diff --git a/src/gxs/rsdataservice.cc b/src/gxs/rsdataservice.cc index 61900e171..af1dd9ecd 100644 --- a/src/gxs/rsdataservice.cc +++ b/src/gxs/rsdataservice.cc @@ -1581,6 +1581,56 @@ int RsDataService::updateMessageMetaData(const MsgLocMetaData& metaData) return 0; } +int RsDataService::updateMessageMetaData(const std::vector& 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 message), which + // is what made "mark all as read" take more than an hour on a large forum. + // We hold mDbMutex for the whole span so no other statement can slip into + // the transaction. + mDb->beginTransaction(); + + int count = 0; + + for(const MsgLocMetaData& metaData : metaList) + { + const RsGxsGroupId& grpId = metaData.msgId.first; + const RsGxsMessageId& msgId = metaData.msgId.second; + + if(mDb->sqlUpdate(MSG_TABLE_NAME, KEY_GRP_ID+ "='" + grpId.toStdString() + "' AND " + KEY_MSG_ID + "='" + msgId.toStdString() + "'", metaData.val) ) + { + // If we use the cache, update the meta data immediately. + if(mUseCache) + { + RetroCursor* c = mDb->sqlQuery(MSG_TABLE_NAME, mMsgMetaColumns, KEY_GRP_ID+ "='" + grpId.toStdString() + "' AND " + KEY_MSG_ID + "='" + msgId.toStdString() + "'", ""); + + c->moveToFirst(); + + // temporarily disable the cache so that we get the value from the DB itself. + mUseCache=false; + auto meta = locked_getMsgMeta(*c, 0); + mUseCache=true; + + if(meta) + mMsgMetaDataCache[grpId].updateMeta(msgId,meta); + + delete c; + } + + ++count; + } + } + + mDb->commitTransaction(); + + return count; +} + int RsDataService::removeMsgs(const GxsMsgReq& msgIds) { RsStackMutex stack(mDbMutex); diff --git a/src/gxs/rsdataservice.h b/src/gxs/rsdataservice.h index e9554f04f..3fd2fb748 100644 --- a/src/gxs/rsdataservice.h +++ b/src/gxs/rsdataservice.h @@ -242,6 +242,13 @@ class RsDataService : public RsGeneralDataService */ int updateMessageMetaData(const MsgLocMetaData& metaData) override; + /*! + * @brief Batch variant persisting all updates in a single DB transaction. + * @param metaList The meta data items to update + * @return the number of items successfully updated + */ + int updateMessageMetaData(const std::vector& metaList) override; + /*! * @param metaData The meta data item to update * @return error code diff --git a/src/gxs/rsgds.h b/src/gxs/rsgds.h index 790a68deb..eb88512da 100644 --- a/src/gxs/rsgds.h +++ b/src/gxs/rsgds.h @@ -25,6 +25,7 @@ #include #include #include +#include #include "inttypes.h" @@ -248,6 +249,25 @@ class RsGeneralDataService */ virtual int updateMessageMetaData(const MsgLocMetaData& metaData) = 0; + /*! + * @brief Update the meta data of several messages at once. + * + * The default implementation just updates them one by one. Stores backed by + * a transactional database should override this to persist the whole batch + * in a single transaction, which is dramatically faster when marking + * thousands of messages (e.g. "mark all as read" on a large forum). + * + * @param metaList the meta data items to update + * @return the number of items successfully updated + */ + virtual int updateMessageMetaData(const std::vector& metaList) + { + int count = 0; + for(const MsgLocMetaData& m : metaList) + count += updateMessageMetaData(m); + return count; + } + /*! * @param metaData */ diff --git a/src/gxs/rsgenexchange.cc b/src/gxs/rsgenexchange.cc index c8773b91c..5621e0448 100644 --- a/src/gxs/rsgenexchange.cc +++ b/src/gxs/rsgenexchange.cc @@ -2093,6 +2093,18 @@ void RsGenExchange::processMsgMetaChanges() GxsMsgReq msgIds; + // First pass: resolve the status masks into absolute values (reads come from + // the in-memory meta cache, so this stays cheap even for thousands of + // messages) and collect every change so they can be persisted together in a + // single DB transaction below, instead of one fsync'd write per message. + std::vector updates; + updates.reserve(metaMap.size()); + + std::vector > tokenIds; + tokenIds.reserve(metaMap.size()); + + std::set failedTokens; + std::map::iterator mit; for (mit = metaMap.begin(); mit != metaMap.end(); ++mit) { @@ -2133,28 +2145,38 @@ void RsGenExchange::processMsgMetaChanges() } } - ok &= mDataStore->updateMessageMetaData(m) == 1; - uint32_t token = mit->first; + // The actual DB write is deferred to the single batched transaction + // below. Collect the resolved change and remember its token. + updates.push_back(m); + tokenIds.push_back(std::make_pair(mit->first, m.msgId)); + + if(!ok) + failedTokens.insert(mit->first); + else if(changed) + msgIds[m.msgId.first].insert(m.msgId.second); + } + + // Second pass: persist every collected change in a single transaction. + int updated = mDataStore->updateMessageMetaData(updates); + bool batchOk = (updated == static_cast(updates.size())); + + for(std::vector >::iterator it = tokenIds.begin(); it != tokenIds.end(); ++it) + { + bool ok = batchOk && (failedTokens.find(it->first) == failedTokens.end()); if(ok) - { - mDataAccess->updatePublicRequestStatus(token, RsTokenService::COMPLETE); - if (changed) - { - msgIds[m.msgId.first].insert(m.msgId.second); - } - } + mDataAccess->updatePublicRequestStatus(it->first, RsTokenService::COMPLETE); else - { - mDataAccess->updatePublicRequestStatus(token, RsTokenService::FAILED); - } + mDataAccess->updatePublicRequestStatus(it->first, RsTokenService::FAILED); - { - RS_STACK_MUTEX(mGenMtx); - mMsgNotify.insert(std::make_pair(token, m.msgId)); - } + RS_STACK_MUTEX(mGenMtx); + mMsgNotify.insert(std::make_pair(it->first, it->second)); } + // If the whole batch failed, don't emit change notifications for it. + if(!batchOk) + msgIds.clear(); + if (!msgIds.empty()) { RS_STACK_MUTEX(mGenMtx); diff --git a/src/retroshare/rsgxschannels.h b/src/retroshare/rsgxschannels.h index 4358e95c5..c9df60830 100644 --- a/src/retroshare/rsgxschannels.h +++ b/src/retroshare/rsgxschannels.h @@ -430,6 +430,22 @@ class RsGxsChannels: public RsGxsIfaceHelper, public RsGxsCommentService */ virtual bool setMessageReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) =0; + /** + * @brief Mark several messages of a channel read/unread in one batch. Blocking. + * + * Like setMessageReadStatus() but queues all the status changes at once: they + * are persisted in a single database transaction and only one event is + * emitted, so it stays cheap even for thousands of posts and is meant to be + * called from a background thread so the GUI never blocks. + * @param[in] channelId channel group identifier + * @param[in] msgIds identifiers of the posts to update + * @param[in] read true to mark as read, false to mark as unread + * @return false on error, true otherwise + */ + virtual bool setMessageReadStatus( const RsGxsGroupId& channelId, + const std::vector& msgIds, + bool read ) =0; + /** * @brief Share channel publishing key * This can be used to authorize other peers to post on the channel diff --git a/src/retroshare/rsgxsforums.h b/src/retroshare/rsgxsforums.h index 3a9181916..cbb0aab84 100644 --- a/src/retroshare/rsgxsforums.h +++ b/src/retroshare/rsgxsforums.h @@ -407,6 +407,23 @@ class RsGxsForums: public RsGxsIfaceHelper */ virtual bool markRead(const RsGxsGrpMsgIdPair& messageId, bool read) = 0; + /** + * @brief Mark several messages of a forum read/unread in one batch. Blocking. + * + * Contrary to calling markRead() in a loop, this queues all the status + * changes at once: they are persisted in a single database transaction and + * only one event is emitted. It therefore stays cheap even for thousands of + * posts (e.g. "mark all as read" on a large forum) and is meant to be called + * from a background thread so the caller (typically the GUI) never blocks. + * @param[in] forumId forum group identifier + * @param[in] msgIds identifiers of the posts to update + * @param[in] read true to mark as read, false to mark as unread + * @return false on error, true otherwise + */ + virtual bool markRead( const RsGxsGroupId& forumId, + const std::vector& msgIds, + bool read ) = 0; + /** * @brief Subscrbe to a forum. Blocking API * @jsonapi{development} diff --git a/src/retroshare/rsposted.h b/src/retroshare/rsposted.h index 412dc6f44..f304d7c41 100644 --- a/src/retroshare/rsposted.h +++ b/src/retroshare/rsposted.h @@ -402,6 +402,22 @@ class RsPosted : public RsGxsIfaceHelper, public RsGxsCommentService */ virtual bool setPostReadStatus(const RsGxsGrpMsgIdPair& msgId, bool read) = 0; + /** + * @brief Mark several posts of a board read/unread in one batch. Blocking. + * + * Like setPostReadStatus() but queues all the status changes at once: they + * are persisted in a single database transaction and only one event is + * emitted, so it stays cheap even for thousands of posts and is meant to be + * called from a background thread so the GUI never blocks. + * @param[in] boardId board group identifier + * @param[in] msgIds identifiers of the posts to update + * @param[in] read true to mark as read, false to mark as unread + * @return false on error, true otherwise + */ + virtual bool setPostReadStatus( const RsGxsGroupId& boardId, + const std::vector& msgIds, + bool read ) = 0; + enum RS_DEPRECATED RankType {TopRankType, HotRankType, NewRankType }; RS_DEPRECATED_FOR(getBoardsInfo) diff --git a/src/services/p3gxschannels.cc b/src/services/p3gxschannels.cc index e71fdb477..b22e46920 100644 --- a/src/services/p3gxschannels.cc +++ b/src/services/p3gxschannels.cc @@ -2041,6 +2041,46 @@ bool p3GxsChannels::setMessageReadStatus(const RsGxsGrpMsgIdPair &msgId, bool re return true; } +bool p3GxsChannels::setMessageReadStatus(const RsGxsGroupId& channelId, const std::vector& msgIds, bool read) +{ + if(msgIds.empty()) + return true; + + uint32_t mask = GXS_SERV::GXS_MSG_STATUS_GUI_NEW | GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + uint32_t status = read ? 0 : GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + + // Queue every status change at once. They are all drained together by a + // single processMsgMetaChanges() tick (one DB transaction), instead of one + // blocking request + one detached thread + one event per message. + std::vector tokens; + tokens.reserve(msgIds.size()); + + for(const RsGxsMessageId& msgId : msgIds) + { + uint32_t token; + setMsgStatusFlags(token, RsGxsGrpMsgIdPair(channelId, msgId), status, mask); + tokens.push_back(token); + } + + // All tokens complete in the same tick, so waiting on the last is enough. + waitToken(tokens.back(), std::chrono::milliseconds(30000)); + + RsGxsGrpMsgIdPair p; + for(uint32_t token : tokens) + acknowledgeMsg(token, p); + + // One single event for the whole batch (consumers only need the group id). + if(rsEvents) + { + auto ev = std::make_shared(); + ev->mChannelGroupId = channelId; + ev->mChannelEventCode = RsChannelEventCode::READ_STATUS_CHANGED; + rsEvents->postEvent(ev); + } + + return true; +} + bool p3GxsChannels::shareChannelKeys( const RsGxsGroupId& channelId, const std::set& peers) { diff --git a/src/services/p3gxschannels.h b/src/services/p3gxschannels.h index ca15b0755..99b5f6a51 100644 --- a/src/services/p3gxschannels.h +++ b/src/services/p3gxschannels.h @@ -106,6 +106,11 @@ class p3GxsChannels: public RsGenExchange, public RsGxsChannels, /// @see RsGxsChannels bool setMessageReadStatus(const RsGxsGrpMsgIdPair& msgId, bool read) override; + /// @see RsGxsChannels (batch variant) + bool setMessageReadStatus( const RsGxsGroupId& channelId, + const std::vector& msgIds, + bool read ) override; + virtual bool groupShareKeys(const RsGxsGroupId &groupId, const std::set& peers) override; // no tokens... should be cached. diff --git a/src/services/p3gxsforums.cc b/src/services/p3gxsforums.cc index a6c8aa73f..bbe6a5ac4 100644 --- a/src/services/p3gxsforums.cc +++ b/src/services/p3gxsforums.cc @@ -895,6 +895,49 @@ bool p3GxsForums::markRead(const RsGxsGrpMsgIdPair& msgId, bool read) return true; } +bool p3GxsForums::markRead(const RsGxsGroupId& forumId, const std::vector& msgIds, bool read) +{ + if(msgIds.empty()) + return true; + + uint32_t mask = GXS_SERV::GXS_MSG_STATUS_GUI_NEW | GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + uint32_t status = read ? 0 : GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + + // Queue every status change at once. They all land in mMsgLocMetaMap and are + // drained together by a single processMsgMetaChanges() tick, which persists + // them in one DB transaction. This is what keeps "mark all as read" cheap + // instead of one blocking request (and one detached thread) per message. + std::vector tokens; + tokens.reserve(msgIds.size()); + + for(const RsGxsMessageId& msgId : msgIds) + { + uint32_t token; + setMsgStatusFlags(token, RsGxsGrpMsgIdPair(forumId, msgId), status, mask); + tokens.push_back(token); + } + + // Wait for the whole batch to be processed. All tokens complete in the same + // tick, so waiting on the last one is enough. + waitToken(tokens.back(), std::chrono::milliseconds(30000)); + + RsGxsGrpMsgIdPair p; + for(uint32_t token : tokens) + acknowledgeMsg(token, p); + + // A single event for the whole batch (the per-message event storm was part + // of what froze the UI). Consumers only use the group id to refresh counts. + if(rsEvents) + { + auto ev = std::make_shared(); + ev->mForumGroupId = forumId; + ev->mForumEventCode = RsForumEventCode::READ_STATUS_CHANGED; + rsEvents->postEvent(ev); + } + + return true; +} + bool p3GxsForums::subscribeToForum(const RsGxsGroupId& groupId, bool subscribe ) { uint32_t token; diff --git a/src/services/p3gxsforums.h b/src/services/p3gxsforums.h index 31a458777..5bd917b1a 100644 --- a/src/services/p3gxsforums.h +++ b/src/services/p3gxsforums.h @@ -124,6 +124,11 @@ class p3GxsForums: public RsGenExchange, public RsGxsForums, public p3Config, /// @see RsGxsForums::markRead virtual bool markRead(const RsGxsGrpMsgIdPair& messageId, bool read) override; + /// @see RsGxsForums::markRead (batch variant) + virtual bool markRead( const RsGxsGroupId& forumId, + const std::vector& msgIds, + bool read ) override; + /// @see RsGxsForums::updateReputationLevel virtual void updateReputationLevel(uint32_t forum_group_sign_flags,ForumPostEntry& e) const override; diff --git a/src/services/p3posted.cc b/src/services/p3posted.cc index 2bb4f0847..442fa825d 100644 --- a/src/services/p3posted.cc +++ b/src/services/p3posted.cc @@ -707,6 +707,45 @@ bool p3Posted::setPostReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) { return setCommentReadStatus(msgId,read); } +bool p3Posted::setPostReadStatus(const RsGxsGroupId& boardId, const std::vector& msgIds, bool read) +{ + if(msgIds.empty()) + return true; + + uint32_t mask = GXS_SERV::GXS_MSG_STATUS_GUI_NEW | GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + uint32_t status = read ? 0 : GXS_SERV::GXS_MSG_STATUS_GUI_UNREAD; + + // Queue every status change at once. They are all drained together by a + // single processMsgMetaChanges() tick (one DB transaction), instead of one + // blocking request + one detached thread + one event per message. + std::vector tokens; + tokens.reserve(msgIds.size()); + + for(const RsGxsMessageId& msgId : msgIds) + { + uint32_t token; + setMsgStatusFlags(token, RsGxsGrpMsgIdPair(boardId, msgId), status, mask); + tokens.push_back(token); + } + + // All tokens complete in the same tick, so waiting on the last is enough. + waitToken(tokens.back(), std::chrono::milliseconds(30000)); + + RsGxsGrpMsgIdPair p; + for(uint32_t token : tokens) + acknowledgeMsg(token, p); + + // One single event for the whole batch (consumers only need the group id). + if(rsEvents) + { + auto ev = std::make_shared(); + ev->mPostedGroupId = boardId; + ev->mPostedEventCode = RsPostedEventCode::READ_STATUS_CHANGED; + rsEvents->postEvent(ev); + } + + return true; +} bool p3Posted::setCommentReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) { uint32_t token; diff --git a/src/services/p3posted.h b/src/services/p3posted.h index 4adc2cf29..0656f1fe1 100644 --- a/src/services/p3posted.h +++ b/src/services/p3posted.h @@ -125,6 +125,9 @@ virtual void receiveHelperChanges(std::vector& changes) bool setCommentReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) override; bool setPostReadStatus(const RsGxsGrpMsgIdPair &msgId, bool read) override; + bool setPostReadStatus( const RsGxsGroupId& boardId, + const std::vector& msgIds, + bool read ) override; bool getRelatedComments( const RsGxsGroupId& gid,const std::set& msgIds, std::vector &comments ) override;