mirror of
https://github.com/RetroShare/RetroShare.git
synced 2025-05-03 14:45:12 -04:00
added auto-request of missing GXS ids for group posts authors and group authors
This commit is contained in:
csoler
parent
8d886b8ecc
commit
92780b7688
2 changed files with 172 additions and 114 deletions
|
|
@ -77,10 +77,10 @@ RsGenExchange::RsGenExchange(RsGeneralDataService *gds, RsNetworkExchangeService
|
||||||
mAuthenPolicy(authenPolicy),
|
mAuthenPolicy(authenPolicy),
|
||||||
MESSAGE_STORE_PERIOD(messageStorePeriod),
|
MESSAGE_STORE_PERIOD(messageStorePeriod),
|
||||||
mCleaning(false),
|
mCleaning(false),
|
||||||
mLastClean(RSRandom::random_u32() % INTEGRITY_CHECK_PERIOD), // this helps unsynchronising the checks for the different services
|
mLastClean((int)time(NULL) - (int)(RSRandom::random_u32() % MSG_CLEANUP_PERIOD)), // this helps unsynchronising the checks for the different services
|
||||||
mMsgCleanUp(NULL),
|
mMsgCleanUp(NULL),
|
||||||
mChecking(false),
|
mChecking(false),
|
||||||
mLastCheck(RSRandom::random_u32() % INTEGRITY_CHECK_PERIOD), // this helps unsynchronising the checks for the different services
|
mLastCheck((int)time(NULL) - (int)(RSRandom::random_u32() % INTEGRITY_CHECK_PERIOD)), // this helps unsynchronising the checks for the different services
|
||||||
mIntegrityCheck(NULL),
|
mIntegrityCheck(NULL),
|
||||||
CREATE_FAIL(0),
|
CREATE_FAIL(0),
|
||||||
CREATE_SUCCESS(1),
|
CREATE_SUCCESS(1),
|
||||||
|
|
|
||||||
|
|
@ -27,9 +27,13 @@
|
||||||
|
|
||||||
#include "rsgxsutil.h"
|
#include "rsgxsutil.h"
|
||||||
#include "retroshare/rsgxsflags.h"
|
#include "retroshare/rsgxsflags.h"
|
||||||
|
#include "retroshare/rspeers.h"
|
||||||
#include "pqi/pqihash.h"
|
#include "pqi/pqihash.h"
|
||||||
#include "gxs/rsgixs.h"
|
#include "gxs/rsgixs.h"
|
||||||
|
|
||||||
|
static const uint32_t MAX_GXS_IDS_REQUESTS_NET = 10 ; // max number of requests from cache/net (avoids killing the system!)
|
||||||
|
|
||||||
|
//#define GXSUTIL_DEBUG 1
|
||||||
|
|
||||||
RsGxsMessageCleanUp::RsGxsMessageCleanUp(RsGeneralDataService* const dataService, uint32_t messageStorePeriod, uint32_t chunkSize)
|
RsGxsMessageCleanUp::RsGxsMessageCleanUp(RsGeneralDataService* const dataService, uint32_t messageStorePeriod, uint32_t chunkSize)
|
||||||
: mDs(dataService), MESSAGE_STORE_PERIOD(messageStorePeriod), CHUNK_SIZE(chunkSize)
|
: mDs(dataService), MESSAGE_STORE_PERIOD(messageStorePeriod), CHUNK_SIZE(chunkSize)
|
||||||
|
|
@ -118,139 +122,193 @@ void RsGxsIntegrityCheck::run()
|
||||||
|
|
||||||
bool RsGxsIntegrityCheck::check()
|
bool RsGxsIntegrityCheck::check()
|
||||||
{
|
{
|
||||||
|
// first take out all the groups
|
||||||
|
std::map<RsGxsGroupId, RsNxsGrp*> grp;
|
||||||
|
mDs->retrieveNxsGrps(grp, true, true);
|
||||||
|
std::vector<RsGxsGroupId> grpsToDel;
|
||||||
|
GxsMsgReq msgIds;
|
||||||
|
GxsMsgReq grps;
|
||||||
|
|
||||||
// first take out all the groups
|
std::set<RsGxsId> used_gxs_ids ;
|
||||||
std::map<RsGxsGroupId, RsNxsGrp*> grp;
|
std::set<RsGxsGroupId> subscribed_groups ;
|
||||||
mDs->retrieveNxsGrps(grp, true, true);
|
|
||||||
std::vector<RsGxsGroupId> grpsToDel;
|
|
||||||
GxsMsgReq msgIds;
|
|
||||||
GxsMsgReq grps;
|
|
||||||
|
|
||||||
std::set<RsGxsGroupId> subscribed_groups ;
|
// compute hash and compare to stored value, if it fails then simply add it
|
||||||
|
// to list
|
||||||
|
std::map<RsGxsGroupId, RsNxsGrp*>::iterator git = grp.begin();
|
||||||
|
for(; 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);
|
||||||
|
|
||||||
// compute hash and compare to stored value, if it fails then simply add it
|
if(currHash == grp->metaData->mHash)
|
||||||
// to list
|
{
|
||||||
std::map<RsGxsGroupId, RsNxsGrp*>::iterator git = grp.begin();
|
// get all message ids of group
|
||||||
for(; git != grp.end(); ++git)
|
if (mDs->retrieveMsgIds(grp->grpId, msgIds[grp->grpId]) == 1)
|
||||||
{
|
{
|
||||||
RsNxsGrp* grp = git->second;
|
// store the group for retrieveNxsMsgs
|
||||||
RsFileHash currHash;
|
grps[grp->grpId];
|
||||||
pqihash pHash;
|
|
||||||
pHash.addData(grp->grp.bin_data, grp->grp.bin_len);
|
|
||||||
pHash.Complete(currHash);
|
|
||||||
|
|
||||||
if(currHash == grp->metaData->mHash)
|
if(grp->metaData->mSubscribeFlags & GXS_SERV::GROUP_SUBSCRIBE_SUBSCRIBED)
|
||||||
{
|
{
|
||||||
// get all message ids of group
|
subscribed_groups.insert(git->first) ;
|
||||||
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)
|
if(!grp->metaData->mAuthorId.isNull())
|
||||||
{
|
{
|
||||||
subscribed_groups.insert(git->first) ;
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << "TimeStamping group authors' key ID " << grp->metaData->mAuthorId << " in group ID " << grp->grpId << std::endl;
|
||||||
|
#endif
|
||||||
|
|
||||||
if(!grp->metaData->mAuthorId.isNull())
|
used_gxs_ids.insert(grp->metaData->mAuthorId) ;
|
||||||
{
|
}
|
||||||
std::cerr << "TimeStamping group authors' key ID " << grp->metaData->mAuthorId << " in group ID " << grp->metaData->mAuthorId << std::endl;
|
}
|
||||||
mGixs->timeStampKey(grp->metaData->mAuthorId) ;
|
}
|
||||||
}
|
else
|
||||||
}
|
{
|
||||||
}
|
msgIds.erase(msgIds.find(grp->grpId));
|
||||||
else
|
// grpsToDel.push_back(grp->grpId);
|
||||||
{
|
}
|
||||||
msgIds.erase(msgIds.find(grp->grpId));
|
|
||||||
// grpsToDel.push_back(grp->grpId);
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
grpsToDel.push_back(grp->grpId);
|
grpsToDel.push_back(grp->grpId);
|
||||||
}
|
}
|
||||||
delete grp;
|
delete grp;
|
||||||
}
|
}
|
||||||
|
|
||||||
mDs->removeGroups(grpsToDel);
|
mDs->removeGroups(grpsToDel);
|
||||||
|
|
||||||
// now messages
|
// now messages
|
||||||
GxsMsgReq msgsToDel;
|
GxsMsgReq msgsToDel;
|
||||||
GxsMsgResult msgs;
|
GxsMsgResult msgs;
|
||||||
|
|
||||||
mDs->retrieveNxsMsgs(grps, msgs, false, true);
|
mDs->retrieveNxsMsgs(grps, msgs, false, true);
|
||||||
|
|
||||||
// check msg ids and messages
|
// check msg ids and messages
|
||||||
GxsMsgReq::iterator msgIdsIt;
|
GxsMsgReq::iterator msgIdsIt;
|
||||||
for (msgIdsIt = msgIds.begin(); msgIdsIt != msgIds.end(); ++msgIdsIt)
|
for (msgIdsIt = msgIds.begin(); msgIdsIt != msgIds.end(); ++msgIdsIt)
|
||||||
{
|
{
|
||||||
const RsGxsGroupId& grpId = msgIdsIt->first;
|
const RsGxsGroupId& grpId = msgIdsIt->first;
|
||||||
std::vector<RsGxsMessageId> &msgIdV = msgIdsIt->second;
|
std::vector<RsGxsMessageId> &msgIdV = msgIdsIt->second;
|
||||||
|
|
||||||
std::vector<RsGxsMessageId>::iterator msgIdIt;
|
std::vector<RsGxsMessageId>::iterator msgIdIt;
|
||||||
for (msgIdIt = msgIdV.begin(); msgIdIt != msgIdV.end(); ++msgIdIt)
|
for (msgIdIt = msgIdV.begin(); msgIdIt != msgIdV.end(); ++msgIdIt)
|
||||||
{
|
{
|
||||||
const RsGxsMessageId& msgId = *msgIdIt;
|
const RsGxsMessageId& msgId = *msgIdIt;
|
||||||
std::vector<RsNxsMsg*> &nxsMsgV = msgs[grpId];
|
std::vector<RsNxsMsg*> &nxsMsgV = msgs[grpId];
|
||||||
|
|
||||||
std::vector<RsNxsMsg*>::iterator nxsMsgIt;
|
std::vector<RsNxsMsg*>::iterator nxsMsgIt;
|
||||||
for (nxsMsgIt = nxsMsgV.begin(); nxsMsgIt != nxsMsgV.end(); ++nxsMsgIt)
|
for (nxsMsgIt = nxsMsgV.begin(); nxsMsgIt != nxsMsgV.end(); ++nxsMsgIt)
|
||||||
{
|
{
|
||||||
RsNxsMsg *nxsMsg = *nxsMsgIt;
|
RsNxsMsg *nxsMsg = *nxsMsgIt;
|
||||||
if (nxsMsg && msgId == nxsMsg->msgId)
|
if (nxsMsg && msgId == nxsMsg->msgId)
|
||||||
{
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (nxsMsgIt == nxsMsgV.end())
|
if (nxsMsgIt == nxsMsgV.end())
|
||||||
{
|
{
|
||||||
msgsToDel[grpId].push_back(msgId);
|
msgsToDel[grpId].push_back(msgId);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
GxsMsgResult::iterator mit = msgs.begin();
|
GxsMsgResult::iterator mit = msgs.begin();
|
||||||
|
|
||||||
for(; mit != msgs.end(); ++mit)
|
for(; mit != msgs.end(); ++mit)
|
||||||
{
|
{
|
||||||
std::vector<RsNxsMsg*>& msgV = mit->second;
|
std::vector<RsNxsMsg*>& msgV = mit->second;
|
||||||
std::vector<RsNxsMsg*>::iterator vit = msgV.begin();
|
std::vector<RsNxsMsg*>::iterator vit = msgV.begin();
|
||||||
|
|
||||||
for(; vit != msgV.end(); ++vit)
|
for(; vit != msgV.end(); ++vit)
|
||||||
{
|
{
|
||||||
RsNxsMsg* msg = *vit;
|
RsNxsMsg* msg = *vit;
|
||||||
RsFileHash currHash;
|
RsFileHash currHash;
|
||||||
pqihash pHash;
|
pqihash pHash;
|
||||||
pHash.addData(msg->msg.bin_data, msg->msg.bin_len);
|
pHash.addData(msg->msg.bin_data, msg->msg.bin_len);
|
||||||
pHash.Complete(currHash);
|
pHash.Complete(currHash);
|
||||||
|
|
||||||
if(msg->metaData == NULL || currHash != msg->metaData->mHash)
|
if(msg->metaData == NULL || currHash != msg->metaData->mHash)
|
||||||
{
|
{
|
||||||
std::cerr << "(EE) deleting message data with wrong hash or null meta data. meta=" << (void*)msg->metaData << std::endl;
|
std::cerr << "(EE) deleting message data with wrong hash or null meta data. meta=" << (void*)msg->metaData << std::endl;
|
||||||
msgsToDel[msg->grpId].push_back(msg->msgId);
|
msgsToDel[msg->grpId].push_back(msg->msgId);
|
||||||
}
|
}
|
||||||
else if(!msg->metaData->mAuthorId.isNull() && subscribed_groups.find(msg->metaData->mGroupId)!=subscribed_groups.end())
|
else if(!msg->metaData->mAuthorId.isNull() && subscribed_groups.find(msg->metaData->mGroupId)!=subscribed_groups.end())
|
||||||
{
|
{
|
||||||
std::cerr << "TimeStamping message authors' key ID " << msg->metaData->mAuthorId << " in message " << msg->msgId << ", group ID " << msg->grpId<< std::endl;
|
#ifdef GXSUTIL_DEBUG
|
||||||
mGixs->timeStampKey(msg->metaData->mAuthorId) ;
|
std::cerr << "TimeStamping message authors' key ID " << msg->metaData->mAuthorId << " in message " << msg->msgId << ", group ID " << msg->grpId<< std::endl;
|
||||||
}
|
#endif
|
||||||
|
used_gxs_ids.insert(msg->metaData->mAuthorId) ;
|
||||||
|
}
|
||||||
|
|
||||||
delete msg;
|
delete msg;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
mDs->removeMsgs(msgsToDel);
|
mDs->removeMsgs(msgsToDel);
|
||||||
|
|
||||||
RsStackMutex stack(mIntegrityMutex);
|
RsStackMutex stack(mIntegrityMutex);
|
||||||
mDone = true;
|
mDone = true;
|
||||||
|
|
||||||
std::vector<RsGxsGroupId>::iterator grpIt;
|
std::vector<RsGxsGroupId>::iterator grpIt;
|
||||||
for(grpIt = grpsToDel.begin(); grpIt != grpsToDel.end(); ++grpIt)
|
for(grpIt = grpsToDel.begin(); grpIt != grpsToDel.end(); ++grpIt)
|
||||||
{
|
{
|
||||||
mDeletedGrps.push_back(*grpIt);
|
mDeletedGrps.push_back(*grpIt);
|
||||||
}
|
}
|
||||||
mDeletedMsgs = msgsToDel;
|
mDeletedMsgs = msgsToDel;
|
||||||
|
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << "At end of pass, this is the list used GXS ids: " << std::endl;
|
||||||
|
std::cerr << " requesting them to GXS identity service to enforce loading." << std::endl;
|
||||||
|
#endif
|
||||||
|
|
||||||
|
std::list<RsPeerId> connected_friends ;
|
||||||
|
rsPeers->getOnlineList(connected_friends) ;
|
||||||
|
|
||||||
|
std::vector<RsGxsId> gxs_ids ;
|
||||||
|
|
||||||
|
for(std::set<RsGxsId>::const_iterator it(used_gxs_ids.begin());it!=used_gxs_ids.end();++it)
|
||||||
|
{
|
||||||
|
gxs_ids.push_back(*it) ;
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << " " << *it << std::endl;
|
||||||
|
#endif
|
||||||
|
}
|
||||||
|
int nb_requested_not_in_cache = 0;
|
||||||
|
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << " issuing random get on friends for non existing IDs" << std::endl;
|
||||||
|
#endif
|
||||||
|
|
||||||
|
for(;nb_requested_not_in_cache<MAX_GXS_IDS_REQUESTS_NET && gxs_ids.size()>0;)
|
||||||
|
{
|
||||||
|
uint32_t n = RSRandom::random_u32() % gxs_ids.size() ;
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << " requesting ID " << gxs_ids[n] ;
|
||||||
|
#endif
|
||||||
|
|
||||||
|
if(!mGixs->haveKey(gxs_ids[n])) // checks if we have it already in the cache (conservative way to ensure that we atually have it)
|
||||||
|
{
|
||||||
|
mGixs->requestKey(gxs_ids[n],connected_friends);
|
||||||
|
|
||||||
|
++nb_requested_not_in_cache ;
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << " ... from cache/net" << std::endl;
|
||||||
|
#endif
|
||||||
|
}
|
||||||
|
else
|
||||||
|
std::cerr << " ... already in cache" << std::endl;
|
||||||
|
|
||||||
|
gxs_ids[n] = gxs_ids[gxs_ids.size()-1] ;
|
||||||
|
gxs_ids.pop_back() ;
|
||||||
|
}
|
||||||
|
#ifdef GXSUTIL_DEBUG
|
||||||
|
std::cerr << " total actual cache requests: "<< nb_requested_not_in_cache << std::endl;
|
||||||
|
#endif
|
||||||
|
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue