rpc/rpctest.cc

   1 // RPC test and pseudo-documentation.
   2 // generates print statements on failures, but eventually says "rpctest OK"
   3
   4 #include "types.h"
   5 #include "rpc.h"
   6 #include <arpa/inet.h>
   7 #include <getopt.h>
   8 #include <unistd.h>
   9 #include <string.h>
  10
  11 #define NUM_CL 2
  12
  13 char log_thread_prefix = 'r';
  14
  15 static rpcs *server;  // server rpc object
  16 static rpcc *clients[NUM_CL];  // client rpc object
  17 static string dst; //server's ip address
  18 static in_port_t port;
  19
  20 // server-side handlers. they must be methods of some class
  21 // to simplify rpcs::reg(). a server process can have handlers
  22 // from multiple classes.
  23 class srv {
  24     public:
  25         int handle_22(string & r, const string a, const string b);
  26         int handle_fast(int & r, const int a);
  27         int handle_slow(int & r, const int a);
  28         int handle_bigrep(string & r, const size_t a);
  29 };
  30
  31 namespace srv_protocol {
  32     using status = rpc_protocol::status;
  33     REMOTE_PROCEDURE_BASE(0);
  34     REMOTE_PROCEDURE(22, _22, (string &, string, string));
  35     REMOTE_PROCEDURE(23, fast, (int &, int));
  36     REMOTE_PROCEDURE(24, slow, (int &, int));
  37     REMOTE_PROCEDURE(25, bigrep, (string &, size_t));
  38 };
  39
  40 // a handler. a and b are arguments, r is the result.
  41 // there can be multiple arguments but only one result.
  42 // the caller also gets to see the int return value
  43 // as the return value from rpcc::call().
  44 // rpcs::reg() decides how to unmarshall by looking
  45 // at these argument types, so this function definition
  46 // does what a .x file does in SunRPC.
  47 int srv::handle_22(string & r, const string a, string b) {
  48     r = a + b;
  49     return 0;
  50 }
  51
  52 int srv::handle_fast(int & r, const int a) {
  53     r = a + 1;
  54     return 0;
  55 }
  56
  57 int srv::handle_slow(int & r, const int a) {
  58     usleep(random() % 500);
  59     r = a + 2;
  60     return 0;
  61 }
  62
  63 int srv::handle_bigrep(string & r, const size_t len) {
  64     r = string(len, 'x');
  65     return 0;
  66 }
  67
  68 static srv service;
  69
  70 void startserver() {
  71     server = new rpcs(port);
  72     server->reg(srv_protocol::_22, &srv::handle_22, &service);
  73     server->reg(srv_protocol::fast, &srv::handle_fast, &service);
  74     server->reg(srv_protocol::slow, &srv::handle_slow, &service);
  75     server->reg(srv_protocol::bigrep, &srv::handle_bigrep, &service);
  76     server->start();
  77 }
  78
  79 void testmarshall() {
  80     marshall m;
  81     rpc_protocol::request_header rh{1,2,3,4,5};
  82     m.pack_header(rh);
  83     VERIFY(((string)m).size()==rpc_protocol::RPC_HEADER_SZ);
  84     int i = 12345;
  85     unsigned long long l = 1223344455L;
  86     size_t sz = 101010101;
  87     string s = "hallo....";
  88     m << i;
  89     m << l;
  90     m << s;
  91     m << sz;
  92
  93     string b = m;
  94     VERIFY(b.size() == rpc_protocol::RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int)+sizeof(uint32_t));
  95
  96     unmarshall un(b, true);
  97     rpc_protocol::request_header rh1;
  98     un.unpack_header(rh1);
  99     VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0);
 100     int i1;
 101     unsigned long long l1;
 102     string s1;
 103     size_t sz1;
 104     un >> i1;
 105     un >> l1;
 106     un >> s1;
 107     un >> sz1;
 108     VERIFY(un.okdone());
 109     VERIFY(i1==i && l1==l && s1==s && sz1==sz);
 110 }
 111
 112 void client1(size_t cl) {
 113     // test concurrency.
 114     size_t which_cl = cl % NUM_CL;
 115
 116     for(int i = 0; i < 100; i++){
 117         unsigned long arg = (random() % 2000);
 118         string rep;
 119         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
 120         VERIFY(ret == 0);
 121         if ((unsigned long)rep.size()!=arg)
 122             cout << "repsize wrong " << rep.size() << "!=" << arg << endl;
 123         VERIFY((unsigned long)rep.size() == arg);
 124     }
 125
 126     // test rpc replies coming back not in the order of
 127     // the original calls -- i.e. does xid reply dispatch work.
 128     for(int i = 0; i < 100; i++){
 129         int which = (random() % 2);
 130         int arg = (random() % 1000);
 131         int rep;
 132
 133         auto start = steady_clock::now();
 134
 135         int ret = clients[which_cl]->call(which ? srv_protocol::fast : srv_protocol::slow, rep, arg);
 136         auto end = steady_clock::now();
 137         auto diff = duration_cast<milliseconds>(end - start).count();
 138         if (ret != 0)
 139             cout << diff << " ms have elapsed!!!" << endl;
 140         VERIFY(ret == 0);
 141         VERIFY(rep == (which ? arg+1 : arg+2));
 142     }
 143 }
 144
 145 void client2(size_t cl) {
 146     size_t which_cl = cl % NUM_CL;
 147
 148     time_t t1;
 149     time(&t1);
 150
 151     while(time(0) - t1 < 10){
 152         unsigned long arg = (random() % 2000);
 153         string rep;
 154         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
 155         if ((unsigned long)rep.size()!=arg)
 156             cout << "ask for " << arg << " reply got " << rep.size() << " ret " << ret << endl;
 157         VERIFY((unsigned long)rep.size() == arg);
 158     }
 159 }
 160
 161 void client3(void *xx) {
 162     rpcc *c = (rpcc *) xx;
 163
 164     for(int i = 0; i < 4; i++){
 165         int rep = 0;
 166         int ret = c->call_timeout(srv_protocol::slow, milliseconds(300), rep, i);
 167         VERIFY(ret == rpc_protocol::timeout_failure || rep == i+2);
 168     }
 169 }
 170
 171 void simple_tests(rpcc *c) {
 172     cout << "simple_tests" << endl;
 173     // an RPC call to procedure #22.
 174     // rpcc::call() looks at the argument types to decide how
 175     // to marshall the RPC call packet, and how to unmarshall
 176     // the reply packet.
 177     string rep;
 178     int intret = c->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
 179     VERIFY(intret == 0); // this is what handle_22 returns
 180     VERIFY(rep == "hello goodbye");
 181     cout << "   -- string concat RPC .. ok" << endl;
 182
 183     // small request, big reply (perhaps req via UDP, reply via TCP)
 184     intret = c->call_timeout(srv_protocol::bigrep, milliseconds(20000), rep, 70000ul);
 185     VERIFY(intret == 0);
 186     VERIFY(rep.size() == 70000);
 187     cout << "   -- small request, big reply .. ok" << endl;
 188
 189     // specify a timeout value to an RPC that should succeed (udp)
 190     int xx = 0;
 191     intret = c->call_timeout(srv_protocol::fast, milliseconds(300), xx, 77);
 192     VERIFY(intret == 0 && xx == 78);
 193     cout << "   -- no spurious timeout .. ok" << endl;
 194
 195     // specify a timeout value to an RPC that should succeed (tcp)
 196     {
 197         string arg(1000, 'x');
 198         string rep2;
 199         c->call_timeout(srv_protocol::_22, milliseconds(300), rep2, arg, (string)"x");
 200         VERIFY(rep2.size() == 1001);
 201         cout << "   -- no spurious timeout .. ok" << endl;
 202     }
 203
 204     // huge RPC
 205     string big(1000000, 'x');
 206     intret = c->call(srv_protocol::_22, rep, big, (string)"z");
 207     VERIFY(intret == 0);
 208     VERIFY(rep.size() == 1000001);
 209     cout << "   -- huge 1M rpc request .. ok" << endl;
 210
 211     // specify a timeout value to an RPC that should timeout (udp)
 212     string non_existent = "127.0.0.1:7661";
 213     rpcc *c1 = new rpcc(non_existent);
 214     time_t t0 = time(0);
 215     intret = c1->bind(milliseconds(300));
 216     time_t t1 = time(0);
 217     VERIFY(intret < 0 && (t1 - t0) <= 4);
 218     cout << "   -- rpc timeout .. ok" << endl;
 219     cout << "simple_tests OK" << endl;
 220 }
 221
 222 void concurrent_test(size_t nt) {
 223     // create threads that make lots of calls in parallel,
 224     // to test thread synchronization for concurrent calls
 225     // and dispatches.
 226     cout << "start concurrent_test (" << nt << " threads) ...";
 227
 228     vector<thread> th(nt);
 229
 230     for(size_t i = 0; i < nt; i++)
 231         th[i] = thread(client1, i);
 232
 233     for(size_t i = 0; i < nt; i++)
 234         th[i].join();
 235
 236     cout << " OK" << endl;
 237 }
 238
 239 void lossy_test() {
 240     cout << "start lossy_test ...";
 241     VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
 242
 243     if (server) {
 244         delete server;
 245         startserver();
 246     }
 247
 248     for (int i = 0; i < NUM_CL; i++) {
 249         delete clients[i];
 250         clients[i] = new rpcc(dst);
 251         VERIFY(clients[i]->bind()==0);
 252     }
 253
 254     size_t nt = 1;
 255
 256     vector<thread> th(nt);
 257
 258     for(size_t i = 0; i < nt; i++)
 259         th[i] = thread(client2, i);
 260
 261     for(size_t i = 0; i < nt; i++)
 262         th[i].join();
 263
 264     cout << ".. OK" << endl;
 265     VERIFY(setenv("RPC_LOSSY", "0", 1) == 0);
 266 }
 267
 268 void failure_test() {
 269     rpcc *client1;
 270     rpcc *client = clients[0];
 271
 272     cout << "failure_test" << endl;
 273
 274     delete server;
 275
 276     client1 = new rpcc(dst);
 277     VERIFY (client1->bind(milliseconds(3000)) < 0);
 278     cout << "   -- create new client and try to bind to failed server .. failed ok" << endl;
 279
 280     delete client1;
 281
 282     startserver();
 283
 284     string rep;
 285     int intret = client->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
 286     VERIFY(intret == rpc_protocol::oldsrv_failure);
 287     cout << "   -- call recovered server with old client .. failed ok" << endl;
 288
 289     delete client;
 290
 291     clients[0] = client = new rpcc(dst);
 292     VERIFY (client->bind() >= 0);
 293     VERIFY (client->bind() < 0);
 294
 295     intret = client->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
 296     VERIFY(intret == 0);
 297     VERIFY(rep == "hello goodbye");
 298
 299     cout << "   -- delete existing rpc client, create replacement rpc client .. ok" << endl;
 300
 301
 302     size_t nt = 10;
 303     cout << "   -- concurrent test on new rpc client w/ " << nt << " threads ..";
 304
 305     vector<thread> th(nt);
 306
 307     for(size_t i = 0; i < nt; i++)
 308         th[i] = thread(client3, client);
 309
 310     for(size_t i = 0; i < nt; i++)
 311         th[i].join();
 312
 313     cout << "ok" << endl;
 314
 315     delete server;
 316     delete client;
 317
 318     startserver();
 319     clients[0] = client = new rpcc(dst);
 320     VERIFY (client->bind() >= 0);
 321     cout << "   -- delete existing rpc client and server, create replacements.. ok" << endl;
 322
 323     cout << "   -- concurrent test on new client and server w/ " << nt << " threads ..";
 324
 325     for(size_t i = 0; i < nt; i++)
 326         th[i] = thread(client3, client);
 327
 328     for(size_t i = 0; i < nt; i++)
 329         th[i].join();
 330
 331     cout << "ok" << endl;
 332
 333     cout << "failure_test OK" << endl;
 334 }
 335
 336 int main(int argc, char *argv[]) {
 337
 338     setvbuf(stdout, NULL, _IONBF, 0);
 339     setvbuf(stderr, NULL, _IONBF, 0);
 340     int debug_level = 0;
 341
 342     bool isclient = false;
 343     bool isserver = false;
 344
 345     srandom((uint32_t)getpid());
 346     port = 20000 + (getpid() % 10000);
 347
 348     int ch = 0;
 349     while ((ch = getopt(argc, argv, "csd:p:l"))!=-1) {
 350         switch (ch) {
 351             case 'c':
 352                 isclient = true;
 353                 break;
 354             case 's':
 355                 isserver = true;
 356                 break;
 357             case 'd':
 358                 debug_level = atoi(optarg);
 359                 break;
 360             case 'p':
 361                 port = (in_port_t)atoi(optarg);
 362                 break;
 363             case 'l':
 364                 VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
 365                 break;
 366             default:
 367                 break;
 368         }
 369     }
 370
 371     if (!isserver && !isclient)  {
 372         isserver = isclient = true;
 373     }
 374
 375     if (debug_level > 0) {
 376         DEBUG_LEVEL = debug_level;
 377         IF_LEVEL(1) LOG_NONMEMBER("DEBUG LEVEL: " << debug_level);
 378     }
 379
 380     testmarshall();
 381
 382     if (isserver) {
 383         cout << "starting server on port " << port << " RPC_HEADER_SZ " << (int)rpc_protocol::RPC_HEADER_SZ << endl;
 384         startserver();
 385     }
 386
 387     if (isclient) {
 388         // server's address.
 389         dst = "127.0.0.1:" + to_string(port);
 390
 391
 392         // start the client.  bind it to the server.
 393         // starts a thread to listen for replies and hand them to
 394         // the correct waiting caller thread. there should probably
 395         // be only one rpcc per process. you probably need one
 396         // rpcc per server.
 397         for (int i = 0; i < NUM_CL; i++) {
 398             clients[i] = new rpcc(dst);
 399             VERIFY (clients[i]->bind() == 0);
 400         }
 401
 402         simple_tests(clients[0]);
 403         concurrent_test(10);
 404         lossy_test();
 405         if (isserver) {
 406             failure_test();
 407         }
 408
 409         cout << "rpctest OK" << endl;
 410
 411         exit(0);
 412     }
 413
 414     while (1)
 415         usleep(100000);
 416 }