diff --git a/python/ray/worker.py b/python/ray/worker.py index 8a16fd3dc..926dc0730 100644 --- a/python/ray/worker.py +++ b/python/ray/worker.py @@ -377,6 +377,14 @@ class Worker(object): timeout, self.serialization_context) return results + except pyarrow.lib.ArrowInvalid as e: + # TODO(ekl): the local scheduler could include relevant + # metadata in the task kill case for a better error message + invalid_error = RayTaskError( + "", None, + "Invalid return value: likely worker died or was killed " + "while executing the task.") + return [invalid_error] * len(object_ids) except pyarrow.DeserializationCallbackError as e: # Wait a little bit for the import thread to import the class. # If we currently have the worker lock, we need to release it diff --git a/src/common/state/actor_notification_table.cc b/src/common/state/actor_notification_table.cc index d405d6263..1d416365a 100644 --- a/src/common/state/actor_notification_table.cc +++ b/src/common/state/actor_notification_table.cc @@ -15,3 +15,7 @@ void actor_notification_table_subscribe( init_table_callback(db_handle, NIL_ID, __func__, sub_data, retry, NULL, redis_actor_notification_table_subscribe, NULL); } + +void actor_table_mark_removed(DBHandle *db_handle, ActorID actor_id) { + redis_actor_table_mark_removed(db_handle, actor_id); +} diff --git a/src/common/state/actor_notification_table.h b/src/common/state/actor_notification_table.h index 87eddde37..c287722e6 100644 --- a/src/common/state/actor_notification_table.h +++ b/src/common/state/actor_notification_table.h @@ -41,4 +41,13 @@ typedef struct { void *subscribe_context; } ActorNotificationTableSubscribeData; +/** + * Marks an actor as removed. This prevents the actor from being resurrected. + * + * @param db The database handle. + * @param actor_id The actor id to mark as removed. + * @return Void. + */ +void actor_table_mark_removed(DBHandle *db_handle, ActorID actor_id); + #endif /* ACTOR_NOTIFICATION_TABLE_H */ diff --git a/src/common/state/redis.cc b/src/common/state/redis.cc index b6557bd6e..abee2ac5a 100644 --- a/src/common/state/redis.cc +++ b/src/common/state/redis.cc @@ -1576,6 +1576,15 @@ void redis_actor_notification_table_subscribe( } } +void redis_actor_table_mark_removed(DBHandle *db, ActorID actor_id) { + int status = + redisAsyncCommand(db->context, NULL, NULL, "HSET Actor:%b removed \"1\"", + actor_id.id, sizeof(actor_id.id)); + if ((status == REDIS_ERR) || db->subscribe_context->err) { + LOG_REDIS_DEBUG(db->context, "error in redis_actor_table_mark_removed"); + } +} + void redis_object_info_subscribe_callback(redisAsyncContext *c, void *r, void *privdata) { diff --git a/src/common/state/redis.h b/src/common/state/redis.h index f422c9baa..ad2a40442 100644 --- a/src/common/state/redis.h +++ b/src/common/state/redis.h @@ -321,6 +321,15 @@ void redis_driver_table_send_driver_death(TableCallbackData *callback_data); void redis_plasma_manager_send_heartbeat(TableCallbackData *callback_data); +/** + * Marks an actor as removed. This prevents the actor from being resurrected. + * + * @param db The database handle. + * @param actor_id The actor id to mark as removed. + * @return Void. + */ +void redis_actor_table_mark_removed(DBHandle *db, ActorID actor_id); + /** * Subscribe to updates about newly created actors. * diff --git a/src/local_scheduler/local_scheduler.cc b/src/local_scheduler/local_scheduler.cc index b1d16c2a0..a71c36e64 100644 --- a/src/local_scheduler/local_scheduler.cc +++ b/src/local_scheduler/local_scheduler.cc @@ -80,8 +80,8 @@ void kill_worker(LocalSchedulerState *state, handle_worker_removed(state, state->algorithm_state, worker); } else { /* Let the scheduling algorithm process the absence of this worker. */ - handle_actor_worker_disconnect(state, state->algorithm_state, - worker->actor_id); + handle_actor_worker_disconnect(state, state->algorithm_state, worker, + cleanup); } /* Remove the client socket from the event loop so that we don't process the diff --git a/src/local_scheduler/local_scheduler_algorithm.cc b/src/local_scheduler/local_scheduler_algorithm.cc index dffe11ab1..814627f25 100644 --- a/src/local_scheduler/local_scheduler_algorithm.cc +++ b/src/local_scheduler/local_scheduler_algorithm.cc @@ -5,6 +5,7 @@ #include #include "state/task_table.h" +#include "state/actor_notification_table.h" #include "state/local_scheduler_table.h" #include "state/object_table.h" #include "local_scheduler_shared.h" @@ -399,6 +400,33 @@ void handle_actor_worker_connect(LocalSchedulerState *state, dispatch_actor_task(state, algorithm_state, actor_id); } +/** + * Finishes a killed task by inserting dummy objects for each of its returns. + */ +void finish_killed_task(LocalSchedulerState *state, + TaskSpec *spec, + int64_t task_spec_size) { + int64_t num_returns = TaskSpec_num_returns(spec); + for (int i = 0; i < num_returns; i++) { + ObjectID object_id = TaskSpec_return(spec, i); + uint8_t *data = NULL; + // TODO(ekl): this writes an invalid arrow object, which is sufficient to + // signal that the worker failed, but it would be nice to return more + // detailed failure metadata in the future. + Status status = + state->plasma_conn->Create(object_id.to_plasma_id(), 1, NULL, 0, &data); + if (!status.IsPlasmaObjectExists()) { + ARROW_CHECK_OK(status); + ARROW_CHECK_OK(state->plasma_conn->Seal(object_id.to_plasma_id())); + } + if (state->db != NULL) { + Task *task = Task_alloc(spec, task_spec_size, TASK_STATUS_DONE, + get_db_client_id(state->db)); + task_table_update(state->db, task, NULL, NULL, NULL); + } + } +} + /** * Insert a task queue entry into an actor's dispatch queue. The task is * inserted in sorted order by task counter. If this is the first task @@ -418,6 +446,12 @@ void insert_actor_task_queue(LocalSchedulerState *state, ActorID task_handle_id = TaskSpec_actor_handle_id(task_entry.spec); int64_t task_counter = TaskSpec_actor_counter(task_entry.spec); + /* Fail the task immediately; it's destined for a dead actor. */ + if (state->removed_actors.find(actor_id) != state->removed_actors.end()) { + finish_killed_task(state, task_entry.spec, task_entry.task_spec_size); + return; + } + /* Handle the case in which there is no LocalActorInfo struct yet. */ if (algorithm_state->local_actor_infos.count(actor_id) == 0) { /* Create the actor struct with a NULL worker because the worker struct has @@ -492,10 +526,6 @@ void queue_actor_task(LocalSchedulerState *state, ActorID actor_id = TaskSpec_actor_id(spec); DCHECK(!ActorID_equal(actor_id, NIL_ACTOR_ID)); - /* Create a new task queue entry. */ - TaskQueueEntry elt = TaskQueueEntry_init(spec, task_spec_size); - insert_actor_task_queue(state, algorithm_state, elt); - /* Update the task table. */ if (state->db != NULL) { Task *task = Task_alloc(spec, task_spec_size, TASK_STATUS_QUEUED, @@ -512,6 +542,11 @@ void queue_actor_task(LocalSchedulerState *state, } } + // Create a new task queue entry. This must come after the above block because + // insert_actor_task_queue may call task_table_update internally, which must + // come after the prior call to task_table_add_task. + TaskQueueEntry elt = TaskQueueEntry_init(spec, task_spec_size); + insert_actor_task_queue(state, algorithm_state, elt); } /** @@ -1264,8 +1299,30 @@ void handle_worker_removed(LocalSchedulerState *state, void handle_actor_worker_disconnect(LocalSchedulerState *state, SchedulingAlgorithmState *algorithm_state, - ActorID actor_id) { - remove_actor(algorithm_state, actor_id); + LocalSchedulerClient *worker, + bool cleanup) { + /* Fail all in progress or queued tasks of the actor. */ + if (!cleanup) { + if (state->db != NULL) { + actor_table_mark_removed(state->db, worker->actor_id); + } + + if (worker->task_in_progress != NULL) { + TaskSpec *spec = Task_task_spec(worker->task_in_progress); + finish_killed_task(state, spec, worker->task_in_progress->task_spec_size); + } + + state->removed_actors.insert(worker->actor_id); + + CHECK(algorithm_state->local_actor_infos.count(worker->actor_id) != 0); + LocalActorInfo &entry = + algorithm_state->local_actor_infos.find(worker->actor_id)->second; + for (auto &task : *entry.task_queue) { + finish_killed_task(state, task.spec, task.task_spec_size); + } + } + + remove_actor(algorithm_state, worker->actor_id); /* Attempt to dispatch some tasks because some resources may have freed up. */ dispatch_all_tasks(state, algorithm_state); diff --git a/src/local_scheduler/local_scheduler_algorithm.h b/src/local_scheduler/local_scheduler_algorithm.h index d845df92a..67f97f17d 100644 --- a/src/local_scheduler/local_scheduler_algorithm.h +++ b/src/local_scheduler/local_scheduler_algorithm.h @@ -197,12 +197,14 @@ void handle_actor_worker_connect(LocalSchedulerState *state, * * @param state The state of the local scheduler. * @param algorithm_state State maintained by the scheduling algorithm. - * @param actor_id The ID of the actor running on the worker. + * @param worker The worker that was disconnected. + * @param cleanup Whether the disconnect was during cleanup. * @return Void. */ void handle_actor_worker_disconnect(LocalSchedulerState *state, SchedulingAlgorithmState *algorithm_state, - ActorID actor_id); + LocalSchedulerClient *worker, + bool cleanup); /** * This function is called when a worker that was executing a task becomes diff --git a/src/local_scheduler/local_scheduler_shared.h b/src/local_scheduler/local_scheduler_shared.h index 0e066ce36..2232e1267 100644 --- a/src/local_scheduler/local_scheduler_shared.h +++ b/src/local_scheduler/local_scheduler_shared.h @@ -48,6 +48,9 @@ struct LocalSchedulerState { /** A set of driver IDs corresponding to drivers that have been removed. This * is used to make sure we don't execute any tasks belong to dead drivers. */ std::unordered_set removed_drivers; + /** A set of actors IDs corresponding to local actors that have been removed. + * This ensures we can reject any tasks destined for dead actors. */ + std::unordered_set removed_actors; /** List of the process IDs for child processes (workers) started by the * local scheduler that have not sent a REGISTER_PID message yet. */ std::vector child_pids; diff --git a/test/failure_test.py b/test/failure_test.py index 8e55e825d..85bdeea2b 100644 --- a/test/failure_test.py +++ b/test/failure_test.py @@ -277,6 +277,67 @@ class WorkerDeath(unittest.TestCase): ray.worker.cleanup() + def testActorWorkerDying(self): + ray.init(num_workers=0, driver_mode=ray.SILENT_MODE) + + @ray.remote + class Actor(object): + def kill(self): + eval("exit()") + + @ray.remote + def consume(x): + pass + + a = Actor.remote() + [obj], _ = ray.wait([a.kill.remote()], timeout=5000) + self.assertRaises(Exception, lambda: ray.get(obj)) + self.assertRaises(Exception, lambda: ray.get(consume.remote(obj))) + wait_for_errors(b"worker_died", 1) + + ray.worker.cleanup() + + def testActorWorkerDyingFutureTasks(self): + ray.init(num_workers=0, driver_mode=ray.SILENT_MODE) + + @ray.remote + class Actor(object): + def getpid(self): + return os.getpid() + + def sleep(self): + time.sleep(1) + + a = Actor.remote() + pid = ray.get(a.getpid.remote()) + tasks1 = [a.sleep.remote() for _ in range(10)] + os.kill(pid, 9) + time.sleep(0.1) + tasks2 = [a.sleep.remote() for _ in range(10)] + for obj in tasks1 + tasks2: + self.assertRaises(Exception, lambda: ray.get(obj)) + + wait_for_errors(b"worker_died", 1) + + ray.worker.cleanup() + + def testActorWorkerDyingNothingInProgress(self): + ray.init(num_workers=0, driver_mode=ray.SILENT_MODE) + + @ray.remote + class Actor(object): + def getpid(self): + return os.getpid() + + a = Actor.remote() + pid = ray.get(a.getpid.remote()) + os.kill(pid, 9) + time.sleep(0.1) + task2 = a.getpid.remote() + self.assertRaises(Exception, lambda: ray.get(task2)) + + ray.worker.cleanup() + class PutErrorTest(unittest.TestCase):