feat: support global kv cache management.

.gitmodules

@@ -16,3 +16,6 @@
 [submodule "third_party/minja"]
 	path = third_party/minja
 	url = https://gitcode.com/xLLM-AI/minja.git
+[submodule "third_party/smhasher"]
+	path = third_party/smhasher
+	url = https://gitcode.com/xLLM-AI/smhasher.git

CMakeLists.txt

@@ -98,6 +98,7 @@ endif()
 
 # find all dependencies from vcpkg
 find_package(Boost REQUIRED)
+find_package(Boost REQUIRED COMPONENTS serialization)
 find_package(glog CONFIG REQUIRED)
 find_package(gflags CONFIG REQUIRED)
 find_package(leveldb CONFIG REQUIRED)

third_party/CMakeLists.txt

@@ -4,4 +4,5 @@ add_subdirectory(cpprestsdk)
 add_subdirectory(etcd_cpp_apiv3)
 add_subdirectory(brpc)
 add_subdirectory(minja)
-add_subdirectory(sentencepiece)
+add_subdirectory(sentencepiece)
+add_subdirectory(smhasher/src)

vcpkg.json

@@ -38,6 +38,10 @@
       "name": "boost-random",
       "version>=": "1.84.0"
     },
+    {
+      "name": "boost-serialization",
+      "version>=": "1.84.0"
+    },
     {
       "name": "protobuf",
       "version>=": "3.21.12",

xllm_service/common/CMakeLists.txt

@@ -14,6 +14,7 @@ cc_library(
     threadpool.h
     types.h
     utils.h
+    hash_util.h
     xllm/output.h
     xllm/status.h
     xllm/uuid.h
@@ -22,13 +23,14 @@ cc_library(
     json_reader.cpp
     threadpool.cpp
     utils.cpp
+    hash_util.cpp
     xllm/uuid.cpp
   DEPS
     absl::random_random
     absl::strings
     glog::glog
     gflags::gflags
     nlohmann_json::nlohmann_json
+    SMHasherSupport
 )
-target_link_libraries(common PRIVATE OpenSSL::SSL OpenSSL::Crypto)
 add_dependencies(common brpc-static)

xllm_service/common/global_gflags.cpp

@@ -33,11 +33,6 @@ DEFINE_int32(http_server_max_concurrency,
              128,
              "Limit number of requests processed in parallel");
 
-DEFINE_string(rpc_server_host,
-              "",
-              "Rpc server listen address, may be IPV4/IPV6/UDS."
-              " If this is set, the flag port will be ignored");
-
 DEFINE_int32(rpc_server_port, 8889, "Port for xllm rpc service to listen on");
 
 DEFINE_int32(rpc_server_idle_timeout_s,
@@ -55,6 +50,8 @@ DEFINE_string(etcd_addr,
               "0.0.0.0:2379",
               "etcd adderss for save instance meta info");
 
+DEFINE_uint32(murmur_hash3_seed, 1024, "default Murmur Hash seed");
+
 DEFINE_int32(port, 8888, "Port for xllm service to listen on");
 
 DEFINE_int32(num_threads, 32, "Number of threads to process requests");
@@ -81,7 +78,9 @@ DEFINE_int32(idle_timeout_s,
              "Connection will be closed if there is no "
              "read/write operations during the last `idle_timeout_s'");
 
-DEFINE_string(disagg_pd_policy, "RR", "Disaggregated prefill-decode policy.");
+DEFINE_string(load_balance_policy,
+              "RR",
+              "Disaggregated prefill-decode policy.");
 
 DEFINE_int32(detect_disconnected_instance_interval,
              15,

xllm_service/common/global_gflags.h

@@ -27,8 +27,6 @@ DECLARE_int32(http_server_num_threads);
 
 DECLARE_int32(http_server_max_concurrency);
 
-DECLARE_string(rpc_server_host);
-
 DECLARE_int32(rpc_server_port);
 
 DECLARE_int32(rpc_server_idle_timeout_s);
@@ -37,6 +35,8 @@ DECLARE_int32(rpc_server_num_threads);
 
 DECLARE_int32(rpc_server_max_concurrency);
 
+DECLARE_uint32(murmur_hash3_seed);
+
 DECLARE_string(test_instance_addr);
 
 DECLARE_int32(timeout_ms);
@@ -53,7 +53,7 @@ DECLARE_int32(max_concurrency);
 
 DECLARE_string(etcd_addr);
 
-DECLARE_string(disagg_pd_policy);
+DECLARE_string(load_balance_policy);
 
 DECLARE_int32(detect_disconnected_instance_interval);
 

xllm_service/common/hash_util.cpp

@@ -0,0 +1,62 @@
+
+#include "common/hash_util.h"
+
+#include <MurmurHash3.h>
+#include <assert.h>
+
+#include <iomanip>
+#include <iostream>
+#include <mutex>
+#include <sstream>
+#include <thread>
+
+#include "common/global_gflags.h"
+
+namespace xllm_service {
+
+void murmur_hash3(const uint8_t* pre_hash_value,
+                  const Slice<int32_t>& token_ids,
+                  uint8_t* hash_value) {
+  if (pre_hash_value == nullptr) {
+    MurmurHash3_x64_128(reinterpret_cast<const void*>(token_ids.data()),
+                        sizeof(int32_t) * token_ids.size(),
+                        FLAGS_murmur_hash3_seed,
+                        hash_value);
+  } else {
+    uint8_t key[1024];
+
+    int32_t data_len =
+        sizeof(int32_t) * token_ids.size() + MURMUR_HASH3_VALUE_LEN;
+    assert(sizeof(key) > data_len);
+
+    memcpy(key, pre_hash_value, MURMUR_HASH3_VALUE_LEN);
+    memcpy(key + MURMUR_HASH3_VALUE_LEN,
+           reinterpret_cast<const void*>(token_ids.data()),
+           sizeof(int32_t) * token_ids.size());
+
+    // print_hex_array(key, data_len);
+    MurmurHash3_x64_128(reinterpret_cast<const void*>(key),
+                        data_len,
+                        FLAGS_murmur_hash3_seed,
+                        hash_value);
+  }
+}
+
+void print_hex_array(uint8_t* array) {
+  for (size_t i = 0; i < MURMUR_HASH3_VALUE_LEN; ++i) {
+    unsigned char uc = static_cast<unsigned char>(array[i]);
+    std::cout << std::hex << std::setw(2) << std::setfill('0')
+              << static_cast<int>(uc);
+
+    if (i % MURMUR_HASH3_VALUE_LEN == MURMUR_HASH3_VALUE_LEN - 1) {
+      std::cout << std::endl;
+    }
+
+    else {
+      std::cout << " ";
+    }
+  }
+  std::cout << std::dec << std::endl;
+}
+
+}  // namespace xllm_service

xllm_service/common/hash_util.h

@@ -0,0 +1,59 @@
+#pragma once
+
+#include <string.h>
+
+#include <cstdint>
+#include <string>
+#include <unordered_set>
+#include <vector>
+
+#include "common/slice.h"
+
+namespace xllm_service {
+constexpr uint32_t MURMUR_HASH3_VALUE_LEN = 16;
+
+struct Murmur3Key {
+  uint8_t data[MURMUR_HASH3_VALUE_LEN];
+
+  Murmur3Key() {}
+  Murmur3Key(const uint8_t* const input_data) {
+    memcpy(data, input_data, MURMUR_HASH3_VALUE_LEN);
+  }
+  Murmur3Key(const char* const input_data) {
+    memcpy(data, input_data, MURMUR_HASH3_VALUE_LEN);
+  }
+
+  std::string to_string() const {
+    return std::string(reinterpret_cast<const char*>(data),
+                       MURMUR_HASH3_VALUE_LEN);
+  }
+
+  bool operator==(const Murmur3Key& other) {
+    return strncmp(reinterpret_cast<const char*>(data),
+                   reinterpret_cast<const char*>(other.data),
+                   MURMUR_HASH3_VALUE_LEN);
+  }
+};
+
+struct FixedStringKeyHash {
+  size_t operator()(const Murmur3Key& key) const {
+    return std::hash<std::string_view>()(std::string_view(
+        reinterpret_cast<const char*>(key.data), sizeof(key.data)));
+  }
+};
+
+struct FixedStringKeyEqual {
+  bool operator()(const Murmur3Key& left, const Murmur3Key& right) const {
+    return strncmp(reinterpret_cast<const char*>(left.data),
+                   reinterpret_cast<const char*>(right.data),
+                   sizeof(left.data)) == 0;
+  }
+};
+
+void print_hex_array(uint8_t* array);
+
+void murmur_hash3(const uint8_t* pre_hash_value,
+                  const Slice<int32_t>& token_ids,
+                  uint8_t* hash_value);
+
+}  // namespace xllm_service