getting ready for some benchmarks

Makefile

@@ -4,7 +4,7 @@ LIB_PATH = lib/orchlib
 
 CXX = g++
 CPPFLAGS += -I/usr/local/include -pthread
-CXXFLAGS += -std=c++11 -fPIC -I$(SRC_PATH)
+CXXFLAGS += -std=c++11 -fPIC -I$(SRC_PATH) -O3
 LDFLAGS += -L/usr/local/lib -lgrpc++_unsecure -lgrpc -lprotobuf -lpthread -ldl
 PROTOC = protoc
 GRPC_CPP_PLUGIN = grpc_cpp_plugin
@@ -12,12 +12,12 @@ GRPC_CPP_PLUGIN_PATH ?= `which $(GRPC_CPP_PLUGIN)`
 
 vpath %.proto $(PROTOS_PATH)
 
-all: system-check $(LIB_PATH)/liborchlib.so $(SRC_PATH)/server
+all: system-check $(LIB_PATH)/liborchlib.so $(SRC_PATH)/server lib/orchpy/orchpy/liborchlib.so
 
-$(LIB_PATH)/liborchlib.so: $(SRC_PATH)/orchestra.pb.o $(SRC_PATH)/orchestra.grpc.pb.o $(LIB_PATH)/orchlib.o
+$(LIB_PATH)/liborchlib.so: $(SRC_PATH)/types.pb.o $(SRC_PATH)/orchestra.pb.o $(SRC_PATH)/types.grpc.pb.o $(SRC_PATH)/orchestra.grpc.pb.o $(LIB_PATH)/orchlib.o
 	$(CXX) $^ $(LDFLAGS) -shared -o $@
 
-$(SRC_PATH)/server: $(SRC_PATH)/orchestra.pb.o $(SRC_PATH)/orchestra.grpc.pb.o $(SRC_PATH)/server.o
+$(SRC_PATH)/server: $(SRC_PATH)/orchestra.pb.o $(SRC_PATH)/types.pb.o $(SRC_PATH)/types.grpc.pb.o $(SRC_PATH)/orchestra.grpc.pb.o $(SRC_PATH)/server.o
 	$(CXX) $^ $(LDFLAGS) -o $@
 
 .PRECIOUS: ./src/%.grpc.pb.cc
@@ -28,6 +28,9 @@ $(SRC_PATH)/%.grpc.pb.cc: %.proto
 $(SRC_PATH)/%.pb.cc: %.proto
 	$(PROTOC) -I $(PROTOS_PATH) --cpp_out=./src $<
 
