diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index 88335ee8e..c061e37e7 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -1583,6 +1583,56 @@ int RsDataService::updateGroupMetaData(const GrpLocMetaData& meta) return 0; } +int RsDataService::updateGroupMetaData(const std::vector& metaList) +{ + if(metaList.empty()) + return 0; + + RsStackMutex stack(mDbMutex); + + // Persist the whole batch inside a single transaction. Without this, every + // row update is its own implicit transaction (one fsync per group), which + // freezes the calling GXS service for seconds when many updates are queued + // (measured: ~1 s per update, 86 s backlogs on the identities service). + // We hold mDbMutex for the whole span so no other statement can slip into + // the transaction. + mDb->beginTransaction(); + + int count = 0; + + for(const GrpLocMetaData& meta : metaList) + { + const RsGxsGroupId& grpId = meta.grpId; + + if(mDb->sqlUpdate(GRP_TABLE_NAME, KEY_GRP_ID + "='" + grpId.toStdString() + "'", meta.val)) + { + // If we use the cache, update the meta data immediately. + if(mUseCache) + { + RetroCursor* c = mDb->sqlQuery(GRP_TABLE_NAME, mGrpMetaColumns, "grpId='" + grpId.toStdString() + "'", ""); + + c->moveToFirst(); + + // temporarily disable the cache so that we get the value from the DB itself. + mUseCache=false; + auto meta_refreshed = locked_getGrpMeta(*c, 0); + mUseCache=true; + + if(meta_refreshed) + mGrpMetaDataCache.updateMeta(grpId,meta_refreshed); + + delete c; + } + + ++count; + } + } + + mDb->commitTransaction(); + + return count; +} + int RsDataService::updateMessageMetaData(const MsgLocMetaData& metaData) { #ifdef RS_DATA_SERVICE_DEBUG_CACHE diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index e136dc042..0424e4e14 100644 --- a/src/gxs/rsdataservice.h +++ b/src/gxs/rsdataservice.h @@ -255,6 +255,8 @@ public: */ int updateGroupMetaData(const GrpLocMetaData &meta) override; + int updateGroupMetaData(const std::vector& metaList) override; + /*! * Completely clear out data stored in * and returns this to a state diff --git a/src/gxs/rsgds.h b/src/gxs/rsgds.h index e5421e7ef..cd33f4417 100644 --- a/src/gxs/rsgds.h +++ b/src/gxs/rsgds.h @@ -266,6 +266,16 @@ public: */ virtual int updateGroupMetaData(const GrpLocMetaData& meta) = 0; + /*! + * Update a batch of group meta data entries inside a single DB + * transaction. One transaction per entry means one fsync per entry, + * which freezes GXS services for seconds when many updates are queued + * (e.g. identity usage stamps at startup). + * @param metaList the meta data entries to update + * @return the number of entries successfully updated + */ + virtual int updateGroupMetaData(const std::vector& metaList) = 0; + virtual int updateGroupKeys(const RsGxsGroupId& grpId,const RsTlvSecurityKeySet& keys,uint32_t subscribed_flags) = 0 ; /*! diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index 6b5cdf148..15edd3208 100644 --- a/src/gxs/rsgenexchange.cc +++ b/src/gxs/rsgenexchange.cc @@ -2205,6 +2205,13 @@ void RsGenExchange::processGrpMetaChanges() std::list grpChanged; + // Phase 1: process the masks, and collect the entries to write. + + std::vector toWrite; + std::vector writeTokens; + toWrite.reserve(metaMap.size()); + writeTokens.reserve(metaMap.size()); + std::map::iterator mit; for (mit = metaMap.begin(); mit != metaMap.end(); ++mit) { @@ -2215,19 +2222,13 @@ void RsGenExchange::processGrpMetaChanges() RsDbg() << " Processing GrpMetaChange for token " << token << std::endl; #endif // process mask - bool ok = processGrpMask(g.grpId, g.val); - - ok = ok && (mDataStore->updateGroupMetaData(g) == 1); - - if(ok) + if(processGrpMask(g.grpId, g.val)) { - mDataAccess->updatePublicRequestStatus(token, RsTokenService::COMPLETE); - grpChanged.push_back(g.grpId); + toWrite.push_back(g); + writeTokens.push_back(token); } else - { mDataAccess->updatePublicRequestStatus(token, RsTokenService::FAILED); - } { RS_STACK_MUTEX(mGenMtx); @@ -2238,6 +2239,27 @@ void RsGenExchange::processGrpMetaChanges() } } + // Phase 2: write the whole batch in a single DB transaction. One call per + // entry means one fsync per entry, which was measured at ~1 s each and + // freezes the service tick for minutes when a backlog accumulates. + + if(!toWrite.empty()) + { + bool all_ok = mDataStore->updateGroupMetaData(toWrite) == (int)toWrite.size(); + + if(!all_ok) + RsErr() << __PRETTY_FUNCTION__ << " some group meta updates failed in a batch of " << toWrite.size() << " entries." << std::endl; + + for(uint32_t i=0;iupdatePublicRequestStatus(writeTokens[i], RsTokenService::COMPLETE); + grpChanged.push_back(toWrite[i].grpId); + } + else + mDataAccess->updatePublicRequestStatus(writeTokens[i], RsTokenService::FAILED); + } + for(auto& groupId:grpChanged) { RS_STACK_MUTEX(mGenMtx);