function return working with one return value

lib/orchpy/orchpy/worker.pyx

@@ -21,6 +21,7 @@ cdef extern void orch_register_function(void* worker, const char* name, size_t n
 cdef extern size_t orch_remote_call(void* worker, void* request)
 cdef extern size_t orch_push(void* worker, void* value)
 cdef extern Slice orch_get_serialized_obj(void* worker, size_t objref)
+cdef extern void orch_put_obj(void* worker, size_t objref, void* obj)
 
 cdef extern from "Python.h":
     Py_ssize_t PyByteArray_GET_SIZE(object array)
@@ -44,6 +45,7 @@ cdef extern from "../../../build/generated/types.pb.h":
     void set_name(const char* value)
     const string& name()
     Value* mutable_arg(int index);
+    size_t result(int index);
     int arg_size() const;
 
   ctypedef enum DataType:
@@ -239,59 +241,80 @@ cdef class Worker:
 
   cpdef push(self, val):
     result = unison.serialize(val)
-    o = ObjWrapper()
     ptr = <uintptr_t>result.get_value()
-    serialize_into_2(0, ptr)
-    return orch_push(self.context, <void*>ptr)
+    return unison.ObjRef(orch_push(self.context, <void*>ptr), None)
 
   cpdef get_serialized(self, objref):
     cdef Slice slice = orch_get_serialized_obj(self.context, objref)
     data = PyBytes_FromStringAndSize(slice.ptr, slice.size)
     return data
 
+  cpdef put_obj(self, objref, obj):
+    result = unison.serialize(obj)
+    p = <uintptr_t>result.get_value()
+    cdef void* ptr = <void*>p
+    print "before put"
+    orch_put_obj(self.context, objref, <void*>ptr)
+    print "after put"
+
   cpdef do_pull(self, objref):
     cdef Slice slice = orch_get_serialized_obj(self.context, objref)
 
   cpdef pull(self, objref):
-    cdef Slice slice = orch_get_serialized_obj(self.context, objref)
+    print "before get_serialized_obj, getting", objref.get_id()
+    cdef Slice slice = orch_get_serialized_obj(self.context, <size_t>objref.get_id())
+    print "after get_serialized_ob"
     data = PyBytes_FromStringAndSize(slice.ptr, slice.size)
+    print "after get data"
     return unison.deserialize_from_string(data)
 
   cpdef register_function(self, func_name, function, num_args):
     orch_register_function(self.context, func_name, num_args)
     self.functions[func_name] = function
 
-  cpdef main_loop(self):
+  cpdef wait_for_next_task(self):
     result = []
     cdef Call* call = <Call*>orch_wait_for_next_task(self.context)
     cdef int size = call[0].arg_size()
     cdef Obj* obj
     print "hello from python"
     print "size", size
-    return call[0].name(), deserialize_args_from_call(call)
+    args = deserialize_args_from_call(call)
+    print "done deserializing"
+    return call[0].name(), args, call[0].result(0) # TODO: make this return multiple values
 
   cpdef invoke_function(self, name, args):
     return self.functions[name].executor(args)
 
+  cpdef main_loop(self):
+    while True:
+      name, args, returnref = self.wait_for_next_task()
+      print "got returnref", returnref
+      self.functions[name].executor(returnref, args)
+      # self.invoke_function(name, args)
+
+
 global_worker = Worker()
 
 def distributed(types, return_type, worker=global_worker):
     def distributed_decorator(func):
         # deserialize arguments, execute function and serialize result
-        def func_executor(args):
+        def func_executor(returnref, args):
             arguments = []
             for (i, arg) in enumerate(args):
+              print "pulling argument", i
               if type(arg) == unison.ObjRef:
                 if i < len(types) - 1:
-                  arguments.append(worker.get_object(arg, types[i]))
+                  arguments.append(worker.pull(arg))
                 elif i == len(types) - 1 and types[-1] is not None:
-                  arguments.append(global_worker.get_object(arg, types[i]))
+                  arguments.append(global_worker.pull(arg))
                 elif types[-1] is None:
-                  arguments.append(worker.get_object(arg, types[-2]))
+                  arguments.append(worker.pull(arg))
                 else:
                   raise Exception("Passed in " + str(len(args)) + " arguments to function " + func.__name__ + ", which takes only " + str(len(types)) + " arguments.")
               else:
                 arguments.append(arg)
+            print "done pulling argument", i
             # TODO
             # buf = bytearray()
             print "called with arguments", arguments
@@ -300,7 +323,11 @@ def distributed(types, return_type, worker=global_worker):
             #   raise Exception("Return type of " + func.func_name + " does not match the return type specified in the @distributed decorator, was expecting " + str(return_type) + " but received " + str(unison.unison_type(result)))
             # unison.serialize(buf, result)
             # return memoryview(buf).tobytes()
-            return result
+            # return result
+            # obj = ObjWrapper()
+            # serialize_into_2(result, obj.get_value())
+            # print "put was sucessful? seems like so"
+            worker.put_obj(returnref, result)
         # for remotely executing the function
         def func_call(*args, typecheck=False):
           return worker.call(func_call.func_name, func_call.module_name, args)
@@ -311,6 +338,3 @@ def distributed(types, return_type, worker=global_worker):
         func_call.types = types
         return func_call
     return distributed_decorator
-
-def pull(objref, worker=global_worker):
-  return 1

lib/orchpy/setup.py

@@ -31,3 +31,7 @@ setup(
   },
   zip_safe=False
 )
