[store]feat: add conductor

.github/workflows/ci.yml

@@ -261,6 +261,14 @@ jobs:
           echo "✅ Coverage collected successfully"
         fi
 
+    - name: Test mooncake conductor (C++)
+      run: |
+        cd mooncake-conductor
+        cmake -S . -B build -DBUILD_UNIT_TESTS=ON
+        cmake --build build -j4
+        ./build/tests/conductor_test
+      shell: bash
+
     - name: Generate Python version tag
       id: generate_tag_build
       run: |

CMakeLists.txt

@@ -8,56 +8,65 @@ include(mooncake-common/FindJsonCpp.cmake)
 include(mooncake-common/FindGLOG.cmake)
 include(mooncake-common/common.cmake)
 # unit test
-if (BUILD_UNIT_TESTS)
+if(BUILD_UNIT_TESTS)
   enable_testing()
 endif()
 
 option(WITH_TE "build mooncake transfer engine and sample code" ON)
 option(WITH_STORE "build mooncake store library and sample code" ON)
 option(WITH_STORE_GO "build Go bindings for mooncake store" OFF)
+option(WITH_CONDUCTOR "build mooncake conductor service" OFF)
 option(WITH_P2P_STORE "build p2p store library and sample code" OFF)
-option(WITH_RUST_EXAMPLE "build the Rust interface and sample code for the transfer engine" OFF)
+option(WITH_RUST_EXAMPLE
+       "build the Rust interface and sample code for the transfer engine" OFF)
 option(WITH_STORE_RUST "build the Rust bindings for the Mooncake Store" ON)
 option(WITH_EP "build mooncake with expert parallelism support" OFF)
 option(USE_NOF "build mooncake store with NoF SSD pool support" OFF)
 
 include(${CMAKE_CURRENT_SOURCE_DIR}/mooncake-common/SetupPython.cmake)
 add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/extern/pybind11)
 execute_process(
-    COMMAND ${PYTHON_EXECUTABLE} -c "import sys; print(sys.path[-1])"
-    OUTPUT_VARIABLE PYTHON_SYS_PATH
-)
+  COMMAND ${PYTHON_EXECUTABLE} -c "import sys; print(sys.path[-1])"
+  OUTPUT_VARIABLE PYTHON_SYS_PATH)
 string(STRIP ${PYTHON_SYS_PATH} PYTHON_SYS_PATH)
 
-if (USE_ETCD)
+if(USE_ETCD)
   add_compile_definitions(USE_ETCD)
-  if (USE_ETCD_LEGACY)
+  if(USE_ETCD_LEGACY)
     add_compile_definitions(USE_ETCD_LEGACY)
-    message(STATUS "etcd as metadata server support is enabled (etcd-cpp-api-v3)")
+    message(
+      STATUS "etcd as metadata server support is enabled (etcd-cpp-api-v3)")
   else()
     message(STATUS "etcd as metadata server support is enabled (go package)")
   endif()
 endif()
 option(STORE_USE_ETCD "build mooncake store with etcd" OFF)
-if (STORE_USE_ETCD)
+if(STORE_USE_ETCD)
   add_compile_definitions(STORE_USE_ETCD)
 endif()
 option(STORE_USE_REDIS "build mooncake store with redis" OFF)
-if (STORE_USE_REDIS)
+if(STORE_USE_REDIS)
   add_compile_definitions(STORE_USE_REDIS)
 endif()
-option(STORE_USE_K8S_LEASE "build mooncake store with K8s Lease leader election" OFF)
-if (STORE_USE_K8S_LEASE)
-  if (STORE_USE_ETCD)
-    message(FATAL_ERROR "STORE_USE_K8S_LEASE and STORE_USE_ETCD cannot be enabled together because both build Go c-shared HA backends.")
+option(STORE_USE_K8S_LEASE
+       "build mooncake store with K8s Lease leader election" OFF)
+if(STORE_USE_K8S_LEASE)
+  if(STORE_USE_ETCD)
+    message(
+      FATAL_ERROR
+        "STORE_USE_K8S_LEASE and STORE_USE_ETCD cannot be enabled together because both build Go c-shared HA backends."
+    )
   endif()
