1 // RPC test and pseudo-documentation.
2 // generates print statements on failures, but eventually says "rpctest OK"
13 char log_thread_prefix = 'r';
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;
23 // server-side handlers. they must be methods of some class
24 // to simplify rpcs::reg(). a server process can have handlers
25 // from multiple classes.
28 int handle_22(string & r, const string a, const string b);
29 int handle_fast(int & r, const int a);
30 int handle_slow(int & r, const int a);
31 int handle_bigrep(string & r, const size_t a);
34 namespace srv_protocol {
35 using status = rpc_protocol::status;
36 REMOTE_PROCEDURE_BASE(0);
37 REMOTE_PROCEDURE(22, _22, (string &, string, string));
38 REMOTE_PROCEDURE(23, fast, (int &, int));
39 REMOTE_PROCEDURE(24, slow, (int &, int));
40 REMOTE_PROCEDURE(25, bigrep, (string &, size_t));
43 // a handler. a and b are arguments, r is the result.
44 // there can be multiple arguments but only one result.
45 // the caller also gets to see the int return value
46 // as the return value from rpcc::call().
47 // rpcs::reg() decides how to unmarshall by looking
48 // at these argument types, so this function definition
49 // does what a .x file does in SunRPC.
50 int srv::handle_22(string & r, const string a, string b) {
55 int srv::handle_fast(int & r, const int a) {
60 int srv::handle_slow(int & r, const int a) {
61 usleep(random() % 500);
66 int srv::handle_bigrep(string & r, const size_t len) {
73 static void startserver() {
74 server = new rpcs(port);
75 server->reg(srv_protocol::_22, &srv::handle_22, &service);
76 server->reg(srv_protocol::fast, &srv::handle_fast, &service);
77 server->reg(srv_protocol::slow, &srv::handle_slow, &service);
78 server->reg(srv_protocol::bigrep, &srv::handle_bigrep, &service);
82 static void testmarshall() {
84 rpc_protocol::request_header rh{1,2,3,4,5};
86 VERIFY(((string)m).size()==rpc_protocol::RPC_HEADER_SZ);
88 unsigned long long l = 1223344455L;
89 size_t sz = 101010101;
90 string s = "hallo....";
91 string bin("\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x7f\xe5", 12);
99 VERIFY(b.size() == rpc_protocol::RPC_HEADER_SZ+sizeof(i)+sizeof(l)+sizeof(uint32_t)+s.size()+sizeof(uint32_t)+sizeof(uint32_t)+bin.size());
101 unmarshall un(b, true);
102 rpc_protocol::request_header rh1;
104 VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0);
106 unsigned long long l1;
116 VERIFY(i1==i && l1==l && s1==s && sz1==sz && bin1==bin);
119 static void client1(size_t cl) {
121 size_t which_cl = cl % NUM_CL;
123 for(int i = 0; i < 100; i++){
124 unsigned long arg = (random() % 2000);
126 int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
128 if ((unsigned long)rep.size()!=arg)
129 cout << "repsize wrong " << rep.size() << "!=" << arg << endl;
130 VERIFY((unsigned long)rep.size() == arg);
133 // test rpc replies coming back not in the order of
134 // the original calls -- i.e. does xid reply dispatch work.
135 for(int i = 0; i < 100; i++){
136 int which = (random() % 2);
137 int arg = (random() % 1000);
140 auto start = steady_clock::now();
142 int ret = clients[which_cl]->call(which ? srv_protocol::fast : srv_protocol::slow, rep, arg);
143 auto end = steady_clock::now();
144 auto diff = duration_cast<milliseconds>(end - start).count();
146 cout << diff << " ms have elapsed!!!" << endl;
148 VERIFY(rep == (which ? arg+1 : arg+2));
152 static void client2(size_t cl) {
153 size_t which_cl = cl % NUM_CL;
158 while(time(0) - t1 < 10){
159 unsigned long arg = (random() % 2000);
161 int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
162 if ((unsigned long)rep.size()!=arg)
163 cout << "ask for " << arg << " reply got " << rep.size() << " ret " << ret << endl;
164 VERIFY((unsigned long)rep.size() == arg);
168 static void client3(void *xx) {
169 rpcc *c = (rpcc *) xx;
171 for(int i = 0; i < 4; i++){
173 int ret = c->call_timeout(srv_protocol::slow, milliseconds(300), rep, i);
174 VERIFY(ret == rpc_protocol::timeout_failure || rep == i+2);
178 static void simple_tests(rpcc *c) {
179 cout << "simple_tests" << endl;
180 // an RPC call to procedure #22.
181 // rpcc::call() looks at the argument types to decide how
182 // to marshall the RPC call packet, and how to unmarshall
185 int intret = c->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
186 VERIFY(intret == 0); // this is what handle_22 returns
187 VERIFY(rep == "hello goodbye");
188 cout << " -- string concat RPC .. ok" << endl;
190 // small request, big reply (perhaps req via UDP, reply via TCP)
191 intret = c->call_timeout(srv_protocol::bigrep, milliseconds(20000), rep, 70000ul);
193 VERIFY(rep.size() == 70000);
194 cout << " -- small request, big reply .. ok" << endl;
196 // specify a timeout value to an RPC that should succeed (udp)
198 intret = c->call_timeout(srv_protocol::fast, milliseconds(300), xx, 77);
199 VERIFY(intret == 0 && xx == 78);
200 cout << " -- no spurious timeout .. ok" << endl;
202 // specify a timeout value to an RPC that should succeed (tcp)
204 string arg(1000, 'x');
206 c->call_timeout(srv_protocol::_22, milliseconds(300), rep2, arg, (string)"x");
207 VERIFY(rep2.size() == 1001);
208 cout << " -- no spurious timeout .. ok" << endl;
212 string big(1000000, 'x');
213 intret = c->call(srv_protocol::_22, rep, big, (string)"z");
215 VERIFY(rep.size() == 1000001);
216 cout << " -- huge 1M rpc request .. ok" << endl;
218 // specify a timeout value to an RPC that should timeout (udp)
219 string non_existent = "127.0.0.1:7661";
220 rpcc *c1 = new rpcc(non_existent);
222 intret = c1->bind(milliseconds(300));
224 VERIFY(intret < 0 && (t1 - t0) <= 4);
225 cout << " -- rpc timeout .. ok" << endl;
226 cout << "simple_tests OK" << endl;
229 static void concurrent_test(size_t nt) {
230 // create threads that make lots of calls in parallel,
231 // to test thread synchronization for concurrent calls
233 cout << "start concurrent_test (" << nt << " threads) ...";
235 vector<thread> th(nt);
237 for(size_t i = 0; i < nt; i++)
238 th[i] = thread(client1, i);
240 for(size_t i = 0; i < nt; i++)
243 cout << " OK" << endl;
246 static void lossy_test() {
247 cout << "start lossy_test ...";
248 VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
255 for (int i = 0; i < NUM_CL; i++) {
257 clients[i] = new rpcc(dst);
258 VERIFY(clients[i]->bind()==0);
263 vector<thread> th(nt);
265 for(size_t i = 0; i < nt; i++)
266 th[i] = thread(client2, i);
268 for(size_t i = 0; i < nt; i++)
271 cout << ".. OK" << endl;
272 VERIFY(setenv("RPC_LOSSY", "0", 1) == 0);
275 static void failure_test() {
277 rpcc *client = clients[0];
279 cout << "failure_test" << endl;
283 client1 = new rpcc(dst);
284 VERIFY (client1->bind(milliseconds(3000)) < 0);
285 cout << " -- create new client and try to bind to failed server .. failed ok" << endl;
292 int intret = client->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
293 VERIFY(intret == rpc_protocol::oldsrv_failure);
294 cout << " -- call recovered server with old client .. failed ok" << endl;
298 clients[0] = client = new rpcc(dst);
299 VERIFY (client->bind() >= 0);
300 VERIFY (client->bind() < 0);
302 intret = client->call(srv_protocol::_22, rep, (string)"hello", (string)" goodbye");
304 VERIFY(rep == "hello goodbye");
306 cout << " -- delete existing rpc client, create replacement rpc client .. ok" << endl;
310 cout << " -- concurrent test on new rpc client w/ " << nt << " threads ..";
312 vector<thread> th(nt);
314 for(size_t i = 0; i < nt; i++)
315 th[i] = thread(client3, client);
317 for(size_t i = 0; i < nt; i++)
320 cout << "ok" << endl;
326 clients[0] = client = new rpcc(dst);
327 VERIFY (client->bind() >= 0);
328 cout << " -- delete existing rpc client and server, create replacements.. ok" << endl;
330 cout << " -- concurrent test on new client and server w/ " << nt << " threads ..";
332 for(size_t i = 0; i < nt; i++)
333 th[i] = thread(client3, client);
335 for(size_t i = 0; i < nt; i++)
338 cout << "ok" << endl;
340 cout << "failure_test OK" << endl;
343 int main(int argc, char *argv[]) {
345 setvbuf(stdout, NULL, _IONBF, 0);
346 setvbuf(stderr, NULL, _IONBF, 0);
349 bool isclient = false;
350 bool isserver = false;
352 srandom((uint32_t)getpid());
353 port = 20000 + (getpid() % 10000);
356 while ((ch = getopt(argc, argv, "csd:p:l"))!=-1) {
365 debug_level = atoi(optarg);
368 port = (in_port_t)atoi(optarg);
371 VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
378 if (!isserver && !isclient) {
379 isserver = isclient = true;
382 if (debug_level > 0) {
383 DEBUG_LEVEL = debug_level;
384 IF_LEVEL(1) LOG_NONMEMBER << "DEBUG LEVEL: " << debug_level;
390 cout << "starting server on port " << port << " RPC_HEADER_SZ " << (int)rpc_protocol::RPC_HEADER_SZ << endl;
396 dst = "127.0.0.1:" + to_string(port);
399 // start the client. bind it to the server.
400 // starts a thread to listen for replies and hand them to
401 // the correct waiting caller thread. there should probably
402 // be only one rpcc per process. you probably need one
404 for (int i = 0; i < NUM_CL; i++) {
405 clients[i] = new rpcc(dst);
406 VERIFY (clients[i]->bind() == 0);
409 simple_tests(clients[0]);
416 cout << "rpctest OK" << endl;