refactor: unify the handling of requests and responses into the scheduler.

Author: JimHsiung

xllm_service/CMakeLists.txt

@@ -8,6 +8,7 @@ add_subdirectory(rpc_service)
 add_subdirectory(tokenizer)
 add_subdirectory(chat_template)
 add_subdirectory(http_service)
+add_subdirectory(scheduler)
 
 cc_binary(
   NAME

xllm_service/common/CMakeLists.txt

@@ -32,5 +32,6 @@ cc_library(
     gflags::gflags
     nlohmann_json::nlohmann_json
     SMHasherSupport
+    proto_xllm
 )
 add_dependencies(common brpc-static)

xllm_service/common/call_data.h

@@ -19,12 +19,16 @@ limitations under the License.
 #include <brpc/controller.h>
 #include <butil/iobuf.h>
 #include <glog/logging.h>
-#include <grpcpp/grpcpp.h>
 #include <json2pb/pb_to_json.h>
 
 #include <functional>
 #include <string>
 
+#include "chat.pb.h"
+#include "completion.pb.h"
+
+namespace xllm_service {
+
 // Interface for the classes that are used to handle grpc requests.
 class CallData {
  public:
@@ -111,28 +115,13 @@ class StreamCallData : public CallData {
     return true;
   }
 
-  bool finish_with_error(const grpc::StatusCode& code,
-                         const std::string& error_message) {
-    if (!stream_) {
-      controller_->SetFailed(error_message);
-
-    } else {
-      io_buf_.clear();
-      io_buf_.append(error_message);
-      pa_->Write(io_buf_);
-    }
-
-    return true;
-  }
-
-  bool write_and_finish(Response& response,
-                        grpc::Status grpc_status = grpc::Status::OK) {
+  bool write_and_finish(Response& response) {
     butil::IOBufAsZeroCopyOutputStream json_output(
         &controller_->response_attachment());
     std::string err_msg;
     if (!json2pb::ProtoMessageToJson(
             response, &json_output, json_options_, &err_msg)) {
-      return finish_with_error(grpc::StatusCode::INTERNAL, err_msg);
+      return finish_with_error(err_msg);
     }
 
     if (trace_callback_) {
@@ -229,3 +218,11 @@ class StreamCallData : public CallData {
   json2pb::Pb2JsonOptions json_options_;
   std::function<void(const std::string&)> trace_callback_;
 };
+
+using CompletionCallData = StreamCallData<llm::proto::CompletionRequest,
+                                          llm::proto::CompletionResponse>;
+
+using ChatCallData =
+    StreamCallData<llm::proto::ChatRequest, llm::proto::ChatResponse>;
+
+}  // namespace xllm_service

xllm_service/common/global_gflags.cpp

@@ -15,9 +15,9 @@ limitations under the License.
 
 #include "common/global_gflags.h"
 
-DEFINE_string(http_server_host,
+DEFINE_string(server_host,
               "",
-              "Http server listen address, may be IPV4/IPV6/UDS."
+              "Server listen address, may be IPV4/IPV6/UDS."
               " If this is set, the flag port will be ignored");
 
 DEFINE_int32(http_server_port, 8888, "Port for xllm http service to listen on");
@@ -60,10 +60,6 @@ DEFINE_int32(max_concurrency,
              128,
              "Limit number of requests processed in parallel");
 
-DEFINE_string(test_instance_addr,
-              "0.0.0.0:9999",
-              "Xllm instance listen addr for testing.");
-
 DEFINE_int32(timeout_ms,
              -1,
              "Max duration of bRPC Channel. -1 means wait indefinitely.");
@@ -92,6 +88,4 @@ DEFINE_int32(block_size,
 
 DEFINE_string(tokenizer_path, "", "tokenizer config path.");
 
-DEFINE_string(model_type, "", "model type.");
-
 DEFINE_bool(enable_request_trace, false, "Whether to enable request trace");

xllm_service/common/global_gflags.h

@@ -17,7 +17,7 @@ limitations under the License.
 
 #include <gflags/gflags.h>
 
-DECLARE_string(http_server_host);
+DECLARE_string(server_host);
 
 DECLARE_int32(http_server_port);
 
@@ -37,8 +37,6 @@ DECLARE_int32(rpc_server_max_concurrency);
 
 DECLARE_uint32(murmur_hash3_seed);
 
-DECLARE_string(test_instance_addr);
-
 DECLARE_int32(timeout_ms);
 
 DECLARE_string(listen_addr);
@@ -61,6 +59,4 @@ DECLARE_int32(block_size);
 
 DECLARE_string(tokenizer_path);
 
-DECLARE_string(model_type);
-
 DECLARE_bool(enable_request_trace);

xllm_service/common/options.h

@@ -0,0 +1,76 @@
+/* Copyright 2025 The xLLM Authors. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    https://github.com/jd-opensource/xllm-service/blob/main/LICENSE
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#pragma once
+
+#include <string>
+
+#include "common/macros.h"
+
+namespace xllm_service {
+
+class Options {
+ public:
+  Options() = default;
+  ~Options() = default;
+
+  // http server options
+  PROPERTY(std::string, server_host);
+
+  PROPERTY(int32_t, http_port) = 9998;
+
+  PROPERTY(int32_t, http_idle_timeout_s) = -1;
+
+  PROPERTY(int32_t, http_num_threads) = 32;
+
+  PROPERTY(int32_t, http_max_concurrency) = 0;
+
+  // rpc server options
+  PROPERTY(int32_t, rpc_port) = 9999;
+
+  PROPERTY(int32_t, rpc_idle_timeout_s) = -1;
+
+  PROPERTY(int32_t, rpc_num_threads) = 32;
+
+  PROPERTY(int32_t, rpc_max_concurrency) = 0;
+
+  PROPERTY(int32_t, num_threads) = 32;
+
+  PROPERTY(int32_t, max_concurrency) = 32;
+
+  PROPERTY(int32_t, timeout_ms) = 32;
+
+  // instance manager options
+  PROPERTY(std::string, etcd_addr);
+
+  PROPERTY(int32_t, detect_disconnected_instance_interval) = 15;
+
+  // scheduler options
+  PROPERTY(std::string, load_balance_policy);
+
+  PROPERTY(int32_t, block_size) = 128;
+
+  PROPERTY(uint32_t, murmur_hash3_seed) = 1024;
+
+  PROPERTY(std::string, service_name);
+
+  // tokenizer options
+  PROPERTY(std::string, tokenizer_path);
+
+  // trace options
+  PROPERTY(bool, enable_request_trace) = false;
+};
+
+}  // namespace xllm_service

xllm_service/common/types.h

@@ -36,26 +36,6 @@ using Murmur3KeyCacheMap = std::unordered_map<Murmur3Key,
                                               FixedStringKeyHash,
                                               FixedStringKeyEqual>;
 
-struct HttpServiceConfig {
-  int num_threads = 16;
-  int timeout_ms = -1;
-  std::string test_instance_addr = "";
-  bool enable_request_trace = false;
-};
-
-struct RpcServiceConfig {
-  std::string etcd_addr = "";
-  std::string load_balance_policy = "";
-  int detect_disconnected_instance_interval = 15;  // seconds
-  std::string service_name = "";
-};
-
-struct ModelConfig {
-  int32_t block_size = 16;
-  std::string model_type = "chatglm";
-  std::string tokenizer_path = "";
-};
-
 struct Routing {
   std::string prefill_name;
   std::string decode_name;

xllm_service/common/xllm/output.h

@@ -126,4 +126,7 @@ inline std::optional<std::string> to_string(FinishReason reason) {
 }
 
 }  // namespace llm
+
+using OutputCallback = std::function<bool(llm::RequestOutput output)>;
+
 }  // namespace xllm_service

xllm_service/http_service/CMakeLists.txt

@@ -12,7 +12,7 @@ cc_library(
     request_tracer.cpp
   DEPS
     :common
-    :xllm_rpc_service
+    :scheduler
     absl::random_random
     absl::synchronization
     glog::glog

xllm_service/http_service/main.cpp

@@ -19,6 +19,7 @@ limitations under the License.
 #include <grpcpp/grpcpp.h>
 
 #include "common/global_gflags.h"
+#include "common/options.h"
 #include "http_service/service.h"
 
 int main(int argc, char** argv) {
@@ -31,11 +32,8 @@ int main(int argc, char** argv) {
 
   LOG(INFO) << "Starting xllm http service, port: " << FLAGS_port;
 
-  xllm_service::HttpServiceConfig config;
-  config.num_threads = FLAGS_num_threads;
-  config.timeout_ms = FLAGS_timeout_ms;
-  config.test_instance_addr = FLAGS_test_instance_addr;
-  xllm_service::XllmHttpServiceImpl service_impl(config);
+  xllm_service::Options service_options;
+  xllm_service::XllmHttpServiceImpl service_impl(service_options, nullptr);
 
   // register http methods here
   brpc::Server server;