-  if (USE_ETCD AND NOT USE_ETCD_LEGACY)
-    message(FATAL_ERROR "STORE_USE_K8S_LEASE cannot be enabled with non-legacy USE_ETCD because both build Go c-shared libraries in the same process.")
+  if(USE_ETCD AND NOT USE_ETCD_LEGACY)
+    message(
+      FATAL_ERROR
+        "STORE_USE_K8S_LEASE cannot be enabled with non-legacy USE_ETCD because both build Go c-shared libraries in the same process."
+    )
   endif()
   add_compile_definitions(STORE_USE_K8S_LEASE)
 endif()
 
-if (USE_NOF)
+if(USE_NOF)
   add_compile_definitions(USE_NOF)
 else()
   message(STATUS "USE_NOF=OFF, NoF SSD pool support is disabled")
@@ -73,48 +82,59 @@ include_directories(mooncake-common/etcd)
 include_directories(mooncake-common/k8s-lease)
 include_directories(mooncake-common/include)
 
-if (WITH_TE)
+if(WITH_TE)
   add_subdirectory(mooncake-transfer-engine)
   include_directories(mooncake-transfer-engine/include)
 endif()
 
-if (WITH_STORE)
+if(WITH_STORE)
   message(STATUS "Mooncake Store will be built")
   add_subdirectory(mooncake-store)
   include_directories(mooncake-store/include)
 endif()
 
-if (WITH_STORE_RUST)
-  if (NOT WITH_STORE)
+if(WITH_CONDUCTOR)
+  message(STATUS "Mooncake Conductor will be built")
+  add_subdirectory(mooncake-conductor)
+endif()
+
+if(WITH_STORE_RUST)
+  if(NOT WITH_STORE)
     message(FATAL_ERROR "WITH_STORE_RUST=ON requires WITH_STORE=ON")
   endif()
   message(STATUS "Mooncake Store Rust bindings will be built")
   add_subdirectory(mooncake-store/rust)
 endif()
 
 option(EP_USE_IDE "Enable intelligent indexing for IDEs" OFF)
-if (WITH_EP)
-  if (EP_USE_IDE)
+if(WITH_EP)
+  if(EP_USE_IDE)
     message(WARNING "EP_USE_IDE enabled. DO NOT USE IN PRODUCTION!")
     add_subdirectory(mooncake-ep)
     include_directories(mooncake-ep/include)
     add_subdirectory(mooncake-pg)
     include_directories(mooncake-pg/include)
-  else ()
-    message(STATUS "WITH_EP enabled: building Mooncake EP and PG Python extensions")
+  else()
+    message(
+      STATUS "WITH_EP enabled: building Mooncake EP and PG Python extensions")
     if(USE_CUDA)
       find_package(CUDAToolkit REQUIRED)
       message(STATUS "Detected CUDA version: ${CUDAToolkit_VERSION}")
     endif()
 
-    # EP_TORCH_VERSIONS: semicolon-separated list of PyTorch versions to build for.
-    # Can be set via -DEP_TORCH_VERSIONS="2.9.1;2.8.0" or the EP_TORCH_VERSIONS env var.
-    # Empty means build with the currently-installed torch.
+    # EP_TORCH_VERSIONS: semicolon-separated list of PyTorch versions to build
+    # for. Can be set via -DEP_TORCH_VERSIONS="2.9.1;2.8.0" or the
+    # EP_TORCH_VERSIONS env var. Empty means build with the currently-installed
+    # torch.
     if(NOT EP_TORCH_VERSIONS)
       set(EP_TORCH_VERSIONS "$ENV{EP_TORCH_VERSIONS}")
     endif()
-    set(EP_TORCH_VERSIONS "${EP_TORCH_VERSIONS}" CACHE STRING
-        "PyTorch versions for EP/PG extensions, semicolon-separated (empty = use currently-installed torch)")
+    set(EP_TORCH_VERSIONS
+        "${EP_TORCH_VERSIONS}"
+        CACHE
+          STRING
+          "PyTorch versions for EP/PG extensions, semicolon-separated (empty = use currently-installed torch)"
+    )
 
     # TORCH_CUDA_ARCH_LIST forwarded to the torch CUDA extension build.
     if(NOT TORCH_CUDA_ARCH_LIST)
