feat: implement single-process mode on npu device.

xllm/core/common/options.h

@@ -124,6 +124,8 @@ class Options {
 
   PROPERTY(int32_t, node_rank) = 0;
 
+  PROPERTY(bool, enable_single_process) = false;
+
   PROPERTY(int32_t, dp_size) = 1;
 
   PROPERTY(int32_t, cp_size) = 1;

xllm/core/distributed_runtime/collective_service.cpp

@@ -20,6 +20,8 @@ limitations under the License.
 
 #include <vector>
 
+#include "util/net.h"
+
 namespace xllm {
 
 CollectiveService::CollectiveService(int dp_group_num,
@@ -38,6 +40,27 @@ CollectiveService::CollectiveService(int dp_group_num,
     root_infos_.push_back(root_info);
   }
 #endif
+
+  // Pre-allocate free TCPStore ports for the worst-case communication group
+  // count. The communicator computes a slot index per group from
+  // (group_kind, group_index) — see ProcessGroupSlots in
+  // collective_communicator.cpp — which is bounded by total_num_ slots
+  // across the world/tp/single_rank/dp_local/moe_tp/moe_ep/encoder_dp
+  // sections. We over-allocate by a small constant to absorb future group
+  // additions; the unused ports are released right after binding.
+  //
+  // OS-allocated free ports avoid collisions with random outgoing TCP source
+  // ports inside the local ephemeral range (1024-65535 on common Linux
+  // hosts), which the previous "master_port + offset" scheme could not.
+  const int port_slot_count = std::max(1, 4 * total_num_ + 8);
+  group_ports_.reserve(port_slot_count);
+  for (int i = 0; i < port_slot_count; ++i) {
+    const int port = net::get_local_free_port();
+    CHECK_GT(port, 0) << "Failed to allocate free TCP port for process-group "
+                         "rendezvous; slot="
+                      << i;
+    group_ports_.push_back(port);
+  }
 }
 
 void CollectiveService::Sync(::google::protobuf::RpcController* controller,
@@ -55,6 +78,12 @@ void CollectiveService::Sync(::google::protobuf::RpcController* controller,
 #if defined(USE_NPU)
   to_proto_list(root_infos_, response);
 #endif
+  // Send the master-allocated ports to every worker. All ranks see the same
+  // ordered list, so each worker resolves its process-group slots to the
+  // same TCPStore endpoints the master will bind on.
+  for (const int port : group_ports_) {
+    response->add_group_ports(port);
+  }
 }
 
 std::unordered_map<int32_t, std::string> CollectiveService::wait() {

xllm/core/distributed_runtime/collective_service.h

@@ -40,6 +40,12 @@ class CollectiveService : public proto::Collective {
   // wait all worker connected
   std::unordered_map<int32_t, std::string> wait();
 
+  // Master-allocated free TCPStore ports, in the canonical group order shared
+  // with workers (see CollectiveCommunicator::create_process_groups). The
+  // master node consumes this list locally; workers receive the same list
+  // through CommUniqueIdList.group_ports during Sync.
+  const std::vector<int32_t>& group_ports() const { return group_ports_; }
+
  private:
   DISALLOW_COPY_AND_ASSIGN(CollectiveService);
 
@@ -57,6 +63,10 @@ class CollectiveService : public proto::Collective {
 #endif
   std::mutex mutex_;
   std::unordered_map<int32_t, std::string> addrs_map_;
+  // Free TCPStore ports the master allocated up-front. Returned verbatim to
+  // every worker through Sync so each rank uses the same set of ports for
+  // process-group bootstrap.
+  std::vector<int32_t> group_ports_;
 };
 
 }  // namespace xllm

xllm/core/distributed_runtime/dist_manager.cpp

@@ -30,6 +30,8 @@ limitations under the License.
 #include "remote_worker.h"
 #include "runtime/forward_shared_memory_manager.h"
 #include "runtime/llm_worker_impl.h"
+#include "runtime/worker.h"
+#include "runtime/worker_client.h"
 #include "server/xllm_server_registry.h"
 #include "shm_channel.h"
 #include "util/net.h"
@@ -38,11 +40,22 @@ namespace xllm {
 DistManager::DistManager(const runtime::Options& options)
     : server_name_("CollectiveServer") {
   auto master_node_addr = options.master_node_addr().value_or("");
-  if (!master_node_addr.empty()) {
+  if (options.enable_single_process()) {
+    // Single-process mode: one OS process drives all local devices (one worker
+    // thread per device, sharing one HcclCommInitAll world). Independent of
+    // master_node_addr — single-node single-process needs none. Multi-node
+    // single-process (nnodes > 1) is not yet supported and is rejected inside
+    // setup_single_node_workers.
+    setup_single_node_workers(options);
+  } else if (!master_node_addr.empty()) {
     server_name_.append(std::to_string(options.server_idx()));
     setup_multi_node_workers(options, master_node_addr);
   } else {
-    LOG(FATAL) << "master_node_addr is empty.";
+    LOG(FATAL)
+        << "master_node_addr is empty and single-process mode is not "
+           "enabled. Set XLLM_ENABLE_SINGLE_PROCESS=1 for single-process "
+           "serving, or --master_node_addr for multi-process/multi-node "
+           "serving.";
   }
 }
 
@@ -51,7 +64,7 @@ DistManager::~DistManager() {
   HealthCheckManager::instance().stop_health_check_thread();
 
   XllmServer* collective_server =
-      ServerRegistry::get_instance().get_server(server_name_);
+      ServerRegistry::get_instance().try_get_server(server_name_);
   if (collective_server != nullptr) {
     collective_server->stop();
 
@@ -142,6 +155,110 @@ void setup_numa_affinity_and_isolation(
 
 }  // namespace
 
+void DistManager::setup_single_node_workers(const runtime::Options& options) {
+  const auto& devices = options.devices();
+  CHECK_GT(devices.size(), 0)
+      << "At least one device is required in single-node mode.";
+  CHECK_EQ((devices.size() % options.dp_size()), 0)
+      << "Device size must be divisible by dp size in single-node serving "
+         "mode.";
+  CHECK_EQ(options.nnodes(), 1)
+      << "Multi-node single-process serving (nnodes = " << options.nnodes()
+      << ") is not yet supported. Run single-node single-process "
+         "(XLLM_ENABLE_SINGLE_PROCESS=1 with nnodes=1), or use the "
+         "multi-process launcher (--master_node_addr, one process per device) "
+         "for multi-node.";
+  CHECK_EQ(options.node_rank(), 0)
+      << "Single-process serving requires node_rank=0 (got "
+      << options.node_rank() << ").";
+
+  const std::string& backend = options.backend();
+  CHECK(backend.empty() || backend == "llm")
+      << "Single-node serving currently supports only the llm backend, got: "
+      << backend;
+
+  const int32_t world_size = static_cast<int32_t>(devices.size());
+  const int32_t dp_size = options.dp_size();
+  const int32_t cp_size = options.cp_size();
+  const int32_t ep_size = options.ep_size();
+  const int32_t dp_local_tp_size = world_size / dp_size;
+
+  LOG(INFO) << "Single-node serving world_size = " << world_size
+            << ", dp_size = " << dp_size << ", cp_size = " << cp_size
+            << ", ep_size = " << ep_size << ", tp_size = " << dp_local_tp_size;
+
+  // initialize process groups if there are multiple devices
+  if (devices.size() > 1) {
+#if defined(USE_NPU)
+    // Single-process multi-device shares one HCCL world (HcclCommInitAll) and
+    // routes tensor-parallel collectives through the C++ ProcessGroup. Only the
+    // TORCH kernel backend uses that path; the ATB backend drives its own
+    // process-global comm manager, which cannot rendezvous two ranks inside one
+    // process. Fail fast with actionable guidance instead of hanging.
+    CHECK_EQ(options.npu_kernel_backend(), "TORCH")
+        << "Single-node single-process mode with multiple NPU devices requires "
+           "--npu_kernel_backend=TORCH (got '"
+        << options.npu_kernel_backend()
+        << "'). The ATB backend does not support single-process multi-device "
+           "yet; use the multi-process launcher (set --master_node_addr) for "
+           "ATB.";
+#endif
+    process_groups_ = parallel_state::create_local_process_groups(
+        devices, /*options=*/options);
+  }
+
+  // For data parallelism with dp_size > 1 and dp_size < world_size, build
+  // dp-local process groups so each dp group has its own communication
+  // channel. Mirrors the old setup_single_node_workers behaviour.
+  if (dp_size > 1 && dp_size < world_size) {
+    dp_local_process_groups_.reserve(dp_size);
+    for (int32_t dp_rank = 0; dp_rank < dp_size; ++dp_rank) {
+      auto begin_it = devices.begin();
+      std::advance(begin_it, dp_rank * dp_local_tp_size);
+      auto end_it = devices.begin();
+      std::advance(end_it, (dp_rank + 1) * dp_local_tp_size);
+      std::vector<torch::Device> dp_local_devices(begin_it, end_it);
+      dp_local_process_groups_.emplace_back(
+          parallel_state::create_local_process_groups(dp_local_devices,
+                                                      options));
+    }
+  }
+
+  WorkerType worker_type =
+      (options.task_type() == "generate") ? WorkerType::LLM : WorkerType::ELM;
+
+  for (size_t i = 0; i < devices.size(); ++i) {
+    const int32_t rank = static_cast<int32_t>(i);
+    ProcessGroup* pg = (world_size > 1) ? process_groups_[i].get() : nullptr;
+    ProcessGroup* dp_local_pg =
+        (dp_size > 1 && dp_size < world_size)
+            ? dp_local_process_groups_[i / dp_local_tp_size]
+                                      [i % dp_local_tp_size]
+                                          .get()
+            : nullptr;
+
+    ParallelArgs parallel_args(rank,
+                               world_size,
+                               dp_size,
+                               cp_size,
+                               /*process_group=*/pg,
+                               ep_size);
+    parallel_args.dp_local_process_group_ = dp_local_pg;
+    // For single-node single-process mode the language-model layers expect a
+    // tensor-parallel group. With dp=ep=cp=1 the global process group also
+    // serves as the tp/cp/single-rank group.
+    parallel_args.tp_group_ = pg;
+    parallel_args.cp_group_ = pg;
+    parallel_args.single_rank_group_ = pg;
+    parallel_args.moe_tp_group_ = pg;
+
+    workers_.emplace_back(std::make_unique<Worker>(
+        parallel_args, devices[i], options, worker_type));
+    worker_clients_.emplace_back(
+        std::make_shared<WorkerClient>(workers_.back().get(), options));
+  }
+}
+
 void DistManager::setup_multi_node_workers(
     const runtime::Options& options,
     const std::string& master_node_addr) {

xllm/core/distributed_runtime/dist_manager.h

@@ -34,6 +34,7 @@ class DistManager {
 
  private:
   DISALLOW_COPY_AND_ASSIGN(DistManager);
+  void setup_single_node_workers(const runtime::Options& options);
   void setup_multi_node_workers(const runtime::Options& options,
                                 const std::string& master_node_addr);
 

xllm/core/distributed_runtime/llm_engine.cpp

@@ -70,9 +70,6 @@ LLMEngine::LLMEngine(const runtime::Options& options,
   InterruptionBus::get_instance().subscribe([this](bool interrupted) {
     this->layer_forward_interrupted_ = interrupted;
   });
-  auto master_node_addr = options.master_node_addr().value_or("");
-  CHECK(!master_node_addr.empty())
-      << " LLM need to set master node addr, Please set --master_node_addr.";
   const auto& devices = options_.devices();
   // initialize device monitor
   DeviceMonitor::get_instance().initialize(devices);

xllm/core/distributed_runtime/master.cpp

@@ -203,15 +203,26 @@ Master::Master(const Options& options, EngineType type)
   // cards (honoring *_VISIBLE_DEVICES) and select the single card this process
   // owns by its node_rank, so `devices` holds exactly the card this process
   // uses -- mirroring the historical single-element devices semantics.
+  //
+  // Single-process serving instead drives every visible card from one process
+  // (one worker thread per device), so it takes all visible devices and its
+  // world size is the device count rather than the node count.
   const auto visible_devices = DeviceNameUtils::parse_devices("auto");
-  CHECK_LT(options_.node_rank(), static_cast<int32_t>(visible_devices.size()))
-      << "node_rank " << options_.node_rank()
-      << " exceeds the number of visible devices " << visible_devices.size()
-      << ". Ensure *_VISIBLE_DEVICES exposes all cards used across processes.";
-  const std::vector<torch::Device> devices = {
-      visible_devices[options_.node_rank()]};
-  // World size is the node count (one worker per process).
-  const int32_t global_world_size = options_.nnodes();
+  std::vector<torch::Device> devices;
+  int32_t global_world_size;
+  if (options_.enable_single_process()) {
+    devices = visible_devices;
+    global_world_size = static_cast<int32_t>(visible_devices.size());
+  } else {
+    CHECK_LT(options_.node_rank(), static_cast<int32_t>(visible_devices.size()))
+        << "node_rank " << options_.node_rank()
+        << " exceeds the number of visible devices " << visible_devices.size()
+        << ". Ensure *_VISIBLE_DEVICES exposes all cards used across "
+           "processes.";
+    devices = {visible_devices[options_.node_rank()]};
+    // World size is the node count (one worker per process).
+    global_world_size = options_.nnodes();
+  }
   std::string cp_model_type;
   if (options_.cp_size() > 1 && Platform::uses_model_cp_partition()) {
     cp_model_type = util::get_model_type(model_path, options_.backend());
@@ -328,6 +339,7 @@ Master::Master(const Options& options, EngineType type)
         .master_node_addr(options.master_node_addr())
         .nnodes(options.nnodes())
         .node_rank(options.node_rank())
+        .enable_single_process(options.enable_single_process())
         .dp_size(options.dp_size())
         .ep_size(options.ep_size())
         .max_tokens_per_batch(options_.max_tokens_per_batch())
@@ -386,6 +398,7 @@ Master::Master(const Options& options, EngineType type)
         .master_node_addr(options.master_node_addr())
         .nnodes(options.nnodes())
         .node_rank(options.node_rank())
+        .enable_single_process(options.enable_single_process())
         .dp_size(options.dp_size())
         .ep_size(options.ep_size())
         .cp_size(options_.cp_size())
@@ -440,6 +453,7 @@ Master::Master(const Options& options, EngineType type)
         .master_node_addr(options_.master_node_addr())
         .nnodes(options_.nnodes())
         .node_rank(options_.node_rank())
+        .enable_single_process(options_.enable_single_process())
         .dp_size(options_.dp_size())
         .ep_size(options_.ep_size())
         .cp_size(options_.cp_size())
@@ -506,6 +520,7 @@ Master::Master(const Options& options, EngineType type)
         .master_node_addr(options_.master_node_addr())
         .nnodes(options_.nnodes())
         .node_rank(options_.node_rank())
+        .enable_single_process(options_.enable_single_process())
         .dp_size(options_.dp_size())
         .ep_size(options_.ep_size())
         .cp_size(options_.cp_size())
@@ -543,6 +558,7 @@ Master::Master(const Options& options, EngineType type)
         .output_shm_size(options_.output_shm_size() * 1024 * 1024)
         .is_local(options_.is_local())
         .node_rank(options_.node_rank())
+        .enable_single_process(options_.enable_single_process())
         .enable_schedule_overlap(options_.enable_schedule_overlap())
         .dp_size(options_.dp_size())
         .ep_size(options_.ep_size())

xllm/core/distributed_runtime/worker_server.cpp

@@ -176,6 +176,16 @@ void WorkerServer::create_server(const runtime::Options& options,
     comm = std::move(common_comm);
   }
 
+  // Forward master-allocated TCPStore ports (if any) so the communicator
+  // binds on OS-chosen free ports instead of fixed master_port + offset
+  // values that can collide with random outgoing TCP source ports inside
+  // the local ephemeral range.
+  std::vector<int32_t> group_ports(uids.group_ports().begin(),
+                                   uids.group_ports().end());
+  if (!group_ports.empty()) {
+    comm->set_group_ports(std::move(group_ports));
+  }
+
   comm->create_process_groups(master_node_addr, device);
   parallel_args = comm->parallel_args();
 

xllm/core/framework/config/distributed_config.cpp

@@ -17,11 +17,13 @@ limitations under the License.
 
 #include "core/common/global_flags.h"
 #include "core/framework/config/config_utils.h"
+#include "core/util/env_var.h"
 
 DEFINE_string(master_node_addr,
-              "127.0.0.1:19888",
+              "",
               "The master address for multi-node distributed serving(e.g. "
-              "10.18.1.1:9999).");
+              "10.18.1.1:9999). Required for multi-process/multi-node serving; "
+              "leave empty only when XLLM_ENABLE_SINGLE_PROCESS is set.");
 
 DEFINE_string(
     xtensor_master_node_addr,
@@ -53,6 +55,11 @@ void DistributedConfig::from_flags() {
   XLLM_CONFIG_ASSIGN_FROM_FLAG(xtensor_master_node_addr);
   XLLM_CONFIG_ASSIGN_FROM_FLAG(nnodes);
   XLLM_CONFIG_ASSIGN_FROM_FLAG(node_rank);
+  // Single-process mode is opt-in via environment variable rather than a
+  // gflag, so it can be set from the launch environment without threading a
+  // CLI flag through every entrypoint.
+  enable_single_process(util::get_bool_env("XLLM_ENABLE_SINGLE_PROCESS",
+                                           enable_single_process()));
   XLLM_CONFIG_ASSIGN_FROM_FLAG(etcd_addr);
   XLLM_CONFIG_ASSIGN_FROM_FLAG(etcd_namespace);
   XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_service_routing);
@@ -66,6 +73,7 @@ void DistributedConfig::from_json(const JsonReader& json) {
   XLLM_CONFIG_ASSIGN_FROM_JSON(nnodes);
   // don't read rank-related config
   // XLLM_CONFIG_ASSIGN_FROM_JSON(node_rank);
+  // enable_single_process is sourced from the environment, not JSON.
   XLLM_CONFIG_ASSIGN_FROM_JSON(etcd_addr);
   XLLM_CONFIG_ASSIGN_FROM_JSON(etcd_namespace);
   XLLM_CONFIG_ASSIGN_FROM_JSON(enable_service_routing);
@@ -84,6 +92,8 @@ void DistributedConfig::append_config_json(
   // don't dump rank-related config
   //   APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT(
   //       config_json, default_config, node_rank);
+  APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT(
+      config_json, default_config, enable_single_process);
   APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT(
       config_json, default_config, etcd_addr);
   APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT(