parser hccl config

mindspore_serving/ccsrc/worker/distributed_worker/agent_startup.cc

@@ -16,6 +16,7 @@
 #include "worker/distributed_worker/agent_startup.h"
 #include <fstream>
 #include "worker/distributed_worker/notify_distributed/notify_worker.h"
+#include "common/grpc_server.h"
 
 namespace mindspore {
 namespace serving {
@@ -26,7 +27,7 @@ WorkerAgentStartUp &WorkerAgentStartUp::Instance() {
 }
 
 Status WorkerAgentStartUp::GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port) {
-  return FAILED;
+  return GrpcNotifyDistributeWorker::GetAgentsConfigsFromWorker(worker_ip, worker_port, &config_);
 }
 
 Status WorkerAgentStartUp::GetDistributedServableConfig(DistributedServableConfig *config) {

mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.cc

@@ -59,5 +59,51 @@ grpc::Status MSDistributedImpl::AgentFailed(grpc::ServerContext *context, const
   }
   return grpc::Status::OK;
 }
+
+grpc::Status MSDistributedImpl::AgentConfigAcquire(grpc::ServerContext *context,
+                                                   const proto::AgentConfigAcquireRequest *request,
+                                                   proto::AgentConfigAcquireReply *reply) {
+  Status status(FAILED);
+  DistributedServableConfig agent_config;
+  status = servable_->GetDistributedServableConfig(&agent_config);
+  if (status != SUCCESS) {
+    MSI_LOG(ERROR) << "Get distributed servable config failed";
+    return grpc::Status::CANCELLED;
+  }
+  if (agent_config.rank_list.empty()) {
+    MSI_LOG(ERROR) << "Get distributed servable config failed, config_ not init";
+    return grpc::Status::CANCELLED;
+  }
+
+  MSI_LOG(INFO) << "Begin to set DistributedServableConfig info in reply message";
+  // set reply message:AgentConfigAcquireReply, parameter:rank_table_content
+  reply->set_rank_table_content(agent_config.rank_table_content);
+  // set reply message:AgentConfigAcquireReply, parameter:rank_list
+  auto &agent_rank_list = agent_config.rank_list;
+  for (auto &agent_rank : agent_rank_list) {
+    auto rank_list = reply->add_rank_list();
+    rank_list->set_ip(agent_rank.ip);
+    rank_list->set_device_id(agent_rank.device_id);
+  }
+  // set reply message:AgentConfigAcquireReply, parameter:common_meta
+  auto reply_common_meta = reply->mutable_common_meta();
+  reply_common_meta->set_servable_name(agent_config.common_meta.servable_name);
+  reply_common_meta->set_with_batch_dim(agent_config.common_meta.with_batch_dim);
+  auto &without_batch_dim_inputs_list = agent_config.common_meta.without_batch_dim_inputs;
+  for (auto &without_batch_dim_input : without_batch_dim_inputs_list) {
+    reply_common_meta->add_without_batch_dim_inputs(without_batch_dim_input);
+  }
+  reply_common_meta->set_inputs_count(agent_config.common_meta.inputs_count);
+  reply_common_meta->set_outputs_count(agent_config.common_meta.outputs_count);
+
+  // set reply message:AgentConfigAcquireReply, parameter:distributed_meta
+  auto reply_distributed_meta = reply->mutable_distributed_meta();
+  reply_distributed_meta->set_rank_size(agent_config.distributed_meta.rank_size);
+  reply_distributed_meta->set_stage_size(agent_config.distributed_meta.stage_size);
+  MSI_LOG(INFO) << "Success to set DistributedServableConfig info in reply message";
+
+  return grpc::Status::OK;
+}
+
 }  // namespace serving
 }  // namespace mindspore

mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_process.h

@@ -43,6 +43,8 @@ class MSDistributedImpl final : public MSWorkerImpl {
                          proto::AgentExitReply *reply) override;
   grpc::Status AgentFailed(grpc::ServerContext *context, const proto::AgentFailedRequest *request,
                            proto::AgentFailedReply *reply) override;
+  grpc::Status AgentConfigAcquire(grpc::ServerContext *context, const proto::AgentConfigAcquireRequest *request,
+                                  proto::AgentConfigAcquireReply *reply) override;
 
  private:
   std::shared_ptr<DistributedServable> servable_;

mindspore_serving/ccsrc/worker/distributed_worker/distributed_process/distributed_server.h

@@ -153,6 +153,38 @@ class WorkerAgentFailedContext : public DistributedServiceContext {
   proto::AgentFailedReply response_;
 };
 
