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

Change all instances of JobID to DriverID. (#4431)

Browse Source
This commit is contained in:
justinwyang
2019-04-22 16:28:09 -07:00
committed by Robert Nishihara
parent 06c768823c
commit 8dfc833a8b
21 changed files with 319 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/experimental/state.py
+20 -19
View File
@@ -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):
python/ray/gcs_utils.py
+1 -1
View File
@@ -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)
python/ray/includes/unique_ids.pxd
-16
View File
@@ -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
python/ray/includes/unique_ids.pxi
-1
View File
@@ -19,7 +19,6 @@ from ray.includes.unique_ids cimport (
CConfigID,
CDriverID,
CFunctionID,
CJobID,
CObjectID,
CTaskID,
CUniqueID,
python/ray/worker.py
+7 -6
View File
@@ -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()
]:
src/ray/gcs/client.h
+2 -2
View File
@@ -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<std::string, std::string> &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);
src/ray/gcs/client_test.cc
+150 -139
View File
@@ -29,7 +29,7 @@ class TestGcs : public ::testing::Test {
TestGcs(CommandType command_type) : num_callbacks_(0), command_type_(command_type) {
client_ = std::make_shared<gcs::AsyncGcsClient>("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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> client) {
void TestTableLookup(const DriverID &driver_id,
std::shared_ptr<gcs::AsyncGcsClient> client) {
TaskID task_id = TaskID::from_random();
auto data = std::make_shared<protocol::TaskT>();
data->task_specification = "123";
@@ -108,8 +109,8 @@ void TestTableLookup(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> client) {
void TestLogLookup(const DriverID &driver_id,
std::shared_ptr<gcs::AsyncGcsClient> client) {
// Append some entries to the log at an object ID.
TaskID task_id = TaskID::from_random();
std::vector<std::string> node_manager_ids = {"abc", "def", "ghi"};
@@ -143,7 +145,7 @@ void TestLogLookup(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> client) {
void TestLogAppendAt(const DriverID &driver_id,
std::shared_ptr<gcs::AsyncGcsClient> client) {
TaskID task_id = TaskID::from_random();
std::vector<std::string> node_manager_ids = {"A", "B"};
std::vector<std::shared_ptr<TaskReconstructionDataT>> data_log;
@@ -219,21 +222,22 @@ void TestLogAppendAt(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> c
TEST_F(TestGcsWithAsio, TestLogAppendAt) {
test = this;
TestLogAppendAt(job_id_, client_);
TestLogAppendAt(driver_id_, client_);
}
void TestSet(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> client) {
void TestSet(const DriverID &driver_id, std::shared_ptr<gcs::AsyncGcsClient> client) {
// Add some entries to the set at an object ID.
ObjectID object_id = ObjectID::from_random();
std::vector<std::string> managers = {"abc", "def", "ghi"};
@@ -273,7 +277,7 @@ void TestSet(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<ObjectTableDataT>();
@@ -299,7 +303,7 @@ void TestSet(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> client) {
TEST_F(TestGcsWithAsio, TestSet) {
test = this;
TestSet(job_id_, client_);
TestSet(driver_id_, client_);
}
void TestDeleteKeysFromLog(
const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> client,
const DriverID &driver_id, std::shared_ptr<gcs::AsyncGcsClient> client,
std::vector<std::shared_ptr<TaskReconstructionDataT>> &data_vector) {
std::vector<TaskID> 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<gcs::AsyncGcsClient> client,
std::vector<std::shared_ptr<protocol::TaskT>> &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<gcs::AsyncGcsClient> client,
std::vector<std::shared_ptr<ObjectTableDataT>> &data_vector) {
std::vector<ObjectID> 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<gcs::AsyncGcsClient> client) {
void TestDeleteKeys(const DriverID &driver_id,
std::shared_ptr<gcs::AsyncGcsClient> client) {
// Test delete function for keys of Log.
std::vector<std::shared_ptr<TaskReconstructionDataT>> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<std::shared_ptr<protocol::TaskT>> task_vector;
@@ -507,16 +512,16 @@ void TestDeleteKeys(const JobID &job_id, std::shared_ptr<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> client) {
std::vector<DriverID> 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<gcs::AsyncGcsClient> client) {
std::vector<ObjectID> 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<gcs::AsyncGcsClient> 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<protocol::TaskT>();
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<protocol::TaskT>();
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<gcs::AsyncGcsClient> client) {
// Add a log entry.
DriverID driver_id1 = DriverID::from_random();
std::vector<std::string> driver_ids1 = {"abc", "def", "ghi"};
auto data1 = std::make_shared<DriverTableDataT>();
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<std::string> driver_ids2 = {"jkl", "mno", "pqr"};
auto data2 = std::make_shared<DriverTableDataT>();
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<std::string>(++driver_ids1.begin(), driver_ids1.end());
for (const auto &driver_id : remaining) {
for (const auto &driver_id_it : remaining) {
auto data = std::make_shared<DriverTableDataT>();
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<std::string>(++driver_ids2.begin(), driver_ids2.end());
for (const auto &driver_id : remaining) {
for (const auto &driver_id_it : remaining) {
auto data = std::make_shared<DriverTableDataT>();
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<gcs::AsyncGcsClient> client) {
// Add a set entry.
ObjectID object_id1 = ObjectID::from_random();
std::vector<std::string> managers1 = {"abc", "def", "ghi"};
auto data1 = std::make_shared<ObjectTableDataT>();
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<std::string> managers2 = {"jkl", "mno", "pqr"};
auto data2 = std::make_shared<ObjectTableDataT>();
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<std::string>(++managers1.begin(), managers1.end());
for (const auto &manager : remaining) {
auto data = std::make_shared<ObjectTableDataT>();
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<std::string>(++managers2.begin(), managers2.end());
for (const auto &manager : remaining) {
auto data = std::make_shared<ObjectTableDataT>();
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<gcs::AsyncGcsClient> client) {
// Add a table entry.
TaskID task_id = TaskID::from_random();
std::vector<std::string> task_specs = {"jkl", "mno", "pqr"};
auto data = std::make_shared<protocol::TaskT>();
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<std::string>(++task_specs.begin(), task_specs.end());
for (const auto &task_spec : remaining) {
auto data = std::make_shared<protocol::TaskT>();
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<gcs::AsyncGcsClient> client) {
// Add a log entry.
DriverID driver_id = DriverID::from_random();
DriverID random_driver_id = DriverID::from_random();
std::vector<std::string> driver_ids = {"jkl", "mno", "pqr"};
auto data = std::make_shared<DriverTableDataT>();
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<DriverTableDataT> &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<std::string>(++driver_ids.begin(), driver_ids.end());
for (const auto &remaining_driver_id : remaining) {
auto data = std::make_shared<DriverTableDataT>();
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<gcs::AsyncGcsClient> client) {
// Add a set entry.
ObjectID object_id = ObjectID::from_random();
std::vector<std::string> managers = {"jkl", "mno", "pqr"};
auto data = std::make_shared<ObjectTableDataT>();
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<std::string>(++managers.begin(), managers.end());
for (const auto &manager : remaining) {
auto data = std::make_shared<ObjectTableDataT>();
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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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<gcs::AsyncGcsClient> 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
src/ray/gcs/format/gcs.fbs
+2 -2
View File
@@ -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.
src/ray/gcs/tables.cc
+38 -37
View File
@@ -38,7 +38,7 @@ namespace ray {
namespace gcs {
template <typename ID, typename Data>
Status Log<ID, Data>::Append(const JobID &job_id, const ID &id,
Status Log<ID, Data>::Append(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &dataT, const WriteCallback &done) {
num_appends_++;
auto callback = [this, id, dataT, done](const std::string &data) {
@@ -59,7 +59,7 @@ Status Log<ID, Data>::Append(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
Status Log<ID, Data>::AppendAt(const JobID &job_id, const ID &id,
Status Log<ID, Data>::AppendAt(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &dataT, const WriteCallback &done,
const WriteCallback &failure, int log_length) {
num_appends_++;
@@ -84,7 +84,8 @@ Status Log<ID, Data>::AppendAt(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
Status Log<ID, Data>::Lookup(const JobID &job_id, const ID &id, const Callback &lookup) {
Status Log<ID, Data>::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<ID, Data>::Lookup(const JobID &job_id, const ID &id, const Callback &
}
template <typename ID, typename Data>
Status Log<ID, Data>::Subscribe(const JobID &job_id, const ClientID &client_id,
Status Log<ID, Data>::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<ID, Data>::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 <typename ID, typename Data>
Status Log<ID, Data>::Subscribe(const JobID &job_id, const ClientID &client_id,
Status Log<ID, Data>::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<ID, Data>::Subscribe(const JobID &job_id, const ClientID &client_id,
}
template <typename ID, typename Data>
Status Log<ID, Data>::RequestNotifications(const JobID &job_id, const ID &id,
Status Log<ID, Data>::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<ID, Data>::RequestNotifications(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
Status Log<ID, Data>::CancelNotifications(const JobID &job_id, const ID &id,
Status Log<ID, Data>::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<ID, Data>::CancelNotifications(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
void Log<ID, Data>::Delete(const JobID &job_id, const std::vector<ID> &ids) {
void Log<ID, Data>::Delete(const DriverID &driver_id, const std::vector<ID> &ids) {
if (ids.empty()) {
return;
}
@@ -210,8 +211,8 @@ void Log<ID, Data>::Delete(const JobID &job_id, const std::vector<ID> &ids) {
}
template <typename ID, typename Data>
void Log<ID, Data>::Delete(const JobID &job_id, const ID &id) {
Delete(job_id, std::vector<ID>({id}));
void Log<ID, Data>::Delete(const DriverID &driver_id, const ID &id) {
Delete(driver_id, std::vector<ID>({id}));
}
template <typename ID, typename Data>
@@ -222,7 +223,7 @@ std::string Log<ID, Data>::DebugString() const {
}
template <typename ID, typename Data>
Status Table<ID, Data>::Add(const JobID &job_id, const ID &id,
Status Table<ID, Data>::Add(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &dataT, const WriteCallback &done) {
num_adds_++;
auto callback = [this, id, dataT, done](const std::string &data) {
@@ -240,10 +241,10 @@ Status Table<ID, Data>::Add(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
Status Table<ID, Data>::Lookup(const JobID &job_id, const ID &id, const Callback &lookup,
const FailureCallback &failure) {
Status Table<ID, Data>::Lookup(const DriverID &driver_id, const ID &id,
const Callback &lookup, const FailureCallback &failure) {
num_lookups_++;
return Log<ID, Data>::Lookup(job_id, id,
return Log<ID, Data>::Lookup(driver_id, id,
[lookup, failure](AsyncGcsClient *client, const ID &id,
const std::vector<DataT> &data) {
if (data.empty()) {
@@ -260,12 +261,12 @@ Status Table<ID, Data>::Lookup(const JobID &job_id, const ID &id, const Callback
}
template <typename ID, typename Data>
Status Table<ID, Data>::Subscribe(const JobID &job_id, const ClientID &client_id,
Status Table<ID, Data>::Subscribe(const DriverID &driver_id, const ClientID &client_id,
const Callback &subscribe,
const FailureCallback &failure,
const SubscriptionCallback &done) {
return Log<ID, Data>::Subscribe(
job_id, client_id,
driver_id, client_id,
[subscribe, failure](AsyncGcsClient *client, const ID &id,
const std::vector<DataT> &data) {
RAY_CHECK(data.empty() || data.size() == 1);
@@ -288,7 +289,7 @@ std::string Table<ID, Data>::DebugString() const {
}
template <typename ID, typename Data>
Status Set<ID, Data>::Add(const JobID &job_id, const ID &id,
Status Set<ID, Data>::Add(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &dataT, const WriteCallback &done) {
num_adds_++;
auto callback = [this, id, dataT, done](const std::string &data) {
@@ -306,7 +307,7 @@ Status Set<ID, Data>::Add(const JobID &job_id, const ID &id,
}
template <typename ID, typename Data>
Status Set<ID, Data>::Remove(const JobID &job_id, const ID &id,
Status Set<ID, Data>::Remove(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &dataT, const WriteCallback &done) {
num_removes_++;
auto callback = [this, id, dataT, done](const std::string &data) {
@@ -334,11 +335,11 @@ std::string Set<ID, Data>::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<ErrorTableDataT>();
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<DriverTableDataT>();
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<ClientTableDataT>();
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<ClientID, ClientTableDataT> &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<ActorCheckpointIdDataT> 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<ActorCheckpointIdDataT> data =
std::make_shared<ActorCheckpointIdDataT>();
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<ObjectID, ObjectTableData>;
src/ray/gcs/tables.h
+52 -50
View File
@@ -38,9 +38,9 @@ enum class CommandType { kRegular, kChain };
template <typename ID>
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<void(AsyncGcsClient *client, const ID &id, const DataT &data)>;
virtual Status Append(const JobID &job_id, const ID &id, std::shared_ptr<DataT> &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<DataT> &data, const WriteCallback &done) = 0;
virtual Status AppendAt(const DriverID &driver_id, const ID &task_id,
std::shared_ptr<DataT> &data, const WriteCallback &done,
const WriteCallback &failure, int log_length) = 0;
virtual ~LogInterface(){};
@@ -104,20 +104,20 @@ class Log : public LogInterface<ID, Data>, virtual public PubsubInterface<ID> {
/// 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<DataT> &data,
Status Append(const DriverID &driver_id, const ID &id, std::shared_ptr<DataT> &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<ID, Data>, virtual public PubsubInterface<ID> {
/// \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<DataT> &data,
Status AppendAt(const DriverID &driver_id, const ID &id, std::shared_ptr<DataT> &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<ID, Data>, virtual public PubsubInterface<ID> {
/// \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<ID, Data>, virtual public PubsubInterface<ID> {
/// 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<ID> &ids);
void Delete(const DriverID &driver_id, const std::vector<ID> &ids);
/// Returns debug string for class.
///
@@ -217,7 +217,7 @@ class Log : public LogInterface<ID, Data>, virtual public PubsubInterface<ID> {
/// 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<ID, Data>, virtual public PubsubInterface<ID> {
/// \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<ID, Data>::WriteCallback;
virtual Status Add(const JobID &job_id, const ID &task_id, std::shared_ptr<DataT> &data,
const WriteCallback &done) = 0;
virtual Status Add(const DriverID &driver_id, const ID &task_id,
std::shared_ptr<DataT> &data, const WriteCallback &done) = 0;
virtual ~TableInterface(){};
};
@@ -299,32 +299,32 @@ class Table : private Log<ID, Data>,
/// 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<DataT> &data,
Status Add(const DriverID &driver_id, const ID &id, std::shared_ptr<DataT> &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<ID, Data>,
/// \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<ID, Data>::Delete(job_id, id); }
void Delete(const DriverID &driver_id, const ID &id) {
Log<ID, Data>::Delete(driver_id, id);
}
void Delete(const JobID &job_id, const std::vector<ID> &ids) {
Log<ID, Data>::Delete(job_id, ids);
void Delete(const DriverID &driver_id, const std::vector<ID> &ids) {
Log<ID, Data>::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<ID, Data>::WriteCallback;
virtual Status Add(const JobID &job_id, const ID &id, std::shared_ptr<DataT> &data,
const WriteCallback &done) = 0;
virtual Status Remove(const JobID &job_id, const ID &id, std::shared_ptr<DataT> &data,
const WriteCallback &done) = 0;
virtual Status Add(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &data, const WriteCallback &done) = 0;
virtual Status Remove(const DriverID &driver_id, const ID &id,
std::shared_ptr<DataT> &data, const WriteCallback &done) = 0;
virtual ~SetInterface(){};
};
@@ -406,30 +408,30 @@ class Set : private Log<ID, Data>,
/// 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<DataT> &data,
Status Add(const DriverID &driver_id, const ID &id, std::shared_ptr<DataT> &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<DataT> &data,
Status Remove(const DriverID &driver_id, const ID &id, std::shared_ptr<DataT> &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<ID, Data>::Subscribe(job_id, client_id, subscribe, done);
return Log<ID, Data>::Subscribe(driver_id, client_id, subscribe, done);
}
/// Returns debug string for class.
@@ -547,9 +549,9 @@ class TaskLeaseTable : public Table<TaskID, TaskLeaseData> {
prefix_ = TablePrefix::TASK_LEASE;
}
Status Add(const JobID &job_id, const TaskID &id, std::shared_ptr<TaskLeaseDataT> &data,
const WriteCallback &done) override {
RAY_RETURN_NOT_OK((Table<TaskID, TaskLeaseData>::Add(job_id, id, data, done)));
Status Add(const DriverID &driver_id, const TaskID &id,
std::shared_ptr<TaskLeaseDataT> &data, const WriteCallback &done) override {
RAY_RETURN_NOT_OK((Table<TaskID, TaskLeaseData>::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<ActorID, ActorCheckpointIdData> {
/// 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<DriverID, ErrorTableData> {
/// 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.
src/ray/id_def.h
-1
View File
@@ -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);
src/ray/object_manager/object_directory.cc
+6 -6
View File
@@ -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<ObjectTableDataT> &location_updates) {
// Build the set of current locations based on the entries in the log.
src/ray/raylet/lineage_cache.cc
+3 -3
View File
@@ -359,7 +359,7 @@ void LineageCache::FlushTask(const TaskID &task_id) {
auto root = flatbuffers::GetRoot<protocol::Task>(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
src/ray/raylet/lineage_cache_test.cc
+4 -4
View File
@@ -22,7 +22,7 @@ class MockGcs : public gcs::TableInterface<TaskID, protocol::Task>,
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<protocol::TaskT> &task_data,
const gcs::TableInterface<TaskID, protocol::Task>::WriteCallback &done) {
task_table_[task_id] = task_data;
@@ -43,10 +43,10 @@ class MockGcs : public gcs::TableInterface<TaskID, protocol::Task>,
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<TaskID, protocol::Task>,
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();
src/ray/raylet/monitor.cc
+3 -3
View File
@@ -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<HeartbeatTableDataT>(
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();
}
src/ray/raylet/node_manager.cc
+16 -16
View File
@@ -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<DriverTableDataT> &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<JobID>(*message->driver_id());
const DriverID driver_task_id = from_flatbuf<DriverID>(*message->driver_id());
worker->AssignTaskId(TaskID(driver_task_id));
worker->AssignDriverId(from_flatbuf<DriverID>(*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) {
src/ray/raylet/raylet_client.h
+2 -3
View File
@@ -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
src/ray/raylet/reconstruction_policy.cc
+3 -3
View File
@@ -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);
}
src/ray/raylet/reconstruction_policy_test.cc
+8 -8
View File
@@ -82,7 +82,7 @@ class MockGcs : public gcs::PubsubInterface<TaskID>,
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<TaskLeaseDataT> &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<TaskID>,
}
}
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<TaskID>,
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<TaskReconstructionDataT> &task_data,
const ray::gcs::LogInterface<TaskID, TaskReconstructionData>::WriteCallback
&success_callback,
@@ -132,7 +132,7 @@ class MockGcs : public gcs::PubsubInterface<TaskID>,
MOCK_METHOD4(
Append,
ray::Status(
const JobID &, const TaskID &, std::shared_ptr<TaskReconstructionDataT> &,
const DriverID &, const TaskID &, std::shared_ptr<TaskReconstructionDataT> &,
const ray::gcs::LogInterface<TaskID, TaskReconstructionData>::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); },
src/ray/raylet/task_dependency_manager.cc
+1 -1
View File
@@ -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);
src/ray/raylet/task_dependency_manager_test.cc
+1 -1
View File
@@ -29,7 +29,7 @@ class MockGcs : public gcs::TableInterface<TaskID, TaskLeaseData> {
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<TaskLeaseDataT> &task_data,
const gcs::TableInterface<TaskID, TaskLeaseData>::WriteCallback &done));
};