From 8dfc833a8b6881273a66935da7b5e1229fc39ac9 Mon Sep 17 00:00:00 2001 From: justinwyang <31091446+justinwyang@users.noreply.github.com> Date: Mon, 22 Apr 2019 16:28:09 -0700 Subject: [PATCH] Change all instances of JobID to DriverID. (#4431) --- python/ray/experimental/state.py | 39 +-- python/ray/gcs_utils.py | 2 +- python/ray/includes/unique_ids.pxd | 16 - python/ray/includes/unique_ids.pxi | 1 - python/ray/worker.py | 13 +- src/ray/gcs/client.h | 4 +- src/ray/gcs/client_test.cc | 289 +++++++++--------- src/ray/gcs/format/gcs.fbs | 4 +- src/ray/gcs/tables.cc | 75 ++--- src/ray/gcs/tables.h | 102 ++++--- src/ray/id_def.h | 1 - src/ray/object_manager/object_directory.cc | 12 +- src/ray/raylet/lineage_cache.cc | 6 +- src/ray/raylet/lineage_cache_test.cc | 8 +- src/ray/raylet/monitor.cc | 6 +- src/ray/raylet/node_manager.cc | 32 +- src/ray/raylet/raylet_client.h | 5 +- src/ray/raylet/reconstruction_policy.cc | 6 +- src/ray/raylet/reconstruction_policy_test.cc | 16 +- src/ray/raylet/task_dependency_manager.cc | 2 +- .../raylet/task_dependency_manager_test.cc | 2 +- 21 files changed, 319 insertions(+), 322 deletions(-) diff --git a/python/ray/experimental/state.py b/python/ray/experimental/state.py index b884d0794..31d4b77c6 100644 --- a/python/ray/experimental/state.py +++ b/python/ray/experimental/state.py @@ -840,19 +840,19 @@ class GlobalState(object): return dict(total_available_resources) - def _error_messages(self, job_id): - """Get the error messages for a specific job. + def _error_messages(self, driver_id): + """Get the error messages for a specific driver. Args: - job_id: The ID of the job to get the errors for. + driver_id: The ID of the driver to get the errors for. Returns: - A list of the error messages for this job. + A list of the error messages for this driver. """ - assert isinstance(job_id, ray.DriverID) + assert isinstance(driver_id, ray.DriverID) message = self.redis_client.execute_command( "RAY.TABLE_LOOKUP", ray.gcs_utils.TablePrefix.ERROR_INFO, "", - job_id.binary()) + driver_id.binary()) # If there are no errors, return early. if message is None: @@ -864,7 +864,7 @@ class GlobalState(object): for i in range(gcs_entries.EntriesLength()): error_data = ray.gcs_utils.ErrorTableData.GetRootAsErrorTableData( gcs_entries.Entries(i), 0) - assert job_id.binary() == error_data.JobId() + assert driver_id.binary() == error_data.DriverId() error_message = { "type": decode(error_data.Type()), "message": decode(error_data.ErrorMessage()), @@ -873,31 +873,32 @@ class GlobalState(object): error_messages.append(error_message) return error_messages - def error_messages(self, job_id=None): - """Get the error messages for all jobs or a specific job. + def error_messages(self, driver_id=None): + """Get the error messages for all drivers or a specific driver. Args: - job_id: The specific job to get the errors for. If this is None, - then this method retrieves the errors for all jobs. + driver_id: The specific driver to get the errors for. If this is + None, then this method retrieves the errors for all drivers. Returns: - A dictionary mapping job ID to a list of the error messages for - that job. + A dictionary mapping driver ID to a list of the error messages for + that driver. """ - if job_id is not None: - assert isinstance(job_id, ray.DriverID) - return self._error_messages(job_id) + if driver_id is not None: + assert isinstance(driver_id, ray.DriverID) + return self._error_messages(driver_id) error_table_keys = self.redis_client.keys( ray.gcs_utils.TablePrefix_ERROR_INFO_string + "*") - job_ids = [ + driver_ids = [ key[len(ray.gcs_utils.TablePrefix_ERROR_INFO_string):] for key in error_table_keys ] return { - binary_to_hex(job_id): self._error_messages(ray.DriverID(job_id)) - for job_id in job_ids + binary_to_hex(driver_id): self._error_messages( + ray.DriverID(driver_id)) + for driver_id in driver_ids } def actor_checkpoint_info(self, actor_id): diff --git a/python/ray/gcs_utils.py b/python/ray/gcs_utils.py index 3b204470f..15eec6c81 100644 --- a/python/ray/gcs_utils.py +++ b/python/ray/gcs_utils.py @@ -76,7 +76,7 @@ def construct_error_message(driver_id, error_type, message, timestamp): message_offset = builder.CreateString(message) ray.core.generated.ErrorTableData.ErrorTableDataStart(builder) - ray.core.generated.ErrorTableData.ErrorTableDataAddJobId( + ray.core.generated.ErrorTableData.ErrorTableDataAddDriverId( builder, driver_offset) ray.core.generated.ErrorTableData.ErrorTableDataAddType( builder, error_type_offset) diff --git a/python/ray/includes/unique_ids.pxd b/python/ray/includes/unique_ids.pxd index cadbdfea2..a607b2a86 100644 --- a/python/ray/includes/unique_ids.pxd +++ b/python/ray/includes/unique_ids.pxd @@ -32,67 +32,51 @@ cdef extern from "ray/id.h" namespace "ray" nogil: @staticmethod CActorCheckpointID from_binary(const c_string &binary) - cdef cppclass CActorClassID "ray::ActorClassID"(CUniqueID): @staticmethod CActorClassID from_binary(const c_string &binary) - cdef cppclass CActorID "ray::ActorID"(CUniqueID): @staticmethod CActorID from_binary(const c_string &binary) - cdef cppclass CActorHandleID "ray::ActorHandleID"(CUniqueID): @staticmethod CActorHandleID from_binary(const c_string &binary) - cdef cppclass CClientID "ray::ClientID"(CUniqueID): @staticmethod CClientID from_binary(const c_string &binary) - cdef cppclass CConfigID "ray::ConfigID"(CUniqueID): @staticmethod CConfigID from_binary(const c_string &binary) - cdef cppclass CFunctionID "ray::FunctionID"(CUniqueID): @staticmethod CFunctionID from_binary(const c_string &binary) - cdef cppclass CDriverID "ray::DriverID"(CUniqueID): @staticmethod CDriverID from_binary(const c_string &binary) - - cdef cppclass CJobID "ray::JobID"(CUniqueID): - - @staticmethod - CJobID from_binary(const c_string &binary) - - cdef cppclass CTaskID "ray::TaskID"(CUniqueID): @staticmethod CTaskID from_binary(const c_string &binary) - cdef cppclass CObjectID" ray::ObjectID"(CUniqueID): @staticmethod CObjectID from_binary(const c_string &binary) - cdef cppclass CWorkerID "ray::WorkerID"(CUniqueID): @staticmethod diff --git a/python/ray/includes/unique_ids.pxi b/python/ray/includes/unique_ids.pxi index 0086f76b5..c96668f2b 100644 --- a/python/ray/includes/unique_ids.pxi +++ b/python/ray/includes/unique_ids.pxi @@ -19,7 +19,6 @@ from ray.includes.unique_ids cimport ( CConfigID, CDriverID, CFunctionID, - CJobID, CObjectID, CTaskID, CUniqueID, diff --git a/python/ray/worker.py b/python/ray/worker.py index 3f4072087..f8c66c72a 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -1142,8 +1142,8 @@ def error_info(): """Return information about failed tasks.""" worker = global_worker worker.check_connected() - return (global_state.error_messages(job_id=worker.task_driver_id) + - global_state.error_messages(job_id=DriverID.nil())) + return (global_state.error_messages(driver_id=worker.task_driver_id) + + global_state.error_messages(driver_id=DriverID.nil())) def _initialize_serialization(driver_id, worker=global_worker): @@ -1288,8 +1288,9 @@ def init(redis_address=None, node_ip_address (str): The IP address of the node that we are on. object_id_seed (int): Used to seed the deterministic generation of object IDs. The same value can be used across multiple runs of the - same job in order to generate the object IDs in a consistent - manner. However, the same ID should not be used for different jobs. + same driver in order to generate the object IDs in a consistent + manner. However, the same ID should not be used for different + drivers. local_mode (bool): True if the code should be executed serially without Ray. This is useful for debugging. ignore_reinit_error: True if we should suppress errors from calling @@ -1663,8 +1664,8 @@ def listen_error_messages_raylet(worker, task_error_queue, threads_stopped): assert gcs_entry.EntriesLength() == 1 error_data = ray.gcs_utils.ErrorTableData.GetRootAsErrorTableData( gcs_entry.Entries(0), 0) - job_id = error_data.JobId() - if job_id not in [ + driver_id = error_data.DriverId() + if driver_id not in [ worker.task_driver_id.binary(), DriverID.nil().binary() ]: diff --git a/src/ray/gcs/client.h b/src/ray/gcs/client.h index 062af0dc4..7a5d8ef0e 100644 --- a/src/ray/gcs/client.h +++ b/src/ray/gcs/client.h @@ -109,10 +109,10 @@ class RAY_EXPORT AsyncGcsClient { class SyncGcsClient { Status LogEvent(const std::string &key, const std::string &value, double timestamp); Status NotifyError(const std::map &error_info); - Status RegisterFunction(const JobID &job_id, const FunctionID &function_id, + Status RegisterFunction(const DriverID &driver_id, const FunctionID &function_id, const std::string &language, const std::string &name, const std::string &data); - Status RetrieveFunction(const JobID &job_id, const FunctionID &function_id, + Status RetrieveFunction(const DriverID &driver_id, const FunctionID &function_id, std::string *name, std::string *data); Status AddExport(const std::string &driver_id, std::string &export_data); diff --git a/src/ray/gcs/client_test.cc b/src/ray/gcs/client_test.cc index 69b0cbac7..d2d225c0a 100644 --- a/src/ray/gcs/client_test.cc +++ b/src/ray/gcs/client_test.cc @@ -29,7 +29,7 @@ class TestGcs : public ::testing::Test { TestGcs(CommandType command_type) : num_callbacks_(0), command_type_(command_type) { client_ = std::make_shared("127.0.0.1", 6379, command_type_, /*is_test_client=*/true); - job_id_ = JobID::from_random(); + driver_id_ = DriverID::from_random(); } virtual ~TestGcs() { @@ -49,7 +49,7 @@ class TestGcs : public ::testing::Test { uint64_t num_callbacks_; gcs::CommandType command_type_; std::shared_ptr client_; - JobID job_id_; + DriverID driver_id_; }; TestGcs *test; @@ -82,7 +82,8 @@ class TestGcsWithChainAsio : public TestGcsWithAsio { TestGcsWithChainAsio() : TestGcsWithAsio(gcs::CommandType::kChain){}; }; -void TestTableLookup(const JobID &job_id, std::shared_ptr client) { +void TestTableLookup(const DriverID &driver_id, + std::shared_ptr client) { TaskID task_id = TaskID::from_random(); auto data = std::make_shared(); data->task_specification = "123"; @@ -108,8 +109,8 @@ void TestTableLookup(const JobID &job_id, std::shared_ptr c }; // Add the task, then do a lookup. - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id, data, add_callback)); - RAY_CHECK_OK(client->raylet_task_table().Lookup(job_id, task_id, lookup_callback, + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id, data, add_callback)); + RAY_CHECK_OK(client->raylet_task_table().Lookup(driver_id, task_id, lookup_callback, failure_callback)); // Run the event loop. The loop will only stop if the Lookup callback is // called (or an assertion failure). @@ -121,7 +122,7 @@ void TestTableLookup(const JobID &job_id, std::shared_ptr c #define TEST_MACRO(FIXTURE, TEST) \ TEST_F(FIXTURE, TEST) { \ test = this; \ - TEST(job_id_, client_); \ + TEST(driver_id_, client_); \ } TEST_MACRO(TestGcsWithAsio, TestTableLookup); @@ -129,7 +130,8 @@ TEST_MACRO(TestGcsWithAsio, TestTableLookup); TEST_MACRO(TestGcsWithChainAsio, TestTableLookup); #endif -void TestLogLookup(const JobID &job_id, std::shared_ptr client) { +void TestLogLookup(const DriverID &driver_id, + std::shared_ptr client) { // Append some entries to the log at an object ID. TaskID task_id = TaskID::from_random(); std::vector node_manager_ids = {"abc", "def", "ghi"}; @@ -143,7 +145,7 @@ void TestLogLookup(const JobID &job_id, std::shared_ptr cli ASSERT_EQ(data->node_manager_id, d.node_manager_id); }; RAY_CHECK_OK( - client->task_reconstruction_log().Append(job_id, task_id, data, add_callback)); + client->task_reconstruction_log().Append(driver_id, task_id, data, add_callback)); } // Check that lookup returns the added object entries. @@ -162,7 +164,7 @@ void TestLogLookup(const JobID &job_id, std::shared_ptr cli // Do a lookup at the object ID. RAY_CHECK_OK( - client->task_reconstruction_log().Lookup(job_id, task_id, lookup_callback)); + client->task_reconstruction_log().Lookup(driver_id, task_id, lookup_callback)); // Run the event loop. The loop will only stop if the Lookup callback is // called (or an assertion failure). test->Start(); @@ -171,10 +173,10 @@ void TestLogLookup(const JobID &job_id, std::shared_ptr cli TEST_F(TestGcsWithAsio, TestLogLookup) { test = this; - TestLogLookup(job_id_, client_); + TestLogLookup(driver_id_, client_); } -void TestTableLookupFailure(const JobID &job_id, +void TestTableLookupFailure(const DriverID &driver_id, std::shared_ptr client) { TaskID task_id = TaskID::from_random(); @@ -189,7 +191,7 @@ void TestTableLookupFailure(const JobID &job_id, }; // Lookup the task. We have not done any writes, so the key should be empty. - RAY_CHECK_OK(client->raylet_task_table().Lookup(job_id, task_id, lookup_callback, + RAY_CHECK_OK(client->raylet_task_table().Lookup(driver_id, task_id, lookup_callback, failure_callback)); // Run the event loop. The loop will only stop if the failure callback is // called (or an assertion failure). @@ -201,7 +203,8 @@ TEST_MACRO(TestGcsWithAsio, TestTableLookupFailure); TEST_MACRO(TestGcsWithChainAsio, TestTableLookupFailure); #endif -void TestLogAppendAt(const JobID &job_id, std::shared_ptr client) { +void TestLogAppendAt(const DriverID &driver_id, + std::shared_ptr client) { TaskID task_id = TaskID::from_random(); std::vector node_manager_ids = {"A", "B"}; std::vector> data_log; @@ -219,21 +222,22 @@ void TestLogAppendAt(const JobID &job_id, std::shared_ptr c }; // Will succeed. - RAY_CHECK_OK(client->task_reconstruction_log().Append(job_id, task_id, data_log.front(), + RAY_CHECK_OK(client->task_reconstruction_log().Append(driver_id, task_id, + data_log.front(), /*done callback=*/nullptr)); // Append at index 0 will fail. RAY_CHECK_OK(client->task_reconstruction_log().AppendAt( - job_id, task_id, data_log[1], + driver_id, task_id, data_log[1], /*done callback=*/nullptr, failure_callback, /*log_length=*/0)); // Append at index 2 will fail. RAY_CHECK_OK(client->task_reconstruction_log().AppendAt( - job_id, task_id, data_log[1], + driver_id, task_id, data_log[1], /*done callback=*/nullptr, failure_callback, /*log_length=*/2)); // Append at index 1 will succeed. RAY_CHECK_OK(client->task_reconstruction_log().AppendAt( - job_id, task_id, data_log[1], + driver_id, task_id, data_log[1], /*done callback=*/nullptr, failure_callback, /*log_length=*/1)); auto lookup_callback = [node_manager_ids]( @@ -247,7 +251,7 @@ void TestLogAppendAt(const JobID &job_id, std::shared_ptr c test->Stop(); }; RAY_CHECK_OK( - client->task_reconstruction_log().Lookup(job_id, task_id, lookup_callback)); + client->task_reconstruction_log().Lookup(driver_id, task_id, lookup_callback)); // Run the event loop. The loop will only stop if the Lookup callback is // called (or an assertion failure). test->Start(); @@ -256,10 +260,10 @@ void TestLogAppendAt(const JobID &job_id, std::shared_ptr c TEST_F(TestGcsWithAsio, TestLogAppendAt) { test = this; - TestLogAppendAt(job_id_, client_); + TestLogAppendAt(driver_id_, client_); } -void TestSet(const JobID &job_id, std::shared_ptr client) { +void TestSet(const DriverID &driver_id, std::shared_ptr client) { // Add some entries to the set at an object ID. ObjectID object_id = ObjectID::from_random(); std::vector managers = {"abc", "def", "ghi"}; @@ -273,7 +277,7 @@ void TestSet(const JobID &job_id, std::shared_ptr client) { ASSERT_EQ(data->manager, d.manager); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id, data, add_callback)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id, data, add_callback)); } // Check that lookup returns the added object entries. @@ -286,7 +290,7 @@ void TestSet(const JobID &job_id, std::shared_ptr client) { }; // Do a lookup at the object ID. - RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, lookup_callback)); + RAY_CHECK_OK(client->object_table().Lookup(driver_id, object_id, lookup_callback)); for (auto &manager : managers) { auto data = std::make_shared(); @@ -299,7 +303,7 @@ void TestSet(const JobID &job_id, std::shared_ptr client) { test->IncrementNumCallbacks(); }; RAY_CHECK_OK( - client->object_table().Remove(job_id, object_id, data, remove_entry_callback)); + client->object_table().Remove(driver_id, object_id, data, remove_entry_callback)); } // Check that the entries are removed. @@ -313,7 +317,7 @@ void TestSet(const JobID &job_id, std::shared_ptr client) { }; // Do a lookup at the object ID. - RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, lookup_callback2)); + RAY_CHECK_OK(client->object_table().Lookup(driver_id, object_id, lookup_callback2)); // Run the event loop. The loop will only stop if the Lookup callback is // called (or an assertion failure). test->Start(); @@ -322,11 +326,11 @@ void TestSet(const JobID &job_id, std::shared_ptr client) { TEST_F(TestGcsWithAsio, TestSet) { test = this; - TestSet(job_id_, client_); + TestSet(driver_id_, client_); } void TestDeleteKeysFromLog( - const JobID &job_id, std::shared_ptr client, + const DriverID &driver_id, std::shared_ptr client, std::vector> &data_vector) { std::vector ids; TaskID task_id; @@ -341,7 +345,7 @@ void TestDeleteKeysFromLog( test->IncrementNumCallbacks(); }; RAY_CHECK_OK( - client->task_reconstruction_log().Append(job_id, task_id, data, add_callback)); + client->task_reconstruction_log().Append(driver_id, task_id, data, add_callback)); } for (const auto &task_id : ids) { // Check that lookup returns the added object entries. @@ -353,12 +357,12 @@ void TestDeleteKeysFromLog( test->IncrementNumCallbacks(); }; RAY_CHECK_OK( - client->task_reconstruction_log().Lookup(job_id, task_id, lookup_callback)); + client->task_reconstruction_log().Lookup(driver_id, task_id, lookup_callback)); } if (ids.size() == 1) { - client->task_reconstruction_log().Delete(job_id, ids[0]); + client->task_reconstruction_log().Delete(driver_id, ids[0]); } else { - client->task_reconstruction_log().Delete(job_id, ids); + client->task_reconstruction_log().Delete(driver_id, ids); } for (const auto &task_id : ids) { auto lookup_callback = [task_id](gcs::AsyncGcsClient *client, const TaskID &id, @@ -368,11 +372,11 @@ void TestDeleteKeysFromLog( test->IncrementNumCallbacks(); }; RAY_CHECK_OK( - client->task_reconstruction_log().Lookup(job_id, task_id, lookup_callback)); + client->task_reconstruction_log().Lookup(driver_id, task_id, lookup_callback)); } } -void TestDeleteKeysFromTable(const JobID &job_id, +void TestDeleteKeysFromTable(const DriverID &driver_id, std::shared_ptr client, std::vector> &data_vector, bool stop_at_end) { @@ -388,7 +392,7 @@ void TestDeleteKeysFromTable(const JobID &job_id, ASSERT_EQ(data->task_specification, d.task_specification); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id, data, add_callback)); + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id, data, add_callback)); } for (const auto &task_id : ids) { auto task_lookup_callback = [task_id](gcs::AsyncGcsClient *client, const TaskID &id, @@ -396,13 +400,13 @@ void TestDeleteKeysFromTable(const JobID &job_id, ASSERT_EQ(id, task_id); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->raylet_task_table().Lookup(job_id, task_id, task_lookup_callback, - nullptr)); + RAY_CHECK_OK(client->raylet_task_table().Lookup(driver_id, task_id, + task_lookup_callback, nullptr)); } if (ids.size() == 1) { - client->raylet_task_table().Delete(job_id, ids[0]); + client->raylet_task_table().Delete(driver_id, ids[0]); } else { - client->raylet_task_table().Delete(job_id, ids); + client->raylet_task_table().Delete(driver_id, ids); } auto expected_failure_callback = [](AsyncGcsClient *client, const TaskID &id) { ASSERT_TRUE(true); @@ -411,17 +415,17 @@ void TestDeleteKeysFromTable(const JobID &job_id, auto undesired_callback = [](gcs::AsyncGcsClient *client, const TaskID &id, const protocol::TaskT &data) { ASSERT_TRUE(false); }; for (size_t i = 0; i < ids.size(); ++i) { - RAY_CHECK_OK(client->raylet_task_table().Lookup(job_id, task_id, undesired_callback, - expected_failure_callback)); + RAY_CHECK_OK(client->raylet_task_table().Lookup( + driver_id, task_id, undesired_callback, expected_failure_callback)); } if (stop_at_end) { auto stop_callback = [](AsyncGcsClient *client, const TaskID &id) { test->Stop(); }; RAY_CHECK_OK( - client->raylet_task_table().Lookup(job_id, ids[0], nullptr, stop_callback)); + client->raylet_task_table().Lookup(driver_id, ids[0], nullptr, stop_callback)); } } -void TestDeleteKeysFromSet(const JobID &job_id, +void TestDeleteKeysFromSet(const DriverID &driver_id, std::shared_ptr client, std::vector> &data_vector) { std::vector ids; @@ -436,7 +440,7 @@ void TestDeleteKeysFromSet(const JobID &job_id, ASSERT_EQ(data->manager, d.manager); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id, data, add_callback)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id, data, add_callback)); } for (const auto &object_id : ids) { // Check that lookup returns the added object entries. @@ -447,12 +451,12 @@ void TestDeleteKeysFromSet(const JobID &job_id, ASSERT_EQ(data.size(), 1); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, lookup_callback)); + RAY_CHECK_OK(client->object_table().Lookup(driver_id, object_id, lookup_callback)); } if (ids.size() == 1) { - client->object_table().Delete(job_id, ids[0]); + client->object_table().Delete(driver_id, ids[0]); } else { - client->object_table().Delete(job_id, ids); + client->object_table().Delete(driver_id, ids); } for (const auto &object_id : ids) { auto lookup_callback = [object_id](gcs::AsyncGcsClient *client, const ObjectID &id, @@ -461,12 +465,13 @@ void TestDeleteKeysFromSet(const JobID &job_id, ASSERT_TRUE(data.size() == 0); test->IncrementNumCallbacks(); }; - RAY_CHECK_OK(client->object_table().Lookup(job_id, object_id, lookup_callback)); + RAY_CHECK_OK(client->object_table().Lookup(driver_id, object_id, lookup_callback)); } } // Test delete function for keys of Log or Table. -void TestDeleteKeys(const JobID &job_id, std::shared_ptr client) { +void TestDeleteKeys(const DriverID &driver_id, + std::shared_ptr client) { // Test delete function for keys of Log. std::vector> task_reconstruction_vector; auto AppendTaskReconstructionData = [&task_reconstruction_vector](size_t add_count) { @@ -479,7 +484,7 @@ void TestDeleteKeys(const JobID &job_id, std::shared_ptr cl // Test one element case. AppendTaskReconstructionData(1); ASSERT_EQ(task_reconstruction_vector.size(), 1); - TestDeleteKeysFromLog(job_id, client, task_reconstruction_vector); + TestDeleteKeysFromLog(driver_id, client, task_reconstruction_vector); // Test the case for more than one elements and less than // maximum_gcs_deletion_batch_size. AppendTaskReconstructionData(RayConfig::instance().maximum_gcs_deletion_batch_size() / @@ -487,14 +492,14 @@ void TestDeleteKeys(const JobID &job_id, std::shared_ptr cl ASSERT_GT(task_reconstruction_vector.size(), 1); ASSERT_LT(task_reconstruction_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromLog(job_id, client, task_reconstruction_vector); + TestDeleteKeysFromLog(driver_id, client, task_reconstruction_vector); // Test the case for more than maximum_gcs_deletion_batch_size. // The Delete function will split the data into two commands. AppendTaskReconstructionData(RayConfig::instance().maximum_gcs_deletion_batch_size() / 2); ASSERT_GT(task_reconstruction_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromLog(job_id, client, task_reconstruction_vector); + TestDeleteKeysFromLog(driver_id, client, task_reconstruction_vector); // Test delete function for keys of Table. std::vector> task_vector; @@ -507,16 +512,16 @@ void TestDeleteKeys(const JobID &job_id, std::shared_ptr cl }; AppendTaskData(1); ASSERT_EQ(task_vector.size(), 1); - TestDeleteKeysFromTable(job_id, client, task_vector, false); + TestDeleteKeysFromTable(driver_id, client, task_vector, false); AppendTaskData(RayConfig::instance().maximum_gcs_deletion_batch_size() / 2); ASSERT_GT(task_vector.size(), 1); ASSERT_LT(task_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromTable(job_id, client, task_vector, false); + TestDeleteKeysFromTable(driver_id, client, task_vector, false); AppendTaskData(RayConfig::instance().maximum_gcs_deletion_batch_size() / 2); ASSERT_GT(task_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromTable(job_id, client, task_vector, true); + TestDeleteKeysFromTable(driver_id, client, task_vector, true); test->Start(); ASSERT_GT(test->NumCallbacks(), @@ -534,25 +539,25 @@ void TestDeleteKeys(const JobID &job_id, std::shared_ptr cl // Test one element case. AppendObjectData(1); ASSERT_EQ(object_vector.size(), 1); - TestDeleteKeysFromSet(job_id, client, object_vector); + TestDeleteKeysFromSet(driver_id, client, object_vector); // Test the case for more than one elements and less than // maximum_gcs_deletion_batch_size. AppendObjectData(RayConfig::instance().maximum_gcs_deletion_batch_size() / 2); ASSERT_GT(object_vector.size(), 1); ASSERT_LT(object_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromSet(job_id, client, object_vector); + TestDeleteKeysFromSet(driver_id, client, object_vector); // Test the case for more than maximum_gcs_deletion_batch_size. // The Delete function will split the data into two commands. AppendObjectData(RayConfig::instance().maximum_gcs_deletion_batch_size() / 2); ASSERT_GT(object_vector.size(), RayConfig::instance().maximum_gcs_deletion_batch_size()); - TestDeleteKeysFromSet(job_id, client, object_vector); + TestDeleteKeysFromSet(driver_id, client, object_vector); } TEST_F(TestGcsWithAsio, TestDeleteKey) { test = this; - TestDeleteKeys(job_id_, client_); + TestDeleteKeys(driver_id_, client_); } // Task table callbacks. @@ -594,7 +599,7 @@ void TaskLookupAfterUpdateFailure(gcs::AsyncGcsClient *client, const TaskID &id) test->Stop(); } -void TestLogSubscribeAll(const JobID &job_id, +void TestLogSubscribeAll(const DriverID &driver_id, std::shared_ptr client) { std::vector driver_ids; for (int i = 0; i < 3; i++) { @@ -628,7 +633,7 @@ void TestLogSubscribeAll(const JobID &job_id, // subscribed, we will append to the key several times and check that we get // notified for each. RAY_CHECK_OK(client->driver_table().Subscribe( - job_id, ClientID::nil(), notification_callback, subscribe_callback)); + driver_id, ClientID::nil(), notification_callback, subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called (or an assertion failure). @@ -639,10 +644,10 @@ void TestLogSubscribeAll(const JobID &job_id, TEST_F(TestGcsWithAsio, TestLogSubscribeAll) { test = this; - TestLogSubscribeAll(job_id_, client_); + TestLogSubscribeAll(driver_id_, client_); } -void TestSetSubscribeAll(const JobID &job_id, +void TestSetSubscribeAll(const DriverID &driver_id, std::shared_ptr client) { std::vector object_ids; for (int i = 0; i < 3; i++) { @@ -673,7 +678,8 @@ void TestSetSubscribeAll(const JobID &job_id, // Callback for subscription success. We are guaranteed to receive // notifications after this is called. - auto subscribe_callback = [job_id, object_ids, managers](gcs::AsyncGcsClient *client) { + auto subscribe_callback = [driver_id, object_ids, + managers](gcs::AsyncGcsClient *client) { // We have subscribed. Do the writes to the table. for (size_t i = 0; i < object_ids.size(); i++) { for (size_t j = 0; j < managers.size(); j++) { @@ -682,7 +688,8 @@ void TestSetSubscribeAll(const JobID &job_id, for (int k = 0; k < 3; k++) { // Add the same entry several times. // Expect no notification if the entry already exists. - RAY_CHECK_OK(client->object_table().Add(job_id, object_ids[i], data, nullptr)); + RAY_CHECK_OK( + client->object_table().Add(driver_id, object_ids[i], data, nullptr)); } } } @@ -694,7 +701,7 @@ void TestSetSubscribeAll(const JobID &job_id, // Remove the same entry several times. // Expect no notification if the entry doesn't exist. RAY_CHECK_OK( - client->object_table().Remove(job_id, object_ids[i], data, nullptr)); + client->object_table().Remove(driver_id, object_ids[i], data, nullptr)); } } } @@ -704,7 +711,7 @@ void TestSetSubscribeAll(const JobID &job_id, // subscribed, we will append to the key several times and check that we get // notified for each. RAY_CHECK_OK(client->object_table().Subscribe( - job_id, ClientID::nil(), notification_callback, subscribe_callback)); + driver_id, ClientID::nil(), notification_callback, subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called (or an assertion failure). @@ -715,10 +722,10 @@ void TestSetSubscribeAll(const JobID &job_id, TEST_F(TestGcsWithAsio, TestSetSubscribeAll) { test = this; - TestSetSubscribeAll(job_id_, client_); + TestSetSubscribeAll(driver_id_, client_); } -void TestTableSubscribeId(const JobID &job_id, +void TestTableSubscribeId(const DriverID &driver_id, std::shared_ptr client) { // Add a table entry. TaskID task_id1 = TaskID::from_random(); @@ -754,29 +761,29 @@ void TestTableSubscribeId(const JobID &job_id, // The callback for subscription success. Once we've subscribed, request // notifications for only one of the keys, then write to both keys. - auto subscribe_callback = [job_id, task_id1, task_id2, task_specs1, + auto subscribe_callback = [driver_id, task_id1, task_id2, task_specs1, task_specs2](gcs::AsyncGcsClient *client) { // Request notifications for one of the keys. RAY_CHECK_OK(client->raylet_task_table().RequestNotifications( - job_id, task_id2, client->client_table().GetLocalClientId())); + driver_id, task_id2, client->client_table().GetLocalClientId())); // Write both keys. We should only receive notifications for the key that // we requested them for. for (const auto &task_spec : task_specs1) { auto data = std::make_shared(); data->task_specification = task_spec; - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id1, data, nullptr)); + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id1, data, nullptr)); } for (const auto &task_spec : task_specs2) { auto data = std::make_shared(); data->task_specification = task_spec; - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id2, data, nullptr)); + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id2, data, nullptr)); } }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. RAY_CHECK_OK(client->raylet_task_table().Subscribe( - job_id, client->client_table().GetLocalClientId(), notification_callback, + driver_id, client->client_table().GetLocalClientId(), notification_callback, failure_callback, subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. @@ -794,21 +801,21 @@ TEST_MACRO(TestGcsWithAsio, TestTableSubscribeId); TEST_MACRO(TestGcsWithChainAsio, TestTableSubscribeId); #endif -void TestLogSubscribeId(const JobID &job_id, +void TestLogSubscribeId(const DriverID &driver_id, std::shared_ptr client) { // Add a log entry. DriverID driver_id1 = DriverID::from_random(); std::vector driver_ids1 = {"abc", "def", "ghi"}; auto data1 = std::make_shared(); data1->driver_id = driver_ids1[0]; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id1, data1, nullptr)); + RAY_CHECK_OK(client->driver_table().Append(driver_id, driver_id1, data1, nullptr)); // Add a log entry at a second key. DriverID driver_id2 = DriverID::from_random(); std::vector driver_ids2 = {"jkl", "mno", "pqr"}; auto data2 = std::make_shared(); data2->driver_id = driver_ids2[0]; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id2, data2, nullptr)); + RAY_CHECK_OK(client->driver_table().Append(driver_id, driver_id2, data2, nullptr)); // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. @@ -829,32 +836,32 @@ void TestLogSubscribeId(const JobID &job_id, // The callback for subscription success. Once we've subscribed, request // notifications for only one of the keys, then write to both keys. - auto subscribe_callback = [job_id, driver_id1, driver_id2, driver_ids1, + auto subscribe_callback = [driver_id, driver_id1, driver_id2, driver_ids1, driver_ids2](gcs::AsyncGcsClient *client) { // Request notifications for one of the keys. RAY_CHECK_OK(client->driver_table().RequestNotifications( - job_id, driver_id2, client->client_table().GetLocalClientId())); + driver_id, driver_id2, client->client_table().GetLocalClientId())); // Write both keys. We should only receive notifications for the key that // we requested them for. auto remaining = std::vector(++driver_ids1.begin(), driver_ids1.end()); - for (const auto &driver_id : remaining) { + for (const auto &driver_id_it : remaining) { auto data = std::make_shared(); - data->driver_id = driver_id; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id1, data, nullptr)); + data->driver_id = driver_id_it; + RAY_CHECK_OK(client->driver_table().Append(driver_id, driver_id1, data, nullptr)); } remaining = std::vector(++driver_ids2.begin(), driver_ids2.end()); - for (const auto &driver_id : remaining) { + for (const auto &driver_id_it : remaining) { auto data = std::make_shared(); - data->driver_id = driver_id; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id2, data, nullptr)); + data->driver_id = driver_id_it; + RAY_CHECK_OK(client->driver_table().Append(driver_id, driver_id2, data, nullptr)); } }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. - RAY_CHECK_OK( - client->driver_table().Subscribe(job_id, client->client_table().GetLocalClientId(), - notification_callback, subscribe_callback)); + RAY_CHECK_OK(client->driver_table().Subscribe( + driver_id, client->client_table().GetLocalClientId(), notification_callback, + subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. test->Start(); @@ -865,24 +872,24 @@ void TestLogSubscribeId(const JobID &job_id, TEST_F(TestGcsWithAsio, TestLogSubscribeId) { test = this; - TestLogSubscribeId(job_id_, client_); + TestLogSubscribeId(driver_id_, client_); } -void TestSetSubscribeId(const JobID &job_id, +void TestSetSubscribeId(const DriverID &driver_id, std::shared_ptr client) { // Add a set entry. ObjectID object_id1 = ObjectID::from_random(); std::vector managers1 = {"abc", "def", "ghi"}; auto data1 = std::make_shared(); data1->manager = managers1[0]; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id1, data1, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id1, data1, nullptr)); // Add a set entry at a second key. ObjectID object_id2 = ObjectID::from_random(); std::vector managers2 = {"jkl", "mno", "pqr"}; auto data2 = std::make_shared(); data2->manager = managers2[0]; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id2, data2, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id2, data2, nullptr)); // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. @@ -905,32 +912,32 @@ void TestSetSubscribeId(const JobID &job_id, // The callback for subscription success. Once we've subscribed, request // notifications for only one of the keys, then write to both keys. - auto subscribe_callback = [job_id, object_id1, object_id2, managers1, + auto subscribe_callback = [driver_id, object_id1, object_id2, managers1, managers2](gcs::AsyncGcsClient *client) { // Request notifications for one of the keys. RAY_CHECK_OK(client->object_table().RequestNotifications( - job_id, object_id2, client->client_table().GetLocalClientId())); + driver_id, object_id2, client->client_table().GetLocalClientId())); // Write both keys. We should only receive notifications for the key that // we requested them for. auto remaining = std::vector(++managers1.begin(), managers1.end()); for (const auto &manager : remaining) { auto data = std::make_shared(); data->manager = manager; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id1, data, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id1, data, nullptr)); } remaining = std::vector(++managers2.begin(), managers2.end()); for (const auto &manager : remaining) { auto data = std::make_shared(); data->manager = manager; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id2, data, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id2, data, nullptr)); } }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. - RAY_CHECK_OK( - client->object_table().Subscribe(job_id, client->client_table().GetLocalClientId(), - notification_callback, subscribe_callback)); + RAY_CHECK_OK(client->object_table().Subscribe( + driver_id, client->client_table().GetLocalClientId(), notification_callback, + subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. test->Start(); @@ -941,17 +948,17 @@ void TestSetSubscribeId(const JobID &job_id, TEST_F(TestGcsWithAsio, TestSetSubscribeId) { test = this; - TestSetSubscribeId(job_id_, client_); + TestSetSubscribeId(driver_id_, client_); } -void TestTableSubscribeCancel(const JobID &job_id, +void TestTableSubscribeCancel(const DriverID &driver_id, std::shared_ptr client) { // Add a table entry. TaskID task_id = TaskID::from_random(); std::vector task_specs = {"jkl", "mno", "pqr"}; auto data = std::make_shared(); data->task_specification = task_specs[0]; - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id, data, nullptr)); + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id, data, nullptr)); // The failure callback should not be called since all keys are non-empty // when notifications are requested. @@ -979,31 +986,32 @@ void TestTableSubscribeCancel(const JobID &job_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. - auto subscribe_callback = [job_id, task_id, task_specs](gcs::AsyncGcsClient *client) { + auto subscribe_callback = [driver_id, task_id, + task_specs](gcs::AsyncGcsClient *client) { // Request notifications, then cancel immediately. We should receive a // notification for the current value at the key. RAY_CHECK_OK(client->raylet_task_table().RequestNotifications( - job_id, task_id, client->client_table().GetLocalClientId())); + driver_id, task_id, client->client_table().GetLocalClientId())); RAY_CHECK_OK(client->raylet_task_table().CancelNotifications( - job_id, task_id, client->client_table().GetLocalClientId())); + driver_id, task_id, client->client_table().GetLocalClientId())); // Write to the key. Since we canceled notifications, we should not receive // a notification for these writes. auto remaining = std::vector(++task_specs.begin(), task_specs.end()); for (const auto &task_spec : remaining) { auto data = std::make_shared(); data->task_specification = task_spec; - RAY_CHECK_OK(client->raylet_task_table().Add(job_id, task_id, data, nullptr)); + RAY_CHECK_OK(client->raylet_task_table().Add(driver_id, task_id, data, nullptr)); } // Request notifications again. We should receive a notification for the // current value at the key. RAY_CHECK_OK(client->raylet_task_table().RequestNotifications( - job_id, task_id, client->client_table().GetLocalClientId())); + driver_id, task_id, client->client_table().GetLocalClientId())); }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. RAY_CHECK_OK(client->raylet_task_table().Subscribe( - job_id, client->client_table().GetLocalClientId(), notification_callback, + driver_id, client->client_table().GetLocalClientId(), notification_callback, failure_callback, subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. @@ -1018,21 +1026,21 @@ TEST_MACRO(TestGcsWithAsio, TestTableSubscribeCancel); TEST_MACRO(TestGcsWithChainAsio, TestTableSubscribeCancel); #endif -void TestLogSubscribeCancel(const JobID &job_id, +void TestLogSubscribeCancel(const DriverID &driver_id, std::shared_ptr client) { // Add a log entry. - DriverID driver_id = DriverID::from_random(); + DriverID random_driver_id = DriverID::from_random(); std::vector driver_ids = {"jkl", "mno", "pqr"}; auto data = std::make_shared(); data->driver_id = driver_ids[0]; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id, data, nullptr)); + RAY_CHECK_OK(client->driver_table().Append(driver_id, random_driver_id, data, nullptr)); // The callback for a notification from the object table. This should only be // received for the object that we requested notifications for. - auto notification_callback = [driver_id, driver_ids]( + auto notification_callback = [random_driver_id, driver_ids]( gcs::AsyncGcsClient *client, const UniqueID &id, const std::vector &data) { - ASSERT_EQ(id, driver_id); + ASSERT_EQ(id, random_driver_id); // Check that we get a duplicate notification for the first write. We get a // duplicate notification because the log is append-only and notifications // are canceled after the first write, then requested again. @@ -1049,32 +1057,34 @@ void TestLogSubscribeCancel(const JobID &job_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. - auto subscribe_callback = [job_id, driver_id, driver_ids](gcs::AsyncGcsClient *client) { + auto subscribe_callback = [driver_id, random_driver_id, + driver_ids](gcs::AsyncGcsClient *client) { // Request notifications, then cancel immediately. We should receive a // notification for the current value at the key. RAY_CHECK_OK(client->driver_table().RequestNotifications( - job_id, driver_id, client->client_table().GetLocalClientId())); + driver_id, random_driver_id, client->client_table().GetLocalClientId())); RAY_CHECK_OK(client->driver_table().CancelNotifications( - job_id, driver_id, client->client_table().GetLocalClientId())); + driver_id, random_driver_id, client->client_table().GetLocalClientId())); // Append to the key. Since we canceled notifications, we should not // receive a notification for these writes. auto remaining = std::vector(++driver_ids.begin(), driver_ids.end()); for (const auto &remaining_driver_id : remaining) { auto data = std::make_shared(); data->driver_id = remaining_driver_id; - RAY_CHECK_OK(client->driver_table().Append(job_id, driver_id, data, nullptr)); + RAY_CHECK_OK( + client->driver_table().Append(driver_id, random_driver_id, data, nullptr)); } // Request notifications again. We should receive a notification for the // current values at the key. RAY_CHECK_OK(client->driver_table().RequestNotifications( - job_id, driver_id, client->client_table().GetLocalClientId())); + driver_id, random_driver_id, client->client_table().GetLocalClientId())); }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. - RAY_CHECK_OK( - client->driver_table().Subscribe(job_id, client->client_table().GetLocalClientId(), - notification_callback, subscribe_callback)); + RAY_CHECK_OK(client->driver_table().Subscribe( + driver_id, client->client_table().GetLocalClientId(), notification_callback, + subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. test->Start(); @@ -1086,17 +1096,17 @@ void TestLogSubscribeCancel(const JobID &job_id, TEST_F(TestGcsWithAsio, TestLogSubscribeCancel) { test = this; - TestLogSubscribeCancel(job_id_, client_); + TestLogSubscribeCancel(driver_id_, client_); } -void TestSetSubscribeCancel(const JobID &job_id, +void TestSetSubscribeCancel(const DriverID &driver_id, std::shared_ptr client) { // Add a set entry. ObjectID object_id = ObjectID::from_random(); std::vector managers = {"jkl", "mno", "pqr"}; auto data = std::make_shared(); data->manager = managers[0]; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id, data, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id, data, nullptr)); // The callback for a notification from the object table. This should only be // received for the object that we requested notifications for. @@ -1131,32 +1141,33 @@ void TestSetSubscribeCancel(const JobID &job_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. - auto subscribe_callback = [job_id, object_id, managers](gcs::AsyncGcsClient *client) { + auto subscribe_callback = [driver_id, object_id, + managers](gcs::AsyncGcsClient *client) { // Request notifications, then cancel immediately. We should receive a // notification for the current value at the key. RAY_CHECK_OK(client->object_table().RequestNotifications( - job_id, object_id, client->client_table().GetLocalClientId())); + driver_id, object_id, client->client_table().GetLocalClientId())); RAY_CHECK_OK(client->object_table().CancelNotifications( - job_id, object_id, client->client_table().GetLocalClientId())); + driver_id, object_id, client->client_table().GetLocalClientId())); // Add to the key. Since we canceled notifications, we should not // receive a notification for these writes. auto remaining = std::vector(++managers.begin(), managers.end()); for (const auto &manager : remaining) { auto data = std::make_shared(); data->manager = manager; - RAY_CHECK_OK(client->object_table().Add(job_id, object_id, data, nullptr)); + RAY_CHECK_OK(client->object_table().Add(driver_id, object_id, data, nullptr)); } // Request notifications again. We should receive a notification for the // current values at the key. RAY_CHECK_OK(client->object_table().RequestNotifications( - job_id, object_id, client->client_table().GetLocalClientId())); + driver_id, object_id, client->client_table().GetLocalClientId())); }; // Subscribe to notifications for this client. This allows us to request and // receive notifications for specific keys. - RAY_CHECK_OK( - client->object_table().Subscribe(job_id, client->client_table().GetLocalClientId(), - notification_callback, subscribe_callback)); + RAY_CHECK_OK(client->object_table().Subscribe( + driver_id, client->client_table().GetLocalClientId(), notification_callback, + subscribe_callback)); // Run the event loop. The loop will only stop if the registered subscription // callback is called for the requested key. test->Start(); @@ -1168,7 +1179,7 @@ void TestSetSubscribeCancel(const JobID &job_id, TEST_F(TestGcsWithAsio, TestSetSubscribeCancel) { test = this; - TestSetSubscribeCancel(job_id_, client_); + TestSetSubscribeCancel(driver_id_, client_); } void ClientTableNotification(gcs::AsyncGcsClient *client, const ClientID &client_id, @@ -1185,7 +1196,7 @@ void ClientTableNotification(gcs::AsyncGcsClient *client, const ClientID &client ASSERT_EQ(cached_client.is_insertion, is_insertion); } -void TestClientTableConnect(const JobID &job_id, +void TestClientTableConnect(const DriverID &driver_id, std::shared_ptr client) { // Register callbacks for when a client gets added and removed. The latter // event will stop the event loop. @@ -1207,10 +1218,10 @@ void TestClientTableConnect(const JobID &job_id, TEST_F(TestGcsWithAsio, TestClientTableConnect) { test = this; - TestClientTableConnect(job_id_, client_); + TestClientTableConnect(driver_id_, client_); } -void TestClientTableDisconnect(const JobID &job_id, +void TestClientTableDisconnect(const DriverID &driver_id, std::shared_ptr client) { // Register callbacks for when a client gets added and removed. The latter // event will stop the event loop. @@ -1238,10 +1249,10 @@ void TestClientTableDisconnect(const JobID &job_id, TEST_F(TestGcsWithAsio, TestClientTableDisconnect) { test = this; - TestClientTableDisconnect(job_id_, client_); + TestClientTableDisconnect(driver_id_, client_); } -void TestClientTableImmediateDisconnect(const JobID &job_id, +void TestClientTableImmediateDisconnect(const DriverID &driver_id, std::shared_ptr client) { // Register callbacks for when a client gets added and removed. The latter // event will stop the event loop. @@ -1267,10 +1278,10 @@ void TestClientTableImmediateDisconnect(const JobID &job_id, TEST_F(TestGcsWithAsio, TestClientTableImmediateDisconnect) { test = this; - TestClientTableImmediateDisconnect(job_id_, client_); + TestClientTableImmediateDisconnect(driver_id_, client_); } -void TestClientTableMarkDisconnected(const JobID &job_id, +void TestClientTableMarkDisconnected(const DriverID &driver_id, std::shared_ptr client) { ClientTableDataT local_client_info = client->client_table().GetLocalClient(); local_client_info.node_manager_address = "127.0.0.1"; @@ -1293,7 +1304,7 @@ void TestClientTableMarkDisconnected(const JobID &job_id, TEST_F(TestGcsWithAsio, TestClientTableMarkDisconnected) { test = this; - TestClientTableMarkDisconnected(job_id_, client_); + TestClientTableMarkDisconnected(driver_id_, client_); } #undef TEST_MACRO diff --git a/src/ray/gcs/format/gcs.fbs b/src/ray/gcs/format/gcs.fbs index dbd181d9b..7acb24d27 100644 --- a/src/ray/gcs/format/gcs.fbs +++ b/src/ray/gcs/format/gcs.fbs @@ -202,8 +202,8 @@ table ActorTableData { } table ErrorTableData { - // The ID of the job that the error is for. - job_id: string; + // The ID of the driver that the error is for. + driver_id: string; // The type of the error. type: string; // The error message. diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index 9b41c9460..93e40a392 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -38,7 +38,7 @@ namespace ray { namespace gcs { template -Status Log::Append(const JobID &job_id, const ID &id, +Status Log::Append(const DriverID &driver_id, const ID &id, std::shared_ptr &dataT, const WriteCallback &done) { num_appends_++; auto callback = [this, id, dataT, done](const std::string &data) { @@ -59,7 +59,7 @@ Status Log::Append(const JobID &job_id, const ID &id, } template -Status Log::AppendAt(const JobID &job_id, const ID &id, +Status Log::AppendAt(const DriverID &driver_id, const ID &id, std::shared_ptr &dataT, const WriteCallback &done, const WriteCallback &failure, int log_length) { num_appends_++; @@ -84,7 +84,8 @@ Status Log::AppendAt(const JobID &job_id, const ID &id, } template -Status Log::Lookup(const JobID &job_id, const ID &id, const Callback &lookup) { +Status Log::Lookup(const DriverID &driver_id, const ID &id, + const Callback &lookup) { num_lookups_++; auto callback = [this, id, lookup](const std::string &data) { if (lookup != nullptr) { @@ -109,7 +110,7 @@ Status Log::Lookup(const JobID &job_id, const ID &id, const Callback & } template -Status Log::Subscribe(const JobID &job_id, const ClientID &client_id, +Status Log::Subscribe(const DriverID &driver_id, const ClientID &client_id, const Callback &subscribe, const SubscriptionCallback &done) { auto subscribe_wrapper = [subscribe](AsyncGcsClient *client, const ID &id, @@ -118,11 +119,11 @@ Status Log::Subscribe(const JobID &job_id, const ClientID &client_id, RAY_CHECK(notification_mode != GcsTableNotificationMode::REMOVE); subscribe(client, id, data); }; - return Subscribe(job_id, client_id, subscribe_wrapper, done); + return Subscribe(driver_id, client_id, subscribe_wrapper, done); } template -Status Log::Subscribe(const JobID &job_id, const ClientID &client_id, +Status Log::Subscribe(const DriverID &driver_id, const ClientID &client_id, const NotificationCallback &subscribe, const SubscriptionCallback &done) { RAY_CHECK(subscribe_callback_index_ == -1) @@ -166,7 +167,7 @@ Status Log::Subscribe(const JobID &job_id, const ClientID &client_id, } template -Status Log::RequestNotifications(const JobID &job_id, const ID &id, +Status Log::RequestNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id) { RAY_CHECK(subscribe_callback_index_ >= 0) << "Client requested notifications on a key before Subscribe completed"; @@ -176,7 +177,7 @@ Status Log::RequestNotifications(const JobID &job_id, const ID &id, } template -Status Log::CancelNotifications(const JobID &job_id, const ID &id, +Status Log::CancelNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id) { RAY_CHECK(subscribe_callback_index_ >= 0) << "Client canceled notifications on a key before Subscribe completed"; @@ -186,7 +187,7 @@ Status Log::CancelNotifications(const JobID &job_id, const ID &id, } template -void Log::Delete(const JobID &job_id, const std::vector &ids) { +void Log::Delete(const DriverID &driver_id, const std::vector &ids) { if (ids.empty()) { return; } @@ -210,8 +211,8 @@ void Log::Delete(const JobID &job_id, const std::vector &ids) { } template -void Log::Delete(const JobID &job_id, const ID &id) { - Delete(job_id, std::vector({id})); +void Log::Delete(const DriverID &driver_id, const ID &id) { + Delete(driver_id, std::vector({id})); } template @@ -222,7 +223,7 @@ std::string Log::DebugString() const { } template -Status Table::Add(const JobID &job_id, const ID &id, +Status Table::Add(const DriverID &driver_id, const ID &id, std::shared_ptr &dataT, const WriteCallback &done) { num_adds_++; auto callback = [this, id, dataT, done](const std::string &data) { @@ -240,10 +241,10 @@ Status Table::Add(const JobID &job_id, const ID &id, } template -Status Table::Lookup(const JobID &job_id, const ID &id, const Callback &lookup, - const FailureCallback &failure) { +Status Table::Lookup(const DriverID &driver_id, const ID &id, + const Callback &lookup, const FailureCallback &failure) { num_lookups_++; - return Log::Lookup(job_id, id, + return Log::Lookup(driver_id, id, [lookup, failure](AsyncGcsClient *client, const ID &id, const std::vector &data) { if (data.empty()) { @@ -260,12 +261,12 @@ Status Table::Lookup(const JobID &job_id, const ID &id, const Callback } template -Status Table::Subscribe(const JobID &job_id, const ClientID &client_id, +Status Table::Subscribe(const DriverID &driver_id, const ClientID &client_id, const Callback &subscribe, const FailureCallback &failure, const SubscriptionCallback &done) { return Log::Subscribe( - job_id, client_id, + driver_id, client_id, [subscribe, failure](AsyncGcsClient *client, const ID &id, const std::vector &data) { RAY_CHECK(data.empty() || data.size() == 1); @@ -288,7 +289,7 @@ std::string Table::DebugString() const { } template -Status Set::Add(const JobID &job_id, const ID &id, +Status Set::Add(const DriverID &driver_id, const ID &id, std::shared_ptr &dataT, const WriteCallback &done) { num_adds_++; auto callback = [this, id, dataT, done](const std::string &data) { @@ -306,7 +307,7 @@ Status Set::Add(const JobID &job_id, const ID &id, } template -Status Set::Remove(const JobID &job_id, const ID &id, +Status Set::Remove(const DriverID &driver_id, const ID &id, std::shared_ptr &dataT, const WriteCallback &done) { num_removes_++; auto callback = [this, id, dataT, done](const std::string &data) { @@ -334,11 +335,11 @@ std::string Set::DebugString() const { Status ErrorTable::PushErrorToDriver(const DriverID &driver_id, const std::string &type, const std::string &error_message, double timestamp) { auto data = std::make_shared(); - data->job_id = driver_id.binary(); + data->driver_id = driver_id.binary(); data->type = type; data->error_message = error_message; data->timestamp = timestamp; - return Append(JobID(driver_id), driver_id, data, /*done_callback=*/nullptr); + return Append(DriverID(driver_id), driver_id, data, /*done_callback=*/nullptr); } std::string ErrorTable::DebugString() const { @@ -351,7 +352,7 @@ Status ProfileTable::AddProfileEventBatch(const ProfileTableData &profile_events // call "Pack" and undo the "UnPack". profile_events.UnPackTo(data.get()); - return Append(JobID::nil(), UniqueID::from_random(), data, + return Append(DriverID::nil(), UniqueID::from_random(), data, /*done_callback=*/nullptr); } @@ -363,7 +364,7 @@ Status DriverTable::AppendDriverData(const DriverID &driver_id, bool is_dead) { auto data = std::make_shared(); data->driver_id = driver_id.binary(); data->is_dead = is_dead; - return Append(JobID(driver_id), driver_id, data, /*done_callback=*/nullptr); + return Append(DriverID(driver_id), driver_id, data, /*done_callback=*/nullptr); } void ClientTable::RegisterClientAddedCallback(const ClientTableCallback &callback) { @@ -494,13 +495,13 @@ Status ClientTable::Connect(const ClientTableDataT &local_client) { // Callback to request notifications from the client table once we've // successfully subscribed. auto subscription_callback = [this](AsyncGcsClient *c) { - RAY_CHECK_OK(RequestNotifications(JobID::nil(), client_log_key_, client_id_)); + RAY_CHECK_OK(RequestNotifications(DriverID::nil(), client_log_key_, client_id_)); }; // Subscribe to the client table. - RAY_CHECK_OK(Subscribe(JobID::nil(), client_id_, notification_callback, + RAY_CHECK_OK(Subscribe(DriverID::nil(), client_id_, notification_callback, subscription_callback)); }; - return Append(JobID::nil(), client_log_key_, data, add_callback); + return Append(DriverID::nil(), client_log_key_, data, add_callback); } Status ClientTable::Disconnect(const DisconnectCallback &callback) { @@ -509,12 +510,12 @@ Status ClientTable::Disconnect(const DisconnectCallback &callback) { auto add_callback = [this, callback](AsyncGcsClient *client, const ClientID &id, const ClientTableDataT &data) { HandleConnected(client, data); - RAY_CHECK_OK(CancelNotifications(JobID::nil(), client_log_key_, id)); + RAY_CHECK_OK(CancelNotifications(DriverID::nil(), client_log_key_, id)); if (callback != nullptr) { callback(); } }; - RAY_RETURN_NOT_OK(Append(JobID::nil(), client_log_key_, data, add_callback)); + RAY_RETURN_NOT_OK(Append(DriverID::nil(), client_log_key_, data, add_callback)); // We successfully added the deletion entry. Mark ourselves as disconnected. disconnected_ = true; return Status::OK(); @@ -524,7 +525,7 @@ ray::Status ClientTable::MarkDisconnected(const ClientID &dead_client_id) { auto data = std::make_shared(); data->client_id = dead_client_id.binary(); data->is_insertion = false; - return Append(JobID::nil(), client_log_key_, data, nullptr); + return Append(DriverID::nil(), client_log_key_, data, nullptr); } void ClientTable::GetClient(const ClientID &client_id, @@ -544,7 +545,7 @@ const std::unordered_map &ClientTable::GetAllClients Status ClientTable::Lookup(const Callback &lookup) { RAY_CHECK(lookup != nullptr); - return Log::Lookup(JobID::nil(), client_log_key_, lookup); + return Log::Lookup(DriverID::nil(), client_log_key_, lookup); } std::string ClientTable::DebugString() const { @@ -555,10 +556,10 @@ std::string ClientTable::DebugString() const { return result.str(); } -Status ActorCheckpointIdTable::AddCheckpointId(const JobID &job_id, +Status ActorCheckpointIdTable::AddCheckpointId(const DriverID &driver_id, const ActorID &actor_id, const ActorCheckpointID &checkpoint_id) { - auto lookup_callback = [this, checkpoint_id, job_id, actor_id]( + auto lookup_callback = [this, checkpoint_id, driver_id, actor_id]( ray::gcs::AsyncGcsClient *client, const UniqueID &id, const ActorCheckpointIdDataT &data) { std::shared_ptr copy = @@ -574,20 +575,20 @@ Status ActorCheckpointIdTable::AddCheckpointId(const JobID &job_id, << actor_id; copy->timestamps.erase(copy->timestamps.begin()); copy->checkpoint_ids.erase(0, kUniqueIDSize); - client_->actor_checkpoint_table().Delete(job_id, checkpoint_id); + client_->actor_checkpoint_table().Delete(driver_id, checkpoint_id); } - RAY_CHECK_OK(Add(job_id, actor_id, copy, nullptr)); + RAY_CHECK_OK(Add(driver_id, actor_id, copy, nullptr)); }; - auto failure_callback = [this, checkpoint_id, job_id, actor_id]( + auto failure_callback = [this, checkpoint_id, driver_id, actor_id]( ray::gcs::AsyncGcsClient *client, const UniqueID &id) { std::shared_ptr data = std::make_shared(); data->actor_id = id.binary(); data->timestamps.push_back(current_sys_time_ms()); data->checkpoint_ids = checkpoint_id.binary(); - RAY_CHECK_OK(Add(job_id, actor_id, data, nullptr)); + RAY_CHECK_OK(Add(driver_id, actor_id, data, nullptr)); }; - return Lookup(job_id, actor_id, lookup_callback, failure_callback); + return Lookup(driver_id, actor_id, lookup_callback, failure_callback); } template class Log; diff --git a/src/ray/gcs/tables.h b/src/ray/gcs/tables.h index 54f7a68da..b22910832 100644 --- a/src/ray/gcs/tables.h +++ b/src/ray/gcs/tables.h @@ -38,9 +38,9 @@ enum class CommandType { kRegular, kChain }; template class PubsubInterface { public: - virtual Status RequestNotifications(const JobID &job_id, const ID &id, + virtual Status RequestNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id) = 0; - virtual Status CancelNotifications(const JobID &job_id, const ID &id, + virtual Status CancelNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id) = 0; virtual ~PubsubInterface(){}; }; @@ -51,9 +51,9 @@ class LogInterface { using DataT = typename Data::NativeTableType; using WriteCallback = std::function; - virtual Status Append(const JobID &job_id, const ID &id, std::shared_ptr &data, - const WriteCallback &done) = 0; - virtual Status AppendAt(const JobID &job_id, const ID &task_id, + virtual Status Append(const DriverID &driver_id, const ID &id, + std::shared_ptr &data, const WriteCallback &done) = 0; + virtual Status AppendAt(const DriverID &driver_id, const ID &task_id, std::shared_ptr &data, const WriteCallback &done, const WriteCallback &failure, int log_length) = 0; virtual ~LogInterface(){}; @@ -104,20 +104,20 @@ class Log : public LogInterface, virtual public PubsubInterface { /// Append a log entry to a key. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is added to the GCS. /// \param data Data to append to the log. TODO(rkn): This can be made const, /// right? /// \param done Callback that is called once the data has been written to the /// GCS. /// \return Status - Status Append(const JobID &job_id, const ID &id, std::shared_ptr &data, + Status Append(const DriverID &driver_id, const ID &id, std::shared_ptr &data, const WriteCallback &done); /// Append a log entry to a key if and only if the log has the given number /// of entries. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is added to the GCS. /// \param data Data to append to the log. /// \param done Callback that is called if the data was appended to the log. @@ -126,25 +126,25 @@ class Log : public LogInterface, virtual public PubsubInterface { /// \param log_length The number of entries that the log must have for the /// append to succeed. /// \return Status - Status AppendAt(const JobID &job_id, const ID &id, std::shared_ptr &data, + Status AppendAt(const DriverID &driver_id, const ID &id, std::shared_ptr &data, const WriteCallback &done, const WriteCallback &failure, int log_length); /// Lookup the log values at a key asynchronously. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is looked up in the GCS. /// \param lookup Callback that is called after lookup. If the callback is /// called with an empty vector, then there was no data at the key. /// \return Status - Status Lookup(const JobID &job_id, const ID &id, const Callback &lookup); + Status Lookup(const DriverID &driver_id, const ID &id, const Callback &lookup); /// Subscribe to any Append operations to this table. The caller may choose /// to subscribe to all Appends, or to subscribe only to keys that it /// requests notifications for. This may only be called once per Log /// instance. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param client_id The type of update to listen to. If this is nil, then a /// message for each Add to the table will be received. Else, only /// messages for the given client will be received. In the latter @@ -155,7 +155,7 @@ class Log : public LogInterface, virtual public PubsubInterface { /// \param done Callback that is called when subscription is complete and we /// are ready to receive messages. /// \return Status - Status Subscribe(const JobID &job_id, const ClientID &client_id, + Status Subscribe(const DriverID &driver_id, const ClientID &client_id, const Callback &subscribe, const SubscriptionCallback &done); /// Request notifications about a key in this table. @@ -167,37 +167,37 @@ class Log : public LogInterface, virtual public PubsubInterface { /// notifications can be requested, the caller must first call `Subscribe`, /// with the same `client_id`. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the key to request notifications for. /// \param client_id The client who is requesting notifications. Before /// notifications can be requested, a call to `Subscribe` to this /// table with the same `client_id` must complete successfully. /// \return Status - Status RequestNotifications(const JobID &job_id, const ID &id, + Status RequestNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id); /// Cancel notifications about a key in this table. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the key to request notifications for. /// \param client_id The client who originally requested notifications. /// \return Status - Status CancelNotifications(const JobID &job_id, const ID &id, + Status CancelNotifications(const DriverID &driver_id, const ID &id, const ClientID &client_id); /// Delete an entire key from redis. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data to delete from the GCS. /// \return Void. - void Delete(const JobID &job_id, const ID &id); + void Delete(const DriverID &driver_id, const ID &id); /// Delete several keys from redis. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param ids The vector of IDs to delete from the GCS. /// \return Void. - void Delete(const JobID &job_id, const std::vector &ids); + void Delete(const DriverID &driver_id, const std::vector &ids); /// Returns debug string for class. /// @@ -217,7 +217,7 @@ class Log : public LogInterface, virtual public PubsubInterface { /// an additional parameter notification_mode in NotificationCallback. Therefore this /// function supports notifications of remove operations. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param client_id The type of update to listen to. If this is nil, then a /// message for each Add to the table will be received. Else, only /// messages for the given client will be received. In the latter @@ -228,7 +228,7 @@ class Log : public LogInterface, virtual public PubsubInterface { /// \param done Callback that is called when subscription is complete and we /// are ready to receive messages. /// \return Status - Status Subscribe(const JobID &job_id, const ClientID &client_id, + Status Subscribe(const DriverID &driver_id, const ClientID &client_id, const NotificationCallback &subscribe, const SubscriptionCallback &done); @@ -261,8 +261,8 @@ class TableInterface { public: using DataT = typename Data::NativeTableType; using WriteCallback = typename Log::WriteCallback; - virtual Status Add(const JobID &job_id, const ID &task_id, std::shared_ptr &data, - const WriteCallback &done) = 0; + virtual Status Add(const DriverID &driver_id, const ID &task_id, + std::shared_ptr &data, const WriteCallback &done) = 0; virtual ~TableInterface(){}; }; @@ -299,32 +299,32 @@ class Table : private Log, /// Add an entry to the table. This overwrites any existing data at the key. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is added to the GCS. /// \param data Data that is added to the GCS. /// \param done Callback that is called once the data has been written to the /// GCS. /// \return Status - Status Add(const JobID &job_id, const ID &id, std::shared_ptr &data, + Status Add(const DriverID &driver_id, const ID &id, std::shared_ptr &data, const WriteCallback &done); /// Lookup an entry asynchronously. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is looked up in the GCS. /// \param lookup Callback that is called after lookup if there was data the /// key. /// \param failure Callback that is called after lookup if there was no data /// at the key. /// \return Status - Status Lookup(const JobID &job_id, const ID &id, const Callback &lookup, + Status Lookup(const DriverID &driver_id, const ID &id, const Callback &lookup, const FailureCallback &failure); /// Subscribe to any Add operations to this table. The caller may choose to /// subscribe to all Adds, or to subscribe only to keys that it requests /// notifications for. This may only be called once per Table instance. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param client_id The type of update to listen to. If this is nil, then a /// message for each Add to the table will be received. Else, only /// messages for the given client will be received. In the latter @@ -337,14 +337,16 @@ class Table : private Log, /// \param done Callback that is called when subscription is complete and we /// are ready to receive messages. /// \return Status - Status Subscribe(const JobID &job_id, const ClientID &client_id, + Status Subscribe(const DriverID &driver_id, const ClientID &client_id, const Callback &subscribe, const FailureCallback &failure, const SubscriptionCallback &done); - void Delete(const JobID &job_id, const ID &id) { Log::Delete(job_id, id); } + void Delete(const DriverID &driver_id, const ID &id) { + Log::Delete(driver_id, id); + } - void Delete(const JobID &job_id, const std::vector &ids) { - Log::Delete(job_id, ids); + void Delete(const DriverID &driver_id, const std::vector &ids) { + Log::Delete(driver_id, ids); } /// Returns debug string for class. @@ -369,10 +371,10 @@ class SetInterface { public: using DataT = typename Data::NativeTableType; using WriteCallback = typename Log::WriteCallback; - virtual Status Add(const JobID &job_id, const ID &id, std::shared_ptr &data, - const WriteCallback &done) = 0; - virtual Status Remove(const JobID &job_id, const ID &id, std::shared_ptr &data, - const WriteCallback &done) = 0; + virtual Status Add(const DriverID &driver_id, const ID &id, + std::shared_ptr &data, const WriteCallback &done) = 0; + virtual Status Remove(const DriverID &driver_id, const ID &id, + std::shared_ptr &data, const WriteCallback &done) = 0; virtual ~SetInterface(){}; }; @@ -406,30 +408,30 @@ class Set : private Log, /// Add an entry to the set. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is added to the GCS. /// \param data Data to add to the set. /// \param done Callback that is called once the data has been written to the /// GCS. /// \return Status - Status Add(const JobID &job_id, const ID &id, std::shared_ptr &data, + Status Add(const DriverID &driver_id, const ID &id, std::shared_ptr &data, const WriteCallback &done); /// Remove an entry from the set. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param id The ID of the data that is removed from the GCS. /// \param data Data to remove from the set. /// \param done Callback that is called once the data has been written to the /// GCS. /// \return Status - Status Remove(const JobID &job_id, const ID &id, std::shared_ptr &data, + Status Remove(const DriverID &driver_id, const ID &id, std::shared_ptr &data, const WriteCallback &done); - Status Subscribe(const JobID &job_id, const ClientID &client_id, + Status Subscribe(const DriverID &driver_id, const ClientID &client_id, const NotificationCallback &subscribe, const SubscriptionCallback &done) { - return Log::Subscribe(job_id, client_id, subscribe, done); + return Log::Subscribe(driver_id, client_id, subscribe, done); } /// Returns debug string for class. @@ -547,9 +549,9 @@ class TaskLeaseTable : public Table { prefix_ = TablePrefix::TASK_LEASE; } - Status Add(const JobID &job_id, const TaskID &id, std::shared_ptr &data, - const WriteCallback &done) override { - RAY_RETURN_NOT_OK((Table::Add(job_id, id, data, done))); + Status Add(const DriverID &driver_id, const TaskID &id, + std::shared_ptr &data, const WriteCallback &done) override { + RAY_RETURN_NOT_OK((Table::Add(driver_id, id, data, done))); // Mark the entry for expiration in Redis. It's okay if this command fails // since the lease entry itself contains the expiration period. In the // worst case, if the command fails, then a client that looks up the lease @@ -584,11 +586,11 @@ class ActorCheckpointIdTable : public Table { /// Add a checkpoint id to an actor, and remove a previous checkpoint if the /// total number of checkpoints in GCS exceeds the max allowed value. /// - /// \param job_id The ID of the job (= driver). + /// \param driver_id The ID of the job (= driver). /// \param actor_id ID of the actor. /// \param checkpoint_id ID of the checkpoint. /// \return Status. - Status AddCheckpointId(const JobID &job_id, const ActorID &actor_id, + Status AddCheckpointId(const DriverID &driver_id, const ActorID &actor_id, const ActorCheckpointID &checkpoint_id); }; @@ -627,7 +629,7 @@ class ErrorTable : private Log { /// duplicate messages currently cause failures (the GCS doesn't allow it). A /// natural way to do this is to have finer-grained time stamps. /// - /// \param job_id The ID of the job that generated the error. If the error + /// \param driver_id The ID of the job that generated the error. If the error /// should be pushed to all jobs, then this should be nil. /// \param type The type of the error. /// \param error_message The error message to push. diff --git a/src/ray/id_def.h b/src/ray/id_def.h index 8e8b2b3fb..b63bc9947 100644 --- a/src/ray/id_def.h +++ b/src/ray/id_def.h @@ -5,7 +5,6 @@ // NOTE: This file should NOT be included in any file other than id.h. DEFINE_UNIQUE_ID(TaskID); -DEFINE_UNIQUE_ID(JobID); DEFINE_UNIQUE_ID(ObjectID); DEFINE_UNIQUE_ID(FunctionID); DEFINE_UNIQUE_ID(ActorClassID); diff --git a/src/ray/object_manager/object_directory.cc b/src/ray/object_manager/object_directory.cc index f9ec35365..99ed0851c 100644 --- a/src/ray/object_manager/object_directory.cc +++ b/src/ray/object_manager/object_directory.cc @@ -71,7 +71,7 @@ void ObjectDirectory::RegisterBackend() { } }; RAY_CHECK_OK(gcs_client_->object_table().Subscribe( - JobID::nil(), gcs_client_->client_table().GetLocalClientId(), + DriverID::nil(), gcs_client_->client_table().GetLocalClientId(), object_notification_callback, nullptr)); } @@ -84,7 +84,7 @@ ray::Status ObjectDirectory::ReportObjectAdded( data->manager = client_id.binary(); data->object_size = object_info.data_size; ray::Status status = - gcs_client_->object_table().Add(JobID::nil(), object_id, data, nullptr); + gcs_client_->object_table().Add(DriverID::nil(), object_id, data, nullptr); return status; } @@ -97,7 +97,7 @@ ray::Status ObjectDirectory::ReportObjectRemoved( data->manager = client_id.binary(); data->object_size = object_info.data_size; ray::Status status = - gcs_client_->object_table().Remove(JobID::nil(), object_id, data, nullptr); + gcs_client_->object_table().Remove(DriverID::nil(), object_id, data, nullptr); return status; }; @@ -157,7 +157,7 @@ ray::Status ObjectDirectory::SubscribeObjectLocations(const UniqueID &callback_i if (it == listeners_.end()) { it = listeners_.emplace(object_id, LocationListenerState()).first; status = gcs_client_->object_table().RequestNotifications( - JobID::nil(), object_id, gcs_client_->client_table().GetLocalClientId()); + DriverID::nil(), object_id, gcs_client_->client_table().GetLocalClientId()); } auto &listener_state = it->second; // TODO(hme): Make this fatal after implementing Pull suppression. @@ -185,7 +185,7 @@ ray::Status ObjectDirectory::UnsubscribeObjectLocations(const UniqueID &callback entry->second.callbacks.erase(callback_id); if (entry->second.callbacks.empty()) { status = gcs_client_->object_table().CancelNotifications( - JobID::nil(), object_id, gcs_client_->client_table().GetLocalClientId()); + DriverID::nil(), object_id, gcs_client_->client_table().GetLocalClientId()); listeners_.erase(entry); } return status; @@ -208,7 +208,7 @@ ray::Status ObjectDirectory::LookupLocations(const ObjectID &object_id, // SubscribeObjectLocations call, so look up the object's locations // directly from the GCS. status = gcs_client_->object_table().Lookup( - JobID::nil(), object_id, + DriverID::nil(), object_id, [this, callback](gcs::AsyncGcsClient *client, const ObjectID &object_id, const std::vector &location_updates) { // Build the set of current locations based on the entries in the log. diff --git a/src/ray/raylet/lineage_cache.cc b/src/ray/raylet/lineage_cache.cc index 949dc9eca..57761cd06 100644 --- a/src/ray/raylet/lineage_cache.cc +++ b/src/ray/raylet/lineage_cache.cc @@ -359,7 +359,7 @@ void LineageCache::FlushTask(const TaskID &task_id) { auto root = flatbuffers::GetRoot(fbb.GetBufferPointer()); root->UnPackTo(task_data.get()); RAY_CHECK_OK( - task_storage_.Add(JobID(task->TaskData().GetTaskSpecification().DriverId()), + task_storage_.Add(DriverID(task->TaskData().GetTaskSpecification().DriverId()), task_id, task_data, task_callback)); // We successfully wrote the task, so mark it as committing. @@ -373,7 +373,7 @@ bool LineageCache::SubscribeTask(const TaskID &task_id) { if (unsubscribed) { // Request notifications for the task if we haven't already requested // notifications for it. - RAY_CHECK_OK(task_pubsub_.RequestNotifications(JobID::nil(), task_id, client_id_)); + RAY_CHECK_OK(task_pubsub_.RequestNotifications(DriverID::nil(), task_id, client_id_)); } // Return whether we were previously unsubscribed to this task and are now // subscribed. @@ -386,7 +386,7 @@ bool LineageCache::UnsubscribeTask(const TaskID &task_id) { if (subscribed) { // Cancel notifications for the task if we previously requested // notifications for it. - RAY_CHECK_OK(task_pubsub_.CancelNotifications(JobID::nil(), task_id, client_id_)); + RAY_CHECK_OK(task_pubsub_.CancelNotifications(DriverID::nil(), task_id, client_id_)); subscribed_tasks_.erase(it); } // Return whether we were previously subscribed to this task and are now diff --git a/src/ray/raylet/lineage_cache_test.cc b/src/ray/raylet/lineage_cache_test.cc index 1ed0dcc84..af411066e 100644 --- a/src/ray/raylet/lineage_cache_test.cc +++ b/src/ray/raylet/lineage_cache_test.cc @@ -22,7 +22,7 @@ class MockGcs : public gcs::TableInterface, notification_callback_ = notification_callback; } - Status Add(const JobID &job_id, const TaskID &task_id, + Status Add(const DriverID &driver_id, const TaskID &task_id, std::shared_ptr &task_data, const gcs::TableInterface::WriteCallback &done) { task_table_[task_id] = task_data; @@ -43,10 +43,10 @@ class MockGcs : public gcs::TableInterface, notification_callback_(client, task_id, data); } }; - return Add(JobID::nil(), task_id, task_data, callback); + return Add(DriverID::nil(), task_id, task_data, callback); } - Status RequestNotifications(const JobID &job_id, const TaskID &task_id, + Status RequestNotifications(const DriverID &driver_id, const TaskID &task_id, const ClientID &client_id) { subscribed_tasks_.insert(task_id); if (task_table_.count(task_id) == 1) { @@ -56,7 +56,7 @@ class MockGcs : public gcs::TableInterface, return ray::Status::OK(); } - Status CancelNotifications(const JobID &job_id, const TaskID &task_id, + Status CancelNotifications(const DriverID &driver_id, const TaskID &task_id, const ClientID &client_id) { subscribed_tasks_.erase(task_id); return ray::Status::OK(); diff --git a/src/ray/raylet/monitor.cc b/src/ray/raylet/monitor.cc index d18edbad8..51e035b1b 100644 --- a/src/ray/raylet/monitor.cc +++ b/src/ray/raylet/monitor.cc @@ -35,7 +35,7 @@ void Monitor::Start() { HandleHeartbeat(id, heartbeat_data); }; RAY_CHECK_OK(gcs_client_.heartbeat_table().Subscribe( - JobID::nil(), ClientID::nil(), heartbeat_callback, nullptr, nullptr)); + DriverID::nil(), ClientID::nil(), heartbeat_callback, nullptr, nullptr)); Tick(); } @@ -67,7 +67,7 @@ void Monitor::Tick() { error_message << "The node with client ID " << client_id << " has been marked dead because the monitor" << " has missed too many heartbeats from it."; - // We use the nil JobID to broadcast the message to all drivers. + // We use the nil DriverID to broadcast the message to all drivers. RAY_CHECK_OK(gcs_client_.error_table().PushErrorToDriver( DriverID::nil(), type, error_message.str(), current_time_ms())); } @@ -88,7 +88,7 @@ void Monitor::Tick() { batch->batch.push_back(std::unique_ptr( new HeartbeatTableDataT(heartbeat.second))); } - RAY_CHECK_OK(gcs_client_.heartbeat_batch_table().Add(JobID::nil(), ClientID::nil(), + RAY_CHECK_OK(gcs_client_.heartbeat_batch_table().Add(DriverID::nil(), ClientID::nil(), batch, nullptr)); heartbeat_buffer_.clear(); } diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index b8daec203..267dc7a30 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -105,7 +105,7 @@ ray::Status NodeManager::RegisterGcs() { lineage_cache_.HandleEntryCommitted(task_id); }; RAY_RETURN_NOT_OK(gcs_client_->raylet_task_table().Subscribe( - JobID::nil(), gcs_client_->client_table().GetLocalClientId(), + DriverID::nil(), gcs_client_->client_table().GetLocalClientId(), task_committed_callback, nullptr, nullptr)); const auto task_lease_notification_callback = [this](gcs::AsyncGcsClient *client, @@ -129,7 +129,7 @@ ray::Status NodeManager::RegisterGcs() { reconstruction_policy_.HandleTaskLeaseNotification(task_id, 0); }; RAY_RETURN_NOT_OK(gcs_client_->task_lease_table().Subscribe( - JobID::nil(), gcs_client_->client_table().GetLocalClientId(), + DriverID::nil(), gcs_client_->client_table().GetLocalClientId(), task_lease_notification_callback, task_lease_empty_callback, nullptr)); // Register a callback to handle actor notifications. @@ -144,7 +144,7 @@ ray::Status NodeManager::RegisterGcs() { }; RAY_RETURN_NOT_OK(gcs_client_->actor_table().Subscribe( - JobID::nil(), ClientID::nil(), actor_notification_callback, nullptr)); + DriverID::nil(), ClientID::nil(), actor_notification_callback, nullptr)); // Register a callback on the client table for new clients. auto node_manager_client_added = [this](gcs::AsyncGcsClient *client, const UniqueID &id, @@ -166,7 +166,7 @@ ray::Status NodeManager::RegisterGcs() { HeartbeatBatchAdded(heartbeat_batch); }; RAY_RETURN_NOT_OK(gcs_client_->heartbeat_batch_table().Subscribe( - JobID::nil(), ClientID::nil(), heartbeat_batch_added, + DriverID::nil(), ClientID::nil(), heartbeat_batch_added, /*subscribe_callback=*/nullptr, /*done_callback=*/nullptr)); @@ -176,8 +176,8 @@ ray::Status NodeManager::RegisterGcs() { const std::vector &driver_data) { HandleDriverTableUpdate(client_id, driver_data); }; - RAY_RETURN_NOT_OK(gcs_client_->driver_table().Subscribe(JobID::nil(), ClientID::nil(), - driver_table_handler, nullptr)); + RAY_RETURN_NOT_OK(gcs_client_->driver_table().Subscribe( + DriverID::nil(), ClientID::nil(), driver_table_handler, nullptr)); // Start sending heartbeats to the GCS. last_heartbeat_at_ms_ = current_time_ms(); @@ -269,7 +269,7 @@ void NodeManager::Heartbeat() { } ray::Status status = heartbeat_table.Add( - JobID::nil(), gcs_client_->client_table().GetLocalClientId(), heartbeat_data, + DriverID::nil(), gcs_client_->client_table().GetLocalClientId(), heartbeat_data, /*success_callback=*/nullptr); RAY_CHECK_OK_PREPEND(status, "Heartbeat failed"); @@ -348,7 +348,7 @@ void NodeManager::ClientAdded(const ClientTableDataT &client_data) { error_message << "Failed to connect to ray node " << client_id << " with status: " << status.ToString() << ". This may be since the node was recently removed."; - // We use the nil JobID to broadcast the message to all drivers. + // We use the nil DriverID to broadcast the message to all drivers. RAY_CHECK_OK(gcs_client_->error_table().PushErrorToDriver( DriverID::nil(), type, error_message.str(), current_time_ms())); return; @@ -514,7 +514,7 @@ void NodeManager::PublishActorStateTransition( RAY_CHECK_OK(redis_context->RunArgvAsync(args)); } }; - RAY_CHECK_OK(gcs_client_->actor_table().AppendAt(JobID::nil(), actor_id, + RAY_CHECK_OK(gcs_client_->actor_table().AppendAt(DriverID::nil(), actor_id, actor_notification, success_callback, failure_callback, log_length)); } @@ -719,7 +719,7 @@ void NodeManager::ProcessClientMessage( for (const auto &object_id : object_ids) { creating_task_ids.push_back(ComputeTaskId(object_id)); } - gcs_client_->raylet_task_table().Delete(JobID::nil(), creating_task_ids); + gcs_client_->raylet_task_table().Delete(DriverID::nil(), creating_task_ids); } } break; case protocol::MessageType::PrepareActorCheckpointRequest: { @@ -752,7 +752,7 @@ void NodeManager::ProcessRegisterClientRequestMessage( // message is actually the ID of the driver task, while client_id represents the // real driver ID, which can associate all the tasks/actors for a given driver, // which is set to the worker ID. - const JobID driver_task_id = from_flatbuf(*message->driver_id()); + const DriverID driver_task_id = from_flatbuf(*message->driver_id()); worker->AssignTaskId(TaskID(driver_task_id)); worker->AssignDriverId(from_flatbuf(*message->client_id())); worker_pool_.RegisterDriver(std::move(worker)); @@ -1065,7 +1065,7 @@ void NodeManager::ProcessPrepareActorCheckpointRequest( // Write checkpoint data to GCS. RAY_CHECK_OK(gcs_client_->actor_checkpoint_table().Add( - JobID::nil(), checkpoint_id, checkpoint_data, + DriverID::nil(), checkpoint_id, checkpoint_data, [worker, actor_id, this](ray::gcs::AsyncGcsClient *client, const ActorCheckpointID &checkpoint_id, const ActorCheckpointDataT &data) { @@ -1074,7 +1074,7 @@ void NodeManager::ProcessPrepareActorCheckpointRequest( // Save this actor-to-checkpoint mapping, and remove old checkpoints associated // with this actor. RAY_CHECK_OK(gcs_client_->actor_checkpoint_id_table().AddCheckpointId( - JobID::nil(), actor_id, checkpoint_id)); + DriverID::nil(), actor_id, checkpoint_id)); // Send reply to worker. flatbuffers::FlatBufferBuilder fbb; auto reply = ray::protocol::CreatePrepareActorCheckpointReply( @@ -1414,7 +1414,7 @@ void NodeManager::SubmitTask(const Task &task, const Lineage &uncommitted_lineag HandleActorStateTransition(actor_id, ActorRegistration(data.back())); } }; - RAY_CHECK_OK(gcs_client_->actor_table().Lookup(JobID::nil(), spec.ActorId(), + RAY_CHECK_OK(gcs_client_->actor_table().Lookup(DriverID::nil(), spec.ActorId(), lookup_callback)); actor_creation_dummy_object = spec.ActorCreationDummyObjectId(); } else { @@ -1835,7 +1835,7 @@ void NodeManager::FinishAssignedActorTask(Worker &worker, const Task &task) { RAY_LOG(DEBUG) << "Looking up checkpoint " << checkpoint_id << " for actor " << actor_id; RAY_CHECK_OK(gcs_client_->actor_checkpoint_table().Lookup( - JobID::nil(), checkpoint_id, + DriverID::nil(), checkpoint_id, [this, actor_id, new_actor_data](ray::gcs::AsyncGcsClient *client, const UniqueID &checkpoint_id, const ActorCheckpointDataT &checkpoint_data) { @@ -1905,7 +1905,7 @@ void NodeManager::FinishAssignedActorTask(Worker &worker, const Task &task) { void NodeManager::HandleTaskReconstruction(const TaskID &task_id) { // Retrieve the task spec in order to re-execute the task. RAY_CHECK_OK(gcs_client_->raylet_task_table().Lookup( - JobID::nil(), task_id, + DriverID::nil(), task_id, /*success_callback=*/ [this](ray::gcs::AsyncGcsClient *client, const TaskID &task_id, const ray::protocol::TaskT &task_data) { diff --git a/src/ray/raylet/raylet_client.h b/src/ray/raylet/raylet_client.h index d9cd63121..ff66ff462 100644 --- a/src/ray/raylet/raylet_client.h +++ b/src/ray/raylet/raylet_client.h @@ -13,7 +13,6 @@ using ray::ActorCheckpointID; using ray::ActorID; using ray::ClientID; using ray::DriverID; -using ray::JobID; using ray::ObjectID; using ray::TaskID; using ray::UniqueID; @@ -170,7 +169,7 @@ class RayletClient { ClientID GetClientID() const { return client_id_; } - JobID GetDriverID() const { return driver_id_; } + DriverID GetDriverID() const { return driver_id_; } bool IsWorker() const { return is_worker_; } @@ -179,7 +178,7 @@ class RayletClient { private: const ClientID client_id_; const bool is_worker_; - const JobID driver_id_; + const DriverID driver_id_; const Language language_; /// A map from resource name to the resource IDs that are currently reserved /// for this worker. Each pair consists of the resource ID and the fraction diff --git a/src/ray/raylet/reconstruction_policy.cc b/src/ray/raylet/reconstruction_policy.cc index d75f8799f..ee328e027 100644 --- a/src/ray/raylet/reconstruction_policy.cc +++ b/src/ray/raylet/reconstruction_policy.cc @@ -50,7 +50,7 @@ void ReconstructionPolicy::SetTaskTimeout( // required by the task are no longer needed soon after. If the // task is still required after this initial period, then we now // subscribe to task lease notifications. - RAY_CHECK_OK(task_lease_pubsub_.RequestNotifications(JobID::nil(), task_id, + RAY_CHECK_OK(task_lease_pubsub_.RequestNotifications(DriverID::nil(), task_id, client_id_)); it->second.subscribed = true; } @@ -108,7 +108,7 @@ void ReconstructionPolicy::AttemptReconstruction(const TaskID &task_id, reconstruction_entry->num_reconstructions = reconstruction_attempt; reconstruction_entry->node_manager_id = client_id_.binary(); RAY_CHECK_OK(task_reconstruction_log_.AppendAt( - JobID::nil(), task_id, reconstruction_entry, + DriverID::nil(), task_id, reconstruction_entry, /*success_callback=*/ [this](gcs::AsyncGcsClient *client, const TaskID &task_id, const TaskReconstructionDataT &data) { @@ -197,7 +197,7 @@ void ReconstructionPolicy::Cancel(const ObjectID &object_id) { // Cancel notifications for the task lease if we were subscribed to them. if (it->second.subscribed) { RAY_CHECK_OK( - task_lease_pubsub_.CancelNotifications(JobID::nil(), task_id, client_id_)); + task_lease_pubsub_.CancelNotifications(DriverID::nil(), task_id, client_id_)); } listening_tasks_.erase(it); } diff --git a/src/ray/raylet/reconstruction_policy_test.cc b/src/ray/raylet/reconstruction_policy_test.cc index c5678d6ce..d9fb92388 100644 --- a/src/ray/raylet/reconstruction_policy_test.cc +++ b/src/ray/raylet/reconstruction_policy_test.cc @@ -82,7 +82,7 @@ class MockGcs : public gcs::PubsubInterface, failure_callback_ = failure_callback; } - void Add(const JobID &job_id, const TaskID &task_id, + void Add(const DriverID &driver_id, const TaskID &task_id, std::shared_ptr &task_lease_data) { task_lease_table_[task_id] = task_lease_data; if (subscribed_tasks_.count(task_id) == 1) { @@ -90,7 +90,7 @@ class MockGcs : public gcs::PubsubInterface, } } - Status RequestNotifications(const JobID &job_id, const TaskID &task_id, + Status RequestNotifications(const DriverID &driver_id, const TaskID &task_id, const ClientID &client_id) { subscribed_tasks_.insert(task_id); auto entry = task_lease_table_.find(task_id); @@ -102,14 +102,14 @@ class MockGcs : public gcs::PubsubInterface, return ray::Status::OK(); } - Status CancelNotifications(const JobID &job_id, const TaskID &task_id, + Status CancelNotifications(const DriverID &driver_id, const TaskID &task_id, const ClientID &client_id) { subscribed_tasks_.erase(task_id); return ray::Status::OK(); } Status AppendAt( - const JobID &job_id, const TaskID &task_id, + const DriverID &driver_id, const TaskID &task_id, std::shared_ptr &task_data, const ray::gcs::LogInterface::WriteCallback &success_callback, @@ -132,7 +132,7 @@ class MockGcs : public gcs::PubsubInterface, MOCK_METHOD4( Append, ray::Status( - const JobID &, const TaskID &, std::shared_ptr &, + const DriverID &, const TaskID &, std::shared_ptr &, const ray::gcs::LogInterface::WriteCallback &)); private: @@ -323,7 +323,7 @@ TEST_F(ReconstructionPolicyTest, TestReconstructionSuppressed) { task_lease_data->node_manager_id = ClientID::from_random().binary(); task_lease_data->acquired_at = current_sys_time_ms(); task_lease_data->timeout = 2 * test_period; - mock_gcs_.Add(JobID::nil(), task_id, task_lease_data); + mock_gcs_.Add(DriverID::nil(), task_id, task_lease_data); // Listen for an object. reconstruction_policy_->ListenAndMaybeReconstruct(object_id); @@ -351,7 +351,7 @@ TEST_F(ReconstructionPolicyTest, TestReconstructionContinuallySuppressed) { task_lease_data->node_manager_id = ClientID::from_random().binary(); task_lease_data->acquired_at = current_sys_time_ms(); task_lease_data->timeout = reconstruction_timeout_ms_; - mock_gcs_.Add(JobID::nil(), task_id, task_lease_data); + mock_gcs_.Add(DriverID::nil(), task_id, task_lease_data); }); // Run the test for much longer than the reconstruction timeout. Run(reconstruction_timeout_ms_ * 2); @@ -405,7 +405,7 @@ TEST_F(ReconstructionPolicyTest, TestSimultaneousReconstructionSuppressed) { task_reconstruction_data->node_manager_id = ClientID::from_random().binary(); task_reconstruction_data->num_reconstructions = 0; RAY_CHECK_OK( - mock_gcs_.AppendAt(JobID::nil(), task_id, task_reconstruction_data, nullptr, + mock_gcs_.AppendAt(DriverID::nil(), task_id, task_reconstruction_data, nullptr, /*failure_callback=*/ [](ray::gcs::AsyncGcsClient *client, const TaskID &task_id, const TaskReconstructionDataT &data) { ASSERT_TRUE(false); }, diff --git a/src/ray/raylet/task_dependency_manager.cc b/src/ray/raylet/task_dependency_manager.cc index 2f1b64a87..fe4364c44 100644 --- a/src/ray/raylet/task_dependency_manager.cc +++ b/src/ray/raylet/task_dependency_manager.cc @@ -263,7 +263,7 @@ void TaskDependencyManager::AcquireTaskLease(const TaskID &task_id) { task_lease_data->node_manager_id = client_id_.hex(); task_lease_data->acquired_at = current_sys_time_ms(); task_lease_data->timeout = it->second.lease_period; - RAY_CHECK_OK(task_lease_table_.Add(JobID::nil(), task_id, task_lease_data, nullptr)); + RAY_CHECK_OK(task_lease_table_.Add(DriverID::nil(), task_id, task_lease_data, nullptr)); auto period = boost::posix_time::milliseconds(it->second.lease_period / 2); it->second.lease_timer->expires_from_now(period); diff --git a/src/ray/raylet/task_dependency_manager_test.cc b/src/ray/raylet/task_dependency_manager_test.cc index e0d30bf9e..5126d8255 100644 --- a/src/ray/raylet/task_dependency_manager_test.cc +++ b/src/ray/raylet/task_dependency_manager_test.cc @@ -29,7 +29,7 @@ class MockGcs : public gcs::TableInterface { public: MOCK_METHOD4( Add, - ray::Status(const JobID &job_id, const TaskID &task_id, + ray::Status(const DriverID &driver_id, const TaskID &task_id, std::shared_ptr &task_data, const gcs::TableInterface::WriteCallback &done)); };