diff --git a/lib/orchpy/orchpy/worker.pyx b/lib/orchpy/orchpy/worker.pyx index b2bd5d868..6d234995d 100644 --- a/lib/orchpy/orchpy/worker.pyx +++ b/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 = result.get_value() - serialize_into_2(0, ptr) - return orch_push(self.context, ptr) + return unison.ObjRef(orch_push(self.context, 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 = result.get_value() + cdef void* ptr = p + print "before put" + orch_put_obj(self.context, objref, 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, 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 = 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 diff --git a/lib/orchpy/setup.py b/lib/orchpy/setup.py index c5d13c814..231eea778 100644 --- a/lib/orchpy/setup.py +++ b/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]) diff --git a/src/orchlib.cc b/src/orchlib.cc index 2f953d179..ef600a2bf 100644 --- a/src/orchlib.cc +++ b/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); } diff --git a/src/orchlib.h b/src/orchlib.h index 7879a0f44..bac8e5ca7 100644 --- a/src/orchlib.h +++ b/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); } diff --git a/src/scheduler.cc b/src/scheduler.cc index a16f897da..3877f396f 100644 --- a/src/scheduler.cc +++ b/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 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 avail_workers_lock(avail_workers_lock_); + std::lock_guard fntable_lock(fntable_lock_); + std::lock_guard 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, WorkerId workerid 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 << "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 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; } } diff --git a/src/worker.cc b/src/worker.cc index f3a18d578..057735e3b 100644 --- a/src/worker.cc +++ b/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(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(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) { diff --git a/src/worker.h b/src/worker.h index 2ac0a4483..b72d9aba4 100644 --- a/src/worker.h +++ b/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 diff --git a/test/testprogram.py b/test/testprogram.py new file mode 100644 index 000000000..e207977a5 --- /dev/null +++ b/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() diff --git a/test/testrecv.py b/test/testrecv.py new file mode 100644 index 000000000..68295916e --- /dev/null +++ b/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)