X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/4e881433f37417ccbda89c09ffdf936855d462d4..ab9eee5d7f1fbe7a3fe6229d4a78136efb14371b:/rpc/rpc.cc diff --git a/rpc/rpc.cc b/rpc/rpc.cc index 0c3a97d..e33b25e 100644 --- a/rpc/rpc.cc +++ b/rpc/rpc.cc @@ -52,7 +52,7 @@ // x exited worker threads). // -#include "rpc.h" +#include "include/rpc/rpc.h" #include #include @@ -60,68 +60,76 @@ #include #include -inline void set_rand_seed() { - auto now = time_point_cast(steady_clock::now()); - srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid()); -} +using std::list; +using namespace std::chrono; -static sockaddr_in make_sockaddr(const string &hostandport); +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_(), destroy_wait_ (false), xid_rep_done_(-1) +rpcc::rpcc(const string & d) : dst_(make_sockaddr(d)) { - if (retrans) { - set_rand_seed(); - clt_nonce_ = (nonce_t)random(); - } else { - // special client nonce 0 means this client does not - // require at-most-once logic from the server - // because it uses tcp and never retries a failed connection - clt_nonce_ = 0; - } + clt_nonce_ = (nonce_t)global->random_generator(); char *loss_env = getenv("RPC_LOSSY"); 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("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)); + lock ml(m_); + cancel(ml); + + lock cl(chan_m_); + 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; - int ret = call_timeout(rpc_protocol::bind, to, 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; -}; +} + +shared_ptr rpcc::bind_cached(const string & destination) { + auto client = global->get_handle(destination); + lock cl = lock(client->bind_m_); + if (!client->bind_done_) { + LOG_NONMEMBER << "bind(\"" << destination << "\")"; + int ret = client->bind(1000ms); + if (ret < 0) { + LOG_NONMEMBER << "bind failure! " << destination << " " << ret; + client.reset(); + } else { + LOG_NONMEMBER << "bind succeeded " << destination; + } + } + return client; +} + +void rpcc::unbind_cached(const string & destination) { + global->erase_handle(destination); +} // Cancel all outstanding calls -void rpcc::cancel(void) { - lock ml(m_); +void rpcc::cancel(lock & m_lock) { + VERIFY(m_lock); if (calls_.size()) { - LOG("force callers to fail"); - for (auto &p : calls_) { + 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; @@ -131,21 +139,22 @@ void rpcc::cancel(void) { destroy_wait_ = true; while (calls_.size () > 0) - destroy_wait_c_.wait(ml); + destroy_wait_c_.wait(m_lock); - LOG("done"); + LOG << "done"; } } -int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { +int rpcc::call_marshalled(const rpc_protocol::proc_t & proc, milliseconds to, string & rep, const marshall & req) { caller ca(0, &rep); xid_t xid_rep; + string datagram; { 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 ((proc.id != rpc_protocol::bind.id && !bind_done_) || (proc.id == rpc_protocol::bind.id && bind_done_)) { + IF_LEVEL(1) LOG << "rpcc has not been bound to dst or binding twice"; return rpc_protocol::bind_failure; } @@ -155,9 +164,9 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { ca.xid = xid_++; calls_[ca.xid] = &ca; - req.pack_header(rpc_protocol::request_header{ - ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front() - }); + datagram = marshall::datagram(rpc_protocol::request_header{ + ca.xid, proc.id, clt_nonce_, srv_nonce_, xid_rep_window_.front() + }, req); xid_rep = xid_rep_window_.front(); } @@ -169,7 +178,7 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { while (1) { if (transmit) { - get_refconn(ch); + get_latest_connection(ch); if (ch) { if (reachable_) { request forgot; @@ -182,29 +191,29 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { } if (forgot.isvalid()) ch->send(forgot.buf); - ch->send(req); + ch->send(datagram); } - 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.id + << " 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; } } @@ -212,11 +221,9 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { if (nextdeadline >= finaldeadline) break; - if (retrans_ && (!ch || ch->isdead())) { - // since connection is dead, retransmit - // on the new connection + // retransmit on new connection if connection is dead + if (!ch || ch->isdead()) transmit = true; - } } { @@ -226,7 +233,7 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { // may need to update the xid again here, in case the // packet times out before it's even sent by the channel. // I don't think there's any harm in maybe doing it twice - update_xid_rep(ca.xid); + update_xid_rep(ca.xid, ml); if (destroy_wait_) destroy_wait_c_.notify_one(); @@ -236,7 +243,7 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { { lock ml(m_); if (!dup_req_.isvalid()) { - dup_req_.buf = req; + dup_req_.buf = datagram; dup_req_.xid = ca.xid; } if (xid_rep > xid_rep_done_) @@ -245,15 +252,15 @@ int rpcc::call1(proc_id_t proc, marshall &req, string &rep, milliseconds to) { 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.id + << " 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); + return ca.done ? ca.intret : rpc_protocol::timeout_failure; } -void rpcc::get_refconn(shared_ptr & ch) { +void rpcc::get_latest_connection(shared_ptr & ch) { lock ml(chan_m_); if (!chan_ || chan_->isdead()) chan_ = connection::to_dst(dst_, this, lossytest_); @@ -262,29 +269,22 @@ void rpcc::get_refconn(shared_ptr & ch) { ch = chan_; } -// PollMgr's thread is being used to -// make this upcall from connection object to rpcc. -// this funtion must not block. -// -// this function keeps no reference for connection *c -bool -rpcc::got_pdu(const shared_ptr &, const string & b) -{ - unmarshall rep(b, true); +// Runs in poll_mgr's thread as an upcall from the connection object to the +// rpcc. Does not call blocking RPC handlers. +bool rpcc::got_pdu(const shared_ptr &, const string & b) { rpc_protocol::reply_header h; - rep.unpack_header(h); - if (!rep.ok()) { - IF_LEVEL(1) LOG("unmarshall header failed!!!"); + if (!unmarshall::datagram(b, h)) { + IF_LEVEL(1) LOG << "unmarshall header failed!!!"; return true; } lock ml(m_); - update_xid_rep(h.xid); + 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]; @@ -294,7 +294,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; } @@ -302,10 +302,8 @@ rpcc::got_pdu(const shared_ptr &, const string & b) return true; } -// assumes thread holds mutex m -void -rpcc::update_xid_rep(int xid) -{ +void rpcc::update_xid_rep(xid_t xid, lock & m_lock) { + VERIFY(m_lock); if (xid <= xid_rep_window_.front()) return; @@ -325,15 +323,12 @@ compress: } } -rpcs::rpcs(in_port_t p1) - : port_(p1), reachable_ (true) +rpcs::rpcs(in_port_t p1) : port_(p1) { - set_rand_seed(); - nonce_ = (nonce_t)random(); - IF_LEVEL(2) LOG("created with nonce " << nonce_); + nonce_ = (nonce_t)global->random_generator(); + IF_LEVEL(2) LOG << "created with nonce " << nonce_; reg(rpc_protocol::bind, &rpcs::rpcbind, this); - dispatchpool_ = unique_ptr(new thread_pool(6, false)); } void rpcs::start() { @@ -345,50 +340,40 @@ rpcs::~rpcs() { // must delete listener before dispatchpool listener_ = nullptr; dispatchpool_ = nullptr; - free_reply_window(); } 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; } return dispatchpool_->addJob(std::bind(&rpcs::dispatch, this, c, b)); } -void rpcs::reg1(proc_id_t proc, handler *h) { - lock pl(procs_m_); - VERIFY(procs_.count(proc) == 0); - procs_[proc] = h; - VERIFY(procs_.count(proc) >= 1); -} - void rpcs::dispatch(shared_ptr c, const string & buf) { - unmarshall req(buf, true); - rpc_protocol::request_header h; - req.unpack_header(h); - proc_id_t proc = h.proc; - if (!req.ok()) { - IF_LEVEL(1) LOG("unmarshall header failed"); + auto req = unmarshall::datagram(buf, h); + + if (!req) { + 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); + proc_id_t proc = h.proc; + + 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 << proc; rh.ret = rpc_protocol::oldsrv_failure; - rep.pack_header(rh); - c->send(rep); + c->send(marshall::datagram(rh)); return; } @@ -397,66 +382,57 @@ 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; } f = procs_[proc]; } - rpcs::rpcstate_t stat; - string b1; - - if (h.clt_nonce) { - // have i seen this client before? - { - lock rwl(reply_window_m_); - // if we don't know about this clt_nonce, create a cleanup object - 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)); - } - } - - // save the latest good connection to the client - { - lock rwl(conns_m_); - if (conns_.find(h.clt_nonce) == conns_.end()) - conns_[h.clt_nonce] = c; - else if (conns_[h.clt_nonce]->create_time < c->create_time) - conns_[h.clt_nonce] = c; + // have i seen this client before? + { + lock rwl(reply_window_m_); + // if we don't know about this clt_nonce, create a cleanup object + 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); } + } - 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; + // save the latest good connection to the client + { + lock rwl(conns_m_); + if (conns_.find(h.clt_nonce) == conns_.end()) + conns_[h.clt_nonce] = c; + else if (conns_[h.clt_nonce]->create_time < c->create_time) + conns_[h.clt_nonce] = c; } - switch (stat) { + string stored_reply; + + switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, stored_reply)) { case NEW: // new request - rh.ret = (*f)(req, rep); - if (rh.ret == rpc_protocol::unmarshal_args_failure) { - LOG("failed to unmarshall the arguments. You are " << - "probably calling RPC 0x" << hex << proc << " with the wrong " << - "types of arguments."); - VERIFY(0); - } - VERIFY(rh.ret >= 0); + { + marshall 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" << std::hex << proc << " with the wrong " + << "types of arguments."; + VERIFY(0); + } + VERIFY(rh.ret >= 0); - rep.pack_header(rh); - b1 = rep; + stored_reply = marshall::datagram(rh, 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 " << stored_reply.size() << " for rpc " + << h.xid << ", proc " << std::hex << proc << " ret " << std::dec + << rh.ret << ", clt " << h.clt_nonce; - if (h.clt_nonce > 0) { - // only record replies for clients that require at-most-once logic - add_reply(h.clt_nonce, h.xid, b1); - } + add_reply(h.clt_nonce, h.xid, stored_reply); // get the latest connection to the client { @@ -465,18 +441,17 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { c = conns_[h.clt_nonce]; } - c->send(rep); + c->send(stored_reply); break; case INPROGRESS: // server is working on this request break; case DONE: // duplicate and we still have the response - c->send(b1); + c->send(stored_reply); 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); - c->send(rep); + c->send(marshall::datagram(rh)); break; } } @@ -496,12 +471,12 @@ void rpcs::dispatch(shared_ptr c, const string & buf) { // 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(nonce_t clt_nonce, xid_t xid, +rpcs::check_duplicate_and_update(nonce_t clt_nonce, xid_t xid, xid_t xid_rep, string & b) { lock rwl(reply_window_m_); - list &l = reply_window_[clt_nonce]; + list & l = reply_window_[clt_nonce]; VERIFY(l.size() > 0); VERIFY(xid >= xid_rep); @@ -543,36 +518,29 @@ rpcs::checkduplicate_and_update(nonce_t clt_nonce, xid_t xid, // rpcs::dispatch calls add_reply when it is sending a reply to an RPC, // 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(nonce_t clt_nonce, xid_t xid, const string & b) { lock rwl(reply_window_m_); // remember the RPC reply value - list &l = reply_window_[clt_nonce]; + list & l = reply_window_[clt_nonce]; list::iterator it = l.begin(); // skip to our place in the list 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); } } -void rpcs::free_reply_window(void) { - lock rwl(reply_window_m_); - reply_window_.clear(); -} - -int rpcs::rpcbind(nonce_t &r) { - IF_LEVEL(2) LOG("called return nonce " << nonce_); +rpc_protocol::status rpcs::rpcbind(nonce_t & r) { + IF_LEVEL(2) LOG << "called return nonce " << nonce_; r = nonce_; return 0; } -static sockaddr_in make_sockaddr(const string &hostandport) { +static sockaddr_in make_sockaddr(const string & hostandport) { string host = "127.0.0.1"; string port = hostandport; auto colon = hostandport.find(':'); @@ -581,7 +549,7 @@ static sockaddr_in make_sockaddr(const string &hostandport) { port = hostandport.substr(colon+1); } - sockaddr_in dst{}; // zero initialize + sockaddr_in dst = sockaddr_in(); // zero initialize dst.sin_family = AF_INET; struct in_addr a{inet_addr(host.c_str())}; @@ -592,12 +560,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; }