X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5a5c578e2e358a121cdb9234a6cb11c4ecfbf323..24bebc0ecf83446c7371eff69042322aab34976a:/rpc/rpc.cc?ds=sidebyside diff --git a/rpc/rpc.cc b/rpc/rpc.cc index ad3fcd9..9f1d90c 100644 --- a/rpc/rpc.cc +++ b/rpc/rpc.cc @@ -63,8 +63,6 @@ #include #include -#include "jsl_log.h" - const rpcc::TO rpcc::to_max = { 120000 }; const rpcc::TO rpcc::to_min = { 1000 }; @@ -95,16 +93,13 @@ rpcc::rpcc(const string & d, bool retrans) : // xid starts with 1 and latest received reply starts with 0 xid_rep_window_.push_back(0); - jsl_log(JSL_DBG_2, "rpcc::rpcc cltn_nonce is %d lossy %d\n", - clt_nonce_, lossytest_); + IF_LEVEL(2) LOG("rpcc::rpcc 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() -{ - jsl_log(JSL_DBG_2, "rpcc::~rpcc delete nonce %d channo=%d\n", - clt_nonce_, chan_?chan_->channo():-1); +rpcc::~rpcc() { + IF_LEVEL(2) LOG("rpcc::~rpcc delete nonce " << clt_nonce_ << " channo=" << (chan_?chan_->channo():-1)); if(chan_){ chan_->closeconn(); chan_->decref(); @@ -112,9 +107,7 @@ rpcc::~rpcc() VERIFY(calls_.size() == 0); } -int -rpcc::bind(TO to) -{ +int rpcc::bind(TO to) { unsigned int r; int ret = call_timeout(rpc_const::bind, to, r, 0); if(ret == 0){ @@ -122,22 +115,19 @@ rpcc::bind(TO to) bind_done_ = true; srv_nonce_ = r; } else { - jsl_log(JSL_DBG_2, "rpcc::bind %s failed %d\n", - inet_ntoa(dst_.sin_addr), ret); + IF_LEVEL(2) LOG("rpcc::bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret); } return ret; }; // Cancel all outstanding calls - void -rpcc::cancel(void) -{ +void rpcc::cancel(void) { lock ml(m_); LOG("rpcc::cancel: force callers to fail"); for(auto &p : calls_){ caller *ca = p.second; - jsl_log(JSL_DBG_2, "rpcc::cancel: force caller to fail\n"); + IF_LEVEL(2) LOG("rpcc::cancel: force caller to fail"); { lock cl(ca->m); ca->done = true; @@ -153,10 +143,7 @@ rpcc::cancel(void) LOG("rpcc::cancel: done"); } -int -rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, - TO to) -{ +int rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, TO to) { caller ca(0, &rep); int xid_rep; @@ -165,7 +152,7 @@ rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, if((proc != rpc_const::bind && !bind_done_) || (proc == rpc_const::bind && bind_done_)){ - jsl_log(JSL_DBG_1, "rpcc::call1 rpcc has not been bound to dst or binding twice\n"); + IF_LEVEL(1) LOG("rpcc::call1 rpcc has not been bound to dst or binding twice"); return rpc_const::bind_failure; } @@ -206,10 +193,9 @@ rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, ch->send((char *)forgot.buf.c_str(), forgot.buf.size()); ch->send(req.cstr(), req.size()); } - else jsl_log(JSL_DBG_1, "not reachable\n"); - jsl_log(JSL_DBG_2, - "rpcc::call1 %u just sent req proc %x xid %d clt_nonce %d\n", - clt_nonce_, proc, ca.xid, clt_nonce_); + else IF_LEVEL(1) LOG("not reachable"); + IF_LEVEL(2) LOG("rpcc::call1 " << 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 } @@ -226,14 +212,14 @@ rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, { lock cal(ca.m); while (!ca.done){ - jsl_log(JSL_DBG_2, "rpcc:call1: wait\n"); + IF_LEVEL(2) LOG("rpcc:call1: wait"); if(ca.c.wait_until(cal, nextdeadline) == cv_status::timeout){ - jsl_log(JSL_DBG_2, "rpcc::call1: timeout\n"); + IF_LEVEL(2) LOG("rpcc::call1: timeout"); break; } } if(ca.done){ - jsl_log(JSL_DBG_2, "rpcc::call1: reply received\n"); + IF_LEVEL(2) LOG("rpcc::call1: reply received"); break; } } @@ -273,10 +259,9 @@ rpcc::call1(proc_t proc, marshall &req, unmarshall &rep, lock cal(ca.m); - jsl_log(JSL_DBG_2, - "rpcc::call1 %u call done for req proc %x xid %d %s:%d done? %d ret %d \n", - clt_nonce_, proc, ca.xid, inet_ntoa(dst_.sin_addr), - ntohs(dst_.sin_port), ca.done, ca.intret); + IF_LEVEL(2) LOG("rpcc::call1 " << 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); if(ch) ch->decref(); @@ -316,7 +301,7 @@ rpcc::got_pdu(connection *, char *b, size_t sz) rep.unpack_reply_header(&h); if(!rep.ok()){ - jsl_log(JSL_DBG_1, "rpcc:got_pdu unmarshall header failed!!!\n"); + IF_LEVEL(1) LOG("rpcc:got_pdu unmarshall header failed!!!"); return true; } @@ -325,7 +310,7 @@ rpcc::got_pdu(connection *, char *b, size_t sz) update_xid_rep(h.xid); if(calls_.find(h.xid) == calls_.end()){ - jsl_log(JSL_DBG_2, "rpcc::got_pdu xid %d no pending request\n", h.xid); + IF_LEVEL(2) LOG("rpcc::got_pdu xid " << h.xid << " no pending request"); return true; } caller *ca = calls_[h.xid]; @@ -335,8 +320,7 @@ rpcc::got_pdu(connection *, char *b, size_t sz) ca->un->take_in(rep); ca->intret = h.ret; if(ca->intret < 0){ - jsl_log(JSL_DBG_2, "rpcc::got_pdu: RPC reply error for xid %d intret %d\n", - h.xid, ca->intret); + IF_LEVEL(2) LOG("rpcc::got_pdu: RPC reply error for xid " << h.xid << " intret " << ca->intret); } ca->done = 1; } @@ -373,7 +357,7 @@ rpcs::rpcs(unsigned int p1, size_t count) { set_rand_seed(); nonce_ = (unsigned int)random(); - jsl_log(JSL_DBG_2, "rpcs::rpcs created with nonce %d\n", nonce_); + IF_LEVEL(2) LOG("rpcs::rpcs created with nonce " << nonce_); char *loss_env = getenv("RPC_LOSSY"); if(loss_env != NULL){ @@ -398,7 +382,7 @@ bool rpcs::got_pdu(connection *c, char *b, size_t sz) { if(!reachable_){ - jsl_log(JSL_DBG_1, "rpcss::got_pdu: not reachable\n"); + IF_LEVEL(1) LOG("rpcss::got_pdu: not reachable"); return true; } @@ -440,8 +424,8 @@ rpcs::updatestat(proc_t proc) if(clt.second.size() > maxrep) maxrep = clt.second.size(); } - jsl_log(JSL_DBG_1, "REPLY WINDOW: clients %d total reply %lu max per client %lu\n", - (int) reply_window_.size()-1, totalrep, maxrep); + IF_LEVEL(1) LOG("REPLY WINDOW: clients " << (reply_window_.size()-1) << " total reply " << + totalrep << " max per client " << maxrep); curr_counts_ = counting_; } } @@ -458,23 +442,21 @@ rpcs::dispatch(djob_t *j) proc_t proc = h.proc; if(!req.ok()){ - jsl_log(JSL_DBG_1, "rpcs:dispatch unmarshall header failed!!!\n"); + IF_LEVEL(1) LOG("rpcs:dispatch unmarshall header failed!!!"); c->decref(); return; } - jsl_log(JSL_DBG_2, - "rpcs::dispatch: rpc %d (proc %x, last_rep %d) from clt %u for srv instance %u \n", - h.xid, proc, h.xid_rep, h.clt_nonce, h.srv_nonce); + IF_LEVEL(2) LOG("rpcs::dispatch: 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); // is client sending to an old instance of server? if(h.srv_nonce != 0 && h.srv_nonce != nonce_){ - jsl_log(JSL_DBG_2, - "rpcs::dispatch: rpc for an old server instance %u (current %u) proc %x\n", - h.srv_nonce, nonce_, h.proc); + IF_LEVEL(2) LOG("rpcs::dispatch: 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()); @@ -486,10 +468,9 @@ rpcs::dispatch(djob_t *j) { lock pl(procs_m_); if(procs_.count(proc) < 1){ - fprintf(stderr, "rpcs::dispatch: unknown proc %x.\n", - proc); + cerr << "rpcs::dispatch: unknown proc " << hex << proc << "." << endl; c->decref(); - VERIFY(0); + VERIFY(0); return; } @@ -508,9 +489,8 @@ 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 - jsl_log(JSL_DBG_2, - "rpcs::dispatch: new client %u xid %d chan %d, total clients %d\n", - h.clt_nonce, h.xid, c->channo(), (int)reply_window_.size()-1); + IF_LEVEL(2) LOG("rpcs::dispatch: new client " << h.clt_nonce << " xid " << h.xid << + " chan " << c->channo() << ", total clients " << (reply_window_.size()-1)); } } @@ -542,10 +522,9 @@ rpcs::dispatch(djob_t *j) rh.ret = (*f)(req, rep); if (rh.ret == rpc_const::unmarshal_args_failure) { - fprintf(stderr, "rpcs::dispatch: failed to" - " unmarshall the arguments. You are" - " probably calling RPC 0x%x with wrong" - " types of arguments.\n", proc); + cerr << "rpcs::dispatch: 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); @@ -553,9 +532,8 @@ rpcs::dispatch(djob_t *j) rep.pack_reply_header(rh); rep.take_buf(&b1,&sz1); - jsl_log(JSL_DBG_2, - "rpcs::dispatch: sending and saving reply of size %lu for rpc %d, proc %x ret %d, clt %u\n", - sz1, h.xid, proc, rh.ret, h.clt_nonce); + IF_LEVEL(2) LOG("rpcs::dispatch: sending and saving reply of size " << sz1 << " for rpc " << + h.xid << ", proc " << hex << proc << " ret " << dec << rh.ret << ", clt " << h.clt_nonce); if(h.clt_nonce > 0){ // only record replies for clients that require at-most-once logic @@ -584,8 +562,7 @@ rpcs::dispatch(djob_t *j) c->send(b1, sz1); break; case FORGOTTEN: // very old request and we don't have the response anymore - jsl_log(JSL_DBG_2, "rpcs::dispatch: very old request %d from %u\n", - h.xid, h.clt_nonce); + IF_LEVEL(2) LOG("rpcs::dispatch: 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()); @@ -674,7 +651,7 @@ rpcs::add_reply(unsigned int clt_nonce, int xid, 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) { - fprintf(stderr, "Could not find reply struct in add_reply"); + cerr << "Could not find reply struct in add_reply" << endl; l.insert(it, reply_t(xid, b, sz)); } else { *it = reply_t(xid, b, sz); @@ -694,7 +671,7 @@ void rpcs::free_reply_window(void) { } int rpcs::rpcbind(unsigned int &r, int) { - jsl_log(JSL_DBG_2, "rpcs::rpcbind called return nonce %u\n", nonce_); + IF_LEVEL(2) LOG("rpcs::rpcbind called return nonce " << nonce_); r = nonce_; return 0; } @@ -838,7 +815,7 @@ sockaddr_in make_sockaddr(const string &host, const string &port) { struct hostent *hp = gethostbyname(host.c_str()); if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) { - fprintf(stderr, "cannot find host name %s\n", host.c_str()); + cerr << "cannot find host name " << host << endl; exit(1); } memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));