Start and clean up workers from the local scheduler. #250

stephanie-wang · 2017-02-07T00:26:19Z

Workers can be started from either the local scheduler or the Python script, as before. Workers send their process IDs to the local scheduler upon initial connection, so that the local scheduler can clean up worker child processes.

robertnishihara · 2017-02-07T00:31:17Z

python/ray/services.py

-  time.sleep(0)
+
+  # Allow the process one second to exit gracefully.
+  p.terminate()


Why reverse the order of kill and terminate? Doesn't it make sense to start with kill?

kill is stronger than terminate. I'm doing terminate first to give the local scheduler a chance to catch the signal and clean up state. Previously it wasn't really an issue because it just needed to free memory that would get cleaned up on exit anyway, but now there are worker processes to clean up.

robertnishihara · 2017-02-07T00:40:47Z

python/ray/worker.py

@@ -1523,6 +1523,7 @@ def process_task(task): # wrapping these lines in a function should cause the lo
                                        "function_name": function_name})

  check_main_thread()
+  worker.photon_client.register_pid()


I think it is more appropriate for this to be part of photon_connect. I needed something similar for the actor PR (to pass an "actor ID" from the worker to the local scheduler), so we should probably use the same mechanism.

I did it like this 5b22750#diff-2a906576526ef8fb8a2de97b6e0bc3ceR7

What do you think about doing it that way? (Your PR will be ready first)

