diff --git a/src/scheduler.cc b/src/scheduler.cc index f07e48f8e..9071b5808 100644 --- a/src/scheduler.cc +++ b/src/scheduler.cc @@ -544,7 +544,7 @@ std::pair SchedulerService::register_worker(const std::str RAY_LOG(RAY_INFO, "registering worker " << worker_address << " connected to object store " << objstore_address); ObjStoreId objstoreid = std::numeric_limits::max(); // TODO: HACK: num_attempts is a hack - for (int num_attempts = 0; num_attempts < 5; ++num_attempts) { + for (int num_attempts = 0; num_attempts < 30; ++num_attempts) { auto objstores = GET(objstores_); for (size_t i = 0; i < objstores->size(); ++i) { if ((*objstores)[i].address == objstore_address) { diff --git a/src/worker.cc b/src/worker.cc index 29081303e..9da8a7c3e 100644 --- a/src/worker.cc +++ b/src/worker.cc @@ -88,13 +88,24 @@ bool Worker::kill_workers(ClientContext &context) { } void Worker::register_worker(const std::string& worker_address, const std::string& objstore_address, bool is_driver) { + unsigned int retry_wait_milliseconds = 20; RegisterWorkerRequest request; request.set_worker_address(worker_address); request.set_objstore_address(objstore_address); request.set_is_driver(is_driver); RegisterWorkerReply reply; - ClientContext context; - Status status = scheduler_stub_->RegisterWorker(&context, request, &reply); + grpc::StatusCode status_code = grpc::UNAVAILABLE; + // TODO: HACK: retrying is a hack + for (int i = 0; i < 5; ++i) { + ClientContext context; + status_code = scheduler_stub_->RegisterWorker(&context, request, &reply).error_code(); + if (status_code != grpc::UNAVAILABLE) { + break; + } + // Note that each pass through the loop may take substantially longer than + // retry_wait_milliseconds because grpc may do its own retrying. + std::this_thread::sleep_for(std::chrono::milliseconds(retry_wait_milliseconds)); + } workerid_ = reply.workerid(); objstoreid_ = reply.objstoreid(); segmentpool_ = std::make_shared(objstoreid_, false);