x exited worker threads).
*/
+#include "types.h"
#include "rpc.h"
#include <sys/types.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <unistd.h>
-#include "lock.h"
-
-#include "jsl_log.h"
-#include "tprintf.h"
-#include "lang/verify.h"
const rpcc::TO rpcc::to_max = { 120000 };
const rpcc::TO rpcc::to_min = { 1000 };
-rpcc::caller::caller(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::nanoseconds>(std::chrono::steady_clock::now());
+inline void set_rand_seed() {
+ auto now = time_point_cast<nanoseconds>(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){
// 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();
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){
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_);
- tprintf("rpcc::cancel: force callers to fail");
+ 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;
destroy_wait_ = true;
destroy_wait_c_.wait(ml);
}
- tprintf("rpcc::cancel: done");
+ 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;
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;
}
ca.xid = xid_++;
calls_[ca.xid] = &ca;
- req.pack_req_header({ca.xid, (int)proc, clt_nonce_, srv_nonce_, 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<std::chrono::steady_clock> 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;
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
}
- if(finaldeadline == std::chrono::time_point<std::chrono::steady_clock>::min())
+ if(finaldeadline == time_point<steady_clock>::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<std::chrono::steady_clock>::min();
+ finaldeadline = time_point<steady_clock>::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;
}
}
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();
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;
}
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];
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;
}
{
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){
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;
}
void
-rpcs::reg1(unsigned int proc, handler *h)
+rpcs::reg1(proc_t proc, handler *h)
{
lock pl(procs_m_);
VERIFY(procs_.count(proc) == 0);
}
void
-rpcs::updatestat(unsigned int proc)
+rpcs::updatestat(proc_t proc)
{
lock cl(count_m_);
counts_[proc]++;
curr_counts_--;
if(curr_counts_ == 0){
- tprintf("RPC STATS: ");
+ LOG("RPC STATS: ");
for (auto i = counts_.begin(); i != counts_.end(); i++)
- tprintf("%x:%lu ", i->first, i->second);
+ LOG(hex << i->first << ":" << dec << i->second);
lock rwl(reply_window_m_);
- std::map<unsigned int,std::list<reply_t> >::iterator clt;
size_t 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();
+ 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 %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_;
}
}
request_header h;
req.unpack_req_header(&h);
- unsigned int proc = (unsigned int)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 %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());
{
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;
}
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));
}
}
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);
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
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());
{
lock rwl(reply_window_m_);
- std::list<reply_t> &l = reply_window_[clt_nonce];
+ list<reply_t> &l = reply_window_[clt_nonce];
VERIFY(l.size() > 0);
VERIFY(xid >= xid_rep);
int past_xid_rep = l.begin()->xid;
- std::list<reply_t>::iterator start = l.begin(), it;
- it = ++start;
+ list<reply_t>::iterator start = l.begin(), it = ++start;
if (past_xid_rep < xid_rep || past_xid_rep == -1) {
// scan for deletion candidates
{
lock rwl(reply_window_m_);
// remember the RPC reply value
- std::list<reply_t> &l = reply_window_[clt_nonce];
- std::list<reply_t>::iterator it = l.begin();
+ list<reply_t> &l = reply_window_[clt_nonce];
+ list<reply_t>::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)
-{
+void rpcs::free_reply_window(void) {
lock rwl(reply_window_m_);
- for (auto clt = reply_window_.begin(); clt != reply_window_.end(); clt++){
- for (auto 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(unsigned int &r, int)
-{
- 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;
}
marshall & operator<<(marshall &m, uint64_t x) { return m << (uint32_t)(x>>32) << (uint32_t)x; }
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;
}
void
-unmarshall::rawbytes(std::string &ss, size_t n)
+unmarshall::rawbytes(string &ss, size_t n)
{
VERIFY(ensure(n));
ss.assign(buf_+index_, n);
unmarshall & operator>>(unmarshall &u, size_t &x) { uint32_t xx; u.rawbytes<uint32_t>(xx); x = xx; return u; }
unmarshall & operator>>(unmarshall &u, uint64_t &x) { u.rawbytes<uint64_t>(x); return u; }
unmarshall & operator>>(unmarshall &u, int64_t &x) { u.rawbytes<int64_t>(x); return u; }
-unmarshall & operator>>(unmarshall &u, std::string &s) {
+unmarshall & operator>>(unmarshall &u, string &s) {
unsigned sz = u.grab<unsigned>();
if(u.ok())
u.rawbytes(s, sz);
}
/*---------------auxilary function--------------*/
-void
-make_sockaddr(const std::string &hostandport, struct sockaddr_in *dst) {
+sockaddr_in make_sockaddr(const string &hostandport) {
auto colon = hostandport.find(':');
- if (colon == std::string::npos)
- make_sockaddr("127.0.0.1", hostandport, dst);
+ if (colon == string::npos)
+ return make_sockaddr("127.0.0.1", hostandport);
else
- make_sockaddr(hostandport.substr(0, colon), hostandport.substr(colon+1), dst);
+ return make_sockaddr(hostandport.substr(0, colon), hostandport.substr(colon+1));
}
-void
-make_sockaddr(const std::string &host, const std::string &port, struct sockaddr_in *dst) {
- bzero(dst, sizeof(*dst));
- dst->sin_family = AF_INET;
+sockaddr_in make_sockaddr(const string &host, const string &port) {
+ sockaddr_in dst;
+ bzero(&dst, sizeof(dst));
+ dst.sin_family = AF_INET;
struct in_addr a{inet_addr(host.c_str())};
if(a.s_addr != INADDR_NONE)
- dst->sin_addr.s_addr = a.s_addr;
+ dst.sin_addr.s_addr = a.s_addr;
else {
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));
- dst->sin_addr.s_addr = a.s_addr;
+ dst.sin_addr.s_addr = a.s_addr;
}
- dst->sin_port = hton((uint16_t)std::stoi(port));
+ dst.sin_port = hton((uint16_t)stoi(port));
+ return dst;
}
projects / invirt/third/libt4.git / blobdiff