Give better error message when no workers have been registered. (#419)

protos/ray.proto

@@ -105,7 +105,8 @@ message SubmitTaskRequest {
 
 message SubmitTaskReply {
   repeated uint64 result = 1; // Object IDs of the function return values
-  bool function_registered = 2; // True if the function was registered; false otherwise
+  bool function_registered = 2; // True if the function was registered, false otherwise
+  bool no_workers = 3; // True if no workers have registered with the scheduler, false otherwise
 }
 
 message RequestObjRequest {

src/raylib.cc

@@ -785,12 +785,15 @@ static PyObject* submit_task(PyObject* self, PyObject* args) {
   SubmitTaskRequest request;
   request.set_allocated_task(task);
   SubmitTaskReply reply = worker->submit_task(&request);
+  request.release_task(); // TODO: Make sure that task is not moved, otherwise capsule pointer needs to be updated
+  if (reply.no_workers()) {
+    PyErr_SetString(RayError, "No workers have registered with the scheduler, so this function cannot be run.");
+    return NULL;
+  }
   if (!reply.function_registered()) {
-    request.release_task();
-    PyErr_SetString(RayError, "task: function not registered");
+    PyErr_SetString(RayError, "No worker has registered this function with the scheduler.");
     return NULL;
   }
-  request.release_task(); // TODO: Make sure that task is not moved, otherwise capsule pointer needs to be updated
   int size = reply.result_size();
   PyObject* list = PyList_New(size);
   std::vector<ObjectID> result_objectids;

src/scheduler.cc

@@ -116,12 +116,29 @@ SchedulerService::SchedulerService(SchedulingAlgorithmType scheduling_algorithm)
 Status SchedulerService::SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) {
   std::unique_ptr<Task> task(new Task(request->task())); // need to copy, because request is const
   size_t num_return_vals;
+  // If there are no workers, then we will set this to true below.
+  reply->set_no_workers(false);
   {
     auto fntable = GET(fntable_);
     FnTable::const_iterator fn = fntable->find(task->name());
     if (fn == fntable->end()) {
       num_return_vals = 0;
       reply->set_function_registered(false);
+      // Check if there are any workers registered with the scheduler, so that
+      // we can tell the worker if there aren't so that it can display a better
+      // error message.
+      int num_live_workers = 0;
+      auto workers = GET(workers_);
+      for (size_t i = 0; i < workers->size(); ++i) {
+        WorkerHandle* worker = &(*workers)[i];
+        // Check if this is a driver and that it is still connected.
+        if (worker->current_task != ROOT_OPERATION && worker->worker_stub) {
+          num_live_workers += 1;
+        }
+      }
+      if (num_live_workers == 0) {
+        reply->set_no_workers(true);
+      }
     } else {
       num_return_vals = fn->second.num_return_vals();
       reply->set_function_registered(true);

src/worker.h

@@ -53,7 +53,7 @@ class Worker {
   // Submit a remote task to the scheduler. If the function in the task is not
   // registered with the scheduler, we will sleep for retry_wait_milliseconds
   // and try to resubmit the task to the scheduler up to max_retries more times.
-  SubmitTaskReply submit_task(SubmitTaskRequest* request, int max_retries = 120, int retry_wait_milliseconds = 500);
+  SubmitTaskReply submit_task(SubmitTaskRequest* request, int max_retries = 10, int retry_wait_milliseconds = 500);
   // Requests the scheduler to kill workers
   bool kill_workers(ClientContext &context);
   // send request to the scheduler to register this worker