Changing hard coded ports for objstore and workers to choose unused ports (#365)

* let grpc choose unused worker and object store ports

* Add objstore addresses to scheduler info to bring back test

src/ipc.cc

@@ -6,6 +6,7 @@
 
 #include <stdlib.h>
 #include "ray/ray.h"
+#include "utils.h"
 
 ObjHandle::ObjHandle(SegmentId segmentid, size_t size, IpcPointer ipcpointer, size_t metadata_offset)
   : segmentid_(segmentid), size_(size), ipcpointer_(ipcpointer), metadata_offset_(metadata_offset)
@@ -82,13 +83,16 @@ bool MessageQueue<>::receive(void * object, size_t size) {
   return true;
 }
 
-MemorySegmentPool::MemorySegmentPool(ObjStoreId objstoreid, bool create) : objstoreid_(objstoreid), create_mode_(create) { }
+MemorySegmentPool::MemorySegmentPool(ObjStoreId objstoreid, std::string& objstore_address, bool create) : objstoreid_(objstoreid), objstore_address_(objstore_address), create_mode_(create) {
+  std::string::iterator split_point = split_ip_address(objstore_address);
+  objstore_port_.assign(split_point, objstore_address.end());
+}
 
 // creates a memory segment if it is not already there; if the pool is in create mode,
 // space is allocated, if it is in open mode, the shared memory is mapped into the process
 void MemorySegmentPool::open_segment(SegmentId segmentid, size_t size) {
-  RAY_LOG(RAY_DEBUG, "Opening segmentid " << segmentid << " on object store " << objstoreid_ << " with create_mode_ = " << create_mode_);
-  RAY_CHECK(segmentid == segments_.size() || !create_mode_, "Object store " << objstoreid_ << " is attempting to open segmentid " << segmentid << " on the object store, but segments_.size() = " << segments_.size());
+  RAY_LOG(RAY_DEBUG, "Opening segmentid " << segmentid << " on object store " << objstoreid_ << " with port " << objstore_port_ << " with create_mode_ = " << create_mode_);
+  RAY_CHECK(segmentid == segments_.size() || !create_mode_, "Object store " << objstoreid_ << " with port " << objstore_port_ << " is attempting to open segmentid " << segmentid << " on the object store, but segments_.size() = " << segments_.size());
   if (segmentid >= segments_.size()) { // resize and initialize segments_
     int current_size = segments_.size();
     segments_.resize(segmentid + 1);
@@ -156,7 +160,7 @@ uint8_t* MemorySegmentPool::get_address(ObjHandle pointer) {
 
 // returns the name of the segment
 std::string MemorySegmentPool::get_segment_name(SegmentId segmentid) {
-  return std::string("ray-{BC200A09-2465-431D-AEC7-2F8530B04535}-objstore-") + std::to_string(objstoreid_) + std::string("-segment-") + std::to_string(segmentid);
+  return std::string("ray-{BC200A09-2465-431D-AEC7-2F8530B04535}-objstore-") + std::to_string(objstoreid_) + "-" + objstore_port_ + std::string("-segment-") + std::to_string(segmentid);
 }
 
 MemorySegmentPool::~MemorySegmentPool() {

src/ipc.h

@@ -117,7 +117,7 @@ enum SegmentStatusType {UNOPENED = 0, OPENED = 1, CLOSED = 2};
 
 class MemorySegmentPool {
 public:
-  MemorySegmentPool(ObjStoreId objstoreid, bool create); // can be used in two modes: create mode and open mode (see above)
+  MemorySegmentPool(ObjStoreId objstoreid, std::string& objstore_address, bool create); // can be used in two modes: create mode and open mode (see above)
   ~MemorySegmentPool();
   ObjHandle allocate(size_t nbytes); // allocate memory, potentially creating a new segment (only run on object store)
   void deallocate(ObjHandle pointer); // deallocate object, potentially deallocating a new segment (only run on object store)
@@ -131,6 +131,10 @@ private:
   void close_segment(SegmentId segmentid); // close a segment
   bool create_mode_; // true in the object stores, false on the workers
   ObjStoreId objstoreid_; // the identity of the associated object store
+  // The address of the object store.
+  std::string objstore_address_;
+  // The port of the object store. This is used to help avoid name collisions.
+  std::string objstore_port_;
   size_t page_size_ = bip::mapped_region::get_page_size();
   std::vector<std::pair<std::unique_ptr<bip::managed_shared_memory>, SegmentStatusType> > segments_;
 };

src/objstore.cc

@@ -39,16 +39,28 @@ void ObjStoreService::get_data_from(ObjectID objectid, ObjStore::Stub& stub) {
   RAY_LOG(RAY_DEBUG, "finished streaming data, objectid was " << objectid << " and size was " << num_bytes);
 }
 
-ObjStoreService::ObjStoreService(const std::string& objstore_address, std::shared_ptr<Channel> scheduler_channel)
-  : scheduler_stub_(Scheduler::NewStub(scheduler_channel)), objstore_address_(objstore_address) {
-  RAY_CHECK(recv_queue_.connect(std::string("queue:") + objstore_address + std::string(":obj"), true), "error connecting recv_queue_");
+ObjStoreService::ObjStoreService(const std::string& scheduler_address)
+    : scheduler_address_(scheduler_address) {
+}
+
+void ObjStoreService::register_objstore() {
+  RAY_CHECK(!objstore_address_.empty(), "The object store address must be set before register_objstore is called.");
+  // Create the scheduler stub.
+  auto scheduler_channel = grpc::CreateChannel(scheduler_address_, grpc::InsecureChannelCredentials());
+  scheduler_stub_ = Scheduler::NewStub(scheduler_channel);
+
+  // Create message queue to receive requests from workers.
+  std::string recv_queue_name = std::string("queue:") + objstore_address_ + std::string(":obj");
+  RAY_LOG(RAY_INFO, "Object store creating queue with name " << recv_queue_name << " to receive requests from workers.");
+  RAY_CHECK(recv_queue_.connect(recv_queue_name, true), "error connecting recv_queue_");
+  // Register the objecet store with the scheduler.
   ClientContext context;
   RegisterObjStoreRequest request;
-  request.set_objstore_address(objstore_address);
+  request.set_objstore_address(objstore_address_);
   RegisterObjStoreReply reply;
   scheduler_stub_->RegisterObjStore(&context, request, &reply);
   objstoreid_ = reply.objstoreid();
-  segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, true);
+  segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, objstore_address_, true);
 }
 
 // this method needs to be protected by a objstores_lock_
@@ -319,20 +331,41 @@ void ObjStoreService::start_objstore_service() {
   });
 }
 
-void start_objstore(const char* scheduler_addr, const char* objstore_addr) {
-  auto scheduler_channel = grpc::CreateChannel(scheduler_addr, grpc::InsecureChannelCredentials());
-  RAY_LOG(RAY_INFO, "object store " << objstore_addr << " connected to scheduler " << scheduler_addr);
-  std::string objstore_address(objstore_addr);
-  ObjStoreService service(objstore_address, scheduler_channel);
-  service.start_objstore_service();
-  std::string::iterator split_point = split_ip_address(objstore_address);
-  std::string port;
-  port.assign(split_point, objstore_address.end());
+void set_logfile(const char* log_file_prefix, const std::string& node_ip_address, int port) {
+  if (log_file_prefix) {
+    std::string log_file_name = std::string(log_file_prefix) + "objstore-" + node_ip_address + "-" + std::to_string(port) + ".log";
+    create_log_dir_or_die(log_file_name.c_str());
+    global_ray_config.log_to_file = true;
+    global_ray_config.logfile.open(log_file_name);
+  } else {
+    std::cout << "object store: writing logs to stdout; you can change this by passing --log-file-prefix <fileprefix> to ./objstore" << std::endl;
+    global_ray_config.log_to_file = false;
+  }
+}
+
+void start_objstore(const std::string& scheduler_address, const std::string& node_ip_address, const char* log_file_prefix) {
+  // Initialize the object store.
+  ObjStoreService service(scheduler_address);
+  int port;
   ServerBuilder builder;
-  builder.AddListeningPort(std::string("0.0.0.0:") + port, grpc::InsecureServerCredentials());
+  // Get GRPC to assign an unused port.
+  builder.AddListeningPort(std::string("0.0.0.0:0"), grpc::InsecureServerCredentials(), &port);
   builder.RegisterService(&service);
   std::unique_ptr<Server> server(builder.BuildAndStart());
-
+  if (server == nullptr) {
+    RAY_CHECK(false, "Failed to create the object store server.")
+  }
+  // Set the object store address.
+  service.set_objstore_address(node_ip_address + ":" + std::to_string(port));
+  // Set the logfile.
+  set_logfile(log_file_prefix, node_ip_address, port);
+  // Register the object store with the scheduler.
+  service.register_objstore();
+  // Launch a thread to process incoming messages in the message queue from
+  // the workers.
+  service.start_objstore_service();
+  // Process incoming GRPC calls. These may come from the schedeler or from
+  // other object stores. This method does not return.
   server->Wait();
 }
 
@@ -341,20 +374,12 @@ RayConfig global_ray_config;
 int main(int argc, char** argv) {
   RAY_CHECK_GE(argc, 3, "object store: expected at least two arguments (scheduler ip address and object store ip address)");
 
+  const char* log_file_prefix = nullptr;
   if (argc > 3) {
-    const char* log_file_name = get_cmd_option(argv, argv + argc, "--log-file-name");
-    if (log_file_name) {
-      std::cout << "object store: writing to log file " << log_file_name << std::endl;
-      create_log_dir_or_die(log_file_name);
-      global_ray_config.log_to_file = true;
-      global_ray_config.logfile.open(log_file_name);
-    } else {
-      std::cout << "object store: writing logs to stdout; you can change this by passing --log-file-name <filename> to ./scheduler" << std::endl;
-      global_ray_config.log_to_file = false;
-    }
+    log_file_prefix = get_cmd_option(argv, argv + argc, "--log-file-prefix");
   }
 
-  start_objstore(argv[1], argv[2]);
+  start_objstore(argv[1], argv[2], log_file_prefix);
 
   return 0;
 }

src/objstore.h

@@ -37,7 +37,12 @@ enum MemoryStatusType {READY = 0, NOT_READY = 1, DEALLOCATED = 2, NOT_PRESENT =
 
 class ObjStoreService final : public ObjStore::Service {
 public:
-  ObjStoreService(const std::string& objstore_address, std::shared_ptr<Channel> scheduler_channel);
+  ObjStoreService(const std::string& scheduler_address);
+  // Create the scheduler stub, register the object store with the scheduler,
+  // and create a message queue for workers to connect to.
+  void register_objstore();
+  // Set the object store address.
+  void set_objstore_address(const std::string& objstore_address) { objstore_address_ = objstore_address; }
 
   Status StartDelivery(ServerContext* context, const StartDeliveryRequest* request, AckReply* reply) override;
   Status StreamObjTo(ServerContext* context, const StreamObjToRequest* request, ServerWriter<ObjChunk>* writer) override;
@@ -57,6 +62,7 @@ private:
   void object_ready(ObjectID objectid, size_t metadata_offset);
 
   static const size_t CHUNK_SIZE;
+  std::string scheduler_address_;
   std::string objstore_address_;
   ObjStoreId objstoreid_; // id of this objectstore in the scheduler object store table
   std::shared_ptr<MemorySegmentPool> segmentpool_;

src/raylib.cc

@@ -665,12 +665,12 @@ static PyObject* create_worker(PyObject* self, PyObject* args) {
   // The object store address can be the empty string, in which case the
   // scheduler will choose the object store address.
   const char* objstore_address;
-  PyObject* is_driver_obj;
-  if (!PyArg_ParseTuple(args, "sssO", &node_ip_address, &scheduler_address, &objstore_address, &is_driver_obj)) {
+  Mode mode;
+  if (!PyArg_ParseTuple(args, "sssi", &node_ip_address, &scheduler_address, &objstore_address, &mode)) {
     return NULL;
   }
-  bool is_driver = PyObject_IsTrue(is_driver_obj);
-  Worker* worker = new Worker(std::string(scheduler_address));
+  bool is_driver = (mode != Mode::WORKER_MODE);
+  Worker* worker = new Worker(std::string(node_ip_address), std::string(scheduler_address), mode);
   worker->register_worker(std::string(node_ip_address), std::string(objstore_address), is_driver);
 
   PyObject* t = PyTuple_New(2);
@@ -800,12 +800,12 @@ static PyObject* submit_task(PyObject* self, PyObject* args) {
   return list;
 }
 
-static PyObject* notify_task_completed(PyObject* self, PyObject* args) {
+static PyObject* ready_for_new_task(PyObject* self, PyObject* args) {
   Worker* worker;
   if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
     return NULL;
   }
-  worker->notify_task_completed();
+  worker->ready_for_new_task();
   Py_RETURN_NONE;
 }
 
@@ -920,18 +920,36 @@ static PyObject* scheduler_info(PyObject* self, PyObject* args) {
   SchedulerInfoReply reply;
   worker->scheduler_info(context, request, reply);
 
+  // Unpack the target object reference information.
   PyObject* target_objectid_list = PyList_New(reply.target_objectid_size());
   for (size_t i = 0; i < reply.target_objectid_size(); ++i) {
     PyList_SetItem(target_objectid_list, i, PyInt_FromLong(reply.target_objectid(i)));
   }
+  // Unpack the reference count information.
   PyObject* reference_count_list = PyList_New(reply.reference_count_size());
   for (size_t i = 0; i < reply.reference_count_size(); ++i) {
     PyList_SetItem(reference_count_list, i, PyInt_FromLong(reply.reference_count(i)));
   }
+  // Unpack the available worker information.
+  PyObject* available_worker_list = PyList_New(reply.avail_worker_size());
+  for (size_t i = 0; i < reply.avail_worker_size(); ++i) {
+    PyList_SetItem(available_worker_list, i, PyInt_FromLong(reply.avail_worker(i)));
+  }
+  // Unpack the object store information.
+  PyObject* objstore_list = PyList_New(reply.objstore_size());
+  for (size_t i = 0; i < reply.objstore_size(); ++i) {
+    PyObject* objstore_data = PyDict_New();
+    set_dict_item_and_transfer_ownership(objstore_data, PyString_FromString("objstoreid"), PyInt_FromLong(reply.objstore(i).objstoreid()));
+    set_dict_item_and_transfer_ownership(objstore_data, PyString_FromString("address"), PyString_FromStringAndSize(reply.objstore(i).address().data(), reply.objstore(i).address().size()));
+    PyList_SetItem(objstore_list, i, objstore_data);
+  }
 
+  // Store the unpacked values in a dictionary to return.
   PyObject* dict = PyDict_New();
   set_dict_item_and_transfer_ownership(dict, PyString_FromString("target_objectids"), target_objectid_list);
   set_dict_item_and_transfer_ownership(dict, PyString_FromString("reference_counts"), reference_count_list);
+  set_dict_item_and_transfer_ownership(dict, PyString_FromString("available_workers"), available_worker_list);
+  set_dict_item_and_transfer_ownership(dict, PyString_FromString("objstores"), objstore_list);
   return dict;
 }
 
@@ -1059,7 +1077,7 @@ static PyMethodDef RayLibMethods[] = {
  { "alias_objectids", alias_objectids, METH_VARARGS, "make two objectids refer to the same object" },
  { "wait_for_next_message", wait_for_next_message, METH_VARARGS, "get next message from scheduler (blocking)" },
  { "submit_task", submit_task, METH_VARARGS, "call a remote function" },
- { "notify_task_completed", notify_task_completed, METH_VARARGS, "notify the scheduler that a task has been completed" },
+ { "ready_for_new_task", ready_for_new_task, METH_VARARGS, "notify the scheduler that a task has been completed" },
  { "start_worker_service", start_worker_service, METH_VARARGS, "start the worker service" },
  { "scheduler_info", scheduler_info, METH_VARARGS, "get info about scheduler state" },
  { "task_info", task_info, METH_VARARGS, "get information about task statuses and failures" },

src/scheduler.cc

@@ -215,6 +215,7 @@ Status SchedulerService::RegisterObjStore(ServerContext* context, const Register
 }
 
 Status SchedulerService::RegisterWorker(ServerContext* context, const RegisterWorkerRequest* request, RegisterWorkerReply* reply) {
+  std::string worker_address = request->worker_address();
   std::string objstore_address = request->objstore_address();
   std::string node_ip_address = request->node_ip_address();
   bool is_driver = request->is_driver();
@@ -250,19 +251,11 @@ Status SchedulerService::RegisterWorker(ServerContext* context, const RegisterWo
   } else {
     RAY_CHECK_NEQ(objstoreid, std::numeric_limits<size_t>::max(), "Object store with address " << objstore_address << " not yet registered.");
   }
-  // Populate the worker information and generate a worker address.
+  // Populate the worker information.
   WorkerId workerid;
-  std::string worker_address;
   {
     auto workers = GET(workers_);
     workerid = workers->size();
-    // Generate a random port number. This is currently a hack to avoid reusing
-    // port numbers when we run the tests.
-    std::random_device rd;
-    std::mt19937 rng(rd());
-    std::uniform_int_distribution<int> uni(0, 10000);
-    int port_number = 40000 + uni(rng);
-    worker_address = node_ip_address + ":" + std::to_string(port_number);
     workers->push_back(WorkerHandle());
     auto channel = grpc::CreateChannel(worker_address, grpc::InsecureChannelCredentials());
     (*workers)[workerid].channel = channel;
@@ -279,7 +272,6 @@ Status SchedulerService::RegisterWorker(ServerContext* context, const RegisterWo
   RAY_LOG(RAY_INFO, "Finished registering worker with workerid " << workerid << ", worker address " << worker_address << " on node with IP address " << node_ip_address << ", is_driver = " << is_driver << ", assigned to object store with id " << objstoreid << " and address " << objstore_address);
   reply->set_workerid(workerid);
   reply->set_objstoreid(objstoreid);
-  reply->set_worker_address(worker_address);
   reply->set_objstore_address(objstore_address);
   schedule();
   return Status::OK;
@@ -724,27 +716,40 @@ void SchedulerService::get_info(const SchedulerInfoRequest& request, SchedulerIn
   auto avail_workers = GET(avail_workers_);
   auto task_queue = GET(task_queue_);
   auto reference_counts = GET(reference_counts_);
+  auto objstores = GET(objstores_);
   auto target_objectids = GET(target_objectids_);
   auto function_table = reply->mutable_function_table();
+  // Return info about the reference counts.
   for (int i = 0; i < reference_counts->size(); ++i) {
     reply->add_reference_count((*reference_counts)[i]);
   }
+  // Return info about the target objectids.
   for (int i = 0; i < target_objectids->size(); ++i) {
     reply->add_target_objectid((*target_objectids)[i]);
   }
+  // Return info about the function table.
   for (const auto& entry : *fntable) {
     (*function_table)[entry.first].set_num_return_vals(entry.second.num_return_vals());
     for (const WorkerId& worker : entry.second.workers()) {
       (*function_table)[entry.first].add_workerid(worker);
     }
   }
+  // Return info about the task queue.
   for (const auto& entry : *task_queue) {
     reply->add_operationid(entry);
   }
+  // Return info about the available workers.
   for (const WorkerId& entry : *avail_workers) {
     reply->add_avail_worker(entry);
   }
+  // Return info about the computation graph.
   computation_graph->to_protobuf(reply->mutable_computation_graph());
+  // Return info about the object stores.
+  for (int i = 0; i < objstores->size(); ++i) {
+    ObjstoreData* objstore_data = reply->add_objstore();
+    objstore_data->set_objstoreid(i);
+    objstore_data->set_address((*objstores)[i].address);
+  }
 }
 
 // pick_objstore must be called with a canonical_objectid
@@ -1064,6 +1069,9 @@ void start_scheduler_service(const char* service_addr, SchedulingAlgorithmType s
   builder.AddListeningPort(std::string("0.0.0.0:") + port, grpc::InsecureServerCredentials());
   builder.RegisterService(&service);
   std::unique_ptr<Server> server(builder.BuildAndStart());
+  if (server == nullptr) {
+    RAY_CHECK(false, "Failed to create the scheduler server.")
+  }
   server->Wait();
 }
 

src/worker.cc

@@ -9,11 +9,8 @@ extern "C" {
   static PyObject *RayError;
 }
 
-inline WorkerServiceImpl::WorkerServiceImpl(const std::string& worker_address, Mode mode)
-  : worker_address_(worker_address),
-    mode_(mode) {
-  RAY_CHECK(send_queue_.connect(worker_address_, false), "error connecting send_queue_");
-}
+inline WorkerServiceImpl::WorkerServiceImpl(Mode mode)
+  : mode_(mode) {}
 
 Status WorkerServiceImpl::ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, AckReply* reply) {
   RAY_CHECK(mode_ == Mode::WORKER_MODE, "ExecuteTask can only be called on workers.");
@@ -87,10 +84,23 @@ Status WorkerServiceImpl::PrintErrorMessage(ServerContext* context, const PrintE
   return Status::OK;
 }
 
-Worker::Worker(const std::string& scheduler_address)
-    : scheduler_address_(scheduler_address) {
-  auto scheduler_channel = grpc::CreateChannel(scheduler_address, grpc::InsecureChannelCredentials());
+void WorkerServiceImpl::connect_to_queue() {
+  RAY_LOG(RAY_DEBUG, "Worker service creating queue with name " << worker_address_ << " to commmunicate with worker.");
+  RAY_CHECK(send_queue_.connect(worker_address_, true), "error connecting send_queue_");
+}
+
+Worker::Worker(const std::string& node_ip_address, const std::string& scheduler_address, Mode mode)
+    : node_ip_address_(node_ip_address),
+      scheduler_address_(scheduler_address),
+      mode_(mode) {
+  // Connect to the scheduler service.
+  RAY_LOG(RAY_DEBUG, "Worker creating a scheduler stub.")
+  auto scheduler_channel = grpc::CreateChannel(scheduler_address_, grpc::InsecureChannelCredentials());
   scheduler_stub_ = Scheduler::NewStub(scheduler_channel);
+  // Start the worker service. This will find an unused port which is stored in
+  // worker_port_. This also sets up a message queue between the worker and the
+  // worker service.
+  start_worker_service(mode_);
 }
 
 
@@ -122,6 +132,7 @@ void Worker::register_worker(const std::string& node_ip_address, const std::stri
   unsigned int retry_wait_milliseconds = 20;
   RegisterWorkerRequest request;
   request.set_node_ip_address(node_ip_address);
+  request.set_worker_address(worker_address_);
   // The object store address can be the empty string, in which case the
   // scheduler will assign an object store address.
   request.set_objstore_address(objstore_address);
@@ -142,11 +153,15 @@ void Worker::register_worker(const std::string& node_ip_address, const std::stri
   workerid_ = reply.workerid();
   objstoreid_ = reply.objstoreid();
   objstore_address_ = reply.objstore_address();
-  worker_address_ = reply.worker_address();
-  segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, false);
-  RAY_CHECK(receive_queue_.connect(worker_address_, true), "error connecting receive_queue_");
-  RAY_CHECK(request_obj_queue_.connect(std::string("queue:") + objstore_address_ + std::string(":obj"), false), "error connecting request_obj_queue_");
-  RAY_CHECK(receive_obj_queue_.connect(std::string("queue:") + objstore_address_ + std::string(":worker:") + std::to_string(workerid_) + std::string(":obj"), true), "error connecting receive_obj_queue_");
+  segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, objstore_address_, false);
+  // Connect to the queue for sending requests to the object store.
+  std::string request_obj_queue_name = std::string("queue:") + objstore_address_ + std::string(":obj");
+  RAY_LOG(RAY_DEBUG, "Worker connecting to queue with name " << request_obj_queue_name << " to send requests to the object store.");
+  RAY_CHECK(request_obj_queue_.connect(request_obj_queue_name, false), "error connecting request_obj_queue_");
+  // Create a queue for receiving messages from the object store.
+  std::string receive_obj_queue_name = std::string("queue:") + objstore_address_ + std::string(":worker:") + std::to_string(workerid_) + std::string(":obj");
+  RAY_LOG(RAY_DEBUG, "Worker creating queue with name " << receive_obj_queue_name << " to receive messages from the object store.");
+  RAY_CHECK(receive_obj_queue_.connect(receive_obj_queue_name, true), "error connecting receive_obj_queue_");
   connected_ = true;
   return;
 }
@@ -374,7 +389,7 @@ std::unique_ptr<WorkerMessage> Worker::receive_next_message() {
   return std::unique_ptr<WorkerMessage>(message_ptr);
 }
 
-void Worker::notify_task_completed() {
+void Worker::ready_for_new_task() {
   RAY_CHECK(connected_, "Attempted to perform notify_task_completed but failed.");
   ClientContext context;
   ReadyForNewTaskRequest request;
@@ -389,7 +404,7 @@ void Worker::disconnect() {
   // return.
   server_ptr_->Shutdown();
   // Wait for the thread that launched the worker service to return.
-  worker_server_thread_->join();
+  worker_server_thread_.join();
 }
 
 // TODO(rkn): Should we be using pointers or references? And should they be const?
@@ -430,34 +445,49 @@ void Worker::export_reusable_variable(const std::string& name, const std::string
 // (in our case running in the main thread), whereas the WorkerService will
 // run in a separate thread and potentially utilize multiple threads.
 void Worker::start_worker_service(Mode mode) {
-  const char* service_addr = worker_address_.c_str();
+  RAY_LOG(RAY_DEBUG, "Worker is starting the worker service.");
+  // Signal when the worker service has started.
+  std::condition_variable worker_service_started;
+  // Lock for the above condition.
+  std::mutex worker_service_started_mutex;
   // Launch a new thread for running the worker service. We store this as a
   // field so that we can clean it up when we disconnect the worker.
-  worker_server_thread_ = std::unique_ptr<std::thread>(new std::thread([this, service_addr, mode]() {
-    std::string service_address(service_addr);
-    std::string::iterator split_point = split_ip_address(service_address);
-    std::string port;
-    port.assign(split_point, service_address.end());
-    // Create the worker service.
-    WorkerServiceImpl service(service_address, mode);
+  worker_server_thread_ = std::thread([this, mode, &worker_service_started]() {
     ServerBuilder builder;
-    builder.AddListeningPort(std::string("0.0.0.0:") + port, grpc::InsecureServerCredentials());
+    // Get GRPC to assign an unused port number.
+    builder.AddListeningPort(std::string("0.0.0.0:0"), grpc::InsecureServerCredentials(), &worker_port_);
+    // Create and start the worker service.
+    WorkerServiceImpl service(mode);
     builder.RegisterService(&service);
     std::unique_ptr<Server> server(builder.BuildAndStart());
     server_ptr_ = server.get();
-    RAY_LOG(RAY_INFO, "worker server listening on " << service_address);
-    // If this is part of a worker process (and not a driver process), then tell
-    // the scheduler that it is ready to start receiving tasks.
-    if (mode == Mode::WORKER_MODE) {
-      ClientContext context;
-      ReadyForNewTaskRequest request;
-      request.set_workerid(workerid_);
-      AckReply reply;
-      scheduler_stub_->ReadyForNewTask(&context, request, &reply);
+    if (server == nullptr) {
+      RAY_CHECK(false, "Failed to create the worker server.")
     }
+    RAY_LOG(RAY_DEBUG, "Worker service listening on " << worker_address_);
+    worker_address_ = node_ip_address_ + ":" + std::to_string(worker_port_);
+    service.set_worker_address(worker_address_);
+    // Connect the worker service by a queue to the worker object.
+    service.connect_to_queue();
+    // Use the condition variable to notify the outside thread that the worker
+    // service has been started.
+    // TODO(rkn): Once this has been called, the outside thread will notify the
+    // scheduler that the worker is ready to receive tasks. This can happen
+    // before server->Wait() is called below. What happens to messages sent from
+    // the scheduler before the call to server->Wait()?
+    worker_service_started.notify_all();
     // Wait for work and process work. This method does not return until
     // Shutdown is called from a different thread.
     server->Wait();
     RAY_LOG(RAY_INFO, "Worker service thread returning.")
-  }));
+  });
+  {
+    // Wait until we know the worker service has been started.
+    std::unique_lock<std::mutex> lock(worker_service_started_mutex);
+    worker_service_started.wait(lock);
+  }
+  // Connect to the queue for receiving messages from the worker service.
+  std::string receive_queue_name = worker_address_;
+  RAY_LOG(RAY_DEBUG, "Worker connecting to queue with name " << receive_queue_name << " to commmunicate with worker service.");
+  RAY_CHECK(receive_queue_.connect(receive_queue_name, false), "error connecting receive_queue_");
 }

src/worker.h

@@ -1,6 +1,8 @@
 #ifndef RAY_WORKER_H
 #define RAY_WORKER_H
 
+#include <condition_variable>
+#include <mutex>
 #include <iostream>
 #include <memory>
 #include <string>
@@ -30,12 +32,16 @@ enum Mode {SCRIPT_MODE, WORKER_MODE, PYTHON_MODE, SILENT_MODE};
 
 class WorkerServiceImpl final : public WorkerService::Service {
 public:
-  WorkerServiceImpl(const std::string& worker_address, Mode mode);
+  WorkerServiceImpl(Mode mode);
   Status ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, AckReply* reply) override;
   Status ImportRemoteFunction(ServerContext* context, const ImportRemoteFunctionRequest* request, AckReply* reply) override;
   Status Die(ServerContext* context, const DieRequest* request, AckReply* reply) override;
   Status ImportReusableVariable(ServerContext* context, const ImportReusableVariableRequest* request, AckReply* reply) override;
   Status PrintErrorMessage(ServerContext* context, const PrintErrorMessageRequest* request, AckReply* reply) override;
+  // Set worker address.
+  void set_worker_address(const std::string& worker_address) { worker_address_ = worker_address; }
+  // Connect the worker service to the worker object via a queue.
+  void connect_to_queue();
 private:
   std::string worker_address_;
   MessageQueue<WorkerMessage*> send_queue_;
@@ -46,8 +52,10 @@ private:
 
 class Worker {
  public:
-  Worker(const std::string& scheduler_address);
-
+  // This constructor constructs a stub for the scheduler service. It also
+  // starts the worker service, which also sets up a message queue between the
+  // worker and the worker service.
+  Worker(const std::string& node_ip_address, const std::string& scheduler_address, Mode mode);
   // 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.
@@ -84,16 +92,16 @@ class Worker {
   void register_remote_function(const std::string& name, size_t num_return_vals);
   // Notify the scheduler that a failure has occurred.
   void notify_failure(FailedType type, const std::string& name, const std::string& error_message);
-  // Start the worker server which accepts commands from the scheduler. For
-  // workers, these commands are stored in the message queue, which is read by
-  // the Python interpreter. For drivers, these commands are only for printing
-  // error messages.
+  // Start the worker server which accepts commands from the scheduler. This
+  // also creates a message queue that worker service uses to send messages to
+  // the worker. The queue is read by the Python interpreter. For drivers, these
+  // commands are only for printing error messages.
   void start_worker_service(Mode mode);
   // wait for next task from the RPC system. If null, it means there are no more tasks and the worker should shut down.
   std::unique_ptr<WorkerMessage> receive_next_message();
   // tell the scheduler that we are done with the current task and request the
   // next one.
-  void notify_task_completed();
+  void ready_for_new_task();
   // disconnect the worker
   void disconnect();
   // return connected_
@@ -113,8 +121,8 @@ class Worker {
   bool connected_;
   const size_t CHUNK_SIZE = 8 * 1024;
   std::unique_ptr<Scheduler::Stub> scheduler_stub_;
-  std::unique_ptr<std::thread> worker_server_thread_;
   Server* server_ptr_;
+  std::thread worker_server_thread_;
   MessageQueue<WorkerMessage*> receive_queue_;
   bip::managed_shared_memory segment_;
   WorkerId workerid_;
@@ -122,6 +130,9 @@ class Worker {
   std::string scheduler_address_;
   std::string objstore_address_;
   std::string worker_address_;
+  std::string node_ip_address_;
+  int worker_port_;
+  Mode mode_;
   MessageQueue<ObjRequest> request_obj_queue_;
   MessageQueue<ObjHandle> receive_obj_queue_;
   std::shared_ptr<MemorySegmentPool> segmentpool_;