+class WorkerAgentConfigAcquireContext : public DistributedServiceContext {
+ public:
+  WorkerAgentConfigAcquireContext(MSDistributedImpl *service_impl, proto::MSWorker::AsyncService *async_service,
+                                  grpc::ServerCompletionQueue *cq)
+      : DistributedServiceContext(service_impl, async_service, cq), responder_(&ctx_) {}
+
+  ~WorkerAgentConfigAcquireContext() = default;
+  static Status EnqueueRequest(MSDistributedImpl *service_impl, proto::MSWorker::AsyncService *async_service,
+                               grpc::ServerCompletionQueue *cq) {
+    auto call = new WorkerAgentConfigAcquireContext(service_impl, async_service, cq);
+    call->StartEnqueueRequest();
+    return SUCCESS;
+  }
+
+  void StartEnqueueRequest() override {
+    state_ = STATE::PROCESS;
+    async_service_->RequestAgentConfigAcquire(&ctx_, &request_, &responder_, cq_, cq_, this);
+  }
+
+  void HandleRequest() override {
+    EnqueueRequest(dist_service_impl_, async_service_, cq_);
+    state_ = STATE::FINISH;
+    grpc::Status status = dist_service_impl_->AgentConfigAcquire(&ctx_, &request_, &response_);
+    responder_.Finish(response_, status, this);
+  }
+
+ private:
+  grpc::ServerAsyncResponseWriter<proto::AgentConfigAcquireReply> responder_;
+  proto::AgentConfigAcquireRequest request_;
+  proto::AgentConfigAcquireReply response_;
+};
+
 class DistributedWorkerGrpcServer : public WorkerGrpcServer {
  public:
   DistributedWorkerGrpcServer(const std::string &host, int32_t port, MSDistributedImpl *service_impl)
@@ -165,6 +197,7 @@ class DistributedWorkerGrpcServer : public WorkerGrpcServer {
     WorkerAgentRegisterContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get());
     WorkerAgentExitContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get());
     WorkerAgentFailedContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get());
+    WorkerAgentConfigAcquireContext::EnqueueRequest(distributed_service_impl_, &svc_, cq_.get());
     return SUCCESS;
   }
 

mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.cc

@@ -157,7 +157,211 @@ Status DistributedServable::StartServable(const std::string &servable_directory,
   return SUCCESS;
 }
 
