projects
/
invirt/third/libt4.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Global destructor clean-ups and python test fixes
[invirt/third/libt4.git]
/
rpc
/
rpc.cc
diff --git
a/rpc/rpc.cc
b/rpc/rpc.cc
index
a451c9f
..
de33675
100644
(file)
--- a/
rpc/rpc.cc
+++ b/
rpc/rpc.cc
@@
-60,6
+60,9
@@
#include <unistd.h>
#include <string.h>
#include <unistd.h>
#include <string.h>
+using std::list;
+using namespace std::chrono;
+
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());
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());
@@
-89,7
+92,7
@@
rpcc::~rpcc() {
}
int rpcc::bind(milliseconds to) {
}
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_);
rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r);
if (ret == 0) {
lock ml(m_);
@@
-143,7
+146,7
@@
int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
ca.xid = xid_++;
calls_[ca.xid] = &ca;
ca.xid = xid_++;
calls_[ca.xid] = &ca;
- req.pack_header(rpc_protocol::request_header{
+ req.write_header(rpc_protocol::request_header{
ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
});
xid_rep = xid_rep_window_.front();
ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
});
xid_rep = xid_rep_window_.front();
@@
-173,20
+176,20
@@
int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
ch->send(req);
}
else IF_LEVEL(1) LOG << "not reachable";
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_;
+ IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc
+ << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_;
}
transmit = false; // only send once on a given channel
}
}
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";
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 (ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout) {
IF_LEVEL(2) LOG << "timeout";
break;
}
IF_LEVEL(2) LOG << "timeout";
break;
}
@@
-231,8
+234,8
@@
int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
lock cal(ca.m);
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) << ":"
+ IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc
+ << " 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
<< ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret;
// destruction of req automatically frees its buffer
@@
-258,7
+261,7
@@
rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
{
unmarshall rep(b, true);
rpc_protocol::reply_header h;
{
unmarshall rep(b, true);
rpc_protocol::reply_header h;
- rep.unpack_header(h);
+ rep.read_header(h);
if (!rep.ok()) {
IF_LEVEL(1) LOG << "unmarshall header failed!!!";
if (!rep.ok()) {
IF_LEVEL(1) LOG << "unmarshall header failed!!!";
@@
-342,7
+345,7
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
unmarshall req(buf, true);
rpc_protocol::request_header h;
unmarshall req(buf, true);
rpc_protocol::request_header h;
- req.unpack_header(h);
+ req.read_header(h);
proc_id_t proc = h.proc;
if (!req.ok()) {
proc_id_t proc = h.proc;
if (!req.ok()) {
@@
-350,8
+353,8
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
return;
}
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 " << 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};
marshall rep;
rpc_protocol::reply_header rh{h.xid,0};
@@
-359,9
+362,9
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
// 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
// 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;
+ << " (current " << nonce_ << ") proc " << std::hex << h.proc;
rh.ret = rpc_protocol::oldsrv_failure;
rh.ret = rpc_protocol::oldsrv_failure;
- rep.pack_header(rh);
+ rep.write_header(rh);
c->send(rep);
return;
}
c->send(rep);
return;
}
@@
-371,7
+374,7
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
{
lock pl(procs_m_);
if (procs_.count(proc) < 1) {
{
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;
}
VERIFY(0);
return;
}
@@
-404,20
+407,21
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
case NEW: // new request
switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
case NEW: // new request
- rh.ret = (*f)(forward<unmarshall>(req), 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 "
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 "
+ << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
<< "types of arguments.";
VERIFY(0);
}
VERIFY(rh.ret >= 0);
<< "types of arguments.";
VERIFY(0);
}
VERIFY(rh.ret >= 0);
- rep.pack_header(rh);
+ rep.write_header(rh);
b1 = rep;
IF_LEVEL(2) LOG << "sending and saving reply of size " << b1.size() << " for rpc "
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;
+ << 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, b1);
@@
-438,7
+442,7
@@
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
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;
rh.ret = rpc_protocol::atmostonce_failure;
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;
rh.ret = rpc_protocol::atmostonce_failure;
- rep.pack_header(rh);
+ rep.write_header(rh);
c->send(rep);
break;
}
c->send(rep);
break;
}
@@
-554,6
+558,6
@@
static sockaddr_in make_sockaddr(const string & hostandport) {
memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));
dst.sin_addr.s_addr = a.s_addr;
}
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;
}
return dst;
}