NOTE: the local scheduler should never terminate a driver, so if we do it in photon_connect, we will need the option of passing -1 or something to indicate that it is a driver. (but this will be useful anyway because the local scheduler really ought to know which clients are drivers (for computing resource capacity).

Also, the purpose of this should be documented (e.g., "We pass the process ID to the local scheduler so that the local scheduler can terminate this worker if it wants to.")

Oh yeah, makes sense to do it as part of photon_connect.

The local scheduler won't ever terminate the driver. It only terminates child processes. If you look at the handler for REGISTER_PID messages, it matches up worker clients with child processes according to pid.

robertnishihara · 2017-02-07T00:44:50Z

src/photon/photon_scheduler.c

+void kill_process(pid_t pid) {
+  /* Kill the worker violently. */
+  kill(pid, SIGKILL);
+  waitpid(pid, NULL, 0);


I'm a little nervous that in pathological situations this could cause the local scheduler to hang forever.

Even during normal usage, do we really want to local scheduler to block here?

Yeah, I'm not sure about this either...ideally we would mark the worker as dead, and then clean up the rest of the state after receiving some signal that it died. I can try to look into that.

robertnishihara · 2017-02-07T00:50:06Z

src/photon/photon.h

@@ -50,7 +44,7 @@ typedef struct scheduling_algorithm_state scheduling_algorithm_state;
 *  scheduler. */
 typedef struct {
  /** The script to use when starting a new worker. */
-  char *start_worker_command;
+  const char **start_worker_command;


Why split the command into different strings? Is this so you can use fork instead of popen?

It's for execvp.

robertnishihara · 2017-02-07T00:57:55Z

src/photon/photon_scheduler.c

+  /* We can't start a worker if we don't have the path to the worker script. */
+  CHECK(state->config.start_worker_command != NULL);
+  /* Launch the process to create the worker. */
+  pid_t pid = fork();


Why use fork instead of popen? With fork the new worker process will have some leftover state from the local scheduler, right?

It calls execvp soon after, which if successful, completely replaces the process image.

popen didn't seem necessary in this case, since we're already communicating with the process through sockets. Also, I needed the PID to be returned to the local scheduler so that it could later figure out if a worker process was a child or not.

robertnishihara · 2017-02-08T21:32:57Z

src/photon/photon_scheduler.c

@@ -46,7 +67,7 @@ void kill_worker(local_scheduler_client *worker, bool wait) {
  /* If the worker has registered a process ID with us, use to send a kill
   * signal. Else, the worker will exit since it won't be able to connect. */
  if (worker->pid != 0) {
-    kill_process(worker->pid);
+    kill(pid, SIGKILL);


where is pid defined? Doesn't this need to be worker->pid?

robertnishihara · 2017-02-08T21:39:23Z

python/ray/services.py

+    num_workers_per_local_scheduler[i] += 1
+    num_workers -= 1
+    i += 1
+    i %= num_local_schedulers


Minor, but I would rewrite lines 495-500 as

for i in range(num_local_schedulers): num_workers_per_local_scheduler[i % num_local_schedulers] += 1

That's what it's doing, right?

robertnishihara · 2017-02-08T22:16:01Z

src/photon/photon_scheduler.c

@@ -345,6 +509,8 @@ void new_client_connection(event_loop *loop,
  local_scheduler_client *worker = malloc(sizeof(local_scheduler_client));
  worker->sock = new_socket;
  worker->task_in_progress = NULL;
+  worker->pid = 0;


Can a normal PID equal 0?

Not any worker PID. PID 0 is reserved by the kernel.

robertnishihara · 2017-02-08T22:16:52Z

src/photon/photon.h

@@ -84,6 +81,10 @@ typedef struct {
   *  no task is running on the worker, this will be NULL. This is used to
   *  update the task table. */
  task *task_in_progress;
+  /** The process ID of the client. */


Maybe document that if this is 0, that means that the client has not connected to the local scheduler yet.

robertnishihara · 2017-02-09T00:00:21Z

src/photon/photon_scheduler.c

@@ -500,6 +669,13 @@ int main(int argc, char *argv[]) {
  if (!node_ip_address) {
    LOG_FATAL("please specify the node IP address with -h switch");
  }
+  int num_workers = 0;
+  if (num_workers_str) {
+    num_workers = strtol(num_workers_str, NULL, 10);


We should check that strtol succeeded and that num_workers_str was not malformed

Ability to kill workers in photon scheduler Test for old method of starting workers Common codepath for killing workers Common codepath for killing workers Photon test case for starting and killing workers fix build Fix component failure test Register a worker's pid as part of initial connection Address comments and revert photon_connect Set PATH during travis install Fix

atumanov · 2017-02-10T01:52:30Z

src/photon/test/photon_tests.c

@@ -476,12 +515,77 @@ TEST task_multi_dependency_test(void) {
  PASS();
 }

+TEST start_kill_workers_test(void) {
+  /* Start some workers. */
+  int num_workers = 4;


we should add a separate test (a stress test if you will) that tries to do this with 100 workers. I've tried it with 100 workers just now, and it wasn't working for me.

atumanov · 2017-02-10T01:56:34Z

src/photon/test/photon_tests.c

+  destroy_photon_mock(photon);
+  PASS();
+}
+


would be great to add a test that dynamically changes the number of workers : e.g.,
start 20 workers
repeat {
kill 2 workers
add 1 worker
} until num_workers > 0
This dynamic test will make sure we are not leaking worker state at runtime and that we're accounting for workers correctly (with asserts before/after each _worker operation )

atumanov · 2017-02-10T04:22:41Z

ok, we've discussed with Stephanie that we should do a more graceful shutdown of workers when we try to kill or evict them. It's best to send the worker a SIGTERM signal and implement a SIGTERM handler in the worker to allow the worker to close all of its sockets. In the meantime, the local scheduler schedules a timer to send a followup SIGKILL to the evicted worker after 1ms. The reason for this is that, as things are right now, the worker ends up in an inconsistent state and attempts to communicate on sockets that are still open, fails and generates core files while failing. This is reproduceable on a Mac with num_workers >= 10 . Example core dump:

(lldb) bt
* thread #1: tid = 0x0000, 0x00007fff9b1b5f06 libsystem_kernel.dylib`__pthread_kill + 10, stop reason = signal SIGSTOP
  * frame #0: 0x00007fff9b1b5f06 libsystem_kernel.dylib`__pthread_kill + 10
    frame #1: 0x00007fff931a04ec libsystem_pthread.dylib`pthread_kill + 90
    frame #2: 0x00007fff8bda26df libsystem_c.dylib`abort + 129
    frame #3: 0x0000000105d91281 libphoton.so`photon_connect(photon_socket="/tmp/photon_socket_1115438165") + 321 at photon_client.c:14
    frame #4: 0x0000000105d8e4ee libphoton.so`PyPhotonClient_init(self=0x00000001002ac3d8, args=0x00000001034e3bd0, kwds=0x0000000000000000) + 78 at photon_extension.c:24
    frame #5: 0x0000000100062465 libpython2.7.dylib`type_call + 229
    frame #6: 0x000000010000c761 libpython2.7.dylib`PyObject_Call + 97
    frame #7: 0x00000001000a35a5 libpython2.7.dylib`PyEval_EvalFrameEx + 4405
    frame #8: 0x00000001000ac7dc libpython2.7.dylib`PyEval_EvalCodeEx + 2092
    frame #9: 0x00000001000ac988 libpython2.7.dylib`fast_function + 296
    frame #10: 0x00000001000a338a libpython2.7.dylib`PyEval_EvalFrameEx + 3866
    frame #11: 0x00000001000ac7dc libpython2.7.dylib`PyEval_EvalCodeEx + 2092
    frame #12: 0x00000001000ac856 libpython2.7.dylib`PyEval_EvalCode + 54
    frame #13: 0x00000001000cb3d4 libpython2.7.dylib`PyRun_FileExFlags + 164
    frame #14: 0x00000001000cc709 libpython2.7.dylib`PyRun_SimpleFileExFlags + 409
    frame #15: 0x00000001000e01ca libpython2.7.dylib`Py_Main + 2938
    frame #16: 0x0000000100000f14 python`start + 52

code where this worker is failing:

  pid_t my_pid = getpid();
  int success = write_message(result->conn, REGISTER_PID, sizeof(my_pid),
                              (uint8_t *) &my_pid);
  CHECKM(success == 0, "Unable to register worker with local scheduler");
  return result;
}

Another example core dump:

(lldb) bt
* thread #1: tid = 0x0000, 0x00007fff9b1b5f06 libsystem_kernel.dylib`__pthread_kill + 10, stop reason = signal SIGSTOP
  * frame #0: 0x00007fff9b1b5f06 libsystem_kernel.dylib`__pthread_kill + 10
    frame #1: 0x00007fff931a04ec libsystem_pthread.dylib`pthread_kill + 90
    frame #2: 0x00007fff8bda26df libsystem_c.dylib`abort + 129
    frame #3: 0x0000000105d8f739 libphoton.so`photon_get_task(conn=0x000000010184fc20) + 297 at photon_client.c:58
    frame #4: 0x0000000105d8c5b2 libphoton.so`PyPhotonClient_get_task(self=0x00000001002ac3d8) + 34 at photon_extension.c:49
    frame #5: 0x00000001000a2f66 libpython2.7.dylib`PyEval_EvalFrameEx + 2806
    frame #6: 0x00000001000ac7dc libpython2.7.dylib`PyEval_EvalCodeEx + 2092
    frame #7: 0x00000001000ac988 libpython2.7.dylib`fast_function + 296
    frame #8: 0x00000001000a338a libpython2.7.dylib`PyEval_EvalFrameEx + 3866
    frame #9: 0x00000001000ac7dc libpython2.7.dylib`PyEval_EvalCodeEx + 2092
    frame #10: 0x00000001000ac856 libpython2.7.dylib`PyEval_EvalCode + 54
    frame #11: 0x00000001000cb3d4 libpython2.7.dylib`PyRun_FileExFlags + 164
    frame #12: 0x00000001000cc709 libpython2.7.dylib`PyRun_SimpleFileExFlags + 409
    frame #13: 0x00000001000e01ca libpython2.7.dylib`Py_Main + 2938
    frame #14: 0x0000000100000f14 python`start + 52

Here, the worker is trying to send the GET_TASK message to photon .

Takeaway, we should let the worker clean up its socket connections and gracefully self-terminate.

atumanov · 2017-02-10T04:36:46Z

src/photon/photon_scheduler.c

+
+  /* Clean up the task in progress. */
+  if (worker->task_in_progress) {
+    /* TODO(swang): Update the task table to mark the task as lost. */


So, this will stall all the downstream tasks that depend on any objects produced by this task_in_progress. We should really have a mechanism for re-scheduling the task in flight. One possibility is to
just update the task table entry for this task with status = TASK_STATUS_KILLED and subscribe to this task status in the global scheduler. The GS handler would then re-schedule the task.
Without this, we are likely to see hangs.

also, this task will never report done back to photon, so we need to do resource accounting for this task. Otherwise we will leak dynamic resources.

basically, you need this code in this if block:

task_spec *spec = task_task_spec(worker->task_in_progress); /* Return dynamic resources back for the task in progress. */ for (int i = 0; i < MAX_RESOURCE_INDEX; i++) { state->dynamic_resources[i] += task_spec_get_required_resource(spec, i); /* Sanity-check resource vector boundary conditions. */ CHECK(state->dynamic_resources[i] <= state->static_resources[i]); }

Yes, this PR is a little incomplete right now, since workers are only killed during cleanup, when reporting task death doesn't matter anymore. I'm hoping to implement reporting task death properly in a separate PR for task preemption.

the resource accounting bit should still be addressed. I'm happy to do it in a separate PR, along with photon_tests updates that would expose and test this condition.

robertnishihara · 2017-02-11T06:42:25Z

src/photon/photon_scheduler.c

+ *        ignored.
+ * @return Void.
+ */
+void kill_worker(local_scheduler_client *worker, bool wait) {


When we kill a worker, we probably need to let the scheduling algorithm do some clean up as well (in particular, the scheduling algorithm needs to remove the worker struct from its list of available workers, otherwise it will try to assign a task to that worker).

robertnishihara reviewed Feb 7, 2017

View reviewed changes

robertnishihara reviewed Feb 8, 2017

View reviewed changes

robertnishihara reviewed Feb 9, 2017

View reviewed changes

stephanie-wang force-pushed the start-workers branch from 434b11e to 391dc29 Compare February 9, 2017 21:09

stephanie-wang added 2 commits February 9, 2017 17:43

Fix photon test case to accept clients on plasma manager fd

a676b8a

stephanie-wang force-pushed the start-workers branch from eda4b68 to a676b8a Compare February 10, 2017 01:44

atumanov reviewed Feb 10, 2017

View reviewed changes

robertnishihara merged commit 2b8e648 into ray-project:master Feb 10, 2017

robertnishihara deleted the start-workers branch February 10, 2017 20:46

robertnishihara reviewed Feb 11, 2017

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Start and clean up workers from the local scheduler. #250

Start and clean up workers from the local scheduler. #250

stephanie-wang commented Feb 7, 2017

robertnishihara Feb 7, 2017

stephanie-wang Feb 7, 2017

robertnishihara Feb 7, 2017

robertnishihara Feb 7, 2017

stephanie-wang Feb 7, 2017

robertnishihara Feb 7, 2017

stephanie-wang Feb 7, 2017

robertnishihara Feb 7, 2017

stephanie-wang Feb 7, 2017

robertnishihara Feb 7, 2017

stephanie-wang Feb 7, 2017

robertnishihara Feb 8, 2017

robertnishihara Feb 8, 2017

robertnishihara Feb 8, 2017

stephanie-wang Feb 8, 2017

robertnishihara Feb 8, 2017

robertnishihara Feb 9, 2017

atumanov Feb 10, 2017

atumanov Feb 10, 2017

atumanov commented Feb 10, 2017

atumanov Feb 10, 2017

atumanov Feb 10, 2017

atumanov Feb 10, 2017

stephanie-wang Feb 10, 2017

atumanov Feb 11, 2017

robertnishihara Feb 11, 2017

Start and clean up workers from the local scheduler. #250

Start and clean up workers from the local scheduler. #250

Conversation

stephanie-wang commented Feb 7, 2017

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

atumanov commented Feb 10, 2017

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