Raylet task dispatch and throttling worker startup #1912

atumanov · 2018-04-16T21:42:53Z

What do these changes do?

There are two parts to this PR:

separation of task placement and local task dispatch decisions in the scheduler
worker startup throttling, by keeping track of workers in the process of being started.

Minor:

kill raylet_monitor on ray stop

…h locally available resournces

…extraneous workers

AmplabJenkins · 2018-04-16T22:11: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/4952/
Test FAILed.

stephanie-wang

Looks good! I left a few low-level comments.

Just so we can start thinking about it, does it make sense to put ScheduleTasks on a timer in the future?

stephanie-wang · 2018-04-17T00:38:28Z

python/ray/scripts/scripts.py

@@ -306,7 +306,7 @@ def stop():
    subprocess.call(
        [
            "killall global_scheduler plasma_store plasma_manager "
-            "local_scheduler raylet"
+            "local_scheduler raylet raylet_monitor"


Oops, thanks :)

stephanie-wang · 2018-04-17T00:40:14Z

src/ray/raylet/node_manager.cc

+    return;
+  }
+  // Early return if there are no resources available.
+  const ClientID &my_client_id = gcs_client_->client_table().GetLocalClientId();


Does it make sense to move this inside the for loop, in case the resources go to zero after assignment of a task?

stephanie-wang · 2018-04-17T00:40:52Z

src/ray/raylet/node_manager.cc

@@ -386,7 +404,8 @@ void NodeManager::ScheduleTasks() {

  // Extract decision for this local scheduler.
  std::unordered_set<TaskID, UniqueIDHasher> local_task_ids;
-  // Iterate over (taskid, clientid) pairs, extract tasks to run on the local client.
+  // Iterate over (taskid, clientid) pairs, extract tasks assigned to the local node.
+  // TODO(atumanov): move the assigned tasks to scheduled and call DispatchTasks().


Is this TODO still valid?

stephanie-wang · 2018-04-17T00:44:58Z

src/ray/raylet/worker_pool.cc

+  return static_cast<uint32_t>(actor_pool_.size() + pool_.size());
+}
+
+void WorkerPool::StartWorker(bool force_start) {


The force_start parameter doesn't seem to be used. Can we remove this?

it's not used yet, but I think it will be useful soon. The idea here is to let the caller decide if they want to start a worker no matter what (i.e., if we want raylet to pre-start num_cpus workers, without considering in-flight worker pool. Otherwise, starting num_cpus workers will be serialized.

btw, to be clear, the implementation does use this flag.

I see. Couldn't we just do the check for started_worker_pids.empty() outside of WorkerPool::Start() though, through your modified WorkerPool::Size()? And in the pre-start case, we can just call StartWorker without checking the worker pool size at all.

I thought about it and decided against it, because I'd prefer to encapsulate this as an implementation detail and not expose the in-flight worker status through the public interface of the worker pool. I just think it'd be preferable for us to keep thinking about the worker pool as an opaque, self-managing container of workers.

I know that this is how we do it in the local_scheduler, but we're in the proper OO land now :) If I had to choose, I'd rather drop the flag than expose the in-flight worker status. The disadvantage of dropping the flag is that all calls to StartWorker will serialize starting k workers.

Ah, I see. Okay, I'm fine with this. Can you document the force_start parameter in the header file though?

by the way, with the latest change, the force_start is now used in the constructor to speed up starting num_workers workers.

stephanie-wang · 2018-04-17T00:49:02Z

src/ray/raylet/node_manager.cc

+    }
+    // We have enough resources for this task. Assign task.
+    // TODO(atumanov): perform the task state/queue transition inside AssignTask.
+    auto scheduled_tasks =


Can we call this something other than scheduled_tasks? It's a little confusing given the other scheduled_tasks local variable here.

atumanov · 2018-04-17T05:24:18Z

retest this please

stephanie-wang · 2018-04-17T05:24:25Z

src/ray/raylet/worker_pool.cc