-Status DistributedServable::InitConfigOnStartup(const std::string &rank_table_json_file) { return FAILED; }
+std::string RealPath(const char *path) {
+  // Return absolute path when path is accessible
+  std::string res;
+  char resolved_path[PATH_MAX] = {0};
+  if (realpath(path, resolved_path) != nullptr) {
+    res = resolved_path;
+  }
+
+  return res;
+}
+
+Status DistributedServable::InitConfigOnStartup(const std::string &rank_table_json_file) {
+  std::string rank_table_json_abs_path = RealPath(rank_table_json_file.c_str());
+  if (rank_table_json_abs_path.empty()) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "failed to get realpath of：" << rank_table_json_file.c_str();
+  }
+
+  MSI_LOG(INFO) << "Begin to parser rank table json file: " << rank_table_json_file.c_str();
+  json rank_table_json;
+  std::ifstream json_file(rank_table_json_abs_path);
+  if (!json_file.is_open()) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "failed to open rank table file：" << rank_table_json_file.c_str();
+  }
+
+  json_file >> rank_table_json;
+  if (!rank_table_json.is_object()) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << rank_table_json_file.c_str() << " is not json object";
+  }
+
+  if (rank_table_json.find("group_list") != rank_table_json.end()) {
+    return ParserRankTableWithGroupList(rank_table_json_file, rank_table_json);
+  } else {
+    return ParserRankTableWithServerList(rank_table_json_file, rank_table_json);
+  }
+}
+
+json DistributedServable::ParserArrayInJson(const json &json_array, const std::string &str) {
+  json temp_array;
+  auto iter = json_array.find(str);
+  if (iter == json_array.end()) {
+    MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not find";
+    return temp_array;
+  }
+  if (!iter->is_array()) {
+    MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not array";
+    return temp_array;
+  }
+  temp_array = json_array.at(str);
+  return temp_array;
+}
+
+json DistributedServable::ParserStringInJson(const json &json_str, const std::string &str) {
+  json temp_str;
+  auto iter = json_str.find(str);
+  if (iter == json_str.end()) {
+    MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not find";
+    return temp_str;
+  }
+  if (!iter->is_string()) {
+    MSI_LOG_ERROR << "Check rank table file failed" << str << "in file is not string";
+    return temp_str;
+  }
+  temp_str = json_str.at(str);
+  return temp_str;
+}
+
+Status DistributedServable::ParserRankTableWithGroupList(const std::string &rank_table_json_file,
+                                                         const json &rank_table_json) {
+  MSI_LOG_INFO << "Begin to parser rank table with group list";
+  auto server_list = ParserArrayInJson(rank_table_json, "group_list");
+  if (server_list.empty()) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "group_list attr is empty in" << rank_table_json_file.c_str();
+  }
+
+  size_t rank_id = 0;
+  for (auto &server : server_list) {
+    auto instance_list = ParserArrayInJson(server, "instance_list");
+    if (instance_list.empty()) {
+      return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "instance_list attr is empty in" << rank_table_json_file.c_str();
+    }
+
+    for (auto &instance : instance_list) {
+      auto str_server_id = ParserStringInJson(instance, "server_id");
+      if (str_server_id.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_id attr is empty in" << rank_table_json_file.c_str();
+      }
+
+      OneRankConfig one_rank_config;
+      one_rank_config.ip = str_server_id;
+
+      auto devices = ParserArrayInJson(instance, "devices");
+      if (devices.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "devices attr is empty in" << rank_table_json_file.c_str();
+      }
+
+      auto str_device_id = ParserStringInJson(devices.at(0), "device_id");
+      if (str_device_id.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device_id attr is empty in" << rank_table_json_file.c_str();
+      }
+      auto temp_str_device_id = str_device_id.get<std::string>();
+      uint32_t temp_device_id;
+      auto status = ConvertStr2Int(rank_table_json_file, temp_str_device_id, "device_id", &temp_device_id);
+      if (status != SUCCESS) {
+        MSI_LOG_ERROR << "Convert device_id from string to int failed";
+        return status;
+      }
+
+      auto str_rank_id = ParserStringInJson(instance, "rank_id");
+      if (str_rank_id.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "rank_id attr is empty in" << rank_table_json_file.c_str();
+      }
+      auto temp_str_rank_id = str_rank_id.get<std::string>();
+      uint32_t temp_rank_id;
+      status = ConvertStr2Int(rank_table_json_file, temp_str_rank_id, "rank_id", &temp_rank_id);
+      if (status != SUCCESS) {
+        MSI_LOG_ERROR << "Convert rank_id from string to int failed";
+        return status;
+      }
+
+      if (rank_id != temp_rank_id) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS)
+               << "device size not match rank_id in" << rank_table_json_file.c_str();
+      }
+      rank_id++;
+      one_rank_config.device_id = temp_device_id;
+      config_.rank_list.push_back(one_rank_config);
+    }
+  }
+  MSI_LOG(INFO) << "Success parser rank table json file with group list and save to DistributedServableConfig";
+
+  return SUCCESS;
+}
+Status DistributedServable::ConvertStr2Int(const std::string &rank_table_json_file, const std::string &para_str,
+                                           const std::string &para_key, uint32_t *para_int) const {
+  try {
+    *para_int = std::stoi(para_str);
+  } catch (std::invalid_argument &) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS)
+           << para_key << "attr is invalid argument in" << rank_table_json_file.c_str();
+  } catch (std::out_of_range &) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS)
+           << para_key << "attr is out of range in" << rank_table_json_file.c_str();
+  } catch (...) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS)
+           << "convert" << para_key << "attr exception occurred in" << rank_table_json_file.c_str();
+  }
+  return SUCCESS;
+}
+
+Status DistributedServable::ParserRankTableWithServerList(const std::string &rank_table_json_file,
+                                                          const json &rank_table_json) {
+  MSI_LOG_INFO << "Begin to parser rank table with server list";
+  auto server_list = ParserArrayInJson(rank_table_json, "server_list");
+  if (server_list.empty()) {
+    return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_list attr is empty in" << rank_table_json_file.c_str();
+  }
+
+  size_t rank_id = 0;
+  for (auto &server : server_list) {
+    auto server_id = ParserStringInJson(server, "server_id");
+    if (server_id.empty()) {
+      return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "server_id attr is empty in" << rank_table_json_file.c_str();
+    }
+
+    auto device_list = ParserArrayInJson(server, "device");
+    if (device_list.empty()) {
+      return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device attr is empty in" << rank_table_json_file.c_str();
+    }
+
+    for (auto &device : device_list) {
+      OneRankConfig one_rank_config;
+      one_rank_config.ip = server_id;
+      auto str_device_id = ParserStringInJson(device, "device_id");
+      if (str_device_id.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "device_id attr is empty in" << rank_table_json_file.c_str();
+      }
+      auto temp_str_device_id = str_device_id.get<std::string>();
+      uint32_t temp_device_id;
+      auto status = ConvertStr2Int(rank_table_json_file, temp_str_device_id, "device_id", &temp_device_id);
+      if (status != SUCCESS) {
+        MSI_LOG_ERROR << "Convert device_id from string to int failed";
+        return status;
+      }
+
+      auto str_rank_id = ParserStringInJson(device, "rank_id");
+      if (str_rank_id.empty()) {
+        return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "rank_id attr is empty in" << rank_table_json_file.c_str();
+      }
+      auto temp_str_rank_id = str_rank_id.get<std::string>();
+      uint32_t temp_rank_id;
+      status = ConvertStr2Int(rank_table_json_file, temp_str_rank_id, "rank_id", &temp_rank_id);
+      if (status != SUCCESS) {
+        MSI_LOG_ERROR << "Convert rank_id from string to int failed";
+        return status;
+      }
+
+      rank_id++;
+      one_rank_config.device_id = temp_device_id;
+      config_.rank_list.push_back(one_rank_config);
+    }
+  }
+  MSI_LOG(INFO) << "Success parser rank table json file with server list and save to DistributedServableConfig";
+
+  return SUCCESS;
+}
 
 Status DistributedServable::WaitAgentsReady(uint64_t wait_agents_time_in_seconds) {
   MSI_LOG_INFO << "Begin waiting ready of all agents";

mindspore_serving/ccsrc/worker/distributed_worker/distributed_servable.h

@@ -17,14 +17,17 @@
 #ifndef MINDSPORE_SERVING_WORKER_DISTRIBUTED_SERVABLE_H
 #define MINDSPORE_SERVING_WORKER_DISTRIBUTED_SERVABLE_H
 
+#include <fstream>
 #include <vector>
 #include <string>
 #include <map>
 #include <memory>
+#include <nlohmann/json.hpp>
 #include "worker/sevable_base.h"
 #include "worker/distributed_worker/common.h"
 #include "worker/distributed_worker/notify_agent/base_notify_agent.h"
 
+using nlohmann::json;
 namespace mindspore {
 namespace serving {
 
@@ -85,6 +88,15 @@ class MS_API DistributedServable : public ServableBase {
   Status CheckRankConfig();
   void SetWaitAgentsPromise(bool flag);
   // agent stubs
+  Status ParserRankTableWithGroupList(const std::string &rank_table_json_file, const json &rank_table_json);
+
+  Status ParserRankTableWithServerList(const std::string &rank_table_json_file, const json &rank_table_json);
+
+  json ParserArrayInJson(const json &json_array, const std::string &str);
+
+  json ParserStringInJson(const json &json_str, const std::string &str);
+  Status ConvertStr2Int(const std::string &rank_table_json_file, const std::string &para_str,
+                        const std::string &para_key, uint32_t *para_int) const;
 };
 
 }  // namespace serving

mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.cc

@@ -104,5 +104,60 @@ Status GrpcNotifyDistributeWorker::NotifyFailed(const std::string &worker_ip, ui
   return INFER_STATUS_LOG_ERROR(SYSTEM_ERROR) << "Failed to notify failure of agent";
 }
 
+Status GrpcNotifyDistributeWorker::GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port,
+                                                              DistributedServableConfig *config) {
+  const int32_t REGISTER_TIME_OUT = 60;
+  const int32_t REGISTER_INTERVAL = 1;
+  auto loop = REGISTER_TIME_OUT;
+  while (loop-- && !ExitSignalHandle::Instance().HasStopped()) {
+    auto address = worker_ip + ":" + std::to_string(worker_port);
+    auto channel = GrpcServer::CreateChannel(address);
+    auto stub = proto::MSWorker::NewStub(channel);
+
+    grpc::ClientContext context;
+    proto::AgentConfigAcquireRequest request;
+    proto::AgentConfigAcquireReply reply;
+    std::chrono::system_clock::time_point deadline =
+      std::chrono::system_clock::now() + std::chrono::seconds(REGISTER_INTERVAL);
+    context.set_deadline(deadline);
+    grpc::Status status = stub->AgentConfigAcquire(&context, request, &reply);
+    if (status.ok()) {
+      MSI_LOG(INFO) << "Success to get Agents configs from Worker, and begin to parser";
+      // parser reply message:AgentConfigAcquireReply, parameter:rank_table_content
+      config->rank_table_content = reply.rank_table_content();
+      // parser reply message:AgentConfigAcquireReply, parameter:rank_list
+      for (auto &temp_rank : reply.rank_list()) {
+        OneRankConfig ome_rank_config;
+        ome_rank_config.ip = temp_rank.ip();
+        ome_rank_config.device_id = temp_rank.device_id();
+        config->rank_list.push_back(ome_rank_config);
+      }
+      // parser reply message:AgentConfigAcquireReply, parameter:common_meta
+      auto &temp_common_meta = reply.common_meta();
+      config->common_meta.servable_name = temp_common_meta.servable_name();
+      config->common_meta.with_batch_dim = temp_common_meta.with_batch_dim();
+      for (auto &temp_without_batch_dim_inputs : temp_common_meta.without_batch_dim_inputs()) {
+        config->common_meta.without_batch_dim_inputs.push_back(temp_without_batch_dim_inputs);
+      }
+      config->common_meta.inputs_count = temp_common_meta.inputs_count();
+      config->common_meta.outputs_count = temp_common_meta.outputs_count();
+
+      // parser reply message:AgentConfigAcquireReply, parameter:distributed_meta
+      auto &temp_distributed_meta = reply.distributed_meta();
+      config->distributed_meta.rank_size = temp_distributed_meta.rank_size();
+      config->distributed_meta.stage_size = temp_distributed_meta.stage_size();
+      MSI_LOG(INFO) << "Success to parser reply message and save to DistributedServableConfig";
+
+      return SUCCESS;
+    }
+    MSI_LOG_INFO << "Grpc message: " << status.error_code() << ", " << status.error_message();
+    std::this_thread::sleep_for(std::chrono::milliseconds(REGISTER_INTERVAL * 1000));
+  }
+  if (ExitSignalHandle::Instance().HasStopped()) {
+    return INFER_STATUS_LOG_WARNING(FAILED) << "Agent exit, stop get Agents configs from Worker";
+  }
+  return INFER_STATUS_LOG_ERROR(SYSTEM_ERROR) << "Failed to get Agents configs from Worker";
+}
+
 }  // namespace serving
 }  // namespace mindspore