+lib/orchpy/orchpy/liborchlib.so:
+	cp -f lib/orchlib/liborchlib.so lib/orchpy/orchpy/liborchlib.so
+
 clean:
 	rm -f $(SRC_PATH)/*.o $(LIB_PATH)/*.o $(SRC_PATH)/*.pb.cc $(SRC_PATH)/*.pb.h $(LIB_PATH)/orchlib.so $(SRC_PATH)/server
 

lib/orchlib/orchlib.cc

@@ -1,9 +1,15 @@
 #include <iostream>
 #include <memory>
 #include <string>
+#include <thread>
 
 #include <grpc++/grpc++.h>
 
+using grpc::Server;
+using grpc::ServerBuilder;
+using grpc::ServerContext;
+using grpc::Status;
+
 #include "orchestra.grpc.pb.h"
 #include "orchlib.h"
 
@@ -16,7 +22,7 @@ class Client {
   Client(std::shared_ptr<Channel> channel)
       : stub_(Orchestra::NewStub(channel)) {}
 
-  void RemoteCall(const std::string& name) {
+  size_t RemoteCall(const std::string& name) {
     RemoteCallRequest request;
     request.set_name(name);
 
@@ -25,6 +31,14 @@ class Client {
 
     Status status = stub_->RemoteCall(&context, request, &reply);
 
+    return reply.result();
+  }
+
+  void RegisterWorker() {
+    RegisterWorkerRequest request;
+    RegisterWorkerReply reply;
+    ClientContext context;
+    Status status = stub_->RegisterWorker(&context, request, &reply);
     return;
   }
 
@@ -32,14 +46,34 @@ class Client {
   std::unique_ptr<Orchestra::Stub> stub_;
 };
 
+class WorkerServiceImpl final : public Worker::Service {
+  Status InvokeCall(ServerContext* context, const InvokeCallRequest* request,
+    InvokeCallReply* reply) override {
+      std::cout << "invoke call request" << std::endl;
+      return Status::OK;
+  }
+};
+
+void start_server() {
+  std::string server_address("0.0.0.0:50053");
+  WorkerServiceImpl service;
+  ServerBuilder builder;
+  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+  builder.RegisterService(&service);
+  std::unique_ptr<Server> server(builder.BuildAndStart());
+  std::cout << "Server listening on " << server_address << std::endl;
+  server->Wait();
+}
+
 void* orch_create_context(const char* server_addr) {
   Client* client = new Client(grpc::CreateChannel("localhost:50052", grpc::InsecureChannelCredentials()));
+  client->RegisterWorker();
   return client;
 }
 
 size_t orch_remote_call(void* context, const char* name, void* args) {
   Client* client = (Client*)context;
-  client->RemoteCall(std::string(name));
+  return client->RemoteCall(std::string(name));
 }
 
 int main(int argc, char** argv) {

lib/orchlib/orchlib.h

@@ -3,4 +3,9 @@ extern "C" {
 void* orch_create_context(const char* server_addr);
 size_t orch_remote_call(void* context, const char* name, void* args);
 
+void* orch_arglist_create();
+void orch_arglist_add_ref(void* arglist, size_t ref);
+void orch_arglist_add_string(void* arglist, const char* str);
+void orch_arglist_destroy(void* arglist);
+
 }

lib/orchpy/orchpy/__init__.py

@@ -1 +1,6 @@
-from .context import Context
+import os, ctypes
+
+_orchlib_handle = ctypes.CDLL(
+   os.path.join(os.path.dirname(os.path.abspath(__file__)), 'liborchlib.so'),
+   ctypes.RTLD_GLOBAL
+)

lib/orchpy/setup.py

@@ -7,10 +7,19 @@ setup(
   name = "orchestra",
   version = "0.1.dev0",
   ext_modules = cythonize([
-    Extension("orchpy/context",
-      sources = ["orchpy/context.pyx"], libraries=["orchlib"],
-      library_dirs=['../orchlib/'])],
-    compiler_directives={'language_level': 3}),
+    Extension("orchpy/worker",
+      sources = ["orchpy/worker.pyx"],
+      extra_link_args=["-Iorchpy -lorchlib"]),
+    Extension("orchpy/unison",
+      include_dirs = ["../../src/"],
+      sources = ["orchpy/unison.pyx"],
+      extra_link_args=["-Iorchpy -lorchlib"],
+      language = "c++")],
+    compiler_directives={'language_level': 2}), # switch to 3 for python 3
   use_2to3=True,
-  packages=find_packages()
+  packages=find_packages(),
+  package_data = {
+    'orchpy': ['liborchlib.so']
+  },
+  zip_safe=False
 )

protos/orchestra.proto

@@ -1,5 +1,7 @@
 syntax = "proto3";
 
+import "types.proto";
+
 message RegisterWorkerRequest {
   string address = 1;
 }
@@ -8,14 +10,9 @@ message RegisterWorkerReply {
   uint64 workerid = 1;
 }
 
-message Value {
-  uint64 ref = 1; // for pass by reference
-  bytes data = 2; // for pass by value
-}
-
 message RemoteCallRequest {
   string name = 1;
-  repeated Value arg = 2;
+  Values arg = 2;
 }
 
 message RemoteCallReply {
@@ -32,5 +29,18 @@ service Orchestra {
   rpc RemoteCall(RemoteCallRequest) returns (RemoteCallReply);
   // rpc PushObject
   // rpc PullObject(PullObjectRequest)
-  // rpc DeliverRequest
+}
+
+message InvokeCallRequest {
+
+}
+
+message InvokeCallReply {
+
+}
+
+service Worker {
+  rpc InvokeCall(InvokeCallRequest) returns (InvokeCallReply);
+  // rpc PushObj(PushObjRequest) returns (PushObjReply);
+  // rpc RequestTransfer(RequestTransferRequest) returns (RequestTransferReply);
 }

src/server.cc

@@ -1,6 +1,7 @@
 #include <iostream>
 #include <memory>
 #include <string>
+#include <mutex>
 
 #include <grpc++/grpc++.h>
 
@@ -10,19 +11,55 @@ using grpc::Server;
 using grpc::ServerBuilder;
 using grpc::ServerContext;
 using grpc::Status;
-// using helloworld::HelloRequest;
-// using helloworld::HelloReply;
-// using helloworld::Greeter;
+
+typedef size_t ObjRef;
+typedef size_t WorkerId;
+typedef std::vector<std::vector<WorkerId> > ObjTable;
+
+class OrchestraScheduler {
+
+};
+
+class OrchestraServer {
+  ObjTable objtable;
+  std::mutex mutex;
+public:
+  ObjRef register_new_object() {
+    mutex.lock();
+    ObjRef result = objtable.size();
+    // std::cout << "size " << result << std::endl;
+    objtable.push_back(std::vector<WorkerId>());
+    mutex.unlock();
+    return result;
+  }
+  void register_object(ObjRef objref, WorkerId workerid) {
+    mutex.lock();
+    objtable[objref].push_back(workerid);
+    mutex.unlock();
+  }
+};
 
 // Logic and data behind the server's behavior.
 class OrchestraServiceImpl final : public Orchestra::Service {
+  ObjTable objtable;
+  std::unique_ptr<OrchestraServer> server;
+public:
+  OrchestraServiceImpl() : server(new OrchestraServer()) {
+  }
   Status RemoteCall(ServerContext* context, const RemoteCallRequest* request,
                   RemoteCallReply* reply) override {
-    std::cout << "called" << std::endl;
+    // std::cout << "called" << std::endl;
+    ObjRef objref = server->register_new_object();
+    reply->set_result(objref);
     // std::string prefix("Hello ");
     // reply->set_message(prefix + request->name());
     return Status::OK;
   }
+  Status RegisterWorker(ServerContext* context, const RegisterWorkerRequest* request,
+    RegisterWorkerReply* reply) override {
+      std::cout << "register worker" << std::endl;
+      return Status::OK;
+  }
 };
 
 void RunServer() {