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