X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/4b9798f44ae94deabf87dd534337b55259272950..24bebc0ecf83446c7371eff69042322aab34976a:/rpc/rpc.cc diff --git a/rpc/rpc.cc b/rpc/rpc.cc index a3ffef1..9f1d90c 100644 --- a/rpc/rpc.cc +++ b/rpc/rpc.cc @@ -54,6 +54,7 @@ x exited worker threads). */ +#include "types.h" #include "rpc.h" #include @@ -61,37 +62,22 @@ #include #include #include -#include "lock.h" - -#include "jsl_log.h" -#include "lang/verify.h" const rpcc::TO rpcc::to_max = { 120000 }; const rpcc::TO rpcc::to_min = { 1000 }; -rpcc::caller::caller(unsigned int xxid, unmarshall *xun) -: xid(xxid), un(xun), done(false) -{ -} - -rpcc::caller::~caller() -{ -} - -inline -void set_rand_seed() -{ - auto now = std::chrono::time_point_cast(std::chrono::steady_clock::now()); - srandom((int)now.time_since_epoch().count()^((int)getpid())); +inline void set_rand_seed() { + auto now = time_point_cast(steady_clock::now()); + srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid()); } -rpcc::rpcc(sockaddr_in d, bool retrans) : - dst_(d), srv_nonce_(0), bind_done_(false), xid_(1), lossytest_(0), +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) { if(retrans){ set_rand_seed(); - clt_nonce_ = random(); + clt_nonce_ = (unsigned int)random(); } else { // special client nonce 0 means this client does not // require at-most-once logic from the server @@ -107,16 +93,13 @@ rpcc::rpcc(sockaddr_in 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(); @@ -124,33 +107,27 @@ rpcc::~rpcc() VERIFY(calls_.size() == 0); } -int -rpcc::bind(TO to) -{ - int r; +int rpcc::bind(TO to) { + unsigned int r; int ret = call_timeout(rpc_const::bind, to, r, 0); if(ret == 0){ lock ml(m_); 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_); - printf("rpcc::cancel: force callers to fail\n"); - std::map::iterator iter; - for(iter = calls_.begin(); iter != calls_.end(); iter++){ - caller *ca = iter->second; + 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; @@ -163,22 +140,19 @@ rpcc::cancel(void) destroy_wait_ = true; destroy_wait_c_.wait(ml); } - printf("rpcc::cancel: done\n"); + LOG("rpcc::cancel: done"); } -int -rpcc::call1(unsigned int 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; + int xid_rep; { lock ml(m_); 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; } @@ -189,17 +163,13 @@ rpcc::call1(unsigned int proc, marshall &req, unmarshall &rep, ca.xid = xid_++; calls_[ca.xid] = &ca; - req_header h(ca.xid, proc, clt_nonce_, srv_nonce_, - xid_rep_window_.front()); - req.pack_req_header(h); - xid_rep = xid_rep_window_.front(); + req.pack_req_header({ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()}); + xid_rep = xid_rep_window_.front(); } TO curr_to; - std::chrono::time_point finaldeadline = - std::chrono::steady_clock::now() + - std::chrono::milliseconds(to.to), - nextdeadline; + auto finaldeadline = steady_clock::now() + milliseconds(to.to), + nextdeadline = finaldeadline; curr_to.to = to_min.to; @@ -223,34 +193,33 @@ rpcc::call1(unsigned int 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 %u 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 } - if(finaldeadline == std::chrono::time_point::min()) + if(finaldeadline == time_point::min()) break; - nextdeadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(curr_to.to); + nextdeadline = steady_clock::now() + milliseconds(curr_to.to); if(nextdeadline > finaldeadline) { nextdeadline = finaldeadline; - finaldeadline = std::chrono::time_point::min(); + finaldeadline = time_point::min(); } { lock cal(ca.m); while (!ca.done){ - jsl_log(JSL_DBG_2, "rpcc:call1: wait\n"); - if(ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout){ - jsl_log(JSL_DBG_2, "rpcc::call1: timeout\n"); + IF_LEVEL(2) LOG("rpcc:call1: wait"); + if(ca.c.wait_until(cal, nextdeadline) == cv_status::timeout){ + 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; } } @@ -290,10 +259,9 @@ rpcc::call1(unsigned int proc, marshall &req, unmarshall &rep, lock cal(ca.m); - jsl_log(JSL_DBG_2, - "rpcc::call1 %u call done for req proc %x xid %u %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(); @@ -326,14 +294,14 @@ rpcc::get_refconn(connection **ch) // // this function keeps no reference for connection *c bool -rpcc::got_pdu(connection *c, char *b, int sz) +rpcc::got_pdu(connection *, char *b, size_t sz) { unmarshall rep(b, sz); reply_header h; 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; } @@ -342,7 +310,7 @@ rpcc::got_pdu(connection *c, char *b, int 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]; @@ -352,8 +320,7 @@ rpcc::got_pdu(connection *c, char *b, int 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; } @@ -363,15 +330,13 @@ rpcc::got_pdu(connection *c, char *b, int sz) // assumes thread holds mutex m void -rpcc::update_xid_rep(unsigned int xid) +rpcc::update_xid_rep(int xid) { - std::list::iterator it; - if(xid <= xid_rep_window_.front()){ return; } - for (it = xid_rep_window_.begin(); it != xid_rep_window_.end(); it++){ + for (auto it = xid_rep_window_.begin(); it != xid_rep_window_.end(); it++){ if(*it > xid){ xid_rep_window_.insert(it, xid); goto compress; @@ -380,19 +345,19 @@ rpcc::update_xid_rep(unsigned int xid) xid_rep_window_.push_back(xid); compress: - it = xid_rep_window_.begin(); + auto it = xid_rep_window_.begin(); for (it++; it != xid_rep_window_.end(); it++){ while (xid_rep_window_.front() + 1 == *it) xid_rep_window_.pop_front(); } } -rpcs::rpcs(unsigned int p1, int count) +rpcs::rpcs(unsigned int p1, size_t count) : port_(p1), counting_(count), curr_counts_(count), lossytest_(0), reachable_ (true) { set_rand_seed(); - nonce_ = random(); - jsl_log(JSL_DBG_2, "rpcs::rpcs created with nonce %d\n", nonce_); + nonce_ = (unsigned int)random(); + IF_LEVEL(2) LOG("rpcs::rpcs created with nonce " << nonce_); char *loss_env = getenv("RPC_LOSSY"); if(loss_env != NULL){ @@ -414,16 +379,16 @@ rpcs::~rpcs() } bool -rpcs::got_pdu(connection *c, char *b, int sz) +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; } djob_t *j = new djob_t(c, b, sz); c->incref(); - bool succ = dispatchpool_->addJob(std::bind(&rpcs::dispatch, this, j)); + bool succ = dispatchpool_->addJob(bind(&rpcs::dispatch, this, j)); if(!succ || !reachable_){ c->decref(); delete j; @@ -432,7 +397,7 @@ rpcs::got_pdu(connection *c, char *b, int sz) } void -rpcs::reg1(unsigned int proc, handler *h) +rpcs::reg1(proc_t proc, handler *h) { lock pl(procs_m_); VERIFY(procs_.count(proc) == 0); @@ -441,30 +406,26 @@ rpcs::reg1(unsigned int proc, handler *h) } void -rpcs::updatestat(unsigned int proc) +rpcs::updatestat(proc_t proc) { lock cl(count_m_); counts_[proc]++; curr_counts_--; if(curr_counts_ == 0){ - std::map::iterator i; - printf("RPC STATS: "); - for (i = counts_.begin(); i != counts_.end(); i++){ - printf("%x:%d ", i->first, i->second); - } - printf("\n"); + LOG("RPC STATS: "); + for (auto i = counts_.begin(); i != counts_.end(); i++) + LOG(hex << i->first << ":" << dec << i->second); lock rwl(reply_window_m_); - std::map >::iterator clt; - unsigned int totalrep = 0, maxrep = 0; - for (clt = reply_window_.begin(); clt != reply_window_.end(); clt++){ - totalrep += clt->second.size(); - if(clt->second.size() > maxrep) - maxrep = clt->second.size(); + size_t totalrep = 0, maxrep = 0; + for (auto clt : reply_window_) { + totalrep += clt.second.size(); + if(clt.second.size() > maxrep) + maxrep = clt.second.size(); } - jsl_log(JSL_DBG_1, "REPLY WINDOW: clients %d total reply %d max per client %d\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_; } } @@ -476,28 +437,26 @@ rpcs::dispatch(djob_t *j) unmarshall req(j->buf, j->sz); delete j; - req_header h; + request_header h; req.unpack_req_header(&h); - int proc = h.proc; + 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 %u (proc %x, last_rep %u) 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()); @@ -509,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; } @@ -520,8 +478,8 @@ rpcs::dispatch(djob_t *j) } rpcs::rpcstate_t stat; - char *b1; - int sz1; + char *b1 = nullptr; + size_t sz1 = 0; if(h.clt_nonce){ // have i seen this client before? @@ -531,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)); } } @@ -565,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); @@ -576,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 %d for rpc %u, 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 @@ -607,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 %u 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()); @@ -632,22 +586,21 @@ rpcs::dispatch(djob_t *j) // DONE: seen this xid, previous reply returned in *b and *sz. // FORGOTTEN: might have seen this xid, but deleted previous reply. rpcs::rpcstate_t -rpcs::checkduplicate_and_update(unsigned int clt_nonce, unsigned int xid, - unsigned int xid_rep, char **b, int *sz) +rpcs::checkduplicate_and_update(unsigned int clt_nonce, int xid, + int xid_rep, char **b, size_t *sz) { lock rwl(reply_window_m_); - std::list &l = reply_window_[clt_nonce]; + list &l = reply_window_[clt_nonce]; VERIFY(l.size() > 0); VERIFY(xid >= xid_rep); - unsigned int past_xid_rep = l.begin()->xid; + int past_xid_rep = l.begin()->xid; - std::list::iterator start = l.begin(), it; - it = ++start; + list::iterator start = l.begin(), it = ++start; - if (past_xid_rep < xid_rep || past_xid_rep == (unsigned int)-1) { + 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) @@ -657,7 +610,7 @@ rpcs::checkduplicate_and_update(unsigned int clt_nonce, unsigned int xid, l.begin()->xid = xid_rep; } - if (xid < past_xid_rep && past_xid_rep != (unsigned int)-1) + if (xid < past_xid_rep && past_xid_rep != -1) return FORGOTTEN; // skip non-deletion candidates @@ -687,287 +640,153 @@ rpcs::checkduplicate_and_update(unsigned int clt_nonce, unsigned int xid, // free_reply_window() and checkduplicate_and_update is responsible for // calling free(b). void -rpcs::add_reply(unsigned int clt_nonce, unsigned int xid, - char *b, int sz) +rpcs::add_reply(unsigned int clt_nonce, int xid, + char *b, size_t sz) { lock rwl(reply_window_m_); // remember the RPC reply value - std::list &l = reply_window_[clt_nonce]; - std::list::iterator it = l.begin(); + 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) { - 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); } } -void -rpcs::free_reply_window(void) -{ - std::map >::iterator clt; - std::list::iterator it; - +void rpcs::free_reply_window(void) { lock rwl(reply_window_m_); - for (clt = reply_window_.begin(); clt != reply_window_.end(); clt++){ - for (it = clt->second.begin(); it != clt->second.end(); it++){ - if (it->cb_present) - free(it->buf); + for (auto clt : reply_window_) { + for (auto it : clt.second){ + if (it.cb_present) + free(it.buf); } - clt->second.clear(); + clt.second.clear(); } reply_window_.clear(); } -// rpc handler -int -rpcs::rpcbind(int &r, int a) -{ - jsl_log(JSL_DBG_2, "rpcs::rpcbind called return nonce %u\n", nonce_); +int rpcs::rpcbind(unsigned int &r, int) { + IF_LEVEL(2) LOG("rpcs::rpcbind called return nonce " << nonce_); r = nonce_; return 0; } -void -marshall::rawbyte(unsigned char x) -{ - if(_ind >= _capa){ - _capa *= 2; - VERIFY (_buf != NULL); - _buf = (char *)realloc(_buf, _capa); - VERIFY(_buf); - } - _buf[_ind++] = x; -} - -void -marshall::rawbytes(const char *p, int n) -{ - if((_ind+n) > _capa){ - _capa = _capa > n? 2*_capa:(_capa+n); - VERIFY (_buf != NULL); - _buf = (char *)realloc(_buf, _capa); - VERIFY(_buf); - } - memcpy(_buf+_ind, p, n); - _ind += n; -} - marshall & -operator<<(marshall &m, bool x) -{ +operator<<(marshall &m, uint8_t x) { m.rawbyte(x); return m; } marshall & -operator<<(marshall &m, unsigned char x) -{ - m.rawbyte(x); +operator<<(marshall &m, uint16_t x) { + x = hton(x); + m.rawbytes((char *)&x, 2); return m; } marshall & -operator<<(marshall &m, char x) -{ - m << (unsigned char) x; +operator<<(marshall &m, uint32_t x) { + x = hton(x); + m.rawbytes((char *)&x, 4); return m; } +marshall & operator<<(marshall &m, int32_t x) { return m << (uint32_t) x; } +marshall & operator<<(marshall &m, int8_t x) { return m << (uint8_t)x; } +marshall & operator<<(marshall &m, bool x) { return m << (uint8_t)x; } +marshall & operator<<(marshall &m, int16_t x) { return m << (uint16_t)x; } +marshall & operator<<(marshall &m, uint64_t x) { return m << (uint32_t)(x>>32) << (uint32_t)x; } marshall & -operator<<(marshall &m, unsigned short x) -{ - m.rawbyte((x >> 8) & 0xff); - m.rawbyte(x & 0xff); - return m; -} - -marshall & -operator<<(marshall &m, short x) -{ - m << (unsigned short) x; - return m; -} - -marshall & -operator<<(marshall &m, unsigned int x) -{ - // network order is big-endian - m.rawbyte((x >> 24) & 0xff); - m.rawbyte((x >> 16) & 0xff); - m.rawbyte((x >> 8) & 0xff); - m.rawbyte(x & 0xff); - return m; -} - -marshall & -operator<<(marshall &m, int x) -{ - m << (unsigned int) x; - return m; -} - -marshall & -operator<<(marshall &m, const std::string &s) -{ +operator<<(marshall &m, const string &s) { m << (unsigned int) s.size(); m.rawbytes(s.data(), s.size()); return m; } -marshall & -operator<<(marshall &m, unsigned long long x) -{ - m << (unsigned int) (x >> 32); - m << (unsigned int) x; - return m; +void marshall::pack_req_header(const request_header &h) { + size_t saved_sz = index_; + //leave the first 4-byte empty for channel to fill size of pdu + index_ = sizeof(rpc_sz_t); + *this << h.xid << h.proc << h.clt_nonce << h.srv_nonce << h.xid_rep; + index_ = saved_sz; } -void -marshall::pack(int x) -{ - rawbyte((x >> 24) & 0xff); - rawbyte((x >> 16) & 0xff); - rawbyte((x >> 8) & 0xff); - rawbyte(x & 0xff); -} - -void -unmarshall::unpack(int *x) -{ - (*x) = (rawbyte() & 0xff) << 24; - (*x) |= (rawbyte() & 0xff) << 16; - (*x) |= (rawbyte() & 0xff) << 8; - (*x) |= rawbyte() & 0xff; +void marshall::pack_reply_header(const reply_header &h) { + size_t saved_sz = index_; + //leave the first 4-byte empty for channel to fill size of pdu + index_ = sizeof(rpc_sz_t); + *this << h.xid << h.ret; + index_ = saved_sz; } // take the contents from another unmarshall object void unmarshall::take_in(unmarshall &another) { - if(_buf) - free(_buf); - another.take_buf(&_buf, &_sz); - _ind = RPC_HEADER_SZ; - _ok = _sz >= RPC_HEADER_SZ?true:false; + if(buf_) + free(buf_); + another.take_buf(&buf_, &sz_); + index_ = RPC_HEADER_SZ; + ok_ = sz_ >= RPC_HEADER_SZ?true:false; } -bool -unmarshall::okdone() -{ - if(ok() && _ind == _sz){ - return true; - } else { - return false; - } +inline bool +unmarshall::ensure(size_t n) { + if (index_+n > sz_) + ok_ = false; + return ok_; } -unsigned int +inline uint8_t unmarshall::rawbyte() { - char c = 0; - if(_ind >= _sz) - _ok = false; - else - c = _buf[_ind++]; - return c; -} - -unmarshall & -operator>>(unmarshall &u, bool &x) -{ - x = (bool) u.rawbyte() ; - return u; -} - -unmarshall & -operator>>(unmarshall &u, unsigned char &x) -{ - x = (unsigned char) u.rawbyte() ; - return u; -} - -unmarshall & -operator>>(unmarshall &u, char &x) -{ - x = (char) u.rawbyte(); - return u; -} - - -unmarshall & -operator>>(unmarshall &u, unsigned short &x) -{ - x = (u.rawbyte() & 0xff) << 8; - x |= u.rawbyte() & 0xff; - return u; -} - -unmarshall & -operator>>(unmarshall &u, short &x) -{ - x = (u.rawbyte() & 0xff) << 8; - x |= u.rawbyte() & 0xff; - return u; -} - -unmarshall & -operator>>(unmarshall &u, unsigned int &x) -{ - x = (u.rawbyte() & 0xff) << 24; - x |= (u.rawbyte() & 0xff) << 16; - x |= (u.rawbyte() & 0xff) << 8; - x |= u.rawbyte() & 0xff; - return u; + if (!ensure(1)) + return 0; + return (uint8_t)buf_[index_++]; } -unmarshall & -operator>>(unmarshall &u, int &x) -{ - x = (u.rawbyte() & 0xff) << 24; - x |= (u.rawbyte() & 0xff) << 16; - x |= (u.rawbyte() & 0xff) << 8; - x |= u.rawbyte() & 0xff; - return u; -} - -unmarshall & -operator>>(unmarshall &u, unsigned long long &x) +void +unmarshall::rawbytes(string &ss, size_t n) { - unsigned int h, l; - u >> h; - u >> l; - x = l | ((unsigned long long) h << 32); - return u; + VERIFY(ensure(n)); + ss.assign(buf_+index_, n); + index_ += n; } -unmarshall & -operator>>(unmarshall &u, std::string &s) -{ - unsigned sz; - u >> sz; +template +void +unmarshall::rawbytes(T &t) +{ + const size_t n = sizeof(T); + VERIFY(ensure(n)); + memcpy(&t, buf_+index_, n); + t = ntoh(t); + index_ += n; +} + +unmarshall & operator>>(unmarshall &u, bool &x) { x = (bool)u.rawbyte(); return u; } +unmarshall & operator>>(unmarshall &u, uint8_t &x) { x = u.rawbyte(); return u; } +unmarshall & operator>>(unmarshall &u, int8_t &x) { x = (int8_t)u.rawbyte(); return u; } +unmarshall & operator>>(unmarshall &u, uint16_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, int16_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, uint32_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, int32_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, size_t &x) { uint32_t xx; u.rawbytes(xx); x = xx; return u; } +unmarshall & operator>>(unmarshall &u, uint64_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, int64_t &x) { u.rawbytes(x); return u; } +unmarshall & operator>>(unmarshall &u, string &s) { + unsigned sz = u.grab(); if(u.ok()) u.rawbytes(s, sz); return u; } -void -unmarshall::rawbytes(std::string &ss, unsigned int n) -{ - if((_ind+n) > (unsigned)_sz){ - _ok = false; - } else { - std::string tmps = std::string(_buf+_ind, n); - swap(ss, tmps); - VERIFY(ss.size() == n); - _ind += n; - } -} - 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) && @@ -975,43 +794,33 @@ bool operator<(const sockaddr_in &a, const sockaddr_in &b){ } /*---------------auxilary function--------------*/ -void -make_sockaddr(const char *hostandport, struct sockaddr_in *dst){ - - char host[200]; - const char *localhost = "127.0.0.1"; - const char *port = index(hostandport, ':'); - if(port == NULL){ - memcpy(host, localhost, strlen(localhost)+1); - port = hostandport; - } else { - memcpy(host, hostandport, port-hostandport); - host[port-hostandport] = '\0'; - port++; - } - - make_sockaddr(host, port, dst); - +sockaddr_in make_sockaddr(const string &hostandport) { + auto colon = hostandport.find(':'); + if (colon == string::npos) + return make_sockaddr("127.0.0.1", hostandport); + else + return make_sockaddr(hostandport.substr(0, colon), hostandport.substr(colon+1)); } -void -make_sockaddr(const char *host, const char *port, struct sockaddr_in *dst){ +sockaddr_in make_sockaddr(const string &host, const string &port) { + sockaddr_in dst; + bzero(&dst, sizeof(dst)); + dst.sin_family = AF_INET; - in_addr_t a; + struct in_addr a{inet_addr(host.c_str())}; - bzero(dst, sizeof(*dst)); - dst->sin_family = AF_INET; + if(a.s_addr != INADDR_NONE) + dst.sin_addr.s_addr = a.s_addr; + else { + struct hostent *hp = gethostbyname(host.c_str()); - a = inet_addr(host); - if(a != INADDR_NONE){ - dst->sin_addr.s_addr = a; - } else { - struct hostent *hp = gethostbyname(host); - if(hp == 0 || hp->h_length != 4){ - fprintf(stderr, "cannot find host name %s\n", host); + if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) { + cerr << "cannot find host name " << host << endl; exit(1); } - dst->sin_addr.s_addr = ((struct in_addr *)(hp->h_addr))->s_addr; + memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t)); + dst.sin_addr.s_addr = a.s_addr; } - dst->sin_port = htons(atoi(port)); + dst.sin_port = hton((uint16_t)stoi(port)); + return dst; }