text: #include <iostream>
  
  #include "ray/common/id.h"
  #include "ray/common/ray_config.h"
  #include "ray/common/status.h"
  #include "ray/common/task/task_common.h"
  #include "ray/raylet/raylet.h"
  #include "ray/stats/stats.h"
  
  #include "gflags/gflags.h"
  
  DEFINE_string(raylet_socket_name, "", "The socket name of raylet.");
  DEFINE_string(store_socket_name, "", "The socket name of object store.");
  DEFINE_int32(object_manager_port, -1, "The port of object manager.");
  DEFINE_int32(node_manager_port, -1, "The port of node manager.");
  DEFINE_string(node_ip_address, "", "The ip address of this node.");
  DEFINE_string(redis_address, "", "The ip address of redis server.");
  DEFINE_int32(redis_port, -1, "The port of redis server.");
  DEFINE_int32(num_initial_workers, 0, "Number of initial workers.");
  DEFINE_int32(maximum_startup_concurrency, 1, "Maximum startup concurrency");
  DEFINE_string(static_resource_list, "", "The static resource list of this node.");
  DEFINE_string(config_list, "", "The raylet config list of this node.");
  DEFINE_string(python_worker_command, "", "Python worker command.");
  DEFINE_string(java_worker_command, "", "Java worker command.");
  DEFINE_string(redis_password, "", "The password of redis.");
  DEFINE_string(temp_dir, "", "Temporary directory.");
  DEFINE_string(session_dir, "", "The path of this ray session directory.");
  DEFINE_bool(disable_stats, false, "Whether disable the stats.");
  DEFINE_string(stat_address, "127.0.0.1:8888", "The address that we report metrics to.");
  DEFINE_bool(enable_stdout_exporter, false,
              "Whether enable the stdout exporter for stats.");
  
  #ifndef RAYLET_TEST
  
  int main(int argc, char *argv[]) {
    InitShutdownRAII ray_log_shutdown_raii(ray::RayLog::StartRayLog,
                                           ray::RayLog::ShutDownRayLog, argv[0],
                                           ray::RayLogLevel::INFO,
                                           /*log_dir=*/"");
    ray::RayLog::InstallFailureSignalHandler();
  
    gflags::ParseCommandLineFlags(&argc, &argv, true);
    const std::string raylet_socket_name = FLAGS_raylet_socket_name;
    const std::string store_socket_name = FLAGS_store_socket_name;
    const int object_manager_port = static_cast<int>(FLAGS_object_manager_port);
    const int node_manager_port = static_cast<int>(FLAGS_node_manager_port);
    const std::string node_ip_address = FLAGS_node_ip_address;
    const std::string redis_address = FLAGS_redis_address;
    const int redis_port = static_cast<int>(FLAGS_redis_port);
    const int num_initial_workers = static_cast<int>(FLAGS_num_initial_workers);
    const int maximum_startup_concurrency =
        static_cast<int>(FLAGS_maximum_startup_concurrency);
    const std::string static_resource_list = FLAGS_static_resource_list;
    const std::string config_list = FLAGS_config_list;
    const std::string python_worker_command = FLAGS_python_worker_command;
    const std::string java_worker_command = FLAGS_java_worker_command;
    const std::string redis_password = FLAGS_redis_password;
    const std::string temp_dir = FLAGS_temp_dir;
    const std::string session_dir = FLAGS_session_dir;
    const bool disable_stats = FLAGS_disable_stats;
    const std::string stat_address = FLAGS_stat_address;
    const bool enable_stdout_exporter = FLAGS_enable_stdout_exporter;
    gflags::ShutDownCommandLineFlags();
  
    // Initialize stats.
    const ray::stats::TagsType global_tags = {
        {ray::stats::JobNameKey, "raylet"},
        {ray::stats::VersionKey, "0.9.0.dev"},
        {ray::stats::NodeAddressKey, node_ip_address}};
    ray::stats::Init(stat_address, global_tags, disable_stats, enable_stdout_exporter);
  
    // Configuration for the node manager.
    ray::raylet::NodeManagerConfig node_manager_config;
    std::unordered_map<std::string, double> static_resource_conf;
    std::unordered_map<std::string, std::string> raylet_config;
  
    // Parse the configuration list.
    std::istringstream config_string(config_list);
    std::string config_name;
    std::string config_value;
  
    while (std::getline(config_string, config_name, ',')) {
      RAY_CHECK(std::getline(config_string, config_value, ','));
      // TODO(rkn): The line below could throw an exception. What should we do about this?
      raylet_config[config_name] = config_value;
    }
  
    RayConfig::instance().initialize(raylet_config);
  
    // Parse the resource list.
    std::istringstream resource_string(static_resource_list);
    std::string resource_name;
    std::string resource_quantity;
  
    while (std::getline(resource_string, resource_name, ',')) {
      RAY_CHECK(std::getline(resource_string, resource_quantity, ','));
      // TODO(rkn): The line below could throw an exception. What should we do about this?
      static_resource_conf[resource_name] = std::stod(resource_quantity);
    }
  
    node_manager_config.resource_config = ray::ResourceSet(std::move(static_resource_conf));
    RAY_LOG(DEBUG) << "Starting raylet with static resource configuration: "
                   << node_manager_config.resource_config.ToString();
    node_manager_config.node_manager_address = node_ip_address;
    node_manager_config.node_manager_port = node_manager_port;
    node_manager_config.num_initial_workers = num_initial_workers;
    node_manager_config.maximum_startup_concurrency = maximum_startup_concurrency;
  
    if (!python_worker_command.empty()) {
      node_manager_config.worker_commands.emplace(
          make_pair(ray::Language::PYTHON, SplitStrByWhitespaces(python_worker_command)));
    }
    if (!java_worker_command.empty()) {
      node_manager_config.worker_commands.emplace(
          make_pair(ray::Language::JAVA, SplitStrByWhitespaces(java_worker_command)));
    }
    if (python_worker_command.empty() && java_worker_command.empty()) {
      RAY_CHECK(0)
          << "Either Python worker command or Java worker command should be provided.";
    }
  
    node_manager_config.heartbeat_period_ms =
        RayConfig::instance().raylet_heartbeat_timeout_milliseconds();
    node_manager_config.debug_dump_period_ms =
        RayConfig::instance().debug_dump_period_milliseconds();
    node_manager_config.fair_queueing_enabled =
        RayConfig::instance().fair_queueing_enabled();
    node_manager_config.max_lineage_size = RayConfig::instance().max_lineage_size();
    node_manager_config.store_socket_name = store_socket_name;
    node_manager_config.temp_dir = temp_dir;
    node_manager_config.session_dir = session_dir;
  
    // Configuration for the object manager.
    ray::ObjectManagerConfig object_manager_config;
    object_manager_config.object_manager_port = object_manager_port;
    object_manager_config.store_socket_name = store_socket_name;
    object_manager_config.pull_timeout_ms =
        RayConfig::instance().object_manager_pull_timeout_ms();
    object_manager_config.push_timeout_ms =
        RayConfig::instance().object_manager_push_timeout_ms();
  
    int num_cpus = static_cast<int>(static_resource_conf["CPU"]);
    object_manager_config.rpc_service_threads_number =
        std::min(std::max(2, num_cpus / 4), 8);
    object_manager_config.object_chunk_size =
        RayConfig::instance().object_manager_default_chunk_size();
  
    RAY_LOG(DEBUG) << "Starting object manager with configuration: \n"
                   << "rpc_service_threads_number = "
                   << object_manager_config.rpc_service_threads_number
                   << ", object_chunk_size = " << object_manager_config.object_chunk_size;
  
    // Initialize the node manager.
    boost::asio::io_service main_service;
  
    // Initialize gcs client
    ray::gcs::GcsClientOptions client_options(redis_address, redis_port, redis_password);
    auto gcs_client = std::make_shared<ray::gcs::RedisGcsClient>(client_options);
    RAY_CHECK_OK(gcs_client->Connect(main_service));
  
    std::unique_ptr<ray::raylet::Raylet> server(new ray::raylet::Raylet(
        main_service, raylet_socket_name, node_ip_address, redis_address, redis_port,
        redis_password, node_manager_config, object_manager_config, gcs_client));
  
    // Destroy the Raylet on a SIGTERM. The pointer to main_service is
    // guaranteed to be valid since this function will run the event loop
    // instead of returning immediately.
    // We should stop the service and remove the local socket file.
    auto handler = [&main_service, &raylet_socket_name, &server, &gcs_client](
                       const boost::system::error_code &error, int signal_number) {
      RAY_LOG(INFO) << "Raylet received SIGTERM, shutting down...";
      auto shutdown_callback = [&server, &main_service, &gcs_client]() {
        server.reset();
        gcs_client->Disconnect();
        main_service.stop();
      };
      RAY_CHECK_OK(gcs_client->client_table().Disconnect(shutdown_callback));
      // Give a timeout for this Disconnect operation.
      boost::posix_time::milliseconds stop_timeout(800);
      boost::asio::deadline_timer timer(main_service);
      timer.expires_from_now(stop_timeout);
      timer.async_wait([shutdown_callback](const boost::system::error_code &error) {
        if (!error) {
          RAY_LOG(INFO) << "Disconnect from client table timed out, forcing shutdown.";
          shutdown_callback();
        }
      });
      remove(raylet_socket_name.c_str());
    };
    boost::asio::signal_set signals(main_service, SIGTERM);
    signals.async_wait(handler);
  
    main_service.run();
  }
  #endif
  