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 using std::list;
  64 using namespace std::chrono;
  65
  66 inline void set_rand_seed() {
  67     auto now = time_point_cast<nanoseconds>(steady_clock::now());
  68     srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid());
  69 }
  70
  71 static sockaddr_in make_sockaddr(const string & hostandport);
  72
  73 rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
  74 {
  75     set_rand_seed();
  76     clt_nonce_ = (nonce_t)random();
  77
  78     char *loss_env = getenv("RPC_LOSSY");
  79     if (loss_env)
  80         lossytest_ = atoi(loss_env);
  81
  82     IF_LEVEL(2) LOG << "cltn_nonce is " << clt_nonce_ << " lossy " << lossytest_;
  83 }
  84
  85 // IMPORTANT: destruction should happen only when no external threads
  86 // are blocked inside rpcc or will use rpcc in the future
  87 rpcc::~rpcc() {
  88     lock ml(m_);
  89     cancel(ml);
  90     IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1);
  91     chan_.reset();
  92     VERIFY(calls_.size() == 0);
  93 }
  94
  95 int rpcc::bind(milliseconds to) {
  96     nonce_t r = 0;
  97     rpc_protocol::status ret = call_timeout(rpc_protocol::bind, to, r);
  98     if (ret == 0) {
  99         lock ml(m_);
 100         bind_done_ = true;
 101         srv_nonce_ = r;
 102     } else {
 103         IF_LEVEL(2) LOG << "bind " << inet_ntoa(dst_.sin_addr) << " failed " << ret;
 104     }
 105     return ret;
 106 }
 107
 108 shared_ptr<rpcc> rpcc::bind_cached(const string & destination) {
 109     auto client = global->get_handle(destination);
 110     lock cl = lock(client->bind_m_);
 111     if (!client->bind_done_) {
 112         LOG_NONMEMBER << "bind(\"" << destination << "\")";
 113         int ret = client->bind(milliseconds(1000));
 114         if (ret < 0) {
 115             LOG_NONMEMBER << "bind failure! " << destination << " " << ret;
 116             client.reset();
 117         } else {
 118             LOG_NONMEMBER << "bind succeeded " << destination;
 119         }
 120     }
 121     return client;
 122 }
 123
 124 void rpcc::unbind_cached(const string & destination) {
 125     global->erase_handle(destination);
 126 }
 127
 128 // Cancel all outstanding calls
 129 void rpcc::cancel(lock & m_lock) {
 130     VERIFY(m_lock);
 131     if (calls_.size()) {
 132         LOG << "force callers to fail";
 133         for (auto & p : calls_) {
 134             caller *ca = p.second;
 135
 136             IF_LEVEL(2) LOG << "force caller to fail";
 137
 138             lock cl(ca->m);
 139             ca->done = true;
 140             ca->intret = rpc_protocol::cancel_failure;
 141             ca->c.notify_one();
 142         }
 143
 144         destroy_wait_ = true;
 145         while (calls_.size () > 0)
 146             destroy_wait_c_.wait(m_lock);
 147
 148         LOG << "done";
 149     }
 150 }
 151
 152 int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
 153
 154     caller ca(0, &rep);
 155     xid_t xid_rep;
 156     {
 157         lock ml(m_);
 158
 159         if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) {
 160             IF_LEVEL(1) LOG << "rpcc has not been bound to dst or binding twice";
 161             return rpc_protocol::bind_failure;
 162         }
 163
 164         if (destroy_wait_)
 165             return rpc_protocol::cancel_failure;
 166
 167         ca.xid = xid_++;
 168         calls_[ca.xid] = &ca;
 169
 170         req.write_header(rpc_protocol::request_header{
 171                 ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
 172                 });
 173         xid_rep = xid_rep_window_.front();
 174     }
 175
 176     milliseconds curr_to = rpc::to_min;
 177     auto finaldeadline = steady_clock::now() + to;
 178
 179     bool transmit = true;
 180     shared_ptr<connection> ch;
 181
 182     while (1) {
 183         if (transmit) {
 184             get_latest_connection(ch);
 185             if (ch) {
 186                 if (reachable_) {
 187                     request forgot;
 188                     {
 189                         lock ml(m_);
 190                         if (dup_req_.isvalid() && xid_rep_done_ > dup_req_.xid) {
 191                             forgot = dup_req_;
 192                             dup_req_.clear();
 193                         }
 194                     }
 195                     if (forgot.isvalid())
 196                         ch->send(forgot.buf);
 197                     ch->send(req);
 198                 }
 199                 else IF_LEVEL(1) LOG << "not reachable";
 200                 IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc
 201                                 << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_;
 202             }
 203             transmit = false; // only send once on a given channel
 204         }
 205
 206         auto nextdeadline = std::min(steady_clock::now() + curr_to, finaldeadline);
 207         curr_to *= 2;
 208
 209         {
 210             lock cal(ca.m);
 211             while (!ca.done) {
 212                 IF_LEVEL(2) LOG << "wait";
 213                 if (ca.c.wait_until(cal, nextdeadline) == std::cv_status::timeout) {
 214                     IF_LEVEL(2) LOG << "timeout";
 215                     break;
 216                 }
 217             }
 218             if (ca.done) {
 219                 IF_LEVEL(2) LOG << "reply received";
 220                 break;
 221             }
 222         }
 223
 224         if (nextdeadline >= finaldeadline)
 225             break;
 226
 227         // retransmit on new connection if connection is dead
 228         if (!ch || ch->isdead())
 229             transmit = true;
 230     }
 231
 232     {
 233         // no locking of ca.m since only this thread changes ca.xid
 234         lock ml(m_);
 235         calls_.erase(ca.xid);
 236         // may need to update the xid again here, in case the
 237         // packet times out before it's even sent by the channel.
 238         // I don't think there's any harm in maybe doing it twice
 239         update_xid_rep(ca.xid, ml);
 240
 241         if (destroy_wait_)
 242             destroy_wait_c_.notify_one();
 243     }
 244
 245     if (ca.done && lossytest_)
 246     {
 247         lock ml(m_);
 248         if (!dup_req_.isvalid()) {
 249             dup_req_.buf = req;
 250             dup_req_.xid = ca.xid;
 251         }
 252         if (xid_rep > xid_rep_done_)
 253             xid_rep_done_ = xid_rep;
 254     }
 255
 256     lock cal(ca.m);
 257
 258     IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc
 259                     << " xid " << std::dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":"
 260                     << ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret;
 261
 262     // destruction of req automatically frees its buffer
 263     return (ca.done? ca.intret : rpc_protocol::timeout_failure);
 264 }
 265
 266 void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
 267     lock ml(chan_m_);
 268     if (!chan_ || chan_->isdead())
 269         chan_ = connection::to_dst(dst_, this, lossytest_);
 270
 271     if (chan_)
 272         ch = chan_;
 273 }
 274
 275 // PollMgr's thread is being used to
 276 // make this upcall from connection object to rpcc.
 277 // this funtion must not block.
 278 //
 279 // this function keeps no reference for connection *c
 280 bool
 281 rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
 282 {
 283     unmarshall rep(b, true);
 284     rpc_protocol::reply_header h;
 285     rep.read_header(h);
 286
 287     if (!rep.ok()) {
 288         IF_LEVEL(1) LOG << "unmarshall header failed!!!";
 289         return true;
 290     }
 291
 292     lock ml(m_);
 293
 294     update_xid_rep(h.xid, ml);
 295
 296     if (calls_.find(h.xid) == calls_.end()) {
 297         IF_LEVEL(2) LOG << "xid " << h.xid << " no pending request";
 298         return true;
 299     }
 300     caller *ca = calls_[h.xid];
 301
 302     lock cl(ca->m);
 303     if (!ca->done) {
 304         *ca->rep = b;
 305         ca->intret = h.ret;
 306         if (ca->intret < 0) {
 307             IF_LEVEL(2) LOG << "RPC reply error for xid " << h.xid << " intret " << ca->intret;
 308         }
 309         ca->done = 1;
 310     }
 311     ca->c.notify_all();
 312     return true;
 313 }
 314
 315 void rpcc::update_xid_rep(xid_t xid, lock & m_lock) {
 316     VERIFY(m_lock);
 317     if (xid <= xid_rep_window_.front())
 318         return;
 319
 320     for (auto it = xid_rep_window_.begin(); it != xid_rep_window_.end(); it++) {
 321         if (*it > xid) {
 322             xid_rep_window_.insert(it, xid);
 323             goto compress;
 324         }
 325     }
 326     xid_rep_window_.push_back(xid);
 327
 328 compress:
 329     auto it = xid_rep_window_.begin();
 330     for (it++; it != xid_rep_window_.end(); it++) {
 331         while (xid_rep_window_.front() + 1 == *it)
 332             xid_rep_window_.pop_front();
 333     }
 334 }
 335
 336 rpcs::rpcs(in_port_t p1) : port_(p1)
 337 {
 338     set_rand_seed();
 339     nonce_ = (nonce_t)random();
 340     IF_LEVEL(2) LOG << "created with nonce " << nonce_;
 341
 342     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
 343 }
 344
 345 void rpcs::start() {
 346     char *loss_env = getenv("RPC_LOSSY");
 347     listener_.reset(new connection_listener(this, port_, loss_env ? atoi(loss_env) : 0));
 348 }
 349
 350 rpcs::~rpcs() {
 351     // must delete listener before dispatchpool
 352     listener_ = nullptr;
 353     dispatchpool_ = nullptr;
 354 }
 355
 356 bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
 357     if (!reachable_) {
 358         IF_LEVEL(1) LOG << "not reachable";
 359         return true;
 360     }
 361
 362     return dispatchpool_->addJob(std::bind(&rpcs::dispatch, this, c, b));
 363 }
 364
 365 void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
 366     unmarshall req(buf, true);
 367
 368     rpc_protocol::request_header h;
 369     req.read_header(h);
 370     proc_id_t proc = h.proc;
 371
 372     if (!req.ok()) {
 373         IF_LEVEL(1) LOG << "unmarshall header failed";
 374         return;
 375     }
 376
 377     IF_LEVEL(2) LOG << "rpc " << h.xid << " (proc " << std::hex << proc << ", last_rep "
 378                     << std::dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce;
 379
 380     marshall rep;
 381     rpc_protocol::reply_header rh{h.xid,0};
 382
 383     // is client sending to an old instance of server?
 384     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {
 385         IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce
 386                         << " (current " << nonce_ << ") proc " << std::hex << h.proc;
 387         rh.ret = rpc_protocol::oldsrv_failure;
 388         rep.write_header(rh);
 389         c->send(rep);
 390         return;
 391     }
 392
 393     handler *f;
 394     // is RPC proc a registered procedure?
 395     {
 396         lock pl(procs_m_);
 397         if (procs_.count(proc) < 1) {
 398             LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;
 399             VERIFY(0);
 400             return;
 401         }
 402
 403         f = procs_[proc];
 404     }
 405
 406     // have i seen this client before?
 407     {
 408         lock rwl(reply_window_m_);
 409         // if we don't know about this clt_nonce, create a cleanup object
 410         if (reply_window_.find(h.clt_nonce) == reply_window_.end()) {
 411             VERIFY (reply_window_[h.clt_nonce].size() == 0); // create
 412             reply_window_[h.clt_nonce].push_back(reply_t(-1)); // store starting reply xid
 413             IF_LEVEL(2) LOG << "new client " << h.clt_nonce << " xid " << h.xid
 414                             << " chan " << c->fd << ", total clients " << (reply_window_.size()-1);
 415         }
 416     }
 417
 418     // save the latest good connection to the client
 419     {
 420         lock rwl(conns_m_);
 421         if (conns_.find(h.clt_nonce) == conns_.end())
 422             conns_[h.clt_nonce] = c;
 423         else if (conns_[h.clt_nonce]->create_time < c->create_time)
 424             conns_[h.clt_nonce] = c;
 425     }
 426
 427     string b1;
 428
 429     switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
 430         case NEW: // new request
 431             rh.ret = (*f)(std::forward<unmarshall>(req), rep);
 432             if (rh.ret == rpc_protocol::unmarshall_args_failure) {
 433                 LOG << "failed to unmarshall the arguments. You are "
 434                     << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
 435                     << "types of arguments.";
 436                 VERIFY(0);
 437             }
 438             VERIFY(rh.ret >= 0);
 439
 440             rep.write_header(rh);
 441             b1 = rep;
 442
 443             IF_LEVEL(2) LOG << "sending and saving reply of size " << b1.size() << " for rpc "
 444                             << h.xid << ", proc " << std::hex << proc << " ret " << std::dec
 445                             << rh.ret << ", clt " << h.clt_nonce;
 446
 447             add_reply(h.clt_nonce, h.xid, b1);
 448
 449             // get the latest connection to the client
 450             {
 451                 lock rwl(conns_m_);
 452                 if (c->isdead())
 453                     c = conns_[h.clt_nonce];
 454             }
 455
 456             c->send(rep);
 457             break;
 458         case INPROGRESS: // server is working on this request
 459             break;
 460         case DONE: // duplicate and we still have the response
 461             c->send(b1);
 462             break;
 463         case FORGOTTEN: // very old request and we don't have the response anymore
 464             IF_LEVEL(2) LOG << "very old request " << h.xid << " from " << h.clt_nonce;
 465             rh.ret = rpc_protocol::atmostonce_failure;
 466             rep.write_header(rh);
 467             c->send(rep);
 468             break;
 469     }
 470 }
 471
 472 // rpcs::dispatch calls this when an RPC request arrives.
 473 //
 474 // checks to see if an RPC with xid from clt_nonce has already been received.
 475 // if not, remembers the request in reply_window_.
 476 //
 477 // deletes remembered requests with XIDs <= xid_rep; the client
 478 // says it has received a reply for every RPC up through xid_rep.
 479 // frees the reply_t::buf of each such request.
 480 //
 481 // returns one of:
 482 //   NEW: never seen this xid before.
 483 //   INPROGRESS: seen this xid, and still processing it.
 484 //   DONE: seen this xid, previous reply returned in b.
 485 //   FORGOTTEN: might have seen this xid, but deleted previous reply.
 486 rpcs::rpcstate_t
 487 rpcs::check_duplicate_and_update(nonce_t clt_nonce, xid_t xid,
 488         xid_t xid_rep, string & b)
 489 {
 490     lock rwl(reply_window_m_);
 491
 492     list<reply_t> & l = reply_window_[clt_nonce];
 493
 494     VERIFY(l.size() > 0);
 495     VERIFY(xid >= xid_rep);
 496
 497     xid_t past_xid_rep = l.begin()->xid;
 498
 499     list<reply_t>::iterator start = l.begin(), it = ++start;
 500
 501     if (past_xid_rep < xid_rep || past_xid_rep == -1) {
 502         // scan for deletion candidates
 503         while (it != l.end() && it->xid < xid_rep)
 504             it++;
 505         l.erase(start, it);
 506         l.begin()->xid = xid_rep;
 507     }
 508
 509     if (xid < past_xid_rep && past_xid_rep != -1)
 510         return FORGOTTEN;
 511
 512     // skip non-deletion candidates
 513     while (it != l.end() && it->xid < xid)
 514         it++;
 515
 516     // if it's in the list it must be right here
 517     if (it != l.end() && it->xid == xid) {
 518         if (it->cb_present) {
 519             // return information about the remembered reply
 520             b = it->buf;
 521             return DONE;
 522         }
 523         return INPROGRESS;
 524     } else {
 525         // remember that a new request has arrived
 526         l.insert(it, reply_t(xid));
 527         return NEW;
 528     }
 529 }
 530
 531 // rpcs::dispatch calls add_reply when it is sending a reply to an RPC,
 532 // and passes the return value in b.
 533 // add_reply() should remember b.
 534 void rpcs::add_reply(nonce_t clt_nonce, xid_t xid, const string & b) {
 535     lock rwl(reply_window_m_);
 536     // remember the RPC reply value
 537     list<reply_t> & l = reply_window_[clt_nonce];
 538     list<reply_t>::iterator it = l.begin();
 539     // skip to our place in the list
 540     for (it++; it != l.end() && it->xid < xid; it++);
 541     // there should already be an entry, so whine if there isn't
 542     if (it == l.end() || it->xid != xid) {
 543         LOG << "Could not find reply struct in add_reply";
 544         l.insert(it, reply_t(xid, b));
 545     } else {
 546         *it = reply_t(xid, b);
 547     }
 548 }
 549
 550 rpc_protocol::status rpcs::rpcbind(nonce_t & r) {
 551     IF_LEVEL(2) LOG << "called return nonce " << nonce_;
 552     r = nonce_;
 553     return 0;
 554 }
 555
 556 static sockaddr_in make_sockaddr(const string & hostandport) {
 557     string host = "127.0.0.1";
 558     string port = hostandport;
 559     auto colon = hostandport.find(':');
 560     if (colon != string::npos) {
 561         host = hostandport.substr(0, colon);
 562         port = hostandport.substr(colon+1);
 563     }
 564
 565     sockaddr_in dst = sockaddr_in(); // zero initialize
 566     dst.sin_family = AF_INET;
 567
 568     struct in_addr a{inet_addr(host.c_str())};
 569
 570     if (a.s_addr != INADDR_NONE)
 571         dst.sin_addr.s_addr = a.s_addr;
 572     else {
 573         struct hostent *hp = gethostbyname(host.c_str());
 574
 575         if (!hp || hp->h_length != 4 || hp->h_addrtype != AF_INET) {
 576             LOG_NONMEMBER << "cannot find host name " << host;
 577             exit(1);
 578         }
 579         memcpy(&a, hp->h_addr_list[0], sizeof(in_addr_t));
 580         dst.sin_addr.s_addr = a.s_addr;
 581     }
 582     dst.sin_port = hton((in_port_t)std::stoi(port));
 583     return dst;
 584 }