renaming Call -> Task (#70)

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

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

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

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<string, FnTableEntry> 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 {
 
 }

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

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<Call*>(PyCapsule_GetPointer(object, "call"));
+int PyObjectToTask(PyObject* object, Task **task) {
+  if (PyCapsule_IsValid(object, "task")) {
+    *task = static_cast<Task*>(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<Call*>(PyCapsule_GetPointer(capsule, "call"));
+void TaskCapsule_Destructor(PyObject* capsule) {
+  Task* obj = static_cast<Task*>(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<ObjRef> 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<ObjRef> 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<ObjRef> 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<ObjRef> 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<void*>(call), "call", &CallCapsule_Destructor);
+  return PyCapsule_New(static_cast<void*>(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<ObjRef> 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<ObjRef> 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<ObjRef> 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<void*>(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<void*>(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<ObjRef> 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" },

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<Call> 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> 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<std::mutex> 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> call, WorkerId workerid) {
+void SchedulerService::assign_task(std::unique_ptr<Task> 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> 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<std::mutex> objtable_lock(objtable_lock_);
       auto &objstores = objtable_[canonical_objref];
       std::lock_guard<std::mutex> workers_lock(workers_lock_);
@@ -258,11 +258,11 @@ void SchedulerService::submit_task(std::unique_ptr<Call> 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<std::mutex> 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<size_t>::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();

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> call, WorkerId workerid);
+  void assign_task(std::unique_ptr<Task> 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<WorkerId, ObjStoreId> 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<std::unique_ptr<Call> > task_queue_;
+  std::deque<std::unique_ptr<Task> > task_queue_;
   std::mutex task_queue_lock_;
   // List of pending pull calls.
   std::vector<std::pair<WorkerId, ObjRef> > pull_queue_;

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<Channel> 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() {

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<Call*> send_queue_;
+  Task task_; // copy of the current task
+  MessageQueue<Task*> send_queue_;
 };
 
 class Worker {
  public:
   Worker(const std::string& worker_address, std::shared_ptr<Channel> scheduler_channel, std::shared_ptr<Channel> 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> scheduler_stub_;
   std::unique_ptr<ObjStore::Stub> objstore_stub_;
   std::thread worker_server_thread_;
-  MessageQueue<Call*> receive_queue_;
+  MessageQueue<Task*> receive_queue_;
   managed_shared_memory segment_;
   WorkerId workerid_;
   ObjStoreId objstoreid_;

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()

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()