Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2024-12-24 23:19:29 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Added Cache's to retroshare configuration.

Browse Source 
* Completed CacheConfig serialisation.
* Switched to Push Cache Mode (faster and more efficient).
* Added CancelCacheTransfer() if we have duplicate HASH / CACHEID.
* Store remote and local caches.
* cleanup all cache directories.
* Broke Cache tests :(  - because of dependency of mAuthMgr/mConnMgr.



git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@343 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
drbob 2008-02-09 12:47:45 +00:00
parent 1bb4283986
commit 276a6bbd99
21 changed files with 796 additions and 352 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
libretroshare/src/dbase/Makefile
View File

@ -10,9 +10,11 @@ include $(RS_TOP_DIR)/scripts/config.mk
RSOBJ = findex.o fimonitor.o cachestrapper.o fistore.o \ RSOBJ = findex.o fimonitor.o cachestrapper.o fistore.o \
rsexpr.o rsexpr.o
TESTOBJ = fitest2.o fisavetest.o ficachetest.o searchtest.o TESTOBJ = fitest2.o fisavetest.o searchtest.o
#ficachetest.o
TESTS = fitest2 fisavetest ficachetest searchtest TESTS = fitest2 fisavetest searchtest
#ficachetest
ifeq ($(OS),Linux) ifeq ($(OS),Linux)
TESTOBJ += fimontest.o TESTOBJ += fimontest.o

445
libretroshare/src/dbase/cachestrapper.cc
View File

@ -22,14 +22,17 @@
*/ */
#include "dbase/cachestrapper.h" #include "dbase/cachestrapper.h"
#include "serialiser/rsconfigitems.h"
#include "pqi/p3connmgr.h"
#include "util/rsdir.h"
#include <iostream> #include <iostream>
#include <sstream> #include <sstream>
#include <iomanip> #include <iomanip>
/** /***/
* #define CS_DEBUG 1 #define CS_DEBUG 1
*/ /***/
bool operator<(const CacheId &a, const CacheId &b) bool operator<(const CacheId &a, const CacheId &b)
{ {
@ -59,8 +62,8 @@ std::ostream &operator<<(std::ostream &out, const CacheData &d)
* *
********************************* Cache Store / Source *************************/ ********************************* Cache Store / Source *************************/
CacheSource::CacheSource(uint16_t t, bool m, std::string cachedir) CacheSource::CacheSource(uint16_t t, bool m, CacheStrapper *cs, std::string cachedir)
:cacheType(t), multiCache(m), cacheDir(cachedir) :cacheType(t), multiCache(m), mStrapper(cs), cacheDir(cachedir)
{ {
return; return;
} }
@ -108,6 +111,9 @@ bool CacheSource::refreshCache(const CacheData &data)
} }
unlockData(); /* UNLOCK MUTEX */ unlockData(); /* UNLOCK MUTEX */
if (mStrapper) /* allow testing without full feedback */
mStrapper->refreshCache(data);
return ret; return ret;
} }
@ -189,8 +195,10 @@ void CacheSource::listCaches(std::ostream &out)
} }
CacheStore::CacheStore(uint16_t t, bool m, CacheTransfer *cft, std::string cachedir) CacheStore::CacheStore(uint16_t t, bool m,
:cacheType(t), multiCache(m), cacheTransfer(cft), cacheDir(cachedir) CacheStrapper *cs, CacheTransfer *cft, std::string cachedir)
:cacheType(t), multiCache(m), mStrapper(cs),
cacheTransfer(cft), cacheDir(cachedir)
{ {
/* not much */ /* not much */
return; return;
@ -292,6 +300,28 @@ bool CacheStore::locked_getStoredCache(CacheData &data)
bool CacheStore::getAllStoredCaches(std::list<CacheData> &data)
{
lockData(); /* LOCK MUTEX */
std::map<RsPeerId, CacheSet>::iterator pit;
for(pit = caches.begin(); pit != caches.end(); pit++)
{
CacheSet::iterator cit;
/* look for subid */
for(cit = (pit->second).begin();
cit != (pit->second).end(); cit++)
{
data.push_back(cit->second);
}
}
unlockData(); /* UNLOCK MUTEX */
return true;
}
/* input from CacheStrapper. /* input from CacheStrapper.
* check if we want to download it... * check if we want to download it...
* determine the new name/path * determine the new name/path
@ -438,6 +468,12 @@ void CacheStore::locked_storeCacheEntry(const CacheData &data)
{ {
(pit->second)[0] = data; (pit->second)[0] = data;
} }
/* tell the strapper we've loaded one */
if (mStrapper)
{
mStrapper->refreshCacheStore(data);
}
return; return;
} }
@ -447,11 +483,9 @@ void CacheStore::locked_storeCacheEntry(const CacheData &data)
* *
********************************* CacheStrapper ********************************/ ********************************* CacheStrapper ********************************/
CacheStrapper::CacheStrapper(RsPeerId id, time_t period) CacheStrapper::CacheStrapper(p3AuthMgr *am, p3ConnectMgr *cm)
:ownId(id), queryPeriod(period) :p3Config(CONFIG_TYPE_CACHE), mAuthMgr(am), mConnMgr(cm)
{ {
/* add OwnId */
addPeerId(ownId);
return; return;
} }
@ -467,12 +501,22 @@ void CacheStrapper::addCachePair(CachePair set)
void CacheStrapper::statusChange(const std::list<pqipeer> &plist) void CacheStrapper::statusChange(const std::list<pqipeer> &plist)
{ {
std::list<pqipeer>::const_iterator it; std::list<pqipeer>::const_iterator it;
std::map<RsPeerId, CacheTS>::iterator mit;
for(it = plist.begin(); it != plist.end(); it++) for(it = plist.begin(); it != plist.end(); it++)
{ {
if (status.end() == (mit = status.find(it->id))) if (it->actions & RS_PEER_CONNECTED)
{ {
addPeerId(it->id); /* grab all the cache ids and add */
std::map<CacheId,CacheData> hashs;
std::map<CacheId,CacheData>::iterator cit;
handleCacheQuery(it->id, hashs);
RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/
for(cit = hashs.begin(); cit != hashs.end(); cit++)
{
mCacheUpdates.push_back(std::make_pair(it->id, cit->second));
}
} }
} }
} }
@ -480,50 +524,51 @@ void CacheStrapper::statusChange(const std::list<pqipeer> &plist)
/**************** from pqimonclient ********************/ /**************** from pqimonclient ********************/
void CacheStrapper::addPeerId(RsPeerId pid) void CacheStrapper::refreshCache(const CacheData &data)
{ {
std::map<RsPeerId, CacheTS>::iterator it; /* we've received an update
* send to all online peers + self
*/
/* just reset it for the moment */ std::list<std::string> ids;
CacheTS ts; std::list<std::string>::iterator it;
ts.query = 0;
ts.answer = 0;
status[pid] = ts; mConnMgr->getOnlineList(ids);
}
bool CacheStrapper::removePeerId(RsPeerId pid) RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/
{ for(it = ids.begin(); it != ids.end(); it++)
std::map<RsPeerId, CacheTS>::iterator it;
if (status.end() != (it = status.find(pid)))
{ {
status.erase(it); mCacheUpdates.push_back(std::make_pair(*it, data));
return true;
} }
return false;
mCacheUpdates.push_back(std::make_pair(mConnMgr->getOwnId(), data));
IndicateConfigChanged(); /**** INDICATE MSG CONFIG CHANGED! *****/
} }
void CacheStrapper::refreshCacheStore(const CacheData &data)
{
/* indicate to save data */
IndicateConfigChanged(); /**** INDICATE MSG CONFIG CHANGED! *****/
}
bool CacheStrapper::getCacheUpdates(std::list<std::pair<RsPeerId, CacheData> > &updates)
{
RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/
updates = mCacheUpdates;
mCacheUpdates.clear();
return true;
}
/* pass to correct CacheSet */ /* pass to correct CacheSet */
void CacheStrapper::recvCacheResponse(CacheData &data, time_t ts) void CacheStrapper::recvCacheResponse(CacheData &data, time_t ts)
{ {
/* update internal data first */
std::map<RsPeerId, CacheTS>::iterator it;
if (status.end() == status.find(data.pid))
{
/* add it in */
CacheTS d;
d.query = 0;
d.answer = 0;
status[data.pid] = d;
}
it = status.find(data.pid); /* will always succeed */
/* update status */
(it -> second).answer = ts;
/* find cache store */ /* find cache store */
std::map<uint16_t, CachePair>::iterator it2; std::map<uint16_t, CachePair>::iterator it2;
if (caches.end() == (it2 = caches.find(data.cid.type))) if (caches.end() == (it2 = caches.find(data.cid.type)))
@ -534,10 +579,12 @@ void CacheStrapper::recvCacheResponse(CacheData &data, time_t ts)
/* notify the CacheStore */ /* notify the CacheStore */
(it2 -> second).store -> availableCache(data); (it2 -> second).store -> availableCache(data);
} }
/* generate periodically or at a change */ /* generate periodically or at a change */
#if 0
bool CacheStrapper::sendCacheQuery(std::list<RsPeerId> &id, time_t ts) bool CacheStrapper::sendCacheQuery(std::list<RsPeerId> &id, time_t ts)
{ {
/* iterate through peers, and see who we haven't got an answer from recently */ /* iterate through peers, and see who we haven't got an answer from recently */
@ -553,6 +600,7 @@ bool CacheStrapper::sendCacheQuery(std::list<RsPeerId> &id, time_t ts)
} }
return (id.size() > 0); return (id.size() > 0);
} }
#endif
void CacheStrapper::handleCacheQuery(RsPeerId id, std::map<CacheId,CacheData> &hashs) void CacheStrapper::handleCacheQuery(RsPeerId id, std::map<CacheId,CacheData> &hashs)
@ -575,8 +623,8 @@ void CacheStrapper::listCaches(std::ostream &out)
{ {
/* can overwrite for more control! */ /* can overwrite for more control! */
std::map<uint16_t, CachePair>::iterator it; std::map<uint16_t, CachePair>::iterator it;
out << "CacheStrapper::listCaches() [" << ownId; out << "CacheStrapper::listCaches() [" << mConnMgr->getOwnId();
out << "] Total Peers: " << status.size() << " Total Caches: " << caches.size(); out << "] " << " Total Caches: " << caches.size();
out << std::endl; out << std::endl;
for(it = caches.begin(); it != caches.end(); it++) for(it = caches.begin(); it != caches.end(); it++)
{ {
@ -592,6 +640,7 @@ void CacheStrapper::listCaches(std::ostream &out)
void CacheStrapper::listPeerStatus(std::ostream &out) void CacheStrapper::listPeerStatus(std::ostream &out)
{ {
#if 0
std::map<RsPeerId, CacheTS>::iterator it; std::map<RsPeerId, CacheTS>::iterator it;
out << "CacheStrapper::listPeerStatus() [" << ownId; out << "CacheStrapper::listPeerStatus() [" << ownId;
out << "] Total Peers: " << status.size() << " Total Caches: " << caches.size(); out << "] Total Peers: " << status.size() << " Total Caches: " << caches.size();
@ -604,6 +653,7 @@ void CacheStrapper::listPeerStatus(std::ostream &out)
out << std::endl; out << std::endl;
} }
return; return;
#endif
} }
@ -620,6 +670,256 @@ bool CacheStrapper::findCache(std::string hash, CacheData &data)
} }
return false; return false;
} }
/***************************************************************************/
/****************************** CONFIGURATION HANDLING *********************/
/***************************************************************************/
/**** OVERLOADED FROM p3Config ****/
RsSerialiser *CacheStrapper::setupSerialiser()
{
RsSerialiser *rss = new RsSerialiser();
/* add in the types we need! */
rss->addSerialType(new RsCacheConfigSerialiser());
return rss;
}
std::list<RsItem *> CacheStrapper::saveList(bool &cleanup)
{
std::list<RsItem *> saveData;
/* it can delete them! */
cleanup = true;
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::saveList()" << std::endl;
#endif
/* iterate through the Caches (local first) */
std::list<CacheData>::iterator cit;
std::list<CacheData> ownCaches;
std::list<CacheData> remoteCaches;
std::string ownId = mConnMgr->getOwnId();
std::map<uint16_t, CachePair>::iterator it;
for(it = caches.begin(); it != caches.end(); it++)
{
std::map<CacheId, CacheData>::iterator tit;
std::map<CacheId, CacheData> ownTmp;
(it->second).source -> cachesAvailable(ownId, ownTmp);
(it->second).store -> getAllStoredCaches(remoteCaches);
for(tit = ownTmp.begin(); tit != ownTmp.end(); tit++)
{
ownCaches.push_back(tit->second);
}
}
for(cit = ownCaches.begin(); cit != ownCaches.end(); cit++)
{
RsCacheConfig *rscc = new RsCacheConfig();
rscc->pid = cit->pid;
//rscc->pname = cit->pname;
rscc->cachetypeid = cit->cid.type;
rscc->cachesubid = cit->cid.subid;
rscc->path = cit->path;
rscc->name = cit->name;
rscc->hash = cit->hash;
rscc->size = cit->size;
rscc->recvd = cit->recvd;
saveData.push_back(rscc);
}
for(cit = remoteCaches.begin(); cit != remoteCaches.end(); cit++)
{
if (cit->pid == ownId)
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() discarding Own Remote Cache";
std::cerr << std::endl;
#endif
continue; /* skip own caches -> will get transferred anyway */
}
RsCacheConfig *rscc = new RsCacheConfig();
rscc->pid = cit->pid;
//rscc->pname = cit->pname;
rscc->cachetypeid = cit->cid.type;
rscc->cachesubid = cit->cid.subid;
rscc->path = cit->path;
rscc->name = cit->name;
rscc->hash = cit->hash;
rscc->size = cit->size;
rscc->recvd = cit->recvd;
saveData.push_back(rscc);
}
/* list completed! */
return saveData;
}
bool CacheStrapper::loadList(std::list<RsItem *> load)
{
std::list<RsItem *>::iterator it;
RsCacheConfig *rscc;
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Item Count: " << load.size();
std::cerr << std::endl;
#endif
std::list<CacheData> ownCaches;
std::list<CacheData> remoteCaches;
std::string ownId = mConnMgr->getOwnId();
std::map<std::string, std::list<std::string> > saveFiles;
std::map<std::string, std::list<std::string> >::iterator sit;
for(it = load.begin(); it != load.end(); it++)
{
/* switch on type */
if (NULL != (rscc = dynamic_cast<RsCacheConfig *>(*it)))
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Item: ";
std::cerr << std::endl;
rscc->print(std::cerr, 10);
std::cerr << std::endl;
#endif
CacheData cd;
cd.pid = rscc->pid;
cd.pname = mAuthMgr->getName(cd.pid);
cd.cid.type = rscc->cachetypeid;
cd.cid.subid = rscc->cachesubid;
cd.path = rscc->path;
cd.name = rscc->name;
cd.hash = rscc->hash;
cd.size = rscc->size;
cd.recvd = rscc->recvd;
/* store files that we want to keep */
(saveFiles[cd.path]).push_back(cd.name);
std::map<uint16_t, CachePair>::iterator it2;
if (caches.end() == (it2 = caches.find(cd.cid.type)))
{
/* error - don't have this type of cache */
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Can't Find Cache discarding";
std::cerr << std::endl;
#endif
}
else
{
if (cd.pid == ownId)
{
/* load local */
(it2 -> second).source -> loadLocalCache(cd);
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() loaded Local";
std::cerr << std::endl;
#endif
}
else
{
/* load remote */
(it2 -> second).store -> loadCache(cd);
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() loaded Remote";
std::cerr << std::endl;
#endif
}
}
/* cleanup */
delete (*it);
}
else
{
/* cleanup */
delete (*it);
}
}
/* assemble a list of dirs to clean (union of cache dirs) */
std::list<std::string> cacheDirs;
std::list<std::string>::iterator dit, fit;
std::map<uint16_t, CachePair>::iterator cit;
for(cit = caches.begin(); cit != caches.end(); cit++)
{
std::string lcdir = (cit->second).source->getCacheDir();
std::string rcdir = (cit->second).store->getCacheDir();
if (cacheDirs.end() == std::find(cacheDirs.begin(), cacheDirs.end(), lcdir))
{
cacheDirs.push_back(lcdir);
}
if (cacheDirs.end() == std::find(cacheDirs.begin(), cacheDirs.end(), rcdir))
{
cacheDirs.push_back(rcdir);
}
}
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Files To Save:" << std::endl;
#endif
for(sit = saveFiles.begin(); sit != saveFiles.end(); sit++)
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Files To Save in dir: <" << sit->first << ">" << std::endl;
#endif
for(fit = (sit->second).begin(); fit != (sit->second).end(); fit++)
{
#ifdef CS_DEBUG
std::cerr << "\tFile: " << *fit << std::endl;
#endif
}
}
std::list<std::string> emptyList;
for(dit = cacheDirs.begin(); dit != cacheDirs.end(); dit++)
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Cleaning cache dir: <" << *dit << ">" << std::endl;
#endif
sit = saveFiles.find(*dit);
if (sit != saveFiles.end())
{
for(fit = (sit->second).begin(); fit != (sit->second).end(); fit++)
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() Keeping File: " << *fit << std::endl;
#endif
}
RsDirUtil::cleanupDirectory(*dit, sit->second);
}
else
{
#ifdef CS_DEBUG
std::cerr << "CacheStrapper::loadList() No Files to save here!" << std::endl;
#endif
RsDirUtil::cleanupDirectory(*dit, emptyList);
}
}
return true;
}
/********************************* CacheStrapper ********************************* /********************************* CacheStrapper *********************************
@ -631,6 +931,43 @@ bool CacheStrapper::findCache(std::string hash, CacheData &data)
/* request from CacheStore */ /* request from CacheStore */
bool CacheTransfer::RequestCache(CacheData &data, CacheStore *cbStore) bool CacheTransfer::RequestCache(CacheData &data, CacheStore *cbStore)
{ {
/* check for a previous request -> and cancel
*
* - if duplicate pid, cid -> cancel old transfer
* - if duplicate hash -> Fail Transfer
*/
std::map<std::string, CacheData>::iterator dit;
std::map<std::string, CacheStore *>::iterator sit;
for(dit = cbData.begin(); dit != cbData.end(); dit++)
{
if (((dit->second).pid == data.pid) &&
((dit->second).cid.type == data.cid.type) &&
((dit->second).cid.subid == data.cid.subid))
{
/* cancel old transfer */
CancelCacheFile(dit->second.pid, dit->second.path,
dit->second.hash, dit->second.size);
sit = cbStores.find(dit->second.hash);
cbData.erase(dit);
cbStores.erase(sit);
break;
}
}
/* find in store.... */
sit = cbStores.find(data.hash);
if (sit != cbStores.end())
{
/* Duplicate Current Request */
cbStore -> failedCache(data);
return false;
}
/* store request */ /* store request */
cbData[data.hash] = data; cbData[data.hash] = data;
cbStores[data.hash] = cbStore; cbStores[data.hash] = cbStore;
@ -657,6 +994,18 @@ bool CacheTransfer::RequestCacheFile(RsPeerId id, std::string path, std::string
return true; return true;
} }
/* to be overloaded */
bool CacheTransfer::CancelCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size)
{
std::cerr << "CacheTransfer::CancelCacheFile() : from:" << id << " #";
std::cerr << hash << " size: " << size;
std::cerr << " savepath: " << path << std::endl;
std::cerr << "CacheTransfer::CancelCacheFile() Dummy fn";
std::cerr << std::endl;
return true;
}
/* internal completion -> does cb */ /* internal completion -> does cb */
bool CacheTransfer::CompletedCache(std::string hash) bool CacheTransfer::CompletedCache(std::string hash)

