// x exited worker threads).
//
-#include "rpc.h"
+#include "include/rpc/rpc.h"
#include <arpa/inet.h>
#include <netinet/tcp.h>
#include <unistd.h>
#include <string.h>
-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());
-}
+using std::list;
+using namespace std::chrono;
static sockaddr_in make_sockaddr(const string & hostandport);
rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
{
- set_rand_seed();
- clt_nonce_ = (nonce_t)random();
+ clt_nonce_ = (nonce_t)global->random_generator();
char *loss_env = getenv("RPC_LOSSY");
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));
+ 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;
+ 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> 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");
+ 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;
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, milliseconds to, string & rep, marshall & req) {
+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;
}
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();
}
}
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;
}
}
{
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_)
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_latest_connection(shared_ptr<connection> & 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<connection> &, 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<connection> &, 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;
}
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;
}
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);
}
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;
}
}
void rpcs::dispatch(shared_ptr<connection> 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;
}
{
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];
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);
}
}
conns_[h.clt_nonce] = c;
}
- string b1;
+ string stored_reply;
- switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
+ switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, stored_reply)) {
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.");
- VERIFY(0);
- }
- VERIFY(rh.ret >= 0);
+ {
+ marshall rep;
+ rh.ret = (*f)(std::forward<unmarshall>(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;
- add_reply(h.clt_nonce, h.xid, b1);
+ add_reply(h.clt_nonce, h.xid, stored_reply);
// get the latest connection to the client
{
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;
}
}
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;
}
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())};
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;
}
projects / invirt/third/libt4.git / blobdiff