const flatbuffers::String &object_id, const protocol::Address &address) {

rpc::ObjectReference ref;

ref.set_object_id(object_id.str());

ref.mutable_owner_address()->set_node_id(address.node_id()->str());

ref.mutable_owner_address()->set_ip_address(address.ip_address()->str());

ref.mutable_owner_address()->set_port(address.port());

ref.mutable_owner_address()->set_worker_id(address.worker_id()->str());

return ref;

const flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>> &object_ids,

const flatbuffers::Vector<flatbuffers::Offset<protocol::Address>> &owner_addresses) {

RAY_CHECK(object_ids.size() == owner_addresses.size());

std::vector<rpc::ObjectReference> refs;

refs.reserve(object_ids.size());

for (int64_t i = 0; i < object_ids.size(); i++) {

refs.push_back(

FlatbufferToSingleObjectReference(*object_ids.Get(i), *owner_addresses.Get(i)));

}

return refs;

const flatbuffers::Vector<flatbuffers::Offset<flatbuffers::String>> &vector) {

std::vector<ObjectID> ids;

ids.reserve(vector.size());

for (int64_t i = 0; i < vector.size(); i++) {

ids.push_back(ObjectID::FromBinary(vector.Get(i)->str()));

}

return ids;

const WorkerID &wid,

const std::string &ctx,

int sig) {

// Guard against targeting the raylet's own process group if isolation failed.

pid_t raylet_pgid = getpgid(0);

if (raylet_pgid == saved_pgid) {

RAY_LOG(WARNING).WithField(wid)

<< ctx

<< ": skipping PG cleanup: worker pgid equals raylet pgid (isolation failed): "

<< saved_pgid;

return;

}

RAY_LOG(INFO).WithField(wid) << ctx << ": sending "

<< (sig == SIGKILL ? "SIGKILL" : "SIGTERM")

<< " to pgid=" << saved_pgid;

auto err = KillProcessGroup(saved_pgid, sig);

if (err && *err) {

RAY_LOG(WARNING).WithField(wid)

<< ctx << ": failed to send " << (sig == SIGKILL ? "SIGKILL" : "SIGTERM")

<< " to process group " << saved_pgid << ": " << err->message()

<< ", errno=" << err->value();

}

int start_index,

int end_index) {

std::vector<std::string> enum_names;

enum_names.reserve(start_index);

for (int i = 0; i < start_index; ++i) {

enum_names.emplace_back("EmptyMessageType");

}

size_t i = 0;

while (true) {

const char *name = enum_names_ptr[i];

if (name == nullptr) {

break;

}

enum_names.emplace_back(name);

i++;

}

RAY_CHECK(static_cast<size_t>(end_index) == enum_names.size() - 1)

<< "Message Type mismatch!";

return enum_names;

instrumented_io_context &io_service,

const NodeID &self_node_id,

std::string self_node_name,

const NodeManagerConfig &config,

gcs::GcsClient &gcs_client,

rpc::ClientCallManager &client_call_manager,

rpc::CoreWorkerClientPool &worker_rpc_pool,

rpc::RayletClientPool &raylet_client_pool,

pubsub::SubscriberInterface &core_worker_subscriber,

ClusterResourceScheduler &cluster_resource_scheduler,

LocalLeaseManagerInterface &local_lease_manager,

ClusterLeaseManagerInterface &cluster_lease_manager,

IObjectDirectory &object_directory,

ObjectManagerInterface &object_manager,

LocalObjectManagerInterface &local_object_manager,

LeaseDependencyManager &lease_dependency_manager,

WorkerPoolInterface &worker_pool,

absl::flat_hash_map<LeaseID, std::shared_ptr<WorkerInterface>> &leased_workers,

std::shared_ptr<plasma::PlasmaClientInterface> store_client,

std::unique_ptr<core::experimental::MutableObjectProviderInterface>

mutable_object_provider,

std::function<void(const rpc::NodeDeathInfo &)> shutdown_raylet_gracefully,

AddProcessToCgroupHook add_process_to_system_cgroup_hook,

std::unique_ptr<CgroupManagerInterface> cgroup_manager,

std::atomic_bool &shutting_down,

PlacementGroupResourceManager &placement_group_resource_manager,

boost::asio::basic_socket_acceptor<local_stream_protocol> acceptor,

local_stream_socket socket,

ray::observability::MetricInterface &memory_manager_worker_eviction_total_count,

ray::observability::MetricInterface

&node_manager_unexpected_worker_failure_total_count)

: self_node_id_(self_node_id),

self_node_name_(std::move(self_node_name)),

io_service_(io_service),

gcs_client_(gcs_client),

shutdown_raylet_gracefully_(std::move(shutdown_raylet_gracefully)),

