X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/f2170465073de34adf89161d4287182b518352c4..2546a41ad36fdc9ef6471cb35a1d56930ae1b527:/rpc/rpc.h diff --git a/rpc/rpc.h b/rpc/rpc.h index c0420a5..d81a5dd 100644 --- a/rpc/rpc.h +++ b/rpc/rpc.h @@ -10,10 +10,11 @@ #include "thr_pool.h" #include "marshall.h" #include "connection.h" +#include "lock.h" -#ifdef DMALLOC -#include "dmalloc.h" -#endif +using std::string; +using std::map; +using std::list; class rpc_const { public: @@ -43,8 +44,8 @@ class rpcc : public chanmgr { unmarshall *un; int intret; bool done; - std::mutex m; - std::condition_variable c; + mutex m; + cond c; }; void get_refconn(connection **ch); @@ -62,27 +63,27 @@ class rpcc : public chanmgr { connection *chan_; - std::mutex m_; // protect insert/delete to calls[] - std::mutex chan_m_; + mutex m_; // protect insert/delete to calls[] + mutex chan_m_; bool destroy_wait_; - std::condition_variable destroy_wait_c_; + cond destroy_wait_c_; - std::map calls_; - std::list xid_rep_window_; + map calls_; + list xid_rep_window_; struct request { request() { clear(); } void clear() { buf.clear(); xid = -1; } bool isvalid() { return xid != -1; } - std::string buf; + string buf; int xid; }; struct request dup_req_; int xid_rep_done_; public: - rpcc(sockaddr_in d, bool retrans=true); + rpcc(const string & d, bool retrans=true); ~rpcc(); struct TO { @@ -99,27 +100,26 @@ class rpcc : public chanmgr { void set_reachable(bool r) { reachable_ = r; } void cancel(); - - int islossy() { return lossytest_ > 0; } - int call1(unsigned int proc, + int islossy() { return lossytest_ > 0; } + + int call1(proc_t proc, marshall &req, unmarshall &rep, TO to); bool got_pdu(connection *c, char *b, size_t sz); - template - int call_m(unsigned int proc, marshall &req, R & r, TO to); + int call_m(proc_t proc, marshall &req, R & r, TO to); template - inline int call(unsigned int proc, R & r, const Args&... args); + inline int call(proc_t proc, R & r, const Args&... args); template - inline int call_timeout(unsigned int proc, TO to, R & r, const Args&... args); + inline int call_timeout(proc_t proc, TO to, R & r, const Args&... args); }; template int -rpcc::call_m(unsigned int proc, marshall &req, R & r, TO to) +rpcc::call_m(proc_t proc, marshall &req, R & r, TO to) { unmarshall u; int intret = call1(proc, req, u, to); @@ -136,13 +136,13 @@ rpcc::call_m(unsigned int proc, marshall &req, R & r, TO to) } template inline int -rpcc::call(unsigned int proc, R & r, const Args&... args) +rpcc::call(proc_t proc, R & r, const Args&... args) { return call_timeout(proc, rpcc::to_max, r, args...); } template inline int -rpcc::call_timeout(unsigned int proc, const rpcc::TO to, R & r, const Args&... args) +rpcc::call_timeout(proc_t proc, const rpcc::TO to, R & r, const Args&... args) { marshall m{args...}; return call_m(proc, m, r, to); @@ -191,7 +191,7 @@ class rpcs : public chanmgr { // 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_; + map > reply_window_; void free_reply_window(void); void add_reply(unsigned int clt_nonce, int xid, char *b, size_t sz); @@ -200,26 +200,26 @@ class rpcs : public chanmgr { int xid, int rep_xid, char **b, size_t *sz); - void updatestat(unsigned int proc); + void updatestat(proc_t proc); // latest connection to the client - std::map conns_; + map conns_; // counting const size_t counting_; size_t curr_counts_; - std::map counts_; + map counts_; int lossytest_; bool reachable_; // map proc # to function - std::map procs_; + 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_ + mutex procs_m_; // protect insert/delete to procs[] + mutex count_m_; //protect modification of counts + mutex reply_window_m_; // protect reply window et al + mutex conss_m_; // protect conns_ protected: @@ -233,7 +233,7 @@ class rpcs : public chanmgr { void dispatch(djob_t *); // internal handler registration - void reg1(unsigned int proc, handler *); + void reg1(proc_t proc, handler *); ThrPool* dispatchpool_; tcpsconn* listener_; @@ -249,7 +249,7 @@ class rpcs : public chanmgr { bool got_pdu(connection *c, char *b, size_t sz); - template void reg(unsigned int proc, F f, C *c=nullptr); + template void reg(proc_t proc, F f, C *c=nullptr); }; struct ReturnOnFailure { @@ -258,12 +258,11 @@ struct ReturnOnFailure { } }; -template void rpcs::reg(unsigned int proc, F f, C *c) { +template void rpcs::reg(proc_t proc, F f, C *c) { reg1(proc, marshalled_func::wrap(f, c)); } -void make_sockaddr(const std::string &hostandport, struct sockaddr_in *dst); -void make_sockaddr(const std::string &host, const std::string &port, struct - sockaddr_in *dst); +sockaddr_in make_sockaddr(const string &hostandport); +sockaddr_in make_sockaddr(const string &host, const string &port); #endif