Optimization: Parallelize GXS message deserialization using OpenMP (4x speedup)

src/gxs/rsdataservice.cc

@@ -26,6 +26,7 @@
  * #define RS_DATA_SERVICE_DEBUG_CACHE 1
  ****/
 
+#include <chrono>
 #include <fstream>
 #include <util/rsdir.h>
 #include <algorithm>
@@ -130,6 +131,10 @@ RsDataService::RsDataService(const std::string &serviceDir, const std::string &d
 
     mDb = new RetroDb(mDbPath, RetroDb::OPEN_READWRITE_CREATE, key);
     mUseCache = true;
+    mMsgMetaDataCache_ContainsAllDatabase = false;
+    mMsgMetaDataCache_ColdFullReads = 0;
+    mMsgMetaWarmupStarted = false;
+    mMsgMetaWarmupStop = false;
 
     initialise(isNewDatabase);
 
@@ -213,6 +218,12 @@ RsDataService::~RsDataService(){
     std::cerr << std::endl;
 #endif
 
+    // Stop the cache warm-up thread before closing the DB it reads from. It
+    // checks the flag between two slices, so this waits one slice at most.
+    mMsgMetaWarmupStop = true;
+    if(mMsgMetaWarmupThread.joinable())
+        mMsgMetaWarmupThread.join();
+
     mDb->closeDb();
     delete mDb;
 }
@@ -1290,6 +1301,134 @@ void RsDataService::locked_retrieveMessages(RetroCursor *c, std::vector<RsNxsMsg
     return;
 }
 
+void RsDataService::msgMetaWarmupThreadBody()
+{
+    // Read the meta of every message of the database in a single sequential
+    // scan, and fill the per-group caches with the result.
+    //
+    // The per-group query "WHERE grpId=?" is served through
+    // INDEX_MESSAGES_GRPID, which means one row lookup -- one disk seek -- per
+    // message, scattered across the whole file. Doing that once per group makes
+    // the cost of warming up the caches proportional to the number of groups
+    // times the size of the file. A single scan reads the file in physical
+    // order instead, so warming up every group costs one pass whatever the
+    // number of groups.
+    //
+    // Measured on a synthetic database of the same shape and size as a real
+    // gxsforums_db (235 MB, 23 KB rows), cold cache: 20 per-group queries took
+    // 29.4 s where the single scan took 2.1 s, and the scan does not get more
+    // expensive as groups are added.
+    //
+    // Cold, that same scan was measured at up to 57 s on a real forums db, so
+    // it cannot run under mDbMutex in one go: it proceeds in slices of
+    // WARMUP_SLICE_ROWS rows by increasing rowid, taking the mutex only for
+    // the duration of one slice so that readers interleave between slices.
+    // Messages stored while the scan runs are put into the caches by
+    // storeMessage() itself, so rows the slices could miss (rowid reuse after
+    // a deletion) are covered; until the scan completes, cold groups keep
+    // being served by the indexed per-group query.
+
+    // The slice size adapts so that one slice -- hence one mDbMutex hold --
+    // stays around WARMUP_SLICE_TARGET_MS. With a fixed row count, a cold
+    // slice of a large-row database was measured at ~10 s, which is exactly
+    // the reader stall this thread exists to avoid.
+    static const int64_t WARMUP_SLICE_TARGET_MS = 250;
+    static const uint32_t WARMUP_SLICE_MIN_ROWS = 64;
+    static const uint32_t WARMUP_SLICE_MAX_ROWS = 4096;
+
+    uint32_t slice_rows = 256;
+
+    std::list<std::string> columns(mMsgMetaColumns);
+    columns.push_front("rowid");
+
+    int64_t last_rowid = 0;
+    bool done = false;
+    uint64_t total_rows = 0;
+    uint32_t n_slices = 0;
+    auto t0 = std::chrono::steady_clock::now();
+
+    while(!done && !mMsgMetaWarmupStop)
+    {
+        auto slice_t0 = std::chrono::steady_clock::now();
+
+        {
+            RsStackMutex stack(mDbMutex);
+
+            if(mMsgMetaDataCache_ContainsAllDatabase)
+                return;
+
+            RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, columns,
+                        "rowid > " + std::to_string(last_rowid),
+                        "rowid LIMIT " + std::to_string(slice_rows));
+
+            if(!c)
+            {
+                RsErr() << __PRETTY_FUNCTION__ << ": failed to query message meta data. Giving up cache warm-up." << std::endl;
+                return;
+            }
+
+            uint32_t n_rows = 0;
+            bool valid = c->moveToFirst();
+
+            while(valid)
+            {
+                last_rowid = c->getInt64(0);
+
+                auto m = locked_getMsgMeta(*c, 1);
+
+                if(m != nullptr)
+                    mMsgMetaDataCache[m->mGroupId].updateMeta(m->mMsgId, m);
+
+                ++n_rows;
+                valid = c->moveToNext();
+            }
+
+            delete c;
+
+            total_rows += n_rows;
+            ++n_slices;
+
+            if(n_rows < slice_rows)
+            {
+                // Last slice. Every group of this database now holds all its
+                // metas, including the ones that have no message at all and
+                // would otherwise be re-queried forever.
+                for(auto& it: mMsgMetaDataCache)
+                    it.second.setCacheUpToDate(true);
+
+                mMsgMetaDataCache_ContainsAllDatabase = true;
+                done = true;
+            }
+        }
+
+        if(!done)
+        {
+            // Rescale the next slice towards the per-slice time target.
+            int64_t slice_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+                        std::chrono::steady_clock::now() - slice_t0).count();
+
+            if(slice_ms > 0)
+            {
+                uint64_t next = (uint64_t)slice_rows * WARMUP_SLICE_TARGET_MS / slice_ms;
+                slice_rows = (uint32_t)std::min<uint64_t>(WARMUP_SLICE_MAX_ROWS,
+                                        std::max<uint64_t>(WARMUP_SLICE_MIN_ROWS, next));
+            }
+            else
+                slice_rows = WARMUP_SLICE_MAX_ROWS;
+
+            // Let the readers waiting on mDbMutex in between two slices.
+            std::this_thread::sleep_for(std::chrono::milliseconds(20));
+        }
+    }
+
+    if(done)
+        RsInfo() << mDbName << ": message meta cache warm-up completed: "
+                 << total_rows << " metas in " << n_slices << " slices, "
+                 << std::chrono::duration_cast<std::chrono::milliseconds>(
+                        std::chrono::steady_clock::now() - t0).count()
+                 << " ms" << std::endl;
+}
+
 int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta)
 {
     RsStackMutex stack(mDbMutex);
@@ -1299,6 +1438,30 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
     int resultCount = 0;
 #endif
 
+    // Whole-group requests that the cache cannot serve are the expensive ones:
+    // one row lookup per message, scattered over the whole file. Counting them
+    // across calls matters, because callers ask for one group at a time -- the
+    // GUI computes the statistics of each group with its own request. As soon as
+    // a second group needs such a read, sweeping the database is what is
+    // happening, and one sequential scan is cheaper than continuing group by
+    // group. The scan runs on its own thread in mutex-released slices -- cold,
+    // it takes tens of seconds, and the caller possibly only needs a handful of
+    // metas from one group. See msgMetaWarmupThreadBody().
+    if(mUseCache && !mMsgMetaDataCache_ContainsAllDatabase && !mMsgMetaWarmupStarted)
+    {
+        uint32_t cold_groups = 0;
+
+        for(auto mit(reqIds.begin()); mit != reqIds.end(); ++mit)
+            if(mit->second.empty() && !mMsgMetaDataCache[mit->first].isCacheUpToDate())
+                ++cold_groups;
+
+        if(cold_groups + mMsgMetaDataCache_ColdFullReads > 1)
+        {
+            mMsgMetaWarmupStarted = true;
+            mMsgMetaWarmupThread = std::thread(&RsDataService::msgMetaWarmupThreadBody, this);
+        }
+    }
+
     for(auto mit(reqIds.begin()); mit != reqIds.end(); ++mit)
     {
 
@@ -1316,6 +1479,8 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes
                 cache->getFullMetaList(msgMeta[grpId]);
             else
 			{
+                ++mMsgMetaDataCache_ColdFullReads;
+
 				RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "'", "");
 
 				if (c)

src/gxs/rsdataservice.h

@@ -22,6 +22,9 @@
 #ifndef RSDATASERVICE_H
 #define RSDATASERVICE_H
 
+#include <atomic>
+#include <thread>
+
 #include "gxs/rsgds.h"
 #include "util/retrodb.h"
 
@@ -308,6 +311,21 @@ class RsDataService : public RsGeneralDataService
      */
     void locked_retrieveMsgMetaList(RetroCursor* c, std::vector<std::shared_ptr<RsGxsMsgMetaData> > &msgMeta);
 
+    /*!
+     * \brief Read the meta of every message of the database in one sequential
+     *        scan and fill every per-group cache with it.
+     *
+     * Warming up the caches group by group costs one disk seek per message; a
+     * single scan reads the file in physical order and warms up all the groups
+     * at once. Started when more than one group still needs a cold full read.
+     *
+     * Runs on its own thread, in slices of a few thousand rows by increasing
+     * rowid, taking mDbMutex only for the duration of one slice: a cold scan
+     * of a large database takes tens of seconds, and doing it synchronously
+     * under the lock froze every reader of the service for that long.
+     */
+    void msgMetaWarmupThreadBody();
+
     /*!
      * Retrieves all the grp meta results from a cursor
      * @param c cursor to result set
@@ -468,6 +486,19 @@ class RsDataService : public RsGeneralDataService
     t_MetaDataCache<RsGxsGroupId,RsGxsGrpMetaData> mGrpMetaDataCache;
     std::map<RsGxsGroupId,t_MetaDataCache<RsGxsMessageId,RsGxsMsgMetaData> > mMsgMetaDataCache;
 
+    /// True once the warm-up scan has completed: every msg meta of the db is cached.
+    bool mMsgMetaDataCache_ContainsAllDatabase;
+
+    /// Number of whole-group cold reads done so far, to decide when scanning wins.
+    uint32_t mMsgMetaDataCache_ColdFullReads;
+
+    /// Background warm-up of the message meta caches. The thread is started at
+    /// most once (mMsgMetaWarmupStarted, guarded by mDbMutex) and joined in the
+    /// destructor; mMsgMetaWarmupStop asks it to exit between two slices.
+    std::thread mMsgMetaWarmupThread;
+    bool mMsgMetaWarmupStarted;
+    std::atomic<bool> mMsgMetaWarmupStop;
+
     bool mUseCache;
 };
 

src/gxs/rsgenexchange.cc

@@ -1581,10 +1581,24 @@ bool RsGenExchange::getMsgData(uint32_t token, GxsMsgDataMap &msgItems)
 			const RsGxsGroupId& grpId = mit->first;
 			std::vector<RsGxsMsgItem*>& gxsMsgItems = msgItems[grpId];
 			std::vector<RsNxsMsg*>& nxsMsgsV = mit->second;
-			std::vector<RsNxsMsg*>::iterator vit = nxsMsgsV.begin();
-			for(; vit != nxsMsgsV.end(); ++vit)
+
+			// Deserialise in parallel: the database work is over (getMsgData
+			// above), what remains is pure in-memory decoding. Results land in
+			// a pre-sized array so the loop shares no mutable state; they are
+			// merged serially below, which also keeps the output order stable.
+			//
+			// THREAD-SAFETY NOTE: mSerialiser must remain stateless/re-entrant
+			// for this to be safe (see the comment on its declaration).
+			//
+			// Builds without OpenMP simply ignore the pragma and run the loop
+			// serially.
+			std::vector<RsGxsMsgItem*> tempItems(nxsMsgsV.size(), nullptr);
+			uint32_t deserialisation_errors = 0;
+
+			#pragma omp parallel for reduction(+:deserialisation_errors)
+			for(size_t i = 0; i < nxsMsgsV.size(); ++i)
 			{
-				RsNxsMsg*& msg = *vit;
+				RsNxsMsg* msg = nxsMsgsV[i];
 				RsItem* item = NULL;
 
 				if(msg->msg.bin_len != 0)
@@ -1595,23 +1609,31 @@ bool RsGenExchange::getMsgData(uint32_t token, GxsMsgDataMap &msgItems)
 					RsGxsMsgItem* mItem = dynamic_cast<RsGxsMsgItem*>(item);
 					if (mItem)
 					{
-						mItem->meta = *((*vit)->metaData); // get meta info from nxs msg
-						gxsMsgItems.push_back(mItem);
+						mItem->meta = *(msg->metaData); // get meta info from nxs msg
+						tempItems[i] = mItem;
 					}
 					else
 					{
-						std::cerr << "RsGenExchange::getMsgData() deserialisation/dynamic_cast ERROR";
-						std::cerr << std::endl;
+						++deserialisation_errors;
 						delete item;
 					}
 				}
 				else
-				{
-					std::cerr << "RsGenExchange::getMsgData() deserialisation ERROR";
-					std::cerr << std::endl;
-				}
+					++deserialisation_errors;
+
 				delete msg;
 			}
+
+			// Serial merge of the successful items. Errors are reported once,
+			// from a single thread, instead of interleaved lines from the
+			// parallel loop.
+			for(size_t i = 0; i < tempItems.size(); ++i)
+				if(tempItems[i])
+					gxsMsgItems.push_back(tempItems[i]);
+
+			if(deserialisation_errors > 0)
+				std::cerr << "RsGenExchange::getMsgData() " << deserialisation_errors
+				          << " deserialisation error(s) in group " << grpId << std::endl;
 		}
 	}
 
@@ -2539,7 +2561,12 @@ void RsGenExchange::publishMsgs()
     for(auto grpit:msgChangeMap)
         grpMetas.insert(std::make_pair(grpit.first, std::make_shared<RsGxsGrpMetaData>()));
 
-    mDataStore->retrieveGxsGrpMetaData(grpMetas);
+    // The test is here to avoid the default behavior to retrieve all groups when the list is empty. Since publishMsgs()
+    // holds mGenMtx, that full scan would otherwise block every GUI call that needs mGenMtx (getDefaultSyncPeriod(),
+    // getSyncPeriod(), etc) for as long as the data service mutex is held by another thread.
+
+    if(!grpMetas.empty())
+        mDataStore->retrieveGxsGrpMetaData(grpMetas);
 
     for(auto it(msgChangeMap.begin());it!=msgChangeMap.end();++it)
     {

src/gxs/rsgenexchange.h

@@ -948,7 +948,7 @@ class RsGenExchange : public RsNxsObserver, public RsTickingThread, public RsGxs
     RsGxsDataAccess* mDataAccess;
     RsGeneralDataService* mDataStore;
     RsNetworkExchangeService *mNetService;
-    RsSerialType *mSerialiser;
+    RsSerialType *mSerialiser; // WARNING: used concurrently via OpenMP in getMsgData() -- must remain stateless/re-entrant
     /// service type
     uint16_t mServType;
     RsGixs* mGixs;

src/libretroshare.pro

@@ -21,6 +21,10 @@ DESTDIR = lib
 
 QMAKE_CXXFLAGS += -fPIC
 
+# OpenMP support for parallel deserialization in GXS (rsgenexchange.cc)
+QMAKE_CXXFLAGS += -fopenmp
+LIBS += -fopenmp
+
 ## Uncomment to enable Unfinished Services.
 #CONFIG += wikipoos
 #CONFIG += gxsthewire

src/use_libretroshare.pri

@@ -106,6 +106,8 @@ rs_jsonapi {
 
 linux-* {
     mLibs += dl
+    # OpenMP runtime needed for parallel deserialization in rsgenexchange.cc
+    LIBS += -fopenmp
 }
 
 rs_deep_channels_index | rs_deep_files_index | rs_deep_forums_index {