+  return static_cast<uint32_t>(actor_pool_.size() + pool_.size());
+}
+
+void WorkerPool::StartWorker(bool force_start) {


I see. Couldn't we just do the check for started_worker_pids.empty() outside of WorkerPool::Start() though, through your modified WorkerPool::Size()? And in the pre-start case, we can just call StartWorker without checking the worker pool size at all.

stephanie-wang · 2018-04-17T05:27:00Z

src/ray/raylet/node_manager.cc

+        cluster_resource_map_[my_client_id].GetAvailableResources();
+    if (local_resources.IsEmpty()) {
+      // Early return if there are no resources available.
+      return;


Hmm, I just realized there's an issue here where a task that requires zero resources (e.g., an actor task) might not get scheduled. Was that why you had this check outside of the loop at first? Although even having it outside of the loop wouldn't fix this issue...

Not sure what the best approach here is. Should we just always loop through all the tasks for now?

honestly, I think zero-resource tasks are just wrong... How can you have an active thread of execution running without having it acquire at least some CPU resource?
Yes, this is why I was asking about actor tasks. Yes, I think we have no choice but to drop the early termination check.

AmplabJenkins · 2018-04-17T06:09:04Z

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

AmplabJenkins · 2018-04-17T06:28:08Z

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

…tor tasks

AmplabJenkins · 2018-04-17T07:15:14Z

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

AmplabJenkins · 2018-04-17T07:58:57Z

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

AmplabJenkins · 2018-04-17T08:08:44Z

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

AmplabJenkins · 2018-04-17T20:36:26Z

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

stephanie-wang · 2018-04-17T20:35:37Z

src/ray/raylet/worker_pool.cc

@@ -16,27 +16,46 @@ WorkerPool::WorkerPool(int num_workers, const std::vector<std::string> &worker_c
  // become zombies instead of dying gracefully.
  signal(SIGCHLD, SIG_IGN);
  for (int i = 0; i < num_workers; i++) {
-    StartWorker();
+    // Force-start num_workers workers.
+    StartWorker(true);


I learned this tip from Zongheng to write this as StartWorker(/*force_start*/=true) to make it clear to the reader what the parameter is. :)

doesn't this introduce C-style comments? I'd rather not, if there's a cleaner, pythonic way, that'd be great.

stephanie-wang · 2018-04-17T20:39:50Z

src/ray/raylet/worker_pool.cc

 WorkerPool::~WorkerPool() {
  // Kill all registered workers. NOTE(swang): This assumes that the registered
  // workers were started by the pool.
+  // TODO(atumanov): remove killed workers from the pool.


Does this TODO mean we should also kill all the PIDs in started_worker_pids_? Either way, we should probably do that here in this PR.

stephanie-wang · 2018-04-17T20:42:09Z

src/ray/raylet/worker_pool_test.cc

+      : WorkerPool(worker_command) {}
+
+  void StartWorker(pid_t pid, bool force_start = false) {
+    AddStartedWorker(pid);


We probably just want to keep the second call to AddStartedWorker, right?

AmplabJenkins · 2018-04-17T20:49:49Z

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

stephanie-wang

Can you remove the TODO that you fixed? Looks good!

AmplabJenkins · 2018-04-17T22:02:56Z

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

AmplabJenkins · 2018-04-17T22:16:30Z

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

AmplabJenkins · 2018-04-18T06:40:34Z

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

* master: Handle interrupts correctly for ASIO synchronous reads and writes. (ray-project#1929) [DataFrame] Adding read methods and tests (ray-project#1712) Allow task_table_update to fail when tasks are finished. (ray-project#1927) [rllib] Contribute DDPG to RLlib (ray-project#1877) [xray] Workers blocked in a `ray.get` release their resources (ray-project#1920) Raylet task dispatch and throttling worker startup (ray-project#1912) [DataFrame] Eval fix (ray-project#1903) [tune] Polishing docs (ray-project#1846) [tune] [rllib] Automatically determine RLlib resources and add queueing mechanism for autoscaling (ray-project#1848) Preemptively push local arguments for actor tasks (ray-project#1901) [tune] Allow fetching pinned objects from trainable functions (ray-project#1895) Multithreading refactor for ObjectManager. (ray-project#1911) Add slice functionality (ray-project#1832) [DataFrame] Pass read_csv kwargs to _infer_column (ray-project#1894) Addresses missed comments from multichunk object transfer PR. (ray-project#1908) Allow numpy arrays to be passed by value into tasks (and inlined in the task spec). (ray-project#1816) [xray] Lineage cache requests notifications from the GCS about remote tasks (ray-project#1834) Fix UI issue for non-json-serializable task arguments. (ray-project#1892) Remove unnecessary calls to .hex() for object IDs. (ray-project#1910) Allow multiple raylets to be started on a single machine. (ray-project#1904) # Conflicts: # python/ray/rllib/__init__.py # python/ray/rllib/dqn/dqn.py

* master: updates (ray-project#1958) Pin Cython in autoscaler development example. (ray-project#1951) Incorporate C++ Buffer management and Seal global threadpool fix from arrow (ray-project#1950) [XRay] Add consistency check for protocol between node_manager and local_scheduler_client (ray-project#1944) Remove smart_open install. (ray-project#1943) [DataFrame] Fully implement append, concat and join (ray-project#1932) [DataFrame] Fix for __getitem__ string indexing (ray-project#1939) [DataFrame] Implementing write methods (ray-project#1918) [rllib] arr[end] was excluded when end is not None (ray-project#1931) [DataFrame] Implementing API correct groupby with aggregation methods (ray-project#1914) Handle interrupts correctly for ASIO synchronous reads and writes. (ray-project#1929) [DataFrame] Adding read methods and tests (ray-project#1712) Allow task_table_update to fail when tasks are finished. (ray-project#1927) [rllib] Contribute DDPG to RLlib (ray-project#1877) [xray] Workers blocked in a `ray.get` release their resources (ray-project#1920) Raylet task dispatch and throttling worker startup (ray-project#1912) [DataFrame] Eval fix (ray-project#1903)

atumanov added 2 commits April 16, 2018 13:44

separate task placement and task dispatch; throttle task dispatch wit…

d27ccdf

…h locally available resournces

keep track of worker's being started/in flight and suppress starting …

c503917

…extraneous workers

atumanov requested a review from stephanie-wang April 16, 2018 21:42

stephanie-wang requested changes Apr 17, 2018

View reviewed changes

cleanup comments

d1af21a

stephanie-wang reviewed Apr 17, 2018

View reviewed changes

atumanov added 4 commits April 16, 2018 23:33

remove early termination in task dispatch to support zero-resource ac…

84638b9

…tor tasks

info -> debug

21c559f

add documentation

fce161f

linting

188a60b

atumanov added 2 commits April 17, 2018 12:48

mock the worker pool for testing

0329905

some linting

4236a8f

stephanie-wang reviewed Apr 17, 2018

View reviewed changes

kill all workers in flight; clear the worker pool in dtor

6bc95af

stephanie-wang approved these changes Apr 17, 2018

View reviewed changes

remove fixed todo

ae0d653

lint

2562a11

stephanie-wang merged commit 1c965fc into ray-project:master Apr 18, 2018

stephanie-wang deleted the raylet_task_dispatch branch April 18, 2018 17:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Raylet task dispatch and throttling worker startup #1912

Raylet task dispatch and throttling worker startup #1912

atumanov commented Apr 16, 2018 •

edited

Loading

AmplabJenkins commented Apr 16, 2018

stephanie-wang left a comment

stephanie-wang Apr 17, 2018

stephanie-wang Apr 17, 2018

stephanie-wang Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov Apr 17, 2018

atumanov Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov Apr 17, 2018

atumanov Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov commented Apr 17, 2018

stephanie-wang Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

stephanie-wang Apr 17, 2018

atumanov Apr 17, 2018

stephanie-wang Apr 17, 2018

stephanie-wang Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

stephanie-wang left a comment

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 18, 2018

Raylet task dispatch and throttling worker startup #1912

Raylet task dispatch and throttling worker startup #1912

Conversation

atumanov commented Apr 16, 2018 • edited Loading

What do these changes do?

AmplabJenkins commented Apr 16, 2018

stephanie-wang 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

atumanov commented Apr 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 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

AmplabJenkins commented Apr 17, 2018

stephanie-wang left a comment

Choose a reason for hiding this comment

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 17, 2018

AmplabJenkins commented Apr 18, 2018

atumanov commented Apr 16, 2018 •

edited

Loading