X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/46fb2b4bbe3a0a8516ab04cfafa895a882c70f86..5d99dbf06a14904944f5593c63705934bdfdcfb7:/rpc/rpc.cc diff --git a/rpc/rpc.cc b/rpc/rpc.cc index 90d9608..62003dd 100644 --- a/rpc/rpc.cc +++ b/rpc/rpc.cc @@ -71,6 +71,8 @@ inline void set_rand_seed() { srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid()); } +static sockaddr_in make_sockaddr(const string &hostandport); + rpcc::rpcc(const string & d, bool retrans) : dst_(make_sockaddr(d)), srv_nonce_(0), bind_done_(false), xid_(1), lossytest_(0), retrans_(retrans), reachable_(true), chan_(NULL), destroy_wait_ (false), xid_rep_done_(-1) @@ -86,20 +88,19 @@ rpcc::rpcc(const string & d, bool retrans) : } char *loss_env = getenv("RPC_LOSSY"); - if(loss_env != NULL){ + if(loss_env) lossytest_ = atoi(loss_env); - } // xid starts with 1 and latest received reply starts with 0 xid_rep_window_.push_back(0); - IF_LEVEL(2) LOG("rpcc::rpcc cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_); + IF_LEVEL(2) LOG("cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_); } // IMPORTANT: destruction should happen only when no external threads // are blocked inside rpcc or will use rpcc in the future rpcc::~rpcc() { - IF_LEVEL(2) LOG("rpcc::~rpcc delete nonce " << clt_nonce_ << " channo=" << (chan_?chan_->channo():-1)); + IF_LEVEL(2) LOG("delete nonce " << clt_nonce_ << " channo=" << (chan_?chan_->channo():-1)); if(chan_){ chan_->closeconn(); chan_->decref(); @@ -115,7 +116,7 @@ int rpcc::bind(TO to) { bind_done_ = true; srv_nonce_ = r; } else { - IF_LEVEL(2) LOG("rpcc::bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret); + IF_LEVEL(2) LOG("bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret); } return ret; }; @@ -123,11 +124,11 @@ int rpcc::bind(TO to) { // Cancel all outstanding calls void rpcc::cancel(void) { lock ml(m_); - LOG("rpcc::cancel: force callers to fail"); + LOG("force callers to fail"); for(auto &p : calls_){ caller *ca = p.second; - IF_LEVEL(2) LOG("rpcc::cancel: force caller to fail"); + IF_LEVEL(2) LOG("force caller to fail"); { lock cl(ca->m); ca->done = true; @@ -140,10 +141,10 @@ void rpcc::cancel(void) { destroy_wait_ = true; destroy_wait_c_.wait(ml); } - LOG("rpcc::cancel: done"); + LOG("done"); } -int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { +int rpcc::call1(proc_t proc, marshall &req, string &rep, TO to) { caller ca(0, &rep); int xid_rep; @@ -152,7 +153,7 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { if((proc != rpc_const::bind && !bind_done_) || (proc == rpc_const::bind && bind_done_)){ - IF_LEVEL(1) LOG("rpcc::call1 rpcc has not been bound to dst or binding twice"); + IF_LEVEL(1) LOG("rpcc has not been bound to dst or binding twice"); return rpc_const::bind_failure; } @@ -163,7 +164,7 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { ca.xid = xid_++; calls_[ca.xid] = &ca; - req.pack_req_header({ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()}); + req.pack_header(request_header{ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()}); xid_rep = xid_rep_window_.front(); } @@ -190,11 +191,11 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { } } if (forgot.isvalid()) - ch->send((char *)forgot.buf.c_str(), forgot.buf.size()); - ch->send(req.cstr(), req.size()); + ch->send(forgot.buf); + ch->send(req); } else IF_LEVEL(1) LOG("not reachable"); - IF_LEVEL(2) LOG("rpcc::call1 " << clt_nonce_ << " just sent req proc " << hex << proc << + IF_LEVEL(2) LOG(clt_nonce_ << " just sent req proc " << hex << proc << " xid " << dec << ca.xid << " clt_nonce " << clt_nonce_); } transmit = false; // only send once on a given channel @@ -212,14 +213,14 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { { lock cal(ca.m); while (!ca.done){ - IF_LEVEL(2) LOG("rpcc:call1: wait"); + IF_LEVEL(2) LOG("wait"); if(ca.c.wait_until(cal, nextdeadline) == cv_status::timeout){ - IF_LEVEL(2) LOG("rpcc::call1: timeout"); + IF_LEVEL(2) LOG("timeout"); break; } } if(ca.done){ - IF_LEVEL(2) LOG("rpcc::call1: reply received"); + IF_LEVEL(2) LOG("reply received"); break; } } @@ -250,7 +251,7 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { { lock ml(m_); if (!dup_req_.isvalid()) { - dup_req_.buf.assign(req.cstr(), req.size()); + dup_req_.buf = req; dup_req_.xid = ca.xid; } if (xid_rep > xid_rep_done_) @@ -259,7 +260,7 @@ int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { lock cal(ca.m); - IF_LEVEL(2) LOG("rpcc::call1 " << clt_nonce_ << " call done for req proc " << hex << proc << + IF_LEVEL(2) LOG(clt_nonce_ << " call done for req proc " << hex << proc << " xid " << dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":" << ntohs(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret); @@ -294,14 +295,14 @@ rpcc::get_refconn(connection **ch) // // this function keeps no reference for connection *c bool -rpcc::got_pdu(connection *, char *b, size_t sz) +rpcc::got_pdu(connection *, const string & b) { - unmarshall rep(b, sz); + unmarshall rep(b, true); reply_header h; - rep.unpack_reply_header(&h); + rep.unpack_header(h); if(!rep.ok()){ - IF_LEVEL(1) LOG("rpcc:got_pdu unmarshall header failed!!!"); + IF_LEVEL(1) LOG("unmarshall header failed!!!"); return true; } @@ -310,17 +311,17 @@ rpcc::got_pdu(connection *, char *b, size_t sz) update_xid_rep(h.xid); if(calls_.find(h.xid) == calls_.end()){ - IF_LEVEL(2) LOG("rpcc::got_pdu xid " << h.xid << " no pending request"); + IF_LEVEL(2) LOG("xid " << h.xid << " no pending request"); return true; } caller *ca = calls_[h.xid]; lock cl(ca->m); if(!ca->done){ - ca->un->take_in(rep); + *ca->rep = b; ca->intret = h.ret; if(ca->intret < 0){ - IF_LEVEL(2) LOG("rpcc::got_pdu: RPC reply error for xid " << h.xid << " intret " << ca->intret); + IF_LEVEL(2) LOG("RPC reply error for xid " << h.xid << " intret " << ca->intret); } ca->done = 1; } @@ -353,21 +354,17 @@ compress: } rpcs::rpcs(unsigned int p1, size_t count) - : port_(p1), counting_(count), curr_counts_(count), lossytest_(0), reachable_ (true) + : port_(p1), counting_(count), curr_counts_(count), reachable_ (true) { set_rand_seed(); nonce_ = (unsigned int)random(); - IF_LEVEL(2) LOG("rpcs::rpcs created with nonce " << nonce_); - - char *loss_env = getenv("RPC_LOSSY"); - if(loss_env != NULL){ - lossytest_ = atoi(loss_env); - } + IF_LEVEL(2) LOG("created with nonce " << nonce_); reg(rpc_const::bind, &rpcs::rpcbind, this); - dispatchpool_ = new ThrPool(6,false); + dispatchpool_ = new ThrPool(6, false); - listener_ = new tcpsconn(this, port_, lossytest_); + char *loss_env = getenv("RPC_LOSSY"); + listener_ = new tcpsconn(this, port_, loss_env ? atoi(loss_env) : 0); } rpcs::~rpcs() @@ -379,14 +376,14 @@ rpcs::~rpcs() } bool -rpcs::got_pdu(connection *c, char *b, size_t sz) +rpcs::got_pdu(connection *c, const string & b) { - if(!reachable_){ - IF_LEVEL(1) LOG("rpcss::got_pdu: not reachable"); - return true; - } + if(!reachable_){ + IF_LEVEL(1) LOG("not reachable"); + return true; + } - djob_t *j = new djob_t(c, b, sz); + djob_t *j = new djob_t{c, b}; c->incref(); bool succ = dispatchpool_->addJob(bind(&rpcs::dispatch, this, j)); if(!succ || !reachable_){ @@ -434,32 +431,32 @@ void rpcs::dispatch(djob_t *j) { connection *c = j->conn; - unmarshall req(j->buf, j->sz); + unmarshall req(j->buf, true); delete j; request_header h; - req.unpack_req_header(&h); + req.unpack_header(h); proc_t proc = h.proc; if(!req.ok()){ - IF_LEVEL(1) LOG("rpcs:dispatch unmarshall header failed!!!"); + IF_LEVEL(1) LOG("unmarshall header failed!!!"); c->decref(); return; } - IF_LEVEL(2) LOG("rpcs::dispatch: rpc " << h.xid << " (proc " << hex << proc << ", last_rep " << + IF_LEVEL(2) LOG("rpc " << h.xid << " (proc " << hex << proc << ", last_rep " << dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce); marshall rep; - reply_header rh(h.xid,0); + reply_header rh{h.xid,0}; // is client sending to an old instance of server? if(h.srv_nonce != 0 && h.srv_nonce != nonce_){ - IF_LEVEL(2) LOG("rpcs::dispatch: rpc for an old server instance " << h.srv_nonce << + IF_LEVEL(2) LOG("rpc for an old server instance " << h.srv_nonce << " (current " << nonce_ << ") proc " << hex << h.proc); rh.ret = rpc_const::oldsrv_failure; - rep.pack_reply_header(rh); - c->send(rep.cstr(),rep.size()); + rep.pack_header(rh); + c->send(rep); return; } @@ -468,7 +465,7 @@ rpcs::dispatch(djob_t *j) { lock pl(procs_m_); if(procs_.count(proc) < 1){ - cerr << "rpcs::dispatch: unknown proc " << hex << proc << "." << endl; + cerr << "unknown proc " << hex << proc << "." << endl; c->decref(); VERIFY(0); return; @@ -478,8 +475,7 @@ rpcs::dispatch(djob_t *j) } rpcs::rpcstate_t stat; - char *b1 = nullptr; - size_t sz1 = 0; + string b1; if(h.clt_nonce){ // have i seen this client before? @@ -489,7 +485,7 @@ rpcs::dispatch(djob_t *j) if(reply_window_.find(h.clt_nonce) == reply_window_.end()){ VERIFY (reply_window_[h.clt_nonce].size() == 0); // create reply_window_[h.clt_nonce].push_back(reply_t(-1)); // store starting reply xid - IF_LEVEL(2) LOG("rpcs::dispatch: new client " << h.clt_nonce << " xid " << h.xid << + IF_LEVEL(2) LOG("new client " << h.clt_nonce << " xid " << h.xid << " chan " << c->channo() << ", total clients " << (reply_window_.size()-1)); } } @@ -507,37 +503,36 @@ rpcs::dispatch(djob_t *j) } } - stat = checkduplicate_and_update(h.clt_nonce, h.xid, - h.xid_rep, &b1, &sz1); + stat = checkduplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1); } else { // this client does not require at most once logic stat = NEW; } - switch (stat){ + switch (stat) { case NEW: // new request - if(counting_){ + if (counting_){ updatestat(proc); } rh.ret = (*f)(req, rep); if (rh.ret == rpc_const::unmarshal_args_failure) { - cerr << "rpcs::dispatch: failed to unmarshall the arguments. You are " << + cerr << "failed to unmarshall the arguments. You are " << "probably calling RPC 0x" << hex << proc << " with the wrong " << "types of arguments." << endl; VERIFY(0); } VERIFY(rh.ret >= 0); - rep.pack_reply_header(rh); - rep.take_buf(&b1,&sz1); + rep.pack_header(rh); + b1 = rep; - IF_LEVEL(2) LOG("rpcs::dispatch: sending and saving reply of size " << sz1 << " for rpc " << + IF_LEVEL(2) LOG("sending and saving reply of size " << b1.size() << " for rpc " << h.xid << ", proc " << hex << proc << " ret " << dec << rh.ret << ", clt " << h.clt_nonce); - if(h.clt_nonce > 0){ + if (h.clt_nonce > 0) { // only record replies for clients that require at-most-once logic - add_reply(h.clt_nonce, h.xid, b1, sz1); + add_reply(h.clt_nonce, h.xid, b1); } // get the latest connection to the client @@ -550,22 +545,18 @@ rpcs::dispatch(djob_t *j) } } - c->send(b1, sz1); - if(h.clt_nonce == 0){ - // reply is not added to at-most-once window, free it - free(b1); - } + c->send(rep); break; case INPROGRESS: // server is working on this request break; case DONE: // duplicate and we still have the response - c->send(b1, sz1); + c->send(b1); break; case FORGOTTEN: // very old request and we don't have the response anymore - IF_LEVEL(2) LOG("rpcs::dispatch: very old request " << h.xid << " from " << h.clt_nonce); + IF_LEVEL(2) LOG("very old request " << h.xid << " from " << h.clt_nonce); rh.ret = rpc_const::atmostonce_failure; - rep.pack_reply_header(rh); - c->send(rep.cstr(),rep.size()); + rep.pack_header(rh); + c->send(rep); break; } c->decref(); @@ -583,11 +574,11 @@ rpcs::dispatch(djob_t *j) // returns one of: // NEW: never seen this xid before. // INPROGRESS: seen this xid, and still processing it. -// DONE: seen this xid, previous reply returned in *b and *sz. +// DONE: seen this xid, previous reply returned in b. // FORGOTTEN: might have seen this xid, but deleted previous reply. rpcs::rpcstate_t rpcs::checkduplicate_and_update(unsigned int clt_nonce, int xid, - int xid_rep, char **b, size_t *sz) + int xid_rep, string & b) { lock rwl(reply_window_m_); @@ -602,10 +593,8 @@ rpcs::checkduplicate_and_update(unsigned int clt_nonce, int xid, if (past_xid_rep < xid_rep || past_xid_rep == -1) { // scan for deletion candidates - for (; it != l.end() && it->xid < xid_rep; it++) { - if (it->cb_present) - free(it->buf); - } + while (it != l.end() && it->xid < xid_rep) + it++; l.erase(start, it); l.begin()->xid = xid_rep; } @@ -621,12 +610,10 @@ rpcs::checkduplicate_and_update(unsigned int clt_nonce, int xid, if (it != l.end() && it->xid == xid) { if (it->cb_present) { // return information about the remembered reply - *b = it->buf; - *sz = it->sz; + b = it->buf; return DONE; - } else { - return INPROGRESS; } + return INPROGRESS; } else { // remember that a new request has arrived l.insert(it, reply_t(xid)); @@ -635,14 +622,11 @@ rpcs::checkduplicate_and_update(unsigned int clt_nonce, int xid, } // rpcs::dispatch calls add_reply when it is sending a reply to an RPC, -// and passes the return value in b and sz. -// add_reply() should remember b and sz. -// free_reply_window() and checkduplicate_and_update is responsible for -// calling free(b). -void -rpcs::add_reply(unsigned int clt_nonce, int xid, - char *b, size_t sz) -{ +// and passes the return value in b. +// add_reply() should remember b. +// free_reply_window() and checkduplicate_and_update are responsible for +// cleaning up the remembered values. +void rpcs::add_reply(unsigned int clt_nonce, int xid, const string & b) { lock rwl(reply_window_m_); // remember the RPC reply value list &l = reply_window_[clt_nonce]; @@ -652,38 +636,26 @@ rpcs::add_reply(unsigned int clt_nonce, int xid, // there should already be an entry, so whine if there isn't if (it == l.end() || it->xid != xid) { cerr << "Could not find reply struct in add_reply" << endl; - l.insert(it, reply_t(xid, b, sz)); + l.insert(it, reply_t(xid, b)); } else { - *it = reply_t(xid, b, sz); + *it = reply_t(xid, b); } } void rpcs::free_reply_window(void) { lock rwl(reply_window_m_); - for (auto clt : reply_window_) { - for (auto it : clt.second){ - if (it.cb_present) - free(it.buf); - } - clt.second.clear(); - } reply_window_.clear(); } int rpcs::rpcbind(unsigned int &r, int) { - IF_LEVEL(2) LOG("rpcs::rpcbind called return nonce " << nonce_); + IF_LEVEL(2) LOG("called return nonce " << nonce_); r = nonce_; return 0; } -bool operator<(const sockaddr_in &a, const sockaddr_in &b){ - return ((a.sin_addr.s_addr < b.sin_addr.s_addr) || - ((a.sin_addr.s_addr == b.sin_addr.s_addr) && - ((a.sin_port < b.sin_port)))); -} +static sockaddr_in make_sockaddr(const string &host, const string &port); -/*---------------auxilary function--------------*/ -sockaddr_in make_sockaddr(const string &hostandport) { +static sockaddr_in make_sockaddr(const string &hostandport) { auto colon = hostandport.find(':'); if (colon == string::npos) return make_sockaddr("127.0.0.1", hostandport); @@ -691,7 +663,7 @@ sockaddr_in make_sockaddr(const string &hostandport) { return make_sockaddr(hostandport.substr(0, colon), hostandport.substr(colon+1)); } -sockaddr_in make_sockaddr(const string &host, const string &port) { +static sockaddr_in make_sockaddr(const string &host, const string &port) { sockaddr_in dst; bzero(&dst, sizeof(dst)); dst.sin_family = AF_INET;