Merge pull request #357 from jolavillette/fix/gxs-grpmeta-batch-write

Batch group meta data updates in a single DB transaction

src/gxs/rsdataservice.cc

@@ -1583,6 +1583,56 @@ int RsDataService::updateGroupMetaData(const GrpLocMetaData& meta)
     return 0;
 }
 
+int RsDataService::updateGroupMetaData(const std::vector<GrpLocMetaData>& 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

src/gxs/rsdataservice.h

@@ -255,6 +255,8 @@ public:
      */
     int updateGroupMetaData(const GrpLocMetaData &meta) override;
 
+    int updateGroupMetaData(const std::vector<GrpLocMetaData>& metaList) override;
+
     /*!
      * Completely clear out data stored in
      * and returns this to a state

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<GrpLocMetaData>& metaList) = 0;
+
     virtual int updateGroupKeys(const RsGxsGroupId& grpId,const RsTlvSecurityKeySet& keys,uint32_t subscribed_flags) = 0 ;
 
     /*!

src/gxs/rsgenexchange.cc

@@ -2205,6 +2205,13 @@ void RsGenExchange::processGrpMetaChanges()
 
     std::list<RsGxsGroupId> grpChanged;
 
+    // Phase 1: process the masks, and collect the entries to write.
+
+    std::vector<GrpLocMetaData> toWrite;
+    std::vector<uint32_t> writeTokens;
+    toWrite.reserve(metaMap.size());
+    writeTokens.reserve(metaMap.size());
+
     std::map<uint32_t, GrpLocMetaData>::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;i<toWrite.size();++i)
+            if(all_ok)
+            {
+                mDataAccess->updatePublicRequestStatus(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);