X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/ebd5aef6dc92accb509b1cc67eaf72159f35cdfa..06282fd37814c4a9d53bca089b048709b368f5b3:/rpc/rpc.cc?ds=inline diff --git a/rpc/rpc.cc b/rpc/rpc.cc index 6985498..de33675 100644 --- a/rpc/rpc.cc +++ b/rpc/rpc.cc @@ -60,6 +60,9 @@ #include #include +using std::list; +using namespace std::chrono; + inline void set_rand_seed() { auto now = time_point_cast(steady_clock::now()); srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid()); @@ -76,27 +79,27 @@ rpcc::rpcc(const string & d) : dst_(make_sockaddr(d)) if (loss_env) lossytest_ = atoi(loss_env); - IF_LEVEL(2) LOG("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() { cancel(); - IF_LEVEL(2) LOG("delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1)); + IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1); chan_.reset(); VERIFY(calls_.size() == 0); } int rpcc::bind(milliseconds to) { - nonce_t r; + nonce_t r = 0; rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r); if (ret == 0) { lock ml(m_); bind_done_ = true; srv_nonce_ = r; } else { - IF_LEVEL(2) LOG("bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret); + IF_LEVEL(2) LOG << "bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret; } return ret; } @@ -105,11 +108,11 @@ int rpcc::bind(milliseconds to) { void rpcc::cancel(void) { lock ml(m_); if (calls_.size()) { - LOG("force callers to fail"); + LOG << "force callers to fail"; for (auto & p : calls_) { caller *ca = p.second; - IF_LEVEL(2) LOG("force caller to fail"); + IF_LEVEL(2) LOG << "force caller to fail"; lock cl(ca->m); ca->done = true; @@ -121,7 +124,7 @@ void rpcc::cancel(void) { while (calls_.size () > 0) destroy_wait_c_.wait(ml); - LOG("done"); + LOG << "done"; } } @@ -133,7 +136,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) { lock ml(m_); if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) { - IF_LEVEL(1) LOG("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_protocol::bind_failure; } @@ -143,7 +146,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) { ca.xid = xid_++; calls_[ca.xid] = &ca; - req.pack_header(rpc_protocol::request_header{ + req.write_header(rpc_protocol::request_header{ ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front() }); xid_rep = xid_rep_window_.front(); @@ -172,27 +175,27 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) { ch->send(forgot.buf); ch->send(req); } - else IF_LEVEL(1) LOG("not reachable"); - IF_LEVEL(2) LOG(clt_nonce_ << " just sent req proc " << hex << proc << - " xid " << dec << ca.xid << " clt_nonce " << clt_nonce_); + else IF_LEVEL(1) LOG << "not reachable"; + IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc + << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_; } transmit = false; // only send once on a given channel } - auto nextdeadline = min(steady_clock::now() + curr_to, finaldeadline); + auto nextdeadline = std::min(steady_clock::now() + curr_to, finaldeadline); curr_to *= 2; { lock cal(ca.m); while (!ca.done) { - IF_LEVEL(2) LOG("wait"); - if (ca.c.wait_until(cal, nextdeadline) == cv_status::timeout) { - IF_LEVEL(2) LOG("timeout"); + IF_LEVEL(2) LOG << "wait"; + if (ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout) { + IF_LEVEL(2) LOG << "timeout"; break; } } if (ca.done) { - IF_LEVEL(2) LOG("reply received"); + IF_LEVEL(2) LOG << "reply received"; break; } } @@ -231,9 +234,9 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) { lock cal(ca.m); - IF_LEVEL(2) LOG(clt_nonce_ << " call done for req proc " << hex << proc << - " xid " << dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":" << - ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret); + IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc + << " xid " << std::dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":" + << ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret; // destruction of req automatically frees its buffer return (ca.done? ca.intret : rpc_protocol::timeout_failure); @@ -258,10 +261,10 @@ rpcc::got_pdu(const shared_ptr &, const string & b) { unmarshall rep(b, true); rpc_protocol::reply_header h; - rep.unpack_header(h); + rep.read_header(h); if (!rep.ok()) { - IF_LEVEL(1) LOG("unmarshall header failed!!!"); + IF_LEVEL(1) LOG << "unmarshall header failed!!!"; return true; } @@ -270,7 +273,7 @@ rpcc::got_pdu(const shared_ptr &, const string & b) update_xid_rep(h.xid, ml); if (calls_.find(h.xid) == calls_.end()) { - IF_LEVEL(2) LOG("xid " << h.xid << " no pending request"); + IF_LEVEL(2) LOG << "xid " << h.xid << " no pending request"; return true; } caller *ca = calls_[h.xid]; @@ -280,7 +283,7 @@ rpcc::got_pdu(const shared_ptr &, const string & b) *ca->rep = b; ca->intret = h.ret; if (ca->intret < 0) { - IF_LEVEL(2) LOG("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; } @@ -313,7 +316,7 @@ rpcs::rpcs(in_port_t p1) : port_(p1) { set_rand_seed(); nonce_ = (nonce_t)random(); - IF_LEVEL(2) LOG("created with nonce " << nonce_); + IF_LEVEL(2) LOG << "created with nonce " << nonce_; reg(rpc_protocol::bind, &rpcs::rpcbind, this); } @@ -331,7 +334,7 @@ rpcs::~rpcs() { bool rpcs::got_pdu(const shared_ptr & c, const string & b) { if (!reachable_) { - IF_LEVEL(1) LOG("not reachable"); + IF_LEVEL(1) LOG << "not reachable"; return true; } @@ -342,26 +345,26 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { unmarshall req(buf, true); rpc_protocol::request_header h; - req.unpack_header(h); + req.read_header(h); proc_id_t proc = h.proc; if (!req.ok()) { - IF_LEVEL(1) LOG("unmarshall header failed"); + IF_LEVEL(1) LOG << "unmarshall header failed"; return; } - 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); + IF_LEVEL(2) LOG << "rpc " << h.xid << " (proc " << std::hex << proc << ", last_rep " + << std::dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce; marshall rep; rpc_protocol::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("rpc for an old server instance " << h.srv_nonce << - " (current " << nonce_ << ") proc " << hex << h.proc); + IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce + << " (current " << nonce_ << ") proc " << std::hex << h.proc; rh.ret = rpc_protocol::oldsrv_failure; - rep.pack_header(rh); + rep.write_header(rh); c->send(rep); return; } @@ -371,7 +374,7 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { { lock pl(procs_m_); if (procs_.count(proc) < 1) { - LOG("unknown proc 0x" << hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_); + LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_; VERIFY(0); return; } @@ -386,8 +389,8 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { 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("new client " << h.clt_nonce << " xid " << h.xid << - " chan " << c->fd << ", total clients " << (reply_window_.size()-1)); + IF_LEVEL(2) LOG << "new client " << h.clt_nonce << " xid " << h.xid + << " chan " << c->fd << ", total clients " << (reply_window_.size()-1); } } @@ -404,20 +407,21 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) { case NEW: // new request - rh.ret = (*f)(forward(req), rep); + rh.ret = (*f)(std::forward(req), rep); if (rh.ret == rpc_protocol::unmarshall_args_failure) { - LOG("failed to unmarshall the arguments. You are " << - "probably calling RPC 0x" << hex << proc << " with the wrong " << - "types of arguments."); + LOG << "failed to unmarshall the arguments. You are " + << "probably calling RPC 0x" << std::hex << proc << " with the wrong " + << "types of arguments."; VERIFY(0); } VERIFY(rh.ret >= 0); - rep.pack_header(rh); + rep.write_header(rh); b1 = rep; - 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_LEVEL(2) LOG << "sending and saving reply of size " << b1.size() << " for rpc " + << h.xid << ", proc " << std::hex << proc << " ret " << std::dec + << rh.ret << ", clt " << h.clt_nonce; add_reply(h.clt_nonce, h.xid, b1); @@ -436,9 +440,9 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { c->send(b1); break; case FORGOTTEN: // very old request and we don't have the response anymore - IF_LEVEL(2) LOG("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_protocol::atmostonce_failure; - rep.pack_header(rh); + rep.write_header(rh); c->send(rep); break; } @@ -515,7 +519,7 @@ void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) { for (it++; it != l.end() && it->xid < xid; it++); // there should already be an entry, so whine if there isn't if (it == l.end() || it->xid != xid) { - LOG("Could not find reply struct in add_reply"); + LOG << "Could not find reply struct in add_reply"; l.insert(it, reply_t(xid, b)); } else { *it = reply_t(xid, b); @@ -523,7 +527,7 @@ void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) { } rpc_protocol::status rpcs::rpcbind(nonce_t & r) { - IF_LEVEL(2) LOG("called return nonce " << nonce_); + IF_LEVEL(2) LOG << "called return nonce " << nonce_; r = nonce_; return 0; } @@ -548,12 +552,12 @@ static sockaddr_in make_sockaddr(const string & hostandport) { struct hostent *hp = gethostbyname(host.c_str()); if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) { - LOG_NONMEMBER("cannot find host name " << host); + LOG_NONMEMBER << "cannot find host name " << host; exit(1); } memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t)); dst.sin_addr.s_addr = a.s_addr; } - dst.sin_port = hton((in_port_t)stoi(port)); + dst.sin_port = hton((in_port_t)std::stoi(port)); return dst; }