From fdbc30f0d7d6828252e98600cd7c41c830f79334 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Wed, 10 Feb 2016 12:12:19 -0800 Subject: [PATCH] getting ready for some benchmarks --- Makefile | 11 +++++---- lib/orchlib/orchlib.cc | 38 +++++++++++++++++++++++++++-- lib/orchlib/orchlib.h | 5 ++++ lib/orchpy/orchpy/__init__.py | 7 +++++- lib/orchpy/setup.py | 19 +++++++++++---- protos/orchestra.proto | 24 +++++++++++++------ src/server.cc | 45 +++++++++++++++++++++++++++++++---- 7 files changed, 126 insertions(+), 23 deletions(-) diff --git a/Makefile b/Makefile index 25ecc27d3..d2afb705d 100644 --- a/Makefile +++ b/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 diff --git a/lib/orchlib/orchlib.cc b/lib/orchlib/orchlib.cc index ca87b2755..9b9ddffb8 100644 --- a/lib/orchlib/orchlib.cc +++ b/lib/orchlib/orchlib.cc @@ -1,9 +1,15 @@ #include #include #include +#include #include +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) : 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 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(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) { diff --git a/lib/orchlib/orchlib.h b/lib/orchlib/orchlib.h index 694c3389c..f732c317b 100644 --- a/lib/orchlib/orchlib.h +++ b/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); + } diff --git a/lib/orchpy/orchpy/__init__.py b/lib/orchpy/orchpy/__init__.py index 0cebf2bfd..09ae04de5 100644 --- a/lib/orchpy/orchpy/__init__.py +++ b/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 +) diff --git a/lib/orchpy/setup.py b/lib/orchpy/setup.py index e0b12601e..c2002b6ed 100644 --- a/lib/orchpy/setup.py +++ b/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 ) diff --git a/protos/orchestra.proto b/protos/orchestra.proto index 9e1b49511..6761b9b03 100644 --- a/protos/orchestra.proto +++ b/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); } diff --git a/src/server.cc b/src/server.cc index 2147ac8dc..4501748a7 100644 --- a/src/server.cc +++ b/src/server.cc @@ -1,6 +1,7 @@ #include #include #include +#include #include @@ -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 > 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()); + 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 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() {