mindspore_serving/ccsrc/worker/distributed_worker/notify_distributed/notify_worker.h

@@ -37,6 +37,8 @@ class MS_API GrpcNotifyDistributeWorker {
   Status Unregister();
   // from start up, not agent
   static Status NotifyFailed(const std::string &worker_ip, uint32_t worker_port);
+  static Status GetAgentsConfigsFromWorker(const std::string &worker_ip, uint32_t worker_port,
+                                           DistributedServableConfig *config);
 
  private:
   std::string distributed_worker_ip_;

mindspore_serving/proto/ms_distributed.proto

@@ -27,6 +27,19 @@ message AgentSpec {
   repeated Tensor outputs = 4;
 }
 
+message CommonServableMeta {
+  string servable_name = 1;
+  bool with_batch_dim = 2;
+  repeated int64 without_batch_dim_inputs = 3;
+  int64 inputs_count = 4;
+  int64 outputs_count = 5;
+}
+
+message DistributedServableMeta {
+  int64 rank_size = 1;
+  int64 stage_size = 2;
+}
+
 message AgentRegisterRequest {
   repeated AgentSpec agent_spec = 1;
   string address = 2;
@@ -51,3 +64,17 @@ message AgentFailedRequest {
 message AgentFailedReply {
   ErrorMsg error_msg = 1;
 }
+
+message AgentConfigAcquireRequest {
+}
+
+message AgentConfigAcquireReply {
+  message OneRankConfig {
+    string ip = 1;
+    int64 device_id = 2;
+  }
+  string rank_table_content = 1;
+  repeated OneRankConfig rank_list = 2;
+  CommonServableMeta common_meta = 3;
+  DistributedServableMeta distributed_meta = 4;
+}

mindspore_serving/proto/ms_worker.proto

@@ -30,4 +30,5 @@ service MSWorker {
   rpc AgentExit(AgentExitRequest) returns (AgentExitReply) {}
   rpc AgentRegister(AgentRegisterRequest) returns (AgentRegisterReply) {}
   rpc AgentFailed(AgentFailedRequest) returns (AgentFailedReply) {}
+  rpc AgentConfigAcquire(AgentConfigAcquireRequest) returns (AgentConfigAcquireReply) {}
 }