@@ -123,23 +143,27 @@ if (WITH_EP)
     if(NOT TORCH_CUDA_ARCH_LIST)
       set(TORCH_CUDA_ARCH_LIST "8.0;9.0")
     endif()
-    set(TORCH_CUDA_ARCH_LIST "${TORCH_CUDA_ARCH_LIST}" CACHE STRING
-        "CUDA arch list for EP/PG extension builds (e.g. \"8.0;9.0\")")
+    set(TORCH_CUDA_ARCH_LIST
+        "${TORCH_CUDA_ARCH_LIST}"
+        CACHE STRING
+              "CUDA arch list for EP/PG extension builds (e.g. \"8.0;9.0\")")
 
     # Staging directory: EP/PG .so files are placed here during make and later
     # injected into the wheel AFTER auditwheel, so patchelf never touches the
     # CUDA fatbins (which would cause cudaErrorInvalidKernelImage at runtime).
     set(EP_PG_STAGING_DIR "${CMAKE_BINARY_DIR}/ep_pg_staging")
 
-    # Convert semicolon-separated lists to pipe-separated strings so they survive
-    # CMake's COMMAND list-splitting (semicolons are CMake list separators).
+    # Convert semicolon-separated lists to pipe-separated strings so they
+    # survive CMake's COMMAND list-splitting (semicolons are CMake list
+    # separators).
     string(REPLACE ";" "|" _ep_torch_versions_pipe "${EP_TORCH_VERSIONS}")
     string(REPLACE ";" "|" _torch_cuda_arch_list_pipe "${TORCH_CUDA_ARCH_LIST}")
 
-    add_custom_target(mooncake_ep_ext ALL
+    add_custom_target(
+      mooncake_ep_ext ALL
       COMMAND ${CMAKE_COMMAND} -E make_directory "${EP_PG_STAGING_DIR}"
-      COMMAND ${CMAKE_COMMAND}
-        "-DSOURCE_DIR=${CMAKE_CURRENT_SOURCE_DIR}/mooncake-ep"
+      COMMAND
+        ${CMAKE_COMMAND} "-DSOURCE_DIR=${CMAKE_CURRENT_SOURCE_DIR}/mooncake-ep"
         "-DEP_CUDA_MAJOR=${CUDAToolkit_VERSION_MAJOR}"
         "-DEP_CUDA_MINOR=${CUDAToolkit_VERSION_MINOR}"
         "-DEP_TORCH_VERSIONS=${_ep_torch_versions_pipe}"
@@ -148,17 +172,17 @@ if (WITH_EP)
         "-DENGINE_SO_PATH=$<TARGET_FILE:engine>"
         "-DPython3_EXECUTABLE=${Python3_EXECUTABLE}"
         "-DEP_USE_MUSA=$<IF:$<BOOL:${USE_MUSA}>,1,0>"
-        "-DEP_USE_MACA=$<IF:$<BOOL:${USE_MACA}>,1,0>"
-        -P "${CMAKE_CURRENT_SOURCE_DIR}/mooncake-ep/BuildEpExt.cmake"
+        "-DEP_USE_MACA=$<IF:$<BOOL:${USE_MACA}>,1,0>" -P
+        "${CMAKE_CURRENT_SOURCE_DIR}/mooncake-ep/BuildEpExt.cmake"
       COMMENT "Building Mooncake EP Python extension(s)"
       DEPENDS engine
-      VERBATIM
-    )
+      VERBATIM)
 
