Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2025-02-03 11:00:14 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #2203 from csoler/v0.6-CacheTests_2

Browse Source 
Cache tests 2
This commit is contained in:
csoler 2021-01-07 18:07:55 +01:00 committed by GitHub
commit d0b8c7dd69
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 612 additions and 679 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

184
libretroshare/src/gxs/rsdataservice.cc
View File

@ -123,6 +123,7 @@ RsDataService::RsDataService(const std::string &serviceDir, const std::string &d
bool isNewDatabase = !RsDirUtil::fileExists(mDbPath);
mDb = new RetroDb(mDbPath, RetroDb::OPEN_READWRITE_CREATE, key);
mUseCache = true;
initialise(isNewDatabase);
@ -482,7 +483,7 @@ bool RsDataService::finishReleaseUpdate(int release, bool result)
return result;
}
RsGxsGrpMetaData* RsDataService::locked_getGrpMeta(RetroCursor &c, int colOffset,bool use_cache)
std::shared_ptr<RsGxsGrpMetaData> RsDataService::locked_getGrpMeta(RetroCursor& c, int colOffset)
{
#ifdef RS_DATA_SERVICE_DEBUG
std::cerr << "RsDataService::locked_getGrpMeta()" << std::endl;
@ -499,16 +500,16 @@ RsGxsGrpMetaData* RsDataService::locked_getGrpMeta(RetroCursor &c, int colOffset
std::string tempId;
c.getString(mColGrpMeta_GrpId + colOffset, tempId);
RsGxsGrpMetaData* grpMeta ;
std::shared_ptr<RsGxsGrpMetaData> grpMeta ;
RsGxsGroupId grpId(tempId) ;
if(grpId.isNull()) // not in the DB!
return nullptr;
if(use_cache)
if(mUseCache)
grpMeta = mGrpMetaDataCache.getOrCreateMeta(grpId);
else
grpMeta = new RsGxsGrpMetaData();
grpMeta = std::make_shared<RsGxsGrpMetaData>();
if(!grpMeta->mGroupId.isNull()) // the grpMeta is already initialized because it comes from the cache
return grpMeta;
@ -596,11 +597,7 @@ RsGxsGrpMetaData* RsDataService::locked_getGrpMeta(RetroCursor &c, int colOffset
if(ok)
return grpMeta;
else
{
if(!use_cache)
delete grpMeta;
return NULL;
}
}
RsNxsGrp* RsDataService::locked_getGroup(RetroCursor &c)
@ -645,9 +642,8 @@ RsNxsGrp* RsDataService::locked_getGroup(RetroCursor &c)
return NULL;
}
RsGxsMsgMetaData* RsDataService::locked_getMsgMeta(RetroCursor &c, int colOffset,bool use_cache)
std::shared_ptr<RsGxsMsgMetaData> RsDataService::locked_getMsgMeta(RetroCursor &c, int colOffset)
{
bool ok = true;
uint32_t data_len = 0,
offset = 0;
@ -667,12 +663,12 @@ RsGxsMsgMetaData* RsDataService::locked_getMsgMeta(RetroCursor &c, int colOffset
if(group_id.isNull() || msg_id.isNull())
return nullptr;
RsGxsMsgMetaData* msgMeta = nullptr;
std::shared_ptr<RsGxsMsgMetaData> msgMeta;
if(use_cache)
if(mUseCache)
msgMeta = mMsgMetaDataCache[group_id].getOrCreateMeta(msg_id);
else
msgMeta = new RsGxsMsgMetaData();
msgMeta = std::make_shared<RsGxsMsgMetaData>();
if(!msgMeta->mGroupId.isNull()) // we cannot do that because the cursor needs to advance. Is there a method to skip some data in the db?
return msgMeta;
@ -712,10 +708,8 @@ RsGxsMsgMetaData* RsDataService::locked_getMsgMeta(RetroCursor &c, int colOffset
if(ok)
return msgMeta;
else if(!use_cache)
delete msgMeta;
return NULL;
return nullptr;
}
@ -754,10 +748,9 @@ RsNxsMsg* RsDataService::locked_getMessage(RetroCursor &c)
if(ok)
return msg;
else
delete msg;
return NULL;
delete msg;
return nullptr;
}
int RsDataService::storeMessage(const std::list<RsNxsMsg*>& msg)
@ -842,8 +835,10 @@ int RsDataService::storeMessage(const std::list<RsNxsMsg*>& msg)
// This is needed so that mLastPost is correctly updated in the group meta when it is re-loaded.
mGrpMetaDataCache.clear(msgMetaPtr->mGroupId);
mMsgMetaDataCache[msgMetaPtr->mGroupId].updateMeta(msgMetaPtr->mMsgId,*msgMetaPtr);
if(mUseCache)
mMsgMetaDataCache[msgMetaPtr->mGroupId].updateMeta(msgMetaPtr->mMsgId,*msgMetaPtr);
delete *mit;
}
// finish transaction
@ -944,6 +939,8 @@ int RsDataService::storeGroup(const std::list<RsNxsGrp*>& grp)
std::cerr << "\t For GroupId: " << grpMetaPtr->mGroupId.toStdString();
std::cerr << std::endl;
}
delete *sit;
}
// finish transaction
bool ret = mDb->commitTransaction();
@ -1020,6 +1017,8 @@ int RsDataService::updateGroup(const std::list<RsNxsGrp *> &grp)
mDb->sqlUpdate(GRP_TABLE_NAME, "grpId='" + grpPtr->grpId.toStdString() + "'", cv);
mGrpMetaDataCache.updateMeta(grpMetaPtr->mGroupId,*grpMetaPtr);
delete *sit;
}
// finish transaction
bool ret = mDb->commitTransaction();
@ -1059,7 +1058,7 @@ bool RsDataService::validSize(RsNxsGrp* grp) const
return false;
}
int RsDataService::retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp *> &grp, bool withMeta, bool /* cache */)
int RsDataService::retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp *> &grp, bool withMeta)
{
#ifdef RS_DATA_SERVICE_DEBUG_TIME
rstime::RsScopeTimer timer("");
@ -1067,8 +1066,8 @@ int RsDataService::retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp *> &grp, bool
int requestedGroups = grp.size();
#endif
if(grp.empty()){
if(grp.empty())
{
RsStackMutex stack(mDbMutex);
RetroCursor* c = mDb->sqlQuery(GRP_TABLE_NAME, withMeta ? mGrpColumnsWithMeta : mGrpColumns, "", "");
@ -1091,8 +1090,9 @@ int RsDataService::retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp *> &grp, bool
delete c;
}
}else{
}
else
{
RsStackMutex stack(mDbMutex);
std::map<RsGxsGroupId, RsNxsGrp *>::iterator mit = grp.begin();
@ -1138,8 +1138,8 @@ int RsDataService::retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp *> &grp, bool
return 1;
}
void RsDataService::locked_retrieveGroups(RetroCursor* c, std::vector<RsNxsGrp*>& grps, int metaOffset){
void RsDataService::locked_retrieveGroups(RetroCursor* c, std::vector<RsNxsGrp*>& grps, int metaOffset)
{
if(c){
bool valid = c->moveToFirst();
@ -1150,7 +1150,7 @@ void RsDataService::locked_retrieveGroups(RetroCursor* c, std::vector<RsNxsGrp*>
if(g)
{
if (metaOffset)
g->metaData = locked_getGrpMeta(*c, metaOffset,false);
g->metaData = new RsGxsGrpMetaData(*locked_getGrpMeta(*c, metaOffset));
else
g->metaData = nullptr;
@ -1161,9 +1161,7 @@ void RsDataService::locked_retrieveGroups(RetroCursor* c, std::vector<RsNxsGrp*>
}
}
int RsDataService::retrieveNxsMsgs(
const GxsMsgReq &reqIds, GxsMsgResult &msg, bool /* cache */,
bool withMeta )
int RsDataService::retrieveNxsMsgs(const GxsMsgReq &reqIds, GxsMsgResult &msg, bool withMeta)
{
#ifdef RS_DATA_SERVICE_DEBUG_TIME
rstime::RsScopeTimer timer("");
@ -1186,9 +1184,7 @@ int RsDataService::retrieveNxsMsgs(
RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, withMeta ? mMsgColumnsWithMeta : mMsgColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "'", "");
if(c)
{
locked_retrieveMessages(c, msgSet, withMeta ? mColMsg_WithMetaOffset : 0);
}
delete c;
}
@ -1238,7 +1234,7 @@ void RsDataService::locked_retrieveMessages(RetroCursor *c, std::vector<RsNxsMsg
if(m){
if (metaOffset)
m->metaData = locked_getMsgMeta(*c, metaOffset,false);
m->metaData = new RsGxsMsgMetaData(*locked_getMsgMeta(*c, metaOffset));
else
m->metaData = nullptr;
@ -1267,20 +1263,23 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
// if vector empty then request all messages
t_MetaDataCache<RsGxsMessageId,RsGxsMsgMetaData>& cache(mMsgMetaDataCache[grpId]);
// The pointer here is a trick to not initialize a new cache entry when cache is disabled, while keeping the unique variable all along.
t_MetaDataCache<RsGxsMessageId,RsGxsMsgMetaData> *cache(mUseCache? (&mMsgMetaDataCache[grpId]) : nullptr);
if(msgIdV.empty())
{
if(cache.isCacheUpToDate())
cache.getFullMetaList(msgMeta[grpId]);
if(mUseCache && cache->isCacheUpToDate())
cache->getFullMetaList(msgMeta[grpId]);
else
{
RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "'", "");
if (c)
{
locked_retrieveMsgMetaList(c, msgMeta[grpId]);
cache.setCacheUpToDate(true);
locked_retrieveMsgMetaList(c, msgMeta[grpId]);
if(mUseCache)
cache->setCacheUpToDate(true);
}
delete c;
}
@ -1297,7 +1296,7 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
{
const RsGxsMessageId& msgId = *sit;
RsGxsMsgMetaData *meta = cache.getMeta(msgId);
auto meta = mUseCache?cache->getMeta(msgId): (std::shared_ptr<RsGxsMsgMetaData>());
if(meta)
metaSet.push_back(meta);
@ -1306,7 +1305,7 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "' AND " + KEY_MSG_ID + "='" + msgId.toStdString() + "'", "");
c->moveToFirst();
RsGxsMsgMetaData* meta = locked_getMsgMeta(*c, 0,true);
auto meta = locked_getMsgMeta(*c, 0);
if(meta)
metaSet.push_back(meta);
@ -1328,7 +1327,7 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
return 1;
}
void RsDataService::locked_retrieveGrpMetaList(RetroCursor *c, std::map<RsGxsGroupId,RsGxsGrpMetaData *>& grpMeta)
void RsDataService::locked_retrieveGrpMetaList(RetroCursor *c, std::map<RsGxsGroupId,std::shared_ptr<RsGxsGrpMetaData> >& grpMeta)
{
if(!c)
{
@ -1340,15 +1339,16 @@ void RsDataService::locked_retrieveGrpMetaList(RetroCursor *c, std::map<RsGxsGro
while(valid)
{
RsGxsGrpMetaData* m = locked_getGrpMeta(*c, 0,true);
auto m = locked_getGrpMeta(*c, 0);
if(m)
if(m != nullptr)
grpMeta[m->mGroupId] = m;
valid = c->moveToNext();
}
}
void RsDataService::locked_retrieveMsgMetaList(RetroCursor *c, std::vector<const RsGxsMsgMetaData *>& msgMeta)
void RsDataService::locked_retrieveMsgMetaList(RetroCursor *c, std::vector<std::shared_ptr<RsGxsMsgMetaData> >& msgMeta)
{
if(!c)
{
@ -1357,17 +1357,18 @@ void RsDataService::locked_retrieveMsgMetaList(RetroCursor *c, std::vector<const
}
bool valid = c->moveToFirst();
while(valid){
const RsGxsMsgMetaData* m = locked_getMsgMeta(*c, 0,true);
while(valid)
{
auto m = locked_getMsgMeta(*c, 0);
if(m != NULL)
if(m != nullptr)
msgMeta.push_back(m);
valid = c->moveToNext();
}
}
int RsDataService::retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp)
int RsDataService::retrieveGxsGrpMetaData(std::map<RsGxsGroupId,std::shared_ptr<RsGxsGrpMetaData> >& grp)
{
#ifdef RS_DATA_SERVICE_DEBUG
std::cerr << "RsDataService::retrieveGxsGrpMetaData()";
@ -1384,7 +1385,7 @@ int RsDataService::retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp)
if(grp.empty())
{
if(mGrpMetaDataCache.isCacheUpToDate()) // grab all the stash from the cache, so as to avoid decryption costs.
if(mUseCache && mGrpMetaDataCache.isCacheUpToDate()) // grab all the stash from the cache, so as to avoid decryption costs.
{
#ifdef RS_DATA_SERVICE_DEBUG_CACHE
std::cerr << (void*)this << ": RsDataService::retrieveGxsGrpMetaData() retrieving all from cache!" << std::endl;
@ -1403,42 +1404,23 @@ int RsDataService::retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp)
if(c)
{
locked_retrieveGrpMetaList(c,grp);
locked_retrieveGrpMetaList(c,grp);
mGrpMetaDataCache.setCacheUpToDate(true);
if(mUseCache)
mGrpMetaDataCache.setCacheUpToDate(true);
}
delete c;
#ifdef RS_DATA_SERVICE_DEBUG_TIME
resultCount += grp.size();
#endif
// if(c)
// {
// bool valid = c->moveToFirst();
//
// while(valid)
// {
// RsGxsGrpMetaData* g = locked_getGrpMeta(*c, 0,true);
//
// if(g)
// {
// grp[g->mGroupId] = g;
//#ifdef RS_DATA_SERVICE_DEBUG_CACHE
// std::cerr << (void *)this << " " << mDbName << ": Retrieving (all) Grp metadata grpId=" << g->mGroupId << std::endl;
//#endif
// }
// valid = c->moveToNext();
//
// }
// delete c;
// }
}
}
else
{
for(auto mit(grp.begin()); mit != grp.end(); ++mit)
{
RsGxsGrpMetaData *meta = mGrpMetaDataCache.getMeta(mit->first) ;
auto meta = mUseCache?mGrpMetaDataCache.getMeta(mit->first): (std::shared_ptr<RsGxsGrpMetaData>()) ;
if(meta)
mit->second = meta;
@ -1452,7 +1434,8 @@ int RsDataService::retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp)
RetroCursor* c = mDb->sqlQuery(GRP_TABLE_NAME, mGrpMetaColumns, "grpId='" + grpId.toStdString() + "'", "");
c->moveToFirst();
RsGxsGrpMetaData* meta = locked_getGrpMeta(*c, 0,true);
auto meta = locked_getGrpMeta(*c, 0);
if(meta)
mit->second = meta;
@ -1463,33 +1446,6 @@ int RsDataService::retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp)
delete c;
// if(c)
// {
// bool valid = c->moveToFirst();
//
//#ifdef RS_DATA_SERVICE_DEBUG_CACHE
// if(!valid)
// std::cerr << " Empty query! GrpId " << grpId << " is not in database" << std::endl;
//#endif
// while(valid)
// {
// RsGxsGrpMetaData* g = locked_getGrpMeta(*c, 0,true);
//
// if(g)
// {
// grp[g->mGroupId] = g;
//#ifdef RS_DATA_SERVICE_DEBUG_CACHE
// std::cerr << ". Got it. Updating cache." << std::endl;
//#endif
// }
// valid = c->moveToNext();
//
//#ifdef RS_DATA_SERVICE_DEBUG_TIME
// ++resultCount;
//#endif
// }
// delete c;
// }
#ifdef RS_DATA_SERVICE_DEBUG_CACHE
else
std::cerr << ". not found!" << std::endl;
@ -1734,30 +1690,28 @@ void RsDataService::debug_printCacheSize()
{
RS_STACK_MUTEX(mDbMutex);
uint32_t nb_items,nb_items_on_deadlist;
uint64_t total_size,total_size_of_deadlist;
uint32_t nb_items;
uint64_t total_size;
mGrpMetaDataCache.debug_computeSize(nb_items, nb_items_on_deadlist, total_size,total_size_of_deadlist);
mGrpMetaDataCache.debug_computeSize(nb_items, total_size);
RsDbg() << "Cache size: " << std::endl;
RsDbg() << " Groups: " << " total: " << nb_items << " (dead: " << nb_items_on_deadlist << "), size: " << total_size << " (Dead: " << total_size_of_deadlist << ")" << std::endl;
RsDbg() << "[CACHE] Cache size: " << std::endl;
RsDbg() << "[CACHE] Groups: " << " total: " << nb_items << ", size: " << total_size << std::endl;
nb_items = 0,nb_items_on_deadlist = 0;
total_size = 0,total_size_of_deadlist = 0;
nb_items = 0;
total_size = 0;
for(auto& it:mMsgMetaDataCache)
{
uint32_t tmp_nb_items,tmp_nb_items_on_deadlist;
uint64_t tmp_total_size,tmp_total_size_of_deadlist;
uint32_t tmp_nb_items;
uint64_t tmp_total_size;
it.second.debug_computeSize(tmp_nb_items, tmp_nb_items_on_deadlist, tmp_total_size,tmp_total_size_of_deadlist);
it.second.debug_computeSize(tmp_nb_items, tmp_total_size);
nb_items += tmp_nb_items;
nb_items_on_deadlist += tmp_nb_items_on_deadlist;
total_size += tmp_total_size;
total_size_of_deadlist += tmp_total_size_of_deadlist;
}
RsDbg() << " Msgs: " << " total: " << nb_items << " (dead: " << nb_items_on_deadlist << "), size: " << total_size << " (Dead: " << total_size_of_deadlist << ")" << std::endl;
RsDbg() << "[CACHE] Msgs: " << " total: " << nb_items << ", size: " << total_size << std::endl;
}

128
libretroshare/src/gxs/rsdataservice.h
View File

@ -38,35 +38,29 @@ public:
template<class ID, class MetaDataClass> class t_MetaDataCache
{
public:
t_MetaDataCache() : mCache_ContainsAllMetas(false) {}
virtual ~t_MetaDataCache()
{
for(auto it: mMetas)
delete it.second;
for(auto it: mOldCachedItems)
delete it.second;
}
t_MetaDataCache()
: mCache_ContainsAllMetas(false)
{}
virtual ~t_MetaDataCache() = default;
bool isCacheUpToDate() const { return mCache_ContainsAllMetas ; }
void setCacheUpToDate(bool b) { mCache_ContainsAllMetas = b; }
void getFullMetaList(std::map<ID,MetaDataClass*>& mp) const { mp = mMetas ; }
void getFullMetaList(std::vector<const MetaDataClass*>& mp) const { for(auto& m:mMetas) mp.push_back(m.second) ; }
void getFullMetaList(std::map<ID,std::shared_ptr<MetaDataClass> >& mp) const { mp = mMetas ; }
void getFullMetaList(std::vector<std::shared_ptr<MetaDataClass> >& mp) const { for(auto& m:mMetas) mp.push_back(m.second) ; }
MetaDataClass *getMeta(const ID& id)
std::shared_ptr<MetaDataClass> getMeta(const ID& id)
{
auto itt = mMetas.find(id);
if(itt != mMetas.end())
return itt->second ;
else
return NULL;
return nullptr;
}
MetaDataClass *getOrCreateMeta(const ID& id)
std::shared_ptr<MetaDataClass> getOrCreateMeta(const ID& id)
{
MetaDataClass *meta = nullptr;
auto it = mMetas.find(id) ;
if(it != mMetas.end())
@ -74,28 +68,20 @@ public:
#ifdef RS_DATA_SERVICE_DEBUG
RsDbg() << __PRETTY_FUNCTION__ << ": getting group meta " << grpId << " from cache." << std::endl;
#endif
meta = it->second ;
return it->second ;
}
else
{
#ifdef RS_DATA_SERVICE_DEBUG
RsDbg() << __PRETTY_FUNCTION__ << ": group meta " << grpId << " not in cache. Loading it from DB..." << std::endl;
#endif
meta = new MetaDataClass();
mMetas[id] = meta ;
}
return meta;
return (mMetas[id] = std::make_shared<MetaDataClass>());
}
}
void updateMeta(const ID& id,const MetaDataClass& meta)
{
auto it = mMetas.find(id) ;
if(it != mMetas.end())
*(it->second) = meta ;
else
mMetas[id] = new MetaDataClass(meta) ;
mMetas[id] = std::make_shared<MetaDataClass>(meta); // create a new shared_ptr to possibly replace the previous one
}
void clear(const ID& id)
@ -114,39 +100,20 @@ public:
std::cerr << "(II) moving database cache entry " << (void*)(*it).second << " to dead list." << std::endl;
#endif
mOldCachedItems.push_back(std::make_pair(now,it->second)) ;
mMetas.erase(it) ;
mCache_ContainsAllMetas = false;
}
// We also take that opportunity to delete old entries.
auto it2(mOldCachedItems.begin());
while(it2!=mOldCachedItems.end() && (*it2).first + CACHE_ENTRY_GRACE_PERIOD < now)
{
#ifdef RS_DATA_SERVICE_DEBUG
std::cerr << "(II) deleting old GXS database cache entry " << (void*)(*it2).second << ", " << now - (*it2).first << " seconds old." << std::endl;
#endif
delete (*it2).second ;
it2 = mOldCachedItems.erase(it2) ;
}
}
void debug_computeSize(uint32_t& nb_items, uint32_t& nb_items_on_deadlist, uint64_t& total_size,uint64_t& total_size_of_deadlist) const
void debug_computeSize(uint32_t& nb_items, uint64_t& total_size) const
{
nb_items = mMetas.size();
nb_items_on_deadlist = mOldCachedItems.size();
total_size = 0;
total_size_of_deadlist = 0;
for(auto it:mMetas) total_size += it.second->serial_size();
for(auto it:mOldCachedItems) total_size_of_deadlist += it.second->serial_size();
}
private:
std::map<ID,MetaDataClass*> mMetas;
std::list<std::pair<rstime_t,MetaDataClass*> > mOldCachedItems ; // dead list, where items get deleted after being unused for a while. This is due to not using smart ptrs.
std::map<ID,std::shared_ptr<MetaDataClass> > mMetas;
static const uint32_t CACHE_ENTRY_GRACE_PERIOD = 600 ; // Unused items are deleted 10 minutes after last usage.
@ -165,40 +132,34 @@ public:
* Retrieves all msgs
* @param reqIds requested msg ids (grpId,msgId), leave msg list empty to get all msgs for the grp
* @param msg result of msg retrieval
* @param cache IGNORED whether to store results of this retrieval in memory
* for faster later retrieval
* @param strictFilter if true do not request any message if reqIds is empty
* @param withMeta true will also retrieve metadata
* @return error code
*/
int retrieveNxsMsgs(
const GxsMsgReq& reqIds, GxsMsgResult& msg, bool cache,
bool withMeta = false );
int retrieveNxsMsgs(const GxsMsgReq& reqIds, GxsMsgResult& msg, bool withMeta = false) override;
/*!
* Retrieves groups, if empty, retrieves all grps, if map is not empty
* only retrieve entries, if entry cannot be found, it is removed from map
* @param grp retrieved groups
* @param withMeta this initialise the metaData member of the nxsgroups retrieved
* @param cache whether to store retrieval in mem for faster later retrieval
* @return error code
*/
int retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp*>& grp, bool withMeta, bool cache);
int retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp*>& grp, bool withMeta) override;
/*!
* Retrieves meta data of all groups stored (most current versions only)
* @param cache whether to store retrieval in mem for faster later retrieval
* @param grp output group meta data
* @return error code
*/
int retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp);
int retrieveGxsGrpMetaData(std::map<RsGxsGroupId, std::shared_ptr<RsGxsGrpMetaData> > &grp) override;
/*!
* Retrieves meta data of all groups stored (most current versions only)
* @param grpIds grpIds for which to retrieve meta data
* @param msgMeta meta data result as map of grpIds to array of metadata for that grpId
* @param cache whether to store retrieval in mem for faster later retrieval
* @return error code
*/
int retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta);
int retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta) override;
/*!
* remove msgs in data store
@ -206,21 +167,21 @@ public:
* @param msgIds ids of messages to be removed
* @return error code
*/
int removeMsgs(const GxsMsgReq& msgIds);
int removeMsgs(const GxsMsgReq& msgIds) override;
/*!
* remove groups in data store listed in grpIds param
* @param grpIds ids of groups to be removed
* @return error code
*/
int removeGroups(const std::vector<RsGxsGroupId>& grpIds);
int removeGroups(const std::vector<RsGxsGroupId>& grpIds) override;
/*!
* Retrieves all group ids in store
* @param grpIds all grpids in store is inserted into this vector
* @return error code
*/
int retrieveGroupIds(std::vector<RsGxsGroupId> &grpIds);
int retrieveGroupIds(std::vector<RsGxsGroupId> &grpIds) override;
/*!
* Retrives all msg ids in store
@ -228,50 +189,57 @@ public:
* @param msgId msgsids retrieved
* @return error code
*/
int retrieveMsgIds(const RsGxsGroupId& grpId, RsGxsMessageId::std_set& msgId);
int retrieveMsgIds(const RsGxsGroupId& grpId, RsGxsMessageId::std_set& msgId) override;
/*!
* @return the cache size set for this RsGeneralDataService in bytes
*/
uint32_t cacheSize() const;
uint32_t cacheSize() const override;
/*!
* \brief serviceType
* \return
* The service type for the current data service.
*/
virtual uint16_t serviceType() const override { return mServType; }
/*!
* @param size size of cache to set in bytes
*/
int setCacheSize(uint32_t size);
int setCacheSize(uint32_t size) override;
/*!
* Stores a list of signed messages into data store
* @param msg map of message and decoded meta data information
* @return error code
*/
int storeMessage(const std::list<RsNxsMsg*>& msg);
int storeMessage(const std::list<RsNxsMsg*>& msg) override;
/*!
* Stores a list of groups in data store
* @param grp map of group and decoded meta data
* @return error code
*/
int storeGroup(const std::list<RsNxsGrp*>& grp);
int storeGroup(const std::list<RsNxsGrp*>& grp) override;
/*!
* Updates group entries in Db
* @param grp map of group and decoded meta data
* @return error code
*/
int updateGroup(const std::list<RsNxsGrp*>& grsp);
int updateGroup(const std::list<RsNxsGrp*>& grsp) override;
/*!
* @param metaData The meta data item to update
* @return error code
*/
int updateMessageMetaData(const MsgLocMetaData& metaData);
int updateMessageMetaData(const MsgLocMetaData& metaData) override;
/*!
* @param metaData The meta data item to update
* @return error code
*/
int updateGroupMetaData(const GrpLocMetaData &meta);
int updateGroupMetaData(const GrpLocMetaData &meta) override;
/*!
* Completely clear out data stored in
@ -279,10 +247,10 @@ public:
* as it was when first constructed
* @return error code
*/
int resetDataStore();
int resetDataStore() override;
bool validSize(RsNxsMsg* msg) const;
bool validSize(RsNxsGrp* grp) const;
bool validSize(RsNxsMsg* msg) const override;
bool validSize(RsNxsGrp* grp) const override;
/*!
* Convenience function used to only update group keys. This is used when sending
@ -290,7 +258,7 @@ public:
* @return SQL error code
*/
int updateGroupKeys(const RsGxsGroupId& grpId,const RsTlvSecurityKeySet& keys, uint32_t subscribe_flags) ;
int updateGroupKeys(const RsGxsGroupId& grpId,const RsTlvSecurityKeySet& keys, uint32_t subscribe_flags) override;
void debug_printCacheSize() ;
@ -316,26 +284,26 @@ private:
* @param c cursor to result set
* @param msgMeta message metadata retrieved from cursor are stored here
*/
void locked_retrieveMsgMetaList(RetroCursor* c, std::vector<const RsGxsMsgMetaData*>& msgMeta);
void locked_retrieveMsgMetaList(RetroCursor* c, std::vector<std::shared_ptr<RsGxsMsgMetaData> > &msgMeta);
/*!
* Retrieves all the grp meta results from a cursor
* @param c cursor to result set
* @param grpMeta group metadata retrieved from cursor are stored here
*/
void locked_retrieveGrpMetaList(RetroCursor *c, std::map<RsGxsGroupId,RsGxsGrpMetaData *>& grpMeta);
void locked_retrieveGrpMetaList(RetroCursor *c, std::map<RsGxsGroupId, std::shared_ptr<RsGxsGrpMetaData> > &grpMeta);
/*!
* extracts a msg meta item from a cursor at its
* current position
*/
RsGxsMsgMetaData* locked_getMsgMeta(RetroCursor& c, int colOffset, bool use_cache);
std::shared_ptr<RsGxsMsgMetaData> locked_getMsgMeta(RetroCursor& c, int colOffset);
/*!
* extracts a grp meta item from a cursor at its
* current position
*/
RsGxsGrpMetaData* locked_getGrpMeta(RetroCursor& c, int colOffset, bool use_cache);
std::shared_ptr<RsGxsGrpMetaData> locked_getGrpMeta(RetroCursor& c, int colOffset);
/*!
* extracts a msg item from a cursor at its
@ -477,6 +445,8 @@ private:
t_MetaDataCache<RsGxsGroupId,RsGxsGrpMetaData> mGrpMetaDataCache;
std::map<RsGxsGroupId,t_MetaDataCache<RsGxsMessageId,RsGxsMsgMetaData> > mMsgMetaDataCache;
bool mUseCache;
};
#endif // RSDATASERVICE_H

35
libretroshare/src/gxs/rsgds.h
View File

@ -19,8 +19,8 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>. *
* *
*******************************************************************************/
#ifndef RSGDS_H
#define RSGDS_H
#pragma once
#include <set>
#include <map>
@ -56,18 +56,16 @@ struct MsgLocMetaData {
ContentValue val;
};
typedef std::map<RsGxsGroupId,RsGxsGrpMetaData*> RsGxsGrpMetaTemporaryMap;
/*!
* This allows modification of local
* meta data items of a group
*/
struct GrpLocMetaData {
GrpLocMetaData() = default;
GrpLocMetaData(const GrpLocMetaData& meta): grpId(meta.grpId), val(meta.val) {}
GrpLocMetaData() = default;
GrpLocMetaData(const GrpLocMetaData& meta): grpId(meta.grpId), val(meta.val) {}
RsGxsGroupId grpId;
ContentValue val;
RsGxsGroupId grpId;
ContentValue val;
};
/*!
@ -144,9 +142,7 @@ public:
* @param strictFilter if true do not request any message if reqIds is empty
* @return error code
*/
virtual int retrieveNxsMsgs(
const GxsMsgReq& reqIds, GxsMsgResult& msg, bool cache,
bool withMeta = false ) = 0;
virtual int retrieveNxsMsgs( const GxsMsgReq& reqIds, GxsMsgResult& msg, bool withMeta = false ) = 0;
/*!
* Retrieves all groups stored. Caller owns the memory and is supposed to delete the RsNxsGrp pointers after use.
@ -155,7 +151,7 @@ public:
* @param cache whether to store retrieval in mem for faster later retrieval
* @return error code
*/
virtual int retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp*>& grp, bool withMeta, bool cache) = 0;
virtual int retrieveNxsGrps(std::map<RsGxsGroupId, RsNxsGrp*>& grp, bool withMeta) = 0;
/*!
* Retrieves meta data of all groups stored (most current versions only)
@ -166,13 +162,12 @@ public:
* , if grpId is failed to be retrieved it will be erased from map
* @return error code
*/
virtual int retrieveGxsGrpMetaData(RsGxsGrpMetaTemporaryMap& grp) = 0;
virtual int retrieveGxsGrpMetaData(std::map<RsGxsGroupId,std::shared_ptr<RsGxsGrpMetaData> >& grp) = 0;
/*!
* Retrieves meta data of all groups stored (most current versions only)
* @param grpIds grpIds for which to retrieve meta data
* @param msgMeta meta data result as map of grpIds to array of metadata for that grpId
* @param cache whether to store retrieval in mem for faster later retrieval
* @return error code
*/
virtual int retrieveGxsMsgMetaData(const GxsMsgReq& msgIds, GxsMsgMetaResult& msgMeta) = 0;
@ -211,6 +206,13 @@ public:
*/
virtual uint32_t cacheSize() const = 0;
/*!
* \brief serviceType
* \return
* The service type for the current data service.
*/
virtual uint16_t serviceType() const = 0 ;
/*!
* @param size size of cache to set in bytes
*/
@ -274,8 +276,3 @@ public:
virtual bool validSize(RsNxsGrp* grp) const = 0 ;
};
#endif // RSGDS_H

