diff --git a/doc/reference-counting.md b/doc/reference-counting.md index 9fa6914ad..c9e7cbe8d 100644 --- a/doc/reference-counting.md +++ b/doc/reference-counting.md @@ -47,8 +47,8 @@ will notify the scheduler that those object references are in the object store. Then when the scheduler deallocates the object, we call `DecrementRefCount` for the object references that it holds internally (the scheduler keeps track of these internal object references in the `contained_objrefs_` data structure). -3. To handle the third case, we increment in the `serialize_call` method and -decrement in the `deserialize_call` method. +3. To handle the third case, we increment in the `serialize_task` method and +decrement in the `deserialize_task` method. ## How to Handle Aliasing Reference counting interacts with aliasing. Since multiple object references diff --git a/lib/orchpy/orchpy/serialization.py b/lib/orchpy/orchpy/serialization.py index 9217550a4..db76618f6 100644 --- a/lib/orchpy/orchpy/serialization.py +++ b/lib/orchpy/orchpy/serialization.py @@ -29,11 +29,11 @@ def deserialize(worker_capsule, capsule): primitive_obj = orchpy.lib.deserialize_object(worker_capsule, capsule) return from_primitive(primitive_obj) -def serialize_call(worker_capsule, func_name, args): +def serialize_task(worker_capsule, func_name, args): primitive_args = [(arg if isinstance(arg, orchpy.lib.ObjRef) else to_primitive(arg)) for arg in args] - return orchpy.lib.serialize_call(worker_capsule, func_name, primitive_args) + return orchpy.lib.serialize_task(worker_capsule, func_name, primitive_args) -def deserialize_call(worker_capsule, call): - func_name, primitive_args, return_objrefs = orchpy.lib.deserialize_call(worker_capsule, call) +def deserialize_task(worker_capsule, task): + func_name, primitive_args, return_objrefs = orchpy.lib.deserialize_task(worker_capsule, task) args = [(arg if isinstance(arg, orchpy.lib.ObjRef) else from_primitive(arg)) for arg in primitive_args] return func_name, args, return_objrefs diff --git a/lib/orchpy/orchpy/worker.py b/lib/orchpy/orchpy/worker.py index 33235c8c1..0bcf85abc 100644 --- a/lib/orchpy/orchpy/worker.py +++ b/lib/orchpy/orchpy/worker.py @@ -43,10 +43,10 @@ class Worker(object): orchpy.lib.register_function(self.handle, function.func_name, len(function.return_types)) self.functions[function.func_name] = function - def remote_call(self, func_name, args): + def submit_task(self, func_name, args): """Tell the scheduler to schedule the execution of the function with name `func_name` with arguments `args`. Retrieve object references for the outputs of the function from the scheduler and immediately return them.""" - call_capsule = serialization.serialize_call(self.handle, func_name, args) - objrefs = orchpy.lib.remote_call(self.handle, call_capsule) + task_capsule = serialization.serialize_task(self.handle, func_name, args) + objrefs = orchpy.lib.submit_task(self.handle, task_capsule) return objrefs # We make `global_worker` a global variable so that there is one worker per worker process. @@ -86,15 +86,15 @@ def main_loop(worker=global_worker): if not orchpy.lib.connected(worker.handle): raise Exception("Worker is attempting to enter main_loop but has not been connected yet.") orchpy.lib.start_worker_service(worker.handle) - def process_call(call): # wrapping these calls in a function should cause the local variables to go out of scope more quickly, which is useful for inspecting reference counts - func_name, args, return_objrefs = serialization.deserialize_call(worker.handle, call) + def process_task(task): # wrapping these lines in a function should cause the local variables to go out of scope more quickly, which is useful for inspecting reference counts + func_name, args, return_objrefs = serialization.deserialize_task(worker.handle, task) arguments = get_arguments_for_execution(worker.functions[func_name], args, worker) # get args from objstore outputs = worker.functions[func_name].executor(arguments) # execute the function store_outputs_in_objstore(return_objrefs, outputs, worker) # store output in local object store orchpy.lib.notify_task_completed(worker.handle) # notify the scheduler that the task has completed while True: - call = orchpy.lib.wait_for_next_task(worker.handle) - process_call(call) + task = orchpy.lib.wait_for_next_task(worker.handle) + process_task(task) def distributed(arg_types, return_types, worker=global_worker): def distributed_decorator(func): @@ -108,7 +108,7 @@ def distributed(arg_types, return_types, worker=global_worker): def func_call(*args): """This is what gets run immediately when a worker calls a distributed function.""" check_arguments(func_call, list(args)) # throws an exception if args are invalid - objrefs = worker.remote_call(func_call.func_name, list(args)) + objrefs = worker.submit_task(func_call.func_name, list(args)) return objrefs[0] if len(objrefs) == 1 else objrefs func_call.func_name = "{}.{}".format(func.__module__, func.__name__) func_call.executor = func_executor diff --git a/protos/orchestra.proto b/protos/orchestra.proto index 9c63f8ad1..9c1a40adc 100644 --- a/protos/orchestra.proto +++ b/protos/orchestra.proto @@ -24,7 +24,7 @@ service Scheduler { // 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); + rpc SubmitTask(SubmitTaskRequest) returns (SubmitTaskReply); // Increment the count of the object reference rpc IncrementCount(ChangeCountRequest) returns (AckReply); // Decrement the count of the object reference @@ -76,11 +76,11 @@ message RegisterFunctionRequest { uint64 num_return_vals = 3; // Number of return values of the function } -message RemoteCallRequest { - Call call = 1; // Contains name of the function to be executed and arguments +message SubmitTaskRequest { + Task task = 1; // Contains name of the function to be executed and arguments } -message RemoteCallReply { +message SubmitTaskReply { repeated uint64 result = 1; // Object references of the function return values } @@ -139,7 +139,7 @@ message FnTableEntry { } message SchedulerInfoReply { - repeated Call task = 1; // Tasks on the task queue + repeated Task task = 1; // Tasks on the task queue repeated uint64 avail_worker = 3; // List of workers waiting to get a task assigned map function_table = 2; // Table of all available remote function repeated uint64 target_objref = 4; // The target_objrefs_ data structure @@ -211,13 +211,13 @@ message ObjStoreInfoReply { // Workers service WorkerService { - rpc InvokeCall(InvokeCallRequest) returns (InvokeCallReply); // Scheduler calls a function from the worker + rpc ExecuteTask(ExecuteTaskRequest) returns (ExecuteTaskReply); // Scheduler calls a function from the worker } -message InvokeCallRequest { - Call call = 1; // Contains name of the function to be executed and arguments +message ExecuteTaskRequest { + Task task = 1; // Contains name of the function to be executed and arguments } -message InvokeCallReply { +message ExecuteTaskReply { } diff --git a/protos/types.proto b/protos/types.proto index 0eb487b1e..af19503f3 100644 --- a/protos/types.proto +++ b/protos/types.proto @@ -50,7 +50,7 @@ message Value { Obj obj = 2; // For pass by value } -message Call { +message Task { string name = 1; // Name of the function call repeated Value arg = 2; // List of arguments, can be either object references or protobuf descriptions of object passed by value repeated uint64 result = 3; // Object references for result diff --git a/src/orchpylib.cc b/src/orchpylib.cc index d408b3ce9..9c2e0b92d 100644 --- a/src/orchpylib.cc +++ b/src/orchpylib.cc @@ -123,12 +123,12 @@ static PyObject *OrchPyError; // Pass arguments from Python to C++ -int PyObjectToCall(PyObject* object, Call **call) { - if (PyCapsule_IsValid(object, "call")) { - *call = static_cast(PyCapsule_GetPointer(object, "call")); +int PyObjectToTask(PyObject* object, Task **task) { + if (PyCapsule_IsValid(object, "task")) { + *task = static_cast(PyCapsule_GetPointer(object, "task")); return 1; } else { - PyErr_SetString(PyExc_TypeError, "must be a 'call' capsule"); + PyErr_SetString(PyExc_TypeError, "must be a 'task' capsule"); return 0; } } @@ -175,8 +175,8 @@ void WorkerCapsule_Destructor(PyObject* capsule) { delete obj; } -void CallCapsule_Destructor(PyObject* capsule) { - Call* obj = static_cast(PyCapsule_GetPointer(capsule, "call")); +void TaskCapsule_Destructor(PyObject* capsule) { + Task* obj = static_cast(PyCapsule_GetPointer(capsule, "task")); delete obj; } @@ -488,59 +488,59 @@ PyObject* deserialize_object(PyObject* self, PyObject* args) { if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToObj, &obj)) { return NULL; } - std::vector objrefs; // This is a vector of all the objrefs that are serialized in this call, including objrefs that are contained in Python objects that are passed by value. + std::vector objrefs; // This is a vector of all the objrefs that are serialized in this task, including objrefs that are contained in Python objects that are passed by value. return deserialize(worker_capsule, *obj, objrefs); // TODO(rkn): Should we do anything with objrefs? } -PyObject* serialize_call(PyObject* self, PyObject* args) { +PyObject* serialize_task(PyObject* self, PyObject* args) { PyObject* worker_capsule; - Call* call = new Call(); // TODO: to be freed in capsul destructor + Task* task = new Task(); // TODO: to be freed in capsule destructor char* name; int len; PyObject* arguments; if (!PyArg_ParseTuple(args, "Os#O", &worker_capsule, &name, &len, &arguments)) { return NULL; } - call->set_name(name, len); - std::vector objrefs; // This is a vector of all the objrefs that are serialized in this call, including objrefs that are contained in Python objects that are passed by value. + task->set_name(name, len); + std::vector objrefs; // This is a vector of all the objrefs that are serialized in this task, including objrefs that are contained in Python objects that are passed by value. if (PyList_Check(arguments)) { for (size_t i = 0, size = PyList_Size(arguments); i < size; ++i) { PyObject* element = PyList_GetItem(arguments, i); if (PyObject_IsInstance(element, (PyObject*)&PyObjRefType)) { ObjRef objref = ((PyObjRef*) element)->val; - call->add_arg()->set_ref(objref); + task->add_arg()->set_ref(objref); objrefs.push_back(objref); } else { - Obj* arg = call->add_arg()->mutable_obj(); + Obj* arg = task->add_arg()->mutable_obj(); serialize(worker_capsule, PyList_GetItem(arguments, i), arg, objrefs); } } } else { - PyErr_SetString(OrchPyError, "serialize_call: second argument needs to be a list"); + PyErr_SetString(OrchPyError, "serialize_task: second argument needs to be a list"); return NULL; } Worker* worker; PyObjectToWorker(worker_capsule, &worker); if (objrefs.size() > 0) { - ORCH_LOG(ORCH_REFCOUNT, "In serialize_call, calling increment_reference_count for contained objrefs"); + ORCH_LOG(ORCH_REFCOUNT, "In serialize_task, calling increment_reference_count for contained objrefs"); worker->increment_reference_count(objrefs); } - return PyCapsule_New(static_cast(call), "call", &CallCapsule_Destructor); + return PyCapsule_New(static_cast(task), "task", &TaskCapsule_Destructor); } -PyObject* deserialize_call(PyObject* self, PyObject* args) { +PyObject* deserialize_task(PyObject* self, PyObject* args) { PyObject* worker_capsule; - Call* call; - if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToCall, &call)) { + Task* task; + if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToTask, &task)) { return NULL; } - std::vector objrefs; // This is a vector of all the objrefs that were serialized in this call, including objrefs that are contained in Python objects that are passed by value. - PyObject* string = PyString_FromStringAndSize(call->name().c_str(), call->name().size()); - int argsize = call->arg_size(); + std::vector objrefs; // This is a vector of all the objrefs that were serialized in this task, including objrefs that are contained in Python objects that are passed by value. + PyObject* string = PyString_FromStringAndSize(task->name().c_str(), task->name().size()); + int argsize = task->arg_size(); PyObject* arglist = PyList_New(argsize); for (int i = 0; i < argsize; ++i) { - const Value& val = call->arg(i); + const Value& val = task->arg(i); if (!val.has_obj()) { PyList_SetItem(arglist, i, make_pyobjref(worker_capsule, val.ref())); objrefs.push_back(val.ref()); @@ -551,14 +551,14 @@ PyObject* deserialize_call(PyObject* self, PyObject* args) { Worker* worker; PyObjectToWorker(worker_capsule, &worker); worker->decrement_reference_count(objrefs); - int resultsize = call->result_size(); + int resultsize = task->result_size(); std::vector result_objrefs; PyObject* resultlist = PyList_New(resultsize); for (int i = 0; i < resultsize; ++i) { - PyList_SetItem(resultlist, i, make_pyobjref(worker_capsule, call->result(i))); - result_objrefs.push_back(call->result(i)); + PyList_SetItem(resultlist, i, make_pyobjref(worker_capsule, task->result(i))); + result_objrefs.push_back(task->result(i)); } - worker->decrement_reference_count(result_objrefs); // The corresponding increment is done in RemoteCall in the scheduler. + worker->decrement_reference_count(result_objrefs); // The corresponding increment is done in SubmitTask in the scheduler. PyObject* t = PyTuple_New(3); // We set the items of the tuple using PyTuple_SetItem, because that transfers ownership to the tuple. PyTuple_SetItem(t, 0, string); PyTuple_SetItem(t, 1, arglist); @@ -607,22 +607,22 @@ PyObject* wait_for_next_task(PyObject* self, PyObject* args) { if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) { return NULL; } - Call* call = worker->receive_next_task(); - return PyCapsule_New(static_cast(call), "call", NULL); // This call is owned by the C++ worker class, so we do not deallocate it. + Task* task = worker->receive_next_task(); + return PyCapsule_New(static_cast(task), "task", NULL); // This task is owned by the C++ worker class, so we do not deallocate it. } -PyObject* remote_call(PyObject* self, PyObject* args) { +PyObject* submit_task(PyObject* self, PyObject* args) { PyObject* worker_capsule; - Call* call; - if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToCall, &call)) { + Task* task; + if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToTask, &task)) { return NULL; } Worker* worker; PyObjectToWorker(worker_capsule, &worker); - RemoteCallRequest request; - request.set_allocated_call(call); - RemoteCallReply reply = worker->remote_call(&request); - request.release_call(); // TODO: Make sure that call is not moved, otherwise capsule pointer needs to be updated + 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 int size = reply.result_size(); PyObject* list = PyList_New(size); std::vector result_objrefs; @@ -630,7 +630,7 @@ PyObject* remote_call(PyObject* self, PyObject* args) { PyList_SetItem(list, i, make_pyobjref(worker_capsule, reply.result(i))); result_objrefs.push_back(reply.result(i)); } - worker->decrement_reference_count(result_objrefs); // The corresponding increment is done in RemoteCall in the scheduler. + worker->decrement_reference_count(result_objrefs); // The corresponding increment is done in SubmitTask in the scheduler. return list; } @@ -761,8 +761,8 @@ static PyMethodDef OrchPyLibMethods[] = { { "put_arrow", put_arrow, METH_VARARGS, "put an arrow array on the local object store"}, { "get_arrow", get_arrow, METH_VARARGS, "get an arrow array from the local object store"}, { "is_arrow", is_arrow, METH_VARARGS, "is the object in the local object store an arrow object?"}, - { "serialize_call", serialize_call, METH_VARARGS, "serialize a call to protocol buffers" }, - { "deserialize_call", deserialize_call, METH_VARARGS, "deserialize a call from protocol buffers" }, + { "serialize_task", serialize_task, METH_VARARGS, "serialize a task to protocol buffers" }, + { "deserialize_task", deserialize_task, METH_VARARGS, "deserialize a task from protocol buffers" }, { "create_worker", create_worker, METH_VARARGS, "connect to the scheduler and the object store" }, { "disconnect", disconnect, METH_VARARGS, "disconnect the worker from the scheduler and the object store" }, { "connected", connected, METH_VARARGS, "check if the worker is connected to the scheduler and the object store" }, @@ -773,7 +773,7 @@ static PyMethodDef OrchPyLibMethods[] = { { "request_object" , request_object, METH_VARARGS, "request an object to be delivered to the local object store" }, { "alias_objrefs", alias_objrefs, METH_VARARGS, "make two objrefs refer to the same object" }, { "wait_for_next_task", wait_for_next_task, METH_VARARGS, "get next task from scheduler (blocking)" }, - { "remote_call", remote_call, METH_VARARGS, "call a remote function" }, + { "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" }, { "start_worker_service", start_worker_service, METH_VARARGS, "start the worker service" }, { "scheduler_info", scheduler_info, METH_VARARGS, "get info about scheduler state" }, diff --git a/src/scheduler.cc b/src/scheduler.cc index 50e01764b..d76c8b337 100644 --- a/src/scheduler.cc +++ b/src/scheduler.cc @@ -8,8 +8,8 @@ SchedulerService::SchedulerService(SchedulingAlgorithmType scheduling_algorithm) : scheduling_algorithm_(scheduling_algorithm) {} -Status SchedulerService::RemoteCall(ServerContext* context, const RemoteCallRequest* request, RemoteCallReply* reply) { - std::unique_ptr task(new Call(request->call())); // need to copy, because request is const +Status SchedulerService::SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) { + std::unique_ptr task(new Task(request->task())); // need to copy, because request is const fntable_lock_.lock(); if (fntable_.find(task->name()) == fntable_.end()) { @@ -29,8 +29,8 @@ Status SchedulerService::RemoteCall(ServerContext* context, const RemoteCallRequ } { std::lock_guard reference_counts_lock(reference_counts_lock_); // we grab this lock because increment_ref_count assumes it has been acquired - increment_ref_count(result_objrefs); // We increment once so the objrefs don't go out of scope before we reply to the worker that called RemoteCall. The corresponding decrement will happen in remote_call in orchpylib. - increment_ref_count(result_objrefs); // We increment once so the objrefs don't go out of scope before the task is scheduled on the worker. The corresponding decrement will happen in deserialize_call in orchpylib. + increment_ref_count(result_objrefs); // We increment once so the objrefs don't go out of scope before we reply to the worker that called SubmitTask. The corresponding decrement will happen in submit_task in orchpylib. + increment_ref_count(result_objrefs); // We increment once so the objrefs don't go out of scope before the task is scheduled on the worker. The corresponding decrement will happen in deserialize_task in orchpylib. } task_queue_lock_.lock(); @@ -232,15 +232,15 @@ void SchedulerService::schedule() { perform_notify_aliases(); // See what we can do in alias_notification_queue_ } -void SchedulerService::submit_task(std::unique_ptr call, WorkerId workerid) { +void SchedulerService::assign_task(std::unique_ptr task, WorkerId workerid) { // submit task assumes that the canonical objrefs for its arguments are all ready, that is has_canonical_objref() is true for all of the call's arguments ClientContext context; - InvokeCallRequest request; - InvokeCallReply reply; + ExecuteTaskRequest request; + ExecuteTaskReply reply; ORCH_LOG(ORCH_INFO, "starting to send arguments"); - for (size_t i = 0; i < call->arg_size(); ++i) { - if (!call->arg(i).has_obj()) { - ObjRef objref = call->arg(i).ref(); + for (size_t i = 0; i < task->arg_size(); ++i) { + if (!task->arg(i).has_obj()) { + ObjRef objref = task->arg(i).ref(); ObjRef canonical_objref = get_canonical_objref(objref); { // Notify the relevant objstore about potential aliasing when it's ready @@ -249,7 +249,7 @@ void SchedulerService::submit_task(std::unique_ptr call, WorkerId workerid } attempt_notify_alias(get_store(workerid), objref, canonical_objref); - ORCH_LOG(ORCH_DEBUG, "call contains object ref " << canonical_objref); + ORCH_LOG(ORCH_DEBUG, "task contains object ref " << canonical_objref); std::lock_guard objtable_lock(objtable_lock_); auto &objstores = objtable_[canonical_objref]; std::lock_guard workers_lock(workers_lock_); @@ -258,11 +258,11 @@ void SchedulerService::submit_task(std::unique_ptr call, WorkerId workerid } } } - request.set_allocated_call(call.release()); // protobuf object takes ownership - Status status = workers_[workerid].worker_stub->InvokeCall(&context, request, &reply); + request.set_allocated_task(task.release()); // protobuf object takes ownership + Status status = workers_[workerid].worker_stub->ExecuteTask(&context, request, &reply); } -bool SchedulerService::can_run(const Call& task) { +bool SchedulerService::can_run(const Task& task) { std::lock_guard lock(objtable_lock_); for (int i = 0; i < task.arg_size(); ++i) { if (!task.arg(i).has_obj()) { @@ -410,8 +410,8 @@ void SchedulerService::get_info(const SchedulerInfoRequest& request, SchedulerIn } } for (const auto& entry : task_queue_) { - Call* call = reply->add_task(); - call->CopyFrom(*entry); + Task* task = reply->add_task(); + task->CopyFrom(*entry); } for (const WorkerId& entry : avail_workers_) { reply->add_avail_worker(entry); @@ -491,10 +491,10 @@ void SchedulerService::schedule_tasks_naively() { // The use of erase(it) below invalidates the iterator, but we // immediately break out of the inner loop, so the iterator is not used // after the erase - const Call& task = *(*it); + const Task& task = *(*it); auto& workers = fntable_[task.name()].workers(); if (std::binary_search(workers.begin(), workers.end(), workerid) && can_run(task)) { - submit_task(std::move(*it), workerid); + assign_task(std::move(*it), workerid); task_queue_.erase(it); std::swap(avail_workers_[i], avail_workers_[avail_workers_.size() - 1]); avail_workers_.pop_back(); @@ -516,7 +516,7 @@ void SchedulerService::schedule_tasks_location_aware() { auto bestit = task_queue_.end(); // keep track of the task that fits the worker best so far size_t min_num_shipped_objects = std::numeric_limits::max(); // number of objects that need to be transfered for this worker for (auto it = task_queue_.begin(); it != task_queue_.end(); ++it) { - const Call& task = *(*it); + const Task& task = *(*it); auto& workers = fntable_[task.name()].workers(); if (std::binary_search(workers.begin(), workers.end(), workerid) && can_run(task)) { // determine how many objects would need to be shipped @@ -542,7 +542,7 @@ void SchedulerService::schedule_tasks_location_aware() { } // if we found a suitable task if (bestit != task_queue_.end()) { - submit_task(std::move(*bestit), workerid); + assign_task(std::move(*bestit), workerid); task_queue_.erase(bestit); std::swap(avail_workers_[i], avail_workers_[avail_workers_.size() - 1]); avail_workers_.pop_back(); diff --git a/src/scheduler.h b/src/scheduler.h index 63d6d241c..db299ec9a 100644 --- a/src/scheduler.h +++ b/src/scheduler.h @@ -50,7 +50,7 @@ class SchedulerService : public Scheduler::Service { public: SchedulerService(SchedulingAlgorithmType scheduling_algorithm); - Status RemoteCall(ServerContext* context, const RemoteCallRequest* request, RemoteCallReply* reply) override; + Status SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) override; Status PushObj(ServerContext* context, const PushObjRequest* request, PushObjReply* reply) override; Status RequestObj(ServerContext* context, const RequestObjRequest* request, AckReply* reply) override; Status AliasObjRefs(ServerContext* context, const AliasObjRefsRequest* request, AckReply* reply) override; @@ -69,9 +69,9 @@ public: // assign a task to a worker void schedule(); // execute a task on a worker and ship required object references - void submit_task(std::unique_ptr call, WorkerId workerid); + void assign_task(std::unique_ptr task, WorkerId workerid); // checks if the dependencies of the task are met - bool can_run(const Call& task); + bool can_run(const Task& task); // register a worker and its object store (if it has not been registered yet) std::pair register_worker(const std::string& worker_address, const std::string& objstore_address); // register a new object with the scheduler and return its object reference @@ -144,7 +144,7 @@ private: FnTable fntable_; std::mutex fntable_lock_; // List of pending tasks. - std::deque > task_queue_; + std::deque > task_queue_; std::mutex task_queue_lock_; // List of pending pull calls. std::vector > pull_queue_; diff --git a/src/worker.cc b/src/worker.cc index d1b0373ff..002c4521f 100644 --- a/src/worker.cc +++ b/src/worker.cc @@ -8,11 +8,11 @@ extern "C" { static PyObject *OrchPyError; } -Status WorkerServiceImpl::InvokeCall(ServerContext* context, const InvokeCallRequest* request, InvokeCallReply* reply) { - call_ = request->call(); // Copy call - ORCH_LOG(ORCH_INFO, "invoked task " << request->call().name()); - Call* callptr = &call_; - send_queue_.send(&callptr); +Status WorkerServiceImpl::ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, ExecuteTaskReply* reply) { + task_ = request->task(); // Copy task + ORCH_LOG(ORCH_INFO, "invoked task " << request->task().name()); + Task* taskptr = &task_; + send_queue_.send(&taskptr); return Status::OK; } @@ -24,13 +24,13 @@ Worker::Worker(const std::string& worker_address, std::shared_ptr sched connected_ = true; } -RemoteCallReply Worker::remote_call(RemoteCallRequest* request) { +SubmitTaskReply Worker::submit_task(SubmitTaskRequest* request) { if (!connected_) { - ORCH_LOG(ORCH_FATAL, "Attempting to perform remote_call, but connected_ = " << connected_ << "."); + ORCH_LOG(ORCH_FATAL, "Attempting to perform submit_task, but connected_ = " << connected_ << "."); } - RemoteCallReply reply; + SubmitTaskReply reply; ClientContext context; - Status status = scheduler_stub_->RemoteCall(&context, *request, &reply); + Status status = scheduler_stub_->SubmitTask(&context, *request, &reply); return reply; } @@ -257,10 +257,10 @@ void Worker::register_function(const std::string& name, size_t num_return_vals) scheduler_stub_->RegisterFunction(&context, request, &reply); } -Call* Worker::receive_next_task() { - Call* call; - receive_queue_.receive(&call); - return call; +Task* Worker::receive_next_task() { + Task* task; + receive_queue_.receive(&task); + return task; } void Worker::notify_task_completed() { diff --git a/src/worker.h b/src/worker.h index 3453de16a..bdec7dfd5 100644 --- a/src/worker.h +++ b/src/worker.h @@ -29,19 +29,19 @@ public: : worker_address_(worker_address) { send_queue_.connect(worker_address_, false); } - Status InvokeCall(ServerContext* context, const InvokeCallRequest* request, InvokeCallReply* reply) override; + Status ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, ExecuteTaskReply* reply) override; private: std::string worker_address_; - Call call_; // copy of the current call - MessageQueue send_queue_; + Task task_; // copy of the current task + MessageQueue send_queue_; }; class Worker { public: Worker(const std::string& worker_address, std::shared_ptr scheduler_channel, std::shared_ptr objstore_channel); - // submit a remote call to the scheduler - RemoteCallReply remote_call(RemoteCallRequest* request); + // submit a remote task to the scheduler + SubmitTaskReply submit_task(SubmitTaskRequest* request); // send request to the scheduler to register this worker void register_worker(const std::string& worker_address, const std::string& objstore_address); // get a new object reference that is registered with the scheduler @@ -70,7 +70,7 @@ class Worker { // it in the message queue, which is read by the Python interpreter void start_worker_service(); // wait for next task from the RPC system - Call* receive_next_task(); + Task* receive_next_task(); // tell the scheduler that we are done with the current task and request the next one void notify_task_completed(); // disconnect the worker @@ -86,7 +86,7 @@ class Worker { std::unique_ptr scheduler_stub_; std::unique_ptr objstore_stub_; std::thread worker_server_thread_; - MessageQueue receive_queue_; + MessageQueue receive_queue_; managed_shared_memory segment_; WorkerId workerid_; ObjStoreId objstoreid_; diff --git a/test/microbenchmarks.py b/test/microbenchmarks.py index b6406bd00..ed0ccf16e 100644 --- a/test/microbenchmarks.py +++ b/test/microbenchmarks.py @@ -14,7 +14,7 @@ class MicroBenchmarkTest(unittest.TestCase): test_path = os.path.join(test_dir, "testrecv.py") services.start_singlenode_cluster(return_drivers=False, num_workers_per_objstore=3, worker_path=test_path) - # measure the time required to submit a remote call to the scheduler + # measure the time required to submit a remote task to the scheduler elapsed_times = [] for _ in range(1000): start_time = time.time() @@ -30,7 +30,7 @@ class MicroBenchmarkTest(unittest.TestCase): print " worst: {}".format(elapsed_times[999]) self.assertTrue(average_elapsed_time < 0.00038) - # measure the time required to submit a remote call to the scheduler (where the remote call returns one value) + # measure the time required to submit a remote task to the scheduler (where the remote task returns one value) elapsed_times = [] for _ in range(1000): start_time = time.time() @@ -46,7 +46,7 @@ class MicroBenchmarkTest(unittest.TestCase): print " worst: {}".format(elapsed_times[999]) self.assertTrue(average_elapsed_time < 0.001) - # measure the time required to submit a remote call to the scheduler and pull the result + # measure the time required to submit a remote task to the scheduler and pull the result elapsed_times = [] for _ in range(1000): start_time = time.time() diff --git a/test/runtest.py b/test/runtest.py index c91c4bbeb..86908e59a 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -117,13 +117,13 @@ class ObjStoreTest(unittest.TestCase): class SchedulerTest(unittest.TestCase): - def testCall(self): + def testRemoteTask(self): test_dir = os.path.dirname(os.path.abspath(__file__)) test_path = os.path.join(test_dir, "testrecv.py") [w] = services.start_singlenode_cluster(return_drivers=True, num_workers_per_objstore=1, worker_path=test_path) value_before = "test_string" - objref = w.remote_call("test_functions.print_string", [value_before]) + objref = w.submit_task("test_functions.print_string", [value_before]) time.sleep(0.2) @@ -172,11 +172,11 @@ class APITest(unittest.TestCase): test_path = os.path.join(test_dir, "testrecv.py") [w] = services.start_singlenode_cluster(return_drivers=True, num_workers_per_objstore=3, worker_path=test_path) - objref = w.remote_call("test_functions.test_alias_f", []) + objref = w.submit_task("test_functions.test_alias_f", []) self.assertTrue(np.alltrue(orchpy.pull(objref[0], w) == np.ones([3, 4, 5]))) - objref = w.remote_call("test_functions.test_alias_g", []) + objref = w.submit_task("test_functions.test_alias_g", []) self.assertTrue(np.alltrue(orchpy.pull(objref[0], w) == np.ones([3, 4, 5]))) - objref = w.remote_call("test_functions.test_alias_h", []) + objref = w.submit_task("test_functions.test_alias_h", []) self.assertTrue(np.alltrue(orchpy.pull(objref[0], w) == np.ones([3, 4, 5]))) services.cleanup()