+
+extension_mod = Extension("symphony", ["orchpy/symphony.cpp"], include_dirs=["../../build/generated/"])
+
+setup(name = "symphony", ext_modules=[extension_mod])

src/orchlib.cc

@@ -28,6 +28,10 @@ slice orch_get_serialized_obj(Worker* worker, ObjRef objref) {
   return worker->get_serialized_obj(objref);
 }
 
+void orch_put_obj(Worker* worker, size_t objref, const Obj* obj) {
+  worker->put_obj(objref, obj);
+}
+
 void orch_register_function(Worker* worker, const char* name, size_t num_return_vals) {
   worker->register_function(std::string(name), num_return_vals);
 }

src/orchlib.h

@@ -21,6 +21,7 @@ size_t orch_remote_call(Worker* worker, RemoteCallRequest* request);
 size_t orch_push(Worker* worker, Obj* value);
 Call* orch_wait_for_next_task(Worker* worker);
 slice orch_get_serialized_obj(Worker* worker, size_t objref);
+void orch_put_obj(Worker* worker, size_t objref, const Obj* obj);
 void orch_register_function(Worker* worker, const char* name, size_t num_return_vals);
 
 }

src/scheduler.cc

@@ -15,6 +15,8 @@ Status SchedulerService::RemoteCall(ServerContext* context, const RemoteCallRequ
   tasks_lock_.lock();
   tasks_.emplace_back(std::move(task));
   tasks_lock_.unlock();
+
+  schedule();
   return Status::OK;
 }
 
@@ -23,6 +25,7 @@ Status SchedulerService::PushObj(ServerContext* context, const PushObjRequest* r
   ObjStoreId objstoreid = get_store(request->workerid());
   add_location(objref, objstoreid);
   reply->set_objref(objref);
+  schedule();
   return Status::OK;
 }
 
@@ -34,12 +37,14 @@ Status SchedulerService::RegisterWorker(ServerContext* context, const RegisterWo
   WorkerId workerid = register_worker(request->worker_address(), request->objstore_address());
   std::cout << "registered worker with workerid" << workerid << std::endl;
   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;
   register_function(request->fnname(), request->workerid(), request->num_return_vals());
+  schedule();
   return Status::OK;
 }
 
@@ -49,24 +54,23 @@ Status SchedulerService::GetDebugInfo(ServerContext* context, const GetDebugInfo
 }
 
 void SchedulerService::schedule() {
-  // TODO: work out a better strategy here
-  WorkerId workerid = 0;
-  {
-    std::lock_guard<std::mutex> lock(avail_workers_lock_);
-    if (avail_workers_.size() == 0)
-      return;
-    workerid = avail_workers_.back();
-    std::cout << "got available worker" << workerid << std::endl;
-    avail_workers_.pop_back();
-  }
-  // TODO: think about locking here
-  for (auto it = tasks_.begin(); it != tasks_.end(); ++it) {
-    const Call& 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);
-      tasks_.erase(it);
-      return;
+  // TODO: don't recheck if nothing changed
+  std::lock_guard<std::mutex> avail_workers_lock(avail_workers_lock_);
+  std::lock_guard<std::mutex> fntable_lock(fntable_lock_);
+  std::lock_guard<std::mutex> tasks_lock(tasks_lock_);
+  for (int i = 0; i < avail_workers_.size(); ++i) {
+    WorkerId workerid = avail_workers_[i];
+    for (auto it = tasks_.begin(); it != tasks_.end(); ++it) {
+      const Call& 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);
+        tasks_.erase(it);
+        std::swap(avail_workers_[i], avail_workers_[avail_workers_.size() - 1]);
+        avail_workers_.pop_back();
+        i -= 1;
+        break;
+      }
     }
   }
 }
