[xray] Track `ray.get` calls as task dependencies #2362

stephanie-wang · 2018-07-06T22:46:19Z

What do these changes do?

The task dependency manager is used to make any dependencies required by a task available, either by fetching it from a remote node, or eventually, by reconstructing it. This was previously used for scheduling submitted tasks with object arguments, but the logic needed for a task that calls ray.get is similar. This PR directs that logic through the task dependency manager as well. This also includes some cleanup in the SchedulingQueue code.

AmplabJenkins · 2018-07-07T00:49:30Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6518/
Test FAILed.

AmplabJenkins · 2018-07-07T00:54:30Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6519/
Test FAILed.

robertnishihara · 2018-07-07T06:32:48Z

python/ray/worker.py

+                    self.local_scheduler_client.reconstruct_objects(
+                        ray_object_ids_to_fetch[i:(
+                            i + ray._config.worker_fetch_request_size())],
+                        False)


Did this batching work? Legacy Ray doesn't seem to support batching here

ray/src/local_scheduler/local_scheduler.cc

Line 1123 in e3534c4

RAY_CHECK(message->object_ids()->size() == 1);

Oops, thanks, yeah it should only work for xray.

robertnishihara · 2018-07-07T06:33:38Z

src/ray/raylet/scheduling_queue.h

@@ -12,7 +12,8 @@ namespace ray {

 namespace raylet {

-enum TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING };
+enum class TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING, BLOCKED };


thank you for switching to enum class :)

robertnishihara · 2018-07-07T22:10:45Z

src/ray/raylet/node_manager.cc

+      // Subscribe to the objects required by the ray.get. These objects will
+      // be fetched and/or reconstructed as necessary, until the objects become
+      // local or are unsubscribed.
+      task_dependency_manager_.SubscribeDependencies(current_task_id,


What does it mean if this is called and current_task_id is nil?

Does that mean it's a driver? Won't that be an issue if there are multiple drivers?

Yes, that means it's a driver.

Ah yes, I think you're right. We'll have to include the driver ID in the initial request to register a client with the raylet.

robertnishihara · 2018-07-07T22:19:28Z

src/ray/raylet/scheduling_queue.cc

+      it++;
+    }
+  }
+}


I've had to do awkward things with the scheduling queues before, not sure what the best way to fix this is, but some kinds of queries are a bit hard to do, e.g.,

ray/src/ray/raylet/node_manager.cc

Lines 384 to 391 in 35f4a30

// TODO(rkn): This is too heavyweight just to get the task's driver ID.

auto const it = std::find_if(

running_tasks.begin(), running_tasks.end(), [task_id](const Task &task) {

return task.GetTaskSpecification().TaskId() == task_id;

});

RAY_CHECK(running_tasks.size() != 0);

RAY_CHECK(it != running_tasks.end());

JobID job_id = it->GetTaskSpecification().DriverId();

Yeah, until we have a way to do fast lookup of a task's state, I think this is the best option for now...

For that snippet that you posted, maybe it makes sense to just remove the task from the queue? I don't know if we should be leaving that task in the running tasks queue anyway.

Oh good point, I commented about this above.

robertnishihara · 2018-07-07T22:20:16Z

src/ray/raylet/scheduling_queue.h

