From f74d4f64756e1925d9ecb20dbe8bc8b73a64f47e Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 29 Jul 2026 19:14:38 +0200 Subject: [PATCH 1/7] gxs: warm up the message meta caches with one scan instead of one query per group Reading the meta of a whole group runs SELECT ... WHERE grpId=?, which INDEX_MESSAGES_GRPID serves with one row lookup per message. Since the payload blob lives in the same row, those lookups are scattered over the whole file: warming up the cache of N groups costs N passes of random I/O over a database that is hundreds of megabytes. When more than one group still needs a cold full read, read the meta of every message in a single sequential scan instead and fill every per-group cache from it. The file is then read in physical order, and the cost no longer grows with 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, 20 groups), cold cache: 20 per-group queries 29449 ms one sequential scan 2146 ms 13.7x and the scan does not get more expensive as groups are added, where the per-group path grows linearly with them. On a node subscribed to hundreds of forums this is the difference between tens of seconds of startup and a fixed couple of seconds. Nothing else changes: same columns, same cache contents, same values returned. Callers and public API are untouched, and no database schema or format is modified. Trade-off: the scan fills the cache for groups that were not requested yet. That is the same memory the cache reaches as soon as those groups are browsed, but it is reached up front rather than progressively. The scan reports itself through the existing opt-in profiler, so the gain is verifiable on a real profile rather than taken on trust: GXS-PROF loadAllMsgMetaInOneScan db=gxsforums_db groups=571 metas=48213 in 2100ms Stacked on the channel loading branch: both reshape the same function, and this one reuses the profiler introduced there. Co-Authored-By: Claude Opus 5 (1M context) --- src/gxs/rsdataservice.cc | 71 ++++++++++++++++++++++++++++++++++++++++ src/gxs/rsdataservice.h | 16 +++++++++ 2 files changed, 87 insertions(+) diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index c061e37e7..9be4dadc7 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -130,6 +130,8 @@ 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; initialise(isNewDatabase); @@ -1290,6 +1292,54 @@ void RsDataService::locked_retrieveMessages(RetroCursor *c, std::vectorsqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, "", ""); + + if(!c) + { + RsErr() << __PRETTY_FUNCTION__ << ": failed to query message meta data" << std::endl; + return; + } + + bool valid = c->moveToFirst(); + + while(valid) + { + auto m = locked_getMsgMeta(*c, 0); + + if(m != nullptr) + mMsgMetaDataCache[m->mGroupId].updateMeta(m->mMsgId, m); + + valid = c->moveToNext(); + } + + delete c; + + // 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; +} + int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta) { RsStackMutex stack(mDbMutex); @@ -1299,6 +1349,25 @@ 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. See locked_loadAllMsgMetaInOneScan(). + if(mUseCache && !mMsgMetaDataCache_ContainsAllDatabase) + { + 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) + locked_loadAllMsgMetaInOneScan(); + } + for(auto mit(reqIds.begin()); mit != reqIds.end(); ++mit) { @@ -1316,6 +1385,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) diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index 0424e4e14..e5e33f836 100644 --- a/src/gxs/rsdataservice.h +++ b/src/gxs/rsdataservice.h @@ -308,6 +308,16 @@ class RsDataService : public RsGeneralDataService */ void locked_retrieveMsgMetaList(RetroCursor* c, std::vector > &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. Called when more than one group still needs a cold full read. + */ + void locked_loadAllMsgMetaInOneScan(); + /*! * Retrieves all the grp meta results from a cursor * @param c cursor to result set @@ -468,6 +478,12 @@ class RsDataService : public RsGeneralDataService t_MetaDataCache mGrpMetaDataCache; std::map > mMsgMetaDataCache; + /// True once locked_loadAllMsgMetaInOneScan() has run: no point scanning twice. + bool mMsgMetaDataCache_ContainsAllDatabase; + + /// Number of whole-group cold reads done so far, to decide when scanning wins. + uint32_t mMsgMetaDataCache_ColdFullReads; + bool mUseCache; }; From d02b5dd2b3c1132b1fc4900f8217ef5142932539 Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 5 Aug 2026 11:27:40 +0200 Subject: [PATCH 2/7] gxs: run the msg meta warm-up scan on its own thread, in mutex-released slices The warm-up scan was triggered synchronously inside retrieveGxsMsgMetaData by the second cold whole-group request, and ran under mDbMutex in one go. Cold page cache, it was measured at up to 57 s on a real gxsforums_db (235 MB): the caller -- possibly asking for a handful of metas from one group -- and every other reader of the service froze for that long at startup. Keep the trigger and the sequential scan, but run it on a dedicated thread in slices of 4096 rows by increasing rowid, taking mDbMutex only for the duration of one slice so readers interleave. Until the scan completes, cold groups keep being served by the indexed per-group query. Messages stored while the scan runs are cached by storeMessage() itself, so rowid reuse after deletions cannot leave a hole. The thread is joined in the destructor before the DB is closed. Co-Authored-By: Claude Fable 5 --- src/gxs/rsdataservice.cc | 102 ++++++++++++++++++++++++++++++--------- src/gxs/rsdataservice.h | 21 ++++++-- 2 files changed, 97 insertions(+), 26 deletions(-) diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index 9be4dadc7..cef80630a 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -26,6 +26,7 @@ * #define RS_DATA_SERVICE_DEBUG_CACHE 1 ****/ +#include #include #include #include @@ -132,6 +133,8 @@ RsDataService::RsDataService(const std::string &serviceDir, const std::string &d mUseCache = true; mMsgMetaDataCache_ContainsAllDatabase = false; mMsgMetaDataCache_ColdFullReads = 0; + mMsgMetaWarmupStarted = false; + mMsgMetaWarmupStop = false; initialise(isNewDatabase); @@ -215,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; } @@ -1292,7 +1301,7 @@ void RsDataService::locked_retrieveMessages(RetroCursor *c, std::vectorsqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, "", ""); + static const uint32_t WARMUP_SLICE_ROWS = 4096; - if(!c) - { - RsErr() << __PRETTY_FUNCTION__ << ": failed to query message meta data" << std::endl; - return; - } + std::list columns(mMsgMetaColumns); + columns.push_front("rowid"); - bool valid = c->moveToFirst(); + int64_t last_rowid = 0; + bool done = false; - while(valid) + while(!done && !mMsgMetaWarmupStop) { - auto m = locked_getMsgMeta(*c, 0); + { + RsStackMutex stack(mDbMutex); - if(m != nullptr) - mMsgMetaDataCache[m->mGroupId].updateMeta(m->mMsgId, m); + if(mMsgMetaDataCache_ContainsAllDatabase) + return; - valid = c->moveToNext(); - } + RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, columns, + "rowid > " + std::to_string(last_rowid), + "rowid LIMIT " + std::to_string(WARMUP_SLICE_ROWS)); + + if(!c) + { + RsErr() << __PRETTY_FUNCTION__ << ": failed to query message meta data. Giving up cache warm-up." << std::endl; + return; + } - delete c; + uint32_t n_rows = 0; + bool valid = c->moveToFirst(); - // 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); + 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; - mMsgMetaDataCache_ContainsAllDatabase = true; + if(n_rows < WARMUP_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; + } + } + + // Let the readers waiting on mDbMutex in between two slices. + if(!done) + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + } } int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta) @@ -1355,8 +1406,10 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes // 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. See locked_loadAllMsgMetaInOneScan(). - if(mUseCache && !mMsgMetaDataCache_ContainsAllDatabase) + // 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; @@ -1365,7 +1418,10 @@ int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaRes ++cold_groups; if(cold_groups + mMsgMetaDataCache_ColdFullReads > 1) - locked_loadAllMsgMetaInOneScan(); + { + mMsgMetaWarmupStarted = true; + mMsgMetaWarmupThread = std::thread(&RsDataService::msgMetaWarmupThreadBody, this); + } } for(auto mit(reqIds.begin()); mit != reqIds.end(); ++mit) diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index e5e33f836..eadd9f1b5 100644 --- a/src/gxs/rsdataservice.h +++ b/src/gxs/rsdataservice.h @@ -22,6 +22,9 @@ #ifndef RSDATASERVICE_H #define RSDATASERVICE_H +#include +#include + #include "gxs/rsgds.h" #include "util/retrodb.h" @@ -314,9 +317,14 @@ class RsDataService : public RsGeneralDataService * * 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. Called when more than one group still needs a cold full read. + * 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 locked_loadAllMsgMetaInOneScan(); + void msgMetaWarmupThreadBody(); /*! * Retrieves all the grp meta results from a cursor @@ -478,12 +486,19 @@ class RsDataService : public RsGeneralDataService t_MetaDataCache mGrpMetaDataCache; std::map > mMsgMetaDataCache; - /// True once locked_loadAllMsgMetaInOneScan() has run: no point scanning twice. + /// 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 mMsgMetaWarmupStop; + bool mUseCache; }; From b26e445273ec8ed739bfa7a916cb335af01b47c2 Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 5 Aug 2026 13:24:48 +0200 Subject: [PATCH 3/7] gxs: log a summary line when the msg meta cache warm-up completes One line per database (rows, slices, duration) so the background warm-up can be observed and validated from the logs. Co-Authored-By: Claude Fable 5 --- src/gxs/rsdataservice.cc | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index cef80630a..0504831f2 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -1335,6 +1335,9 @@ void RsDataService::msgMetaWarmupThreadBody() 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) { @@ -1372,6 +1375,9 @@ void RsDataService::msgMetaWarmupThreadBody() delete c; + total_rows += n_rows; + ++n_slices; + if(n_rows < WARMUP_SLICE_ROWS) { // Last slice. Every group of this database now holds all its @@ -1389,6 +1395,13 @@ void RsDataService::msgMetaWarmupThreadBody() if(!done) 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::steady_clock::now() - t0).count() + << " ms" << std::endl; } int RsDataService::retrieveGxsMsgMetaData(const GxsMsgReq& reqIds, GxsMsgMetaResult& msgMeta) From 4a61ccee0097605bf00575cad474f916cc48d34d Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 5 Aug 2026 13:53:52 +0200 Subject: [PATCH 4/7] gxs: adapt the warm-up slice size to a fixed per-slice time budget A fixed 4096-row slice held mDbMutex for ~10 s on a cold large-row database (23492 forum metas warmed in 6 slices of ~10 s each), stalling single-group readers for that long -- the very stall the background scan exists to avoid. Start at 256 rows and rescale each slice towards a 250 ms target, clamped to [64, 4096] rows. Co-Authored-By: Claude Fable 5 --- src/gxs/rsdataservice.cc | 33 +++++++++++++++++++++++++++++---- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index 0504831f2..db6f12c6e 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -1328,7 +1328,15 @@ void RsDataService::msgMetaWarmupThreadBody() // a deletion) are covered; until the scan completes, cold groups keep // being served by the indexed per-group query. - static const uint32_t WARMUP_SLICE_ROWS = 4096; + // 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 columns(mMsgMetaColumns); columns.push_front("rowid"); @@ -1341,6 +1349,8 @@ void RsDataService::msgMetaWarmupThreadBody() while(!done && !mMsgMetaWarmupStop) { + auto slice_t0 = std::chrono::steady_clock::now(); + { RsStackMutex stack(mDbMutex); @@ -1349,7 +1359,7 @@ void RsDataService::msgMetaWarmupThreadBody() RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, columns, "rowid > " + std::to_string(last_rowid), - "rowid LIMIT " + std::to_string(WARMUP_SLICE_ROWS)); + "rowid LIMIT " + std::to_string(slice_rows)); if(!c) { @@ -1378,7 +1388,7 @@ void RsDataService::msgMetaWarmupThreadBody() total_rows += n_rows; ++n_slices; - if(n_rows < WARMUP_SLICE_ROWS) + 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 @@ -1391,9 +1401,24 @@ void RsDataService::msgMetaWarmupThreadBody() } } - // Let the readers waiting on mDbMutex in between two slices. if(!done) + { + // Rescale the next slice towards the per-slice time target. + int64_t slice_ms = std::chrono::duration_cast( + 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(WARMUP_SLICE_MAX_ROWS, + std::max(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) From b6285a8d9a46262cfeae19cdb1c2e4c53a77677c Mon Sep 17 00:00:00 2001 From: jolavillette Date: Sat, 22 Aug 2026 19:01:58 +0200 Subject: [PATCH 5/7] gxs: do not retrieve all group metas in publishMsgs() when nothing was published publishMsgs() builds a grpMetas map out of msgChangeMap and then calls RsDataService::retrieveGxsGrpMetaData() unconditionally. When no message was published - which is the case on virtually every tick - that map is empty, and an empty map means "retrieve ALL groups" for the data service: a full group table scan (with the associated decryption cost when the grp meta cache is cold). That happens with mGenMtx held, so the whole GXS service is frozen for the duration, and so is any GUI call that needs mGenMtx. Observed on a client with 878 forums: a msg meta retrieval held the data service mutex for 35 s, publishMsgs() blocked on it while holding mGenMtx, and the GUI thread hung 27 s inside RsGenExchange::getDefaultSyncPeriod() while building the forums help string. Guard the call the same way processRecvdMessages() already does. --- src/gxs/rsgenexchange.cc | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index 15edd3208..9eea26153 100644 --- a/src/gxs/rsgenexchange.cc +++ b/src/gxs/rsgenexchange.cc @@ -2539,7 +2539,12 @@ void RsGenExchange::publishMsgs() for(auto grpit:msgChangeMap) grpMetas.insert(std::make_pair(grpit.first, std::make_shared())); - 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) { From 49f07b8af075e2abd8974c234207195e24a98484 Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 29 Jul 2026 14:18:00 +0200 Subject: [PATCH 6/7] gxs: parallelise message deserialisation in getMsgData with OpenMP The database work is finished when the loop starts; what remains is pure in-memory decoding, one item per message, independent of each other. Run it through an OpenMP parallel for: results land in a pre-sized array so the loop shares no mutable state, and are merged serially afterwards, which keeps the output order stable and reports deserialisation errors from a single thread instead of interleaving lines from the pool. mSerialiser is used concurrently and must remain stateless/re-entrant; its declaration now says so. Builds without OpenMP ignore the pragma and run the loop serially, unchanged. Restacked on top of perf/gxs-channel-loading, whose version filtering and move semantics reshape the same loop: the two changes are complementary (fewer items to decode, then decoded in parallel). Co-Authored-By: Claude Opus 5 (1M context) --- src/gxs/rsgenexchange.cc | 44 ++++++++++++++++++++++++++++++---------- src/gxs/rsgenexchange.h | 2 +- 2 files changed, 34 insertions(+), 12 deletions(-) diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index 9eea26153..19d0b8f73 100644 --- a/src/gxs/rsgenexchange.cc +++ b/src/gxs/rsgenexchange.cc @@ -1581,10 +1581,24 @@ bool RsGenExchange::getMsgData(uint32_t token, GxsMsgDataMap &msgItems) const RsGxsGroupId& grpId = mit->first; std::vector& gxsMsgItems = msgItems[grpId]; std::vector& nxsMsgsV = mit->second; - std::vector::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 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(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; } } diff --git a/src/gxs/rsgenexchange.h b/src/gxs/rsgenexchange.h index 257b7bc5f..fa4ec0d5a 100644 --- a/src/gxs/rsgenexchange.h +++ b/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; From 3eae1832ec89532308fe82c19b22afc21c820bcf Mon Sep 17 00:00:00 2001 From: jolavillette Date: Wed, 29 Jul 2026 14:18:38 +0200 Subject: [PATCH 7/7] build: enable OpenMP for the parallel GXS deserialisation Adds -fopenmp where the qmake build compiles and links libretroshare. The previous version of this change also dropped rs_deep_forums_index from the xapian link condition in use_libretroshare.pri; that was unrelated and would have broken deep-forum-index builds, so it is not carried over. CMake builds do not set the flag yet: there the pragma is ignored and the loop runs serially, which is correct, just not parallel. Co-Authored-By: Claude Opus 5 (1M context) --- src/libretroshare.pro | 4 ++++ src/use_libretroshare.pri | 2 ++ 2 files changed, 6 insertions(+) diff --git a/src/libretroshare.pro b/src/libretroshare.pro index b5c015375..d055e8738 100644 --- a/src/libretroshare.pro +++ b/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 diff --git a/src/use_libretroshare.pri b/src/use_libretroshare.pri index 85377d042..9706c332b 100644 --- a/src/use_libretroshare.pri +++ b/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 {