77
libretroshare/src/dbase/cachestrapper.h
View File

@ -113,6 +113,7 @@ bool RequestCache(CacheData &data, CacheStore *cbStore); /* request from CacheSt
protected: protected:
/* to be overloaded */ /* to be overloaded */
virtual bool RequestCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size); virtual bool RequestCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size);
virtual bool CancelCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size);
bool CompletedCache(std::string hash); /* internal completion -> does cb */ bool CompletedCache(std::string hash); /* internal completion -> does cb */
bool FailedCache(std::string hash); /* internal completion -> does cb */ bool FailedCache(std::string hash); /* internal completion -> does cb */
@ -134,7 +135,7 @@ typedef std::map<uint16_t, CacheData> CacheSet;
class CacheSource class CacheSource
{ {
public: public:
CacheSource(uint16_t t, bool m, std::string cachedir); CacheSource(uint16_t t, bool m, CacheStrapper *cs, std::string cachedir);
virtual ~CacheSource() {} virtual ~CacheSource() {}
/* called to determine available cache for peer - /* called to determine available cache for peer -
@ -165,8 +166,11 @@ bool findCache(std::string hash, CacheData &data);
protected: protected:
/*** MUTEX LOCKING - TODO uint16_t cacheType; /* for checking */
*/ bool multiCache; /* do we care about subid's */
CacheStrapper *mStrapper;
/*** MUTEX LOCKING */
void lockData(); void lockData();
void unlockData(); void unlockData();
@ -174,9 +178,6 @@ void unlockData();
private: private:
uint16_t cacheType; /* for checking */
bool multiCache; /* do we care about subid's */
std::string cacheDir; std::string cacheDir;
RsMutex cMutex; RsMutex cMutex;
}; };
@ -186,11 +187,12 @@ class CacheStore
{ {
public: public:
CacheStore(uint16_t t, bool m, CacheTransfer *cft, std::string cachedir); CacheStore(uint16_t t, bool m, CacheStrapper *cs, CacheTransfer *cft, std::string cachedir);
virtual ~CacheStore() {} virtual ~CacheStore() {}
/* current stored data */ /* current stored data */
bool getStoredCache(CacheData &data); /* use pid/cid in data */ bool getStoredCache(CacheData &data); /* use pid/cid in data */
bool getAllStoredCaches(std::list<CacheData> &data); /* use pid/cid in data */
/* input from CacheStrapper -> store can then download new data */ /* input from CacheStrapper -> store can then download new data */
void availableCache(const CacheData &data); void availableCache(const CacheData &data);
@ -232,12 +234,15 @@ bool locked_getStoredCache(CacheData &data);
uint16_t cacheType; /* for checking */ uint16_t cacheType; /* for checking */
bool multiCache; /* do we care about subid's */ bool multiCache; /* do we care about subid's */
CacheStrapper *mStrapper;
CacheTransfer *cacheTransfer; CacheTransfer *cacheTransfer;
std::string cacheDir; std::string cacheDir;
RsMutex cMutex;
std::map<RsPeerId, CacheSet> caches; std::map<RsPeerId, CacheSet> caches;
RsMutex cMutex;
}; };
@ -263,54 +268,62 @@ class CachePair
bool operator<(const CachePair &a, const CachePair &b); bool operator<(const CachePair &a, const CachePair &b);
class CacheTS
{
public:
time_t query;
time_t answer;
};
#include "pqi/pqimonitor.h" #include "pqi/pqimonitor.h"
#include "pqi/p3cfgmgr.h"
class CacheStrapper: public pqiMonitor class p3AuthMgr;
class CacheStrapper: public pqiMonitor, public p3Config
{ {
public: public:
CacheStrapper(RsPeerId id, time_t period); CacheStrapper(p3AuthMgr *am, p3ConnectMgr *cm);
virtual ~CacheStrapper() { return; } virtual ~CacheStrapper() { return; }
/************* from pqiMonitor *******************/ /************* from pqiMonitor *******************/
virtual void statusChange(const std::list<pqipeer> &plist); virtual void statusChange(const std::list<pqipeer> &plist);
/************* from pqiMonitor *******************/ /************* from pqiMonitor *******************/
/* Feedback from CacheSources */
void refreshCache(const CacheData &data);
void refreshCacheStore(const CacheData &data);
/* list of Caches to send out */
bool getCacheUpdates(std::list<std::pair<RsPeerId, CacheData> > &updates);
void addCachePair(CachePair pair); void addCachePair(CachePair pair);
void addPeerId(RsPeerId pid);
bool removePeerId(RsPeerId pid);
/*** I/O (1) ***/
/* pass to correct CacheSet */
void recvCacheResponse(CacheData &date, time_t ts);
/* generate periodically or at a change */
bool sendCacheQuery(std::list<RsPeerId> &id, time_t ts);
/*** I/O (2) ***/ /*** I/O (2) ***/
/* handle a DirQuery */ void recvCacheResponse(CacheData &data, time_t ts);
void handleCacheQuery(RsPeerId id, std::map<CacheId, CacheData> &data); void handleCacheQuery(RsPeerId id, std::map<CacheId, CacheData> &data);
/* search through CacheSources. */ /* search through CacheSources. */
bool findCache(std::string hash, CacheData &data); bool findCache(std::string hash, CacheData &data);
/* display */ /* display */
void listCaches(std::ostream &out); void listCaches(std::ostream &out);
void listPeerStatus(std::ostream &out); void listPeerStatus(std::ostream &out);
/* Config */
protected:
/* Key Functions to be overloaded for Full Configuration */
virtual RsSerialiser *setupSerialiser();
virtual std::list<RsItem *> saveList(bool &cleanup);
virtual bool loadList(std::list<RsItem *> load);
private: private:
std::map<RsPeerId, CacheTS> status; /* these are static - so shouldn't need mutex */
p3AuthMgr *mAuthMgr;
p3ConnectMgr *mConnMgr;
std::map<uint16_t, CachePair> caches; std::map<uint16_t, CachePair> caches;
RsPeerId ownId;
time_t queryPeriod; RsMutex csMtx; /* protect below */
std::list<std::pair<RsPeerId, CacheData> > mCacheUpdates;
}; };