146
libretroshare/src/gxs/rsgenexchange.cc
View File

@ -59,6 +59,9 @@ static const uint32_t INDEX_AUTHEN_IDENTITY = 0x00000010; // identity
static const uint32_t INDEX_AUTHEN_PUBLISH = 0x00000020; // publish key
static const uint32_t INDEX_AUTHEN_ADMIN = 0x00000040; // admin key
static const uint32_t MSG_CLEANUP_PERIOD = 60*59; // 59 minutes
static const uint32_t INTEGRITY_CHECK_PERIOD = 60*31; // 31 minutes
#define GXS_MASK "GXS_MASK_HACK"
/*
@ -132,9 +135,6 @@ static const uint32_t INDEX_AUTHEN_ADMIN = 0x00000040; // admin key
// +--- processRoutingClues() ;
//
static const uint32_t MSG_CLEANUP_PERIOD = 60*59; // 59 minutes
static const uint32_t INTEGRITY_CHECK_PERIOD = 60*31; // 31 minutes
RsGenExchange::RsGenExchange(
RsGeneralDataService* gds, RsNetworkExchangeService* ns,
RsSerialType* serviceSerialiser, uint16_t servType, RsGixs* gixs,
@ -162,6 +162,47 @@ RsGenExchange::RsGenExchange(
VALIDATE_MAX_WAITING_TIME(60)
{
mDataAccess = new RsGxsDataAccess(gds);
// Perform an early checking/cleaning of the db. Will eliminate groups and messages that do not match their hash
#ifdef RS_DEEP_CHANNEL_INDEX
// This code is only called because it of deep indexing in channels. But loading
// the entire set of messages in order to provide indexing is pretty bad (very costly and slowly
// eats memory, as many tests have shown. Not because of leaks, but because new threads are
// apparently attributed large stacks and pages of memory are not regained by the system maybe because it thinks
// that RS will use them again.
//
// * the deep check should be implemented differently, in an incremental way. For instance in notifyChanges() of each
// service (e.g. channels here) should update the index when a new message is received. The question to how old messages
// are processed is open. I believe that they shouldn't. New users will progressively process them.
//
// * integrity check (re-hashing of message data) is not needed. Message signature already ensures that all messages received are
// unalterated. The only problem (possibly very rare) is that a message is locally corrupted and not deleted (because of no check).
// It will therefore never be replaced by the correct one from friends. The cost of re-hashing the whole db data regularly
// doesn't counterbalance such a low risk.
//
if(mServType == RS_SERVICE_GXS_TYPE_CHANNELS)
{
std::vector<RsGxsGroupId> grpsToDel;
GxsMsgReq msgsToDel;
RsGxsSinglePassIntegrityCheck::check(mServType,mGixs,mDataStore,
this, mSerialiser,
grpsToDel,msgsToDel);
for(auto& grpId: grpsToDel)
{
uint32_t token2=0;
deleteGroup(token2,grpId);
}
if(!msgsToDel.empty())
{
uint32_t token1=0;
deleteMsgs(token1,msgsToDel);
}
}
#endif
}
void RsGenExchange::setNetworkExchangeService(RsNetworkExchangeService *ns)
@ -169,9 +210,7 @@ void RsGenExchange::setNetworkExchangeService(RsNetworkExchangeService *ns)
if(mNetService != NULL)
std::cerr << "(EE) Cannot override existing network exchange service. Make sure it has been deleted otherwise." << std::endl;
else
{
mNetService = ns ;
}
}
RsGenExchange::~RsGenExchange()
@ -809,12 +848,13 @@ int RsGenExchange::createMessage(RsNxsMsg* msg)
else
{
// get publish key
const RsGxsGrpMetaData* grpMeta = metaMap[id];
const auto& grpMeta = metaMap[id];
uint32_t metaDataLen = meta.serial_size();
uint32_t allMsgDataLen = metaDataLen + msg->msg.bin_len;
char* metaData = new char[metaDataLen];
char* allMsgData = new char[allMsgDataLen]; // msgData + metaData
RsTemporaryMemory metaData(metaDataLen);
RsTemporaryMemory allMsgData(allMsgDataLen);
meta.serialise(metaData, &metaDataLen);
@ -839,9 +879,6 @@ int RsGenExchange::createMessage(RsNxsMsg* msg)
// assign msg id to msg meta
msg->metaData->mMsgId = msg->msgId;
delete[] metaData;
delete[] allMsgData;
}
if(ret_val == SIGN_FAIL)
@ -1299,16 +1336,14 @@ bool RsGenExchange::getPublishedGroupMeta(const uint32_t& token,RsGroupMetaData&
bool RsGenExchange::getGroupMeta(const uint32_t &token, std::list<RsGroupMetaData> &groupInfo)
{
std::list<const RsGxsGrpMetaData*> metaL;
std::list<std::shared_ptr<RsGxsGrpMetaData> > metaL;
bool ok = mDataAccess->getGroupSummary(token, metaL);
RsGroupMetaData m;
for( auto lit = metaL.begin(); lit != metaL.end(); ++lit)
{
const RsGxsGrpMetaData& gMeta = *(*lit);
m = gMeta;
RsGroupMetaData m(gMeta);
RsGroupNetworkStats sts ;
if(mNetService != NULL && mNetService->getGroupNetworkStats(gMeta.mGroupId,sts))
@ -1341,24 +1376,15 @@ bool RsGenExchange::getMsgMeta(const uint32_t &token,
GxsMsgMetaResult result;
bool ok = mDataAccess->getMsgSummary(token, result);
GxsMsgMetaResult::iterator mit = result.begin();
for(; mit != result.end(); ++mit)
for(auto mit=result.begin(); mit != result.end(); ++mit)
{
std::vector<const RsGxsMsgMetaData*>& metaV = mit->second;
auto& metaV = mit->second;
std::vector<RsMsgMetaData>& msgInfoV = msgInfo[mit->first];
std::vector<const RsGxsMsgMetaData*>::iterator vit = metaV.begin();
RsMsgMetaData meta;
for(; vit != metaV.end(); ++vit)
{
const RsGxsMsgMetaData& m = *(*vit);
meta = m;
msgInfoV.push_back(meta);
//delete *vit;
}
metaV.clear();
for(auto vit=metaV.begin(); vit != metaV.end(); ++vit)
msgInfoV.push_back(RsMsgMetaData(*(*vit)));
}
return ok;
@ -1373,20 +1399,12 @@ bool RsGenExchange::getMsgRelatedMeta(const uint32_t &token, GxsMsgRelatedMetaMa
for(; mit != result.end(); ++mit)
{
std::vector<const RsGxsMsgMetaData*>& metaV = mit->second;
auto& metaV = mit->second;
std::vector<RsMsgMetaData>& msgInfoV = msgMeta[mit->first];
std::vector<const RsGxsMsgMetaData*>::iterator vit = metaV.begin();
RsMsgMetaData meta;
for(; vit != metaV.end(); ++vit)
{
const RsGxsMsgMetaData& m = *(*vit);
meta = m;
msgInfoV.push_back(meta);
//delete *vit;
}
metaV.clear();
for(auto vit=metaV.begin(); vit != metaV.end(); ++vit)
msgInfoV.push_back(RsMsgMetaData(*(*vit)));
}
return ok;
@ -1435,7 +1453,7 @@ bool RsGenExchange::retrieveNxsIdentity(const RsGxsGroupId& group_id,RsNxsGrp *&
grp[group_id]=nullptr;
std::map<RsGxsGroupId, RsNxsGrp*>::const_iterator grp_it;
if(! mDataStore->retrieveNxsGrps(grp, true,true) || grp.end()==(grp_it=grp.find(group_id)) || !grp_it->second)
if(! mDataStore->retrieveNxsGrps(grp, true) || grp.end()==(grp_it=grp.find(group_id)) || !grp_it->second)
{
std::cerr << "(EE) Cannot retrieve group data for group " << group_id << " in service " << mServType << std::endl;
return false;
@ -1620,8 +1638,8 @@ bool RsGenExchange::getMsgRelatedData( uint32_t token,
const RsGxsGrpMsgIdPair& msgId = mit->first;
std::vector<RsGxsMsgItem*> &gxsMsgItems = msgItems[msgId];
std::vector<RsNxsMsg*>& nxsMsgsV = mit->second;
std::vector<RsNxsMsg*>::iterator vit = nxsMsgsV.begin();
for(; vit != nxsMsgsV.end(); ++vit)
for(auto vit=nxsMsgsV.begin(); vit != nxsMsgsV.end(); ++vit)
{
RsNxsMsg*& msg = *vit;
RsItem* item = NULL;
@ -2107,15 +2125,14 @@ void RsGenExchange::processMsgMetaChanges()
if(mit != result.end())
{
std::vector<const RsGxsMsgMetaData*>& msgMetaV = mit->second;
const auto& msgMetaV = mit->second;
if(!msgMetaV.empty())
{
const RsGxsMsgMetaData* meta = *(msgMetaV.begin());
const auto& meta = *(msgMetaV.begin());
value = (meta->mMsgStatus & ~mask) | (mask & value);
changed = (static_cast<int64_t>(meta->mMsgStatus) != value);
m.val.put(RsGeneralDataService::MSG_META_STATUS, value);
//delete meta;
ok = true;
}
}
@ -2210,16 +2227,16 @@ bool RsGenExchange::processGrpMask(const RsGxsGroupId& grpId, ContentValue &grpC
// first find out which mask is involved
int32_t value, mask, currValue;
std::string key;
const RsGxsGrpMetaData* grpMeta = NULL;
std::shared_ptr<RsGxsGrpMetaData> grpMeta;
bool ok = false;
RsGxsGrpMetaTemporaryMap grpMetaMap;
std::map<RsGxsGroupId, RsGxsGrpMetaData* >::iterator mit;
grpMetaMap.insert(std::make_pair(grpId, (RsGxsGrpMetaData*)(NULL)));
mDataStore->retrieveGxsGrpMetaData(grpMetaMap);
auto mit = grpMetaMap.find(grpId);
if((mit = grpMetaMap.find(grpId)) != grpMetaMap.end())
if(mit != grpMetaMap.end())
{
grpMeta = mit->second;
@ -2402,15 +2419,15 @@ void RsGenExchange::publishMsgs()
if(rsPeers)
mRoutingClues[msg->metaData->mAuthorId].insert(rsPeers->getOwnId()) ;
computeHash(msg->msg, msg->metaData->mHash);
mDataAccess->addMsgData(msg);
computeHash(msg->msg, msg->metaData->mHash); // (csoler) weird choice: hash should also depend on metadata. Fortunately, msg signature
// signs the holw thing (msg + meta)
mPublishedMsgs[token] = *msg->metaData;
RsGxsMsgItem *msg_item = dynamic_cast<RsGxsMsgItem*>(mSerialiser->deserialise(msg->msg.bin_data,&msg->msg.bin_len)) ;
msg_item->meta = *msg->metaData;
delete msg ;
mDataAccess->addMsgData(msg); // msg is deleted by addMsgData()
msgChangeMap[grpId].push_back(msg_item);
@ -2501,10 +2518,11 @@ void RsGenExchange::processGroupUpdatePublish()
{
GroupUpdatePublish& gup = *vit;
const RsGxsGroupId& groupId = gup.grpItem->meta.mGroupId;
std::map<RsGxsGroupId, RsGxsGrpMetaData*>::iterator mit = grpMeta.find(groupId);
auto mit = grpMeta.find(groupId);
const RsGxsGrpMetaData* meta = NULL;
if(mit == grpMeta.end() || mit->second == NULL)
std::shared_ptr<RsGxsGrpMetaData> meta;
if(mit == grpMeta.end() || mit->second == nullptr)
{
std::cerr << "Error! could not find meta of old group to update!" << std::endl;
mDataAccess->updatePublicRequestStatus(gup.mToken, RsTokenService::FAILED);
@ -2823,7 +2841,7 @@ void RsGenExchange::publishGrps()
RsNxsGrpDataTemporaryMap oldGrpDatas;
oldGrpDatas.insert(std::make_pair(grpId, (RsNxsGrp*)NULL));
if(mDataStore->retrieveNxsGrps(oldGrpDatas,true,false) && oldGrpDatas.size() == 1)
if(mDataStore->retrieveNxsGrps(oldGrpDatas,true) && oldGrpDatas.size() == 1)
{
auto oldGrp = oldGrpDatas[grpId];
c->mOldGroupItem = dynamic_cast<RsGxsGrpItem*>(mSerialiser->deserialise(oldGrp->grp.bin_data,&oldGrp->grp.bin_len));
@ -2840,7 +2858,6 @@ void RsGenExchange::publishGrps()
else
mDataAccess->addGroupData(grp);
delete grp ;
groups_to_subscribe.push_back(grpId) ;
}
else
@ -2960,7 +2977,7 @@ bool RsGenExchange::getGroupKeys(const RsGxsGroupId &grpId, RsTlvSecurityKeySet
if(grpMeta.empty())
return false;
const RsGxsGrpMetaData* meta = grpMeta[grpId];
const auto& meta = grpMeta[grpId];
if(meta == NULL)
return false;
@ -3059,7 +3076,7 @@ void RsGenExchange::processRecvdMessages()
mDataStore->retrieveGxsGrpMetaData(grpMetas);
GxsMsgReq msgIds;
RsNxsMsgDataTemporaryList msgs_to_store;
std::list<RsNxsMsg*> msgs_to_store;
#ifdef GEN_EXCH_DEBUG
std::cerr << " updating received messages:" << std::endl;
@ -3086,7 +3103,7 @@ void RsGenExchange::processRecvdMessages()
#ifdef GEN_EXCH_DEBUG
std::cerr << " deserialised info: grp id=" << msg->grpId << ", msg id=" << msg->msgId ;
#endif
std::map<RsGxsGroupId, RsGxsGrpMetaData*>::iterator mit = grpMetas.find(msg->grpId);
auto mit = grpMetas.find(msg->grpId);
#ifdef GEN_EXCH_DEBUG
std::cerr << " msg info : grp id=" << msg->grpId << ", msg id=" << msg->msgId << std::endl;
@ -3100,7 +3117,7 @@ void RsGenExchange::processRecvdMessages()
continue ;
}
const RsGxsGrpMetaData *grpMeta = mit->second;
const auto& grpMeta = mit->second;
RsTlvSecurityKeySet keys = grpMeta->keys ;
GxsSecurity::createPublicKeysFromPrivateKeys(keys); // make sure we have the public keys that correspond to the private ones, as it happens. Most of the time this call does nothing.
@ -3211,7 +3228,7 @@ void RsGenExchange::processRecvdGroups()
std::cerr << "RsGenExchange::Processing received groups" << std::endl;
#endif
std::list<RsGxsGroupId> grpIds;
RsNxsGrpDataTemporaryList grps_to_store;
std::list<RsNxsGrp*> grps_to_store;
// 1 - retrieve the existing groups so as to check what's not new
std::vector<RsGxsGroupId> existingGrpIds;
@ -3326,7 +3343,7 @@ void RsGenExchange::processRecvdGroups()
mNotifications.push_back(c);
}
mDataStore->storeGroup(grps_to_store);
mDataStore->storeGroup(grps_to_store); // deletes the data after storing it.
#ifdef GEN_EXCH_DEBUG
std::cerr << " adding the following grp ids to notification: " << std::endl;
for(std::list<RsGxsGroupId>::const_iterator it(grpIds.begin());it!=grpIds.end();++it)
@ -3351,7 +3368,7 @@ void RsGenExchange::performUpdateValidation()
for(auto vit(mGroupUpdates.begin()); vit != mGroupUpdates.end(); ++vit)
grpDatas.insert(std::make_pair(vit->newGrp->grpId, (RsNxsGrp*)NULL));
if(grpDatas.empty() || !mDataStore->retrieveNxsGrps(grpDatas,true,false))
if(grpDatas.empty() || !mDataStore->retrieveNxsGrps(grpDatas,true))
{
if(grpDatas.empty())
RsErr() << __PRETTY_FUNCTION__ << " Validation of multiple group updates failed: no group in list!" << std::endl;
@ -3364,8 +3381,7 @@ void RsGenExchange::performUpdateValidation()
#ifdef GEN_EXCH_DEBUG
std::cerr << "RsGenExchange::performUpdateValidation() " << std::endl;
#endif
RsNxsGrpDataTemporaryList grps ;
std::list<RsNxsGrp*> grps; // dont use RsNxsGrpDataTemporaryList because updateGrps will delete the groups
for(auto vit(mGroupUpdates.begin()); vit != mGroupUpdates.end(); ++vit)
{

4
libretroshare/src/gxs/rsgxs.h
View File

@ -29,8 +29,8 @@
/* data types used throughout Gxs from netservice to genexchange */
typedef std::map<RsGxsGroupId, std::vector<const RsGxsMsgMetaData*> > GxsMsgMetaResult;
typedef std::map<RsGxsGrpMsgIdPair, std::vector<const RsGxsMsgMetaData*> > MsgRelatedMetaResult;
typedef std::map<RsGxsGroupId, std::vector<std::shared_ptr<RsGxsMsgMetaData> > > GxsMsgMetaResult;
typedef std::map<RsGxsGrpMsgIdPair, std::vector<std::shared_ptr<RsGxsMsgMetaData> > > MsgRelatedMetaResult;
// Default values that are used throughout GXS code

