wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-27 11:26:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

[Core] Remove Legacy Raylet Code (#9936)

Browse Source 
* Remove a flag and some methods in node manager including HandleDisconnectedActor, ResubmitTask, and HandleTaskReconstruction

* Make actor creator always required + remove raylet transport

* Remove actor reporter + remove FinishAssignedActorCreationTask

* Remove actor tasks.

* Remove finishactortask and switched it to finishactorcreation task

* Remove reconstruction policy.

* Remove lineage cache.

* Formatting.

* Remove actor frontier code.

* Removed build error.

* Revert "Remove reconstruction policy."

This reverts commit 9d25c9bced4da5fbcac5d484d51013345f16513b.

* Recover HandleReconstruction to mark expired objects as failed.
This commit is contained in:
SangBin Cho
2020-08-06 16:37:50 -07:00
committed by GitHub
parent ec2f1a225e
commit 44826878ff
41 changed files with 180 additions and 2751 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.bazelrc
-1
View File
@@ -101,7 +101,6 @@ test:ci --nocache_test_results
test:ci --spawn_strategy=local
test:ci --test_output=errors
test:ci --test_verbose_timeout_warnings
test:ci --test_env=RAY_GCS_ACTOR_SERVICE_ENABLED
aquery:get-toolchain --include_commandline=false
aquery:get-toolchain --noimplicit_deps
BUILD.bazel
-11
View File
@@ -738,17 +738,6 @@ cc_test(
],
)
cc_test(
name = "lineage_cache_test",
srcs = ["src/ray/raylet/lineage_cache_test.cc"],
copts = COPTS,
deps = [
":node_manager_fbs",
":raylet_lib",
"@com_google_googletest//:gtest_main",
],
)
cc_test(
name = "reconstruction_policy_test",
srcs = ["src/ray/raylet/reconstruction_policy_test.cc"],
python/ray/_raylet.pyx
-5
View File
@@ -127,11 +127,6 @@ OPTIMIZED = __OPTIMIZE__
logger = logging.getLogger(__name__)
def gcs_actor_service_enabled():
return (
RayConfig.instance().gcs_actor_service_enabled())
cdef int check_status(const CRayStatus& status) nogil except -1:
if status.ok():
return 0
python/ray/actor.py
+1 -4
View File
@@ -9,7 +9,7 @@ import ray._raylet
import ray.signature as signature
import ray.worker
from ray import ActorClassID, Language
from ray._raylet import PythonFunctionDescriptor, gcs_actor_service_enabled
from ray._raylet import PythonFunctionDescriptor
from ray import cross_language
logger = logging.getLogger(__name__)
@@ -591,9 +591,6 @@ class ActorClass:
worker.current_session_and_job,
original_handle=True)
if name is not None and not gcs_actor_service_enabled():
ray.util.named_actors._register_actor(name, actor_handle)
return actor_handle
python/ray/includes/ray_config.pxd
-2
View File
@@ -87,8 +87,6 @@ cdef extern from "ray/common/ray_config.h" nogil:
int64_t max_direct_call_object_size() const
c_bool gcs_actor_service_enabled() const
c_bool put_small_object_in_memory_store() const
uint32_t max_tasks_in_flight_per_worker() const
python/ray/tests/test_actor_advanced.py
-5
View File
@@ -853,11 +853,6 @@ def test_actor_creation_task_crash(ray_start_regular):
ray.get(ra.f.remote())
@pytest.mark.skipif(
os.environ.get("RAY_GCS_ACTOR_SERVICE_ENABLED") != "true",
reason=("This edge case is not handled when GCS actor management is off. "
"We won't fix this because GCS actor management "
"will be on by default anyway."))
@pytest.mark.parametrize(
"ray_start_regular", [{
"num_cpus": 2,
python/ray/tests/test_gcs_fault_tolerance.py
-10
View File
@@ -1,4 +1,3 @@
import os
import sys
import ray
@@ -21,9 +20,6 @@ def increase(x):
return x + 1
@pytest.mark.skipif(
os.environ.get("RAY_GCS_ACTOR_SERVICE_ENABLED", "true") != "true",
reason=("This testcase can only be run when GCS actor management is on."))
@pytest.mark.parametrize(
"ray_start_regular",
[generate_internal_config_map(num_heartbeats_timeout=20)],
@@ -47,9 +43,6 @@ def test_gcs_server_restart(ray_start_regular):
assert result == 2
@pytest.mark.skipif(
os.environ.get("RAY_GCS_ACTOR_SERVICE_ENABLED", "true") != "true",
reason=("This testcase can only be run when GCS actor management is on."))
@pytest.mark.parametrize(
"ray_start_regular",
[generate_internal_config_map(num_heartbeats_timeout=20)],
@@ -69,9 +62,6 @@ def test_gcs_server_restart_during_actor_creation(ray_start_regular):
assert len(unready) == 0
@pytest.mark.skipif(
os.environ.get("RAY_GCS_ACTOR_SERVICE_ENABLED", "true") != "true",
reason=("This testcase can only be run when GCS actor management is on."))
@pytest.mark.parametrize(
"ray_start_cluster_head",
[generate_internal_config_map(num_heartbeats_timeout=20)],
python/ray/util/__init__.py
+1 -2
View File
@@ -2,7 +2,7 @@ from ray.util import iter
from ray.util.actor_pool import ActorPool
from ray.util.debug import log_once, disable_log_once_globally, \
enable_periodic_logging
from ray.util.named_actors import get_actor, register_actor
from ray.util.named_actors import get_actor
__all__ = [
"ActorPool",
@@ -11,5 +11,4 @@ __all__ = [
"get_actor",
"iter",
"log_once",
"register_actor",
]
python/ray/util/named_actors.py
+2 -69
View File
@@ -1,44 +1,13 @@
import logging
import ray
import ray.cloudpickle as pickle
from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
from ray.gcs_utils import ActorTableData
logger = logging.getLogger(__name__)
def _calculate_key(name):
"""Generate a Redis key with the given name.
Args:
name: The name of the named actor.
Returns:
The key to use for storing a named actor in Redis.
"""
return b"Actor:" + name.encode("ascii")
def _get_actor(name):
if ray._raylet.gcs_actor_service_enabled():
worker = ray.worker.global_worker
handle = worker.core_worker.get_named_actor_handle(name)
else:
actor_name = _calculate_key(name)
pickled_state = _internal_kv_get(actor_name)
if pickled_state is None:
raise ValueError(
"The actor with name={} doesn't exist".format(name))
handle = pickle.loads(pickled_state)
# If the actor state is dead, that means that this name is reusable.
# We don't delete the name entry from key value store when
# the actor is killed because ray.kill is asynchronous,
# and it can cause worker leaks.
actor_info = ray.actors(actor_id=handle._actor_id.hex())
actor_state = actor_info.get("State", None)
if actor_state and actor_state == ActorTableData.DEAD:
raise ValueError("The actor with name={} is dead.".format(name))
worker = ray.worker.global_worker
handle = worker.core_worker.get_named_actor_handle(name)
return handle
@@ -56,39 +25,3 @@ def get_actor(name: str) -> ray.actor.ActorHandle:
logger.warning("ray.util.get_actor has been moved to ray.get_actor and "
"will be removed in the future.")
return _get_actor(name)
def _register_actor(name, actor_handle):
if not isinstance(name, str):
raise TypeError("The name argument must be a string.")
if not isinstance(actor_handle, ray.actor.ActorHandle):
raise TypeError("The actor_handle argument must be an ActorHandle "
"object.")
actor_name = _calculate_key(name)
# First check if the actor already exists.
try:
_get_actor(name)
exists = True
except ValueError:
exists = False
if exists:
raise ValueError("An actor with name={} already exists or there "
"was timeout in getting this actor handle."
.format(name))
# Add the actor to Redis if it does not already exist.
_internal_kv_put(actor_name, pickle.dumps(actor_handle), overwrite=True)
def register_actor(name, actor_handle):
"""Register a named actor under a string key.
Args:
name: The name of the named actor.
actor_handle: The actor object to be associated with this name
"""
logger.warning("ray.util.register_actor is deprecated. To create a "
"named, detached actor, use Actor.options(name=\"name\").")
return _register_actor(name, actor_handle)
src/ray/common/ray_config_def.h
-4
View File
@@ -312,10 +312,6 @@ RAY_CONFIG(bool, plasma_store_as_thread, false)
/// changed. When the address changed, we will resubscribe again.
RAY_CONFIG(int64_t, gcs_service_address_check_interval_milliseconds, 1000)
RAY_CONFIG(bool, gcs_actor_service_enabled,
getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") == nullptr ||
getenv("RAY_GCS_ACTOR_SERVICE_ENABLED") == std::string("true"))
/// The batch size for metrics export.
RAY_CONFIG(int64_t, metrics_report_batch_size, 100)
src/ray/core_worker/actor_manager.cc
-29
View File
@@ -91,35 +91,6 @@ bool ActorManager::AddActorHandle(std::unique_ptr<ActorHandle> actor_handle,
std::placeholders::_1, std::placeholders::_2);
RAY_CHECK_OK(gcs_client_->Actors().AsyncSubscribe(
actor_id, actor_notification_callback, nullptr));
if (!RayConfig::instance().gcs_actor_service_enabled()) {
RAY_CHECK(reference_counter_->SetDeleteCallback(
actor_creation_return_id,
[this, actor_id, is_owner_handle](const ObjectID &object_id) {
if (is_owner_handle) {
// If we own the actor and the actor handle is no longer in scope,
// terminate the actor. We do not do this if the GCS service is
// enabled since then the GCS will terminate the actor for us.
// TODO(sang): Remove this block once gcs_actor_service is enabled by
// default.
RAY_LOG(INFO) << "Owner's handle and creation ID " << object_id
<< " has gone out of scope, sending message to actor "
<< actor_id << " to do a clean exit.";
RAY_CHECK(CheckActorHandleExists(actor_id));
direct_actor_submitter_->KillActor(actor_id,
/*force_kill=*/false,
/*no_restart=*/false);
// TODO(swang): Erase the actor handle once all refs to the actor
// have gone out of scope. We cannot erase it here in case the
// language frontend receives another ref to the same actor. In this
// case, we must remember the last task counter that we sent to the
// actor.
// TODO(ekl) we can't unsubscribe to actor notifications here due to
// https://github.com/ray-project/ray/pull/6885
}
}));
}
}
return inserted;
src/ray/core_worker/actor_reporter.cc
-37
View File
@@ -1,37 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "ray/core_worker/actor_reporter.h"
#include "ray/gcs/pb_util.h"
#include "ray/gcs/redis_accessor.h"
namespace ray {
void ActorReporter::PublishTerminatedActor(const TaskSpecification &actor_creation_task) {
auto actor_id = actor_creation_task.ActorCreationId();
auto data = gcs::CreateActorTableData(actor_creation_task, rpc::Address(),
rpc::ActorTableData::DEAD, 0);
auto update_callback = [actor_id](Status status) {
if (!status.ok()) {
// Only one node at a time should succeed at creating or updating the actor.
RAY_LOG(ERROR) << "Failed to update state to DEAD for actor " << actor_id
<< ", error: " << status.ToString();
}
};
RAY_CHECK_OK(gcs_client_->Actors().AsyncRegister(data, update_callback));
}
} // namespace ray
src/ray/core_worker/actor_reporter.h
-44
View File
@@ -1,44 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#pragma once
#include "ray/gcs/redis_gcs_client.h"
namespace ray {
// TODO(SANG): This class won't be needed once GCS actor mangement becomes the default.
// Interface for testing.
class ActorReporterInterface {
public:
virtual void PublishTerminatedActor(const TaskSpecification &actor_creation_task) = 0;
virtual ~ActorReporterInterface() {}
};
class ActorReporter : public ActorReporterInterface {
public:
ActorReporter(std::shared_ptr<gcs::GcsClient> gcs_client) : gcs_client_(gcs_client) {}
~ActorReporter() {}
/// Called when an actor that we own can no longer be restarted.
void PublishTerminatedActor(const TaskSpecification &actor_creation_task) override;
private:
/// GCS client
std::shared_ptr<gcs::GcsClient> gcs_client_;
};
} // namespace ray
src/ray/core_worker/core_worker.cc
+5 -28
View File
@@ -20,7 +20,6 @@
#include "ray/common/task/task_util.h"
#include "ray/core_worker/context.h"
#include "ray/core_worker/transport/direct_actor_transport.h"
#include "ray/core_worker/transport/raylet_transport.h"
#include "ray/gcs/gcs_client/service_based_gcs_client.h"
#include "ray/stats/stats.h"
#include "ray/util/process.h"
@@ -286,9 +285,6 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
auto execute_task =
std::bind(&CoreWorker::ExecuteTask, this, std::placeholders::_1,
std::placeholders::_2, std::placeholders::_3, std::placeholders::_4);
raylet_task_receiver_ =
std::unique_ptr<CoreWorkerRayletTaskReceiver>(new CoreWorkerRayletTaskReceiver(
worker_context_.GetWorkerID(), local_raylet_client_, execute_task));
direct_task_receiver_ =
std::unique_ptr<CoreWorkerDirectTaskReceiver>(new CoreWorkerDirectTaskReceiver(
worker_context_, task_execution_service_, execute_task,
@@ -378,8 +374,6 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
boost::asio::chrono::milliseconds(kInternalHeartbeatMillis));
internal_timer_.async_wait(boost::bind(&CoreWorker::InternalHeartbeat, this, _1));
actor_reporter_ = std::unique_ptr<ActorReporter>(new ActorReporter(gcs_client_));
plasma_store_provider_.reset(new CoreWorkerPlasmaStoreProvider(
options_.store_socket, local_raylet_client_, reference_counter_,
options_.check_signals,
@@ -407,7 +401,7 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
});
};
task_manager_.reset(new TaskManager(
memory_store_, reference_counter_, actor_reporter_,
memory_store_, reference_counter_,
[this](TaskSpecification &spec, bool delay) {
if (delay) {
// Retry after a delay to emulate the existing Raylet reconstruction
@@ -463,10 +457,8 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
new raylet::RayletClient(std::move(grpc_client)));
};
std::shared_ptr<ActorCreatorInterface> actor_creator = nullptr;
if (RayConfig::instance().gcs_actor_service_enabled()) {
actor_creator = std::make_shared<DefaultActorCreator>(gcs_client_);
}
std::shared_ptr<ActorCreatorInterface> actor_creator =
std::make_shared<DefaultActorCreator>(gcs_client_);
direct_actor_submitter_ = std::shared_ptr<CoreWorkerDirectActorTaskSubmitter>(
new CoreWorkerDirectActorTaskSubmitter(client_factory, memory_store_,
@@ -477,8 +469,9 @@ CoreWorker::CoreWorker(const CoreWorkerOptions &options, const WorkerID &worker_
rpc_address_, local_raylet_client_, client_factory, raylet_client_factory,
memory_store_, task_manager_, local_raylet_id,
RayConfig::instance().worker_lease_timeout_milliseconds(),
std::move(actor_creator),
RayConfig::instance().max_tasks_in_flight_per_worker(),
std::move(actor_creator), boost::asio::steady_timer(io_service_)));
boost::asio::steady_timer(io_service_)));
future_resolver_.reset(new FutureResolver(memory_store_, client_factory, rpc_address_));
// Unfortunately the raylet client has to be constructed after the receivers.
if (direct_task_receiver_ != nullptr) {
@@ -1426,7 +1419,6 @@ const ActorHandle *CoreWorker::GetActorHandle(const ActorID &actor_id) const {
std::pair<const ActorHandle *, Status> CoreWorker::GetNamedActorHandle(
const std::string &name) {
RAY_CHECK(RayConfig::instance().gcs_actor_service_enabled());
RAY_CHECK(!name.empty());
if (options_.is_local_mode) {
return GetNamedActorHandleLocalMode(name);
@@ -1783,20 +1775,6 @@ Status CoreWorker::GetAndPinArgsForExecutor(const TaskSpecification &task,
return Status::OK();
}
void CoreWorker::HandleAssignTask(const rpc::AssignTaskRequest &request,
rpc::AssignTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) {
if (HandleWrongRecipient(WorkerID::FromBinary(request.intended_worker_id()),
send_reply_callback)) {
return;
}
task_queue_length_ += 1;
task_execution_service_.post([=] {
raylet_task_receiver_->HandleAssignTask(request, reply, send_reply_callback);
});
}
void CoreWorker::HandlePushTask(const rpc::PushTaskRequest &request,
rpc::PushTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) {
@@ -1882,7 +1860,6 @@ void CoreWorker::HandleWaitForActorOutOfScope(
const rpc::WaitForActorOutOfScopeRequest &request,
rpc::WaitForActorOutOfScopeReply *reply, rpc::SendReplyCallback send_reply_callback) {
// Currently WaitForActorOutOfScope is only used when GCS actor service is enabled.
RAY_CHECK(RayConfig::instance().gcs_actor_service_enabled());
if (HandleWrongRecipient(WorkerID::FromBinary(request.intended_worker_id()),
send_reply_callback)) {
return;
src/ray/core_worker/core_worker.h
-13
View File
@@ -20,7 +20,6 @@
#include "ray/common/placement_group.h"
#include "ray/core_worker/actor_handle.h"
#include "ray/core_worker/actor_manager.h"
#include "ray/core_worker/actor_reporter.h"
#include "ray/core_worker/common.h"
#include "ray/core_worker/context.h"
#include "ray/core_worker/future_resolver.h"
@@ -31,7 +30,6 @@
#include "ray/core_worker/store_provider/plasma_store_provider.h"
#include "ray/core_worker/transport/direct_actor_transport.h"
#include "ray/core_worker/transport/direct_task_transport.h"
#include "ray/core_worker/transport/raylet_transport.h"
#include "ray/gcs/redis_gcs_client.h"
#include "ray/gcs/subscription_executor.h"
#include "ray/raylet_client/raylet_client.h"
@@ -735,11 +733,6 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
/// post work to the appropriate event loop.
///
/// Implements gRPC server handler.
void HandleAssignTask(const rpc::AssignTaskRequest &request,
rpc::AssignTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) override;
/// Implements gRPC server handler.
void HandlePushTask(const rpc::PushTaskRequest &request, rpc::PushTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) override;
@@ -1035,9 +1028,6 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
// Tracks the currently pending tasks.
std::shared_ptr<TaskManager> task_manager_;
// Interface for publishing actor death event for actor creation failure.
std::shared_ptr<ActorReporter> actor_reporter_;
// Interface to submit tasks directly to other actors.
std::shared_ptr<CoreWorkerDirectActorTaskSubmitter> direct_actor_submitter_;
@@ -1095,9 +1085,6 @@ class CoreWorker : public rpc::CoreWorkerServiceHandler {
/// of that resource allocated for this worker. This is set on task assignment.
std::shared_ptr<ResourceMappingType> resource_ids_ GUARDED_BY(mutex_);
// Interface that receives tasks from the raylet.
std::unique_ptr<CoreWorkerRayletTaskReceiver> raylet_task_receiver_;
/// Common rpc service for all worker modules.
rpc::CoreWorkerGrpcService grpc_service_;
src/ray/core_worker/task_manager.cc
-6
View File
@@ -445,12 +445,6 @@ void TaskManager::MarkPendingTaskFailed(const TaskID &task_id,
const auto object_id = ObjectID::ForTaskReturn(task_id, /*index=*/i + 1);
RAY_UNUSED(in_memory_store_->Put(RayObject(error_type), object_id));
}
if (spec.IsActorCreationTask()) {
// Publish actor death if actor creation task failed after
// a number of retries.
actor_reporter_->PublishTerminatedActor(spec);
}
}
absl::optional<TaskSpecification> TaskManager::GetTaskSpec(const TaskID &task_id) const {
src/ray/core_worker/task_manager.h
-6
View File
@@ -19,7 +19,6 @@
#include "absl/synchronization/mutex.h"
#include "ray/common/id.h"
#include "ray/common/task/task.h"
#include "ray/core_worker/actor_reporter.h"
#include "ray/core_worker/store_provider/memory_store/memory_store.h"
#include "src/ray/protobuf/core_worker.pb.h"
#include "src/ray/protobuf/gcs.pb.h"
@@ -58,13 +57,11 @@ class TaskManager : public TaskFinisherInterface, public TaskResubmissionInterfa
public:
TaskManager(std::shared_ptr<CoreWorkerMemoryStore> in_memory_store,
std::shared_ptr<ReferenceCounter> reference_counter,
std::shared_ptr<ActorReporterInterface> actor_reporter,
RetryTaskCallback retry_task_callback,
const std::function<bool(const ClientID &node_id)> &check_node_alive,
ReconstructObjectCallback reconstruct_object_callback)
: in_memory_store_(in_memory_store),
reference_counter_(reference_counter),
actor_reporter_(actor_reporter),
retry_task_callback_(retry_task_callback),
check_node_alive_(check_node_alive),
reconstruct_object_callback_(reconstruct_object_callback) {
@@ -234,9 +231,6 @@ class TaskManager : public TaskFinisherInterface, public TaskResubmissionInterfa
/// submitted tasks (dependencies and return objects).
std::shared_ptr<ReferenceCounter> reference_counter_;
// Interface for publishing actor creation.
std::shared_ptr<ActorReporterInterface> actor_reporter_;
/// Called when a task should be retried.
const RetryTaskCallback retry_task_callback_;
src/ray/core_worker/test/actor_manager_test.cc
-1
View File
@@ -18,7 +18,6 @@
#include "gtest/gtest.h"
#include "ray/common/task/task_spec.h"
#include "ray/common/test_util.h"
#include "ray/core_worker/actor_reporter.h"
#include "ray/core_worker/reference_count.h"
#include "ray/core_worker/transport/direct_actor_transport.h"
#include "ray/gcs/redis_accessor.h"
src/ray/core_worker/test/direct_task_transport_test.cc
+62 -17
View File
@@ -166,6 +166,22 @@ class MockRayletClient : public WorkerLeaseInterface {
std::list<rpc::ClientCallback<rpc::CancelWorkerLeaseReply>> cancel_callbacks = {};
};
class MockActorCreator : public ActorCreatorInterface {
public:
MockActorCreator() {}
Status RegisterActor(const TaskSpecification &task_spec) override {
return Status::OK();
};
Status AsyncCreateActor(const TaskSpecification &task_spec,
const gcs::StatusCallback &callback) override {
return Status::OK();
}
~MockActorCreator() {}
};
TEST(TestMemoryStore, TestPromoteToPlasma) {
size_t num_plasma_puts = 0;
auto mem = std::make_shared<CoreWorkerMemoryStore>(
@@ -321,8 +337,10 @@ TEST(DirectTaskTransportTest, TestSubmitOneTask) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
@@ -355,8 +373,10 @@ TEST(DirectTaskTransportTest, TestHandleTaskFailure) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -382,8 +402,10 @@ TEST(DirectTaskTransportTest, TestConcurrentWorkerLeases) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -430,8 +452,10 @@ TEST(DirectTaskTransportTest, TestReuseWorkerLease) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -485,8 +509,10 @@ TEST(DirectTaskTransportTest, TestRetryLeaseCancellation) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -538,8 +564,10 @@ TEST(DirectTaskTransportTest, TestConcurrentCancellationAndSubmission) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -588,8 +616,10 @@ TEST(DirectTaskTransportTest, TestWorkerNotReusedOnError) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -629,8 +659,10 @@ TEST(DirectTaskTransportTest, TestWorkerNotReturnedOnExit) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -669,9 +701,10 @@ TEST(DirectTaskTransportTest, TestSpillback) {
return client;
};
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory,
lease_client_factory, store, task_finisher,
ClientID::Nil(), kLongTimeout);
ClientID::Nil(), kLongTimeout, actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -726,9 +759,10 @@ TEST(DirectTaskTransportTest, TestSpillbackRoundTrip) {
};
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto local_raylet_id = ClientID::FromRandom();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory,
lease_client_factory, store, task_finisher,
local_raylet_id, kLongTimeout);
local_raylet_id, kLongTimeout, actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -781,8 +815,10 @@ void TestSchedulingKey(const std::shared_ptr<CoreWorkerMemoryStore> store,
auto worker_client = std::make_shared<MockWorkerClient>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
ASSERT_TRUE(submitter.SubmitTask(same1).ok());
ASSERT_TRUE(submitter.SubmitTask(same2).ok());
@@ -891,9 +927,10 @@ TEST(DirectTaskTransportTest, TestWorkerLeaseTimeout) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(),
/*lease_timeout_ms=*/5);
/*lease_timeout_ms=*/5, actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -943,8 +980,10 @@ TEST(DirectTaskTransportTest, TestKillExecutingTask) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -988,8 +1027,10 @@ TEST(DirectTaskTransportTest, TestKillPendingTask) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -1014,8 +1055,10 @@ TEST(DirectTaskTransportTest, TestKillResolvingTask) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout);
task_finisher, ClientID::Nil(), kLongTimeout,
actor_creator);
std::unordered_map<std::string, double> empty_resources;
ray::FunctionDescriptor empty_descriptor =
ray::FunctionDescriptorBuilder::BuildPython("", "", "", "");
@@ -1042,6 +1085,7 @@ TEST(DirectTaskTransportTest, TestPipeliningConcurrentWorkerLeases) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
// Set max_tasks_in_flight_per_worker to a value larger than 1 to enable the pipelining
// of task submissions. This is done by passing a max_tasks_in_flight_per_worker
@@ -1049,7 +1093,7 @@ TEST(DirectTaskTransportTest, TestPipeliningConcurrentWorkerLeases) {
uint32_t max_tasks_in_flight_per_worker = 10;
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout,
max_tasks_in_flight_per_worker);
actor_creator, max_tasks_in_flight_per_worker);
// Prepare 20 tasks and save them in a vector.
std::unordered_map<std::string, double> empty_resources;
@@ -1110,6 +1154,7 @@ TEST(DirectTaskTransportTest, TestPipeliningReuseWorkerLease) {
auto store = std::make_shared<CoreWorkerMemoryStore>();
auto factory = [&](const rpc::Address &addr) { return worker_client; };
auto task_finisher = std::make_shared<MockTaskFinisher>();
auto actor_creator = std::make_shared<MockActorCreator>();
// Set max_tasks_in_flight_per_worker to a value larger than 1 to enable the pipelining
// of task submissions. This is done by passing a max_tasks_in_flight_per_worker
@@ -1117,7 +1162,7 @@ TEST(DirectTaskTransportTest, TestPipeliningReuseWorkerLease) {
uint32_t max_tasks_in_flight_per_worker = 10;
CoreWorkerDirectTaskSubmitter submitter(address, raylet_client, factory, nullptr, store,
task_finisher, ClientID::Nil(), kLongTimeout,
max_tasks_in_flight_per_worker);
actor_creator, max_tasks_in_flight_per_worker);
// prepare 30 tasks and save them in a vector
std::unordered_map<std::string, double> empty_resources;
src/ray/core_worker/test/task_manager_test.cc
+1 -12
View File
@@ -17,7 +17,6 @@
#include "gtest/gtest.h"
#include "ray/common/task/task_spec.h"
#include "ray/common/test_util.h"
#include "ray/core_worker/actor_reporter.h"
#include "ray/core_worker/reference_count.h"
#include "ray/core_worker/store_provider/memory_store/memory_store.h"
@@ -35,14 +34,6 @@ TaskSpecification CreateTaskHelper(uint64_t num_returns,
return task;
}
class MockActorManager : public ActorReporterInterface {
void PublishTerminatedActor(const TaskSpecification &actor_creation_task) override {
num_terminations += 1;
}
int num_terminations = 0;
};
class TaskManagerTest : public ::testing::Test {
public:
TaskManagerTest(bool lineage_pinning_enabled = false)
@@ -50,8 +41,7 @@ class TaskManagerTest : public ::testing::Test {
reference_counter_(std::shared_ptr<ReferenceCounter>(new ReferenceCounter(
rpc::Address(),
/*distributed_ref_counting_enabled=*/true, lineage_pinning_enabled))),
actor_reporter_(std::shared_ptr<ActorReporterInterface>(new MockActorManager())),
manager_(store_, reference_counter_, actor_reporter_,
manager_(store_, reference_counter_,
[this](TaskSpecification &spec, bool delay) {
num_retries_++;
return Status::OK();
@@ -63,7 +53,6 @@ class TaskManagerTest : public ::testing::Test {
std::shared_ptr<CoreWorkerMemoryStore> store_;
std::shared_ptr<ReferenceCounter> reference_counter_;
std::shared_ptr<ActorReporterInterface> actor_reporter_;
bool all_nodes_alive_ = true;
std::vector<ObjectID> objects_to_recover_;
TaskManager manager_;
src/ray/core_worker/transport/direct_task_transport.cc
+2 -2
View File
@@ -21,7 +21,7 @@ namespace ray {
Status CoreWorkerDirectTaskSubmitter::SubmitTask(TaskSpecification task_spec) {
RAY_LOG(DEBUG) << "Submit task " << task_spec.TaskId();
if (actor_creator_ && task_spec.IsActorCreationTask()) {
if (task_spec.IsActorCreationTask()) {
// Synchronously register the actor to GCS server.
// Previously, we asynchronously registered the actor after all its dependencies were
// resolved. This caused a problem: if the owner of the actor dies before dependencies
@@ -37,7 +37,7 @@ Status CoreWorkerDirectTaskSubmitter::SubmitTask(TaskSpecification task_spec) {
resolver_.ResolveDependencies(task_spec, [this, task_spec]() {
RAY_LOG(DEBUG) << "Task dependencies resolved " << task_spec.TaskId();
if (actor_creator_ && task_spec.IsActorCreationTask()) {
if (task_spec.IsActorCreationTask()) {
// If gcs actor management is enabled, the actor creation task will be sent to
// gcs server directly after the in-memory dependent objects are resolved. For
// more details please see the protocol of actor management based on gcs.
src/ray/core_worker/transport/direct_task_transport.h
+1 -2
View File
@@ -54,10 +54,9 @@ class CoreWorkerDirectTaskSubmitter {
rpc::ClientFactoryFn client_factory, LeaseClientFactoryFn lease_client_factory,
std::shared_ptr<CoreWorkerMemoryStore> store,
std::shared_ptr<TaskFinisherInterface> task_finisher, ClientID local_raylet_id,
int64_t lease_timeout_ms,
int64_t lease_timeout_ms, std::shared_ptr<ActorCreatorInterface> actor_creator,
uint32_t max_tasks_in_flight_per_worker =
RayConfig::instance().max_tasks_in_flight_per_worker(),
std::shared_ptr<ActorCreatorInterface> actor_creator = nullptr,
absl::optional<boost::asio::steady_timer> cancel_timer = absl::nullopt)
: rpc_address_(rpc_address),
local_lease_client_(lease_client),
src/ray/core_worker/transport/raylet_transport.cc
-87
View File
@@ -1,87 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "ray/core_worker/transport/raylet_transport.h"
#include "ray/common/common_protocol.h"
#include "ray/common/task/task.h"
namespace ray {
CoreWorkerRayletTaskReceiver::CoreWorkerRayletTaskReceiver(
const WorkerID &worker_id, std::shared_ptr<raylet::RayletClient> &raylet_client,
const TaskHandler &task_handler)
: worker_id_(worker_id), raylet_client_(raylet_client), task_handler_(task_handler) {}
void CoreWorkerRayletTaskReceiver::HandleAssignTask(
const rpc::AssignTaskRequest &request, rpc::AssignTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) {
const Task task(request.task());
const auto &task_spec = task.GetTaskSpecification();
RAY_LOG(DEBUG) << "Received task " << task_spec.TaskId() << " is create "
<< task_spec.IsActorCreationTask();
// Set the resource IDs for this task.
// TODO: convert the resource map to protobuf and change this.
auto resource_ids = std::make_shared<ResourceMappingType>();
auto resource_infos =
flatbuffers::GetRoot<protocol::ResourceIdSetInfos>(request.resource_ids().data())
->resource_infos();
for (size_t i = 0; i < resource_infos->size(); ++i) {
auto const &fractional_resource_ids = resource_infos->Get(i);
auto &acquired_resources =
(*resource_ids)[string_from_flatbuf(*fractional_resource_ids->resource_name())];
size_t num_resource_ids = fractional_resource_ids->resource_ids()->size();
size_t num_resource_fractions = fractional_resource_ids->resource_fractions()->size();
RAY_CHECK(num_resource_ids == num_resource_fractions);
RAY_CHECK(num_resource_ids > 0);
for (size_t j = 0; j < num_resource_ids; ++j) {
int64_t resource_id = fractional_resource_ids->resource_ids()->Get(j);
double resource_fraction = fractional_resource_ids->resource_fractions()->Get(j);
if (num_resource_ids > 1) {
int64_t whole_fraction = resource_fraction;
RAY_CHECK(whole_fraction == resource_fraction);
}
acquired_resources.push_back(std::make_pair(resource_id, resource_fraction));
}
}
std::vector<std::shared_ptr<RayObject>> results;
ReferenceCounter::ReferenceTableProto borrower_refs;
// NOTE(swang): Distributed ref counting does not work for the raylet
// transport.
auto status = task_handler_(task_spec, resource_ids, &results, &borrower_refs);
if (status.IsSystemExit()) {
return;
}
RAY_LOG(DEBUG) << "Assigned task " << task_spec.TaskId() << " finished execution.";
// Notify raylet that current task is done via a `TaskDone` message. This is to
// ensure that the task is marked as finished by raylet only after previous
// raylet client calls are completed. For example, if the worker sends a
// NotifyUnblocked message that it is no longer blocked in a `ray.get`
// on the normal raylet socket, then completes an assigned task, we
// need to guarantee that raylet gets the former message first before
// marking the task as completed. This is why a `TaskDone` message
// is required - without it, it's possible that raylet receives
// rpc reply first before the NotifyUnblocked message arrives,
// as they use different connections, the `TaskDone` message is sent
// to raylet via the same connection so the order is guaranteed.
RAY_UNUSED(raylet_client_->TaskDone());
// Send rpc reply.
send_reply_callback(status, nullptr, nullptr);
}
} // namespace ray
src/ray/core_worker/transport/raylet_transport.h
-61
View File
@@ -1,61 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#pragma once
#include <list>
#include "ray/common/ray_object.h"
#include "ray/core_worker/reference_count.h"
#include "ray/raylet_client/raylet_client.h"
#include "ray/rpc/worker/core_worker_server.h"
namespace ray {
class CoreWorkerRayletTaskReceiver {
public:
using TaskHandler =
std::function<Status(const TaskSpecification &task_spec,
const std::shared_ptr<ResourceMappingType> &resource_ids,
std::vector<std::shared_ptr<RayObject>> *return_objects,
ReferenceCounter::ReferenceTableProto *borrower_refs)>;
CoreWorkerRayletTaskReceiver(const WorkerID &worker_id,
std::shared_ptr<raylet::RayletClient> &raylet_client,
const TaskHandler &task_handler);
/// Handle a `AssignTask` request.
/// The implementation can handle this request asynchronously. When handling is done,
/// the `send_reply_callback` should be called.
///
/// \param[in] request The request message.
/// \param[out] reply The reply message.
/// \param[in] send_reply_callback The callback to be called when the request is done.
void HandleAssignTask(const rpc::AssignTaskRequest &request,
rpc::AssignTaskReply *reply,
rpc::SendReplyCallback send_reply_callback);
private:
// WorkerID of this worker.
WorkerID worker_id_;
/// Reference to the core worker's raylet client. This is a pointer ref so that it
/// can be initialized by core worker after this class is constructed.
std::shared_ptr<raylet::RayletClient> &raylet_client_;
/// The callback function to process a task.
TaskHandler task_handler_;
/// The callback to process arg wait complete.
std::function<void(int64_t)> on_wait_complete_;
};
} // namespace ray
src/ray/gcs/gcs_server/gcs_actor_manager.cc
+1 -3
View File
@@ -944,9 +944,7 @@ void GcsActorManager::LoadInitialData(const EmptyCallback &done) {
}
// Notify raylets to release unused workers.
if (RayConfig::instance().gcs_actor_service_enabled()) {
gcs_actor_scheduler_->ReleaseUnusedWorkers(node_to_workers);
}
gcs_actor_scheduler_->ReleaseUnusedWorkers(node_to_workers);
RAY_LOG(DEBUG) << "The number of registered actors is " << registered_actors_.size()
<< ", and the number of created actors is " << created_actors_.size();
src/ray/gcs/redis_gcs_client.cc
+1 -5
View File
@@ -68,11 +68,7 @@ Status RedisGcsClient::Connect(boost::asio::io_service &io_service) {
resource_table_.reset(new DynamicResourceTable({primary_context}, this));
worker_table_.reset(new WorkerTable(shard_contexts, this));
if (RayConfig::instance().gcs_actor_service_enabled()) {
actor_accessor_.reset(new RedisActorInfoAccessor(this));
} else {
actor_accessor_.reset(new RedisLogBasedActorInfoAccessor(this));
}
actor_accessor_.reset(new RedisActorInfoAccessor(this));
job_accessor_.reset(new RedisJobInfoAccessor(this));
object_accessor_.reset(new RedisObjectInfoAccessor(this));
src/ray/protobuf/core_worker.proto
-20
View File
@@ -55,24 +55,6 @@ message ActorHandle {
int64 max_task_retries = 9;
}
message AssignTaskRequest {
// The ID of the worker this message is intended for. This is used to
// ensure that workers don't try to execute tasks assigned to workers
// that used to be bound to the same port.
bytes intended_worker_id = 1;
// The task to be pushed.
Task task = 2;
// A list of the resources reserved for this worker.
// TODO(zhijunfu): `resource_ids` is represented as
// flatbutters-serialized bytes, will be moved to protobuf later.
bytes resource_ids = 3;
}
message AssignTaskReply {
}
message ReturnObject {
// Object ID.
bytes object_id = 1;
@@ -286,8 +268,6 @@ message PlasmaObjectReadyReply {
}
service CoreWorkerService {
// Push a task to a worker from the raylet.
rpc AssignTask(AssignTaskRequest) returns (AssignTaskReply);
// Push a task directly to this worker from another.
rpc PushTask(PushTaskRequest) returns (PushTaskReply);
// Reply from raylet that wait for direct actor call args has completed.
src/ray/protobuf/node_manager.proto
-13
View File
@@ -89,17 +89,6 @@ message CancelWorkerLeaseReply {
bool success = 1;
}
message ForwardTaskRequest {
// The ID of the task to be forwarded.
bytes task_id = 1;
// The tasks in the uncommitted lineage of the forwarded task. This
// should include task_id.
repeated Task uncommitted_tasks = 2;
}
message ForwardTaskReply {
}
message PinObjectIDsRequest {
// Address of the owner to ask when to unpin the objects.
Address owner_address = 1;
@@ -171,8 +160,6 @@ service NodeManagerService {
// Cancel a pending lease request. This only returns success if the
// lease request was not yet granted.
rpc CancelWorkerLease(CancelWorkerLeaseRequest) returns (CancelWorkerLeaseReply);
// Forward a task and its uncommitted lineage to the remote node manager.
rpc ForwardTask(ForwardTaskRequest) returns (ForwardTaskReply);
// Pin the provided object IDs.
rpc PinObjectIDs(PinObjectIDsRequest) returns (PinObjectIDsReply);
// Get the current node stats.
src/ray/raylet/format/node_manager.fbs
-24
View File
@@ -55,22 +55,6 @@ enum MessageType:int {
NotifyDirectCallTaskBlocked,
// Notify the current worker is unblocked. This is only used by direct task calls.
NotifyDirectCallTaskUnblocked,
// A request to get the task frontier for an actor, called by the actor when
// saving a checkpoint.
GetActorFrontierRequest,
// The ActorFrontier response to a GetActorFrontierRequest. The raylet
// returns the actor's per-handle task counts and execution dependencies,
// which can later be used as the argument to SetActorFrontier
// when resuming from the checkpoint.
GetActorFrontierReply,
// A request to set the task frontier for an actor, called when resuming from
// a checkpoint. The raylet will update the actor's per-handle task
// counts and execution dependencies, discard any tasks that already executed
// before the checkpoint, and make any tasks on the frontier runnable by
// making their execution dependencies available.
SetActorFrontier,
// A node manager request to process a task forwarded from another node manager.
ForwardTaskRequest,
// Wait for objects to be ready either from local or remote Plasma stores.
WaitRequest,
// The response message to WaitRequest; replies with the objects found and objects
@@ -182,14 +166,6 @@ table RegisterNodeManagerRequest {
client_id: string;
}
table ForwardTaskRequest {
// The ID of the task to be forwarded.
task_id: string;
// The tasks in the uncommitted lineage of the forwarded task. This
// should include task_id.
uncommitted_tasks: [Task];
}
// Mimics the Address protobuf.
table Address {
raylet_id: string;
src/ray/raylet/lineage_cache.cc
-412
View File
@@ -1,412 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "ray/raylet/lineage_cache.h"
#include <sstream>
#include "ray/gcs/redis_gcs_client.h"
#include "ray/stats/stats.h"
namespace ray {
namespace raylet {
LineageEntry::LineageEntry(const Task &task, GcsStatus status)
: status_(status), task_(task) {
ComputeParentTaskIds();
}
GcsStatus LineageEntry::GetStatus() const { return status_; }
bool LineageEntry::SetStatus(GcsStatus new_status) {
if (status_ < new_status) {
status_ = new_status;
return true;
} else {
return false;
}
}
void LineageEntry::ResetStatus(GcsStatus new_status) {
RAY_CHECK(new_status < status_);
status_ = new_status;
}
void LineageEntry::MarkExplicitlyForwarded(const ClientID &node_id) {
forwarded_to_.insert(node_id);
}
bool LineageEntry::WasExplicitlyForwarded(const ClientID &node_id) const {
return forwarded_to_.find(node_id) != forwarded_to_.end();
}
const TaskID LineageEntry::GetEntryId() const {
return task_.GetTaskSpecification().TaskId();
}
const std::unordered_set<TaskID> &LineageEntry::GetParentTaskIds() const {
return parent_task_ids_;
}
void LineageEntry::ComputeParentTaskIds() {
parent_task_ids_.clear();
// A task's parents are the tasks that created its arguments.
for (const auto &dependency : task_.GetDependencies()) {
parent_task_ids_.insert(ObjectID::FromBinary(dependency.object_id()).TaskId());
}
}
const Task &LineageEntry::TaskData() const { return task_; }
Task &LineageEntry::TaskDataMutable() { return task_; }
void LineageEntry::UpdateTaskData(const Task &task) {
task_.CopyTaskExecutionSpec(task);
ComputeParentTaskIds();
}
Lineage::Lineage() {}
boost::optional<const LineageEntry &> Lineage::GetEntry(const TaskID &task_id) const {
auto entry = entries_.find(task_id);
if (entry != entries_.end()) {
return entry->second;
} else {
return boost::optional<const LineageEntry &>();
}
}
boost::optional<LineageEntry &> Lineage::GetEntryMutable(const TaskID &task_id) {
auto entry = entries_.find(task_id);
if (entry != entries_.end()) {
return entry->second;
} else {
return boost::optional<LineageEntry &>();
}
}
void Lineage::RemoveChild(const TaskID &parent_id, const TaskID &child_id) {
auto parent_it = children_.find(parent_id);
RAY_CHECK(parent_it->second.erase(child_id) == 1);
if (parent_it->second.empty()) {
children_.erase(parent_it);
}
}
void Lineage::AddChild(const TaskID &parent_id, const TaskID &child_id) {
auto inserted = children_[parent_id].insert(child_id);
RAY_CHECK(inserted.second);
}
bool Lineage::SetEntry(const Task &task, GcsStatus status) {
// Get the status of the current entry at the key.
auto task_id = task.GetTaskSpecification().TaskId();
auto it = entries_.find(task_id);
bool updated = false;
if (it != entries_.end()) {
if (it->second.SetStatus(status)) {
// We assume here that the new `task` has the same fields as the task
// already in the lineage cache. If this is not true, then it is
// necessary to update the task data of the existing lineage cache entry
// with LineageEntry::UpdateTaskData.
updated = true;
}
} else {
LineageEntry new_entry(task, status);
it = entries_.emplace(std::make_pair(task_id, std::move(new_entry))).first;
updated = true;
// New task data was added to the local cache, so record which tasks it
// depends on. Add all new tasks that it depends on.
for (const auto &parent_id : it->second.GetParentTaskIds()) {
AddChild(parent_id, task_id);
}
}
return updated;
}
boost::optional<LineageEntry> Lineage::PopEntry(const TaskID &task_id) {
auto entry = entries_.find(task_id);
if (entry != entries_.end()) {
LineageEntry entry = std::move(entries_.at(task_id));
// Remove the task's dependencies.
for (const auto &parent_id : entry.GetParentTaskIds()) {
RemoveChild(parent_id, task_id);
}
entries_.erase(task_id);
return entry;
} else {
return boost::optional<LineageEntry>();
}
}
const std::unordered_map<const TaskID, LineageEntry> &Lineage::GetEntries() const {
return entries_;
}
const std::unordered_set<TaskID> &Lineage::GetChildren(const TaskID &task_id) const {
static const std::unordered_set<TaskID> empty_children;
const auto it = children_.find(task_id);
if (it != children_.end()) {
return it->second;
} else {
return empty_children;
}
}
LineageCache::LineageCache(const ClientID &self_node_id,
std::shared_ptr<gcs::GcsClient> gcs_client,
uint64_t max_lineage_size)
: self_node_id_(self_node_id), gcs_client_(gcs_client) {}
/// A helper function to add some uncommitted lineage to the local cache.
void LineageCache::AddUncommittedLineage(const TaskID &task_id,
const Lineage &uncommitted_lineage) {
// TODO(edoakes): remove this method, it's currently only called in unit tests.
RAY_LOG(DEBUG) << "Adding uncommitted task " << task_id << " on " << self_node_id_;
// If the entry is not found in the lineage to merge, then we stop since
// there is nothing to copy into the merged lineage.
auto entry = uncommitted_lineage.GetEntry(task_id);
if (!entry) {
return;
}
RAY_CHECK(entry->GetStatus() == GcsStatus::UNCOMMITTED);
// Insert a copy of the entry into our cache.
const auto &parent_ids = entry->GetParentTaskIds();
// If the insert is successful, then continue the DFS. The insert will fail
// if the new entry has an equal or lower GCS status than the current entry
// in our cache. This also prevents us from traversing the same node twice.
if (lineage_.SetEntry(entry->TaskData(), entry->GetStatus())) {
RAY_CHECK(SubscribeTask(task_id));
for (const auto &parent_id : parent_ids) {
AddUncommittedLineage(parent_id, uncommitted_lineage);
}
}
}
bool LineageCache::CommitTask(const Task &task) {
// TODO(edoakes): remove this method, it's currently only called in unit tests.
const TaskID task_id = task.GetTaskSpecification().TaskId();
RAY_LOG(DEBUG) << "Committing task " << task_id << " on " << self_node_id_;
if (lineage_.SetEntry(task, GcsStatus::UNCOMMITTED) ||
lineage_.GetEntry(task_id)->GetStatus() == GcsStatus::UNCOMMITTED) {
// Attempt to flush the task if the task is uncommitted.
FlushTask(task_id);
return true;
} else {
// The task was already committing (COMMITTING).
return false;
}
}
void LineageCache::FlushAllUncommittedTasks() {
size_t num_flushed = 0;
for (const auto &entry : lineage_.GetEntries()) {
// Flush all tasks that have not yet committed.
if (entry.second.GetStatus() == GcsStatus::UNCOMMITTED) {
RAY_CHECK(UnsubscribeTask(entry.first));
FlushTask(entry.first);
num_flushed++;
}
}
RAY_LOG(DEBUG) << "Flushed " << num_flushed << " uncommitted tasks";
}
void LineageCache::MarkTaskAsForwarded(const TaskID &task_id, const ClientID &node_id) {
RAY_CHECK(!node_id.IsNil());
auto entry = lineage_.GetEntryMutable(task_id);
if (entry) {
entry->MarkExplicitlyForwarded(node_id);
}
}
/// A helper function to get the uncommitted lineage of a task.
void GetUncommittedLineageHelper(const TaskID &task_id, const Lineage &lineage_from,
Lineage &lineage_to, const ClientID &node_id) {
// If the entry is not found in the lineage to merge, then we stop since
// there is nothing to copy into the merged lineage.
auto entry = lineage_from.GetEntry(task_id);
if (!entry) {
return;
}
// If this task has already been forwarded to this node, then we can stop.
if (entry->WasExplicitlyForwarded(node_id)) {
return;
}
// Insert a copy of the entry into lineage_to. If the insert is successful,
// then continue the DFS. The insert will fail if the new entry has an equal
// or lower GCS status than the current entry in lineage_to. This also
// prevents us from traversing the same node twice.
if (lineage_to.SetEntry(entry->TaskData(), entry->GetStatus())) {
for (const auto &parent_id : entry->GetParentTaskIds()) {
GetUncommittedLineageHelper(parent_id, lineage_from, lineage_to, node_id);
}
}
}
Lineage LineageCache::GetUncommittedLineage(const TaskID &task_id,
const ClientID &node_id) const {
Lineage uncommitted_lineage;
// Add all uncommitted ancestors from the lineage cache to the uncommitted
// lineage of the requested task.
GetUncommittedLineageHelper(task_id, lineage_, uncommitted_lineage, node_id);
// The lineage always includes the requested task id, so add the task if it
// wasn't already added. The requested task may not have been added if it was
// already explicitly forwarded to this node before.
if (uncommitted_lineage.GetEntries().empty()) {
auto entry = lineage_.GetEntry(task_id);
if (entry) {
RAY_CHECK(uncommitted_lineage.SetEntry(entry->TaskData(), entry->GetStatus()));
}
}
return uncommitted_lineage;
}
void LineageCache::FlushTask(const TaskID &task_id) {
auto entry = lineage_.GetEntryMutable(task_id);
RAY_CHECK(entry);
RAY_CHECK(entry->GetStatus() < GcsStatus::COMMITTING);
auto task_callback = [this, task_id](Status status) {
RAY_CHECK(status.ok());
HandleEntryCommitted(task_id);
};
auto task = lineage_.GetEntry(task_id);
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->CopyFrom(
task->TaskData().GetTaskSpecification().GetMessage());
task_data->mutable_task()->mutable_task_execution_spec()->CopyFrom(
task->TaskData().GetTaskExecutionSpec().GetMessage());
RAY_CHECK_OK(gcs_client_->Tasks().AsyncAdd(task_data, task_callback));
// We successfully wrote the task, so mark it as committing.
// TODO(swang): Use a batched interface and write with all object entries.
RAY_CHECK(entry->SetStatus(GcsStatus::COMMITTING));
}
bool LineageCache::SubscribeTask(const TaskID &task_id) {
auto inserted = subscribed_tasks_.insert(task_id);
bool unsubscribed = inserted.second;
if (unsubscribed) {
auto subscribe = [this](const TaskID &task_id, const TaskTableData) {
HandleEntryCommitted(task_id);
};
// Subscribe to the task.
RAY_CHECK_OK(gcs_client_->Tasks().AsyncSubscribe(task_id, subscribe,
/*done*/ nullptr));
}
// Return whether we were previously unsubscribed to this task and are now
// subscribed.
return unsubscribed;
}
bool LineageCache::UnsubscribeTask(const TaskID &task_id) {
auto it = subscribed_tasks_.find(task_id);
bool subscribed = (it != subscribed_tasks_.end());
if (subscribed) {
// Cancel subscribe to the task.
RAY_CHECK_OK(gcs_client_->Tasks().AsyncUnsubscribe(task_id));
subscribed_tasks_.erase(it);
}
// Return whether we were previously subscribed to this task and are now
// unsubscribed.
return subscribed;
}
void LineageCache::EvictTask(const TaskID &task_id) {
// If the entry has already been evicted, exit.
auto entry = lineage_.GetEntry(task_id);
if (!entry) {
return;
}
// If the entry has not yet been committed, exit.
if (entry->GetStatus() != GcsStatus::COMMITTED) {
return;
}
// Entries cannot be safely evicted until their parents are all evicted.
for (const auto &parent_id : entry->GetParentTaskIds()) {
if (ContainsTask(parent_id)) {
return;
}
}
// Evict the task.
RAY_LOG(DEBUG) << "Evicting task " << task_id << " on " << self_node_id_;
lineage_.PopEntry(task_id);
// Try to evict the children of the evict task. These are the tasks that have
// a dependency on the evicted task.
const auto children = lineage_.GetChildren(task_id);
for (const auto &child_id : children) {
EvictTask(child_id);
}
}
void LineageCache::HandleEntryCommitted(const TaskID &task_id) {
RAY_LOG(DEBUG) << "Task committed: " << task_id;
auto entry = lineage_.GetEntryMutable(task_id);
if (!entry) {
// The task has already been evicted due to a previous commit notification.
return;
}
// Record the commit acknowledgement and attempt to evict the task.
entry->SetStatus(GcsStatus::COMMITTED);
EvictTask(task_id);
// We got the notification about the task's commit, so no longer need any
// more notifications.
UnsubscribeTask(task_id);
}
const Task &LineageCache::GetTaskOrDie(const TaskID &task_id) const {
const auto &entries = lineage_.GetEntries();
auto it = entries.find(task_id);
RAY_CHECK(it != entries.end());
return it->second.TaskData();
}
bool LineageCache::ContainsTask(const TaskID &task_id) const {
const auto &entries = lineage_.GetEntries();
auto it = entries.find(task_id);
return it != entries.end();
}
const Lineage &LineageCache::GetLineage() const { return lineage_; }
std::string LineageCache::DebugString() const {
std::stringstream result;
result << "LineageCache:";
result << "\n- child map size: " << lineage_.GetChildrenSize();
result << "\n- num subscribed tasks: " << subscribed_tasks_.size();
result << "\n- lineage size: " << lineage_.GetEntries().size();
return result.str();
}
void LineageCache::RecordMetrics() const {
stats::LineageCacheStats().Record(lineage_.GetChildrenSize(),
{{stats::ValueTypeKey, "num_children"}});
stats::LineageCacheStats().Record(subscribed_tasks_.size(),
{{stats::ValueTypeKey, "num_subscribed_tasks"}});
stats::LineageCacheStats().Record(lineage_.GetEntries().size(),
{{stats::ValueTypeKey, "num_lineages"}});
}
} // namespace raylet
} // namespace ray
src/ray/raylet/lineage_cache.h
-327
View File
@@ -1,327 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#pragma once
#include <gtest/gtest_prod.h>
#include <boost/optional.hpp>
#include "ray/common/id.h"
#include "ray/common/status.h"
#include "ray/common/task/task.h"
#include "ray/gcs/redis_gcs_client.h"
namespace ray {
namespace raylet {
using rpc::TaskTableData;
/// The status of a lineage cache entry according to its status in the GCS.
/// Tasks can only transition to a higher GcsStatus (e.g., an UNCOMMITTED state
/// can become COMMITTING but not vice versa). If a task is evicted from the
/// local cache, it implicitly goes back to state `NONE`, after which it may be
/// added to the local cache again (e.g., if it is forwarded to us again).
enum class GcsStatus {
/// The task is not in the lineage cache.
NONE = 0,
/// The task is uncommitted. Unless there is a failure, we will expect a
/// different node to commit this task.
UNCOMMITTED,
/// We flushed this task and are waiting for the commit acknowledgement.
COMMITTING,
// Tasks for which we received a commit acknowledgement, but which we cannot
// evict yet (due to an ancestor that has not been evicted). This is to allow
// a performance optimization that avoids unnecessary subscribes when we
// receive tasks that were already COMMITTED at the sender.
COMMITTED,
};
/// \class LineageEntry
///
/// A task entry in the data lineage. Each entry's parents are the tasks that
/// created the entry's arguments.
class LineageEntry {
public:
/// Create an entry for a task.
///
/// \param task The task data to eventually be written back to the GCS.
/// \param status The status of this entry, according to its write status in
/// the GCS.
LineageEntry(const Task &task, GcsStatus status);
/// Get this entry's GCS status.
///
/// \return The entry's status in the GCS.
GcsStatus GetStatus() const;
/// Set this entry's GCS status. The status is only set if the new status
/// is strictly greater than the entry's previous status, according to the
/// GcsStatus enum.
///
/// \param new_status Set the entry's status to this value if it is greater
/// than the current status.
/// \return Whether the entry was set to the new status.
bool SetStatus(GcsStatus new_status);
/// Reset this entry's GCS status to a lower status. The new status must
/// be lower than the current status.
///
/// \param new_status This must be lower than the current status.
void ResetStatus(GcsStatus new_status);
/// Mark this entry as having been explicitly forwarded to a remote node manager.
///
/// \param node_id The ID of the remote node manager.
void MarkExplicitlyForwarded(const ClientID &node_id);
/// Gets whether this entry was explicitly forwarded to a remote node.
///
/// \param node_id The ID of the remote node manager.
/// \return Whether this entry was explicitly forwarded to the remote node.
bool WasExplicitlyForwarded(const ClientID &node_id) const;
/// Get this entry's ID.
///
/// \return The entry's ID.
const TaskID GetEntryId() const;
/// Get the IDs of this entry's parent tasks. These are the IDs of the tasks
/// that created its arguments.
///
/// \return The IDs of the parent entries.
const std::unordered_set<TaskID> &GetParentTaskIds() const;
/// Get the task data.
///
/// \return The task data.
const Task &TaskData() const;
/// Get a mutable version of the task data.
///
/// \return The task data.
/// TODO(swang): This is pretty ugly.
Task &TaskDataMutable();
/// Update the task data with a new task.
///
/// \return Void.
void UpdateTaskData(const Task &task);
private:
/// Compute cached parent task IDs. This task is dependent on values returned
/// by these tasks.
void ComputeParentTaskIds();
/// The current state of this entry according to its status in the GCS.
GcsStatus status_;
/// The task data to be written to the GCS. This is nullptr if the entry is
/// an object.
// const Task task_;
Task task_;
/// A cached copy of the parent task IDs. This task is dependent on values
/// returned by these tasks.
std::unordered_set<TaskID> parent_task_ids_;
/// IDs of node managers that this task has been explicitly forwarded to.
std::unordered_set<ClientID> forwarded_to_;
};
/// \class Lineage
///
/// A lineage DAG, according to the data dependency graph. Each node is a task,
/// with an outgoing edge to each of its parent tasks. For a given task, the
/// parents are the tasks that created its arguments. Each entry also records
/// the current status in the GCS for that task or object.
class Lineage {
public:
/// Construct an empty Lineage.
Lineage();
/// Get an entry from the lineage.
///
/// \param entry_id The ID of the entry to get.
/// \return An optional reference to the entry. If this is empty, then the
/// entry ID is not in the lineage.
boost::optional<const LineageEntry &> GetEntry(const TaskID &entry_id) const;
boost::optional<LineageEntry &> GetEntryMutable(const TaskID &task_id);
/// Set an entry in the lineage. If an entry with this ID already exists,
/// then the entry is overwritten if and only if the new entry has a higher
/// GCS status than the current. The current entry's object or task data will
/// also be overwritten.
///
/// \param task The task data to set, if status is greater than the current entry.
/// \param status The GCS status.
/// \return Whether the entry was set.
bool SetEntry(const Task &task, GcsStatus status);
/// Delete and return an entry from the lineage.
///
/// \param entry_id The ID of the entry to pop.
/// \return An optional reference to the popped entry. If this is empty, then
/// the entry ID is not in the lineage.
boost::optional<LineageEntry> PopEntry(const TaskID &entry_id);
/// Get all entries in the lineage.
///
/// \return A const reference to the lineage entries.
const std::unordered_map<const TaskID, LineageEntry> &GetEntries() const;
/// Return the IDs of tasks in the lineage that are dependent on the given
/// task.
///
/// \param The ID of the task whose children to get.
/// \return The list of IDs for tasks that are in the lineage and dependent
/// on the given task.
const std::unordered_set<TaskID> &GetChildren(const TaskID &task_id) const;
/// Return the size of the children_ map. This is used for debugging purposes
/// only.
size_t GetChildrenSize() const { return children_.size(); }
private:
/// The lineage entries.
std::unordered_map<const TaskID, LineageEntry> entries_;
/// A mapping from each task in the lineage to its children.
std::unordered_map<TaskID, std::unordered_set<TaskID>> children_;
/// Record the fact that the child task depends on the parent task.
void AddChild(const TaskID &parent_id, const TaskID &child_id);
/// Erase the fact that the child task depends on the parent task.
void RemoveChild(const TaskID &parent_id, const TaskID &child_id);
};
/// \class LineageCache
///
/// A cache of the task table. This consists of all tasks that this node owns,
/// as well as their lineage, that have not yet been added durably
/// ("committed") to the GCS.
///
/// The current policy is to flush each task as soon as it enters the
/// UNCOMMITTED_READY state. For safety, we only evict tasks if they have been
/// committed and if their parents have been all evicted. Thus, the invariant
/// is that if g depends on f, and g has been evicted, then f must have been
/// committed.
class LineageCache {
public:
/// Create a lineage cache for the given task storage system.
/// TODO(swang): Pass in the policy (interface?).
LineageCache(const ClientID &self_node_id, std::shared_ptr<gcs::GcsClient> gcs_client,
uint64_t max_lineage_size);
/// Asynchronously commit a task to the GCS.
///
/// \param task The task to commit. It will be moved to the COMMITTING state.
/// \return Whether the task was successfully committed. This can fail if the
/// task was already in the COMMITTING state.
bool CommitTask(const Task &task);
/// Flush all tasks in the local cache that are not already being
/// committed. This is equivalent to all tasks in the UNCOMMITTED
/// state.
///
/// \return Void.
void FlushAllUncommittedTasks();
/// Add a task and its (estimated) uncommitted lineage to the local cache. We
/// will subscribe to commit notifications for all uncommitted tasks to
/// determine when it is safe to evict the lineage from the local cache.
///
/// \param task_id The ID of the uncommitted task to add.
/// \param uncommitted_lineage The task's uncommitted lineage. These are the
/// tasks that the given task is data-dependent on, but that have not
/// been committed to the GCS. This must contain the given task ID.
/// \return Void.
void AddUncommittedLineage(const TaskID &task_id, const Lineage &uncommitted_lineage);
/// Mark a task as having been explicitly forwarded to a node.
/// The lineage of the task is implicitly assumed to have also been forwarded.
///
/// \param task_id The ID of the task to get the uncommitted lineage for.
/// \param node_id The ID of the node to get the uncommitted lineage for.
void MarkTaskAsForwarded(const TaskID &task_id, const ClientID &node_id);
/// Get the uncommitted lineage of a task that hasn't been forwarded to a node yet.
/// The uncommitted lineage consists of all tasks in the given task's lineage
/// that have not been committed in the GCS, as far as we know.
///
/// \param task_id The ID of the task to get the uncommitted lineage for. If
/// the task is not found, then the returned lineage will be empty.
/// \param node_id The ID of the receiving node.
/// \return The uncommitted, unforwarded lineage of the task. The returned lineage
/// includes the entry for the requested entry_id.
Lineage GetUncommittedLineage(const TaskID &task_id, const ClientID &node_id) const;
/// Handle the commit of a task entry in the GCS. This attempts to evict the
/// task if possible.
///
/// \param task_id The ID of the task entry that was committed.
void HandleEntryCommitted(const TaskID &task_id);
/// Get a task. The task must be in the lineage cache.
///
/// \param task_id The ID of the task to get.
/// \return A const reference to the task data.
const Task &GetTaskOrDie(const TaskID &task_id) const;
/// Get whether the lineage cache contains the task.
///
/// \param task_id The ID of the task to get.
/// \return Whether the task is in the lineage cache.
bool ContainsTask(const TaskID &task_id) const;
/// Get all lineage in the lineage cache.
///
/// \return A const reference to the lineage.
const Lineage &GetLineage() const;
/// Returns debug string for class.
///
/// \return string.
std::string DebugString() const;
/// Record metrics.
void RecordMetrics() const;
private:
FRIEND_TEST(LineageCacheTest, BarReturnsZeroOnNull);
/// Flush a task that is in UNCOMMITTED_READY state.
void FlushTask(const TaskID &task_id);
/// Evict a single task. This should only be called if we are sure that the
/// task has been committed. The task will only be evicted if all of its
/// parents have also been evicted. If successful, then we will also attempt
/// to evict the task's children.
void EvictTask(const TaskID &task_id);
/// Subscribe to notifications for a task. Returns whether the operation
/// was successful (whether we were not already subscribed).
bool SubscribeTask(const TaskID &task_id);
/// Unsubscribe from notifications for a task. Returns whether the operation
/// was successful (whether we were subscribed).
bool UnsubscribeTask(const TaskID &task_id);
/// ID of this node.
ClientID self_node_id_;
/// A client connection to the GCS.
std::shared_ptr<gcs::GcsClient> gcs_client_;
/// All tasks and objects that we are responsible for writing back to the
/// GCS, and the tasks and objects in their lineage.
Lineage lineage_;
/// The tasks that we've subscribed to.
/// We will receive a notification for these tasks on commit.
std::unordered_set<TaskID> subscribed_tasks_;
};
} // namespace raylet
} // namespace ray
src/ray/raylet/lineage_cache_test.cc
-670
View File
@@ -1,670 +0,0 @@
// Copyright 2017 The Ray Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "ray/raylet/lineage_cache.h"
#include <list>
#include <memory>
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "ray/common/task/task.h"
#include "ray/common/task/task_execution_spec.h"
#include "ray/common/task/task_spec.h"
#include "ray/common/task/task_util.h"
#include "ray/common/test_util.h"
#include "ray/gcs/callback.h"
#include "ray/gcs/redis_accessor.h"
#include "ray/gcs/redis_gcs_client.h"
#include "ray/raylet/format/node_manager_generated.h"
namespace ray {
namespace raylet {
const static JobID kDefaultJobId = JobID::FromInt(1);
const static TaskID kDefaultDriverTaskId = TaskID::ForDriverTask(kDefaultJobId);
class MockGcsClient;
class MockTaskInfoAccessor : public gcs::RedisTaskInfoAccessor {
public:
MockTaskInfoAccessor(gcs::RedisGcsClient *gcs_client)
: RedisTaskInfoAccessor(gcs_client) {}
virtual ~MockTaskInfoAccessor() {}
void RegisterSubscribeCallback(
const gcs::SubscribeCallback<TaskID, rpc::TaskTableData> &notification_callback) {
notification_callback_ = notification_callback;
}
Status AsyncAdd(const std::shared_ptr<TaskTableData> &task_data,
const gcs::StatusCallback &done) {
TaskID task_id = TaskID::FromBinary(task_data->task().task_spec().task_id());
task_table_[task_id] = task_data;
auto callback = done;
// If we requested notifications for this task ID, send the notification as
// part of the callback.
if (subscribed_tasks_.count(task_id) == 1) {
callback = [this, done, task_id, task_data](Status status) {
done(status);
// If we're subscribed to the task to be added, also send a
// subscription notification.
notification_callback_(task_id, *task_data);
};
}
callbacks_.push_back({callback, task_id});
num_task_adds_++;
return ray::Status::OK();
}
Status RemoteAdd(std::shared_ptr<TaskTableData> task_data) {
TaskID task_id = TaskID::FromBinary(task_data->task().task_spec().task_id());
task_table_[task_id] = task_data;
// Send a notification after the add if the lineage cache requested
// notifications for this key.
bool send_notification = (subscribed_tasks_.count(task_id) == 1);
auto callback = [this, send_notification, task_id, task_data](Status status) {
if (send_notification) {
notification_callback_(task_id, *task_data);
}
};
return AsyncAdd(task_data, callback);
}
Status AsyncSubscribe(
const TaskID &task_id,
const gcs::SubscribeCallback<TaskID, rpc::TaskTableData> &notification_callback,
const gcs::StatusCallback &done) {
subscribed_tasks_.insert(task_id);
if (task_table_.count(task_id) == 1) {
notification_callbacks_.push_back({notification_callback_, task_id});
}
num_requested_notifications_ += 1;
return ray::Status::OK();
}
Status AsyncUnsubscribe(const TaskID &task_id) {
subscribed_tasks_.erase(task_id);
return ray::Status::OK();
}
void Flush() {
auto callbacks = std::move(callbacks_);
callbacks_.clear();
for (const auto &callback : callbacks) {
callback.first(Status::OK());
}
for (const auto &callback : notification_callbacks_) {
callback.first(callback.second, *task_table_[callback.second]);
}
}
const std::unordered_map<TaskID, std::shared_ptr<TaskTableData>> &TaskTable() const {
return task_table_;
}
const std::unordered_set<TaskID> &SubscribedTasks() const { return subscribed_tasks_; }
const int NumRequestedNotifications() const { return num_requested_notifications_; }
const int NumTaskAdds() const { return num_task_adds_; }
private:
std::unordered_map<TaskID, std::shared_ptr<TaskTableData>> task_table_;
std::vector<std::pair<gcs::StatusCallback, TaskID>> callbacks_;
typedef gcs::SubscribeCallback<TaskID, rpc::TaskTableData> TaskSubscribeCallback;
TaskSubscribeCallback notification_callback_;
std::vector<std::pair<TaskSubscribeCallback, TaskID>> notification_callbacks_;
std::unordered_set<TaskID> subscribed_tasks_;
int num_requested_notifications_ = 0;
int num_task_adds_ = 0;
};
class MockNodeInfoAccessor : public gcs::RedisNodeInfoAccessor {
public:
MockNodeInfoAccessor(gcs::RedisGcsClient *gcs_client, const ClientID &node_id)
: RedisNodeInfoAccessor(gcs_client), node_id_(node_id) {}
const ClientID &GetSelfId() const override { return node_id_; }
private:
ClientID node_id_;
};
class MockGcsClient : public gcs::RedisGcsClient {
public:
MockGcsClient(const gcs::GcsClientOptions &options, const ClientID &node_id)
: RedisGcsClient(options) {
task_table_fake_.reset(new gcs::raylet::TaskTable({nullptr}, this));
task_accessor_.reset(new MockTaskInfoAccessor(this));
node_accessor_.reset(new MockNodeInfoAccessor(this, node_id));
}
gcs::raylet::TaskTable &raylet_task_table() override { return *task_table_fake_; }
MockTaskInfoAccessor &MockTasks() {
return *dynamic_cast<MockTaskInfoAccessor *>(task_accessor_.get());
}
private:
std::unique_ptr<gcs::raylet::TaskTable> task_table_fake_;
};
class LineageCacheTest : public ::testing::Test {
public:
LineageCacheTest() : max_lineage_size_(10), num_notifications_(0) {
gcs::GcsClientOptions options("10.10.10.10", 12100, "");
mock_gcs_ = std::make_shared<MockGcsClient>(options, node_id_);
lineage_cache_.reset(new LineageCache(node_id_, mock_gcs_, max_lineage_size_));
mock_gcs_->MockTasks().RegisterSubscribeCallback(
[this](const TaskID &task_id, const TaskTableData &data) {
lineage_cache_->HandleEntryCommitted(task_id);
num_notifications_++;
});
}
protected:
uint64_t max_lineage_size_;
uint64_t num_notifications_;
ClientID node_id_{ClientID::FromRandom()};
std::shared_ptr<MockGcsClient> mock_gcs_;
std::unique_ptr<LineageCache> lineage_cache_;
};
static inline Task ExampleTask(const std::vector<ObjectID> &arguments,
uint64_t num_returns) {
TaskSpecBuilder builder;
rpc::Address address;
builder.SetCommonTaskSpec(RandomTaskId(), Language::PYTHON,
ray::FunctionDescriptorBuilder::BuildPython("", "", "", ""),
JobID::Nil(), RandomTaskId(), 0, RandomTaskId(), address,
num_returns, {}, {});
for (const auto &arg : arguments) {
builder.AddArg(TaskArgByReference(arg, rpc::Address()));
}
rpc::TaskExecutionSpec execution_spec_message;
execution_spec_message.set_num_forwards(1);
return Task(builder.Build(), TaskExecutionSpecification(execution_spec_message));
}
/// Helper method to create a Lineage object with a single task.
Lineage CreateSingletonLineage(const Task &task) {
Lineage singleton_lineage;
singleton_lineage.SetEntry(task, GcsStatus::UNCOMMITTED);
return singleton_lineage;
}
std::vector<ObjectID> InsertTaskChain(LineageCache &lineage_cache,
std::vector<Task> &inserted_tasks, int chain_size,
const std::vector<ObjectID> &initial_arguments,
int64_t num_returns) {
std::vector<ObjectID> arguments = initial_arguments;
for (int i = 0; i < chain_size; i++) {
auto task = ExampleTask(arguments, num_returns);
Lineage lineage = CreateSingletonLineage(task);
lineage_cache.AddUncommittedLineage(task.GetTaskSpecification().TaskId(), lineage);
inserted_tasks.push_back(task);
arguments.clear();
for (size_t j = 0; j < task.GetTaskSpecification().NumReturns(); j++) {
arguments.push_back(task.GetTaskSpecification().ReturnId(j));
}
}
return arguments;
}
TEST_F(LineageCacheTest, TestGetUncommittedLineage) {
// Insert two independent chains of tasks.
std::vector<Task> tasks1;
auto return_values1 =
InsertTaskChain(*lineage_cache_, tasks1, 3, std::vector<ObjectID>(), 1);
std::vector<TaskID> task_ids1;
for (const auto &task : tasks1) {
task_ids1.push_back(task.GetTaskSpecification().TaskId());
}
std::vector<Task> tasks2;
auto return_values2 =
InsertTaskChain(*lineage_cache_, tasks2, 2, std::vector<ObjectID>(), 2);
std::vector<TaskID> task_ids2;
for (const auto &task : tasks2) {
task_ids2.push_back(task.GetTaskSpecification().TaskId());
}
// Get the uncommitted lineage for the last task (the leaf) of one of the chains.
auto uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(task_ids1.back(), ClientID::Nil());
// Check that the uncommitted lineage is exactly equal to the first chain of tasks.
ASSERT_EQ(task_ids1.size(), uncommitted_lineage.GetEntries().size());
for (auto &task_id : task_ids1) {
ASSERT_TRUE(uncommitted_lineage.GetEntry(task_id));
}
// Insert one task that is dependent on the previous chains of tasks.
std::vector<Task> combined_tasks = tasks1;
combined_tasks.insert(combined_tasks.end(), tasks2.begin(), tasks2.end());
std::vector<ObjectID> combined_arguments = return_values1;
combined_arguments.insert(combined_arguments.end(), return_values2.begin(),
return_values2.end());
InsertTaskChain(*lineage_cache_, combined_tasks, 1, combined_arguments, 1);
std::vector<TaskID> combined_task_ids;
for (const auto &task : combined_tasks) {
combined_task_ids.push_back(task.GetTaskSpecification().TaskId());
}
// Get the uncommitted lineage for the inserted task.
uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(combined_task_ids.back(), ClientID::Nil());
// Check that the uncommitted lineage is exactly equal to the entire set of
// tasks inserted so far.
ASSERT_EQ(combined_task_ids.size(), uncommitted_lineage.GetEntries().size());
for (auto &task_id : combined_task_ids) {
ASSERT_TRUE(uncommitted_lineage.GetEntry(task_id));
}
}
TEST_F(LineageCacheTest, TestDuplicateUncommittedLineage) {
// Insert a chain of tasks.
std::vector<Task> tasks;
auto return_values =
InsertTaskChain(*lineage_cache_, tasks, 3, std::vector<ObjectID>(), 1);
std::vector<TaskID> task_ids;
for (const auto &task : tasks) {
task_ids.push_back(task.GetTaskSpecification().TaskId());
}
// Check that we subscribed to each of the uncommitted tasks.
ASSERT_EQ(mock_gcs_->MockTasks().NumRequestedNotifications(), task_ids.size());
// Check that if we add the same tasks as UNCOMMITTED again, we do not issue
// duplicate subscribe requests.
Lineage duplicate_lineage;
for (const auto &task : tasks) {
duplicate_lineage.SetEntry(task, GcsStatus::UNCOMMITTED);
}
lineage_cache_->AddUncommittedLineage(task_ids.back(), duplicate_lineage);
ASSERT_EQ(mock_gcs_->MockTasks().NumRequestedNotifications(), task_ids.size());
// Check that if we commit one of the tasks, we still do not issue any
// duplicate subscribe requests.
lineage_cache_->CommitTask(tasks.front());
lineage_cache_->AddUncommittedLineage(task_ids.back(), duplicate_lineage);
ASSERT_EQ(mock_gcs_->MockTasks().NumRequestedNotifications(), task_ids.size());
}
TEST_F(LineageCacheTest, TestMarkTaskAsForwarded) {
// Insert chain of tasks.
std::vector<Task> tasks;
auto return_values =
InsertTaskChain(*lineage_cache_, tasks, 4, std::vector<ObjectID>(), 1);
std::vector<TaskID> task_ids;
for (const auto &task : tasks) {
task_ids.push_back(task.GetTaskSpecification().TaskId());
}
auto node_id = ClientID::FromRandom();
auto node_id2 = ClientID::FromRandom();
auto forwarded_task_id = task_ids[task_ids.size() - 2];
auto remaining_task_id = task_ids[task_ids.size() - 1];
lineage_cache_->MarkTaskAsForwarded(forwarded_task_id, node_id);
auto uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(remaining_task_id, node_id);
auto uncommitted_lineage_all =
lineage_cache_->GetUncommittedLineage(remaining_task_id, node_id2);
ASSERT_EQ(1, uncommitted_lineage.GetEntries().size());
ASSERT_EQ(4, uncommitted_lineage_all.GetEntries().size());
ASSERT_TRUE(uncommitted_lineage.GetEntry(remaining_task_id));
// Check that lineage of requested task includes itself, regardless of whether
// it has been forwarded before.
auto uncommitted_lineage_forwarded =
lineage_cache_->GetUncommittedLineage(forwarded_task_id, node_id);
ASSERT_EQ(1, uncommitted_lineage_forwarded.GetEntries().size());
}
TEST_F(LineageCacheTest, TestWritebackReady) {
// Insert a chain of dependent tasks.
size_t num_tasks_flushed = 0;
std::vector<Task> tasks;
InsertTaskChain(*lineage_cache_, tasks, 3, std::vector<ObjectID>(), 1);
// Check that when no tasks have been marked as ready, we do not flush any
// entries.
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
// Check that after marking the first task as ready, we flush only that task.
ASSERT_TRUE(lineage_cache_->CommitTask(tasks.front()));
num_tasks_flushed++;
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
}
TEST_F(LineageCacheTest, TestWritebackOrder) {
// Insert a chain of dependent tasks.
std::vector<Task> tasks;
InsertTaskChain(*lineage_cache_, tasks, 3, std::vector<ObjectID>(), 1);
size_t num_tasks_flushed = tasks.size();
// Mark all tasks as ready. All tasks should be flushed.
for (const auto &task : tasks) {
ASSERT_TRUE(lineage_cache_->CommitTask(task));
}
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
}
TEST_F(LineageCacheTest, TestEvictChain) {
// Create a chain of 3 tasks.
size_t num_tasks_flushed = 0;
std::vector<Task> tasks;
std::vector<ObjectID> arguments;
for (int i = 0; i < 3; i++) {
auto task = ExampleTask(arguments, 1);
tasks.push_back(task);
arguments = {task.GetTaskSpecification().ReturnId(0)};
}
Lineage uncommitted_lineage;
for (const auto &task : tasks) {
uncommitted_lineage.SetEntry(task, GcsStatus::UNCOMMITTED);
}
// Mark the last task as ready to flush.
lineage_cache_->AddUncommittedLineage(tasks.back().GetTaskSpecification().TaskId(),
uncommitted_lineage);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), tasks.size());
ASSERT_TRUE(lineage_cache_->CommitTask(tasks.back()));
num_tasks_flushed++;
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
// Flush acknowledgements. The lineage cache should receive the commit for
// the flushed task, but its lineage should not be evicted yet.
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(lineage_cache_
->GetUncommittedLineage(tasks.back().GetTaskSpecification().TaskId(),
ClientID::Nil())
.GetEntries()
.size(),
tasks.size());
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), tasks.size());
// Simulate executing the task on a remote node and adding it to the GCS.
auto task_id = tasks.at(1).GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(lineage_cache_
->GetUncommittedLineage(tasks.back().GetTaskSpecification().TaskId(),
ClientID::Nil())
.GetEntries()
.size(),
tasks.size());
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), tasks.size());
// Simulate executing the task on a remote node and adding it to the GCS.
task_id = tasks.at(0).GetTaskSpecification().TaskId();
auto task_data_2 = std::make_shared<TaskTableData>();
task_data_2->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data_2));
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), 0);
ASSERT_EQ(lineage_cache_->GetLineage().GetChildrenSize(), 0);
}
TEST_F(LineageCacheTest, TestEvictManyParents) {
// Create some independent tasks.
std::vector<Task> parent_tasks;
std::vector<ObjectID> arguments;
for (int i = 0; i < 10; i++) {
auto task = ExampleTask({}, 1);
parent_tasks.push_back(task);
arguments.push_back(task.GetTaskSpecification().ReturnId(0));
auto lineage = CreateSingletonLineage(task);
lineage_cache_->AddUncommittedLineage(task.GetTaskSpecification().TaskId(), lineage);
}
// Create a child task that is dependent on all of the previous tasks.
auto child_task = ExampleTask(arguments, 1);
auto lineage = CreateSingletonLineage(child_task);
lineage_cache_->AddUncommittedLineage(child_task.GetTaskSpecification().TaskId(),
lineage);
// Flush the child task. Make sure that it remains in the cache, since none
// of its parents have been committed yet, and that the uncommitted lineage
// still includes all of the parent tasks.
size_t total_tasks = parent_tasks.size() + 1;
lineage_cache_->CommitTask(child_task);
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), total_tasks);
ASSERT_EQ(lineage_cache_
->GetUncommittedLineage(child_task.GetTaskSpecification().TaskId(),
ClientID::Nil())
.GetEntries()
.size(),
total_tasks);
// Flush each parent task and check for eviction safety.
for (const auto &parent_task : parent_tasks) {
lineage_cache_->CommitTask(parent_task);
mock_gcs_->MockTasks().Flush();
total_tasks--;
if (total_tasks > 1) {
// Each task should be evicted as soon as its commit is acknowledged,
// since the parent tasks have no dependencies.
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), total_tasks);
ASSERT_EQ(lineage_cache_
->GetUncommittedLineage(child_task.GetTaskSpecification().TaskId(),
ClientID::Nil())
.GetEntries()
.size(),
total_tasks);
} else {
// After the last task has been committed, then the child task should
// also be evicted. The lineage cache should now be empty.
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), 0);
}
}
ASSERT_EQ(lineage_cache_->GetLineage().GetChildrenSize(), 0);
}
TEST_F(LineageCacheTest, TestEviction) {
// Insert a chain of dependent tasks.
uint64_t lineage_size = max_lineage_size_ + 1;
size_t num_tasks_flushed = 0;
std::vector<Task> tasks;
InsertTaskChain(*lineage_cache_, tasks, lineage_size, std::vector<ObjectID>(), 1);
// Check that the last task in the chain still has all tasks in its
// uncommitted lineage.
const auto last_task_id = tasks.back().GetTaskSpecification().TaskId();
auto uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(last_task_id, ClientID::Nil());
ASSERT_EQ(uncommitted_lineage.GetEntries().size(), lineage_size);
// Simulate executing the first task on a remote node and adding it to the
// GCS.
auto it = tasks.begin();
auto task_id = it->GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
it++;
// Check that the remote task is flushed.
num_tasks_flushed++;
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
// Check that the last task in the chain still has all tasks in its
// uncommitted lineage.
ASSERT_EQ(uncommitted_lineage.GetEntries().size(), lineage_size);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(),
lineage_size - num_tasks_flushed);
// Simulate executing all the rest of the tasks except the last one on a
// remote node and adding them to the GCS.
tasks.pop_back();
for (; it != tasks.end(); it++) {
auto task_id = it->GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
// Check that the remote task is flushed.
num_tasks_flushed++;
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(),
lineage_size - num_tasks_flushed);
}
// All tasks have now been flushed. Check that enough lineage has been
// evicted that the uncommitted lineage is now less than the maximum size.
uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(last_task_id, ClientID::Nil());
ASSERT_TRUE(uncommitted_lineage.GetEntries().size() < max_lineage_size_);
// The remaining task should have no uncommitted lineage.
ASSERT_EQ(uncommitted_lineage.GetEntries().size(), 1);
ASSERT_EQ(lineage_cache_->GetLineage().GetChildrenSize(), 1);
}
TEST_F(LineageCacheTest, TestOutOfOrderEviction) {
// Insert a chain of dependent tasks that is more than twice as long as the
// maximum lineage size. This will ensure that we request notifications for
// at most 2 remote tasks.
uint64_t lineage_size = (2 * max_lineage_size_) + 2;
size_t num_tasks_flushed = 0;
std::vector<Task> tasks;
InsertTaskChain(*lineage_cache_, tasks, lineage_size, std::vector<ObjectID>(), 1);
// Check that the last task in the chain still has all tasks in its
// uncommitted lineage.
const auto last_task_id = tasks.back().GetTaskSpecification().TaskId();
auto uncommitted_lineage =
lineage_cache_->GetUncommittedLineage(last_task_id, ClientID::Nil());
ASSERT_EQ(uncommitted_lineage.GetEntries().size(), lineage_size);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), lineage_size);
// Simulate executing the rest of the tasks on a remote node and receiving
// the notifications from the GCS in reverse order of execution.
auto last_task = tasks.front();
tasks.erase(tasks.begin());
for (auto it = tasks.rbegin(); it != tasks.rend(); it++) {
auto task_id = it->GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
// Check that the remote task is flushed.
num_tasks_flushed++;
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), lineage_size);
}
// Flush the last task. The lineage should not get evicted until this task's
// commit is received.
auto task_id = last_task.GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
num_tasks_flushed++;
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), 0);
ASSERT_EQ(lineage_cache_->GetLineage().GetChildrenSize(), 0);
}
TEST_F(LineageCacheTest, TestEvictionUncommittedChildren) {
// Insert a chain of dependent tasks.
size_t num_tasks_flushed = 0;
uint64_t lineage_size = max_lineage_size_ + 1;
std::vector<Task> tasks;
InsertTaskChain(*lineage_cache_, tasks, lineage_size, std::vector<ObjectID>(), 1);
// Add more tasks to the lineage cache that will remain local. Each of these
// tasks is dependent one of the tasks that was forwarded above.
for (const auto &task : tasks) {
auto return_id = task.GetTaskSpecification().ReturnId(0);
auto dependent_task = ExampleTask({return_id}, 1);
auto lineage = CreateSingletonLineage(dependent_task);
lineage_cache_->AddUncommittedLineage(dependent_task.GetTaskSpecification().TaskId(),
lineage);
ASSERT_TRUE(lineage_cache_->CommitTask(dependent_task));
// Once the forwarded tasks are evicted from the lineage cache, we expect
// each of these dependent tasks to be flushed, since all of their
// dependencies have been committed.
num_tasks_flushed++;
}
// Simulate executing the tasks on the remote node in reverse order and
// adding them to the GCS. Lineage at the local node should not get evicted
// until after the final remote task is executed, since a task can only be
// evicted once all of its ancestors have been committed.
for (auto it = tasks.rbegin(); it != tasks.rend(); it++) {
auto task_id = it->GetTaskSpecification().TaskId();
auto task_data = std::make_shared<TaskTableData>();
task_data->mutable_task()->mutable_task_spec()->set_task_id(task_id.Binary());
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), lineage_size * 2);
RAY_CHECK_OK(mock_gcs_->MockTasks().RemoteAdd(task_data));
num_tasks_flushed++;
mock_gcs_->MockTasks().Flush();
ASSERT_EQ(mock_gcs_->MockTasks().TaskTable().size(), num_tasks_flushed);
}
// Check that after the final remote task is executed, all local lineage is
// now evicted.
ASSERT_EQ(lineage_cache_->GetLineage().GetEntries().size(), 0);
ASSERT_EQ(lineage_cache_->GetLineage().GetChildrenSize(), 0);
}
TEST_F(LineageCacheTest, TestFlushAllUncommittedTasks) {
// Insert a chain of tasks.
std::vector<Task> tasks;
auto return_values =
InsertTaskChain(*lineage_cache_, tasks, 3, std::vector<ObjectID>(), 1);
std::vector<TaskID> task_ids;
for (const auto &task : tasks) {
task_ids.push_back(task.GetTaskSpecification().TaskId());
}
// Check that we subscribed to each of the uncommitted tasks.
ASSERT_EQ(mock_gcs_->MockTasks().NumRequestedNotifications(), task_ids.size());
// Flush all uncommitted tasks and make sure we add all tasks to
// the task table.
lineage_cache_->FlushAllUncommittedTasks();
ASSERT_EQ(mock_gcs_->MockTasks().NumTaskAdds(), tasks.size());
// Flush again and make sure there are no new tasks added to the
// task table.
lineage_cache_->FlushAllUncommittedTasks();
ASSERT_EQ(mock_gcs_->MockTasks().NumTaskAdds(), tasks.size());
// Flush all GCS notifications.
mock_gcs_->MockTasks().Flush();
// Make sure that we unsubscribed to the uncommitted tasks before
// we flushed them.
ASSERT_EQ(num_notifications_, 0);
// Flush again and make sure there are no new tasks added to the
// task table.
lineage_cache_->FlushAllUncommittedTasks();
ASSERT_EQ(mock_gcs_->MockTasks().NumTaskAdds(), tasks.size());
}
} // namespace raylet
} // namespace ray
int main(int argc, char **argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
src/ray/raylet/node_manager.cc
+99 -714
View File
File diff suppressed because it is too large Load Diff
src/ray/raylet/node_manager.h
+4 -55
View File
@@ -27,7 +27,6 @@
#include "ray/common/task/scheduling_resources.h"
#include "ray/object_manager/object_manager.h"
#include "ray/raylet/actor_registration.h"
#include "ray/raylet/lineage_cache.h"
#include "ray/raylet/scheduling/scheduling_ids.h"
#include "ray/raylet/scheduling/cluster_resource_scheduler.h"
#include "ray/raylet/scheduling/cluster_task_manager.h"
@@ -246,12 +245,8 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
/// Handle specified task's submission to the local node manager.
///
/// \param task The task being submitted.
/// \param uncommitted_lineage The uncommitted lineage of the task.
/// \param forwarded True if the task has been forwarded from a different
/// node manager and false if it was submitted by a local worker.
/// \return Void.
void SubmitTask(const Task &task, const Lineage &uncommitted_lineage,
bool forwarded = false);
void SubmitTask(const Task &task);
/// Assign a task to a worker. The task is assumed to not be queued in local_queues_.
///
/// \param[in] worker The worker to assign the task to.
@@ -274,25 +269,11 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
/// \param worker The port that the actor is listening on.
std::shared_ptr<ActorTableData> CreateActorTableDataFromCreationTask(
const TaskSpecification &task_spec, int port, const WorkerID &worker_id);
/// Handle a worker finishing an assigned actor task or actor creation task.
/// Handle a worker finishing an assigned actor creation task.
/// \param worker The worker that finished the task.
/// \param task The actor task or actor creation task.
/// \return Void.
void FinishAssignedActorTask(WorkerInterface &worker, const Task &task);
/// Helper function for handling worker to finish its assigned actor task
/// or actor creation task. Gets invoked when tasks's parent actor is known.
///
/// \param parent_actor_id The actor id corresponding to the actor which creates
/// the new actor.
/// \param task_spec Task specification of the actor creation task that created the
/// actor.
/// \param resumed_from_checkpoint If the actor was resumed from a checkpoint.
/// \param port Rpc server port that the actor is listening on.
/// \return Void.
void FinishAssignedActorCreationTask(const ActorID &parent_actor_id,
const TaskSpecification &task_spec,
bool resumed_from_checkpoint, int port,
const WorkerID &worker_id);
void FinishAssignedActorCreationTask(WorkerInterface &worker, const Task &task);
/// Make a placement decision for placeable tasks given the resource_map
/// provided. This will perform task state transitions and task forwarding.
///
@@ -321,13 +302,7 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
/// \return Void.
void HandleTaskReconstruction(const TaskID &task_id,
const ObjectID &required_object_id);
/// Resubmit a task for execution. This is a task that was previously already
/// submitted to a raylet but which must now be re-executed.
///
/// \param task The task being resubmitted.
/// \param required_object_id The object id that triggered the resubmission.
/// \return Void.
void ResubmitTask(const Task &task, const ObjectID &required_object_id);
/// Attempt to forward a task to a remote different node manager. If this
/// fails, the task will be resubmit locally.
///
@@ -553,13 +528,6 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
/// \param message_data A pointer to the message data.
void ProcessNotifyActorResumedFromCheckpoint(const uint8_t *message_data);
/// Update actor frontier when a task finishes.
/// If the task is an actor creation task and the actor was resumed from a checkpoint,
/// restore the frontier from the checkpoint. Otherwise, just extend actor frontier.
///
/// \param task The task that just finished.
void UpdateActorFrontier(const Task &task);
/// Process client message of SetResourceRequest
/// \param client The client that sent the message.
/// \param message_data A pointer to the message data.
@@ -567,18 +535,6 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
void ProcessSetResourceRequest(const std::shared_ptr<ClientConnection> &client,
const uint8_t *message_data);
/// Handle the case where an actor is disconnected, determine whether this
/// actor needs to be restarted and then update actor table.
/// This function needs to be called either when actor process dies or when
/// a node dies.
///
/// \param actor_id Id of this actor.
/// \param was_local Whether the disconnected was on this local node.
/// \param intentional_disconnect Wether the client was intentionally disconnected.
/// \return Void.
void HandleDisconnectedActor(const ActorID &actor_id, bool was_local,
bool intentional_disconnect);
/// Finish assigning a task to a worker.
///
/// \param worker Worker that the task is assigned to.
@@ -631,11 +587,6 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
rpc::CancelWorkerLeaseReply *reply,
rpc::SendReplyCallback send_reply_callback) override;
/// Handle a `ForwardTask` request.
void HandleForwardTask(const rpc::ForwardTaskRequest &request,
rpc::ForwardTaskReply *reply,
rpc::SendReplyCallback send_reply_callback) override;
/// Handle a `PinObjectIDs` request.
void HandlePinObjectIDs(const rpc::PinObjectIDsRequest &request,
rpc::PinObjectIDsReply *reply,
@@ -744,8 +695,6 @@ class NodeManager : public rpc::NodeManagerServiceHandler {
ReconstructionPolicy reconstruction_policy_;
/// A manager to make waiting tasks's missing object dependencies available.
TaskDependencyManager task_dependency_manager_;
/// The lineage cache for the GCS object and task tables.
LineageCache lineage_cache_;
/// A mapping from actor ID to registration information about that actor
/// (including which node manager owns it).
std::unordered_map<ActorID, ActorRegistration> actor_registry_;
src/ray/raylet/worker.cc
-21
View File
@@ -174,27 +174,6 @@ void Worker::AcquireTaskCpuResources(const ResourceIdSet &cpu_resources) {
task_resource_ids_.Release(cpu_resources);
}
Status Worker::AssignTask(const Task &task, const ResourceIdSet &resource_id_set) {
RAY_CHECK(port_ > 0);
rpc::AssignTaskRequest request;
request.set_intended_worker_id(worker_id_.Binary());
request.mutable_task()->mutable_task_spec()->CopyFrom(
task.GetTaskSpecification().GetMessage());
request.mutable_task()->mutable_task_execution_spec()->CopyFrom(
task.GetTaskExecutionSpec().GetMessage());
request.set_resource_ids(resource_id_set.Serialize());
rpc_client_->AssignTask(request, [](Status status, const rpc::AssignTaskReply &reply) {
if (!status.ok()) {
RAY_LOG(DEBUG) << "Worker failed to finish executing task: " << status.ToString();
}
// Worker has finished this task. There's nothing to do here
// and assigning new task will be done when raylet receives
// `TaskDone` message.
});
return Status::OK();
}
void Worker::DirectActorCallArgWaitComplete(int64_t tag) {
RAY_CHECK(port_ > 0);
rpc::DirectActorCallArgWaitCompleteRequest request;
src/ray/raylet/worker.h
-2
View File
@@ -79,7 +79,6 @@ class WorkerInterface {
virtual ResourceIdSet ReleaseTaskCpuResources() = 0;
virtual void AcquireTaskCpuResources(const ResourceIdSet &cpu_resources) = 0;
virtual Status AssignTask(const Task &task, const ResourceIdSet &resource_id_set) = 0;
virtual void DirectActorCallArgWaitComplete(int64_t tag) = 0;
// Setter, geter, and clear methods for allocated_instances_.
@@ -165,7 +164,6 @@ class Worker : public WorkerInterface {
ResourceIdSet ReleaseTaskCpuResources();
void AcquireTaskCpuResources(const ResourceIdSet &cpu_resources);
Status AssignTask(const Task &task, const ResourceIdSet &resource_id_set);
void DirectActorCallArgWaitComplete(int64_t tag);
// Setter, geter, and clear methods for allocated_instances_.
src/ray/rpc/node_manager/node_manager_client.h
-6
View File
@@ -41,12 +41,6 @@ class NodeManagerClient {
new GrpcClient<NodeManagerService>(address, port, client_call_manager));
};
/// Forward a task and its uncommitted lineage.
///
/// \param[in] request The request message.
/// \param[in] callback The callback function that handles reply.
VOID_RPC_CLIENT_METHOD(NodeManagerService, ForwardTask, grpc_client_, )
/// Get current node stats.
VOID_RPC_CLIENT_METHOD(NodeManagerService, GetNodeStats, grpc_client_, )
src/ray/rpc/node_manager/node_manager_server.h
-5
View File
@@ -28,7 +28,6 @@ namespace rpc {
RPC_SERVICE_HANDLER(NodeManagerService, ReturnWorker) \
RPC_SERVICE_HANDLER(NodeManagerService, ReleaseUnusedWorkers) \
RPC_SERVICE_HANDLER(NodeManagerService, CancelWorkerLease) \
RPC_SERVICE_HANDLER(NodeManagerService, ForwardTask) \
RPC_SERVICE_HANDLER(NodeManagerService, PinObjectIDs) \
RPC_SERVICE_HANDLER(NodeManagerService, GetNodeStats) \
RPC_SERVICE_HANDLER(NodeManagerService, GlobalGC) \
@@ -76,10 +75,6 @@ class NodeManagerServiceHandler {
rpc::CancelResourceReserveReply *reply,
rpc::SendReplyCallback send_reply_callback) = 0;
virtual void HandleForwardTask(const ForwardTaskRequest &request,
ForwardTaskReply *reply,
SendReplyCallback send_reply_callback) = 0;
virtual void HandlePinObjectIDs(const PinObjectIDsRequest &request,
PinObjectIDsReply *reply,
SendReplyCallback send_reply_callback) = 0;
src/ray/rpc/worker/core_worker_client.h
-10
View File
@@ -104,14 +104,6 @@ class CoreWorkerClientInterface {
return empty_addr_;
}
/// This is called by the Raylet to assign a task to the worker.
///
/// \param[in] request The request message.
/// \param[in] callback The callback function that handles reply.
/// \return if the rpc call succeeds
virtual void AssignTask(const AssignTaskRequest &request,
const ClientCallback<AssignTaskReply> &callback) {}
/// Push an actor task directly from worker to worker.
///
/// \param[in] request The request message.
@@ -196,8 +188,6 @@ class CoreWorkerClient : public std::enable_shared_from_this<CoreWorkerClient>,
const rpc::Address &Addr() const override { return addr_; }
VOID_RPC_CLIENT_METHOD(CoreWorkerService, AssignTask, grpc_client_, override)
VOID_RPC_CLIENT_METHOD(CoreWorkerService, DirectActorCallArgWaitComplete, grpc_client_,
override)
src/ray/rpc/worker/core_worker_server.h
-2
View File
@@ -27,7 +27,6 @@ namespace rpc {
/// NOTE: See src/ray/core_worker/core_worker.h on how to add a new grpc handler.
#define RAY_CORE_WORKER_RPC_HANDLERS \
RPC_SERVICE_HANDLER(CoreWorkerService, AssignTask) \
RPC_SERVICE_HANDLER(CoreWorkerService, PushTask) \
RPC_SERVICE_HANDLER(CoreWorkerService, DirectActorCallArgWaitComplete) \
RPC_SERVICE_HANDLER(CoreWorkerService, GetObjectStatus) \
@@ -42,7 +41,6 @@ namespace rpc {
RPC_SERVICE_HANDLER(CoreWorkerService, PlasmaObjectReady)
#define RAY_CORE_WORKER_DECLARE_RPC_HANDLERS \
DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(AssignTask) \
DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(PushTask) \
DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(DirectActorCallArgWaitComplete) \
DECLARE_VOID_RPC_SERVICE_HANDLER_METHOD(GetObjectStatus) \
src/ray/stats/metric_defs.h
-4
View File
@@ -61,10 +61,6 @@ static Gauge ObjectManagerStats("object_manager_stats",
"Stat the metric values of object in raylet", "pcs",
{ValueTypeKey});
static Gauge LineageCacheStats("lineage_cache_stats",
"Stats the metric values of lineage cache.", "pcs",
{ValueTypeKey});
static Gauge TaskDependencyManagerStats("task_dependency_manager_stats",
"Stat the metric values of task dependency.",
"pcs", {ValueTypeKey});