X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5fd8cc8409d0efadc07dfe8d6774ad9ff477663d..2546a41ad36fdc9ef6471cb35a1d56930ae1b527:/rpc/rpctest.cc?ds=inline diff --git a/rpc/rpctest.cc b/rpc/rpctest.cc index 74c61d1..bf8a56c 100644 --- a/rpc/rpctest.cc +++ b/rpc/rpctest.cc @@ -3,31 +3,40 @@ #include "rpc.h" #include -#include +#include +#include +#include #include -#include #include +#include +#include #include "jsl_log.h" -#include "gettime.h" #include "lang/verify.h" #define NUM_CL 2 +char log_thread_prefix = 'r'; + +using std::string; +using std::cout; +using std::endl; +using std::vector; +using std::thread; + rpcs *server; // server rpc object rpcc *clients[NUM_CL]; // client rpc object -struct sockaddr_in dst; //server's ip address +string dst; //server's ip address int port; -pthread_attr_t attr; // server-side handlers. they must be methods of some class // to simplify rpcs::reg(). a server process can have handlers // from multiple classes. class srv { public: - int handle_22(const std::string a, const std::string b, std::string & r); - int handle_fast(const int a, int &r); - int handle_slow(const int a, int &r); - int handle_bigrep(const int a, std::string &r); + int handle_22(string & r, const string a, const string b); + int handle_fast(int &r, const int a); + int handle_slow(int &r, const int a); + int handle_bigrep(string &r, const size_t a); }; // a handler. a and b are arguments, r is the result. @@ -38,21 +47,21 @@ class srv { // at these argument types, so this function definition // does what a .x file does in SunRPC. int -srv::handle_22(const std::string a, std::string b, std::string &r) +srv::handle_22(string &r, const string a, string b) { r = a + b; return 0; } int -srv::handle_fast(const int a, int &r) +srv::handle_fast(int &r, const int a) { r = a + 1; return 0; } int -srv::handle_slow(const int a, int &r) +srv::handle_slow(int &r, const int a) { usleep(random() % 5000); r = a + 2; @@ -60,9 +69,9 @@ srv::handle_slow(const int a, int &r) } int -srv::handle_bigrep(const int len, std::string &r) +srv::handle_bigrep(string &r, const size_t len) { - r = std::string(len, 'x'); + r = string((size_t)len, 'x'); return 0; } @@ -70,39 +79,39 @@ srv service; void startserver() { - server = new rpcs(port); - server->reg(22, &service, &srv::handle_22); - server->reg(23, &service, &srv::handle_fast); - server->reg(24, &service, &srv::handle_slow); - server->reg(25, &service, &srv::handle_bigrep); + server = new rpcs((unsigned int)port); + server->reg(22, &srv::handle_22, &service); + server->reg(23, &srv::handle_fast, &service); + server->reg(24, &srv::handle_slow, &service); + server->reg(25, &srv::handle_bigrep, &service); } void testmarshall() { marshall m; - req_header rh(1,2,3,4,5); + request_header rh{1,2,3,4,5}; m.pack_req_header(rh); VERIFY(m.size()==RPC_HEADER_SZ); int i = 12345; unsigned long long l = 1223344455L; - std::string s = std::string("hallo...."); + string s = "hallo...."; m << i; m << l; m << s; char *b; - int sz; + size_t sz; m.take_buf(&b,&sz); - VERIFY(sz == (int)(RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int))); + VERIFY(sz == RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int)); unmarshall un(b,sz); - req_header rh1; + request_header rh1; un.unpack_req_header(&rh1); VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0); int i1; unsigned long long l1; - std::string s1; + string s1; un >> i1; un >> l1; un >> s1; @@ -110,21 +119,19 @@ testmarshall() VERIFY(i1==i && l1==l && s1==s); } -void * -client1(void *xx) +void +client1(size_t cl) { - // test concurrency. - int which_cl = ((unsigned long) xx ) % NUM_CL; + size_t which_cl = cl % NUM_CL; for(int i = 0; i < 100; i++){ int arg = (random() % 2000); - std::string rep; - int ret = clients[which_cl]->call(25, arg, rep); + string rep; + int ret = clients[which_cl]->call(25, rep, arg); VERIFY(ret == 0); - if ((int)rep.size()!=arg) { - printf("repsize wrong %d!=%d\n", (int)rep.size(), arg); - } + if ((int)rep.size()!=arg) + cout << "repsize wrong " << rep.size() << "!=" << arg << endl; VERIFY((int)rep.size() == arg); } @@ -135,158 +142,124 @@ client1(void *xx) int arg = (random() % 1000); int rep; - struct timespec start,end; - clock_gettime(CLOCK_REALTIME, &start); + auto start = std::chrono::steady_clock::now(); - int ret = clients[which_cl]->call(which ? 23 : 24, arg, rep); - clock_gettime(CLOCK_REALTIME, &end); - int diff = diff_timespec(end, start); + int ret = clients[which_cl]->call(which ? 23 : 24, rep, arg); + auto end = std::chrono::steady_clock::now(); + auto diff = std::chrono::duration_cast(end - start).count(); if (ret != 0) - printf("%d ms have elapsed!!!\n", diff); + cout << diff << " ms have elapsed!!!" << endl; VERIFY(ret == 0); VERIFY(rep == (which ? arg+1 : arg+2)); } - - return 0; } -void * -client2(void *xx) +void +client2(size_t cl) { - int which_cl = ((unsigned long) xx ) % NUM_CL; + size_t which_cl = cl % NUM_CL; time_t t1; time(&t1); while(time(0) - t1 < 10){ int arg = (random() % 2000); - std::string rep; - int ret = clients[which_cl]->call(25, arg, rep); - if ((int)rep.size()!=arg) { - printf("ask for %d reply got %d ret %d\n", - arg, (int)rep.size(), ret); - } + string rep; + int ret = clients[which_cl]->call(25, rep, arg); + if ((int)rep.size()!=arg) + cout << "ask for " << arg << " reply got " << rep.size() << " ret " << ret << endl; VERIFY((int)rep.size() == arg); } - return 0; } -void * +void client3(void *xx) { rpcc *c = (rpcc *) xx; for(int i = 0; i < 4; i++){ int rep; - int ret = c->call(24, i, rep, rpcc::to(3000)); + int ret = c->call_timeout(24, rpcc::to(3000), rep, i); VERIFY(ret == rpc_const::timeout_failure || rep == i+2); } - return 0; } void simple_tests(rpcc *c) { - printf("simple_tests\n"); + cout << "simple_tests" << endl; // an RPC call to procedure #22. // rpcc::call() looks at the argument types to decide how // to marshall the RPC call packet, and how to unmarshall // the reply packet. - std::string rep; - int intret = c->call(22, (std::string)"hello", (std::string)" goodbye", rep); + string rep; + int intret = c->call(22, rep, (string)"hello", (string)" goodbye"); VERIFY(intret == 0); // this is what handle_22 returns VERIFY(rep == "hello goodbye"); - printf(" -- string concat RPC .. ok\n"); + cout << " -- string concat RPC .. ok" << endl; // small request, big reply (perhaps req via UDP, reply via TCP) - intret = c->call(25, 70000, rep, rpcc::to(200000)); + intret = c->call_timeout(25, rpcc::to(200000), rep, 70000); VERIFY(intret == 0); VERIFY(rep.size() == 70000); - printf(" -- small request, big reply .. ok\n"); - -#if 0 - // too few arguments - intret = c->call(22, (std::string)"just one", rep); - VERIFY(intret < 0); - printf(" -- too few arguments .. failed ok\n"); - - // too many arguments; proc #23 expects just one. - intret = c->call(23, 1001, 1002, rep); - VERIFY(intret < 0); - printf(" -- too many arguments .. failed ok\n"); - - // wrong return value size - int wrongrep; - intret = c->call(23, (std::string)"hello", (std::string)" goodbye", wrongrep); - VERIFY(intret < 0); - printf(" -- wrong ret value size .. failed ok\n"); -#endif + cout << " -- small request, big reply .. ok" << endl; // specify a timeout value to an RPC that should succeed (udp) int xx = 0; - intret = c->call(23, 77, xx, rpcc::to(3000)); + intret = c->call_timeout(23, rpcc::to(3000), xx, 77); VERIFY(intret == 0 && xx == 78); - printf(" -- no suprious timeout .. ok\n"); + cout << " -- no spurious timeout .. ok" << endl; // specify a timeout value to an RPC that should succeed (tcp) { - std::string arg(1000, 'x'); - std::string rep; - c->call(22, arg, (std::string)"x", rep, rpcc::to(3000)); - VERIFY(rep.size() == 1001); - printf(" -- no suprious timeout .. ok\n"); + string arg(1000, 'x'); + string rep2; + c->call_timeout(22, rpcc::to(3000), rep2, arg, (string)"x"); + VERIFY(rep2.size() == 1001); + cout << " -- no spurious timeout .. ok" << endl; } // huge RPC - std::string big(1000000, 'x'); - intret = c->call(22, big, (std::string)"z", rep); + string big(1000000, 'x'); + intret = c->call(22, rep, big, (string)"z"); VERIFY(rep.size() == 1000001); - printf(" -- huge 1M rpc request .. ok\n"); + cout << " -- huge 1M rpc request .. ok" << endl; // specify a timeout value to an RPC that should timeout (udp) - struct sockaddr_in non_existent; - memset(&non_existent, 0, sizeof(non_existent)); - non_existent.sin_family = AF_INET; - non_existent.sin_addr.s_addr = inet_addr("127.0.0.1"); - non_existent.sin_port = htons(7661); + string non_existent = "127.0.0.1:7661"; rpcc *c1 = new rpcc(non_existent); time_t t0 = time(0); intret = c1->bind(rpcc::to(3000)); time_t t1 = time(0); VERIFY(intret < 0 && (t1 - t0) <= 4); - printf(" -- rpc timeout .. ok\n"); - printf("simple_tests OK\n"); + cout << " -- rpc timeout .. ok" << endl; + cout << "simple_tests OK" << endl; } void -concurrent_test(int nt) +concurrent_test(size_t nt) { // create threads that make lots of calls in parallel, // to test thread synchronization for concurrent calls // and dispatches. - int ret; + cout << "start concurrent_test (" << nt << " threads) ..."; - printf("start concurrent_test (%d threads) ...", nt); + vector th(nt); - pthread_t th[nt]; - for(int i = 0; i < nt; i++){ - ret = pthread_create(&th[i], &attr, client1, (void *) (uintptr_t)i); - VERIFY(ret == 0); - } + for(size_t i = 0; i < nt; i++) + th[i] = thread(client1, i); - for(int i = 0; i < nt; i++){ - VERIFY(pthread_join(th[i], NULL) == 0); - } - printf(" OK\n"); + for(size_t i = 0; i < nt; i++) + th[i].join(); + + cout << " OK" << endl; } void lossy_test() { - int ret; - - printf("start lossy_test ..."); + cout << "start lossy_test ..."; VERIFY(setenv("RPC_LOSSY", "5", 1) == 0); if (server) { @@ -300,16 +273,17 @@ lossy_test() VERIFY(clients[i]->bind()==0); } - int nt = 1; - pthread_t th[nt]; - for(int i = 0; i < nt; i++){ - ret = pthread_create(&th[i], &attr, client2, (void *) (uintptr_t)i); - VERIFY(ret == 0); - } - for(int i = 0; i < nt; i++){ - VERIFY(pthread_join(th[i], NULL) == 0); - } - printf(".. OK\n"); + size_t nt = 1; + + vector th(nt); + + for(size_t i = 0; i < nt; i++) + th[i] = thread(client2, i); + + for(size_t i = 0; i < nt; i++) + th[i].join(); + + cout << ".. OK" << endl; VERIFY(setenv("RPC_LOSSY", "0", 1) == 0); } @@ -319,22 +293,22 @@ failure_test() rpcc *client1; rpcc *client = clients[0]; - printf("failure_test\n"); + cout << "failure_test" << endl; delete server; client1 = new rpcc(dst); VERIFY (client1->bind(rpcc::to(3000)) < 0); - printf(" -- create new client and try to bind to failed server .. failed ok\n"); + cout << " -- create new client and try to bind to failed server .. failed ok" << endl; delete client1; startserver(); - std::string rep; - int intret = client->call(22, (std::string)"hello", (std::string)" goodbye", rep); + string rep; + int intret = client->call(22, rep, (string)"hello", (string)" goodbye"); VERIFY(intret == rpc_const::oldsrv_failure); - printf(" -- call recovered server with old client .. failed ok\n"); + cout << " -- call recovered server with old client .. failed ok" << endl; delete client; @@ -342,27 +316,25 @@ failure_test() VERIFY (client->bind() >= 0); VERIFY (client->bind() < 0); - intret = client->call(22, (std::string)"hello", (std::string)" goodbye", rep); + intret = client->call(22, rep, (string)"hello", (string)" goodbye"); VERIFY(intret == 0); VERIFY(rep == "hello goodbye"); - printf(" -- delete existing rpc client, create replacement rpc client .. ok\n"); + cout << " -- delete existing rpc client, create replacement rpc client .. ok" << endl; - int nt = 10; - int ret; - printf(" -- concurrent test on new rpc client w/ %d threads ..", nt); + size_t nt = 10; + cout << " -- concurrent test on new rpc client w/ " << nt << " threads .."; - pthread_t th[nt]; - for(int i = 0; i < nt; i++){ - ret = pthread_create(&th[i], &attr, client3, (void *) client); - VERIFY(ret == 0); - } + vector th(nt); - for(int i = 0; i < nt; i++){ - VERIFY(pthread_join(th[i], NULL) == 0); - } - printf("ok\n"); + for(size_t i = 0; i < nt; i++) + th[i] = thread(client3, client); + + for(size_t i = 0; i < nt; i++) + th[i].join(); + + cout << "ok" << endl; delete server; delete client; @@ -370,20 +342,19 @@ failure_test() startserver(); clients[0] = client = new rpcc(dst); VERIFY (client->bind() >= 0); - printf(" -- delete existing rpc client and server, create replacements.. ok\n"); + cout << " -- delete existing rpc client and server, create replacements.. ok" << endl; - printf(" -- concurrent test on new client and server w/ %d threads ..", nt); - for(int i = 0; i < nt; i++){ - ret = pthread_create(&th[i], &attr, client3, (void *)client); - VERIFY(ret == 0); - } + cout << " -- concurrent test on new client and server w/ " << nt << " threads .."; - for(int i = 0; i < nt; i++){ - VERIFY(pthread_join(th[i], NULL) == 0); - } - printf("ok\n"); + for(size_t i = 0; i < nt; i++) + th[i] = thread(client3, client); + + for(size_t i = 0; i < nt; i++) + th[i].join(); - printf("failure_test OK\n"); + cout << "ok" << endl; + + cout << "failure_test OK" << endl; } int @@ -397,10 +368,10 @@ main(int argc, char *argv[]) bool isclient = false; bool isserver = false; - srandom(getpid()); + srandom((uint32_t)getpid()); port = 20000 + (getpid() % 10000); - char ch = 0; + int ch = 0; while ((ch = getopt(argc, argv, "csd:p:l"))!=-1) { switch (ch) { case 'c': @@ -417,6 +388,7 @@ main(int argc, char *argv[]) break; case 'l': VERIFY(setenv("RPC_LOSSY", "5", 1) == 0); + break; default: break; } @@ -427,28 +399,20 @@ main(int argc, char *argv[]) } if (debug_level > 0) { - //__loginit.initNow(); - jsl_set_debug(debug_level); + JSL_DEBUG_LEVEL = debug_level; jsl_log(JSL_DBG_1, "DEBUG LEVEL: %d\n", debug_level); } testmarshall(); - pthread_attr_init(&attr); - // set stack size to 32K, so we don't run out of memory - pthread_attr_setstacksize(&attr, 32*1024); - if (isserver) { - printf("starting server on port %d RPC_HEADER_SZ %d\n", port, RPC_HEADER_SZ); + cout << "starting server on port " << port << " RPC_HEADER_SZ " << (int)RPC_HEADER_SZ << endl; startserver(); } if (isclient) { // server's address. - memset(&dst, 0, sizeof(dst)); - dst.sin_family = AF_INET; - dst.sin_addr.s_addr = inet_addr("127.0.0.1"); - dst.sin_port = htons(port); + dst = "127.0.0.1:" + std::to_string(port); // start the client. bind it to the server. @@ -468,7 +432,7 @@ main(int argc, char *argv[]) failure_test(); } - printf("rpctest OK\n"); + cout << "rpctest OK" << endl; exit(0); }