worker_pool_(worker_pool),

client_call_manager_(client_call_manager),

worker_rpc_pool_(worker_rpc_pool),

raylet_client_pool_(raylet_client_pool),

core_worker_subscriber_(core_worker_subscriber),

object_directory_(object_directory),

object_manager_(object_manager),

store_client_(std::move(store_client)),

mutable_object_provider_(std::move(mutable_object_provider)),

periodical_runner_(PeriodicalRunner::Create(io_service)),

report_resources_period_ms_(config.report_resources_period_ms),

initial_config_(config),

lease_dependency_manager_(lease_dependency_manager),

wait_manager_(/*is_object_local*/

[this](const ObjectID &object_id) {

return lease_dependency_manager_.CheckObjectLocal(object_id);

},

/*delay_executor*/

[this](std::function<void()> fn, int64_t delay_ms) {

RAY_UNUSED(execute_after(io_service_,

std::move(fn),

std::chrono::milliseconds(delay_ms)));

}),

runtime_env_agent_port_(config.runtime_env_agent_port),

node_manager_server_("NodeManager",

config.node_manager_port,

config.node_manager_address == "127.0.0.1"),

local_object_manager_(local_object_manager),

leased_workers_(leased_workers),

local_gc_interval_ns_(RayConfig::instance().local_gc_interval_s() * 1e9),

plasma_store_usage_trigger_gc_threshold_(

RayConfig::instance().plasma_store_usage_trigger_gc_threshold()),

local_gc_throttler_(RayConfig::instance().local_gc_min_interval_s() * 1e9),

global_gc_throttler_(RayConfig::instance().global_gc_min_interval_s() * 1e9),

memory_manager_worker_eviction_total_count_(

memory_manager_worker_eviction_total_count),

node_manager_unexpected_worker_failure_total_count_(

node_manager_unexpected_worker_failure_total_count),

cluster_resource_scheduler_(cluster_resource_scheduler),

local_lease_manager_(local_lease_manager),

cluster_lease_manager_(cluster_lease_manager),

record_metrics_period_ms_(config.record_metrics_period_ms),

placement_group_resource_manager_(placement_group_resource_manager),

ray_syncer_(io_service_, self_node_id_.Binary(), 1, 0),

worker_killing_policy_(WorkerKillingPolicyFactory::Create(

config.enable_resource_isolation, *cgroup_manager)),

memory_monitors_(MemoryMonitorFactory::Create(CreateKillWorkersCallback(),

config.enable_resource_isolation,

*cgroup_manager)),

add_process_to_system_cgroup_hook_(std::move(add_process_to_system_cgroup_hook)),

cgroup_manager_(std::move(cgroup_manager)),

shutting_down_(shutting_down),

acceptor_(std::move(acceptor)),

