Repository: ray-project/ray
File: src/ray/core_worker/core_worker_process.cc (C++)

Code before:
    // Initialize global worker instance.
    auto worker = CreateCoreWorker(options_, worker_id_);
    auto write_locked = core_worker_.LockForWrite();
    write_locked.Get() = worker;
    // Initialize metrics agent client.
    // Port > 0 means valid port, -1 means metrics agent not available (minimal install).
    if (options_.metrics_agent_port > 0) {
      metrics_agent_client_ = std::make_unique<ray::rpc::MetricsAgentClientImpl>(
          "127.0.0.1", options_.metrics_agent_port, io_service_, *client_call_manager_);
      // Block until exporters are initialized to avoid a getenv/setenv race
      // POSIX setenv is MT-Unsafe.
      auto metrics_init_promise = std::make_shared<std::promise<void>>();
      metrics_init_future = metrics_init_promise->get_future();
      metrics_agent_client_->WaitForServerReady(
          [this, metrics_init_promise](const Status &server_status) {
            if (server_status.ok()) {
              stats::ConnectOpenCensusExporter(options_.metrics_agent_port);
              stats::InitOpenTelemetryExporter(options_.metrics_agent_port);
            } else {
              RAY_LOG(ERROR)
                  << "Failed to establish connection to the metrics exporter agent. "
                     "Metrics will not be exported. "
                  << "Exporter agent status: " << server_status.ToString();
            }
            metrics_init_promise->set_value();
          });
    } else {
      RAY_LOG(INFO) << "Metrics agent not available. To enable metrics, install Ray "
                       "with dashboard support: `pip install 'ray[default]'`.";
    }
  }

  if (metrics_init_future.has_value()) {
    metrics_init_future->wait();
  }
}

CoreWorkerProcessImpl::~CoreWorkerProcessImpl() {
  RAY_LOG(INFO) << "Destructing CoreWorkerProcessImpl. pid: " << getpid();
  // Shutdown stats module if worker process exits.
  stats::Shutdown();
  if (options_.enable_logging) {
    RayLog::ShutDownRayLog();
  }
}

void CoreWorkerProcess::EnsureInitialized(bool quick_exit) {
  if (core_worker_process != nullptr) {
    return;
  }


Review comment (performance):
Can you make WaitForServerReady synchronous instead? It's unnecessary complexity to call and async method and then block. 

Code after:
    auto write_locked = core_worker_.LockForWrite();
    write_locked.Get() = worker;
    // Initialize metrics agent client.
    // Port > 0 means valid port, -1 means metrics agent not available (minimal install).
    if (options_.metrics_agent_port > 0) {
      // Initialize exporters synchronously to avoid a getenv/setenv race condition.
      // POSIX setenv is MT-Unsafe.
      stats::ConnectOpenCensusExporter(options_.metrics_agent_port);
      stats::InitOpenTelemetryExporter(options_.metrics_agent_port);
    } else {
      RAY_LOG(INFO) << "Metrics agent not available. To enable metrics, install Ray "
                       "with dashboard support: `pip install 'ray[default]'`.";
    }
  }
}

CoreWorkerProcessImpl::~CoreWorkerProcessImpl() {
  RAY_LOG(INFO) << "Destructing CoreWorkerProcessImpl. pid: " << getpid();
  // Shutdown stats module if worker process exits.
  stats::Shutdown();
  if (options_.enable_logging) {
    RayLog::ShutDownRayLog();
  }
}

void CoreWorkerProcess::EnsureInitialized(bool quick_exit) {
  if (core_worker_process != nullptr) {
    return;
  }

  if (quick_exit) {
    RAY_LOG(WARNING) << "The core worker process is not initialized yet or already "
                     << "shutdown.";
    QuickExit();
  } else {
    RAY_CHECK(core_worker_process)
        << "The core worker process is not initialized yet or already "
        << "shutdown.";
  }
}

void CoreWorkerProcessImpl::InitializeSystemConfig() {
  // We have to create a short-time thread here because the RPC request to get the
  // system config from Raylet is asynchronous, and we need to synchronously initialize
  // the system config in the constructor of `CoreWorkerProcessImpl`.
  std::promise<std::string> promise;
  std::thread thread([&] {
    instrumented_io_context io_service{/*emit_metrics=*/false,
                                       /*running_on_single_thread=*/true};
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work(
        io_service.get_executor());