194
libretroshare/src/gxs/rsgxsdataaccess.cc
View File

@ -339,7 +339,7 @@ bool RsGxsDataAccess::cancelRequest(const uint32_t& token)
{
RsStackMutex stack(mDataMutex); /****** LOCKED *****/
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if (!req)
{
return false;
@ -373,11 +373,11 @@ bool RsGxsDataAccess::locked_clearRequest(const uint32_t& token)
return true;
}
bool RsGxsDataAccess::getGroupSummary(const uint32_t& token, std::list<const RsGxsGrpMetaData*>& groupInfo)
bool RsGxsDataAccess::getGroupSummary(const uint32_t& token, std::list<std::shared_ptr<RsGxsGrpMetaData> >& groupInfo)
{
RS_STACK_MUTEX(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -406,7 +406,7 @@ bool RsGxsDataAccess::getGroupData(const uint32_t& token, std::list<RsNxsGrp*>&
{
RS_STACK_MUTEX(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -442,7 +442,7 @@ bool RsGxsDataAccess::getMsgData(const uint32_t& token, NxsMsgDataResult& msgDat
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -468,12 +468,12 @@ bool RsGxsDataAccess::getMsgData(const uint32_t& token, NxsMsgDataResult& msgDat
return true;
}
bool RsGxsDataAccess::getMsgRelatedData(const uint32_t &token, NxsMsgRelatedDataResult &msgData)
bool RsGxsDataAccess::getMsgRelatedData(const uint32_t& token, NxsMsgRelatedDataResult& msgData)
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -506,7 +506,7 @@ bool RsGxsDataAccess::getMsgSummary(const uint32_t& token, GxsMsgMetaResult& msg
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -533,7 +533,7 @@ bool RsGxsDataAccess::getMsgRelatedSummary(const uint32_t &token, MsgRelatedMeta
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -564,7 +564,7 @@ bool RsGxsDataAccess::getMsgRelatedList(const uint32_t &token, MsgRelatedIdResul
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -594,7 +594,7 @@ bool RsGxsDataAccess::getMsgIdList(const uint32_t& token, GxsMsgIdResult& msgIds
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -621,7 +621,7 @@ bool RsGxsDataAccess::getGroupList(const uint32_t& token, std::list<RsGxsGroupId
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -648,7 +648,7 @@ bool RsGxsDataAccess::getGroupStatistic(const uint32_t &token, GxsGroupStatistic
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -672,7 +672,7 @@ bool RsGxsDataAccess::getServiceStatistic(const uint32_t &token, GxsServiceStati
{
RsStackMutex stack(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req == nullptr)
{
@ -691,7 +691,7 @@ bool RsGxsDataAccess::getServiceStatistic(const uint32_t &token, GxsServiceStati
locked_clearRequest(token);
return true;
}
GxsRequest* RsGxsDataAccess::locked_retrieveCompetedRequest(const uint32_t& token)
GxsRequest* RsGxsDataAccess::locked_retrieveCompletedRequest(const uint32_t& token)
{
auto it = mCompletedRequests.find(token) ;
@ -860,7 +860,7 @@ bool RsGxsDataAccess::getGroupSerializedData(GroupSerializedDataReq* req)
for(std::list<RsGxsGroupId>::iterator lit = grpIdsOut.begin();lit != grpIdsOut.end();++lit)
grpData[*lit] = nullptr;
bool ok = mDataStore->retrieveNxsGrps(grpData, true, true);
bool ok = mDataStore->retrieveNxsGrps(grpData, true);
req->mGroupData.clear();
std::map<RsGxsGroupId, RsNxsGrp*>::iterator mit = grpData.begin();
@ -888,7 +888,7 @@ bool RsGxsDataAccess::getGroupData(GroupDataReq* req)
grpData[*lit] = nullptr;
}
bool ok = mDataStore->retrieveNxsGrps(grpData, true, true);
bool ok = mDataStore->retrieveNxsGrps(grpData, true);
std::map<RsGxsGroupId, RsNxsGrp*>::iterator mit = grpData.begin();
for(; mit != grpData.end(); ++mit)
@ -908,23 +908,21 @@ bool RsGxsDataAccess::getGroupSummary(GroupMetaReq* req)
if(grpIdsOut.empty())
return true;
std::list<RsGxsGroupId>::const_iterator lit = grpIdsOut.begin();
for(; lit != grpIdsOut.end(); ++lit)
for(auto lit = grpIdsOut.begin();lit != grpIdsOut.end(); ++lit)
grpMeta[*lit] = nullptr;
mDataStore->retrieveGxsGrpMetaData(grpMeta);
mDataStore->retrieveGxsGrpMetaData(grpMeta);
std::map<RsGxsGroupId, RsGxsGrpMetaData*>::iterator mit = grpMeta.begin();
for(; mit != grpMeta.end(); ++mit)
for(auto mit = grpMeta.begin(); mit != grpMeta.end(); ++mit)
req->mGroupMetaData.push_back(mit->second);
return true;
}
bool RsGxsDataAccess::getGroupList(GroupIdReq* req)
{ return getGroupList(req->mGroupIds, req->Options, req->mGroupIdResult); }
{
return getGroupList(req->mGroupIds, req->Options, req->mGroupIdResult);
}
bool RsGxsDataAccess::getGroupList(const std::list<RsGxsGroupId>& grpIdsIn, const RsTokReqOptions& opts, std::list<RsGxsGroupId>& grpIdsOut)
{
@ -956,7 +954,7 @@ bool RsGxsDataAccess::getMsgData(MsgDataReq* req)
if((opts.mMsgFlagMask || opts.mStatusMask) && msgIdOut.empty())
return true;
mDataStore->retrieveNxsMsgs(msgIdOut, req->mMsgData, true, true);
mDataStore->retrieveNxsMsgs(msgIdOut, req->mMsgData, true);
return true;
}
@ -1030,7 +1028,7 @@ bool RsGxsDataAccess::getMsgMetaDataList( const GxsMsgReq& msgIds, const RsTokRe
//auto& filter( metaFilter[grpId] ); // does the initialization of metaFilter[grpId] and avoids further O(log(n)) calls
std::vector<const RsGxsMsgMetaData*>& metaV = meta_it->second;
auto& metaV = meta_it->second;
if (onlyLatestMsgs) // if we only consider latest messages, we need to first filter out messages with "children"
{
@ -1124,7 +1122,7 @@ bool RsGxsDataAccess::getMsgMetaDataList( const GxsMsgReq& msgIds, const RsTokRe
for(uint32_t i=0;i<metaV.size();++i)
if(metaV[i] != nullptr)
{
const RsGxsMsgMetaData* msgMeta = metaV[i];
const auto& msgMeta = metaV[i];
bool add = false;
/* if we are grabbing thread Head... then parentId == empty. */
@ -1291,8 +1289,7 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
GxsMsgReq msgIds;
msgIds.insert(std::make_pair(grpMsgIdPair.first, std::set<RsGxsMessageId>()));
mDataStore->retrieveGxsMsgMetaData(msgIds, result);
std::vector<const RsGxsMsgMetaData*>& metaV = result[grpMsgIdPair.first];
std::vector<const RsGxsMsgMetaData*>::iterator vit_meta;
auto& metaV = result[grpMsgIdPair.first];
// msg id to relate to
const RsGxsMessageId& msgId = grpMsgIdPair.second;
@ -1300,18 +1297,14 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
std::set<RsGxsMessageId> outMsgIds;
const RsGxsMsgMetaData* origMeta = nullptr;
std::shared_ptr<RsGxsMsgMetaData> origMeta;
for(vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
{
const RsGxsMsgMetaData* meta = *vit_meta;
if(msgId == meta->mMsgId)
for(auto vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
if(msgId == (*vit_meta)->mMsgId)
{
origMeta = meta;
origMeta = *vit_meta;
break;
}
}
if(!origMeta)
{
@ -1320,7 +1313,7 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
}
const RsGxsMessageId& origMsgId = origMeta->mOrigMsgId;
std::map<RsGxsMessageId, const RsGxsMsgMetaData*>& metaMap = filterMap[grpId];
auto& metaMap = filterMap[grpId];
if (onlyLatestMsgs)
{
@ -1329,10 +1322,10 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
// RUN THROUGH ALL MSGS... in map origId -> TS.
std::map<RsGxsMessageId, std::pair<RsGxsMessageId, rstime_t> > origMsgTs;
std::map<RsGxsMessageId, std::pair<RsGxsMessageId, rstime_t> >::iterator oit;
for(vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
{
const RsGxsMsgMetaData* meta = *vit_meta;
for(auto vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
{
auto meta = *vit_meta;
// skip msgs that aren't children.
if (onlyChildMsgs)
@ -1400,38 +1393,31 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
/* first guess is potentially better than Orig (can't be worse!) */
rstime_t latestTs = 0;
RsGxsMessageId latestMsgId;
const RsGxsMsgMetaData* latestMeta=nullptr;
std::shared_ptr<RsGxsMsgMetaData> latestMeta;
for(vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
{
const RsGxsMsgMetaData* meta = *vit_meta;
if (meta->mOrigMsgId == origMsgId)
for(auto vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
if ((*vit_meta)->mOrigMsgId == origMsgId)
{
if (meta->mPublishTs > latestTs)
if ((*vit_meta)->mPublishTs > latestTs)
{
latestTs = meta->mPublishTs;
latestMsgId = meta->mMsgId;
latestMeta = meta;
latestTs = (*vit_meta)->mPublishTs;
latestMsgId = (*vit_meta)->mMsgId;
latestMeta = (*vit_meta);
}
}
}
outMsgIds.insert(latestMsgId);
metaMap.insert(std::make_pair(latestMsgId, latestMeta));
}
}
else if (onlyAllVersions)
{
for(vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
{
const RsGxsMsgMetaData* meta = *vit_meta;
if (meta->mOrigMsgId == origMsgId)
for(auto vit_meta = metaV.begin(); vit_meta != metaV.end(); ++vit_meta)
if ((*vit_meta)->mOrigMsgId == origMsgId)
{
outMsgIds.insert(meta->mMsgId);
metaMap.insert(std::make_pair(meta->mMsgId, meta));
outMsgIds.insert((*vit_meta)->mMsgId);
metaMap.insert(std::make_pair((*vit_meta)->mMsgId, (*vit_meta)));
}
}
}
GxsMsgIdResult filteredOutMsgIds;
@ -1453,7 +1439,7 @@ bool RsGxsDataAccess::getMsgRelatedInfo(MsgRelatedInfoReq *req)
else if(req->Options.mReqType == GXS_REQUEST_TYPE_MSG_RELATED_DATA)
{
GxsMsgResult msgResult;
mDataStore->retrieveNxsMsgs(filteredOutMsgIds, msgResult, false, true);
mDataStore->retrieveNxsMsgs(filteredOutMsgIds, msgResult, true);
req->mMsgDataResult[grpMsgIdPair] = msgResult[grpId];
}
}
@ -1495,7 +1481,7 @@ bool RsGxsDataAccess::getGroupStatistic(GroupStatisticRequest *req)
for(uint32_t i = 0; i < msgMetaV.size(); ++i)
{
const RsGxsMsgMetaData* m = msgMetaV[i];
const auto& m = msgMetaV[i];
req->mGroupStatistic.mTotalSizeOfMsgs += m->mMsgSize + m->serial_size();
if(obsolete_msgs.find(m->mMsgId) != obsolete_msgs.end()) // skip obsolete messages.
@ -1504,24 +1490,19 @@ bool RsGxsDataAccess::getGroupStatistic(GroupStatisticRequest *req)
if (IS_MSG_NEW(m->mMsgStatus))
{
if (m->mParentId.isNull())
{
++req->mGroupStatistic.mNumThreadMsgsNew;
} else {
else
++req->mGroupStatistic.mNumChildMsgsNew;
}
}
if (IS_MSG_UNREAD(m->mMsgStatus))
{
if (m->mParentId.isNull())
{
++req->mGroupStatistic.mNumThreadMsgsUnread;
} else {
else
++req->mGroupStatistic.mNumChildMsgsUnread;
}
}
}
//cleanseMsgMetaMap(metaResult);
return true;
}
@ -1544,7 +1525,7 @@ bool RsGxsDataAccess::getServiceStatistic(ServiceStatisticRequest *req)
for(auto mit = grpMeta.begin(); mit != grpMeta.end(); ++mit)
{
const RsGxsGrpMetaData* m = mit->second;
const auto& m = mit->second;
req->mServiceStatistic.mSizeOfGrps += m->mGrpSize + m->serial_size(RS_GXS_GRP_META_DATA_CURRENT_API_VERSION);
if (IS_GROUP_SUBSCRIBED(m->mSubscribeFlags))
@ -1573,23 +1554,15 @@ bool RsGxsDataAccess::getMsgIdList(MsgIdReq* req)
{
GxsMsgMetaResult result;
mDataStore->retrieveGxsMsgMetaData(req->mMsgIds, result);
GxsMsgMetaResult::iterator mit = result.begin(), mit_end = result.end();
for(; mit != mit_end; ++mit)
for(auto mit = result.begin(); mit != result.end(); ++mit)
{
const RsGxsGroupId grpId = mit->first;
std::vector<const RsGxsMsgMetaData*>& metaV = mit->second;
std::vector<const RsGxsMsgMetaData*>::iterator vit = metaV.begin(),
vit_end = metaV.end();
auto& metaV = mit->second;
for(; vit != vit_end; ++vit)
{
const RsGxsMsgMetaData* meta = *vit;
req->mMsgIdResult[grpId].insert(meta->mMsgId);
}
for(auto vit=metaV.begin(); vit != metaV.end(); ++vit)
req->mMsgIdResult[grpId].insert((*vit)->mMsgId);
}
GxsMsgReq msgIdOut;
@ -1601,25 +1574,6 @@ bool RsGxsDataAccess::getMsgIdList(MsgIdReq* req)
return true;
}
// void RsGxsDataAccess::cleanseMsgMetaMap(GxsMsgMetaResult& result)
// {
// GxsMsgMetaResult::iterator mit = result.begin();
//
// for(; mit !=result.end(); ++mit)
// {
//
// std::vector<RsGxsMsgMetaData*>& msgMetaV = mit->second;
// std::vector<RsGxsMsgMetaData*>::iterator vit = msgMetaV.begin();
// for(; vit != msgMetaV.end(); ++vit)
// {
// delete *vit;
// }
// }
//
// result.clear();
// return;
// }
void RsGxsDataAccess::filterMsgIdList( GxsMsgIdResult& resultsMap, const RsTokReqOptions& opts, const MsgMetaFilter& msgMetas ) const
{
for( GxsMsgIdResult::iterator grpIt = resultsMap.begin(); grpIt != resultsMap.end(); ++grpIt )
@ -1638,16 +1592,13 @@ void RsGxsDataAccess::filterMsgIdList( GxsMsgIdResult& resultsMap, const RsTokRe
for( std::set<RsGxsMessageId>::iterator msgIdIt = msgsIdSet.begin(); msgIdIt != msgsIdSet.end(); )
{
const RsGxsMessageId& msgId(*msgIdIt);
const std::map<RsGxsMessageId, const RsGxsMsgMetaData*>& msgsMetaMap =
cit->second;
const auto& msgsMetaMap = cit->second;
bool keep = false;
std::map<RsGxsMessageId, const RsGxsMsgMetaData*>::const_iterator msgsMetaMapIt;
auto msgsMetaMapIt = msgsMetaMap.find(msgId);
if( (msgsMetaMapIt = msgsMetaMap.find(msgId)) != msgsMetaMap.end() )
{
if( msgsMetaMapIt != msgsMetaMap.end() )
keep = checkMsgFilter(opts, msgsMetaMapIt->second);
}
if(keep)
++msgIdIt;
@ -1663,29 +1614,21 @@ void RsGxsDataAccess::filterMsgIdList( GxsMsgIdResult& resultsMap, const RsTokRe
}
}
void RsGxsDataAccess::filterGrpList(std::list<RsGxsGroupId> &grpIds, const RsTokReqOptions &opts, const GrpMetaFilter &meta) const
void RsGxsDataAccess::filterGrpList(std::list<RsGxsGroupId> &grpIds, const RsTokReqOptions &opts, const GrpMetaFilter& meta) const
{
std::list<RsGxsGroupId>::iterator lit = grpIds.begin();
for(; lit != grpIds.end(); )
for(auto lit = grpIds.begin(); lit != grpIds.end(); )
{
GrpMetaFilter::const_iterator cit = meta.find(*lit);
auto cit = meta.find(*lit);
bool keep = false;
if(cit != meta.end())
{
keep = checkGrpFilter(opts, cit->second);
}
if(keep)
{
++lit;
}else
{
else
lit = grpIds.erase(lit);
}
}
}
@ -1694,7 +1637,7 @@ bool RsGxsDataAccess::checkRequestStatus( uint32_t token, GxsRequestStatus& stat
{
RS_STACK_MUTEX(mDataMutex);
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
#ifdef DATA_DEBUG
RsDbg() << "CheckRequestStatus: token=" << token ;
@ -1756,7 +1699,7 @@ bool RsGxsDataAccess::getGroupData(const RsGxsGroupId& grpId, RsNxsGrp *& grp_da
grps[grpId] = nullptr ;
if(mDataStore->retrieveNxsGrps(grps, false, true)) // the false here is very important: it removes the private key parts.
if(mDataStore->retrieveNxsGrps(grps, false)) // the false here is very important: it removes the private key parts.
{
grp_data = grps.begin()->second;
return true;
@ -1790,7 +1733,7 @@ void RsGxsDataAccess::tokenList(std::list<uint32_t>& tokens)
bool RsGxsDataAccess::locked_updateRequestStatus( uint32_t token, RsTokenService::GxsRequestStatus status )
{
GxsRequest* req = locked_retrieveCompetedRequest(token);
GxsRequest* req = locked_retrieveCompletedRequest(token);
if(req) req->status = status;
else return false;
@ -1843,7 +1786,7 @@ bool RsGxsDataAccess::disposeOfPublicToken(uint32_t token)
return false;
}
bool RsGxsDataAccess::checkGrpFilter(const RsTokReqOptions &opts, const RsGxsGrpMetaData *meta) const
bool RsGxsDataAccess::checkGrpFilter(const RsTokReqOptions &opts, const std::shared_ptr<RsGxsGrpMetaData>& meta) const
{
bool subscribeMatch = false;
@ -1863,8 +1806,7 @@ bool RsGxsDataAccess::checkGrpFilter(const RsTokReqOptions &opts, const RsGxsGrp
return subscribeMatch;
}
bool RsGxsDataAccess::checkMsgFilter(
const RsTokReqOptions& opts, const RsGxsMsgMetaData* meta ) const
bool RsGxsDataAccess::checkMsgFilter(const RsTokReqOptions& opts, const std::shared_ptr<RsGxsMsgMetaData> &meta ) const
{
if (opts.mStatusMask)
{

14
libretroshare/src/gxs/rsgxsdataaccess.h
View File

@ -28,8 +28,8 @@
#include "rsgds.h"
typedef std::map< RsGxsGroupId, std::map<RsGxsMessageId, const RsGxsMsgMetaData*> > MsgMetaFilter;
typedef std::map< RsGxsGroupId, RsGxsGrpMetaData* > GrpMetaFilter;
typedef std::map< RsGxsGroupId, std::map<RsGxsMessageId, std::shared_ptr<RsGxsMsgMetaData> > > MsgMetaFilter;
typedef std::map< RsGxsGroupId, std::shared_ptr<RsGxsGrpMetaData> > GrpMetaFilter;
bool operator<(const std::pair<uint32_t,GxsRequest*>& p1,const std::pair<uint32_t,GxsRequest*>& p2);
@ -220,7 +220,7 @@ public:
* @param token request token to be redeemed
* @param groupInfo
*/
bool getGroupSummary(const uint32_t &token, std::list<const RsGxsGrpMetaData*>& groupInfo);
bool getGroupSummary(const uint32_t &token, std::list<std::shared_ptr<RsGxsGrpMetaData> > &groupInfo);
/*!
*
@ -276,7 +276,7 @@ private:
* @param token the value of the token for the request object handle wanted
* @return the request associated to this token
*/
GxsRequest* locked_retrieveCompetedRequest(const uint32_t& token);
GxsRequest* locked_retrieveCompletedRequest(const uint32_t& token);
/*!
* Add a gxs request to queue
@ -478,7 +478,7 @@ private:
* @param meta meta containing currently defined options for msg
* @return true if msg meta passes all options
*/
bool checkMsgFilter(const RsTokReqOptions& opts, const RsGxsMsgMetaData* meta) const;
bool checkMsgFilter(const RsTokReqOptions& opts, const std::shared_ptr<RsGxsMsgMetaData>& meta) const;
/*!
* This applies the options to the meta to find out if the given group satisfies
@ -487,7 +487,7 @@ private:
* @param meta meta containing currently defined options for group
* @return true if group meta passes all options
*/
bool checkGrpFilter(const RsTokReqOptions& opts, const RsGxsGrpMetaData* meta) const;
bool checkGrpFilter(const RsTokReqOptions& opts, const std::shared_ptr<RsGxsGrpMetaData> &meta) const;
/*!
@ -511,6 +511,8 @@ private:
std::set<std::pair<uint32_t,GxsRequest*> > mRequestQueue;
std::map<uint32_t, GxsRequest*> mCompletedRequests;
bool mUseMetaCache;
};
#endif // RSGXSDATAACCESS_H

45
libretroshare/src/gxs/rsgxsnetservice.cc
View File

@ -637,7 +637,7 @@ void RsGxsNetService::syncWithPeers()
for(RsGxsGrpMetaTemporaryMap::iterator mit = grpMeta.begin(); mit != grpMeta.end(); ++mit)
{
RsGxsGrpMetaData* meta = mit->second;
const auto& meta = mit->second;
// This was commented out because we want to know how many messages are available for unsubscribed groups.
@ -674,7 +674,7 @@ void RsGxsNetService::syncWithPeers()
RsGxsGrpMetaTemporaryMap::const_iterator mmit = toRequest.begin();
for(; mmit != toRequest.end(); ++mmit)
{
const RsGxsGrpMetaData* meta = mmit->second;
const auto& meta = mmit->second;
const RsGxsGroupId& grpId = mmit->first;
RsGxsCircleId encrypt_to_this_circle_id ;
@ -926,9 +926,9 @@ void RsGxsNetService::handleRecvSyncGrpStatistics(RsNxsSyncGrpStatsItem *grs)
RsGxsGrpMetaTemporaryMap grpMetas;
grpMetas[grs->grpId] = NULL;
mDataStore->retrieveGxsGrpMetaData(grpMetas);
mDataStore->retrieveGxsGrpMetaData(grpMetas);
const RsGxsGrpMetaData* grpMeta = grpMetas[grs->grpId];
const auto& grpMeta = grpMetas[grs->grpId];
if(grpMeta == NULL)
{
@ -957,9 +957,9 @@ void RsGxsNetService::handleRecvSyncGrpStatistics(RsNxsSyncGrpStatsItem *grs)
#ifdef NXS_NET_DEBUG_6
GXSNETDEBUG_PG(grs->PeerId(),grs->grpId) << " retrieving message information." << std::endl;
#endif
mDataStore->retrieveGxsMsgMetaData(reqIds, result);
mDataStore->retrieveGxsMsgMetaData(reqIds, result);
const std::vector<const RsGxsMsgMetaData*>& vec(result[grs->grpId]) ;
const auto& vec(result[grs->grpId]) ;
if(vec.empty()) // that means we don't have any, or there isn't any, but since the default is always 0, no need to send.
return ;
@ -2115,7 +2115,7 @@ void RsGxsNetService::updateServerSyncTS()
{
RS_STACK_MUTEX(mNxsMutex) ;
// retrieve all grps and update TS
mDataStore->retrieveGxsGrpMetaData(gxsMap);
mDataStore->retrieveGxsGrpMetaData(gxsMap);
// (cyril) This code was previously removed because it sounded inconsistent: the list of grps normally does not need to be updated when
// new posts arrive. The two (grp list and msg list) are handled independently. Still, when group meta data updates are received,
@ -2196,7 +2196,7 @@ void RsGxsNetService::updateServerSyncTS()
RS_STACK_MUTEX(mNxsMutex) ;
const RsGxsGrpMetaData* grpMeta = mit->second;
const auto& grpMeta = mit->second;
#ifdef TO_REMOVE
// That accounts for modification of the meta data.
@ -2924,7 +2924,7 @@ void RsGxsNetService::locked_genReqMsgTransaction(NxsTransaction* tr)
grpMetaMap[grpId] = NULL;
mDataStore->retrieveGxsGrpMetaData(grpMetaMap);
const RsGxsGrpMetaData* grpMeta = grpMetaMap[grpId];
const auto& grpMeta = grpMetaMap[grpId];
if(grpMeta == NULL) // this should not happen, but just in case...
{
@ -2956,17 +2956,16 @@ void RsGxsNetService::locked_genReqMsgTransaction(NxsTransaction* tr)
reqIds[grpId] = std::set<RsGxsMessageId>();
GxsMsgMetaResult result;
mDataStore->retrieveGxsMsgMetaData(reqIds, result);
std::vector<const RsGxsMsgMetaData*> &msgMetaV = result[grpId];
auto& msgMetaV = result[grpId];
#ifdef NXS_NET_DEBUG_1
GXSNETDEBUG_PG(item->PeerId(),grpId) << " retrieving grp message list..." << std::endl;
GXSNETDEBUG_PG(item->PeerId(),grpId) << " grp locally contains " << msgMetaV.size() << " messsages." << std::endl;
#endif
std::vector<const RsGxsMsgMetaData*>::const_iterator vit = msgMetaV.begin();
std::set<RsGxsMessageId> msgIdSet;
// put ids in set for each searching
for(; vit != msgMetaV.end(); ++vit)
for(auto vit=msgMetaV.begin(); vit != msgMetaV.end(); ++vit)
msgIdSet.insert((*vit)->mMsgId);
msgMetaV.clear();
@ -3230,7 +3229,7 @@ void RsGxsNetService::locked_genReqGrpTransaction(NxsTransaction* tr)
RsNxsSyncGrpItem*& grpSyncItem = *llit;
const RsGxsGroupId& grpId = grpSyncItem->grpId;
std::map<RsGxsGroupId, RsGxsGrpMetaData*>::const_iterator metaIter = grpMetaMap.find(grpId);
auto metaIter = grpMetaMap.find(grpId);
bool haveItem = false;
bool latestVersion = false;
@ -3312,7 +3311,7 @@ void RsGxsNetService::locked_genSendGrpsTransaction(NxsTransaction* tr)
}
if(!grps.empty())
mDataStore->retrieveNxsGrps(grps, false, false);
mDataStore->retrieveNxsGrps(grps, false);
else
{
#ifdef NXS_NET_DEBUG_1
@ -3552,7 +3551,7 @@ void RsGxsNetService::locked_genSendMsgsTransaction(NxsTransaction* tr)
#endif
#endif
mDataStore->retrieveNxsMsgs(msgIds, msgs, false, false);
mDataStore->retrieveNxsMsgs(msgIds, msgs, false);
NxsTransaction* newTr = new NxsTransaction();
newTr->mFlag = NxsTransaction::FLAG_STATE_WAITING_CONFIRM;
@ -4050,7 +4049,7 @@ void RsGxsNetService::handleRecvSyncGroup(RsNxsSyncGrpReqItem *item)
for(auto mit = grp.begin(); mit != grp.end(); ++mit)
{
const RsGxsGrpMetaData* grpMeta = mit->second;
const auto& grpMeta = mit->second;
// Only send info about subscribed groups.
@ -4369,7 +4368,7 @@ void RsGxsNetService::handleRecvSyncMessage(RsNxsSyncMsgReqItem *item,bool item_
grpMetas[item->grpId] = NULL;
mDataStore->retrieveGxsGrpMetaData(grpMetas);
const RsGxsGrpMetaData* grpMeta = grpMetas[item->grpId];
const auto& grpMeta = grpMetas[item->grpId];
if(grpMeta == NULL)
{
@ -4402,7 +4401,7 @@ void RsGxsNetService::handleRecvSyncMessage(RsNxsSyncMsgReqItem *item,bool item_
GxsMsgMetaResult metaResult;
mDataStore->retrieveGxsMsgMetaData(req, metaResult);
std::vector<const RsGxsMsgMetaData*>& msgMetas = metaResult[item->grpId];
auto& msgMetas = metaResult[item->grpId];
#ifdef NXS_NET_DEBUG_0
GXSNETDEBUG_PG(item->PeerId(),item->grpId) << " retrieving message meta data." << std::endl;
@ -4432,7 +4431,7 @@ void RsGxsNetService::handleRecvSyncMessage(RsNxsSyncMsgReqItem *item,bool item_
{
for(auto vit = msgMetas.begin();vit != msgMetas.end(); ++vit)
{
const RsGxsMsgMetaData* m = *vit;
const auto& m = *vit;
// Check reputation
@ -4577,7 +4576,7 @@ void RsGxsNetService::locked_pushMsgRespFromList(std::list<RsNxsItem*>& itemL, c
}
}
bool RsGxsNetService::canSendMsgIds(std::vector<const RsGxsMsgMetaData*>& msgMetas, const RsGxsGrpMetaData& grpMeta, const RsPeerId& sslId,RsGxsCircleId& should_encrypt_id)
bool RsGxsNetService::canSendMsgIds(std::vector<std::shared_ptr<RsGxsMsgMetaData> >& msgMetas, const RsGxsGrpMetaData& grpMeta, const RsPeerId& sslId,RsGxsCircleId& should_encrypt_id)
{
#ifdef NXS_NET_DEBUG_4
GXSNETDEBUG_PG(sslId,grpMeta.mGroupId) << "RsGxsNetService::canSendMsgIds() CIRCLE VETTING" << std::endl;
@ -4929,7 +4928,7 @@ void RsGxsNetService::sharePublishKeysPending()
// Find the publish keys in the retrieved info
const RsGxsGrpMetaData *grpMeta = grpMetaMap[mit->first] ;
const auto& grpMeta = grpMetaMap[mit->first] ;
if(grpMeta == NULL)
{
@ -5014,7 +5013,7 @@ void RsGxsNetService::handleRecvPublishKeys(RsNxsGroupPublishKeyItem *item)
// update the publish keys in this group meta info
const RsGxsGrpMetaData *grpMeta = grpMetaMap[item->grpId] ;
const auto& grpMeta = grpMetaMap[item->grpId] ;
if (!grpMeta) {
std::cerr << "(EE) RsGxsNetService::handleRecvPublishKeys() grpMeta not found." << std::endl;
return ;
@ -5526,7 +5525,7 @@ bool RsGxsNetService::search(const Sha1CheckSum& hashed_group_id,unsigned char *
RsNxsGrpDataTemporaryMap grpDataMap;
{
RS_STACK_MUTEX(mNxsMutex) ;
mDataStore->retrieveNxsGrps(grpDataMap, true, true);
mDataStore->retrieveNxsGrps(grpDataMap, true);
mLastCacheReloadTS = time(NULL);
}

4
libretroshare/src/gxs/rsgxsnetservice.h
View File

@ -399,7 +399,7 @@ private:
* @return false, if you cannot send to this peer, true otherwise
*/
bool canSendGrpId(const RsPeerId& sslId, const RsGxsGrpMetaData& grpMeta, std::vector<GrpIdCircleVet>& toVet, bool &should_encrypt);
bool canSendMsgIds(std::vector<const RsGxsMsgMetaData*>& msgMetas, const RsGxsGrpMetaData&, const RsPeerId& sslId, RsGxsCircleId &should_encrypt_id);
bool canSendMsgIds(std::vector<std::shared_ptr<RsGxsMsgMetaData> > &msgMetas, const RsGxsGrpMetaData&, const RsPeerId& sslId, RsGxsCircleId &should_encrypt_id);
/*!
* \brief checkPermissionsForFriendGroup
@ -626,6 +626,8 @@ private:
std::map<TurtleRequestId,RsGxsGroupId> mSearchRequests;
std::map<RsGxsGroupId,GroupRequestRecord> mSearchedGroups ;
rstime_t mLastCacheReloadTS ;
bool mUseMetaCache;
};
#endif // RSGXSNETSERVICE_H

13
libretroshare/src/gxs/rsgxsrequesttypes.cc
View File

@ -156,13 +156,6 @@ GroupDataReq::~GroupDataReq()
rsstd::delete_all(mGroupData.begin(), mGroupData.end());
}
MsgMetaReq::~MsgMetaReq()
{
for (GxsMsgMetaResult::iterator it = mMsgMetaData.begin(); it != mMsgMetaData.end(); ++it) {
rsstd::delete_all(it->second.begin(), it->second.end());
}
}
MsgDataReq::~MsgDataReq()
{
for (NxsMsgDataResult::iterator it = mMsgData.begin(); it != mMsgData.end(); ++it) {
@ -172,12 +165,8 @@ MsgDataReq::~MsgDataReq()
MsgRelatedInfoReq::~MsgRelatedInfoReq()
{
for (MsgRelatedMetaResult::iterator metaIt = mMsgMetaResult.begin(); metaIt != mMsgMetaResult.end(); ++metaIt) {
rsstd::delete_all(metaIt->second.begin(), metaIt->second.end());
}
for (NxsMsgRelatedDataResult::iterator dataIt = mMsgDataResult.begin(); dataIt != mMsgDataResult.end(); ++dataIt) {
for (NxsMsgRelatedDataResult::iterator dataIt = mMsgDataResult.begin(); dataIt != mMsgDataResult.end(); ++dataIt)
rsstd::delete_all(dataIt->second.begin(), dataIt->second.end());
}
}
std::ostream& MessageSetFlagReq::print(std::ostream& o) const
{

4
libretroshare/src/gxs/rsgxsrequesttypes.h
View File

@ -55,7 +55,7 @@ public:
virtual std::ostream& print(std::ostream& o) const override;
public:
std::list<RsGxsGroupId> mGroupIds;
std::list<const RsGxsGrpMetaData*> mGroupMetaData;
std::list<std::shared_ptr<RsGxsGrpMetaData> > mGroupMetaData;
};
class GroupIdReq : public GxsRequest
@ -98,7 +98,7 @@ public:
class MsgMetaReq : public GxsRequest
{
public:
virtual ~MsgMetaReq();
virtual ~MsgMetaReq() = default;
virtual std::ostream& print(std::ostream& o) const override;

472
libretroshare/src/gxs/rsgxsutil.cc
View File

@ -41,7 +41,7 @@
static const uint32_t MAX_GXS_IDS_REQUESTS_NET = 10 ; // max number of requests from cache/net (avoids killing the system!)
#define DEBUG_GXSUTIL 1
// #define DEBUG_GXSUTIL 1
#ifdef DEBUG_GXSUTIL
#define GXSUTIL_DEBUG() std::cerr << "[" << time(NULL) << "] : GXS_UTIL : " << __FUNCTION__ << " : "
@ -111,7 +111,7 @@ bool RsGxsCleanUp::clean(RsGxsGroupId& next_group_to_check,std::vector<RsGxsGrou
for(; mit != result.end(); ++mit)
{
std::vector<const RsGxsMsgMetaData*>& metaV = mit->second;
auto& metaV = mit->second;
// First, make a map of which message have a child message. This allows to only delete messages that dont have child messages.
// A more accurate way to go would be to compute the time of the oldest message and possibly delete all the branch, but in the
@ -125,7 +125,7 @@ bool RsGxsCleanUp::clean(RsGxsGroupId& next_group_to_check,std::vector<RsGxsGrou
for( uint32_t i=0;i<metaV.size();++i)
{
const RsGxsMsgMetaData* meta = metaV[i];
const auto& meta = metaV[i];
bool have_kids = (messages_with_kids.find(meta->mMsgId)!=messages_with_kids.end());
@ -170,7 +170,9 @@ bool RsGxsCleanUp::clean(RsGxsGroupId& next_group_to_check,std::vector<RsGxsGrou
if(it->first == next_group_to_check)
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "Had the time to test all groups. Will start again at " << it->first << std::endl;
#endif
full_round = true;
break;
}
@ -182,7 +184,9 @@ bool RsGxsCleanUp::clean(RsGxsGroupId& next_group_to_check,std::vector<RsGxsGrou
//if(tm > now + 1) // we spent more than 1 sec on the job already
if(tm > now) // we spent more than 1 sec on the job already
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "Aborting cleanup because it took too much time already. Next group left to be " << it->first << std::endl;
#endif
next_group_to_check = it->first;
full_round = false;
break;
@ -210,245 +214,80 @@ void RsGxsIntegrityCheck::run()
std::vector<RsGxsGroupId> grps_to_delete;
GxsMsgReq msgs_to_delete;
check(mGenExchangeClient->serviceType(), mGixs, mDs
#ifdef RS_DEEP_CHANNEL_INDEX
, mGenExchangeClient, mSerializer
#endif
, mDeletedGrps, mDeletedMsgs);
check(mGenExchangeClient->serviceType(), mGixs, mDs);
RS_STACK_MUTEX(mIntegrityMutex);
mDone = true;
}
bool RsGxsIntegrityCheck::check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds
#ifdef RS_DEEP_CHANNEL_INDEX
, RsGenExchange* mGenExchangeClient, RsSerialType& mSerializer
#endif
, std::vector<RsGxsGroupId>& grpsToDel, GxsMsgReq& msgsToDel)
bool RsGxsIntegrityCheck::check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds)
{
#ifdef RS_DEEP_CHANNEL_INDEX
bool isGxsChannels = mGenExchangeClient->serviceType() == RS_SERVICE_GXS_TYPE_CHANNELS;
std::set<RsGxsGroupId> indexedGroups;
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "Parsing all groups and messages MetaData in service " << std::hex << mds->serviceType() << std::endl;
#endif
// first take out all the groups
std::map<RsGxsGroupId, RsNxsGrp*> grp;
mds->retrieveNxsGrps(grp, true, true);
std::map<RsGxsGroupId, std::shared_ptr<RsGxsGrpMetaData> > grp;
mds->retrieveGxsGrpMetaData(grp);
GxsMsgReq msgIds;
GxsMsgReq grps;
std::map<RsGxsId,RsIdentityUsage> used_gxs_ids ;
std::set<RsGxsGroupId> subscribed_groups ;
// compute hash and compare to stored value, if it fails then simply add it
// to list
for( std::map<RsGxsGroupId, RsNxsGrp*>::iterator git = grp.begin(); git != grp.end(); ++git )
{
RsNxsGrp* grp = git->second;
RsFileHash currHash;
pqihash pHash;
pHash.addData(grp->grp.bin_data, grp->grp.bin_len);
pHash.Complete(currHash);
// Check that message ids...
if(currHash == grp->metaData->mHash)
{
// Get all message ids of group, store them in msgIds, creating the grp entry at the same time.
if (mds->retrieveMsgIds(grp->grpId, msgIds[grp->grpId]) == 1)
{
// store the group for retrieveNxsMsgs
grps[grp->grpId];
if(grp->metaData->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED)
{
subscribed_groups.insert(git->first);
if(!grp->metaData->mAuthorId.isNull())
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "TimeStamping group authors' key ID " << grp->metaData->mAuthorId << " in group ID " << grp->grpId << std::endl;
#endif
if( rsReputations &&
rsReputations->overallReputationLevel(
grp->metaData->mAuthorId ) >
RsReputationLevel::LOCALLY_NEGATIVE )
used_gxs_ids.insert(std::make_pair(grp->metaData->mAuthorId, RsIdentityUsage(RsServiceType(service_type), RsIdentityUsage::GROUP_AUTHOR_KEEP_ALIVE,grp->grpId)));
}
}
}
else
msgIds.erase(msgIds.find(grp->grpId)); // could not get them, so group is removed from list.
#ifdef RS_DEEP_CHANNEL_INDEX
// This should be moved to p3gxschannels. It is really not the place for this here!
if( isGxsChannels
&& grp->metaData->mCircleType == GXS_CIRCLE_TYPE_PUBLIC
&& grp->metaData->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED )
{
RsGxsGrpMetaData meta;
meta.deserialise(grp->meta.bin_data, grp->meta.bin_len);
uint32_t blz = grp->grp.bin_len;
RsItem* rIt = mSerializer.deserialise(grp->grp.bin_data,
&blz);
if( RsGxsChannelGroupItem* cgIt =
dynamic_cast<RsGxsChannelGroupItem*>(rIt) )
{
RsGxsChannelGroup cg;
cgIt->toChannelGroup(cg, false);
cg.mMeta = meta;
indexedGroups.insert(grp->grpId);
DeepChannelsIndex::indexChannelGroup(cg);
}
else
{
std::cerr << __PRETTY_FUNCTION__ << " Group: "
<< meta.mGroupId.toStdString() << " "
<< meta.mGroupName
<< " doesn't seems a channel, please "
<< "report to developers"
<< std::endl;
print_stacktrace();
}
delete rIt;
}
#endif // def RS_DEEP_CHANNEL_INDEX
}
else
{
std::cerr << __PRETTY_FUNCTION__ <<" (EE) deleting group " << grp->grpId << " with wrong hash or null/corrupted meta data. meta=" << grp->metaData << std::endl;
grpsToDel.push_back(grp->grpId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelFromIndex(grp->grpId);
#endif // def RS_DEEP_CHANNEL_INDEX
}
delete grp;
}
// now messages
GxsMsgResult msgs;
mds->retrieveNxsMsgs(grps, msgs, false, true);
// Check msg ids and messages. Go through all message IDs referred to by the db call
// and verify that the message belongs to the nxs msg data that was just retrieved.
for(auto& msgIdsIt:msgIds)
for( auto git = grp.begin(); git != grp.end(); ++git )
{
const RsGxsGroupId& grpId = msgIdsIt.first;
std::set<RsGxsMessageId>& msgIdV = msgIdsIt.second;
const auto& grpMeta = git->second;
std::set<RsGxsMessageId> nxsMsgS;
std::vector<RsNxsMsg*>& nxsMsgV = msgs[grpId];
if (mds->retrieveMsgIds(grpMeta->mGroupId, msgIds[grpMeta->mGroupId]) == 1)
{
if(grpMeta->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED)
{
subscribed_groups.insert(git->first);
// To make the search efficient, we first build a set of msgIds to search in.
// Set build and search are both O(n log(n)).
for(auto& nxsMsg:nxsMsgV)
if(nxsMsg)
nxsMsgS.insert(nxsMsg->msgId);
for (auto& msgId:msgIdV)
if(nxsMsgS.find(msgId) == nxsMsgS.end())
{
msgsToDel[grpId].insert(msgId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelPostFromIndex(grpId, msgId);
#endif // def RS_DEEP_CHANNEL_INDEX
}
if(!grpMeta->mAuthorId.isNull())
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "TimeStamping group authors' key ID " << grpMeta->mAuthorId << " in group ID " << grpMeta->mGroupId << std::endl;
#endif
if( rsReputations && rsReputations->overallReputationLevel( grpMeta->mAuthorId ) > RsReputationLevel::LOCALLY_NEGATIVE )
used_gxs_ids.insert(std::make_pair(grpMeta->mAuthorId, RsIdentityUsage(RsServiceType(service_type), RsIdentityUsage::GROUP_AUTHOR_KEEP_ALIVE,grpMeta->mGroupId)));
}
}
}
else
msgIds.erase(msgIds.find(grpMeta->mGroupId)); // could not get them, so group is removed from list.
}
GxsMsgResult::iterator mit = msgs.begin();
for(; mit != msgs.end(); ++mit)
// now messages
GxsMsgMetaResult msgMetas;
mds->retrieveGxsMsgMetaData(msgIds, msgMetas);
for(auto mit=msgMetas.begin(); mit != msgMetas.end(); ++mit)
{
std::vector<RsNxsMsg*>& msgV = mit->second;
std::vector<RsNxsMsg*>::iterator vit = msgV.begin();
const auto& msgM = mit->second;
for(; vit != msgV.end(); ++vit)
for(auto vit=msgM.begin(); vit != msgM.end(); ++vit)
{
RsNxsMsg* msg = *vit;
RsFileHash currHash;
pqihash pHash;
pHash.addData(msg->msg.bin_data, msg->msg.bin_len);
pHash.Complete(currHash);
const auto& meta = *vit;
if(msg->metaData == NULL || currHash != msg->metaData->mHash)
{
std::cerr << __PRETTY_FUNCTION__ <<" (EE) deleting message " << msg->msgId << " in group " << msg->grpId << " with wrong hash or null/corrupted meta data. meta=" << (void*)msg->metaData << std::endl;
msgsToDel[msg->grpId].insert(msg->msgId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelPostFromIndex(
msg->grpId, msg->msgId );
#endif // def RS_DEEP_CHANNEL_INDEX
}
else if (subscribed_groups.count(msg->metaData->mGroupId))
{
#ifdef RS_DEEP_CHANNEL_INDEX
// This should be moved to p3gxschannels. It is really not the place for this here!
if( isGxsChannels && indexedGroups.count(msg->metaData->mGroupId) )
{
RsGxsMsgMetaData meta;
meta.deserialise(msg->meta.bin_data, &msg->meta.bin_len);
uint32_t blz = msg->msg.bin_len;
RsItem* rIt = mSerializer.deserialise(msg->msg.bin_data,
&blz);
if( RsGxsChannelPostItem* cgIt =
dynamic_cast<RsGxsChannelPostItem*>(rIt) )
{
RsGxsChannelPost cg;
cgIt->toChannelPost(cg, false);
cg.mMeta = meta;
DeepChannelsIndex::indexChannelPost(cg);
}
else if(dynamic_cast<RsGxsCommentItem*>(rIt)) {}
else if(dynamic_cast<RsGxsVoteItem*>(rIt)) {}
else
{
std::cerr << __PRETTY_FUNCTION__ << " Message: "
<< meta.mMsgId.toStdString()
<< " in group: "
<< meta.mGroupId.toStdString() << " "
<< " doesn't seems a channel post, please "
<< "report to developers"
<< std::endl;
print_stacktrace();
}
delete rIt;
}
#endif // def RS_DEEP_CHANNEL_INDEX
if(!msg->metaData->mAuthorId.isNull())
if (subscribed_groups.count(meta->mGroupId))
if(!meta->mAuthorId.isNull())
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "TimeStamping message authors' key ID " << msg->metaData->mAuthorId << " in message " << msg->msgId << ", group ID " << msg->grpId<< std::endl;
GXSUTIL_DEBUG() << "TimeStamping message authors' key ID " << meta->mAuthorId << " in message " << meta->mMsgId << ", group ID " << meta->mGroupId<< std::endl;
#endif
if( rsReputations &&
rsReputations->overallReputationLevel(
msg->metaData->mAuthorId ) >
RsReputationLevel::LOCALLY_NEGATIVE )
used_gxs_ids.insert(std::make_pair(msg->metaData->mAuthorId,RsIdentityUsage(RsServiceType(service_type),
if( rsReputations && rsReputations->overallReputationLevel( meta->mAuthorId ) > RsReputationLevel::LOCALLY_NEGATIVE )
used_gxs_ids.insert(std::make_pair(meta->mAuthorId,RsIdentityUsage(RsServiceType(service_type),
RsIdentityUsage::MESSAGE_AUTHOR_KEEP_ALIVE,
msg->metaData->mGroupId,
msg->metaData->mMsgId,
msg->metaData->mParentId,
msg->metaData->mThreadId))) ;
meta->mGroupId,
meta->mMsgId,
meta->mParentId,
meta->mThreadId))) ;
}
}
delete msg;
}
}
@ -513,6 +352,211 @@ bool RsGxsIntegrityCheck::check(uint16_t service_type, RsGixs *mgixs, RsGeneralD
return true;
}
bool RsGxsSinglePassIntegrityCheck::check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds
#ifdef RS_DEEP_CHANNEL_INDEX
, RsGenExchange* mGenExchangeClient, RsSerialType& mSerializer
#endif
, std::vector<RsGxsGroupId>& grpsToDel, GxsMsgReq& msgsToDel)
{
#ifdef DEBUG_GXSUTIL
GXSUTIL_DEBUG() << "Parsing all groups and messages data in service " << std::hex << mds->serviceType() << " for integrity check. Could take a while..." << std::endl;
#endif
#ifdef RS_DEEP_CHANNEL_INDEX
bool isGxsChannels = mGenExchangeClient->serviceType() == RS_SERVICE_GXS_TYPE_CHANNELS;
std::set<RsGxsGroupId> indexedGroups;
#endif
// first take out all the groups
std::map<RsGxsGroupId, RsNxsGrp*> grp;
mds->retrieveNxsGrps(grp, true);
GxsMsgReq msgIds;
GxsMsgReq grps;
std::map<RsGxsId,RsIdentityUsage> used_gxs_ids ;
std::set<RsGxsGroupId> subscribed_groups ;
// compute hash and compare to stored value, if it fails then simply add it
// to list
for( std::map<RsGxsGroupId, RsNxsGrp*>::iterator git = grp.begin(); git != grp.end(); ++git )
{
RsNxsGrp* grp = git->second;
RsFileHash currHash;
pqihash pHash;
pHash.addData(grp->grp.bin_data, grp->grp.bin_len);
pHash.Complete(currHash);
if(currHash == grp->metaData->mHash)
{
// Get all message ids of group, store them in msgIds, creating the grp entry at the same time.
if (mds->retrieveMsgIds(grp->grpId, msgIds[grp->grpId]) == 1)
{
// store the group for retrieveNxsMsgs
grps[grp->grpId];
if(grp->metaData->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED)
subscribed_groups.insert(git->first);
}
else
msgIds.erase(msgIds.find(grp->grpId)); // could not get them, so group is removed from list.
#ifdef RS_DEEP_CHANNEL_INDEX
// This should be moved to p3gxschannels. It is really not the place for this here!
if( isGxsChannels
&& grp->metaData->mCircleType == GXS_CIRCLE_TYPE_PUBLIC
&& grp->metaData->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED )
{
RsGxsGrpMetaData meta;
meta.deserialise(grp->meta.bin_data, grp->meta.bin_len);
uint32_t blz = grp->grp.bin_len;
RsItem* rIt = mSerializer.deserialise(grp->grp.bin_data,
&blz);
if( RsGxsChannelGroupItem* cgIt =
dynamic_cast<RsGxsChannelGroupItem*>(rIt) )
{
RsGxsChannelGroup cg;
cgIt->toChannelGroup(cg, false);
cg.mMeta = meta;
indexedGroups.insert(grp->grpId);
DeepChannelsIndex::indexChannelGroup(cg);
}
else
{
std::cerr << __PRETTY_FUNCTION__ << " Group: "
<< meta.mGroupId.toStdString() << " "
<< meta.mGroupName
<< " doesn't seems a channel, please "
<< "report to developers"
<< std::endl;
print_stacktrace();
}
delete rIt;
}
#endif // def RS_DEEP_CHANNEL_INDEX
}
else
{
std::cerr << __PRETTY_FUNCTION__ <<" (EE) deleting group " << grp->grpId << " with wrong hash or null/corrupted meta data. meta=" << grp->metaData << std::endl;
grpsToDel.push_back(grp->grpId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelFromIndex(grp->grpId);
#endif // def RS_DEEP_CHANNEL_INDEX
}
delete grp;
}
// now messages
GxsMsgResult msgs;
mds->retrieveNxsMsgs(grps, msgs, true);
// Check msg ids and messages. Go through all message IDs referred to by the db call
// and verify that the message belongs to the nxs msg data that was just retrieved.
for(auto& msgIdsIt:msgIds)
{
const RsGxsGroupId& grpId = msgIdsIt.first;
std::set<RsGxsMessageId>& msgIdV = msgIdsIt.second;
std::set<RsGxsMessageId> nxsMsgS;
std::vector<RsNxsMsg*>& nxsMsgV = msgs[grpId];
// To make the search efficient, we first build a set of msgIds to search in.
// Set build and search are both O(n log(n)).
for(auto& nxsMsg:nxsMsgV)
if(nxsMsg)
nxsMsgS.insert(nxsMsg->msgId);
for (auto& msgId:msgIdV)
if(nxsMsgS.find(msgId) == nxsMsgS.end())
{
msgsToDel[grpId].insert(msgId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelPostFromIndex(grpId, msgId);
#endif // def RS_DEEP_CHANNEL_INDEX
}
}
for(auto mit = msgs.begin(); mit != msgs.end(); ++mit)
{
std::vector<RsNxsMsg*>& msgV = mit->second;
std::vector<RsNxsMsg*>::iterator vit = msgV.begin();
for(; vit != msgV.end(); ++vit)
{
RsNxsMsg* msg = *vit;
RsFileHash currHash;
pqihash pHash;
pHash.addData(msg->msg.bin_data, msg->msg.bin_len);
pHash.Complete(currHash);
if(msg->metaData == NULL || currHash != msg->metaData->mHash)
{
std::cerr << __PRETTY_FUNCTION__ <<" (EE) deleting message " << msg->msgId << " in group " << msg->grpId << " with wrong hash or null/corrupted meta data. meta=" << (void*)msg->metaData << std::endl;
msgsToDel[msg->grpId].insert(msg->msgId);
#ifdef RS_DEEP_CHANNEL_INDEX
if(isGxsChannels)
DeepChannelsIndex::removeChannelPostFromIndex(
msg->grpId, msg->msgId );
#endif // def RS_DEEP_CHANNEL_INDEX
}
else if (subscribed_groups.count(msg->metaData->mGroupId))
{
#ifdef RS_DEEP_CHANNEL_INDEX
// This should be moved to p3gxschannels. It is really not the place for this here!
if( isGxsChannels && indexedGroups.count(msg->metaData->mGroupId) )
{
RsGxsMsgMetaData meta;
meta.deserialise(msg->meta.bin_data, &msg->meta.bin_len);
uint32_t blz = msg->msg.bin_len;
RsItem* rIt = mSerializer.deserialise(msg->msg.bin_data,
&blz);
if( RsGxsChannelPostItem* cgIt =
dynamic_cast<RsGxsChannelPostItem*>(rIt) )
{
RsGxsChannelPost cg;
cgIt->toChannelPost(cg, false);
cg.mMeta = meta;
DeepChannelsIndex::indexChannelPost(cg);
}
else if(dynamic_cast<RsGxsCommentItem*>(rIt)) {}
else if(dynamic_cast<RsGxsVoteItem*>(rIt)) {}
else
{
std::cerr << __PRETTY_FUNCTION__ << " Message: "
<< meta.mMsgId.toStdString()
<< " in group: "
<< meta.mGroupId.toStdString() << " "
<< " doesn't seems a channel post, please "
<< "report to developers"
<< std::endl;
print_stacktrace();
}
delete rIt;
}
#endif // def RS_DEEP_CHANNEL_INDEX
}
delete msg;
}
}
return true;
}
bool RsGxsIntegrityCheck::isDone()
{
RS_STACK_MUTEX(mIntegrityMutex);

37
libretroshare/src/gxs/rsgxsutil.h
View File

@ -102,10 +102,10 @@ public:
}
};
typedef std::map<RsGxsGroupId,RsGxsGrpMetaData*> RsGxsGrpMetaTemporaryMap;
typedef t_RsGxsGenericDataTemporaryMap<RsGxsGroupId,RsNxsGrp> RsNxsGrpDataTemporaryMap;
typedef std::map<RsGxsGroupId, std::shared_ptr<RsGxsGrpMetaData> > RsGxsGrpMetaTemporaryMap; // This map doesn't need to delete elements since it holds
typedef std::map<RsGxsGroupId,std::vector<std::shared_ptr<RsGxsMsgMetaData> > > RsGxsMsgMetaTemporaryMap; // shared_ptr's.
typedef t_RsGxsGenericDataTemporaryMapVector<RsGxsMsgMetaData> RsGxsMsgMetaTemporaryMap ;
typedef t_RsGxsGenericDataTemporaryMap<RsGxsGroupId,RsNxsGrp> RsNxsGrpDataTemporaryMap;
typedef t_RsGxsGenericDataTemporaryMapVector<RsNxsMsg> RsNxsMsgDataTemporaryMap ;
typedef t_RsGxsGenericDataTemporaryList<RsNxsGrp> RsNxsGrpDataTemporaryList ;
@ -175,11 +175,7 @@ public:
RsGenExchange *genex, RsSerialType& gxsSerialiser,
RsGixs *gixs);
static bool check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds
#ifdef RS_DEEP_CHANNEL_INDEX
, RsGenExchange* mGenExchangeClient, RsSerialType& mSerializer
#endif
, std::vector<RsGxsGroupId>& grpsToDel, GxsMsgReq& msgsToDel);
static bool check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds);
bool isDone();
void run();
@ -190,9 +186,6 @@ private:
RsGeneralDataService* const mDs;
RsGenExchange *mGenExchangeClient;
#ifdef RS_DEEP_CHANNEL_INDEX
RsSerialType& mSerializer;
#endif
bool mDone;
RsMutex mIntegrityMutex;
std::vector<RsGxsGroupId> mDeletedGrps;
@ -201,6 +194,28 @@ private:
RsGixs* mGixs;
};
/*!
* Checks the integrity message and groups
* in rsDataService using computed hash
*/
class RsGxsSinglePassIntegrityCheck
{
public:
/*!
*
* @param dataService
* @param mGroupTS
* @param chunkSize
* @param sleepPeriod
*/
static bool check(uint16_t service_type, RsGixs *mgixs, RsGeneralDataService *mds
#ifdef RS_DEEP_CHANNEL_INDEX
, RsGenExchange* mGenExchangeClient, RsSerialType& mSerializer
#endif
, std::vector<RsGxsGroupId>& grpsToDel, GxsMsgReq& msgsToDel);
};
class GroupUpdate
{
public:

4
libretroshare/src/gxstrans/p3gxstrans.cc
View File

@ -361,7 +361,7 @@ void p3GxsTrans::GxsTransIntegrityCleanupThread::run()
// first take out all the groups
std::map<RsGxsGroupId, RsNxsGrp*> grp;
mDs->retrieveNxsGrps(grp, true, true);
mDs->retrieveNxsGrps(grp, true);
#ifdef DEBUG_GXSTRANS
std::cerr << "GxsTransIntegrityCleanupThread::run()" << std::endl;
@ -389,7 +389,7 @@ void p3GxsTrans::GxsTransIntegrityCleanupThread::run()
std::list<RsGxsTransId> received_msgs ;
GxsMsgResult msgs;
mDs->retrieveNxsMsgs(grps, msgs, false, true);
mDs->retrieveNxsMsgs(grps, msgs, true);
for(GxsMsgResult::iterator mit = msgs.begin();mit != msgs.end(); ++mit)
{

3
libretroshare/src/retroshare/rsgxsifacetypes.h
View File

@ -66,6 +66,7 @@ struct RsGroupMetaData : RsSerializable
virtual ~RsGroupMetaData() {}
void operator =(const RsGxsGrpMetaData& rGxsMeta);
RsGroupMetaData(const RsGxsGrpMetaData& rGxsMeta) { operator=(rGxsMeta); }
RsGxsGroupId mGroupId;
std::string mGroupName;
@ -138,7 +139,9 @@ struct RsMsgMetaData : RsSerializable
RsMsgMetaData() : mPublishTs(0), mMsgFlags(0), mMsgStatus(0), mChildTs(0) {}
virtual ~RsMsgMetaData() {}
void operator =(const RsGxsMsgMetaData& rGxsMeta);
RsMsgMetaData(const RsGxsMsgMetaData& rGxsMeta) { operator=(rGxsMeta); }
RsGxsGroupId mGroupId;
RsGxsMessageId mMsgId;

4
retroshare-gui/src/gui/feeds/GxsForumMsgItem.cpp
View File

@ -216,8 +216,8 @@ void GxsForumMsgItem::loadGroup()
void GxsForumMsgItem::loadMessage()
{
#ifdef DEBUG_ITEM
std::cerr << "GxsForumMsgItem::loadMessage()";
#ifndef DEBUG_ITEM
std::cerr << "GxsForumMsgItem::loadMessage(): messageId=" << messageId() << " groupId=" << groupId() ;
std::cerr << std::endl;
#endif