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