socket_(std::move(socket)) {

RAY_LOG(INFO).WithField(kLogKeyNodeID, self_node_id_) << "Initializing NodeManager";

periodical_runner_->RunFnPeriodically(

[this]() { cluster_lease_manager_.ScheduleAndGrantLeases(); },

RayConfig::instance().worker_cap_initial_backoff_delay_ms(),

"NodeManager.ScheduleAndGrantLeases");

periodical_runner_->RunFnPeriodically(

[this]() { CheckForUnexpectedWorkerDisconnects(); },

RayConfig::instance().raylet_check_for_unexpected_worker_disconnect_interval_ms(),

"NodeManager.CheckForUnexpectedWorkerDisconnects");

RAY_CHECK_OK(store_client_->Connect(config.store_socket_name));

// Run the node manager rpc server.

node_manager_server_.RegisterService(

std::make_unique<rpc::NodeManagerGrpcService>(io_service, *this), false);

// Pass auth token from the RPC server to the syncer service

node_manager_server_.RegisterService(std::make_unique<syncer::RaySyncerService>(

ray_syncer_, ray::rpc::AuthenticationTokenLoader::instance().GetToken()));

node_manager_server_.Run();

// GCS will check the health of the service named with the node id.

// Fail to setup this will lead to the health check failure.

node_manager_server_.GetServer().GetHealthCheckService()->SetServingStatus(

self_node_id_.Hex(), true);

worker_pool_.SetNodeManagerPort(GetServerPort());

dashboard_agent_manager_ = CreateDashboardAgentManager(self_node_id, config);

runtime_env_agent_manager_ = CreateRuntimeEnvAgentManager(self_node_id, config);

std::tie(metrics_agent_port_, metrics_export_port_, dashboard_agent_listen_port_) =

WaitForDashboardAgentPorts(self_node_id, config);

runtime_env_agent_port_ = WaitForRuntimeEnvAgentPort(self_node_id, config);

auto runtime_env_agent_client = RuntimeEnvAgentClient::Create(

io_service_,

config.node_manager_address,

runtime_env_agent_port_, /*delay_executor=*/

[this](std::function<void()> task, uint32_t delay_ms) {

return execute_after(

io_service_, std::move(task), std::chrono::milliseconds(delay_ms));

},

shutdown_raylet_gracefully_);

worker_pool_.SetRuntimeEnvAgentClient(std::move(runtime_env_agent_client));

worker_pool_.Start();

periodical_runner_->RunFnPeriodically([this]() { GCWorkerFailureReason(); },

RayConfig::instance().task_failure_entry_ttl_ms(),

"NodeManager.GCTaskFailureReason");

auto register_callback =

[this,

object_manager_port = self_node_info.object_manager_port()](const Status &status) {

RAY_CHECK_OK(status);

RAY_LOG(INFO) << "Raylet of id, " << self_node_id_

<< " started. Raylet consists of node_manager and object_manager."

<< " node_manager address: "

<< BuildAddress(initial_config_.node_manager_address,

initial_config_.node_manager_port)

<< " object_manager address: "

<< BuildAddress(initial_config_.node_manager_address,

object_manager_port)

<< " hostname: " << boost::asio::ip::host_name();

this->RegisterGcs();

};

gcs_client_.Nodes().RegisterSelf(std::move(self_node_info), register_callback);

acceptor_.async_accept(

socket_,

boost::bind(&NodeManager::HandleAccept, this, boost::asio::placeholders::error));

auto on_node_change = [this](const NodeID &node_id,

const rpc::GcsNodeAddressAndLiveness &data) {

if (data.state() == GcsNodeInfo::ALIVE) {

NodeAdded(data);

} else {

RAY_CHECK(data.state() == GcsNodeInfo::DEAD);

NodeRemoved(node_id);

}

};

// If the node resource message is received first and then the node message is

// received, ForwardTask will throw exception, because it can't get node info.

auto on_node_change_subscribe_done = [this](Status status) {

RAY_CHECK_OK(status);

// RESOURCE_VIEW is used to synchronize available resources across Raylets.

//

// LocalResourceManager::CreateSyncMessage will be called periodically to collect

// the local Raylet's usage to broadcast to others (via the GCS). The updates are

// versioned inside of `LocalResourceManager` to avoid unnecessary broadcasts.

//

// NodeManager::ConsumeSyncMessage will be called when a sync message containing

// other Raylets' resource usage is received.

ray_syncer_.Register(

/* message_type */ syncer::MessageType::RESOURCE_VIEW,

/* reporter */ &cluster_resource_scheduler_.GetLocalResourceManager(),

/* receiver */ this,

/* pull_from_reporter_interval_ms */

report_resources_period_ms_);

// COMMANDS is used only to broadcast a global request to call the Python garbage

// collector on all Raylets when the cluster is under memory pressure.

//

// Periodic collection is disabled, so this command is only broadcasted via

// `OnDemandBroadcasting` (which will call NodeManager::CreateSyncMessage).

//

// NodeManager::ConsumeSyncMessage is called to execute the GC command from other

// Raylets.

ray_syncer_.Register(

/* message_type */ syncer::MessageType::COMMANDS,

/* reporter */ this,

/* receiver */ this,

/* pull_from_reporter_interval_ms */ 0);

auto gcs_channel = gcs_client_.GetGcsRpcClient().GetChannel();

ray_syncer_.Connect(kGCSNodeID.Binary(), gcs_channel);

periodical_runner_->RunFnPeriodically(

[this] { TriggerLocalOrGlobalGCIfNeeded(); },

RayConfig::instance().raylet_check_gc_period_milliseconds(),

"NodeManager.CheckGC");

};

// Register a callback to monitor new nodes and a callback to monitor removed nodes.

gcs_client_.Nodes().AsyncSubscribeToNodeAddressAndLivenessChange(

std::move(on_node_change), std::move(on_node_change_subscribe_done));

// Subscribe to all unexpected failure notifications from the local and

// remote raylets. Note that this does not include workers that failed due to

// node failure. These workers can be identified by comparing the node_id

// in their rpc::Address to the ID of a failed raylet.

const auto &worker_failure_handler =

[this](const rpc::WorkerDeltaData &worker_failure_data) {

HandleUnexpectedWorkerFailure(

WorkerID::FromBinary(worker_failure_data.worker_id()));

};

gcs_client_.Workers().AsyncSubscribeToWorkerFailures(worker_failure_handler, nullptr);

// Subscribe to job updates.

const auto job_subscribe_handler = [this](const JobID &job_id,

const JobTableData &job_data) {

// HandleJobStarted is idempotent so it's ok to call it again when the job

// finishes. We always need to call `HandleJobStarted` even when a job has

// finished, because we may have missed the started event (for example,

// because the node wasn't up when the job started). JobStarted +

// JobFinished events both need to be processed because we need to persist

// the job config of dead jobs in order for detached actors to function

// properly.

HandleJobStarted(job_id, job_data);

if (job_data.is_dead()) {

HandleJobFinished(job_id, job_data);

}

};

gcs_client_.Jobs().AsyncSubscribeAll(job_subscribe_handler, nullptr);

periodical_runner_->RunFnPeriodically(

[this] {

DumpDebugState();

WarnResourceDeadlock();

},

RayConfig::instance().debug_dump_period_milliseconds(),

"NodeManager.deadline_timer.debug_state_dump");

uint64_t now_ms = current_time_ms();

last_metrics_recorded_at_ms_ = now_ms;

periodical_runner_->RunFnPeriodically([this] { RecordMetrics(); },

record_metrics_period_ms_,

"NodeManager.deadline_timer.record_metrics");

if (RayConfig::instance().free_objects_period_milliseconds() > 0) {

periodical_runner_->RunFnPeriodically(

[this] { local_object_manager_.FlushFreeObjects(); },

RayConfig::instance().free_objects_period_milliseconds(),

"NodeManager.deadline_timer.flush_free_objects");

if (RayConfig::instance().object_spilling_config().empty()) {

RAY_LOG(INFO) << "Object spilling is disabled because spilling config is "

<< "unspecified";

} else {

periodical_runner_->RunFnPeriodically(

[this] { SpillIfOverPrimaryObjectsThreshold(); },

RayConfig::instance().free_objects_period_milliseconds(),

"NodeManager.deadline_timer.spill_objects_when_over_threshold");

}

}

/// If periodic asio stats print is enabled, it will print it.

const auto event_stats_print_interval_ms =

RayConfig::instance().event_stats_print_interval_ms();

if (event_stats_print_interval_ms != -1 && RayConfig::instance().event_stats()) {

periodical_runner_->RunFnPeriodically(

[this] {

std::stringstream debug_msg;

debug_msg << DebugString() << "\n\n";

RAY_LOG(INFO) << PrependToEachLine(debug_msg.str(), "[state-dump] ");

ReportWorkerOomKillStats();

},

event_stats_print_interval_ms,

"NodeManager.deadline_timer.print_event_loop_stats");

}

// Raylet periodically check whether it's alive in GCS.

// For failure cases, GCS might think this raylet dead, but this

// raylet still think it's alive. This could happen when the cluster setup is wrong,

// for example, there is data loss in the DB.

periodical_runner_->RunFnPeriodically(

[this] {

// Flag to see whether a request is running.

static bool checking = false;

if (checking) {

return;

}

checking = true;

gcs_client_.Nodes().AsyncCheckAlive(

{self_node_id_},

/* timeout_ms = */ 30000,

// capture checking ptr here because vs17 fail to compile

[this, checking_ptr = &checking](const auto &status,

const auto &alive_vec) mutable {

bool alive = alive_vec[0];

if ((status.ok() && !alive)) {

// GCS think this raylet is dead. Fail the node

RAY_LOG(FATAL)

<< "GCS consider this node to be dead. This may happen when "

<< "GCS is not backed by a DB and restarted or there is data loss "

<< "in the DB.";

} else if (status.IsUnauthenticated()) {

RAY_LOG(FATAL)

<< "GCS returned an authentication error. This may happen when "

<< "GCS is not backed by a DB and restarted or there is data loss "

<< "in the DB. Local cluster ID: " << gcs_client_.GetClusterId();

}

*checking_ptr = false;

});

},

RayConfig::instance().raylet_liveness_self_check_interval_ms(),

"NodeManager.GcsCheckAlive");

if (!error) {

ConnectionErrorHandler error_handler =

[this](const std::shared_ptr<ClientConnection> &client,

const boost::system::error_code &err) {

this->HandleClientConnectionError(client, err);

};

MessageHandler message_handler = [this](

const std::shared_ptr<ClientConnection> &client,

int64_t message_type,

const std::vector<uint8_t> &message) {

this->ProcessClientMessage(client, message_type, message.data());

};

// Accept a new local client and dispatch it to the node manager.

auto conn = ClientConnection::Create(message_handler,

error_handler,

std::move(socket_),

"worker",

node_manager_message_enum);

// Begin processing messages. The message handler above is expected to call this to

// continue processing messages.

conn->ProcessMessages();

} else {

RAY_LOG(ERROR) << "Raylet failed to accept new connection: " << error.message();

if (error == boost::asio::error::operation_aborted) {

// The server is being destroyed. Don't continue accepting connections.

return;

}

};

// We're ready to accept another client.

acceptor_.async_accept(

socket_,

boost::bind(&NodeManager::HandleAccept, this, boost::asio::placeholders::error));

rpc::WorkerExitType disconnect_type,

const std::string &disconnect_detail,

bool force) {

// We should disconnect the client first. Otherwise, we'll remove bundle resources

// before actual resources are returned. Subsequent disconnect request that comes

// due to worker dead will be ignored.

DisconnectClient(

worker->Connection(), /*graceful=*/false, disconnect_type, disconnect_detail);

worker->KillAsync(io_service_, force);

if (disconnect_type == rpc::WorkerExitType::SYSTEM_ERROR) {

number_workers_killed_++;

} else if (disconnect_type == rpc::WorkerExitType::NODE_OUT_OF_MEMORY) {

number_workers_killed_by_oom_++;

}

RAY_LOG(DEBUG).WithField(job_id)

<< "HandleJobStarted Driver pid " << job_data.driver_pid()

<< " is dead: " << job_data.is_dead()

<< " driver address: " << job_data.driver_address().ip_address();

worker_pool_.HandleJobStarted(job_id, job_data.config());

// Leases of this job may already arrived but failed to pop a worker because the job

// config is not local yet. So we trigger granting again here to try to

// reschedule these leases.

cluster_lease_manager_.ScheduleAndGrantLeases();

RAY_LOG(DEBUG).WithField(job_id) << "HandleJobFinished";

RAY_CHECK(job_data.is_dead());

// Force kill all the worker processes belonging to the finished job

// so that no worker processes is leaked.

for (const auto &pair : leased_workers_) {

auto &worker = pair.second;

RAY_CHECK(!worker->GetAssignedJobId().IsNil());

if (worker->GetRootDetachedActorId().IsNil() &&

(worker->GetAssignedJobId() == job_id)) {

// Don't kill worker processes belonging to the detached actor

// since those are expected to outlive the job.

RAY_LOG(INFO).WithField(worker->WorkerId()).WithField(job_id)

<< "Killing leased worker because its job finished.";

rpc::ExitRequest request;

request.set_force_exit(true);

worker->rpc_client()->Exit(

request, [this, worker](const ray::Status &status, const rpc::ExitReply &r) {

if (!status.ok()) {

RAY_LOG(WARNING).WithField(worker->WorkerId())

<< "Failed to send exit request to worker "

<< ": " << status.ToString() << ". Killing it using SIGKILL instead.";

// Just kill-9 as a last resort.

worker->KillAsync(io_service_, /* force */ true);

}

});

}

}

worker_pool_.HandleJobFinished(job_id);

std::vector<std::shared_ptr<ClientConnection>> all_connections;

std::vector<std::shared_ptr<WorkerInterface>> all_workers =

worker_pool_.GetAllRegisteredWorkers();

all_connections.reserve(all_workers.size());

for (const auto &worker : all_workers) {

all_connections.push_back(worker->Connection());

}

for (const auto &driver : worker_pool_.GetAllRegisteredDrivers()) {

all_workers.push_back(driver);

all_connections.push_back(driver->Connection());

}

RAY_CHECK_EQ(all_connections.size(), all_workers.size());

// Check if there are any unexpected disconnects on the worker socket connections.

// This will close the connection without processing remaining messages.

std::vector<bool> disconnects = CheckForClientDisconnects(all_connections);

for (size_t i = 0; i < disconnects.size(); i++) {

if (disconnects[i]) {

std::string msg = "Worker connection closed unexpectedly.";

RAY_LOG(DEBUG).WithField(all_workers[i]->WorkerId()) << msg;

DestroyWorker(all_workers[i], rpc::WorkerExitType::SYSTEM_ERROR, msg);

}

}

rpc::ReleaseUnusedBundlesReply *reply,

rpc::SendReplyCallback send_reply_callback) {

RAY_LOG(DEBUG) << "Releasing unused bundles.";

std::unordered_set<BundleID, pair_hash> in_use_bundles;

for (int index = 0; index < request.bundles_in_use_size(); ++index) {

const auto &bundle_id = request.bundles_in_use(index).bundle_id();

in_use_bundles.emplace(PlacementGroupID::FromBinary(bundle_id.placement_group_id()),

bundle_id.bundle_index());

// Add -1 one to the in_use_bundles. It's ok to add it more than one times since

// it's a set.

in_use_bundles.emplace(PlacementGroupID::FromBinary(bundle_id.placement_group_id()),

-1);

}

// Cancel lease requests that are waiting for workers

// to free the acquired pg bundle resources

// so that pg bundle can be returned.

local_lease_manager_.CancelLeases(

[&](const std::shared_ptr<internal::Work> &work) {

const auto bundle_id =

work->lease_.GetLeaseSpecification().PlacementGroupBundleId();

return !bundle_id.first.IsNil() && (0 == in_use_bundles.count(bundle_id)) &&

(work->GetState() == internal::WorkStatus::WAITING_FOR_WORKER);

},

rpc::RequestWorkerLeaseReply::SCHEDULING_CANCELLED_INTENDED,

"The lease request is cancelled because it uses placement group bundles that are "

"not "

"registered to GCS. It can happen upon GCS restart.");

// Kill all workers that are currently associated with the unused bundles.

// NOTE: We can't traverse directly with `leased_workers_`, because `DestroyWorker`

// will delete the element of `leased_workers_`. So we need to filter out

// `workers_associated_with_unused_bundles` separately.

std::vector<std::shared_ptr<WorkerInterface>> workers_associated_with_unused_bundles;

for (const auto &worker_it : leased_workers_) {

auto &worker = worker_it.second;

const auto &bundle_id = worker->GetBundleId();

// We need to filter out the workers used by placement group.

if (!bundle_id.first.IsNil() && 0 == in_use_bundles.count(bundle_id)) {

workers_associated_with_unused_bundles.emplace_back(worker);

}

}

for (const auto &worker : workers_associated_with_unused_bundles) {

RAY_LOG(DEBUG)

.WithField(worker->GetBundleId().first)

.WithField(worker->GetGrantedLeaseId())

.WithField(worker->GetActorId())

.WithField(worker->WorkerId())

<< "Destroying worker since its bundle was unused, bundle index: "

<< worker->GetBundleId().second;

DestroyWorker(worker,

rpc::WorkerExitType::INTENDED_SYSTEM_EXIT,

"Worker exits because it uses placement group bundles that are not "

"registered to GCS. It can happen upon GCS restart.");

}

// Return unused bundle resources.

placement_group_resource_manager_.ReturnUnusedBundle(in_use_bundles);

send_reply_callback(Status::OK(), nullptr, nullptr);

rpc::GetObjectsInfoReply *reply,

rpc::SendReplyCallback send_reply_callback) {

RAY_LOG(DEBUG) << "Received a HandleGetObjectsInfo request";

auto total = std::make_shared<int>(0);

auto count = std::make_shared<int>(0);

auto limit = request.has_limit() ? request.limit() : -1;

// Each worker query will have limit as well.

// At the end there will be limit * num_workers entries returned at max.

QueryAllWorkerStates(

/*on_replied*/

[reply, total, count, limit](const ray::Status &status,

const rpc::GetCoreWorkerStatsReply &r) {

*total += r.core_worker_stats().objects_total();

if (limit != -1 && *count >= limit) {

return;

}

// Currently, instead of counting object one by one, we add all object refs

// returned. This means there can be overflow. TODO(sang): Fix it after

// refactoring this code path.

*count += r.core_worker_stats().object_refs_size();

if (status.ok()) {

reply->add_core_workers_stats()->MergeFrom(r.core_worker_stats());

} else {

RAY_LOG(INFO) << "Failed to query object information from a worker.";

}

},

send_reply_callback,

/*include_memory_info*/ true,

/*include_task_info*/ false,

/*limit*/ limit,

/*on_all_replied*/ [total, reply]() { reply->set_total(*total); });

rpc::GetWorkerFailureCauseRequest request,

rpc::GetWorkerFailureCauseReply *reply,

rpc::SendReplyCallback send_reply_callback) {

const LeaseID lease_id = LeaseID::FromBinary(request.lease_id());

RAY_LOG(DEBUG) << "Received a HandleGetWorkerFailureCause request for lease "

<< lease_id;

auto it = worker_failure_reasons_.find(lease_id);

if (it != worker_failure_reasons_.end()) {

RAY_LOG(DEBUG) << "lease " << lease_id << " has failure reason "

<< ray::gcs::RayErrorInfoToString(it->second.ray_error_info_)

<< ", fail immediately: " << !it->second.should_retry_;

reply->mutable_failure_cause()->CopyFrom(it->second.ray_error_info_);

reply->set_fail_task_immediately(!it->second.should_retry_);

} else {

RAY_LOG(INFO) << "didn't find failure cause for lease " << lease_id;

}

send_reply_callback(Status::OK(), nullptr, nullptr);

rpc::RegisterMutableObjectRequest request,

rpc::RegisterMutableObjectReply *reply,

rpc::SendReplyCallback send_reply_callback) {

ObjectID writer_object_id = ObjectID::FromBinary(request.writer_object_id());

int64_t num_readers = request.num_readers();

ObjectID reader_object_id = ObjectID::FromBinary(request.reader_object_id());

mutable_object_provider_->HandleRegisterMutableObject(

writer_object_id, num_readers, reader_object_id);

send_reply_callback(Status::OK(), nullptr, nullptr);

rpc::PushMutableObjectReply *reply,

rpc::SendReplyCallback send_reply_callback) {

mutable_object_provider_->HandlePushMutableObject(request, reply);

send_reply_callback(Status::OK(), nullptr, nullptr);

const std::function<void(const ray::Status &, const rpc::GetCoreWorkerStatsReply &)>

&on_replied,

rpc::SendReplyCallback &send_reply_callback,

bool include_memory_info,

bool include_task_info,

int64_t limit,

const std::function<void()> &on_all_replied) {

auto all_workers = worker_pool_.GetAllRegisteredWorkers(/* filter_dead_workers */ true,

/*filter_io_workers*/ true);

for (auto &driver :

worker_pool_.GetAllRegisteredDrivers(/* filter_dead_driver */ true)) {

all_workers.push_back(driver);

}

if (all_workers.empty()) {

send_reply_callback(Status::OK(), nullptr, nullptr);

return;

}

// Sort workers for the consistent ordering.

auto sort_func = [](const auto &worker_a, const auto &worker_b) {

// Prioritize drivers over workers. It is because drivers usually have data users

// care more. Note the enum values Driver == 1, Worker == 0.

return (worker_a->GetWorkerType() > worker_b->GetWorkerType())

// If the worker type is the same, order it based on pid (just for consistent

// ordering).

|| ((worker_a->GetWorkerType() == worker_b->GetWorkerType()) &&

(worker_a->GetProcess().GetId() < worker_b->GetProcess().GetId()));

};

std::sort(all_workers.begin(), all_workers.end(), sort_func);

// Query all workers.

auto rpc_replied = std::make_shared<size_t>(0);

auto num_workers = all_workers.size();

bool all_dead = true;

for (const auto &worker : all_workers) {

if (worker->IsDead()) {

*rpc_replied += 1;

continue;

}

all_dead = false;

rpc::GetCoreWorkerStatsRequest request;

request.set_intended_worker_id(worker->WorkerId().Binary());

request.set_include_memory_info(include_memory_info);

request.set_include_task_info(include_task_info);

request.set_limit(limit);

// TODO(sang): Add timeout to the RPC call.

worker->rpc_client()->GetCoreWorkerStats(

request,

[num_workers, rpc_replied, send_reply_callback, on_replied, on_all_replied](

const ray::Status &status, const rpc::GetCoreWorkerStatsReply &r) {

*rpc_replied += 1;

on_replied(status, r);

if (*rpc_replied == num_workers) {

if (on_all_replied) {

on_all_replied();

}

send_reply_callback(Status::OK(), nullptr, nullptr);

}

});

}

if (all_dead) {

send_reply_callback(Status::OK(), nullptr, nullptr);

}

int pending_actor_creations = 0;

int pending_leases = 0;

// Check if any progress is being made on this raylet.

if (worker_pool_.IsWorkerAvailableForScheduling()) {

// Progress is being made in a lease, don't warn.

resource_deadlock_warned_ = 0;

return;

}

auto exemplar = cluster_lease_manager_.AnyPendingLeasesForResourceAcquisition(

&pending_actor_creations, &pending_leases);

// Check if any leases are blocked on resource acquisition.

if (exemplar == nullptr) {

// No pending leases, no need to warn.

resource_deadlock_warned_ = 0;

return;

}

// Push an warning to the driver that a lease is blocked trying to acquire resources.

// To avoid spurious triggers, only take action starting with the second time.

// case resource_deadlock_warned_: 0 => first time, don't do anything yet

// case resource_deadlock_warned_: 1 => second time, print a warning

// case resource_deadlock_warned_: >1 => global gc but don't print any warnings

if (resource_deadlock_warned_++ > 0) {

// Actor references may be caught in cycles, preventing them from being deleted.

// Set should global gc to hopefully free up resource slots.

SetShouldGlobalGC();

// Suppress duplicates warning messages.

if (resource_deadlock_warned_ > 2) {

return;

}

RAY_LOG(WARNING)

<< "The lease with ID " << exemplar->GetLeaseSpecification().LeaseId()

<< " cannot be scheduled right now. You can ignore this message if this "

<< "Ray cluster is expected to auto-scale or if you specified a "

<< "runtime_env for this actor or lease, which may take time to install. "

<< "Otherwise, this is likely due to all cluster resources being claimed "

<< "by actors. To resolve the issue, consider creating fewer actors or "

<< "increasing the resources available to this Ray cluster.\n"

<< "Required resources for this lease: "

<< exemplar->GetLeaseSpecification().GetRequiredPlacementResources().DebugString()

<< "\n"

<< "Available resources on this node: "

<< cluster_resource_scheduler_.GetClusterResourceManager()

.GetNodeResourceViewString(scheduling::NodeID(self_node_id_.Binary()))

<< " In total there are " << pending_leases << " pending leases and "

<< pending_actor_creations << " pending actors on this node.";

RAY_LOG_EVERY_MS(WARNING, 10 * 1000) << cluster_lease_manager_.DebugStr();

}

// Try scheduling leases. Without this, if there's no more leases coming in,

// deadlocked leases are never be scheduled.

cluster_lease_manager_.ScheduleAndGrantLeases();

const NodeID node_id = NodeID::FromBinary(node_info.node_id());

RAY_LOG(DEBUG).WithField(node_id) << "[NodeAdded] Received callback from node id ";

if (node_id == self_node_id_) {

return;

}

// Store address of the new node manager for rpc requests.

remote_node_manager_addresses_[node_id] =

std::make_pair(node_info.node_manager_address(), node_info.node_manager_port());

// Update the resource view if a new message has been sent.

if (auto sync_msg = ray_syncer_.GetSyncMessage(node_id.Binary(),

syncer::MessageType::RESOURCE_VIEW)) {

if (sync_msg) {

ConsumeSyncMessage(sync_msg);

}

}

RAY_LOG(DEBUG).WithField(node_id) << "[NodeRemoved] Received callback from node id ";

if (node_id == self_node_id_) {

if (!shutting_down_) {

std::ostringstream error_message;

error_message

<< "[Timeout] Exiting because this node manager has mistakenly been marked "

"as dead by the GCS: GCS failed to check the health of this node for "

<< RayConfig::instance().health_check_failure_threshold() << " times."

<< " This is likely because the machine or raylet has become overloaded.";

RAY_EVENT(FATAL, "RAYLET_MARKED_DEAD").WithField("node_id", self_node_id_.Hex())

<< error_message.str();

RAY_LOG(FATAL) << error_message.str();

} else {

// No-op since this node is already shutting down, and GCS already knows.

RAY_LOG(INFO).WithField(node_id)

<< "Node is marked as dead by GCS as it's already shutting down.";

return;

}

}

failed_nodes_cache_.insert(node_id);

cluster_lease_manager_.CancelAllLeasesOwnedBy(node_id);

raylet_client_pool_.Disconnect(node_id);

worker_rpc_pool_.Disconnect(node_id);

// Clean up workers that were owned by processes that were on the failed

// node.

for (const auto &[_, worker] : leased_workers_) {

const auto owner_node_id = NodeID::FromBinary(worker->GetOwnerAddress().node_id());

RAY_CHECK(!owner_node_id.IsNil());

if (worker->IsDetachedActor() || owner_node_id != node_id) {

continue;

}

// If the leased worker's owner was on the failed node, then kill the leased

// worker.

RAY_LOG(INFO).WithField(worker->WorkerId()).WithField(owner_node_id)

<< "Killing leased worker because its owner's node died.";

worker->KillAsync(io_service_);

}

// Below, when we remove node_id from all of these data structures, we could

// check that it is actually removed, or log a warning otherwise, but that may

// not be necessary.

// Remove the node from the resource map.

if (!cluster_resource_scheduler_.GetClusterResourceManager().RemoveNode(

scheduling::NodeID(node_id.Binary()))) {

RAY_LOG(DEBUG).WithField(node_id)

<< "Received NodeRemoved callback for an unknown node.";

}

// Remove the node manager address.

const auto node_entry = remote_node_manager_addresses_.find(node_id);

if (node_entry != remote_node_manager_addresses_.end()) {

remote_node_manager_addresses_.erase(node_entry);

}

// Notify the object directory that the node has been removed so that it

// can remove it from any cached locations.

object_directory_.HandleNodeRemoved(node_id);

object_manager_.HandleNodeRemoved(node_id);

RAY_CHECK(!worker_id.IsNil());

RAY_LOG(DEBUG).WithField(worker_id) << "Worker failed";

failed_workers_cache_.insert(worker_id);

cluster_lease_manager_.CancelAllLeasesOwnedBy(worker_id);

for (const auto &[_, worker] : leased_workers_) {

const auto owner_worker_id =

WorkerID::FromBinary(worker->GetOwnerAddress().worker_id());

RAY_CHECK(!owner_worker_id.IsNil());

if (worker->IsDetachedActor() || owner_worker_id != worker_id) {

continue;

}

// If the failed worker was a leased worker's owner, then kill the leased worker.

RAY_LOG(INFO)

.WithField(worker->WorkerId())

.WithField("owner_worker_id", owner_worker_id)

<< "Killing leased worker because its owner died.";

worker->KillAsync(io_service_);

}