8
libretroshare/src/dbase/cachetest.h
View File

@ -32,8 +32,8 @@
class CacheTestSource: public CacheSource class CacheTestSource: public CacheSource
{ {
public: public:
CacheTestSource(std::string dir) CacheTestSource(CacheStrapper *cs, std::string dir)
:CacheSource(TESTID, false, dir) { return; } :CacheSource(cs, TESTID, false, dir) { return; }
}; };
class CacheTestStore: public CacheStore class CacheTestStore: public CacheStore
@ -47,8 +47,8 @@ class CacheTestStore: public CacheStore
class CacheTestMultiSource: public CacheSource class CacheTestMultiSource: public CacheSource
{ {
public: public:
CacheTestMultiSource(std::string dir) CacheTestMultiSource(CacheStrapper *cs, std::string dir)
:CacheSource(TESTID2, true, dir) { return; } :CacheSource(cs, TESTID2, true, dir) { return; }
}; };
class CacheTestMultiStore: public CacheStore class CacheTestMultiStore: public CacheStore

22
libretroshare/src/dbase/ficachetest.cc
View File

@ -52,9 +52,13 @@ int main(int argc, char **argv)
RsPeerId pid2("0x0102"); RsPeerId pid2("0x0102");
RsPeerId pid3("0x0103"); RsPeerId pid3("0x0103");
CacheStrapper sc1(pid1, period); p3ConnectMgr *connMgr1 = NULL;
CacheStrapper sc2(pid2, period); p3ConnectMgr *connMgr2 = NULL;
CacheStrapper sc3(pid3, period); p3ConnectMgr *connMgr3 = NULL;
CacheStrapper sc1(connMgr1);
CacheStrapper sc2(connMgr2);
CacheStrapper sc3(connMgr3);
CacheTransfer ctt1(&sc1); CacheTransfer ctt1(&sc1);
CacheTransfer ctt2(&sc2); CacheTransfer ctt2(&sc2);
CacheTransfer ctt3(&sc3); CacheTransfer ctt3(&sc3);
@ -67,15 +71,15 @@ int main(int argc, char **argv)
std::string nulldir = ""; std::string nulldir = "";
CacheSource *csrc1 = new CacheTestSource(nulldir); CacheSource *csrc1 = new CacheTestSource(&sc1, nulldir);
CacheStore *cstore1 = new CacheTestStore(&ctt1, nulldir); CacheStore *cstore1 = new CacheTestStore(&ctt1, nulldir);
CacheId cid1(TESTID, 0); CacheId cid1(TESTID, 0);
CacheSource *csrc2 = new CacheTestSource(nulldir); CacheSource *csrc2 = new CacheTestSource(&sc2, nulldir);
CacheStore *cstore2 = new CacheTestStore(&ctt2, nulldir); CacheStore *cstore2 = new CacheTestStore(&ctt2, nulldir);
CacheId cid2(TESTID, 0); CacheId cid2(TESTID, 0);
CacheSource *csrc3 = new CacheTestSource(nulldir); CacheSource *csrc3 = new CacheTestSource(&sc3, nulldir);
CacheStore *cstore3 = new CacheTestStore(&ctt3, nulldir); CacheStore *cstore3 = new CacheTestStore(&ctt3, nulldir);
CacheId cid3(TESTID, 0); CacheId cid3(TESTID, 0);
@ -87,12 +91,6 @@ int main(int argc, char **argv)
sc2.addCachePair(cp2); sc2.addCachePair(cp2);
sc3.addCachePair(cp3); sc3.addCachePair(cp3);
sc1.addPeerId(pid2);
sc2.addPeerId(pid1);
sc2.addPeerId(pid3);
sc3.addPeerId(pid2);
/* add in a cache to sc2 */ /* add in a cache to sc2 */
CacheData cdata; CacheData cdata;