-    add_custom_target(mooncake_pg_ext ALL
+    add_custom_target(
+      mooncake_pg_ext ALL
       COMMAND ${CMAKE_COMMAND} -E make_directory "${EP_PG_STAGING_DIR}"
-      COMMAND ${CMAKE_COMMAND}
-        "-DSOURCE_DIR=${CMAKE_CURRENT_SOURCE_DIR}/mooncake-pg"
+      COMMAND
+        ${CMAKE_COMMAND} "-DSOURCE_DIR=${CMAKE_CURRENT_SOURCE_DIR}/mooncake-pg"
         "-DEP_CUDA_MAJOR=${CUDAToolkit_VERSION_MAJOR}"
         "-DEP_CUDA_MINOR=${CUDAToolkit_VERSION_MINOR}"
         "-DEP_TORCH_VERSIONS=${_ep_torch_versions_pipe}"
@@ -167,29 +191,28 @@ if (WITH_EP)
         "-DENGINE_SO_PATH=$<TARGET_FILE:engine>"
         "-DPython3_EXECUTABLE=${Python3_EXECUTABLE}"
         "-DEP_USE_MUSA=$<IF:$<BOOL:${USE_MUSA}>,1,0>"
-        "-DEP_USE_MACA=$<IF:$<BOOL:${USE_MACA}>,1,0>"
-        -P "${CMAKE_CURRENT_SOURCE_DIR}/mooncake-pg/BuildPgExt.cmake"
+        "-DEP_USE_MACA=$<IF:$<BOOL:${USE_MACA}>,1,0>" -P
+        "${CMAKE_CURRENT_SOURCE_DIR}/mooncake-pg/BuildPgExt.cmake"
       COMMENT "Building Mooncake PG Python extension(s)"
       DEPENDS engine mooncake_ep_ext
-      VERBATIM
-    )
-  endif ()
+      VERBATIM)
+  endif()
 endif()
 
 add_subdirectory(mooncake-integration)
 
-if (WITH_STORE_GO AND WITH_STORE)
+if(WITH_STORE_GO AND WITH_STORE)
   add_custom_target(build_store_go DEPENDS mooncake_store transfer_engine)
   add_custom_command(
-      TARGET build_store_go
-      COMMAND bash build.sh ${CMAKE_BINARY_DIR} ${CMAKE_CURRENT_BINARY_DIR} ${USE_ETCD} ${USE_REDIS} ${USE_HTTP} ${USE_ETCD_LEGACY}
-      WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/mooncake-store/go
-  )
+    TARGET build_store_go
+    COMMAND bash build.sh ${CMAKE_BINARY_DIR} ${CMAKE_CURRENT_BINARY_DIR}
+            ${USE_ETCD} ${USE_REDIS} ${USE_HTTP} ${USE_ETCD_LEGACY}
+    WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/mooncake-store/go)
   set_property(TARGET build_store_go PROPERTY EXCLUDE_FROM_ALL FALSE)
   message(STATUS "Mooncake Store Go bindings will be built")
 endif()
 
-if (WITH_P2P_STORE)
+if(WITH_P2P_STORE)
   add_subdirectory(mooncake-p2p-store)
   message(STATUS "P2P Store will be built")
 endif()

docs/source/getting_started/build.md

@@ -179,11 +179,20 @@ The following options can be passed to `cmake ..`.
 | `-DWITH_TE=ON/OFF` | `ON` | Build the Mooncake Transfer Engine component and sample code. |
 | `-DWITH_STORE=ON/OFF` | `ON` | Build the Mooncake Store component. |
 | `-DWITH_STORE_GO=ON/OFF` | `OFF` | Build Go bindings for Mooncake Store when `-DWITH_STORE=ON`. |
+| `-DWITH_CONDUCTOR=ON/OFF` | `OFF` | Build the Mooncake Conductor service. |
 | `-DWITH_P2P_STORE=ON/OFF` | `OFF` | Enable Golang support and build the P2P Store component. Requires Go 1.23+. |
 | `-DWITH_RUST_EXAMPLE=ON/OFF` | `OFF` | Build the Transfer Engine Rust interface and sample code. |
 | `-DWITH_STORE_RUST=ON/OFF` | `ON` | Build Mooncake Store Rust bindings and CMake Rust targets. |
 | `-DWITH_EP=ON/OFF` | `OFF` | Build the EP and PG Python extensions for CUDA. Requires CUDA toolkit and PyTorch. Use `-DEP_TORCH_VERSIONS="2.12.1"` to build for specific PyTorch versions, or leave empty to use the currently installed torch. The CUDA version is detected automatically. |
 