@@ -83,7 +87,7 @@ void SchedulerService::submit_task(std::unique_ptr<Call> call, WorkerId workerid
       auto &objstores = objtable_[call->arg(i).ref()];
       std::lock_guard<std::mutex> workers_lock(workers_lock_);
       if (!std::binary_search(objstores.begin(), objstores.end(), workers_[workerid].objstoreid)) {
-        std::cout << "have to send" << std::endl;
+        std::cout << "lost object store, need to do recovery" << std::endl;
         std::exit(1);
       }
       // if (objstoreid != workers_[workerid].objstoreid) {
@@ -100,7 +104,8 @@ bool SchedulerService::can_run(const Call& task) {
   std::lock_guard<std::mutex> lock(objtable_lock_);
   for (int i = 0; i < task.arg_size(); ++i) {
     if (!task.arg(i).has_obj()) {
-      if (objtable_[task.arg(i).ref()].size() == 0) {
+      ObjRef objref = task.arg(i).ref();
+      if (objref >= objtable_.size() || objtable_[objref].size() == 0) {
         return false;
       }
     }

src/worker.cc

@@ -37,7 +37,7 @@ void Worker::register_worker(const std::string& worker_address, const std::strin
   return;
 }
 
-ObjRef Worker::push_obj(Obj* obj) {
+ObjRef Worker::push_obj(const Obj* obj) {
   // first get objref for the new object
   PushObjRequest push_request;
   PushObjReply push_reply;
@@ -45,6 +45,24 @@ ObjRef Worker::push_obj(Obj* obj) {
   Status push_status = scheduler_stub_->PushObj(&push_context, push_request, &push_reply);
   ObjRef objref = push_reply.objref();
   // then stream the object to the object store
+  put_obj(objref, obj);
+  return objref;
+}
+
+slice Worker::get_serialized_obj(ObjRef objref) {
+  ClientContext context;
+  GetObjRequest request;
+  request.set_objref(objref);
+  GetObjReply reply;
+  objstore_stub_->GetObj(&context, request, &reply);
+  segment_ = managed_shared_memory(open_only, reply.bucket().c_str());
+  slice slice;
+  slice.data = static_cast<char*>(segment_.get_address_from_handle(reply.handle()));
+  slice.len = reply.size();
+  return slice;
+}
+
+void Worker::put_obj(ObjRef objref, const Obj* obj) {
   ObjChunk chunk;
   std::string data;
   obj->SerializeToString(&data);
@@ -66,20 +84,7 @@ ObjRef Worker::push_obj(Obj* obj) {
   }
   writer->WritesDone();
   Status status = writer->Finish();
-  return objref;
-}
-
-slice Worker::get_serialized_obj(ObjRef objref) {
-  ClientContext context;
-  GetObjRequest request;
-  request.set_objref(objref);
-  GetObjReply reply;
-  objstore_stub_->GetObj(&context, request, &reply);
-  segment_ = managed_shared_memory(open_only, reply.bucket().c_str());
-  slice slice;
-  slice.data = static_cast<char*>(segment_.get_address_from_handle(reply.handle()));
-  slice.len = reply.size();
-  return slice;
+  // TODO: error handling
 }
 
 void Worker::register_function(const std::string& name, size_t num_return_vals) {

src/worker.h

@@ -50,9 +50,11 @@ class Worker {
   // send request to the scheduler to register this worker
   void register_worker(const std::string& worker_address, const std::string& objstore_address);
   // push object to local object store, register it with the server and return object reference
-  ObjRef push_obj(Obj* obj);
+  ObjRef push_obj(const Obj* obj);
   // retrieve serialized object from local object store
   slice get_serialized_obj(ObjRef objref);
+  // stores an object to the local object store
+  void put_obj(ObjRef objref, const Obj* obj);
   // register function with scheduler
   void register_function(const std::string& name, size_t num_return_vals);
   // start the worker server which accepts tasks from the scheduler and stores

test/testprogram.py

@@ -0,0 +1,16 @@
+import orchpy.unison as unison
+import orchpy.services as services
+import orchpy.worker as worker
+
+@worker.distributed([str], [str])
+def print_string(string):
+  print "called print_string with", string
+  return string
+
+if __name__ == '__main__':
+  worker.global_worker.connect("127.0.0.1:22221", "127.0.0.1:40000", "127.0.0.1:22222")
+
+  worker.global_worker.register_function("hello_world", print_string, 1)
+  worker.global_worker.start_worker_service()
+
+  worker.global_worker.main_loop()

test/testrecv.py

@@ -0,0 +1,16 @@
+import orchpy.unison as unison
+import orchpy.services as services
+import orchpy.worker as worker
+import sys
+
+if __name__ == '__main__':
+  worker = worker.Worker()
+  worker.connect("127.0.0.1:22221", "127.0.0.1:50000", "127.0.0.1:22222")
+  worker.start_worker_service()
+  worker.register_function("hello_world", None, 0)
+  name, args, returnref = worker.wait_for_next_task()
+  print "received args ", args
+  if args == ["hi"]:
+    sys.exit(0)
+  else:
+    sys.exit(1)