4
libretroshare/src/dbase/fimonitor.cc
View File

@ -38,8 +38,8 @@
#define FIM_DEBUG 1 #define FIM_DEBUG 1
FileIndexMonitor::FileIndexMonitor(std::string cachedir, std::string pid) FileIndexMonitor::FileIndexMonitor(CacheStrapper *cs, std::string cachedir, std::string pid)
:CacheSource(RS_SERVICE_TYPE_FILE_INDEX, false, cachedir), fi(pid), :CacheSource(RS_SERVICE_TYPE_FILE_INDEX, false, cs, cachedir), fi(pid),
pendingDirs(false), pendingForceCacheWrite(false) pendingDirs(false), pendingForceCacheWrite(false)
{ {

2
libretroshare/src/dbase/fimonitor.h
View File

@ -68,7 +68,7 @@ std::string FileIndexMonitor::findRealRoot(std::string base);
class FileIndexMonitor: public CacheSource, public RsThread class FileIndexMonitor: public CacheSource, public RsThread
{ {
public: public:
FileIndexMonitor(std::string cachedir, std::string pid); FileIndexMonitor(CacheStrapper *cs, std::string cachedir, std::string pid);
virtual ~FileIndexMonitor(); virtual ~FileIndexMonitor();
/* external interface for filetransfer */ /* external interface for filetransfer */

2
libretroshare/src/dbase/fimontest.cc
View File

@ -72,7 +72,7 @@ int main(int argc, char **argv)
sleep(1); sleep(1);
FileIndexMonitor mon("", "OWN ID"); FileIndexMonitor mon(NULL, "", "OWN ID");
/* setup monitor */ /* setup monitor */
mon.setPeriod(period); mon.setPeriod(period);

4
libretroshare/src/dbase/fistore.cc
View File

@ -25,9 +25,9 @@
#include "rsiface/rsexpr.h" #include "rsiface/rsexpr.h"
#include "serialiser/rsserviceids.h" #include "serialiser/rsserviceids.h"
FileIndexStore::FileIndexStore(CacheTransfer *cft, FileIndexStore::FileIndexStore(CacheStrapper *cs, CacheTransfer *cft,
NotifyBase *cb_in, RsPeerId ownid, std::string cachedir) NotifyBase *cb_in, RsPeerId ownid, std::string cachedir)
:CacheStore(RS_SERVICE_TYPE_FILE_INDEX, false, cft, cachedir), :CacheStore(RS_SERVICE_TYPE_FILE_INDEX, false, cs, cft, cachedir),
localId(ownid), localindex(NULL), cb(cb_in) localId(ownid), localindex(NULL), cb(cb_in)
{ {
return; return;

2
libretroshare/src/dbase/fistore.h
View File

@ -63,7 +63,7 @@ class FileIndexStore: public CacheStore
{ {
public: public:
FileIndexStore(CacheTransfer *cft, NotifyBase *cb_in, FileIndexStore(CacheStrapper *cs, CacheTransfer *cft, NotifyBase *cb_in,
RsPeerId ownid, std::string cachedir); RsPeerId ownid, std::string cachedir);
virtual ~FileIndexStore(); virtual ~FileIndexStore();

1
libretroshare/src/pqi/p3cfgmgr.h
View File

@ -62,6 +62,7 @@ const uint32_t CONFIG_TYPE_GENERAL = 0x0001;
const uint32_t CONFIG_TYPE_PEERS = 0x0002; const uint32_t CONFIG_TYPE_PEERS = 0x0002;
const uint32_t CONFIG_TYPE_FSERVER = 0x0003; const uint32_t CONFIG_TYPE_FSERVER = 0x0003;
const uint32_t CONFIG_TYPE_MSGS = 0x0004; const uint32_t CONFIG_TYPE_MSGS = 0x0004;
const uint32_t CONFIG_TYPE_CACHE = 0x0005;
class p3ConfigMgr; class p3ConfigMgr;
class p3AuthMgr; class p3AuthMgr;

69
libretroshare/src/rsserver/p3face-startup.cc
View File

@ -474,18 +474,15 @@ int RsServer::StartupRetroShare(RsInit *config)
((AuthXPGP *) mAuthMgr) -> loadCertificates(oldFormat, oldConfigMap); ((AuthXPGP *) mAuthMgr) -> loadCertificates(oldFormat, oldConfigMap);
/**************************************************************************/ /**************************************************************************/
/* setup classes / structures */ /* setup classes / structures */
/**************************************************************************/ /**************************************************************************/
uint32_t queryPeriod = 60; /* query every 1 minutes -> change later to 600+ */
mConnMgr = new p3ConnectMgr(mAuthMgr); mConnMgr = new p3ConnectMgr(mAuthMgr);
p3UpnpMgr *mUpnpMgr = new upnphandler(); p3UpnpMgr *mUpnpMgr = new upnphandler();
p3DhtMgr *mDhtMgr = new OpenDHTMgr(ownId, mConnMgr); p3DhtMgr *mDhtMgr = new OpenDHTMgr(ownId, mConnMgr);
CacheStrapper *mCacheStrapper = new CacheStrapper(ownId, queryPeriod); CacheStrapper *mCacheStrapper = new CacheStrapper(mAuthMgr, mConnMgr);
ftfiler *mCacheTransfer = new ftfiler(mCacheStrapper); ftfiler *mCacheTransfer = new ftfiler(mCacheStrapper);
SecurityPolicy *none = secpolicy_create(); SecurityPolicy *none = secpolicy_create();
@ -497,14 +494,11 @@ int RsServer::StartupRetroShare(RsInit *config)
server->setConfigDir(config->basedir.c_str()); server->setConfigDir(config->basedir.c_str());
server->setSaveDir(config->homePath.c_str()); /* Default Save Dir - config will overwrite */ server->setSaveDir(config->homePath.c_str()); /* Default Save Dir - config will overwrite */
server->setSearchInterface(pqih, mAuthMgr, mConnMgr); server->setSearchInterface(pqih, mAuthMgr, mConnMgr);
server->setFileCallback(ownId, mCacheStrapper, mCacheTransfer, &(getNotify()));
mConfigMgr = new p3ConfigMgr(mAuthMgr, config->basedir, "rs-v0.4.cfg", "rs-v0.4.sgn"); mConfigMgr = new p3ConfigMgr(mAuthMgr, config->basedir, "rs-v0.4.cfg", "rs-v0.4.sgn");
mGeneralConfig = new p3GeneralConfig(); mGeneralConfig = new p3GeneralConfig();
// Setup Peer Interface.
rsPeers = new p3Peers(mConnMgr, mAuthMgr);
/* create Services */ /* create Services */
ad = new p3disc(mAuthMgr, mConnMgr); ad = new p3disc(mAuthMgr, mConnMgr);
msgSrv = new p3MsgService(mConnMgr); msgSrv = new p3MsgService(mConnMgr);
@ -516,13 +510,25 @@ int RsServer::StartupRetroShare(RsInit *config)
pqih -> addService(chatSrv); pqih -> addService(chatSrv);
pqih -> addService(gameLauncher); pqih -> addService(gameLauncher);
/* so need to Monitor too! */ /* create Cache Services */
std::string config_dir = config->basedir;
std::string localcachedir = config_dir + "/cache/local";
std::string remotecachedir = config_dir + "/cache/remote";
mRanking = new p3Ranking(RS_SERVICE_TYPE_RANK,
mCacheStrapper, mCacheTransfer,
localcachedir, remotecachedir, 3600 * 24 * 30);
CachePair cp(mRanking, mRanking, CacheId(RS_SERVICE_TYPE_RANK, 0));
mCacheStrapper -> addCachePair(cp);
/**************************************************************************/ /**************************************************************************/
mConnMgr->setDhtMgr(mDhtMgr); mConnMgr->setDhtMgr(mDhtMgr);
mConnMgr->setUpnpMgr(mUpnpMgr); mConnMgr->setUpnpMgr(mUpnpMgr);
/**************************************************************************/ /**************************************************************************/
/* need to Monitor too! */
mConnMgr->addMonitor(pqih); mConnMgr->addMonitor(pqih);
mConnMgr->addMonitor(mCacheStrapper); mConnMgr->addMonitor(mCacheStrapper);
@ -535,7 +541,8 @@ int RsServer::StartupRetroShare(RsInit *config)
mConfigMgr->addConfiguration("peers.cfg", mConnMgr); mConfigMgr->addConfiguration("peers.cfg", mConnMgr);
mConfigMgr->addConfiguration("general.cfg", mGeneralConfig); mConfigMgr->addConfiguration("general.cfg", mGeneralConfig);
mConfigMgr->addConfiguration("msgs.cfg", msgSrv); mConfigMgr->addConfiguration("msgs.cfg", msgSrv);
mConfigMgr->addConfiguration("cache.cfg", mCacheStrapper);
/**************************************************************************/ /**************************************************************************/
@ -570,6 +577,9 @@ int RsServer::StartupRetroShare(RsInit *config)
/* trigger generalConfig loading for classes that require it */ /* trigger generalConfig loading for classes that require it */
/**************************************************************************/ /**************************************************************************/
pqih->setConfig(mGeneralConfig);
pqih->load_config();
/**************************************************************************/ /**************************************************************************/
/* Force Any Configuration before Startup (After Load) */ /* Force Any Configuration before Startup (After Load) */
@ -621,29 +631,7 @@ int RsServer::StartupRetroShare(RsInit *config)
/* Force Any Last Configuration Options */ /* Force Any Last Configuration Options */
/**************************************************************************/ /**************************************************************************/
server->StartupMonitor();
/****************** setup new stuff ***************/
pqih->load_config();
/* Must be after server->setSearchInterface()
* and.
* Must be before other Caches are added to the Strapper!
* */
server->setFileCallback(ownId, mCacheStrapper, mCacheTransfer, &(getNotify()));
#ifdef PQI_USE_CHANNELS #ifdef PQI_USE_CHANNELS
server->setP3Channel(pqih->getP3Channel()); server->setP3Channel(pqih->getP3Channel());
@ -660,20 +648,9 @@ int RsServer::StartupRetroShare(RsInit *config)
pqih->AddSearchModule(mod); pqih->AddSearchModule(mod);
/* Setup GUI Interfaces. */
rsPeers = new p3Peers(mConnMgr, mAuthMgr);
/* create Cache Services */
std::string config_dir = config->basedir;
std::string localcachedir = config_dir + "/cache/local";
std::string remotecachedir = config_dir + "/cache/remote";
mRanking = new p3Ranking(RS_SERVICE_TYPE_RANK,
mCacheTransfer, localcachedir, remotecachedir, 3600 * 24 * 30);
CachePair cp(mRanking, mRanking, CacheId(RS_SERVICE_TYPE_RANK, 0));
mCacheStrapper -> addCachePair(cp);
/* setup the gui */
rsGameLauncher = gameLauncher; rsGameLauncher = gameLauncher;
rsRanks = new p3Rank(mRanking); rsRanks = new p3Rank(mRanking);
rsMsgs = new p3Msgs(mAuthMgr, msgSrv, chatSrv); rsMsgs = new p3Msgs(mAuthMgr, msgSrv, chatSrv);

350
libretroshare/src/serialiser/rsconfigitems.cc
View File

@ -843,7 +843,7 @@ RsPeerStunItem *RsPeerConfigSerialiser::deserialiseStun(void *data, uint32_t *si
} }
/****************************************************************************/ /****************************************************************************/
RsCacheConfig::~RsCacheConfig() RsCacheConfig::~RsCacheConfig()
@ -851,43 +851,51 @@ RsCacheConfig::~RsCacheConfig()
return; return;
} }
void RsCacheConfig::clear() void RsCacheConfig::clear()
{ {
pid.clear();
cacheid = 0; cachetypeid = 0;
path = ""; cachesubid = 0;
name = ""; path = "";
hash = ""; name = "";
recvd = 0; hash = "";
size = 0;
} recvd = 0;
std::ostream &RsCacheConfig::print(std::ostream &out, uint16_t indent) }
{
printRsItemBase(out, "RsCacheConfig", indent); // begin 'WRITE' check std::ostream &RsCacheConfig::print(std::ostream &out, uint16_t indent)
uint16_t int_Indent = indent + 2; {
printRsItemBase(out, "RsCacheConfig", indent);
printIndent(out, int_Indent); //indent uint16_t int_Indent = indent + 2;
out << "cacheid: " << cacheid << std::endl; // display value of cacheid
printIndent(out, int_Indent); //indent
printIndent(out, int_Indent); out << "pid: " << pid << std::endl; // display value of peerid
out << "path: " << path << std::endl; // display value of path
printIndent(out, int_Indent); //indent
printIndent(out, int_Indent); out << "cacheid: " << cachetypeid << ":" << cachesubid << std::endl; // display value of cacheid
out << "name: " << name << std::endl; // display value of name
printIndent(out, int_Indent);
printIndent(out, int_Indent); out << "path: " << path << std::endl; // display value of path
out << "hash: " << hash << std::endl; // display value of hash
printIndent(out, int_Indent);
printIndent(out, int_Indent); out << "name: " << name << std::endl; // display value of name
out << "recvd: " << recvd << std::endl; // display value of recvd
printIndent(out, int_Indent);
printRsItemEnd(out, "RsCacheConfig", indent); // end of 'WRITE' check out << "hash: " << hash << std::endl; // display value of hash
return out;
} printIndent(out, int_Indent);
out << "size: " << size << std::endl; // display value of size
/**************************************************************************/
printIndent(out, int_Indent);
out << "recvd: " << recvd << std::endl; // display value of recvd
printRsItemEnd(out, "RsCacheConfig", indent); // end of 'WRITE' check
return out;
}
/**************************************************************************/
RsCacheConfigSerialiser::~RsCacheConfigSerialiser() RsCacheConfigSerialiser::~RsCacheConfigSerialiser()
{ {
@ -895,140 +903,148 @@ RsCacheConfigSerialiser::~RsCacheConfigSerialiser()
} }
uint32_t RsCacheConfigSerialiser::size(RsItem *i) uint32_t RsCacheConfigSerialiser::size(RsItem *i)
{ {
RsCacheConfig *item = (RsCacheConfig *) i; RsCacheConfig *item = (RsCacheConfig *) i;
uint32_t s = 8; // to store calculated size, initiailize with size of header
s += 2; /* cacheid */
s += GetTlvStringSize(item->path);
s += GetTlvStringSize(item->name);
s += GetTlvStringSize(item->hash);
s += 2; /* recvd */
return s;
}
bool RsCacheConfigSerialiser::serialise(RsItem *i, void *data, uint32_t *size)
{
RsCacheConfig *item = (RsCacheConfig *) i;
uint32_t tlvsize = RsCacheConfigSerialiser::size(item);
uint32_t offset = 0;
if(*size < tlvsize)
return false; /* not enough space */
*size = tlvsize;
bool ok = true;
ok &=setRsItemHeader(data, tlvsize, item->PacketId(), tlvsize);
std::cerr << "RsCacheConfigSerialiser::serialise() Header: " << ok << std::endl;
std::cerr << "RsCacheConfigSerialiser::serialise() Size: " << size << std::endl;
/* skip the header */
offset += 8;
/* add the mandatory parts first */
ok &= setRawUInt32(data, tlvsize, &offset, item->cacheid);
std::cerr << "RsCacheConfigSerialiser::serialise() cacheid: " << ok << std::endl;
ok &= setRawUInt32(data, tlvsize, &offset, item->recvd);
std::cerr << "RsCacheConfigSerialiser::serialise() recvd: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PATH, item->path);
std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_NAME, item->name);
std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash);
std::cerr << "RsCacheConfigSerialiser::serialise() hash: " << ok << std::endl;
if (offset !=tlvsize)
{
ok = false;
std::cerr << "RsConfigSerialiser::serialisertransfer() Size Error! " << std::endl;
}
return ok;
}
RsItem *RsCacheConfigSerialiser::deserialise(void *data, uint32_t *size)
{/* get the type and size */
uint32_t rstype = getRsItemId(data);
uint32_t rssize = getRsItemSize(data);
uint32_t *offset;
*offset = 0;
uint32_t tlvend = *offset + rssize;
if ((RS_PKT_VERSION1 != getRsItemVersion(rstype)) ||
(RS_PKT_CLASS_CONFIG != getRsItemClass(rstype)) ||
(RS_PKT_TYPE_CACHE_CONFIG != getRsItemType(rstype)) ||
(RS_PKT_SUBTYPE_DEFAULT != getRsItemSubType(rstype)))
{
return NULL; /* wrong type */
}
if (*size < rssize) /* check size */
return NULL; /* not enough data */
/* set the packet length */
*size = rssize;
bool ok = true;
/* ready to load */
RsCacheConfig *item = new RsCacheConfig();
item->clear();
/* skip the header */
offset += 8;
/* get mandatory parts first */
ok &= getRawUInt32(data, rssize, offset, &(item->cacheid));
std::cerr << "RsCacheConfigSerialiser::deserialise() cacheid: " << ok << std::endl;
ok &= getRawUInt32(data, rssize, offset, &(item->recvd));
std::cerr << "RsCacheConfigSerialiser::deserialise() recvd: " << ok << std::endl;
while((*offset) + 2 < tlvend)
{
/* get the next type */
uint16_t tlvsubtype = GetTlvType( &(((uint8_t *) data)[*offset]) );
switch(tlvsubtype)
{
case TLV_TYPE_STR_PATH:
ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->path);
break;
case TLV_TYPE_STR_NAME:
ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->name);
break;
case TLV_TYPE_STR_HASH_SHA1:
ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->hash);
break;
default:
break;
}
if (!ok) uint32_t s = 8; // to store calculated size, initiailize with size of header
{
return false;
} s += GetTlvStringSize(item->pid);
s += 2; /* cachetypeid */
s += 2; /* cachesubid */
s += GetTlvStringSize(item->path);
s += GetTlvStringSize(item->name);
s += GetTlvStringSize(item->hash);
s += 8; /* size */
s += 4; /* recvd */
return s;
}
bool RsCacheConfigSerialiser::serialise(RsItem *i, void *data, uint32_t *size)
{
RsCacheConfig *item = (RsCacheConfig *) i;
uint32_t tlvsize = RsCacheConfigSerialiser::size(item);
uint32_t offset = 0;
if(*size < tlvsize)
return false; /* not enough space */
*size = tlvsize;
bool ok = true;
ok &=setRsItemHeader(data, tlvsize, item->PacketId(), tlvsize);
std::cerr << "RsCacheConfigSerialiser::serialise() Header: " << ok << std::endl;
std::cerr << "RsCacheConfigSerialiser::serialise() Size: " << size << std::endl;
/* skip the header */
offset += 8;
/* add the mandatory parts first */
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PEERID, item->pid);
std::cerr << "RsCacheConfigSerialiser::serialise() peerid: " << ok << std::endl;
ok &= setRawUInt16(data, tlvsize, &offset, item->cachetypeid);
std::cerr << "RsCacheConfigSerialiser::serialise() cacheTypeId: " << ok << std::endl;
ok &= setRawUInt16(data, tlvsize, &offset, item->cachesubid);
std::cerr << "RsCacheConfigSerialiser::serialise() cacheSubId: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PATH, item->path);
std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_NAME, item->name);
std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl;
ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash);
std::cerr << "RsCacheConfigSerialiser::serialise() hash: " << ok << std::endl;
ok &= setRawUInt64(data, tlvsize, &offset, item->size);
std::cerr << "RsCacheConfigSerialiser::serialise() size: " << ok << std::endl;
ok &= setRawUInt32(data, tlvsize, &offset, item->recvd);
std::cerr << "RsCacheConfigSerialiser::serialise() recvd: " << ok << std::endl;
if (offset !=tlvsize)
{
ok = false;
std::cerr << "RsConfigSerialiser::serialisertransfer() Size Error! " << std::endl;
}
return ok;
}
RsItem *RsCacheConfigSerialiser::deserialise(void *data, uint32_t *size)
{/* get the type and size */
uint32_t rstype = getRsItemId(data);
uint32_t rssize = getRsItemSize(data);
uint32_t offset;
offset = 0;
if ((RS_PKT_VERSION1 != getRsItemVersion(rstype)) ||
(RS_PKT_CLASS_CONFIG != getRsItemClass(rstype)) ||
(RS_PKT_TYPE_CACHE_CONFIG != getRsItemType(rstype)) ||
(RS_PKT_SUBTYPE_DEFAULT != getRsItemSubType(rstype)))
{
return NULL; /* wrong type */
}
if (*size < rssize) /* check size */
return NULL; /* not enough data */
/* set the packet length */
*size = rssize;
bool ok = true;
/* ready to load */
RsCacheConfig *item = new RsCacheConfig();
item->clear();
/* skip the header */
offset += 8;
/* get mandatory parts first */
ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_PEERID, item->pid);
std::cerr << "RsCacheConfigSerialiser::deserialise() peerid: " << ok << std::endl;
ok &= getRawUInt16(data, rssize, &offset, &(item->cachetypeid));
std::cerr << "RsCacheConfigSerialiser::serialise() cacheTypeId: " << ok << std::endl;
ok &= getRawUInt16(data, rssize, &offset, &(item->cachesubid));
std::cerr << "RsCacheConfigSerialiser::serialise() cacheSubId: " << ok << std::endl;
ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_PATH, item->path);
std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl;
ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_NAME, item->name);
std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl;
ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash);
std::cerr << "RsCacheConfigSerialiser::deserialise() hash: " << ok << std::endl;
ok &= getRawUInt64(data, rssize, &offset, &(item->size));
std::cerr << "RsCacheConfigSerialiser::deserialise() size: " << ok << std::endl;
ok &= getRawUInt32(data, rssize, &offset, &(item->recvd));
std::cerr << "RsCacheConfigSerialiser::deserialise() recvd: " << ok << std::endl;
if (offset != rssize)
{
/* error */
delete item;
return NULL;
} }
return item; return item;
} }

