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);
}
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;
}
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;
while (calls_.size () > 0)
destroy_wait_c_.wait(ml);
- LOG("done");
+ LOG << "done";
}
}
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;
}
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 " << hex << proc
+ << " xid " << dec << ca.xid << " clt_nonce " << clt_nonce_;
}
transmit = false; // only send once on a given channel
}
{
lock cal(ca.m);
while (!ca.done) {
- IF_LEVEL(2) LOG("wait");
+ IF_LEVEL(2) LOG << "wait";
if (ca.c.wait_until(cal, nextdeadline) == cv_status::timeout) {
- IF_LEVEL(2) LOG("timeout");
+ IF_LEVEL(2) LOG << "timeout";
break;
}
}
if (ca.done) {
- IF_LEVEL(2) LOG("reply received");
+ IF_LEVEL(2) LOG << "reply received";
break;
}
}
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 " << hex << proc
+ << " xid " << 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);
rep.unpack_header(h);
if (!rep.ok()) {
- IF_LEVEL(1) LOG("unmarshall header failed!!!");
+ IF_LEVEL(1) LOG << "unmarshall header failed!!!";
return true;
}
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];
*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;
}
{
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);
}
bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
if (!reachable_) {
- IF_LEVEL(1) LOG("not reachable");
+ IF_LEVEL(1) LOG << "not reachable";
return true;
}
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 " << hex << proc << ", last_rep "
+ << 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 " << hex << h.proc;
rh.ret = rpc_protocol::oldsrv_failure;
rep.pack_header(rh);
c->send(rep);
{
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" << hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;
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
- 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);
}
}
case NEW: // new request
rh.ret = (*f)(forward<unmarshall>(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" << hex << proc << " with the wrong "
+ << "types of arguments.";
VERIFY(0);
}
VERIFY(rh.ret >= 0);
rep.pack_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 " << hex << proc << " ret " << dec << rh.ret << ", clt " << h.clt_nonce;
add_reply(h.clt_nonce, h.xid, b1);
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);
c->send(rep);
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);
}
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;
}
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));