rpc/rpc.cc

   1 //
   2 // The rpcc class handles client-side RPC.  Each rpcc is bound to a single RPC
   3 // server.  The jobs of rpcc include maintaining a connection to server, sending
   4 // RPC requests and waiting for responses, retransmissions, at-most-once delivery
   5 // etc.
   6 //
   7 // The rpcs class handles the server side of RPC.  Each rpcs handles multiple
   8 // connections from different rpcc objects.  The jobs of rpcs include accepting
   9 // connections, dispatching requests to registered RPC handlers, at-most-once
  10 // delivery etc.
  11 //
  12 // Both rpcc and rpcs use the connection class as an abstraction for the
  13 // underlying communication channel.  To send an RPC request/reply, one calls
  14 // connection::send() which blocks until data is sent or the connection has
  15 // failed (thus the caller can free the buffer when send() returns).  When a
  16 // request/reply is received, connection makes a callback into the corresponding
  17 // rpcc or rpcs (see rpcc::got_pdu() and rpcs::got_pdu()).
  18 //
  19 // Thread organization:
  20 // rpcc uses application threads to send RPC requests and blocks to receive the
  21 // reply or error. All connections use a single PollMgr object to perform async
  22 // socket IO.  PollMgr creates a single thread to examine the readiness of socket
  23 // file descriptors and informs the corresponding connection whenever a socket is
  24 // ready to be read or written.  (We use asynchronous socket IO to reduce the
  25 // number of threads needed to manage these connections; without async IO, at
  26 // least one thread is needed per connection to read data without blocking other
  27 // activities.)  Each rpcs object creates one thread for listening on the server
  28 // port and a pool of threads for executing RPC requests.  The thread pool allows
  29 // us to control the number of threads spawned at the server (spawning one thread
  30 // per request will hurt when the server faces thousands of requests).
  31 //
  32 // In order to delete a connection object, we must maintain a reference count.
  33 // For rpcc, multiple client threads might be invoking the rpcc::call() functions
  34 // and thus holding multiple references to the underlying connection object. For
  35 // rpcs, multiple dispatch threads might be holding references to the same
  36 // connection object.  A connection object is deleted only when the underlying
  37 // connection is dead and the reference count reaches zero.
  38 //
  39 // This version of the RPC library explicitly joins exited threads to make sure
  40 // no outstanding references exist before deleting objects.
  41 //
  42 // To delete a rpcc object safely, the users of the library must ensure that
  43 // there are no outstanding calls on the rpcc object.
  44 //
  45 // To delete a rpcs object safely, we do the following in sequence: 1. stop
  46 // accepting new incoming connections. 2. close existing active connections.  3.
  47 // delete the dispatch thread pool which involves waiting for current active RPC
  48 // handlers to finish.  It is interesting how a thread pool can be deleted
  49 // without using thread cancellation. The trick is to inject x "poison pills" for
  50 // a thread pool of x threads. Upon getting a poison pill instead of a normal
  51 // task, a worker thread will exit (and thread pool destructor waits to join all
  52 // x exited worker threads).
  53 //
  54
  55 #include "rpc.h"
  56
  57 #include <arpa/inet.h>
  58 #include <netinet/tcp.h>
  59 #include <netdb.h>
  60 #include <unistd.h>
  61 #include <string.h>
  62
  63 inline void set_rand_seed() {
  64     auto now = time_point_cast<nanoseconds>(steady_clock::now());
  65     srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid());
  66 }
  67
  68 static sockaddr_in make_sockaddr(const string & hostandport);
  69
  70 rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
  71 {
  72     set_rand_seed();
  73     clt_nonce_ = (nonce_t)random();
  74
  75     char *loss_env = getenv("RPC_LOSSY");
  76     if (loss_env)
  77         lossytest_ = atoi(loss_env);
  78
  79     IF_LEVEL(2) LOG("cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_);
  80 }
  81
  82 // IMPORTANT: destruction should happen only when no external threads
  83 // are blocked inside rpcc or will use rpcc in the future
  84 rpcc::~rpcc() {
  85     cancel();
  86     IF_LEVEL(2) LOG("delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1));
  87     chan_.reset();
  88     VERIFY(calls_.size() == 0);
  89 }
  90
  91 int rpcc::bind(milliseconds to) {
  92     nonce_t r;
  93     rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r);
  94     if (ret == 0) {
  95         lock ml(m_);
  96         bind_done_ = true;
  97         srv_nonce_ = r;
  98     } else {
  99         IF_LEVEL(2) LOG("bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret);
 100     }
 101     return ret;
 102 };
 103
 104 // Cancel all outstanding calls
 105 void rpcc::cancel(void) {
 106     lock ml(m_);
 107     if (calls_.size()) {
 108         LOG("force callers to fail");
 109         for (auto & p : calls_) {
 110             caller *ca = p.second;
 111
 112             IF_LEVEL(2) LOG("force caller to fail");
 113
 114             lock cl(ca->m);
 115             ca->done = true;
 116             ca->intret = rpc_protocol::cancel_failure;
 117             ca->c.notify_one();
 118         }
 119
 120         destroy_wait_ = true;
 121         while (calls_.size () > 0)
 122             destroy_wait_c_.wait(ml);
 123
 124         LOG("done");
 125     }
 126 }
 127
 128 int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
 129
 130     caller ca(0, &rep);
 131     xid_t xid_rep;
 132     {
 133         lock ml(m_);
 134
 135         if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) {
 136             IF_LEVEL(1) LOG("rpcc has not been bound to dst or binding twice");
 137             return rpc_protocol::bind_failure;
 138         }
 139
 140         if (destroy_wait_)
 141             return rpc_protocol::cancel_failure;
 142
 143         ca.xid = xid_++;
 144         calls_[ca.xid] = &ca;
 145
 146         req.pack_header(rpc_protocol::request_header{
 147                 ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
 148                 });
 149         xid_rep = xid_rep_window_.front();
 150     }
 151
 152     milliseconds curr_to = rpc::to_min;
 153     auto finaldeadline = steady_clock::now() + to;
 154
 155     bool transmit = true;
 156     shared_ptr<connection> ch;
 157
 158     while (1) {
 159         if (transmit) {
 160             get_latest_connection(ch);
 161             if (ch) {
 162                 if (reachable_) {
 163                     request forgot;
 164                     {
 165                         lock ml(m_);
 166                         if (dup_req_.isvalid() && xid_rep_done_ > dup_req_.xid) {
 167                             forgot = dup_req_;
 168                             dup_req_.clear();
 169                         }
 170                     }
 171                     if (forgot.isvalid())
 172                         ch->send(forgot.buf);
 173                     ch->send(req);
 174                 }
 175                 else IF_LEVEL(1) LOG("not reachable");
 176                 IF_LEVEL(2) LOG(clt_nonce_ << " just sent req proc " << hex << proc <<
 177                                 " xid " << dec << ca.xid << " clt_nonce " << clt_nonce_);
 178             }
 179             transmit = false; // only send once on a given channel
 180         }
 181
 182         auto nextdeadline = min(steady_clock::now() + curr_to, finaldeadline);
 183         curr_to *= 2;
 184
 185         {
 186             lock cal(ca.m);
 187             while (!ca.done) {
 188                 IF_LEVEL(2) LOG("wait");
 189                 if (ca.c.wait_until(cal, nextdeadline) == cv_status::timeout) {
 190                     IF_LEVEL(2) LOG("timeout");
 191                     break;
 192                 }
 193             }
 194             if (ca.done) {
 195                 IF_LEVEL(2) LOG("reply received");
 196                 break;
 197             }
 198         }
 199
 200         if (nextdeadline >= finaldeadline)
 201             break;
 202
 203         // retransmit on new connection if connection is dead
 204         if (!ch || ch->isdead())
 205             transmit = true;
 206     }
 207
 208     {
 209         // no locking of ca.m since only this thread changes ca.xid
 210         lock ml(m_);
 211         calls_.erase(ca.xid);
 212         // may need to update the xid again here, in case the
 213         // packet times out before it's even sent by the channel.
 214         // I don't think there's any harm in maybe doing it twice
 215         update_xid_rep(ca.xid, ml);
 216
 217         if (destroy_wait_)
 218             destroy_wait_c_.notify_one();
 219     }
 220
 221     if (ca.done && lossytest_)
 222     {
 223         lock ml(m_);
 224         if (!dup_req_.isvalid()) {
 225             dup_req_.buf = req;
 226             dup_req_.xid = ca.xid;
 227         }
 228         if (xid_rep > xid_rep_done_)
 229             xid_rep_done_ = xid_rep;
 230     }
 231
 232     lock cal(ca.m);
 233
 234     IF_LEVEL(2) LOG(clt_nonce_ << " call done for req proc " << hex << proc <<
 235                     " xid " << dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":" <<
 236                     ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret);
 237
 238     // destruction of req automatically frees its buffer
 239     return (ca.done? ca.intret : rpc_protocol::timeout_failure);
 240 }
 241
 242 void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
 243     lock ml(chan_m_);
 244     if (!chan_ || chan_->isdead())
 245         chan_ = connection::to_dst(dst_, this, lossytest_);
 246
 247     if (chan_)
 248         ch = chan_;
 249 }
 250
 251 // PollMgr's thread is being used to
 252 // make this upcall from connection object to rpcc.
 253 // this funtion must not block.
 254 //
 255 // this function keeps no reference for connection *c
 256 bool
 257 rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
 258 {
 259     unmarshall rep(b, true);
 260     rpc_protocol::reply_header h;
 261     rep.unpack_header(h);
 262
 263     if (!rep.ok()) {
 264         IF_LEVEL(1) LOG("unmarshall header failed!!!");
 265         return true;
 266     }
 267
 268     lock ml(m_);
 269
 270     update_xid_rep(h.xid, ml);
 271
 272     if (calls_.find(h.xid) == calls_.end()) {
 273         IF_LEVEL(2) LOG("xid " << h.xid << " no pending request");
 274         return true;
 275     }
 276     caller *ca = calls_[h.xid];
 277
 278     lock cl(ca->m);
 279     if (!ca->done) {
 280         *ca->rep = b;
 281         ca->intret = h.ret;
 282         if (ca->intret < 0) {
 283             IF_LEVEL(2) LOG("RPC reply error for xid " << h.xid << " intret " << ca->intret);
 284         }
 285         ca->done = 1;
 286     }
 287     ca->c.notify_all();
 288     return true;
 289 }
 290
 291 void rpcc::update_xid_rep(xid_t xid, lock & m_lock) {
 292     VERIFY(m_lock);
 293     if (xid <= xid_rep_window_.front())
 294         return;
 295
 296     for (auto it = xid_rep_window_.begin(); it != xid_rep_window_.end(); it++) {
 297         if (*it > xid) {
 298             xid_rep_window_.insert(it, xid);
 299             goto compress;
 300         }
 301     }
 302     xid_rep_window_.push_back(xid);
 303
 304 compress:
 305     auto it = xid_rep_window_.begin();
 306     for (it++; it != xid_rep_window_.end(); it++) {
 307         while (xid_rep_window_.front() + 1 == *it)
 308             xid_rep_window_.pop_front();
 309     }
 310 }
 311
 312 rpcs::rpcs(in_port_t p1) : port_(p1)
 313 {
 314     set_rand_seed();
 315     nonce_ = (nonce_t)random();
 316     IF_LEVEL(2) LOG("created with nonce " << nonce_);
 317
 318     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
 319 }
 320
 321 void rpcs::start() {
 322     char *loss_env = getenv("RPC_LOSSY");
 323     listener_.reset(new connection_listener(this, port_, loss_env ? atoi(loss_env) : 0));
 324 }
 325
 326 rpcs::~rpcs() {
 327     // must delete listener before dispatchpool
 328     listener_ = nullptr;
 329     dispatchpool_ = nullptr;
 330 }
 331
 332 bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
 333     if (!reachable_) {
 334         IF_LEVEL(1) LOG("not reachable");
 335         return true;
 336     }
 337
 338     return dispatchpool_->addJob(std::bind(&rpcs::dispatch, this, c, b));
 339 }
 340
 341 void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
 342     unmarshall req(buf, true);
 343
 344     rpc_protocol::request_header h;
 345     req.unpack_header(h);
 346     proc_id_t proc = h.proc;
 347
 348     if (!req.ok()) {
 349         IF_LEVEL(1) LOG("unmarshall header failed");
 350         return;
 351     }
 352
 353     IF_LEVEL(2) LOG("rpc " << h.xid << " (proc " << hex << proc << ", last_rep " <<
 354                     dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce);
 355
 356     marshall rep;
 357     rpc_protocol::reply_header rh{h.xid,0};
 358
 359     // is client sending to an old instance of server?
 360     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {
 361         IF_LEVEL(2) LOG("rpc for an old server instance " << h.srv_nonce <<
 362                         " (current " << nonce_ << ") proc " << hex << h.proc);
 363         rh.ret = rpc_protocol::oldsrv_failure;
 364         rep.pack_header(rh);
 365         c->send(rep);
 366         return;
 367     }
 368
 369     handler *f;
 370     // is RPC proc a registered procedure?
 371     {
 372         lock pl(procs_m_);
 373         if (procs_.count(proc) < 1) {
 374             LOG("unknown proc 0x" << hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_);
 375             VERIFY(0);
 376             return;
 377         }
 378
 379         f = procs_[proc];
 380     }
 381
 382     // have i seen this client before?
 383     {
 384         lock rwl(reply_window_m_);
 385         // if we don't know about this clt_nonce, create a cleanup object
 386         if (reply_window_.find(h.clt_nonce) == reply_window_.end()) {
 387             VERIFY (reply_window_[h.clt_nonce].size() == 0); // create
 388             reply_window_[h.clt_nonce].push_back(reply_t(-1)); // store starting reply xid
 389             IF_LEVEL(2) LOG("new client " << h.clt_nonce << " xid " << h.xid <<
 390                             " chan " << c->fd << ", total clients " << (reply_window_.size()-1));
 391         }
 392     }
 393
 394     // save the latest good connection to the client
 395     {
 396         lock rwl(conns_m_);
 397         if (conns_.find(h.clt_nonce) == conns_.end())
 398             conns_[h.clt_nonce] = c;
 399         else if (conns_[h.clt_nonce]->create_time < c->create_time)
 400             conns_[h.clt_nonce] = c;
 401     }
 402
 403     string b1;
 404
 405     switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
 406         case NEW: // new request
 407             rh.ret = (*f)(forward<unmarshall>(req), rep);
 408             if (rh.ret == rpc_protocol::unmarshall_args_failure) {
 409                 LOG("failed to unmarshall the arguments. You are " <<
 410                     "probably calling RPC 0x" << hex << proc << " with the wrong " <<
 411                     "types of arguments.");
 412                 VERIFY(0);
 413             }
 414             VERIFY(rh.ret >= 0);
 415
 416             rep.pack_header(rh);
 417             b1 = rep;
 418
 419             IF_LEVEL(2) LOG("sending and saving reply of size " << b1.size() << " for rpc " <<
 420                             h.xid << ", proc " << hex << proc << " ret " << dec << rh.ret << ", clt " << h.clt_nonce);
 421
 422             add_reply(h.clt_nonce, h.xid, b1);
 423
 424             // get the latest connection to the client
 425             {
 426                 lock rwl(conns_m_);
 427                 if (c->isdead())
 428                     c = conns_[h.clt_nonce];
 429             }
 430
 431             c->send(rep);
 432             break;
 433         case INPROGRESS: // server is working on this request
 434             break;
 435         case DONE: // duplicate and we still have the response
 436             c->send(b1);
 437             break;
 438         case FORGOTTEN: // very old request and we don't have the response anymore
 439             IF_LEVEL(2) LOG("very old request " << h.xid << " from " << h.clt_nonce);
 440             rh.ret = rpc_protocol::atmostonce_failure;
 441             rep.pack_header(rh);
 442             c->send(rep);
 443             break;
 444     }
 445 }
 446
 447 // rpcs::dispatch calls this when an RPC request arrives.
 448 //
 449 // checks to see if an RPC with xid from clt_nonce has already been received.
 450 // if not, remembers the request in reply_window_.
 451 //
 452 // deletes remembered requests with XIDs <= xid_rep; the client
 453 // says it has received a reply for every RPC up through xid_rep.
 454 // frees the reply_t::buf of each such request.
 455 //
 456 // returns one of:
 457 //   NEW: never seen this xid before.
 458 //   INPROGRESS: seen this xid, and still processing it.
 459 //   DONE: seen this xid, previous reply returned in b.
 460 //   FORGOTTEN: might have seen this xid, but deleted previous reply.
 461 rpcs::rpcstate_t
 462 rpcs::check_duplicate_and_update(nonce_t clt_nonce, xid_t xid,
 463         xid_t xid_rep, string & b)
 464 {
 465     lock rwl(reply_window_m_);
 466
 467     list<reply_t> & l = reply_window_[clt_nonce];
 468
 469     VERIFY(l.size() > 0);
 470     VERIFY(xid >= xid_rep);
 471
 472     xid_t past_xid_rep = l.begin()->xid;
 473
 474     list<reply_t>::iterator start = l.begin(), it = ++start;
 475
 476     if (past_xid_rep < xid_rep || past_xid_rep == -1) {
 477         // scan for deletion candidates
 478         while (it != l.end() && it->xid < xid_rep)
 479             it++;
 480         l.erase(start, it);
 481         l.begin()->xid = xid_rep;
 482     }
 483
 484     if (xid < past_xid_rep && past_xid_rep != -1)
 485         return FORGOTTEN;
 486
 487     // skip non-deletion candidates
 488     while (it != l.end() && it->xid < xid)
 489         it++;
 490
 491     // if it's in the list it must be right here
 492     if (it != l.end() && it->xid == xid) {
 493         if (it->cb_present) {
 494             // return information about the remembered reply
 495             b = it->buf;
 496             return DONE;
 497         }
 498         return INPROGRESS;
 499     } else {
 500         // remember that a new request has arrived
 501         l.insert(it, reply_t(xid));
 502         return NEW;
 503     }
 504 }
 505
 506 // rpcs::dispatch calls add_reply when it is sending a reply to an RPC,
 507 // and passes the return value in b.
 508 // add_reply() should remember b.
 509 void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) {
 510     lock rwl(reply_window_m_);
 511     // remember the RPC reply value
 512     list<reply_t> & l = reply_window_[clt_nonce];
 513     list<reply_t>::iterator it = l.begin();
 514     // skip to our place in the list
 515     for (it++; it != l.end() && it->xid < xid; it++);
 516     // there should already be an entry, so whine if there isn't
 517     if (it == l.end() || it->xid != xid) {
 518         LOG("Could not find reply struct in add_reply");
 519         l.insert(it, reply_t(xid, b));
 520     } else {
 521         *it = reply_t(xid, b);
 522     }
 523 }
 524
 525 rpc_protocol::status rpcs::rpcbind(nonce_t & r) {
 526     IF_LEVEL(2) LOG("called return nonce " << nonce_);
 527     r = nonce_;
 528     return 0;
 529 }
 530
 531 static sockaddr_in make_sockaddr(const string & hostandport) {
 532     string host = "127.0.0.1";
 533     string port = hostandport;
 534     auto colon = hostandport.find(':');
 535     if (colon != string::npos) {
 536         host = hostandport.substr(0, colon);
 537         port = hostandport.substr(colon+1);
 538     }
 539
 540     sockaddr_in dst{}; // zero initialize
 541     dst.sin_family = AF_INET;
 542
 543     struct in_addr a{inet_addr(host.c_str())};
 544
 545     if (a.s_addr != INADDR_NONE)
 546         dst.sin_addr.s_addr = a.s_addr;
 547     else {
 548         struct hostent *hp = gethostbyname(host.c_str());
 549
 550         if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) {
 551             LOG_NONMEMBER("cannot find host name " << host);
 552             exit(1);
 553         }
 554         memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));
 555         dst.sin_addr.s_addr = a.s_addr;
 556     }
 557     dst.sin_port = hton((in_port_t)stoi(port));
 558     return dst;
 559 }