6
libretroshare/src/serialiser/rsconfigitems.h
View File

@ -126,12 +126,14 @@ virtual ~RsCacheConfig();
virtual void clear(); virtual void clear();
std::ostream &print(std::ostream &out, uint16_t indent = 0); std::ostream &print(std::ostream &out, uint16_t indent = 0);
//RsTlvPeerId peerid; /* Mandatory */ std::string pid; /* Mandatory */
uint32_t cacheid; /* Mandatory */ uint16_t cachetypeid; /* Mandatory */
uint16_t cachesubid; /* Mandatory */
std::string path; /* Mandatory */ std::string path; /* Mandatory */
std::string name; /* Mandatory */ std::string name; /* Mandatory */
std::string hash; /* Mandatory */ std::string hash; /* Mandatory */
uint64_t size; /* Mandatory */
uint32_t recvd; /* Mandatory */ uint32_t recvd; /* Mandatory */
}; };

4
libretroshare/src/server/Makefile
View File

@ -9,9 +9,9 @@ include $(RS_TOP_DIR)/scripts/config.mk
RSOBJ = ft.o ftfiler.o hashsearch.o filedexserver.o RSOBJ = ft.o ftfiler.o hashsearch.o filedexserver.o
TESTOBJ = ftcachetest.o TESTOBJ = #ftcachetest.o
TESTS = ftcachetest TESTS = #ftcachetest
all: librs tests all: librs tests

