diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index c061e37e7..db6f12c6e 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 @@ -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 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::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) + 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) { 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) diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index 0424e4e14..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" @@ -308,6 +311,21 @@ 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. 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 mGrpMetaDataCache; std::map > 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 mMsgMetaWarmupStop; + bool mUseCache; }; diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index 15edd3208..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; } } @@ -2539,7 +2561,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) { 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; 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 {