@@ -70,9 +71,12 @@ class SchedulingQueue {
  /// Remove tasks from the task queue.
  ///
  /// \param tasks The set of task IDs to remove from the queue. The
-  ///        corresponding tasks must be contained in the queue.
+  ///        corresponding tasks must be contained in the queue. The IDs of
+  ///        removed tasks will be erased from the set.


robertnishihara · 2018-07-07T22:22:47Z

src/ray/raylet/scheduling_queue.cc

+    FilterStateFromQueue(blocked_tasks_, task_ids, filter_state);
+    break;
+  default:
+    RAY_LOG(ERROR) << "Attempting to filter tasks on unrecognized state "


ERROR -> FATAL

robertnishihara · 2018-07-07T22:22:56Z

src/ray/raylet/scheduling_queue.cc

    break;
  default:
-    RAY_LOG(ERROR) << "Attempting to move tasks from unrecognized state " << src_state;
+    RAY_LOG(ERROR) << "Attempting to move tasks from unrecognized state "


ERROR -> FATAL

AmplabJenkins · 2018-07-20T21:34:02Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6749/
Test FAILed.

AmplabJenkins · 2018-07-20T22:17:30Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6751/
Test FAILed.

robertnishihara

This looks pretty good to me (though there are a few parts I haven't read through carefully yet).

cc @atumanov is also currently working on the scheduling queues code.

robertnishihara · 2018-07-20T23:21:29Z

src/ray/raylet/node_manager.cc

+      // The client is a driver.
+      const std::shared_ptr<Worker> driver = worker_pool_.GetRegisteredDriver(client);
+      RAY_CHECK(driver);
+      auto driver_id = driver->GetAssignedTaskId();


This seems like we're conflating "driver ID" with "task ID of the driver task". Is that correct? Is there a reason for doing that?

I felt like it made sense to treat them as the same thing, since it seems like they pretty much serve the same purpose. What do you think?

I agree they serve the same purpose.

I'm fine with unifying the IDs into the same thing, though I find it a little confusing to refer to it in some places as the "driver ID" and in some places as a "task ID". I think your approach is ok.

However, note that in the current PR, if we actually use it as a "task ID" (e.g., use it to look up something in the task table), that won't work because we're actually just passing in the "worker ID" (which I think is randomly generated) when we connect to the local scheduler.

Do you anticipate ever using the driver ID as a task ID in the node manager?

Ah, I see. I think that's actually fixed since the most recent commit since I changed the driver's frontend to pass in worker.current_task_id instead of the worker ID. Do I understand that correctly?

Oh ok, I think that's correct then. We confusingly set that ID for drivers in two places (both at the top and bottom of the snippet

ray/python/ray/worker.py

Lines 2181 to 2221 in 05490b8

worker.current_task_id = ray.ObjectID(np.random.bytes(20))

# Reset the state of the numpy random number generator.

np.random.set_state(numpy_state)

# Set other fields needed for computing task IDs.

worker.task_index = 0

worker.put_index = 1

# Create an entry for the driver task in the task table. This task is

# added immediately with status RUNNING. This allows us to push errors

# related to this driver task back to the driver. For example, if the

# driver creates an object that is later evicted, we should notify the

# user that we're unable to reconstruct the object, since we cannot

# rerun the driver.

nil_actor_counter = 0

driver_task = ray.local_scheduler.Task(

worker.task_driver_id, ray.ObjectID(NIL_FUNCTION_ID), [], 0,

worker.current_task_id, worker.task_index,

ray.ObjectID(NIL_ACTOR_ID), ray.ObjectID(NIL_ACTOR_ID),

ray.ObjectID(NIL_ACTOR_ID), ray.ObjectID(NIL_ACTOR_ID),

nil_actor_counter, False, [], {"CPU": 0}, worker.use_raylet)

# Add the driver task to the task table.

if not worker.use_raylet:

global_state._execute_command(

driver_task.task_id(), "RAY.TASK_TABLE_ADD",

driver_task.task_id().id(), TASK_STATUS_RUNNING,

NIL_LOCAL_SCHEDULER_ID,

driver_task.execution_dependencies_string(), 0,

ray.local_scheduler.task_to_string(driver_task))

else:

global_state._execute_command(

driver_task.task_id(), "RAY.TABLE_ADD",

ray.gcs_utils.TablePrefix.RAYLET_TASK,

ray.gcs_utils.TablePubsub.RAYLET_TASK,

driver_task.task_id().id(),

driver_task._serialized_raylet_task())

# Set the driver's current task ID to the task ID assigned to the

# driver task.

worker.current_task_id = driver_task.task_id()

robertnishihara · 2018-07-20T23:24:55Z

src/ray/raylet/node_manager.cc

-        // dependency manager.
-        RAY_CHECK_OK(object_manager_.Pull(object_id));
+        if (message->fetch_only()) {
+          RAY_CHECK_OK(object_manager_.Pull(object_id));


Don't we want to Pull in the non-fetch-only case as well?

I interpreted the fetch_only flag as "if true, then just fetch, if false, then fetch + reconstruct".

Oops, thanks!

Ah actually, the reason I did that was because the task dependency manager calls Pull internally once it determines that a dependency is missing. A little unfortunate, but I guess this makes sense for now while we have the fetch_only flag? I'll leave a comment about that.

robertnishihara · 2018-07-20T23:26:00Z

src/ray/raylet/node_manager.cc

+          RAY_CHECK_OK(object_manager_.Pull(object_id));
+        } else {
+          // Add any missing objects to the list to subscribe to in the task
+          // dependency manager.


I think this else might be applied to the wrong if statement.

robertnishihara · 2018-07-20T23:27:26Z

src/ray/raylet/node_manager.cc

-      // Clean up the assigned task's resources, push an error to the driver.
+      // The client is a worker. Handle the case where the worker is killed
+      // while executing a task. Clean up the assigned task's resources, push
+      // an error to the driver.


I can address this in a different PR, but I noticed that when the client dies, we should be removing the in-progress task from the queues, which we are not doing.

robertnishihara · 2018-07-20T23:32:46Z

src/ray/raylet/scheduling_queue.cc

+      it++;
+    }
+  }
+}