57
libretroshare/src/server/filedexserver.cc
View File

@ -217,10 +217,10 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe
/* setup FiStore/Monitor */ /* setup FiStore/Monitor */
std::string localcachedir = config_dir + "/cache/local"; std::string localcachedir = config_dir + "/cache/local";
std::string remotecachedir = config_dir + "/cache/remote"; std::string remotecachedir = config_dir + "/cache/remote";
fiStore = new FileIndexStore(ftFiler, cb, ownId, remotecachedir); fiStore = new FileIndexStore(strapper, ftFiler, cb, ownId, remotecachedir);
/* now setup the FiMon */ /* now setup the FiMon */
fimon = new FileIndexMonitor(localcachedir, ownId); fimon = new FileIndexMonitor(strapper, localcachedir, ownId);
/* setup ftFiler /* setup ftFiler
* to find peer info / savedir * to find peer info / savedir
@ -234,10 +234,7 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe
CachePair cp(fimon, fiStore, CacheId(RS_SERVICE_TYPE_FILE_INDEX, 0)); CachePair cp(fimon, fiStore, CacheId(RS_SERVICE_TYPE_FILE_INDEX, 0));
mCacheStrapper -> addCachePair(cp); mCacheStrapper -> addCachePair(cp);
/* now we can load the cache configuration */ #if 0
//std::string cacheconfig = config_dir + "/caches.cfg";
//mCacheStrapper -> loadCaches(cacheconfig);
/************ TMP HACK LOAD until new serialiser is finished */ /************ TMP HACK LOAD until new serialiser is finished */
/* get filename and hash from configuration */ /* get filename and hash from configuration */
std::string localCacheFile; // = getSSLRoot()->getSetting(LOCAL_CACHE_FILE_KEY); std::string localCacheFile; // = getSSLRoot()->getSetting(LOCAL_CACHE_FILE_KEY);
@ -270,8 +267,14 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe
RsDirUtil::cleanupDirectory(remotecachedir, saveRemoteCaches); /* clean up all */ RsDirUtil::cleanupDirectory(remotecachedir, saveRemoteCaches); /* clean up all */
/************ TMP HACK LOAD until new serialiser is finished */ /************ TMP HACK LOAD until new serialiser is finished */
#endif
return;
}
void filedexserver::StartupMonitor()
{
/* startup the FileMonitor (after cache load) */ /* startup the FileMonitor (after cache load) */
fimon->setPeriod(600); /* 10 minutes */ fimon->setPeriod(600); /* 10 minutes */
/* start it up */ /* start it up */
@ -296,9 +299,10 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe
delete rsft; delete rsft;
} }
mResumeTransferList.clear(); mResumeTransferList.clear();
} }
int filedexserver::FileCacheSave() int filedexserver::FileCacheSave()
{ {
/************ TMP HACK SAVE until new serialiser is finished */ /************ TMP HACK SAVE until new serialiser is finished */
@ -441,29 +445,28 @@ int filedexserver::handleInputQueues()
i_init = i; i_init = i;
while((cr = pqisi -> RequestedSearch()) != NULL) while((cr = pqisi -> RequestedSearch()) != NULL)
{ {
//std::cerr << "filedexserver::handleInputQueues() Recvd RequestedSearch (CacheQuery!)" << std::endl; /* just delete these */
std::ostringstream out; std::ostringstream out;
if (i++ == i_init) out << "Requested Search:" << std::endl;
{
out << "Requested Search:" << std::endl;
}
cr -> print(out); cr -> print(out);
pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out.str()); pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out.str());
delete cr;
}
/* these go to the CacheStrapper (handled immediately) */
std::map<CacheId, CacheData>::iterator it;
std::map<CacheId, CacheData> answer;
RsPeerId id;
mCacheStrapper->handleCacheQuery(id, answer); // Now handle it replacement (pushed cache results)
for(it = answer.begin(); it != answer.end(); it++) {
std::list<std::pair<RsPeerId, CacheData> > cacheUpdates;
std::list<std::pair<RsPeerId, CacheData> >::iterator it;
mCacheStrapper->getCacheUpdates(cacheUpdates);
for(it = cacheUpdates.begin(); it != cacheUpdates.end(); it++)
{ {
//std::cerr << "filedexserver::handleInputQueues() Sending (CacheAnswer!)" << std::endl;
/* construct reply */ /* construct reply */
RsCacheItem *ci = new RsCacheItem(); RsCacheItem *ci = new RsCacheItem();
/* id from incoming */ /* id from incoming */
ci -> PeerId(cr->PeerId()); ci -> PeerId(it->first);
ci -> file.hash = (it->second).hash; ci -> file.hash = (it->second).hash;
ci -> file.name = (it->second).name; ci -> file.name = (it->second).name;
@ -473,13 +476,13 @@ int filedexserver::handleInputQueues()
ci -> cacheSubId = (it->second).cid.subid; ci -> cacheSubId = (it->second).cid.subid;
std::ostringstream out2; std::ostringstream out2;
out2 << "Outgoing CacheStrapper reply -> RsCacheItem:" << std::endl; out2 << "Outgoing CacheStrapper Update -> RsCacheItem:" << std::endl;
ci -> print(out2); ci -> print(out2);
pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out2.str()); std::cerr << out2.str() << std::endl;
//pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out2.str());
pqisi -> SendSearchResult(ci); pqisi -> SendSearchResult(ci);
} }
delete cr;
} }
// now File Input. // now File Input.
@ -545,11 +548,13 @@ int filedexserver::handleOutputQueues()
//std::cerr << "filedexserver::handleOutputQueues()" << std::endl; //std::cerr << "filedexserver::handleOutputQueues()" << std::endl;
int i = 0; int i = 0;
#if 0 /* no more cache queries -> results are pushed */
std::list<RsPeerId> ids; std::list<RsPeerId> ids;
std::list<RsPeerId>::iterator pit; std::list<RsPeerId>::iterator pit;
mCacheStrapper->sendCacheQuery(ids, time(NULL)); mCacheStrapper->sendCacheQuery(ids, time(NULL));
for(pit = ids.begin(); pit != ids.end(); pit++) for(pit = ids.begin(); pit != ids.end(); pit++)
{ {
//std::cerr << "filedexserver::handleOutputQueues() Cache Query for: " << (*pit) << std::endl; //std::cerr << "filedexserver::handleOutputQueues() Cache Query for: " << (*pit) << std::endl;
@ -569,6 +574,8 @@ int filedexserver::handleOutputQueues()
/* send it off */ /* send it off */
pqisi -> SearchSpecific(cr); pqisi -> SearchSpecific(cr);
} }
#endif
/* now see if the filer has any data */ /* now see if the filer has any data */
ftFileRequest *ftr; ftFileRequest *ftr;

