wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-11 11:24:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

clean up queues

Browse Source
This commit is contained in:
Philipp Moritz
2016-03-17 22:56:55 -07:00
parent 60f582b98d
commit c1459f3603
2 changed files with 11 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/worker.h
+5 -7
View File
@@ -6,11 +6,6 @@
#include <string>
#include <thread>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
using namespace boost::interprocess;
#include <grpc++/grpc++.h>
using grpc::Server;
@@ -29,11 +24,14 @@ using grpc::ClientWriter;
class WorkerServiceImpl final : public WorkerService::Service {
public:
WorkerServiceImpl(const std::string& worker_address)
: worker_address_(worker_address) {}
: worker_address_(worker_address) {
send_queue_.connect(worker_address_, false);
}
Status InvokeCall(ServerContext* context, const InvokeCallRequest* request, InvokeCallReply* reply) override;
private:
std::string worker_address_;
Call call_; // copy of the current call
MessageQueue<Call*> send_queue_;
};
class Worker {
@@ -67,7 +65,7 @@ class Worker {
std::unique_ptr<Scheduler::Stub> scheduler_stub_;
std::unique_ptr<ObjStore::Stub> objstore_stub_;
std::thread worker_server_thread_;
std::unique_ptr<message_queue> receive_queue_;
MessageQueue<Call*> receive_queue_;
managed_shared_memory segment_;
WorkerId workerid_;
std::string worker_address_;