rpc/rpc.h

   1 #ifndef rpc_h
   2 #define rpc_h
   3
   4 #include <sys/socket.h>
   5 #include <netinet/in.h>
   6 #include <list>
   7 #include <map>
   8 #include <stdio.h>
   9
  10 #include "thr_pool.h"
  11 #include "marshall.h"
  12 #include "connection.h"
  13
  14 #ifdef DMALLOC
  15 #include "dmalloc.h"
  16 #endif
  17
  18 class rpc_const {
  19         public:
  20                 static const unsigned int bind = 1;   // handler number reserved for bind
  21                 static const int timeout_failure = -1;
  22                 static const int unmarshal_args_failure = -2;
  23                 static const int unmarshal_reply_failure = -3;
  24                 static const int atmostonce_failure = -4;
  25                 static const int oldsrv_failure = -5;
  26                 static const int bind_failure = -6;
  27                 static const int cancel_failure = -7;
  28 };
  29
  30 // rpc client endpoint.
  31 // manages a xid space per destination socket
  32 // threaded: multiple threads can be sending RPCs,
  33 class rpcc : public chanmgr {
  34
  35         private:
  36
  37                 //manages per rpc info
  38                 struct caller {
  39                         caller(unsigned int xxid, unmarshall *un);
  40                         ~caller();
  41
  42                         unsigned int xid;
  43                         unmarshall *un;
  44                         int intret;
  45                         bool done;
  46                         pthread_mutex_t m;
  47                         pthread_cond_t c;
  48                 };
  49
  50                 void get_refconn(connection **ch);
  51                 void update_xid_rep(unsigned int xid);
  52
  53
  54                 sockaddr_in dst_;
  55                 unsigned int clt_nonce_;
  56                 unsigned int srv_nonce_;
  57                 bool bind_done_;
  58                 unsigned int xid_;
  59                 int lossytest_;
  60                 bool retrans_;
  61                 bool reachable_;
  62
  63                 connection *chan_;
  64
  65                 pthread_mutex_t m_; // protect insert/delete to calls[]
  66                 pthread_mutex_t chan_m_;
  67
  68                 bool destroy_wait_;
  69                 pthread_cond_t destroy_wait_c_;
  70
  71                 std::map<int, caller *> calls_;
  72                 std::list<unsigned int> xid_rep_window_;
  73
  74                 struct request {
  75                     request() { clear(); }
  76                     void clear() { buf.clear(); xid = -1; }
  77                     bool isvalid() { return xid != -1; }
  78                     std::string buf;
  79                     int xid;
  80                 };
  81                 struct request dup_req_;
  82                 int xid_rep_done_;
  83         public:
  84
  85                 rpcc(sockaddr_in d, bool retrans=true);
  86                 ~rpcc();
  87
  88                 struct TO {
  89                         int to;
  90                 };
  91                 static const TO to_max;
  92                 static const TO to_min;
  93                 static TO to(int x) { TO t; t.to = x; return t;}
  94
  95                 unsigned int id() { return clt_nonce_; }
  96
  97                 int bind(TO to = to_max);
  98
  99                 void set_reachable(bool r) { reachable_ = r; }
 100
 101                 void cancel();
 102
 103                 int islossy() { return lossytest_ > 0; }
 104
 105                 int call1(unsigned int proc,
 106                                 marshall &req, unmarshall &rep, TO to);
 107
 108                 bool got_pdu(connection *c, char *b, int sz);
 109
 110
 111                 template<class R>
 112                         int call_m(unsigned int proc, marshall &req, R & r, TO to);
 113
 114                 template<class R>
 115                         int call(unsigned int proc, R & r, TO to = to_max);
 116                 template<class R, class A1>
 117                         int call(unsigned int proc, const A1 & a1, R & r, TO to = to_max);
 118                 template<class R, class A1, class A2>
 119                         int call(unsigned int proc, const A1 & a1, const A2 & a2, R & r,
 120                                         TO to = to_max);
 121                 template<class R, class A1, class A2, class A3>
 122                         int call(unsigned int proc, const A1 & a1, const A2 & a2, const A3 & a3,
 123                                         R & r, TO to = to_max);
 124                 template<class R, class A1, class A2, class A3, class A4>
 125                         int call(unsigned int proc, const A1 & a1, const A2 & a2, const A3 & a3,
 126                                         const A4 & a4, R & r, TO to = to_max);
 127                 template<class R, class A1, class A2, class A3, class A4, class A5>
 128                         int call(unsigned int proc, const A1 & a1, const A2 & a2, const A3 & a3,
 129                                         const A4 & a4, const A5 & a5, R & r, TO to = to_max);
 130                 template<class R, class A1, class A2, class A3, class A4, class A5,
 131                         class A6>
 132                                 int call(unsigned int proc, const A1 & a1, const A2 & a2, const A3 & a3,
 133                                                 const A4 & a4, const A5 & a5, const A6 & a6,
 134                                                 R & r, TO to = to_max);
 135                 template<class R, class A1, class A2, class A3, class A4, class A5,
 136                         class A6, class A7>
 137                                 int call(unsigned int proc, const A1 & a1, const A2 & a2, const A3 & a3,
 138                                                 const A4 & a4, const A5 & a5, const A6 &a6, const A7 &a7,
 139                                                 R & r, TO to = to_max);
 140
 141 };
 142
 143 template<class R> int
 144 rpcc::call_m(unsigned int proc, marshall &req, R & r, TO to)
 145 {
 146         unmarshall u;
 147         int intret = call1(proc, req, u, to);
 148         if (intret < 0) return intret;
 149         u >> r;
 150         if(u.okdone() != true) {
 151                 fprintf(stderr, "rpcc::call_m: failed to unmarshall the reply."
 152                        "You are probably calling RPC 0x%x with wrong return "
 153                        "type.\n", proc);
 154                 VERIFY(0);
 155                 return rpc_const::unmarshal_reply_failure;
 156         }
 157         return intret;
 158 }
 159
 160 template<class R> int
 161 rpcc::call(unsigned int proc, R & r, TO to)
 162 {
 163         marshall m;
 164         return call_m(proc, m, r, to);
 165 }
 166
 167 template<class R, class A1> int
 168 rpcc::call(unsigned int proc, const A1 & a1, R & r, TO to)
 169 {
 170         marshall m;
 171         m << a1;
 172         return call_m(proc, m, r, to);
 173 }
 174
 175 template<class R, class A1, class A2> int
 176 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 177                 R & r, TO to)
 178 {
 179         marshall m;
 180         m << a1;
 181         m << a2;
 182         return call_m(proc, m, r, to);
 183 }
 184
 185 template<class R, class A1, class A2, class A3> int
 186 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 187                 const A3 & a3, R & r, TO to)
 188 {
 189         marshall m;
 190         m << a1;
 191         m << a2;
 192         m << a3;
 193         return call_m(proc, m, r, to);
 194 }
 195
 196 template<class R, class A1, class A2, class A3, class A4> int
 197 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 198                 const A3 & a3, const A4 & a4, R & r, TO to)
 199 {
 200         marshall m;
 201         m << a1;
 202         m << a2;
 203         m << a3;
 204         m << a4;
 205         return call_m(proc, m, r, to);
 206 }
 207
 208 template<class R, class A1, class A2, class A3, class A4, class A5> int
 209 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 210                 const A3 & a3, const A4 & a4, const A5 & a5, R & r, TO to)
 211 {
 212         marshall m;
 213         m << a1;
 214         m << a2;
 215         m << a3;
 216         m << a4;
 217         m << a5;
 218         return call_m(proc, m, r, to);
 219 }
 220
 221 template<class R, class A1, class A2, class A3, class A4, class A5,
 222         class A6> int
 223 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 224                 const A3 & a3, const A4 & a4, const A5 & a5,
 225                 const A6 & a6, R & r, TO to)
 226 {
 227         marshall m;
 228         m << a1;
 229         m << a2;
 230         m << a3;
 231         m << a4;
 232         m << a5;
 233         m << a6;
 234         return call_m(proc, m, r, to);
 235 }
 236
 237 template<class R, class A1, class A2, class A3, class A4, class A5,
 238         class A6, class A7> int
 239 rpcc::call(unsigned int proc, const A1 & a1, const A2 & a2,
 240                 const A3 & a3, const A4 & a4, const A5 & a5,
 241                 const A6 & a6, const A7 & a7,
 242                 R & r, TO to)
 243 {
 244         marshall m;
 245         m << a1;
 246         m << a2;
 247         m << a3;
 248         m << a4;
 249         m << a5;
 250         m << a6;
 251         m << a7;
 252         return call_m(proc, m, r, to);
 253 }
 254
 255 bool operator<(const sockaddr_in &a, const sockaddr_in &b);
 256
 257 class handler {
 258         public:
 259                 handler() { }
 260                 virtual ~handler() { }
 261                 virtual int fn(unmarshall &, marshall &) = 0;
 262 };
 263
 264
 265 // rpc server endpoint.
 266 class rpcs : public chanmgr {
 267
 268         typedef enum {
 269                 NEW,  // new RPC, not a duplicate
 270                 INPROGRESS, // duplicate of an RPC we're still processing
 271                 DONE, // duplicate of an RPC we already replied to (have reply)
 272                 FORGOTTEN,  // duplicate of an old RPC whose reply we've forgotten
 273         } rpcstate_t;
 274
 275         private:
 276
 277         // state about an in-progress or completed RPC, for at-most-once.
 278         // if cb_present is true, then the RPC is complete and a reply
 279         // has been sent; in that case buf points to a copy of the reply,
 280         // and sz holds the size of the reply.
 281         struct reply_t {
 282                 reply_t (unsigned int _xid) {
 283                         xid = _xid;
 284                         cb_present = false;
 285                         buf = NULL;
 286                         sz = 0;
 287                 }
 288                 reply_t (unsigned int _xid, char *_buf, int _sz) {
 289                         xid = _xid;
 290                         cb_present = true;
 291                         buf = _buf;
 292                         sz = _sz;
 293                 }
 294                 unsigned int xid;
 295                 bool cb_present; // whether the reply buffer is valid
 296                 char *buf;      // the reply buffer
 297                 int sz;         // the size of reply buffer
 298         };
 299
 300         int port_;
 301         unsigned int nonce_;
 302
 303         // provide at most once semantics by maintaining a window of replies
 304         // per client that that client hasn't acknowledged receiving yet.
 305         // indexed by client nonce.
 306         std::map<unsigned int, std::list<reply_t> > reply_window_;
 307
 308         void free_reply_window(void);
 309         void add_reply(unsigned int clt_nonce, unsigned int xid, char *b, int sz);
 310
 311         rpcstate_t checkduplicate_and_update(unsigned int clt_nonce,
 312                         unsigned int xid, unsigned int rep_xid,
 313                         char **b, int *sz);
 314
 315         void updatestat(unsigned int proc);
 316
 317         // latest connection to the client
 318         std::map<unsigned int, connection *> conns_;
 319
 320         // counting
 321         const int counting_;
 322         int curr_counts_;
 323         std::map<int, int> counts_;
 324
 325         int lossytest_;
 326         bool reachable_;
 327
 328         // map proc # to function
 329         std::map<int, handler *> procs_;
 330
 331         pthread_mutex_t procs_m_; // protect insert/delete to procs[]
 332         pthread_mutex_t count_m_;  //protect modification of counts
 333         pthread_mutex_t reply_window_m_; // protect reply window et al
 334         pthread_mutex_t conss_m_; // protect conns_
 335
 336
 337         protected:
 338
 339         struct djob_t {
 340                 djob_t (connection *c, char *b, int bsz):buf(b),sz(bsz),conn(c) {}
 341                 char *buf;
 342                 int sz;
 343                 connection *conn;
 344         };
 345         void dispatch(djob_t *);
 346
 347         // internal handler registration
 348         void reg1(unsigned int proc, handler *);
 349
 350         ThrPool* dispatchpool_;
 351         tcpsconn* listener_;
 352
 353         public:
 354         rpcs(unsigned int port, int counts=0);
 355         ~rpcs();
 356         inline int port() { return listener_->port(); }
 357         //RPC handler for clients binding
 358         int rpcbind(int a, int &r);
 359
 360         void set_reachable(bool r) { reachable_ = r; }
 361
 362         bool got_pdu(connection *c, char *b, int sz);
 363
 364         // register a handler
 365         template<class S, class A1, class R>
 366                 void reg(unsigned int proc, S*, int (S::*meth)(const A1 a1, R & r));
 367         template<class S, class A1, class A2, class R>
 368                 void reg(unsigned int proc, S*, int (S::*meth)(const A1 a1, const A2,
 369                                         R & r));
 370         template<class S, class A1, class A2, class A3, class R>
 371                 void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2,
 372                                         const A3, R & r));
 373         template<class S, class A1, class A2, class A3, class A4, class R>
 374                 void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2,
 375                                         const A3, const A4, R & r));
 376         template<class S, class A1, class A2, class A3, class A4, class A5, class R>
 377                 void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2,
 378                                         const A3, const A4, const A5,
 379                                         R & r));
 380         template<class S, class A1, class A2, class A3, class A4, class A5, class A6,
 381                 class R>
 382                         void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2,
 383                                                 const A3, const A4, const A5,
 384                                                 const A6, R & r));
 385         template<class S, class A1, class A2, class A3, class A4, class A5, class A6,
 386                 class A7, class R>
 387                         void reg(unsigned int proc, S*, int (S::*meth)(const A1, const A2,
 388                                                 const A3, const A4, const A5,
 389                                                 const A6, const A7,
 390                                                 R & r));
 391 };
 392
 393 template<class S, class A1, class R> void
 394 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, R & r))
 395 {
 396         class h1 : public handler {
 397                 private:
 398                         S * sob;
 399                         int (S::*meth)(const A1 a1, R & r);
 400                 public:
 401                         h1(S *xsob, int (S::*xmeth)(const A1 a1, R & r))
 402                                 : sob(xsob), meth(xmeth) { }
 403                         int fn(unmarshall &args, marshall &ret) {
 404                                 A1 a1;
 405                                 R r;
 406                                 args >> a1;
 407                                 if(!args.okdone())
 408                                         return rpc_const::unmarshal_args_failure;
 409                                 int b = (sob->*meth)(a1, r);
 410                                 ret << r;
 411                                 return b;
 412                         }
 413         };
 414         reg1(proc, new h1(sob, meth));
 415 }
 416
 417 template<class S, class A1, class A2, class R> void
 418 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 419                         R & r))
 420 {
 421         class h1 : public handler {
 422                 private:
 423                         S * sob;
 424                         int (S::*meth)(const A1 a1, const A2 a2, R & r);
 425                 public:
 426                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, R & r))
 427                                 : sob(xsob), meth(xmeth) { }
 428                         int fn(unmarshall &args, marshall &ret) {
 429                                 A1 a1;
 430                                 A2 a2;
 431                                 R r;
 432                                 args >> a1;
 433                                 args >> a2;
 434                                 if(!args.okdone())
 435                                         return rpc_const::unmarshal_args_failure;
 436                                 int b = (sob->*meth)(a1, a2, r);
 437                                 ret << r;
 438                                 return b;
 439                         }
 440         };
 441         reg1(proc, new h1(sob, meth));
 442 }
 443
 444 template<class S, class A1, class A2, class A3, class R> void
 445 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 446                         const A3 a3, R & r))
 447 {
 448         class h1 : public handler {
 449                 private:
 450                         S * sob;
 451                         int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, R & r);
 452                 public:
 453                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3, R & r))
 454                                 : sob(xsob), meth(xmeth) { }
 455                         int fn(unmarshall &args, marshall &ret) {
 456                                 A1 a1;
 457                                 A2 a2;
 458                                 A3 a3;
 459                                 R r;
 460                                 args >> a1;
 461                                 args >> a2;
 462                                 args >> a3;
 463                                 if(!args.okdone())
 464                                         return rpc_const::unmarshal_args_failure;
 465                                 int b = (sob->*meth)(a1, a2, a3, r);
 466                                 ret << r;
 467                                 return b;
 468                         }
 469         };
 470         reg1(proc, new h1(sob, meth));
 471 }
 472
 473 template<class S, class A1, class A2, class A3, class A4, class R> void
 474 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 475                         const A3 a3, const A4 a4,
 476                         R & r))
 477 {
 478         class h1 : public handler {
 479                 private:
 480                         S * sob;
 481                         int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4, R & r);
 482                 public:
 483                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3,
 484                                                 const A4 a4, R & r))
 485                                 : sob(xsob), meth(xmeth)  { }
 486                         int fn(unmarshall &args, marshall &ret) {
 487                                 A1 a1;
 488                                 A2 a2;
 489                                 A3 a3;
 490                                 A4 a4;
 491                                 R r;
 492                                 args >> a1;
 493                                 args >> a2;
 494                                 args >> a3;
 495                                 args >> a4;
 496                                 if(!args.okdone())
 497                                         return rpc_const::unmarshal_args_failure;
 498                                 int b = (sob->*meth)(a1, a2, a3, a4, r);
 499                                 ret << r;
 500                                 return b;
 501                         }
 502         };
 503         reg1(proc, new h1(sob, meth));
 504 }
 505
 506 template<class S, class A1, class A2, class A3, class A4, class A5, class R> void
 507 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 508                         const A3 a3, const A4 a4,
 509                         const A5 a5, R & r))
 510 {
 511         class h1 : public handler {
 512                 private:
 513                         S * sob;
 514                         int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4,
 515                                         const A5 a5, R & r);
 516                 public:
 517                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3,
 518                                                 const A4 a4, const A5 a5, R & r))
 519                                 : sob(xsob), meth(xmeth) { }
 520                         int fn(unmarshall &args, marshall &ret) {
 521                                 A1 a1;
 522                                 A2 a2;
 523                                 A3 a3;
 524                                 A4 a4;
 525                                 A5 a5;
 526                                 R r;
 527                                 args >> a1;
 528                                 args >> a2;
 529                                 args >> a3;
 530                                 args >> a4;
 531                                 args >> a5;
 532                                 if(!args.okdone())
 533                                         return rpc_const::unmarshal_args_failure;
 534                                 int b = (sob->*meth)(a1, a2, a3, a4, a5, r);
 535                                 ret << r;
 536                                 return b;
 537                         }
 538         };
 539         reg1(proc, new h1(sob, meth));
 540 }
 541
 542 template<class S, class A1, class A2, class A3, class A4, class A5, class A6, class R> void
 543 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 544                         const A3 a3, const A4 a4,
 545                         const A5 a5, const A6 a6,
 546                         R & r))
 547 {
 548         class h1 : public handler {
 549                 private:
 550                         S * sob;
 551                         int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4,
 552                                         const A5 a5, const A6 a6, R & r);
 553                 public:
 554                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3,
 555                                                 const A4 a4, const A5 a5, const A6 a6, R & r))
 556                                 : sob(xsob), meth(xmeth) { }
 557                         int fn(unmarshall &args, marshall &ret) {
 558                                 A1 a1;
 559                                 A2 a2;
 560                                 A3 a3;
 561                                 A4 a4;
 562                                 A5 a5;
 563                                 A6 a6;
 564                                 R r;
 565                                 args >> a1;
 566                                 args >> a2;
 567                                 args >> a3;
 568                                 args >> a4;
 569                                 args >> a5;
 570                                 args >> a6;
 571                                 if(!args.okdone())
 572                                         return rpc_const::unmarshal_args_failure;
 573                                 int b = (sob->*meth)(a1, a2, a3, a4, a5, a6, r);
 574                                 ret << r;
 575                                 return b;
 576                         }
 577         };
 578         reg1(proc, new h1(sob, meth));
 579 }
 580
 581 template<class S, class A1, class A2, class A3, class A4, class A5,
 582         class A6, class A7, class R> void
 583 rpcs::reg(unsigned int proc, S*sob, int (S::*meth)(const A1 a1, const A2 a2,
 584                         const A3 a3, const A4 a4,
 585                         const A5 a5, const A6 a6,
 586                         const A7 a7, R & r))
 587 {
 588         class h1 : public handler {
 589                 private:
 590                         S * sob;
 591                         int (S::*meth)(const A1 a1, const A2 a2, const A3 a3, const A4 a4,
 592                                         const A5 a5, const A6 a6, const A7 a7, R & r);
 593                 public:
 594                         h1(S *xsob, int (S::*xmeth)(const A1 a1, const A2 a2, const A3 a3,
 595                                                 const A4 a4, const A5 a5, const A6 a6,
 596                                                 const A7 a7, R & r))
 597                                 : sob(xsob), meth(xmeth) { }
 598                         int fn(unmarshall &args, marshall &ret) {
 599                                 A1 a1;
 600                                 A2 a2;
 601                                 A3 a3;
 602                                 A4 a4;
 603                                 A5 a5;
 604                                 A6 a6;
 605                                 A7 a7;
 606                                 R r;
 607                                 args >> a1;
 608                                 args >> a2;
 609                                 args >> a3;
 610                                 args >> a4;
 611                                 args >> a5;
 612                                 args >> a6;
 613                                 args >> a7;
 614                                 if(!args.okdone())
 615                                         return rpc_const::unmarshal_args_failure;
 616                                 int b = (sob->*meth)(a1, a2, a3, a4, a5, a6, a7, r);
 617                                 ret << r;
 618                                 return b;
 619                         }
 620         };
 621         reg1(proc, new h1(sob, meth));
 622 }
 623
 624
 625 void make_sockaddr(const char *hostandport, struct sockaddr_in *dst);
 626 void make_sockaddr(const char *host, const char *port,
 627                 struct sockaddr_in *dst);
 628
 629 int cmp_timespec(const struct timespec &a, const struct timespec &b);
 630 void add_timespec(const struct timespec &a, int b, struct timespec *result);
 631 int diff_timespec(const struct timespec &a, const struct timespec &b);
 632
 633 #endif