3
libretroshare/src/server/filedexserver.h
View File

@ -154,10 +154,13 @@ bool loadConfigMap(std::map<std::string, std::string> &configMap);
int FileStoreTick(); int FileStoreTick();
int FileCacheSave(); int FileCacheSave();
/* Setup */
void initialiseFileStore(); void initialiseFileStore();
void setFileCallback(std::string ownId, CacheStrapper *strapper, void setFileCallback(std::string ownId, CacheStrapper *strapper,
ftfiler *ft, NotifyBase *cb); ftfiler *ft, NotifyBase *cb);
void StartupMonitor();
/* Controls */
int RequestDirDetails(std::string uid, std::string path, DirDetails &details); int RequestDirDetails(std::string uid, std::string path, DirDetails &details);
int RequestDirDetails(void *ref, DirDetails &details, uint32_t flags); int RequestDirDetails(void *ref, DirDetails &details, uint32_t flags);

40
libretroshare/src/server/ftfiler.cc
View File

@ -173,6 +173,46 @@ bool ftfiler::RequestCacheFile(std::string id, std::string destpath, std::str
return 1; return 1;
} }
bool ftfiler::CancelCacheFile(RsPeerId id, std::string path,
std::string hash, uint64_t size)
{
/* clean up old transfer - just remove it (no callback) */
{
std::ostringstream out;
out << "ftfiler::CancelCacheFile() Looking for: " << hash;
pqioutput(PQL_DEBUG_BASIC, ftfilerzone, out.str());
}
/* iterate through fileItems and check for this one */
std::list<ftFileStatus *>::iterator it;
for(it = recvFiles.begin(); it != recvFiles.end(); it++)
{
if ((hash==(*it)->hash) &&
(size==(*it)->size) &&
((*it)->ftMode == FT_MODE_CACHE))
{
std::ostringstream out;
out << "ftfiler::CancelCacheFile() ";
out << "Match ftFileStatus: " << hash;
pqioutput(PQL_DEBUG_BASIC, ftfilerzone, out.str());
/* same */
std::cerr << "Clearing Failed Cache Transfer: " << (*it)->name;
std::cerr << std::endl;
delete (*it);
it = recvFiles.erase(it);
return true;
}
}
std::cerr << "************* ERROR *****************";
std::cerr << std::endl;
std::cerr << "ftfiler::CancelCacheFile() Failed to Find: " << hash;
std::cerr << std::endl;
return false;
}
ftFileStatus *ftfiler::findRecvFileItem(std::string hash) ftFileStatus *ftfiler::findRecvFileItem(std::string hash)

