#include "lock.h"
#include "jsl_log.h"
+#include "threaded_log.h"
#include "lang/verify.h"
+using std::stoi;
+
const rpcc::TO rpcc::to_max = { 120000 };
const rpcc::TO rpcc::to_min = { 1000 };
-rpcc::caller::caller(unsigned int xxid, unmarshall *xun)
+rpcc::caller::caller(int xxid, unmarshall *xun)
: xid(xxid), un(xun), done(false)
{
}
void set_rand_seed()
{
auto now = std::chrono::time_point_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now());
- srandom((int)now.time_since_epoch().count()^((int)getpid()));
+ 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
int
rpcc::bind(TO to)
{
- int r;
+ unsigned int r;
int ret = call_timeout(rpc_const::bind, to, r, 0);
if(ret == 0){
lock ml(m_);
rpcc::cancel(void)
{
lock ml(m_);
- printf("rpcc::cancel: force callers to fail\n");
- std::map<int,caller*>::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");
{
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,
+rpcc::call1(proc_t proc, marshall &req, unmarshall &rep,
TO to)
{
caller ca(0, &rep);
- int xid_rep;
+ int xid_rep;
{
lock ml(m_);
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;
}
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",
+ "rpcc::call1 %u just sent req proc %x xid %d clt_nonce %d\n",
clt_nonce_, proc, ca.xid, clt_nonce_);
}
transmit = false; // only send once on a given channel
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",
+ "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);
//
// 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;
// assumes thread holds mutex m
void
-rpcc::update_xid_rep(unsigned int xid)
+rpcc::update_xid_rep(int xid)
{
- std::list<unsigned int>::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;
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();
+ nonce_ = (unsigned int)random();
jsl_log(JSL_DBG_2, "rpcs::rpcs created with nonce %d\n", nonce_);
char *loss_env = getenv("RPC_LOSSY");
}
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");
}
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){
- std::map<int, int>::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(std::hex << i->first << ":" << std::dec << i->second);
lock rwl(reply_window_m_);
- std::map<unsigned int,std::list<reply_t> >::iterator clt;
+ map<unsigned int,list<reply_t> >::iterator clt;
- unsigned int totalrep = 0, maxrep = 0;
+ 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();
}
- jsl_log(JSL_DBG_1, "REPLY WINDOW: clients %d total reply %d max per client %d\n",
+ jsl_log(JSL_DBG_1, "REPLY WINDOW: clients %d total reply %lu max per client %lu\n",
(int) reply_window_.size()-1, totalrep, maxrep);
curr_counts_ = counting_;
}
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");
}
jsl_log(JSL_DBG_2,
- "rpcs::dispatch: rpc %u (proc %x, last_rep %u) from clt %u for srv instance %u \n",
+ "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);
marshall rep;
}
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?
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",
+ "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(h.clt_nonce > 0){
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",
+ jsl_log(JSL_DBG_2, "rpcs::dispatch: very old request %d from %u\n",
h.xid, h.clt_nonce);
rh.ret = rpc_const::atmostonce_failure;
rep.pack_reply_header(rh);
// 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<reply_t> &l = reply_window_[clt_nonce];
+ list<reply_t> &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<reply_t>::iterator start = l.begin(), it;
+ list<reply_t>::iterator start = l.begin(), it;
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)
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
// 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<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
void
rpcs::free_reply_window(void)
{
- std::map<unsigned int,std::list<reply_t> >::iterator clt;
- std::list<reply_t>::iterator it;
-
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++){
+ 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);
}
// rpc handler
int
-rpcs::rpcbind(int &r, int a)
+rpcs::rpcbind(unsigned int &r, int)
{
jsl_log(JSL_DBG_2, "rpcs::rpcbind called return nonce %u\n", 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(int x)
-{
- rawbyte((x >> 24) & 0xff);
- rawbyte((x >> 16) & 0xff);
- rawbyte((x >> 8) & 0xff);
- rawbyte(x & 0xff);
+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
-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;
-}
-
-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;
+ if (!ensure(1))
+ return 0;
+ return (uint8_t)buf_[index_++];
}
-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 <class T>
+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<uint16_t>(x); return u; }
+unmarshall & operator>>(unmarshall &u, int16_t &x) { u.rawbytes<int16_t>(x); return u; }
+unmarshall & operator>>(unmarshall &u, uint32_t &x) { u.rawbytes<uint32_t>(x); return u; }
+unmarshall & operator>>(unmarshall &u, int32_t &x) { u.rawbytes<int32_t>(x); return u; }
+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, string &s) {
+ unsigned sz = u.grab<unsigned>();
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) &&
}
/*---------------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) {
+ fprintf(stderr, "cannot find host name %s\n", host.c_str());
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;
}
projects / invirt/third/libt4.git / blobdiff