diff --git a/include/orchestra/orchestra.h b/include/orchestra/orchestra.h index f325a0c8d..f9bd9f8c6 100644 --- a/include/orchestra/orchestra.h +++ b/include/orchestra/orchestra.h @@ -35,6 +35,21 @@ public: typedef std::vector > ObjTable; typedef std::unordered_map FnTable; +#define ORCH_VERBOSE -1 +#define ORCH_INFO 0 +#define ORCH_DEBUG 1 +#define ORCH_FATAL 2 + +#define ORCH_LOG(LEVEL, MESSAGE) \ + if (LEVEL == ORCH_VERBOSE) { \ + \ + } else if (LEVEL == ORCH_FATAL) { \ + std::cerr << "fatal error occured: " << MESSAGE << std::endl; \ + std::exit(1); \ + } else { \ + std::cout << MESSAGE << std::endl; \ + } + class objstore_not_registered_error : public std::runtime_error { public: diff --git a/lib/orchpy/orchpy/services.py b/lib/orchpy/orchpy/services.py index 6d20c4406..99a8b217b 100644 --- a/lib/orchpy/orchpy/services.py +++ b/lib/orchpy/orchpy/services.py @@ -34,8 +34,8 @@ def start_scheduler(scheduler_address): p = subprocess.Popen([os.path.join(_services_path, "scheduler"), scheduler_address]) all_processes.append((p, scheduler_address)) -def start_objstore(objstore_address): - p = subprocess.Popen([os.path.join(_services_path, "objstore"), objstore_address]) +def start_objstore(scheduler_address, objstore_address): + p = subprocess.Popen([os.path.join(_services_path, "objstore"), scheduler_address, objstore_address]) all_processes.append((p, objstore_address)) def start_worker(test_path, scheduler_address, objstore_address, worker_address): diff --git a/protos/orchestra.proto b/protos/orchestra.proto index 0df17ec0f..2f8bb6faf 100644 --- a/protos/orchestra.proto +++ b/protos/orchestra.proto @@ -50,6 +50,15 @@ message PushObjReply { uint64 objref = 1; } +message ObjReadyRequest { + uint64 objref = 1; + uint64 objstoreid = 2; +} + +message WorkerReadyRequest { + uint64 workerid = 1; +} + message ChangeCountRequest { uint64 objref = 1; } @@ -70,14 +79,27 @@ message GetDebugInfoReply { } service Scheduler { + // Register a new worker with the scheduler rpc RegisterWorker(RegisterWorkerRequest) returns (RegisterWorkerReply); + // Register an object store with the scheduler rpc RegisterObjStore(RegisterObjStoreRequest) returns (RegisterObjStoreReply); + // Tell the scheduler that a worker can execute a certain function rpc RegisterFunction(RegisterFunctionRequest) returns (AckReply); + // Asks the scheduler to execute a task, immediately returns an object reference to the result rpc RemoteCall(RemoteCallRequest) returns (RemoteCallReply); + // increment the count of the object reference rpc IncrementCount(ChangeCountRequest) returns (AckReply); + // decrement the count of the object reference rpc DecrementCount(ChangeCountRequest) returns (AckReply); + // request an object reference for an object that will be pushed to an object store rpc PushObj(PushObjRequest) returns (PushObjReply); + // request delivery of an object rpc PullObj(PullObjRequest) returns (AckReply); + // used by an object store to tell the scheduler that an object is ready + rpc ObjReady(ObjReadyRequest) returns (AckReply); + // used by the worker to report back and ask for more work + rpc WorkerReady(WorkerReadyRequest) returns (AckReply); + // get debugging information from the scheduler rpc GetDebugInfo(GetDebugInfoRequest) returns (GetDebugInfoReply); } @@ -117,7 +139,9 @@ message DebugInfoReply { } service ObjStore { + // Request to deliver the data that comes with an object reference to another object store rpc DeliverObj(DeliverObjRequest) returns (AckReply); + // Accept incoming data from another object store rpc StreamObj(stream ObjChunk) returns (AckReply); rpc GetObj(GetObjRequest) returns (GetObjReply); rpc DebugInfo(DebugInfoRequest) returns (DebugInfoReply); diff --git a/src/objstore.cc b/src/objstore.cc index 7e2551b5c..500ca6fe2 100644 --- a/src/objstore.cc +++ b/src/objstore.cc @@ -2,6 +2,7 @@ const size_t ObjStoreClient::CHUNK_SIZE = 8 * 1024; +// this method needs to be protected by a objstore_lock_ Status ObjStoreClient::upload_data_to(slice data, ObjRef objref, ObjStore::Stub& stub) { ObjChunk chunk; ClientContext context; @@ -13,15 +14,31 @@ Status ObjStoreClient::upload_data_to(slice data, ObjRef objref, ObjStore::Stub& chunk.set_totalsize(data.len); chunk.set_data(head + i, std::min(CHUNK_SIZE, data.len - i)); if (!writer->Write(chunk)) { - std::cout << "write failed" << std::endl; - // throw std::runtime_error("write failed"); + ORCH_LOG(ORCH_FATAL, "stream connection prematurely closed") } } writer->WritesDone(); return writer->Finish(); } -void ObjStoreServer::allocate_memory(ObjRef objref, size_t size) { +ObjStoreService::ObjStoreService(const std::string& objstore_address, std::shared_ptr scheduler_channel) + : scheduler_stub_(Scheduler::NewStub(scheduler_channel)) { + ClientContext context; + RegisterObjStoreRequest request; + request.set_address(objstore_address); + RegisterObjStoreReply reply; + scheduler_stub_->RegisterObjStore(&context, request, &reply); + objstoreid_ = reply.objstoreid(); +} + +ObjStoreService::~ObjStoreService() { + for (const auto& segment_name : memory_names_) { + shared_memory_object::remove(segment_name.c_str()); + } +} + +// this method needs to be protected by a memory_lock_ +void ObjStoreService::allocate_memory(ObjRef objref, size_t size) { std::ostringstream stream; stream << "obj-" << memory_names_.size(); std::string name = stream.str(); @@ -37,7 +54,8 @@ void ObjStoreServer::allocate_memory(ObjRef objref, size_t size) { object.ptr.len = size; } -ObjStore::Stub& ObjStoreServer::get_objstore_stub(const std::string& objstore_address) { +// this method needs to be protected by a objstores_lock_ +ObjStore::Stub& ObjStoreService::get_objstore_stub(const std::string& objstore_address) { auto iter = objstores_.find(objstore_address); if (iter != objstores_.end()) return *(iter->second); @@ -46,23 +64,26 @@ ObjStore::Stub& ObjStoreServer::get_objstore_stub(const std::string& objstore_ad return *objstores_[objstore_address]; } -Status ObjStoreServer::DeliverObj(ServerContext* context, const DeliverObjRequest* request, AckReply* reply) { +Status ObjStoreService::DeliverObj(ServerContext* context, const DeliverObjRequest* request, AckReply* reply) { + std::lock_guard objstores_lock(objstores_lock_); ObjStore::Stub& stub = get_objstore_stub(request->objstore_address()); ObjRef objref = request->objref(); - // TODO: Have to introduce wait condition - return ObjStoreClient::upload_data_to(memory_[objref].ptr, objref, stub); + Status status = ObjStoreClient::upload_data_to(memory_[objref].ptr, objref, stub); + return status; } -Status ObjStoreServer::DebugInfo(ServerContext* context, const DebugInfoRequest* request, DebugInfoReply* reply) { +Status ObjStoreService::DebugInfo(ServerContext* context, const DebugInfoRequest* request, DebugInfoReply* reply) { + std::lock_guard memory_lock(memory_lock_); for (const auto& entry : memory_) { reply->add_objref(entry.first); } return Status::OK; } -Status ObjStoreServer::GetObj(ServerContext* context, const GetObjRequest* request, GetObjReply* reply) { +Status ObjStoreService::GetObj(ServerContext* context, const GetObjRequest* request, GetObjReply* reply) { + // TODO(pcm): There is one remaining case where this can fail, i.e. if an object is + // to be delivered from another store but hasn't yet arrived ObjRef objref = request->objref(); - std::cout << "getobj lock"; memory_lock_.lock(); shared_object& object = memory_[objref]; reply->set_bucket(object.name); @@ -70,12 +91,11 @@ Status ObjStoreServer::GetObj(ServerContext* context, const GetObjRequest* reque reply->set_handle(handle); reply->set_size(object.ptr.len); memory_lock_.unlock(); - std::cout << "getobj unlock"; return Status::OK; } -Status ObjStoreServer::StreamObj(ServerContext* context, ServerReader* reader, AckReply* reply) { - std::cout << "stream obj lock" << std::endl; +Status ObjStoreService::StreamObj(ServerContext* context, ServerReader* reader, AckReply* reply) { + ORCH_LOG(ORCH_VERBOSE, "begin to stream data to object store " << objstoreid_); memory_lock_.lock(); ObjChunk chunk; ObjRef objref = 0; @@ -88,31 +108,37 @@ Status ObjStoreServer::StreamObj(ServerContext* context, ServerReader* size_t num_bytes = 0; char* data = memory_[objref].ptr.data; - std::cout << "before loop " << totalsize << std::endl; - do { if (num_bytes + chunk.data().size() > totalsize) { - std::cout << "cancelled" << std::endl; memory_lock_.unlock(); return Status::CANCELLED; } std::memcpy(data, chunk.data().c_str(), chunk.data().size()); data += chunk.data().size(); num_bytes += chunk.data().size(); - std::cout << "looping " << num_bytes << std::endl; } while (reader->Read(&chunk)); - std::cout << "finished" << std::endl; + ORCH_LOG(ORCH_VERBOSE, "finished streaming data, objref was " << objref << " and size was " << num_bytes); + memory_lock_.unlock(); - std::cout << "stream obj unlock" << std::endl; + + ClientContext objready_context; + ObjReadyRequest objready_request; + objready_request.set_objref(objref); + objready_request.set_objstoreid(objstoreid_); + AckReply objready_reply; + scheduler_stub_->ObjReady(&objready_context, objready_request, &objready_reply); + return Status::OK; } -void start_objstore(const char* objstore_address) { - ObjStoreServer service; +void start_objstore(const char* scheduler_addr, const char* objstore_addr) { + auto scheduler_channel = grpc::CreateChannel(scheduler_addr, grpc::InsecureChannelCredentials()); + ORCH_LOG(ORCH_INFO, "object store " << objstore_addr << " connected to scheduler " << scheduler_addr); + std::string objstore_address(objstore_addr); + ObjStoreService service(objstore_address, scheduler_channel); ServerBuilder builder; - - builder.AddListeningPort(std::string(objstore_address), grpc::InsecureServerCredentials()); + builder.AddListeningPort(std::string(objstore_addr), grpc::InsecureServerCredentials()); builder.RegisterService(&service); std::unique_ptr server(builder.BuildAndStart()); @@ -120,11 +146,11 @@ void start_objstore(const char* objstore_address) { } int main(int argc, char** argv) { - if (argc != 2) { + if (argc != 3) { return 1; } - start_objstore(argv[1]); + start_objstore(argv[1], argv[2]); return 0; } diff --git a/src/objstore.h b/src/objstore.h index feed29cff..f93d760f9 100644 --- a/src/objstore.h +++ b/src/objstore.h @@ -35,22 +35,14 @@ struct shared_object { slice ptr; }; -class ObjStoreServer final : public ObjStore::Service { +class ObjStoreService final : public ObjStore::Service { public: - ObjStoreServer() {} - - ~ObjStoreServer() { - for (const auto& segment_name : memory_names_) { - shared_memory_object::remove(segment_name.c_str()); - } - } + ObjStoreService(const std::string& objstore_address, std::shared_ptr scheduler_channel); + ~ObjStoreService(); Status DeliverObj(ServerContext* context, const DeliverObjRequest* request, AckReply* reply) override; - Status DebugInfo(ServerContext* context, const DebugInfoRequest* request, DebugInfoReply* reply) override; - Status GetObj(ServerContext* context, const GetObjRequest* request, GetObjReply* reply) override; - Status StreamObj(ServerContext* context, ServerReader* reader, AckReply* reply) override; private: void allocate_memory(ObjRef objref, size_t size); @@ -62,6 +54,9 @@ private: std::mutex memory_lock_; size_t page_size = mapped_region::get_page_size(); std::unordered_map> objstores_; + std::mutex objstores_lock_; + std::unique_ptr scheduler_stub_; + ObjStoreId objstoreid_; // id of this objectstore in the scheduler object store table }; #endif diff --git a/src/scheduler.cc b/src/scheduler.cc index 4c5e5268a..d8ec05a7f 100644 --- a/src/scheduler.cc +++ b/src/scheduler.cc @@ -1,4 +1,6 @@ #include +#include +#include #include "scheduler.h" @@ -25,7 +27,6 @@ Status SchedulerService::RemoteCall(ServerContext* context, const RemoteCallRequ Status SchedulerService::PushObj(ServerContext* context, const PushObjRequest* request, PushObjReply* reply) { ObjRef objref = register_new_object(); ObjStoreId objstoreid = get_store(request->workerid()); - add_location(objref, objstoreid); reply->set_objref(objref); schedule(); return Status::OK; @@ -35,44 +36,72 @@ Status SchedulerService::PullObj(ServerContext* context, const PullObjRequest* r std::lock_guard objtable_lock(objtable_lock_); ObjRef objref = request->objref(); if (objref >= objtable_.size() || objtable_[objref].size() == 0) { - std::cout << "internal error: no object with objref exists" << std::endl; - std::exit(1); + ORCH_LOG(ORCH_FATAL, "internal error: no object with objref " << objref << " exists"); } - std::mt19937 rng; - std::uniform_int_distribution uni(0, objtable_[objref].size()-1); - ObjStoreId objstoreid = uni(rng); - std::lock_guard objstore_lock(objstores_lock_); + ObjStoreId objstoreid = pick_objstore(objref); + deliver_object(objref, objstoreid, get_store(request->workerid())); + return Status::OK; +} - DeliverObjRequest deliver_request; - ObjStoreId id = get_store(request->workerid()); - deliver_request.set_objstore_address(objstores_[id].address); - deliver_request.set_objref(objref); - AckReply deliver_reply; - ClientContext deliver_context; - objstores_[objstoreid].objstore_stub->DeliverObj(&deliver_context, deliver_request, &deliver_reply); +Status SchedulerService::RegisterObjStore(ServerContext* context, const RegisterObjStoreRequest* request, RegisterObjStoreReply* reply) { + std::lock_guard lock(); + ObjStoreId objstoreid = objstores_.size(); + auto channel = grpc::CreateChannel(request->address(), grpc::InsecureChannelCredentials()); + objstores_.push_back(ObjStoreHandle()); + objstores_[objstoreid].address = request->address(); + objstores_[objstoreid].channel = channel; + objstores_[objstoreid].objstore_stub = ObjStore::NewStub(channel); + reply->set_objstoreid(objstoreid); return Status::OK; } Status SchedulerService::RegisterWorker(ServerContext* context, const RegisterWorkerRequest* request, RegisterWorkerReply* reply) { WorkerId workerid = register_worker(request->worker_address(), request->objstore_address()); - std::cout << "registered worker with workerid" << workerid << std::endl; + ORCH_LOG(ORCH_INFO, "registered worker with workerid " << workerid); reply->set_workerid(workerid); schedule(); return Status::OK; } Status SchedulerService::RegisterFunction(ServerContext* context, const RegisterFunctionRequest* request, AckReply* reply) { - std::cout << "RegisterFunction: workerid is" << request->workerid() << std::endl; + ORCH_LOG(ORCH_INFO, "register function " << request->fnname() << " from workerid " << request->workerid()); register_function(request->fnname(), request->workerid(), request->num_return_vals()); schedule(); return Status::OK; } +Status SchedulerService::ObjReady(ServerContext* context, const ObjReadyRequest* request, AckReply* reply) { + ORCH_LOG(ORCH_VERBOSE, "object " << request->objref() << " ready on store " << request->objstoreid()); + add_location(request->objref(), request->objstoreid()); + schedule(); + return Status::OK; +} + +Status SchedulerService::WorkerReady(ServerContext* context, const WorkerReadyRequest* request, AckReply* reply) { + ORCH_LOG(ORCH_INFO, "worker " << request->workerid() << " reported back"); + { + std::lock_guard lock(avail_workers_lock_); + avail_workers_.push_back(request->workerid()); + } + schedule(); + return Status::OK; +} + Status SchedulerService::GetDebugInfo(ServerContext* context, const GetDebugInfoRequest* request, GetDebugInfoReply* reply) { debug_info(*request, reply); return Status::OK; } +void SchedulerService::deliver_object(ObjRef objref, ObjStoreId from, ObjStoreId to) { + ClientContext context; + AckReply reply; + DeliverObjRequest request; + request.set_objref(objref); + std::lock_guard lock(objstores_lock_); + request.set_objstore_address(objstores_[to].address); + objstores_[from].objstore_stub->DeliverObj(&context, request, &reply); +} + void SchedulerService::schedule() { // TODO: don't recheck if nothing changed std::lock_guard avail_workers_lock(avail_workers_lock_); @@ -99,21 +128,17 @@ void SchedulerService::submit_task(std::unique_ptr call, WorkerId workerid ClientContext context; InvokeCallRequest request; InvokeCallReply reply; - std::cout << "sending arguments now" << std::endl; + ORCH_LOG(ORCH_INFO, "starting to send arguments"); for (size_t i = 0; i < call->arg_size(); ++i) { if (!call->arg(i).has_obj()) { - std::cout << "need to send object ref" << call->arg(i).ref() << std::endl; + ObjRef objref = call->arg(i).ref(); + ORCH_LOG(ORCH_INFO, "call contains object ref " << objref); std::lock_guard objtable_lock(objtable_lock_); auto &objstores = objtable_[call->arg(i).ref()]; std::lock_guard workers_lock(workers_lock_); if (!std::binary_search(objstores.begin(), objstores.end(), workers_[workerid].objstoreid)) { - std::cout << "lost object store, need to do recovery" << std::endl; - std::exit(1); + deliver_object(objref, pick_objstore(objref), workers_[workerid].objstoreid); } - // if (objstoreid != workers_[workerid].objstoreid) { - // std::lock_guard objstores_lock(objstores_lock_); - // objstores_. - // } } } request.set_allocated_call(call.release()); // protobuf object takes ownership @@ -134,25 +159,22 @@ bool SchedulerService::can_run(const Call& task) { } WorkerId SchedulerService::register_worker(const std::string& worker_address, const std::string& objstore_address) { + ORCH_LOG(ORCH_INFO, "registering worker " << worker_address << " connected to object store " << objstore_address); ObjStoreId objstoreid = std::numeric_limits::max(); - objstores_lock_.lock(); - for (size_t i = 0; i < objstores_.size(); ++i) { - std::cout << "address: " << objstores_[i].address << std::endl; - std::cout << "my address: " << objstore_address << std::endl; - if (objstores_[i].address == objstore_address) { - objstoreid = i; + for (int num_attempts = 0; num_attempts < 5; ++num_attempts) { + std::lock_guard lock(objstores_lock_); + for (size_t i = 0; i < objstores_.size(); ++i) { + if (objstores_[i].address == objstore_address) { + objstoreid = i; + } + } + if (objstoreid == std::numeric_limits::max()) { + std::this_thread::sleep_for (std::chrono::milliseconds(100)); } } if (objstoreid == std::numeric_limits::max()) { - // register objstore - objstoreid = objstores_.size(); - auto channel = grpc::CreateChannel(objstore_address, grpc::InsecureChannelCredentials()); - objstores_.push_back(ObjStoreHandle()); - objstores_[objstoreid].address = objstore_address; - objstores_[objstoreid].channel = channel; - objstores_[objstoreid].objstore_stub = ObjStore::NewStub(channel); + ORCH_LOG(ORCH_FATAL, "object store with address " << objstore_address << " not yet registered"); } - objstores_lock_.unlock(); workers_lock_.lock(); WorkerId workerid = workers_.size(); workers_.push_back(WorkerHandle()); @@ -168,36 +190,32 @@ WorkerId SchedulerService::register_worker(const std::string& worker_address, co } ObjRef SchedulerService::register_new_object() { - objtable_lock_.lock(); + std::lock_guard lock(objtable_lock_); ObjRef result = objtable_.size(); objtable_.push_back(std::vector()); - objtable_lock_.unlock(); return result; } void SchedulerService::add_location(ObjRef objref, ObjStoreId objstoreid) { - objtable_lock_.lock(); + std::lock_guard objtable_lock(objtable_lock_); // do a binary search auto pos = std::lower_bound(objtable_[objref].begin(), objtable_[objref].end(), objstoreid); if (pos == objtable_[objref].end() || objstoreid < *pos) { objtable_[objref].insert(pos, objstoreid); } - objtable_lock_.unlock(); } ObjStoreId SchedulerService::get_store(WorkerId workerid) { - workers_lock_.lock(); + std::lock_guard lock(workers_lock_); ObjStoreId result = workers_[workerid].objstoreid; - workers_lock_.unlock(); return result; } void SchedulerService::register_function(const std::string& name, WorkerId workerid, size_t num_return_vals) { - fntable_lock_.lock(); + std::lock_guard lock(fntable_lock_); FnInfo& info = fntable_[name]; info.set_num_return_vals(num_return_vals); info.add_worker(workerid); - fntable_lock_.unlock(); } void SchedulerService::debug_info(const GetDebugInfoRequest& request, GetDebugInfoReply* reply) { @@ -226,6 +244,12 @@ void SchedulerService::debug_info(const GetDebugInfoRequest& request, GetDebugIn avail_workers_lock_.unlock(); } +ObjStoreId SchedulerService::pick_objstore(ObjRef objref) { + std::mt19937 rng; + std::uniform_int_distribution uni(0, objtable_[objref].size()-1); + return uni(rng); +} + void start_scheduler_service(const char* server_address) { SchedulerService service; ServerBuilder builder; diff --git a/src/scheduler.h b/src/scheduler.h index 591abf28c..21ebb1844 100644 --- a/src/scheduler.h +++ b/src/scheduler.h @@ -40,10 +40,15 @@ public: Status RemoteCall(ServerContext* context, const RemoteCallRequest* request, RemoteCallReply* reply) override; Status PushObj(ServerContext* context, const PushObjRequest* request, PushObjReply* reply) override; Status PullObj(ServerContext* context, const PullObjRequest* request, AckReply* reply) override; + Status RegisterObjStore(ServerContext* context, const RegisterObjStoreRequest* request, RegisterObjStoreReply* reply) override; Status RegisterWorker(ServerContext* context, const RegisterWorkerRequest* request, RegisterWorkerReply* reply) override; Status RegisterFunction(ServerContext* context, const RegisterFunctionRequest* request, AckReply* reply) override; + Status ObjReady(ServerContext* context, const ObjReadyRequest* request, AckReply* reply) override; + Status WorkerReady(ServerContext* context, const WorkerReadyRequest* request, AckReply* reply) override; Status GetDebugInfo(ServerContext* context, const GetDebugInfoRequest* request, GetDebugInfoReply* reply) override; + // ask an object store to send object to another objectstore + void deliver_object(ObjRef objref, ObjStoreId from, ObjStoreId to); // assign a task to a worker void schedule(); // execute a task on a worker and ship required object references @@ -63,6 +68,9 @@ public: // get debugging information for the scheduler void debug_info(const GetDebugInfoRequest& request, GetDebugInfoReply* reply); private: + // pick an objectstore that holds a given object (needs protection by objtable_lock_) + ObjStoreId pick_objstore(ObjRef objref); + // Vector of all workers registered in the system. Their index in this vector // is the workerid. std::vector workers_; diff --git a/src/worker.cc b/src/worker.cc index 9b4ba2b77..861c943a7 100644 --- a/src/worker.cc +++ b/src/worker.cc @@ -1,12 +1,11 @@ # include "worker.h" Status WorkerServiceImpl::InvokeCall(ServerContext* context, const InvokeCallRequest* request, InvokeCallReply* reply) { - call_ = request->call(); - std::cout << "invoke call request" << std::endl; + call_ = request->call(); // Copy call + ORCH_LOG(ORCH_INFO, "invoked task " << request->call().name()); try { Call* callptr = &call_; message_queue mq(open_only, worker_address_.c_str()); - std::cout << "before send: num args" << call_.arg_size() << std::endl; mq.send(&callptr, sizeof(Call*), 0); } catch(interprocess_exception &ex){ @@ -15,7 +14,6 @@ Status WorkerServiceImpl::InvokeCall(ServerContext* context, const InvokeCallReq // TODO: return Status; } message_queue::remove(worker_address_.c_str()); - std::cout << "notified server" << std::endl; return Status::OK; } @@ -85,17 +83,16 @@ void Worker::put_object(ObjRef objref, const Obj* obj) { const char* head = data.c_str(); for (size_t i = 0; i < data.length(); i += CHUNK_SIZE) { chunk.set_objref(objref); - std::cout << "chunk totalsize" << std::endl; chunk.set_totalsize(totalsize); chunk.set_data(head + i, std::min(CHUNK_SIZE, data.length() - i)); if (!writer->Write(chunk)) { - std::cout << "write failed" << std::endl; - // TODO: Better error handling: throw std::runtime_error("write failed"); + ORCH_LOG(ORCH_FATAL, "write failed during put_object"); + // TODO(pcm): better error handling } } writer->WritesDone(); Status status = writer->Finish(); - // TODO: error handling + // TODO(pcm): error handling } void Worker::register_function(const std::string& name, size_t num_return_vals) { @@ -120,7 +117,7 @@ void Worker::start_worker_service() { builder.AddListeningPort(server_address, grpc::InsecureServerCredentials()); builder.RegisterService(&service); std::unique_ptr server(builder.BuildAndStart()); - std::cout << "WorkerServer listening on " << server_address << std::endl; + ORCH_LOG(ORCH_INFO, "worker server listening on " << server_address); server->Wait(); }); } @@ -135,8 +132,6 @@ Call* Worker::receive_next_task() { Call* call; while (true) { mq.receive(&call, sizeof(Call*), recvd_size, priority); - std::cout << "got call" << call << std::endl; - std::cout << "after send: num args" << call->arg_size() << std::endl; return call; } } diff --git a/test/runtest.py b/test/runtest.py index 4fd6015db..d385d8b82 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -72,8 +72,11 @@ class ObjStoreTest(unittest.TestCase): worker2_port = new_worker_port() services.start_scheduler(address(IP_ADDRESS, scheduler_port)) - services.start_objstore(address(IP_ADDRESS, objstore1_port)) - services.start_objstore(address(IP_ADDRESS, objstore2_port)) + + time.sleep(0.1) + + services.start_objstore(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, objstore1_port)) + services.start_objstore(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, objstore2_port)) time.sleep(0.2) @@ -111,7 +114,10 @@ class SchedulerTest(unittest.TestCase): worker2_port = new_worker_port() services.start_scheduler(address(IP_ADDRESS, scheduler_port)) - services.start_objstore(address(IP_ADDRESS, objstore_port)) + + time.sleep(0.1) + + services.start_objstore(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, objstore_port)) time.sleep(0.2) @@ -151,7 +157,10 @@ class WorkerTest(unittest.TestCase): worker1_port = new_worker_port() services.start_scheduler(address(IP_ADDRESS, scheduler_port)) - services.start_objstore(address(IP_ADDRESS, objstore_port)) + + time.sleep(0.1) + + services.start_objstore(address(IP_ADDRESS, scheduler_port), address(IP_ADDRESS, objstore_port)) time.sleep(0.2) @@ -184,5 +193,6 @@ class WorkerTest(unittest.TestCase): services.cleanup() + if __name__ == '__main__': unittest.main()