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 @@ private: */ 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 @@ private: 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; };