diff --git a/lib/orchpy/orchpy/unison.pyx b/lib/orchpy/orchpy/unison.pyx index d6b1b017a..f0c1c9ee9 100644 --- a/lib/orchpy/orchpy/unison.pyx +++ b/lib/orchpy/orchpy/unison.pyx @@ -10,7 +10,7 @@ try: except: import pickle -cdef extern from "types.pb.h": +cdef extern from "../../../build/generated/types.pb.h": ctypedef enum DataType: INT32 INT64 diff --git a/lib/orchpy/orchpy/worker.pyx b/lib/orchpy/orchpy/worker.pyx index 5c1bc124b..53f81a2c5 100644 --- a/lib/orchpy/orchpy/worker.pyx +++ b/lib/orchpy/orchpy/worker.pyx @@ -24,16 +24,15 @@ cdef extern from "Python.h": int PyByteArray_Resize(object self, Py_ssize_t size) except -1 char* PyByteArray_AS_STRING(object bytearray) -cdef extern from "types.pb.h": +# cdef extern from "../../../build/generated/orchestra.pb.h": +# cdef cppclass RemoteCallRequest: +# RemoteCallRequest() +# void set_name(const char* value) +# Call* mutable_call() + +cdef extern from "../../../build/generated/types.pb.h": cdef cppclass Values -cdef extern from "orchestra.pb.h": - cdef cppclass RemoteCallRequest: - RemoteCallRequest() - void set_name(const char* value) - Values* mutable_arg() - -cdef extern from "types.pb.h": ctypedef enum DataType: INT32 INT64 @@ -140,13 +139,13 @@ cdef class Worker: def connect(self, server_addr, worker_addr, objstore_addr): self.context = orch_create_context(server_addr, worker_addr, objstore_addr) -# cpdef call(self, name, args): -# cdef RemoteCallRequest* result = new RemoteCallRequest() -# result[0].set_name(name) -# unison.serialize_args_into(args, result[0].mutable_arg()) -# for i in range(10): -# orch_remote_call(self.context, result) -# # return result +# cpdef call(self, name, args): +# cdef RemoteCallRequest* result = new RemoteCallRequest() +# result[0].set_name(name) +# unison.serialize_args_into(args, result[0].mutable_arg()) +# for i in range(10): +# orch_remote_call(self.context, result) +# # return result cpdef do_call(self, ptr): return orch_remote_call(self.context, ptr) diff --git a/protos/orchestra.proto b/protos/orchestra.proto index 28ccd8229..11d4057ed 100644 --- a/protos/orchestra.proto +++ b/protos/orchestra.proto @@ -63,7 +63,8 @@ message FnTableEntry { } message GetDebugInfoReply { - map function_table = 1; + repeated Call task = 1; + map function_table = 2; } service SchedulerServer { diff --git a/src/orchlib.cc b/src/orchlib.cc index 0c4aaf40f..1ca84c873 100644 --- a/src/orchlib.cc +++ b/src/orchlib.cc @@ -25,5 +25,5 @@ slice orch_get_serialized_obj(Worker* worker, ObjRef objref) { } void orch_register_function(Worker* worker, const char* name, size_t num_return_vals) { - // worker->register_function(std::string(name), num_return_vals); + worker->register_function(std::string(name), num_return_vals); } diff --git a/src/scheduler.h b/src/scheduler.h index 2ee32996a..2d21b34c5 100644 --- a/src/scheduler.h +++ b/src/scheduler.h @@ -2,6 +2,7 @@ #define ORCHESTRA_SCHEDULER_H #include +#include #include @@ -51,11 +52,11 @@ public: // returns number of return values of task size_t add_task(const Call& task) { fntable_lock_.lock(); - size_t num_return_vals = 2; // fn_table_[task.name()].num_return_vals(); + size_t num_return_vals = fntable_[task.name()].num_return_vals(); fntable_lock_.unlock(); - // std::unique_ptr task_ptr(new Call(task)); // TODO: perform copy outside + std::unique_ptr task_ptr(new Call(task)); // TODO: perform copy outside tasks_lock_.lock(); - // tasks_.push_back(task_ptr); + tasks_.emplace_back(std::move(task_ptr)); tasks_lock_.unlock(); return num_return_vals; } @@ -131,15 +132,21 @@ public: info.add_worker(workerid); fntable_lock_.unlock(); } - /* - void debug_info(DebugInfoReply* debug_info) { + void debug_info(GetDebugInfoReply* debug_info) { fntable_lock_.lock(); for (const auto& entry : fntable_) { - debug_info-> + auto function_table = debug_info->mutable_function_table(); + (*function_table)[entry.first].set_num_return_vals(entry.second.num_return_vals()); + // TODO: set workerid } - fntable_lock_.lock(); + fntable_lock_.unlock(); + tasks_lock_.lock(); + for (const auto& entry : tasks_) { + Call* call = debug_info->add_task(); + call->CopyFrom(*entry); + } + tasks_lock_.unlock(); } - */ }; #endif diff --git a/src/scheduler_server.h b/src/scheduler_server.h index d6739a23a..0fad71a00 100644 --- a/src/scheduler_server.h +++ b/src/scheduler_server.h @@ -38,6 +38,7 @@ public: return Status::OK; } Status GetDebugInfo(ServerContext* context, const GetDebugInfoRequest* request, GetDebugInfoReply* reply) override { + scheduler_->debug_info(reply); return Status::OK; } }; diff --git a/src/worker.h b/src/worker.h index 14b8bba81..58aa05c29 100644 --- a/src/worker.h +++ b/src/worker.h @@ -121,6 +121,8 @@ class Worker { void register_function(const std::string& name, size_t num_return_vals) { ClientContext context; RegisterFunctionRequest request; + request.set_fnname(name); + request.set_num_return_vals(num_return_vals); AckReply reply; scheduler_stub_->RegisterFunction(&context, request, &reply); } diff --git a/test/runtest.py b/test/runtest.py index 987aeba45..119b5d5b2 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -80,6 +80,14 @@ class SchedulerTest(unittest.TestCase): w = worker.Worker() w.connect("127.0.0.1:22221", "127.0.0.1:40002", "127.0.0.1:22222") w.register_function("hello_world", 2) + reply = scheduler_stub.GetDebugInfo(orchestra_pb2.GetDebugInfoRequest(), TIMEOUT_SECONDS) + self.assertEqual(reply.function_table.items()[0][0], u'hello_world') + + def testCall(self): + scheduler_channel = implementations.insecure_channel('localhost', 22221) + scheduler_stub = orchestra_pb2.beta_create_SchedulerServer_stub(scheduler_channel) + w = worker.Worker() + w.connect("127.0.0.1:22221", "127.0.0.1:40003", "127.0.0.1:22222") """