Oh good point, I commented about this above.

robertnishihara · 2018-07-20T23:53:39Z

src/ray/raylet/scheduling_queue.h

@@ -12,7 +12,8 @@ namespace ray {

 namespace raylet {

-enum TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING };
+enum class TaskState { INIT, PLACEABLE, WAITING, READY, RUNNING, BLOCKED, DRIVER };


thank you for making it enum class

robertnishihara · 2018-07-20T23:57:23Z

src/ray/raylet/scheduling_queue.h

@@ -67,12 +68,31 @@ class SchedulingQueue {
  /// at runtime.
  const std::list<Task> &GetBlockedTasks() const;

+  /// Get the set of driver task IDs.


A general comment, SchedulingQueue feels like a very "shallow" class. In the sense that it has a large API and each method has a fairly simple implementation. Also, we directly expose some of the internal data structures.

These are things that are pointed out as "red flags" in A Philosophy of Software Design. However, I don't have a concrete suggestion here.

cc @pcmoritz

I agree :(

I was actually considering putting the set of driver tasks somewhere else, but I thought it made sense to keep all of the task status information in one place. We should probably rethink some of the methods on the SchedulingQueue.

robertnishihara · 2018-07-21T00:07:06Z

python/ray/worker.py

@@ -1998,6 +2002,8 @@ def connect(info,
    # the correct driver.
    if mode != WORKER_MODE:
        worker.task_driver_id = ray.ObjectID(worker.worker_id)


I'm pretty sure this "driver ID" is different from the task ID of the driver task.

Oops, I mixed this up with the worker.current_task_id. Thanks!

AmplabJenkins · 2018-07-21T00:19:02Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6752/
Test FAILed.

AmplabJenkins · 2018-07-21T02:01:16Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6753/
Test FAILed.

AmplabJenkins · 2018-07-23T14:23:48Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6780/
Test FAILed.

AmplabJenkins · 2018-07-24T22:29:28Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6804/
Test FAILed.

AmplabJenkins · 2018-07-25T02:44:51Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6812/
Test PASSed.

robertnishihara · 2018-07-25T20:50:43Z

src/ray/raylet/node_manager.cc

@@ -766,7 +766,10 @@ void NodeManager::HandleWorkerBlocked(std::shared_ptr<Worker> worker) {
 }

 void NodeManager::HandleWorkerUnblocked(std::shared_ptr<Worker> worker) {
-  RAY_CHECK(worker->IsBlocked());
+  RAY_CHECK(worker);
+  if (!worker->IsBlocked()) {


Why has this changed? Shouldn't this only be called when the worker is blocked?

In legacy ray, a worker always gets marked as blocked if it sends a ReconstructObjects message, so once it sends the NotifyUnblocked message, it's guaranteed to be blocked. The thing that I changed here compared to legacy ray is that the worker only gets marked as blocked if there is an object requested in the ReconstructObjects message that is not local. That means that sometimes, an unblocked worker will send the NotifyUnblocked message.

AmplabJenkins · 2018-07-25T20:51:24Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6837/
Test FAILed.

robertnishihara

Thanks @stephanie-wang, this looks good to me assuming the tests are passing.

robertnishihara · 2018-07-25T21:52:13Z

@eric-jj @kingchin1218 do you understand the Java test failure in the attached log? I'm not really sure how to read what is going wrong.

Log attached.
travis_log_java_error.txt

EDIT: @stephanie-wang in this PR you change the signature of LocalSchedulerConnection_init which is called from the Java code, so I believe that needs to be updated.

robertnishihara · 2018-07-26T02:21:08Z

jenkins, retest this please

robertnishihara · 2018-07-26T02:42:25Z

@stephanie-wang I pushed a few changes showing which part of the Java code needs to be changed. However, the semantics of what I did don't quite match the semantics of the Python side. In particular, on the Python side, we pass in the "driver task ID" and on the Java side I'm not sure if there is a dummy task, so it's passing in something else. Please take a look.

AmplabJenkins · 2018-07-26T03:45:00Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6856/
Test PASSed.

stephanie-wang · 2018-07-26T03:58:02Z

@robertnishihara ah thanks. I took a brief look, but I'm not familiar enough with the code to see how we could do the same thing. @eric-jj @kingchin1218, do you guys know if a driver process gets assigned a dummy task representing the driver? And if so, is there a convenient way to pass it into the Java local scheduler client?

AmplabJenkins · 2018-07-26T04:05:02Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6859/
Test PASSed.

AmplabJenkins · 2018-07-26T16:05:28Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6874/
Test PASSed.

AmplabJenkins · 2018-07-27T03:30:59Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6886/
Test PASSed.

AmplabJenkins · 2018-07-27T05:50:45Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6891/
Test FAILed.

AmplabJenkins · 2018-07-27T16:19:54Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/6908/
Test PASSed.

jovany-wang · 2018-07-28T02:44:59Z

@stephanie-wang @robertnishihara
Sorry guns, I have missed this messages :(
It looks like that you guns have already solved this issue.

I will change the notification-email-addr to my frequently-used one.

raulchen · 2018-07-30T11:28:16Z

src/ray/raylet/worker_pool.cc

+namespace {
+
+// A helper function to remove a worker from a list. Returns true if the worker
+// was found and removed.


it seems that you forgot to update this comment.

## What do these changes do? #2362 left a bug where it assumed that the driver task ID was nil. This fixes the bug to check the `SchedulingQueue` for any driver task IDs instead.

stephanie-wang added 4 commits July 6, 2018 15:03

don't call reconstruct_objects twice in worker.py

88e75ba

SchedulingQueue methods to remove a single task

3cb24b6

Subscribe to task dependencies for a task blocked in ray.get

9f86d3e

Refactor SchedulingQueue enums

a7e481f

stephanie-wang requested review from pcmoritz and robertnishihara July 6, 2018 22:46

doc

9f06f24

robertnishihara reviewed Jul 7, 2018

View reviewed changes

stephanie-wang added 3 commits July 20, 2018 12:52

Fix batching in legacy ray

3d05cdc

Make backend aware of drivers

13815f5

Merge branch 'master' into track-get-dependencies

e3e7118

Merge branch 'master' into track-get-dependencies

407032c

robertnishihara reviewed Jul 20, 2018

View reviewed changes

robertnishihara reviewed Jul 21, 2018

View reviewed changes

fix

c676346

robertnishihara mentioned this pull request Jul 22, 2018

Raise application-level exceptions for some failure scenarios. #2429

Merged

3 tasks

Fix test

21ee113

stephanie-wang added 2 commits July 24, 2018 17:51

lint

ced33cb

Merge branch 'master' into track-get-dependencies

bc740d4

Fix

86b00f8

Fix

ee00169

robertnishihara reviewed Jul 25, 2018

View reviewed changes

robertnishihara approved these changes Jul 25, 2018

View reviewed changes

Pass driver ID into local scheduler client from Java.

d3185d0

Set random task ID for Java driver

ca36b8e

Fix java build

a2f8745

Fix java build

830c06f

lint

5da71af

robertnishihara merged commit 6675361 into ray-project:master Jul 27, 2018

robertnishihara deleted the track-get-dependencies branch July 27, 2018 18:59

robertnishihara mentioned this pull request Jul 27, 2018

[Java Worker] Support raylet on Java #2479

Merged

robertnishihara mentioned this pull request Jul 28, 2018

[xray] Adds a driver table. #2289

Merged

raulchen reviewed Jul 30, 2018

View reviewed changes

stephanie-wang mentioned this pull request Aug 23, 2018

[xray] Fix bug where driver task ID is assumed to be nil #2725

Merged

	// TODO(rkn): This is too heavyweight just to get the task's driver ID.
	auto const it = std::find_if(
	running_tasks.begin(), running_tasks.end(), [task_id](const Task &task) {
	return task.GetTaskSpecification().TaskId() == task_id;
	});
	RAY_CHECK(running_tasks.size() != 0);
	RAY_CHECK(it != running_tasks.end());
	JobID job_id = it->GetTaskSpecification().DriverId();

	worker.current_task_id = ray.ObjectID(np.random.bytes(20))
	# Reset the state of the numpy random number generator.
	np.random.set_state(numpy_state)
	# Set other fields needed for computing task IDs.
	worker.task_index = 0
	worker.put_index = 1

	# Create an entry for the driver task in the task table. This task is
	# added immediately with status RUNNING. This allows us to push errors
	# related to this driver task back to the driver. For example, if the
	# driver creates an object that is later evicted, we should notify the
	# user that we're unable to reconstruct the object, since we cannot
	# rerun the driver.
	nil_actor_counter = 0

	driver_task = ray.local_scheduler.Task(
	worker.task_driver_id, ray.ObjectID(NIL_FUNCTION_ID), [], 0,
	worker.current_task_id, worker.task_index,
	ray.ObjectID(NIL_ACTOR_ID), ray.ObjectID(NIL_ACTOR_ID),
	ray.ObjectID(NIL_ACTOR_ID), ray.ObjectID(NIL_ACTOR_ID),
	nil_actor_counter, False, [], {"CPU": 0}, worker.use_raylet)

	# Add the driver task to the task table.
	if not worker.use_raylet:
	global_state._execute_command(
	driver_task.task_id(), "RAY.TASK_TABLE_ADD",
	driver_task.task_id().id(), TASK_STATUS_RUNNING,
	NIL_LOCAL_SCHEDULER_ID,
	driver_task.execution_dependencies_string(), 0,
	ray.local_scheduler.task_to_string(driver_task))
	else:
	global_state._execute_command(
	driver_task.task_id(), "RAY.TABLE_ADD",
	ray.gcs_utils.TablePrefix.RAYLET_TASK,
	ray.gcs_utils.TablePubsub.RAYLET_TASK,
	driver_task.task_id().id(),
	driver_task._serialized_raylet_task())

	# Set the driver's current task ID to the task ID assigned to the
	# driver task.
	worker.current_task_id = driver_task.task_id()

[xray] Track ray.get calls as task dependencies #2362

[xray] Track ray.get calls as task dependencies #2362

Conversation

stephanie-wang commented Jul 6, 2018

What do these changes do?

AmplabJenkins commented Jul 7, 2018

AmplabJenkins commented Jul 7, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Jul 20, 2018

AmplabJenkins commented Jul 20, 2018

robertnishihara left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Jul 21, 2018

AmplabJenkins commented Jul 21, 2018

AmplabJenkins commented Jul 23, 2018

AmplabJenkins commented Jul 24, 2018

AmplabJenkins commented Jul 25, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Jul 25, 2018

robertnishihara left a comment

Choose a reason for hiding this comment

robertnishihara commented Jul 25, 2018 • edited Loading

robertnishihara commented Jul 26, 2018

robertnishihara commented Jul 26, 2018

AmplabJenkins commented Jul 26, 2018

stephanie-wang commented Jul 26, 2018

AmplabJenkins commented Jul 26, 2018

AmplabJenkins commented Jul 26, 2018

AmplabJenkins commented Jul 27, 2018

AmplabJenkins commented Jul 27, 2018

AmplabJenkins commented Jul 27, 2018

jovany-wang commented Jul 28, 2018

Choose a reason for hiding this comment

[xray] Track `ray.get` calls as task dependencies #2362

[xray] Track `ray.get` calls as task dependencies #2362

robertnishihara commented Jul 25, 2018 •

edited

Loading