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