+To build only the Conductor target from a configured build tree, enable the
+component and build `mooncake_conductor`:
+
+```bash
+cmake -S . -B build -DWITH_CONDUCTOR=ON
+cmake --build build --target mooncake_conductor
+```
+
 ### Build Behavior Options
 
 | Option | Default | Description |

mooncake-conductor/CMakeLists.txt

@@ -0,0 +1,89 @@
+cmake_minimum_required(VERSION 3.16)
+
+if(NOT GLOBAL_CONFIG)
+  project(mooncake-conductor CXX)
+  set(CMAKE_CXX_STANDARD 20)
+  set(CMAKE_CXX_STANDARD_REQUIRED ON)
+  include(../mooncake-common/FindJsonCpp.cmake)
+  include(../mooncake-common/FindGLOG.cmake)
+endif()
+
+find_package(cppzmq REQUIRED)
+find_package(msgpack-cxx REQUIRED)
+# In the top-level build mooncake-common/common.cmake has already found
+# yalantinglibs; its config.cmake mutates the imported target and cannot run
+# twice (the target belongs to the parent scope), so guard it.
+if(NOT TARGET yalantinglibs::yalantinglibs)
+  find_package(yalantinglibs CONFIG REQUIRED)
+endif()
+find_package(Threads REQUIRED)
+
+# ThreadSanitizer build for the concurrency tests (task 6.7): cmake
+# -DCONDUCTOR_CPP_TSAN=ON .. Run with:
+# TSAN_OPTIONS="suppressions=<src>/tests/tsan.supp" ctest (suppressions cover
+# uninstrumented libzmq/glog internals only).
+option(CONDUCTOR_CPP_TSAN "Build mooncake-conductor with ThreadSanitizer" OFF)
+if(CONDUCTOR_CPP_TSAN)
+  add_compile_options(-fsanitize=thread -g)
+  add_link_options(-fsanitize=thread)
+  message(STATUS "conductor-cpp: ThreadSanitizer enabled")
+endif()
+
+# Find xxHash (required for prefix hashing)
+find_path(
+  XXHASH_INCLUDE_DIR
+  NAMES xxhash.h
+  PATHS /usr/include /usr/local/include)
+find_library(
+  XXHASH_LIBRARY
+  NAMES xxhash libxxhash
+  PATHS /usr/lib /usr/local/lib /usr/lib64)
+if(XXHASH_INCLUDE_DIR AND XXHASH_LIBRARY)
+  message(
+    STATUS
+      "conductor-cpp: found xxHash: include=${XXHASH_INCLUDE_DIR} lib=${XXHASH_LIBRARY}"
+  )
+else()
+  message(
+    FATAL_ERROR
+      "xxHash library/header not found. Please install libxxhash-dev or xxhash-devel and try again."
+  )
+endif()
+
+add_library(
+  conductor_cpp_core STATIC
+  src/common/utils.cpp
+  src/prefixindex/prefix_indexer.cpp
+  src/zmq/msg_decoder.cpp
+  src/zmq/zmq_client.cpp
+  src/kvevent/event_handler.cpp
+  src/kvevent/event_manager.cpp
+  src/kvevent/config.cpp)
+
+target_include_directories(
+  conductor_cpp_core PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include
+                            ${XXHASH_INCLUDE_DIR})
+
+target_link_libraries(
+  conductor_cpp_core
+  PUBLIC ${XXHASH_LIBRARY}
+         cppzmq
+         msgpack-cxx
+         yalantinglibs::yalantinglibs
+         glog::glog
+         JsonCpp::JsonCpp
+         Threads::Threads)
+
+if(TARGET asio_shared)
+  target_link_libraries(conductor_cpp_core PUBLIC asio_shared ibverbs)
+endif()
+
+add_executable(mooncake_conductor src/main.cpp)
+target_link_libraries(mooncake_conductor PRIVATE conductor_cpp_core)
+
+install(TARGETS mooncake_conductor DESTINATION bin)
+
+if(BUILD_UNIT_TESTS)
+  enable_testing()
+  add_subdirectory(tests)
+endif()