feat: add roundrobin policy.

Author: Kang-Meng

xllm_service/common/global_gflags.cpp

@@ -78,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

@@ -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/types.h

@@ -45,7 +45,7 @@ struct HttpServiceConfig {
 
 struct RpcServiceConfig {
   std::string etcd_addr = "";
-  std::string disagg_pd_policy = "";
+  std::string load_balance_policy = "";
   int detect_disconnected_instance_interval = 15;  // seconds
   std::string service_name = "";
 };
@@ -163,6 +163,8 @@ struct InstanceMetaInfo {
   // latest heatbeat timestamp
   uint64_t latest_timestamp = 0;
 
+  uint64_t instance_index = -1;
+
   nlohmann::json serialize_to_json() const {
     nlohmann::json json_val;
     json_val["name"] = name;

xllm_service/master.cpp

@@ -31,7 +31,7 @@ Master::Master(const ServerOptions& server_options)
   }
   RpcServiceConfig rpc_config;
   rpc_config.etcd_addr = server_options.etcd_addr;
-  rpc_config.disagg_pd_policy = server_options.disagg_pd_policy;
+  rpc_config.load_balance_policy = server_options.load_balance_policy;
   rpc_config.detect_disconnected_instance_interval =
       server_options.detect_disconnected_instance_interval;
 
@@ -213,7 +213,7 @@ int main(int argc, char* argv[]) {
   server_options.rpc_num_threads = FLAGS_rpc_server_num_threads;
   server_options.rpc_max_concurrency = FLAGS_rpc_server_max_concurrency;
   server_options.etcd_addr = FLAGS_etcd_addr;
-  server_options.disagg_pd_policy = FLAGS_disagg_pd_policy;
+  server_options.load_balance_policy = FLAGS_load_balance_policy;
   server_options.detect_disconnected_instance_interval =
       FLAGS_detect_disconnected_instance_interval;
   server_options.enable_request_trace = FLAGS_enable_request_trace;

xllm_service/master.h

@@ -43,7 +43,7 @@ struct ServerOptions {
   int32_t rpc_num_threads = 32;
   int32_t rpc_max_concurrency = 128;
   std::string etcd_addr = "";
-  std::string disagg_pd_policy = "RR";
+  std::string load_balance_policy = "RR";
   int32_t detect_disconnected_instance_interval = 15;
   int32_t block_size = 16;
   std::string model_type = "chatglm";

xllm_service/rpc_service/CMakeLists.txt

@@ -1,32 +1,28 @@
 include(cc_binary)
 include(cc_library)
 include(cc_test)
+add_subdirectory(etcd_client)
+add_subdirectory(managers)
 add_subdirectory(loadbalance_policy)
 
 cc_library(
   NAME
   xllm_rpc_service
   HDRS
     scheduler.h
-    etcd_client.h
-    instance_mgr.h
-    global_kvcache_mgr.h
     response_handler.h
     service.h
   SRCS
     scheduler.cpp
-    etcd_client.cpp
-    instance_mgr.cpp
-    global_kvcache_mgr.cpp
     response_handler.cpp
     service.cpp
   DEPS
     :common
+    :etcd_client
+    :managers
     :loadbalance_policy
     absl::random_random
     absl::strings
-    cpprest
-    etcd-cpp-api
     glog::glog
     nlohmann_json::nlohmann_json
     proto::proto_rpc_service

xllm_service/rpc_service/etcd_client/CMakeLists.txt

@@ -0,0 +1,17 @@
+include(cc_library)
+include(cc_test)
+
+cc_library(
+  NAME
+  etcd_client
+  HDRS
+    etcd_client.h
+  SRCS
+    etcd_client.cpp
+  DEPS
+    :common
+    cpprest
+    etcd-cpp-api
+    glog::glog
+    nlohmann_json::nlohmann_json
+)

xllm_service/rpc_service/etcd_client.cpp renamed to xllm_service/rpc_service/etcd_client/etcd_client.cpp

@@ -7,8 +7,12 @@ cc_library(
   loadbalance_policy
   HDRS
     loadbalance_policy.h
+    round_robin.h
+    cache_aware_routing.h
   SRCS
-    loadbalance_policy.cpp
+    round_robin.cpp
+    cache_aware_routing.cpp
   DEPS
     :common
+    :managers
 )