X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/130f2d53438eb6193accb445aca52fa8e2fe4158..4b9798f44ae94deabf87dd534337b55259272950:/rpc/rpc.h diff --git a/rpc/rpc.h b/rpc/rpc.h index f7245ad..245f277 100644 --- a/rpc/rpc.h +++ b/rpc/rpc.h @@ -16,15 +16,21 @@ #endif class rpc_const { - public: - static const unsigned int bind = 1; // handler number reserved for bind - static const int timeout_failure = -1; - static const int unmarshal_args_failure = -2; - static const int unmarshal_reply_failure = -3; - static const int atmostonce_failure = -4; - static const int oldsrv_failure = -5; - static const int bind_failure = -6; - static const int cancel_failure = -7; + public: + static const unsigned int bind = 1; // handler number reserved for bind + static const int timeout_failure = -1; + static const int unmarshal_args_failure = -2; + static const int unmarshal_reply_failure = -3; + static const int atmostonce_failure = -4; + static const int oldsrv_failure = -5; + static const int bind_failure = -6; + static const int cancel_failure = -7; +}; + +struct ReturnOnFailure { + static inline int unmarshall_args_failure() { + return rpc_const::unmarshal_args_failure; + } }; // rpc client endpoint. @@ -32,44 +38,44 @@ class rpc_const { // threaded: multiple threads can be sending RPCs, class rpcc : public chanmgr { - private: + private: - //manages per rpc info - struct caller { - caller(unsigned int xxid, unmarshall *un); - ~caller(); + //manages per rpc info + struct caller { + caller(unsigned int xxid, unmarshall *un); + ~caller(); - unsigned int xid; - unmarshall *un; - int intret; - bool done; + unsigned int xid; + unmarshall *un; + int intret; + bool done; std::mutex m; std::condition_variable c; - }; + }; - void get_refconn(connection **ch); - void update_xid_rep(unsigned int xid); + void get_refconn(connection **ch); + void update_xid_rep(unsigned int xid); - sockaddr_in dst_; - unsigned int clt_nonce_; - unsigned int srv_nonce_; - bool bind_done_; - unsigned int xid_; - int lossytest_; - bool retrans_; - bool reachable_; + sockaddr_in dst_; + unsigned int clt_nonce_; + unsigned int srv_nonce_; + bool bind_done_; + unsigned int xid_; + int lossytest_; + bool retrans_; + bool reachable_; - connection *chan_; + connection *chan_; std::mutex m_; // protect insert/delete to calls[] - std::mutex chan_m_; + std::mutex chan_m_; - bool destroy_wait_; + bool destroy_wait_; std::condition_variable destroy_wait_c_; - std::map calls_; - std::list xid_rep_window_; + std::map calls_; + std::list xid_rep_window_; struct request { request() { clear(); } @@ -80,42 +86,42 @@ class rpcc : public chanmgr { }; struct request dup_req_; int xid_rep_done_; - public: + public: - rpcc(sockaddr_in d, bool retrans=true); - ~rpcc(); + rpcc(sockaddr_in d, bool retrans=true); + ~rpcc(); - struct TO { - int to; - }; - static const TO to_max; - static const TO to_min; - static TO to(int x) { TO t; t.to = x; return t;} + struct TO { + int to; + }; + static const TO to_max; + static const TO to_min; + static TO to(int x) { TO t; t.to = x; return t;} - unsigned int id() { return clt_nonce_; } + unsigned int id() { return clt_nonce_; } - int bind(TO to = to_max); + int bind(TO to = to_max); - void set_reachable(bool r) { reachable_ = r; } + void set_reachable(bool r) { reachable_ = r; } - void cancel(); + void cancel(); int islossy() { return lossytest_ > 0; } - int call1(unsigned int proc, - marshall &req, unmarshall &rep, TO to); + int call1(unsigned int proc, + marshall &req, unmarshall &rep, TO to); - bool got_pdu(connection *c, char *b, int sz); + bool got_pdu(connection *c, char *b, int sz); - template - int call_m(unsigned int proc, marshall &req, R & r, TO to); + template + int call_m(unsigned int proc, marshall &req, R & r, TO to); - template - inline int call(unsigned int proc, R & r, const Args&... args); + template + inline int call(unsigned int proc, R & r, const Args&... args); - template - inline int call_timeout(unsigned int proc, TO to, R & r, const Args&... args); + template + inline int call_timeout(unsigned int proc, TO to, R & r, const Args&... args); }; template int @@ -144,382 +150,120 @@ rpcc::call(unsigned int proc, R & r, const Args&... args) template inline int rpcc::call_timeout(unsigned int proc, const rpcc::TO to, R & r, const Args&... args) { - marshall m{args...}; - return call_m(proc, m, r, to); + marshall m{args...}; + return call_m(proc, m, r, to); } bool operator<(const sockaddr_in &a, const sockaddr_in &b); -class handler { - public: - handler() { } - virtual ~handler() { } - virtual int fn(unmarshall &, marshall &) = 0; -}; - - // rpc server endpoint. class rpcs : public chanmgr { - typedef enum { - NEW, // new RPC, not a duplicate - INPROGRESS, // duplicate of an RPC we're still processing - DONE, // duplicate of an RPC we already replied to (have reply) - FORGOTTEN, // duplicate of an old RPC whose reply we've forgotten - } rpcstate_t; + typedef enum { + NEW, // new RPC, not a duplicate + INPROGRESS, // duplicate of an RPC we're still processing + DONE, // duplicate of an RPC we already replied to (have reply) + FORGOTTEN, // duplicate of an old RPC whose reply we've forgotten + } rpcstate_t; - private: + private: // state about an in-progress or completed RPC, for at-most-once. // if cb_present is true, then the RPC is complete and a reply // has been sent; in that case buf points to a copy of the reply, // and sz holds the size of the reply. - struct reply_t { - reply_t (unsigned int _xid) { - xid = _xid; - cb_present = false; - buf = NULL; - sz = 0; - } - reply_t (unsigned int _xid, char *_buf, int _sz) { - xid = _xid; - cb_present = true; - buf = _buf; - sz = _sz; - } - unsigned int xid; - bool cb_present; // whether the reply buffer is valid - char *buf; // the reply buffer - int sz; // the size of reply buffer - }; - - int port_; - unsigned int nonce_; - - // provide at most once semantics by maintaining a window of replies - // per client that that client hasn't acknowledged receiving yet. + struct reply_t { + reply_t (unsigned int _xid) { + xid = _xid; + cb_present = false; + buf = NULL; + sz = 0; + } + reply_t (unsigned int _xid, char *_buf, int _sz) { + xid = _xid; + cb_present = true; + buf = _buf; + sz = _sz; + } + unsigned int xid; + bool cb_present; // whether the reply buffer is valid + char *buf; // the reply buffer + int sz; // the size of reply buffer + }; + + int port_; + unsigned int nonce_; + + // provide at most once semantics by maintaining a window of replies + // per client that that client hasn't acknowledged receiving yet. // indexed by client nonce. - std::map > reply_window_; - - void free_reply_window(void); - void add_reply(unsigned int clt_nonce, unsigned int xid, char *b, int sz); - - rpcstate_t checkduplicate_and_update(unsigned int clt_nonce, - unsigned int xid, unsigned int rep_xid, - char **b, int *sz); - - void updatestat(unsigned int proc); - - // latest connection to the client - std::map conns_; - - // counting - const int counting_; - int curr_counts_; - std::map counts_; - - int lossytest_; - bool reachable_; - - // map proc # to function - std::map procs_; - - std::mutex procs_m_; // protect insert/delete to procs[] - std::mutex count_m_; //protect modification of counts - std::mutex reply_window_m_; // protect reply window et al - std::mutex conss_m_; // protect conns_ - - - protected: - - struct djob_t { - djob_t (connection *c, char *b, int bsz):buf(b),sz(bsz),conn(c) {} - char *buf; - int sz; - connection *conn; - }; - void dispatch(djob_t *); - - // internal handler registration - void reg1(unsigned int proc, handler *); - - ThrPool* dispatchpool_; - tcpsconn* listener_; - - public: - rpcs(unsigned int port, int counts=0); - ~rpcs(); - inline int port() { return listener_->port(); } - //RPC handler for clients binding - int rpcbind(int a, int &r); - - void set_reachable(bool r) { reachable_ = r; } - - bool got_pdu(connection *c, char *b, int sz); - - // register a handler - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1 a1, R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1 a1, const A2, - R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2, - const A3, R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2, - const A3, const A4, R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2, - const A3, const A4, const A5, - R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2, - const A3, const A4, const A5, - const A6, R & r)); - template - void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2, - const A3, const A4, const A5, - const A6, const A7, - R & r)); -}; + std::map > reply_window_; -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - R r; - args >> a1; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + void free_reply_window(void); + void add_reply(unsigned int clt_nonce, unsigned int xid, char *b, int sz); -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - R r; - args >> a1; - args >> a2; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + rpcstate_t checkduplicate_and_update(unsigned int clt_nonce, + unsigned int xid, unsigned int rep_xid, + char **b, int *sz); -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - const A3 a3, R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - A3 a3; - R r; - args >> a1; - args >> a2; - args >> a3; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, a3, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + void updatestat(unsigned int proc); -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - const A3 a3, const A4 a4, - R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, - const A4 a4, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - A3 a3; - A4 a4; - R r; - args >> a1; - args >> a2; - args >> a3; - args >> a4; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, a3, a4, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + // latest connection to the client + std::map conns_; -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - const A3 a3, const A4 a4, - const A5 a5, R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4, - const A5 a5, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, - const A4 a4, const A5 a5, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - A3 a3; - A4 a4; - A5 a5; - R r; - args >> a1; - args >> a2; - args >> a3; - args >> a4; - args >> a5; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, a3, a4, a5, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + // counting + const int counting_; + int curr_counts_; + std::map counts_; -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - const A3 a3, const A4 a4, - const A5 a5, const A6 a6, - R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4, - const A5 a5, const A6 a6, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, - const A4 a4, const A5 a5, const A6 a6, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - A3 a3; - A4 a4; - A5 a5; - A6 a6; - R r; - args >> a1; - args >> a2; - args >> a3; - args >> a4; - args >> a5; - args >> a6; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, a3, a4, a5, a6, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + int lossytest_; + bool reachable_; -template void -rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2, - const A3 a3, const A4 a4, - const A5 a5, const A6 a6, - const A7 a7, R & r)) -{ - class h1 : public handler { - private: - S * sob; - int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4, - const A5 a5, const A6 a6, const A7 a7, R & r); - public: - h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, - const A4 a4, const A5 a5, const A6 a6, - const A7 a7, R & r)) - : sob(xsob), meth(xmeth) { } - int fn(unmarshall &args, marshall &ret) { - A1 a1; - A2 a2; - A3 a3; - A4 a4; - A5 a5; - A6 a6; - A7 a7; - R r; - args >> a1; - args >> a2; - args >> a3; - args >> a4; - args >> a5; - args >> a6; - args >> a7; - if(!args.okdone()) - return rpc_const::unmarshal_args_failure; - int b = (sob->*meth)(a1, a2, a3, a4, a5, a6, a7, r); - ret << r; - return b; - } - }; - reg1(proc, new h1(sob, meth)); -} + // map proc # to function + std::map procs_; + + std::mutex procs_m_; // protect insert/delete to procs[] + std::mutex count_m_; //protect modification of counts + std::mutex reply_window_m_; // protect reply window et al + std::mutex conss_m_; // protect conns_ + + + protected: + + struct djob_t { + djob_t (connection *c, char *b, int bsz):buf(b),sz(bsz),conn(c) {} + char *buf; + int sz; + connection *conn; + }; + void dispatch(djob_t *); + // internal handler registration + void reg1(unsigned int proc, handler *); + + ThrPool* dispatchpool_; + tcpsconn* listener_; + + public: + rpcs(unsigned int port, int counts=0); + ~rpcs(); + inline int port() { return listener_->port(); } + //RPC handler for clients binding + int rpcbind(int &r, int a); + + void set_reachable(bool r) { reachable_ = r; } + + bool got_pdu(connection *c, char *b, int sz); + + template void reg(unsigned int proc, F f, C *c=nullptr); +}; + +template void rpcs::reg(unsigned int proc, F f, C *c) { + reg1(proc, marshalled_func::wrap(f, c)); +} void make_sockaddr(const char *hostandport, struct sockaddr_in *dst); void make_sockaddr(const char *host, const char *port, - struct sockaddr_in *dst); + struct sockaddr_in *dst); #endif