Use brpc Module Interface to refactor ThreadGroup loop

concourse/scripts/build_tarball.bash

@@ -287,12 +287,6 @@ if [ -f ${DEST_DIR}/bin/dss_server ]; then
   patchelf --set-rpath '$ORIGIN/../lib' ${DEST_DIR}/bin/dss_server
 fi
 
-# Preload libmimalloc and libbrpc at launch.
-patchelf --remove-needed libmimalloc.so.2 ${DEST_DIR}/bin/eloqdoc
-patchelf --remove-needed libbrpc.so ${DEST_DIR}/bin/eloqdoc
-patchelf --add-needed libbrpc.so ${DEST_DIR}/bin/eloqdoc
-patchelf --add-needed libmimalloc.so.2 ${DEST_DIR}/bin/eloqdoc
-
 # Config files
 cp ${ELOQDOC_SRC}/concourse/artifact/${DATA_STORE_TYPE}/* ${DEST_DIR}/etc
 

concourse/scripts/build_tarball_open.bash

@@ -205,12 +205,6 @@ if [ -f ${DEST_DIR}/bin/host_manager ]; then
   patchelf --set-rpath '$ORIGIN/../lib' ${DEST_DIR}/bin/host_manager
 fi
 
-# Preload libmimalloc and libbrpc at launch.
-patchelf --remove-needed libmimalloc.so.2 ${DEST_DIR}/bin/eloqdoc
-patchelf --remove-needed libbrpc.so ${DEST_DIR}/bin/eloqdoc
-patchelf --add-needed libbrpc.so ${DEST_DIR}/bin/eloqdoc
-patchelf --add-needed libmimalloc.so.2 ${DEST_DIR}/bin/eloqdoc
-
 # Config files
 cp ${ELOQDOC_SRC}/concourse/artifact/${DATA_STORE_TYPE}/* ${DEST_DIR}/etc
 

concourse/scripts/common.sh

@@ -63,6 +63,7 @@ compile_and_install() {
             -DCMAKE_CXX_FLAGS_DEBUG_INIT="-Wno-error -fPIC" \
             -DCMAKE_BUILD_TYPE=Debug \
             -DEXT_TX_PROC_ENABLED=ON \
+            -DELOQ_MODULE_ENABLED=ON \
             -DSTATISTICS=ON \
             -DUSE_ASAN=OFF \
             -DWITH_DATA_STORE=ELOQDSS_ROCKSDB_CLOUD_S3
@@ -79,6 +80,7 @@ compile_and_install() {
             VARIANT_DIR=Debug \
             LIBPATH=/usr/local/lib \
             CXXFLAGS="-Wno-nonnull -Wno-class-memaccess -Wno-interference-size -Wno-redundant-move" \
+            CPPDEFINES="ELOQ_MODULE_ENABLED" \
             --build-dir=#build \
             --prefix="$PREFIX" \
             --dbg=on \
@@ -110,6 +112,7 @@ compile_and_install_ent() {
             -DCMAKE_CXX_FLAGS_DEBUG_INIT="-Wno-error -fPIC" \
             -DCMAKE_BUILD_TYPE=Debug \
             -DEXT_TX_PROC_ENABLED=ON \
+            -DELOQ_MODULE_ENABLED=ON \
             -DSTATISTICS=ON \
             -DUSE_ASAN=OFF \
             -DWITH_LOG_STATE=ROCKSDB_CLOUD_S3 \
@@ -130,6 +133,7 @@ compile_and_install_ent() {
             VARIANT_DIR=Debug \
             LIBPATH=/usr/local/lib \
             CXXFLAGS="-Wno-nonnull -Wno-class-memaccess -Wno-interference-size -Wno-redundant-move" \
+            CPPDEFINES="ELOQ_MODULE_ENABLED" \
             --build-dir=#build \
             --prefix="$PREFIX" \
             --dbg=on \
@@ -151,7 +155,6 @@ launch_eloqdoc() {
       local bucket_name="$1"
       local bucket_prefix="$2"
       echo "launch eloqdoc with bucket name: $bucket_name, bucket prefix: $bucket_prefix"
-      export LD_PRELOAD=/usr/local/lib/libmimalloc.so
       mkdir -p "$PREFIX/log" "$PREFIX/data"
       sed -i "s|rocksdbCloudEndpointUrl: \"http://[0-9]\+\.[0-9]\+\.[0-9]\+\.[0-9]\+:[0-9]\+\"|rocksdbCloudEndpointUrl: \"${MINIO_ENDPOINT}\"|g" /home/eloq/workspace/mongo/concourse/scripts/store_rocksdb_cloud.yaml
       sed -i "s|txlogRocksDBCloudEndpointUrl: \"http://[0-9]\+\.[0-9]\+\.[0-9]\+\.[0-9]\+:[0-9]\+\"|txlogRocksDBCloudEndpointUrl: \"${MINIO_ENDPOINT}\"|g" /home/eloq/workspace/mongo/concourse/scripts/store_rocksdb_cloud.yaml
@@ -173,7 +176,6 @@ launch_eloqdoc_fast() {
       local bucket_name="$1"
       local bucket_prefix="$2"
       echo "launch eloqdoc fast with bucket name: $bucket_name, bucket prefix: $bucket_prefix"
-      export LD_PRELOAD=/usr/local/lib/libmimalloc.so
       mkdir -p "$PREFIX/log" "$PREFIX/data"
       sed -i "s|rocksdbCloudEndpointUrl: \"http://[0-9]\+\.[0-9]\+\.[0-9]\+\.[0-9]\+:[0-9]\+\"|rocksdbCloudEndpointUrl: \"${MINIO_ENDPOINT}\"|g" /home/eloq/workspace/mongo/concourse/scripts/store_rocksdb_cloud.yaml
       sed -i "s|txlogRocksDBCloudEndpointUrl: \"http://[0-9]\+\.[0-9]\+\.[0-9]\+\.[0-9]\+:[0-9]\+\"|txlogRocksDBCloudEndpointUrl: \"${MINIO_ENDPOINT}\"|g" /home/eloq/workspace/mongo/concourse/scripts/store_rocksdb_cloud.yaml

concourse/scripts/store_rocksdb_cloud.yaml

@@ -25,6 +25,8 @@ storage:
       collectActiveTxTsIntervalSec: 2
       checkpointerDelaySec: 5
       ccProtocol: "OccRead"
+      enableIOuring: true
+      txlogAsyncFsync: true
       # txlogServiceList: ["127.0.0.1:8500"]
       # txlogGroupReplicaNum: 1
       # skipRedoLog: true

docs/how-to-compile.md

@@ -117,20 +117,3 @@ python scripts/buildscripts/scons.py \
 ```
 
 All executable files will be installed to `$INSTALL_PREFIX/bin`, and all libraries will be installed to `$INSTALL_PREFIX/lib`.
-
-## 3 Adjust load order of libmimalloc and libbrpc
-
-EloqDoc depends on libmimalloc and libbrpc, and requires them to load before other libraries.
-
-```bash
-patchelf --remove-needed libmimalloc.so.2 $INSTALL_PREFIX/bin/eloqdoc
-patchelf --remove-needed libbrpc.so $INSTALL_PREFIX/bin/eloqdoc
-patchelf --add-needed libbrpc.so $INSTALL_PREFIX/bin/eloqdoc
-patchelf --add-needed libmimalloc.so.2 $INSTALL_PREFIX/bin/eloqdoc
-```
-
-If you don't adjust load order, then you must set LD_PRELOAD before run EloqDoc.
-
-```bash
-export LD_PRELOAD=/usr/local/lib/libmimalloc.so.2:/usr/lib/libbrpc.so
-```

src/mongo/SConscript

@@ -64,7 +64,6 @@ env.Library(
         'base/init.cpp',
         'base/initializer.cpp',
         'base/initializer_dependency_graph.cpp',
-        'base/local_thread_state.cpp',
         'base/make_string_vector.cpp',
         'base/parse_number.cpp',
         'base/shim.cpp',
@@ -365,6 +364,10 @@ mongod = env.Program(
         'base',
         'mongodmain',
     ],
+    SYSLIBDEPS=[
+        'mimalloc',
+        'brpc',
+    ] if env.TargetOSIs('linux') else [],
     INSTALL_ALIAS=[
         'core',
         'default',

src/mongo/base/string_data.h

@@ -176,7 +176,7 @@ class StringData {
     std::string toString() const {
         return std::string(_data, size());
     }
-    std::string_view toStringView() const{
+    std::string_view toStringView() const {
         return std::string_view(_data, size());
     }
     constexpr char operator[](unsigned pos) const {
@@ -353,3 +353,12 @@ inline std::string operator+(StringData lhs, std::string rhs) {
 }
 
 }  // namespace mongo
+
+namespace std {
+template <>
+struct hash<mongo::StringData> {
+    size_t operator()(const mongo::StringData& sd) const {
+        return std::hash<std::string_view>{}(sd.toStringView());
+    }
+};
+}  // namespace std

src/mongo/db/SConscript

@@ -510,6 +510,7 @@ env.Library(
         'server_recovery.cpp',
         'unclean_shutdown.cpp',
         'coro_sync.cpp',
+        'local_thread_state.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/logical_session_id',
@@ -521,6 +522,7 @@ env.Library(
         '$BUILD_DIR/mongo/util/net/network',
         '$BUILD_DIR/mongo/util/periodic_runner',
     ],
+    SYSLIBDEPS=["brpc"],
 )
 
 env.Library(

src/mongo/db/catalog/database_holder_impl.cpp

@@ -51,7 +51,6 @@
 #include "mongo/util/log.h"
 
 namespace mongo {
-extern thread_local int16_t localThreadId;
 
 namespace {
 
@@ -99,7 +98,7 @@ Database* DatabaseHolderImpl::get(OperationContext* opCtx, StringData ns) {
     const StringData db = _todb(ns);
     invariant(opCtx->lockState()->isDbLockedForMode(db, MODE_IS));
 
-    auto id = static_cast<int16_t>(localThreadId + 1);
+    auto id = static_cast<int16_t>(LocalThread::ID() + 1);
     // std::scoped_lock<std::mutex> lock{_dbMapMutexVector[id]};
     const auto& dbMap = _dbMapVector[id];
     if (auto iter = dbMap.find(db); iter != dbMap.end()) {
@@ -124,7 +123,7 @@ Database* DatabaseHolderImpl::get(OperationContext* opCtx, StringData ns) {
 std::set<std::string> DatabaseHolderImpl::_getNamesWithConflictingCasing_inlock(
     StringData name) const {
     std::set<std::string> duplicates;
-    auto id = static_cast<int16_t>(localThreadId + 1);
+    auto id = static_cast<int16_t>(LocalThread::ID() + 1);
 
     for (const auto& [dbName, dbPtr] : _dbMapVector[id]) {
         // A name that's equal with case-insensitive match must be identical, or it's a duplicate.
@@ -149,7 +148,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
         *justCreated = false;  // Until proven otherwise.
     }
 
-    auto id = static_cast<int16_t>(localThreadId + 1);
+    auto id = static_cast<int16_t>(LocalThread::ID() + 1);
     auto& dbMap = _dbMapVector[id];
 
     // std::scoped_lock<std::mutex> lock(_dbMapMutexVector[id]);
@@ -223,7 +222,7 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns, const std
 
     const StringData dbName = _todb(ns);
 
-    auto id = static_cast<int16_t>(localThreadId + 1);
+    auto id = static_cast<int16_t>(LocalThread::ID() + 1);
     {
         // std::scoped_lock<std::mutex> lock{_dbMapMutexVector[id]};
         auto& dbMap = _dbMapVector[id];
@@ -249,7 +248,7 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns, const std
 void DatabaseHolderImpl::closeAll(OperationContext* opCtx, const std::string& reason) {
     invariant(opCtx->lockState()->isW());
 
-    auto id = static_cast<int16_t>(localThreadId + 1);
+    auto id = static_cast<int16_t>(LocalThread::ID() + 1);
 
     auto& dbMap = _dbMapVector[id];
     // std::scoped_lock<std::mutex> lock{_dbMapMutexVector[i]};

src/mongo/db/client.h

@@ -39,9 +39,9 @@
 #include <boost/optional.hpp>
 
 #include "mongo/base/disallow_copying.h"
-#include "mongo/base/local_thread_state.h"
 #include "mongo/db/client.h"
 #include "mongo/db/coro_sync.h"
+#include "mongo/db/local_thread_state.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/service_context.h"
 #include "mongo/platform/random.h"
@@ -52,6 +52,10 @@
 #include "mongo/util/invariant.h"
 #include "mongo/util/net/hostandport.h"
 
+#ifdef D_USE_CORO_SYNC
+#include "mongo/db/coro_sync.h"
+#endif
+
 namespace mongo {
 
 class Collection;

src/mongo/db/coro_sync.cpp

@@ -2,16 +2,17 @@
 
 #include "mongo/db/coro_sync.h"
 #include "mongo/db/client.h"
+#include "mongo/db/local_thread_state.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
 
-extern thread_local int16_t localThreadId;
+const CoroutineFunctors CoroutineFunctors::Unavailable{};
 
 namespace coro {
 
 void Mutex::lock() {
-    if (localThreadId != -1) {
+    if (LocalThread::ID() != -1) {
         Client* client = Client::getCurrent();
         if (client) {
             const CoroutineFunctors& coro = Client::getCurrent()->coroutineFunctors();
@@ -24,13 +25,13 @@ void Mutex::lock() {
                     (*coro.yieldFuncPtr)();
                 }
             } else {
-                MONGO_LOG(2)
-                    << "ThreadGroup " << localThreadId
+                MONGO_LOG(1)
+                    << "ThreadGroup " << LocalThread::ID()
                     << " call std::mutex::lock because the coroutine context is unavailable.";
                 _mux.lock();
             }
         } else {
-            MONGO_LOG(2) << "ThreadGroup " << localThreadId
+            MONGO_LOG(1) << "ThreadGroup " << LocalThread::ID()
                          << " call std::mutex::lock because the client object is unavailable.";
             _mux.lock();
         }
@@ -41,7 +42,7 @@ void Mutex::lock() {
 
 void ConditionVariable::wait(std::unique_lock<Mutex>& lock) {
     invariant(lock.owns_lock());
-    if (localThreadId != -1) {
+    if (LocalThread::ID() != -1) {
         Client* client = Client::getCurrent();
         if (client) {
             const CoroutineFunctors& coro = Client::getCurrent()->coroutineFunctors();
@@ -51,14 +52,14 @@ void ConditionVariable::wait(std::unique_lock<Mutex>& lock) {
                 (*coro.yieldFuncPtr)();
                 lock.lock();
             } else {
-                MONGO_LOG(2) << "ThreadGroup " << localThreadId
+                MONGO_LOG(1) << "ThreadGroup " << LocalThread::ID()
                              << " call std::condition_variable::wait because the coroutine context "
                                 "is unavailable.";
                 _cv.wait(reinterpret_cast<std::unique_lock<std::mutex>&>(lock));
             }
         } else {
-            MONGO_LOG(2)
-                << "ThreadGroup " << localThreadId
+            MONGO_LOG(1)
+                << "ThreadGroup " << LocalThread::ID()
                 << " call std::condition_variable::wait because the client object is unavailable.";
             _cv.wait(reinterpret_cast<std::unique_lock<std::mutex>&>(lock));
         }

src/mongo/db/coro_sync.h

@@ -1,11 +1,33 @@
 #pragma once
 
 #include <condition_variable>
+#include <cstdint>
+#include <functional>
 #include <mutex>
 
 #include "mongo/util/time_support.h"
 
-namespace mongo::coro {
+namespace mongo {
+
+struct CoroutineFunctors {
+    const std::function<void()>* yieldFuncPtr{nullptr};
+    const std::function<void()>* resumeFuncPtr{nullptr};
+    const std::function<void()>* longResumeFuncPtr{nullptr};
+    const std::function<void(uint16_t)>* migrateThreadGroupFuncPtr{nullptr};
+
+    const static CoroutineFunctors Unavailable;
+
+    friend bool operator==(const CoroutineFunctors& lhs, const CoroutineFunctors& rhs) {
+        return lhs.yieldFuncPtr == rhs.yieldFuncPtr && lhs.resumeFuncPtr == rhs.resumeFuncPtr &&
+            lhs.longResumeFuncPtr == rhs.longResumeFuncPtr &&
+            lhs.migrateThreadGroupFuncPtr == rhs.migrateThreadGroupFuncPtr;
+    }
+    friend bool operator!=(const CoroutineFunctors& lhs, const CoroutineFunctors& rhs) {
+        return !(lhs == rhs);
+    }
+};
+
+namespace coro {
 /**
  * coro::Mutex can be used in both pthread/coroutine.
  * coro::Mutex can be cast to std::mutex.
@@ -90,4 +112,5 @@ class ConditionVariable {
 private:
     std::condition_variable _cv;
 };
-}  // namespace mongo::coro
+}  // namespace coro
+}  // namespace mongo