3
libretroshare/src/server/ftfiler.h
View File

@ -159,6 +159,9 @@ virtual ~ftfiler() { return; }
virtual bool RequestCacheFile(std::string id, std::string path, virtual bool RequestCacheFile(std::string id, std::string path,
std::string hash, uint64_t size); std::string hash, uint64_t size);
virtual bool CancelCacheFile(RsPeerId id, std::string path,
std::string hash, uint64_t size);
virtual int getFile(std::string name, std::string hash, virtual int getFile(std::string name, std::string hash,
uint64_t size, std::string destpath); uint64_t size, std::string destpath);

41
libretroshare/src/services/p3ranking.cc
View File

@ -42,11 +42,11 @@ std::string generateRandomLinkId();
#define RANK_DEBUG 1 #define RANK_DEBUG 1
p3Ranking::p3Ranking(uint16_t type, CacheTransfer *cft, p3Ranking::p3Ranking(uint16_t type, CacheStrapper *cs, CacheTransfer *cft,
std::string sourcedir, std::string storedir, std::string sourcedir, std::string storedir,
uint32_t storePeriod) uint32_t storePeriod)
:CacheSource(type, true, sourcedir), :CacheSource(type, true, cs, sourcedir),
CacheStore(type, true, cft, storedir), CacheStore(type, true, cs, cft, storedir),
mStorePeriod(storePeriod) mStorePeriod(storePeriod)
{ {
@ -64,7 +64,35 @@ p3Ranking::p3Ranking(uint16_t type, CacheTransfer *cft,
bool p3Ranking::loadLocalCache(const CacheData &data) bool p3Ranking::loadLocalCache(const CacheData &data)
{ {
/* ignore Local Cache -> just use remote caches */ std::string filename = data.path + '/' + data.name;
std::string hash = data.hash;
//uint64_t size = data.size;
std::string source = data.pid;
#ifdef RANK_DEBUG
std::cerr << "p3Ranking::loadLocalCache()";
std::cerr << std::endl;
std::cerr << "\tSource: " << source;
std::cerr << std::endl;
std::cerr << "\tFilename: " << filename;
std::cerr << std::endl;
std::cerr << "\tHash: " << hash;
std::cerr << std::endl;
std::cerr << "\tSize: " << data.size;
std::cerr << std::endl;
#endif
loadRankFile(filename, source);
{
RsStackMutex stack(mRankMtx); /********** STACK LOCKED MTX ******/
mRepublish = false;
}
if (data.size > 0) /* don't refresh zero sized caches */
{
refreshCache(data);
}
return true; return true;
} }
@ -90,6 +118,11 @@ int p3Ranking::loadCache(const CacheData &data)
loadRankFile(filename, source); loadRankFile(filename, source);
CacheStore::lockData(); /***** LOCK ****/
locked_storeCacheEntry(data);
CacheStore::unlockData(); /***** UNLOCK ****/
return 1; return 1;
} }

2
libretroshare/src/services/p3ranking.h
View File

@ -65,7 +65,7 @@ class p3Ranking: public CacheSource, public CacheStore
{ {
public: public:
p3Ranking(uint16_t type, CacheTransfer *cft, p3Ranking(uint16_t type, CacheStrapper *cs, CacheTransfer *cft,
std::string sourcedir, std::string storedir, std::string sourcedir, std::string storedir,
uint32_t storePeriod); uint32_t storePeriod);