Raw
1 /*
2 * test-simple-ipc.c: verify that the Inter-Process Communication works.
3 */
4
5 #include "test-tool.h"
6 #include "gettext.h"
7 #include "simple-ipc.h"
8 #include "parse-options.h"
9 #include "thread-utils.h"
10 #include "strvec.h"
11 #include "run-command.h"
12 #include "trace2.h"
13
14 #ifndef SUPPORTS_SIMPLE_IPC
15 int cmd__simple_ipc(int argc, const char **argv)
16 {
17 die("simple IPC not available on this platform");
18 }
19 #else
20
21 /*
22 * The test daemon defines an "application callback" that supports a
23 * series of commands (see `test_app_cb()`).
24 *
25 * Unknown commands are caught here and we send an error message back
26 * to the client process.
27 */
28 static int app__unhandled_command(const char *command,
29 ipc_server_reply_cb *reply_cb,
30 struct ipc_server_reply_data *reply_data)
31 {
32 struct strbuf buf = STRBUF_INIT;
33 int ret;
34
35 strbuf_addf(&buf, "unhandled command: %s", command);
36 ret = reply_cb(reply_data, buf.buf, buf.len);
37 strbuf_release(&buf);
38
39 return ret;
40 }
41
42 /*
43 * Reply with a single very large buffer. This is to ensure that
44 * long response are properly handled -- whether the chunking occurs
45 * in the kernel or in the (probably pkt-line) layer.
46 */
47 #define BIG_ROWS (10000)
48 static int app__big_command(ipc_server_reply_cb *reply_cb,
49 struct ipc_server_reply_data *reply_data)
50 {
51 struct strbuf buf = STRBUF_INIT;
52 int row;
53 int ret;
54
55 for (row = 0; row < BIG_ROWS; row++)
56 strbuf_addf(&buf, "big: %.75d\n", row);
57
58 ret = reply_cb(reply_data, buf.buf, buf.len);
59 strbuf_release(&buf);
60
61 return ret;
62 }
63
64 /*
65 * Reply with a series of lines. This is to ensure that we can incrementally
66 * compute the response and chunk it to the client.
67 */
68 #define CHUNK_ROWS (10000)
69 static int app__chunk_command(ipc_server_reply_cb *reply_cb,
70 struct ipc_server_reply_data *reply_data)
71 {
72 struct strbuf buf = STRBUF_INIT;
73 int row;
74 int ret;
75
76 for (row = 0; row < CHUNK_ROWS; row++) {
77 strbuf_setlen(&buf, 0);
78 strbuf_addf(&buf, "big: %.75d\n", row);
79 ret = reply_cb(reply_data, buf.buf, buf.len);
80 }
81
82 strbuf_release(&buf);
83
84 return ret;
85 }
86
87 /*
88 * Slowly reply with a series of lines. This is to model an expensive to
89 * compute chunked response (which might happen if this callback is running
90 * in a thread and is fighting for a lock with other threads).
91 */
92 #define SLOW_ROWS (1000)
93 #define SLOW_DELAY_MS (10)
94 static int app__slow_command(ipc_server_reply_cb *reply_cb,
95 struct ipc_server_reply_data *reply_data)
96 {
97 struct strbuf buf = STRBUF_INIT;
98 int row;
99 int ret;
100
101 for (row = 0; row < SLOW_ROWS; row++) {
102 strbuf_setlen(&buf, 0);
103 strbuf_addf(&buf, "big: %.75d\n", row);
104 ret = reply_cb(reply_data, buf.buf, buf.len);
105 sleep_millisec(SLOW_DELAY_MS);
106 }
107
108 strbuf_release(&buf);
109
110 return ret;
111 }
112
113 /*
114 * The client sent a command followed by a (possibly very) large buffer.
115 */
116 static int app__sendbytes_command(const char *received, size_t received_len,
117 ipc_server_reply_cb *reply_cb,
118 struct ipc_server_reply_data *reply_data)
119 {
120 struct strbuf buf_resp = STRBUF_INIT;
121 const char *p = "?";
122 int len_ballast = 0;
123 int k;
124 int errs = 0;
125 int ret;
126
127 /*
128 * The test is setup to send:
129 * "sendbytes" SP <n * char>
130 */
131 if (received_len < strlen("sendbytes "))
132 BUG("received_len is short in app__sendbytes_command");
133
134 if (skip_prefix(received, "sendbytes ", &p))
135 len_ballast = strlen(p);
136
137 /*
138 * Verify that the ballast is n copies of a single letter.
139 * And that the multi-threaded IO layer didn't cross the streams.
140 */
141 for (k = 1; k < len_ballast; k++)
142 if (p[k] != p[0])
143 errs++;
144
145 if (errs)
146 strbuf_addf(&buf_resp, "errs:%d\n", errs);
147 else
148 strbuf_addf(&buf_resp, "rcvd:%c%08d\n", p[0], len_ballast);
149
150 ret = reply_cb(reply_data, buf_resp.buf, buf_resp.len);
151
152 strbuf_release(&buf_resp);
153
154 return ret;
155 }
156
157 /*
158 * An arbitrary fixed address to verify that the application instance
159 * data is handled properly.
160 */
161 static int my_app_data = 42;
162
163 static ipc_server_application_cb test_app_cb;
164
165 /*
166 * This is the "application callback" that sits on top of the
167 * "ipc-server". It completely defines the set of commands supported
168 * by this application.
169 */
170 static int test_app_cb(void *application_data,
171 const char *command, size_t command_len,
172 ipc_server_reply_cb *reply_cb,
173 struct ipc_server_reply_data *reply_data)
174 {
175 /*
176 * Verify that we received the application-data that we passed
177 * when we started the ipc-server. (We have several layers of
178 * callbacks calling callbacks and it's easy to get things mixed
179 * up (especially when some are "void*").)
180 */
181 if (application_data != (void*)&my_app_data)
182 BUG("application_cb: application_data pointer wrong");
183
184 if (command_len == 4 && !strncmp(command, "quit", 4)) {
185 /*
186 * The client sent a "quit" command. This is an async
187 * request for the server to shutdown.
188 *
189 * We DO NOT send the client a response message
190 * (because we have nothing to say and the other
191 * server threads have not yet stopped).
192 *
193 * Tell the ipc-server layer to start shutting down.
194 * This includes: stop listening for new connections
195 * on the socket/pipe and telling all worker threads
196 * to finish/drain their outgoing responses to other
197 * clients.
198 *
199 * This DOES NOT force an immediate sync shutdown.
200 */
201 return SIMPLE_IPC_QUIT;
202 }
203
204 if (command_len == 4 && !strncmp(command, "ping", 4)) {
205 const char *answer = "pong";
206 return reply_cb(reply_data, answer, strlen(answer));
207 }
208
209 if (command_len == 3 && !strncmp(command, "big", 3))
210 return app__big_command(reply_cb, reply_data);
211
212 if (command_len == 5 && !strncmp(command, "chunk", 5))
213 return app__chunk_command(reply_cb, reply_data);
214
215 if (command_len == 4 && !strncmp(command, "slow", 4))
216 return app__slow_command(reply_cb, reply_data);
217
218 if (command_len >= 10 && starts_with(command, "sendbytes "))
219 return app__sendbytes_command(command, command_len,
220 reply_cb, reply_data);
221
222 return app__unhandled_command(command, reply_cb, reply_data);
223 }
224
225 struct cl_args
226 {
227 const char *subcommand;
228 const char *path;
229 const char *token;
230
231 int nr_threads;
232 int max_wait_sec;
233 int bytecount;
234 int batchsize;
235
236 char bytevalue;
237 };
238
239 static struct cl_args cl_args = {
240 .subcommand = NULL,
241 .path = "ipc-test",
242 .token = NULL,
243
244 .nr_threads = 5,
245 .max_wait_sec = 60,
246 .bytecount = 1024,
247 .batchsize = 10,
248
249 .bytevalue = 'x',
250 };
251
252 /*
253 * This process will run as a simple-ipc server and listen for IPC commands
254 * from client processes.
255 */
256 static int daemon__run_server(void)
257 {
258 int ret;
259
260 struct ipc_server_opts opts = {
261 .nr_threads = cl_args.nr_threads,
262 };
263
264 /*
265 * Synchronously run the ipc-server. We don't need any application
266 * instance data, so pass an arbitrary pointer (that we'll later
267 * verify made the round trip).
268 */
269 ret = ipc_server_run(cl_args.path, &opts, test_app_cb, (void*)&my_app_data);
270 if (ret == -2)
271 error("socket/pipe already in use: '%s'", cl_args.path);
272 else if (ret == -1)
273 error_errno("could not start server on: '%s'", cl_args.path);
274
275 return ret;
276 }
277
278 static start_bg_wait_cb bg_wait_cb;
279
280 static int bg_wait_cb(const struct child_process *cp UNUSED,
281 void *cb_data UNUSED)
282 {
283 int s = ipc_get_active_state(cl_args.path);
284
285 switch (s) {
286 case IPC_STATE__LISTENING:
287 /* child is "ready" */
288 return 0;
289
290 case IPC_STATE__NOT_LISTENING:
291 case IPC_STATE__PATH_NOT_FOUND:
292 /* give child more time */
293 return 1;
294
295 default:
296 case IPC_STATE__INVALID_PATH:
297 case IPC_STATE__OTHER_ERROR:
298 /* all the time in world won't help */
299 return -1;
300 }
301 }
302
303 static int daemon__start_server(void)
304 {
305 struct child_process cp = CHILD_PROCESS_INIT;
306 enum start_bg_result sbgr;
307
308 strvec_push(&cp.args, "test-tool");
309 strvec_push(&cp.args, "simple-ipc");
310 strvec_push(&cp.args, "run-daemon");
311 strvec_pushf(&cp.args, "--name=%s", cl_args.path);
312 strvec_pushf(&cp.args, "--threads=%d", cl_args.nr_threads);
313
314 cp.no_stdin = 1;
315 cp.no_stdout = 1;
316 cp.no_stderr = 1;
317
318 sbgr = start_bg_command(&cp, bg_wait_cb, NULL, cl_args.max_wait_sec);
319
320 switch (sbgr) {
321 case SBGR_READY:
322 return 0;
323
324 default:
325 case SBGR_ERROR:
326 case SBGR_CB_ERROR:
327 return error("daemon failed to start");
328
329 case SBGR_TIMEOUT:
330 return error("daemon not online yet");
331
332 case SBGR_DIED:
333 return error("daemon terminated");
334 }
335 }
336
337 /*
338 * This process will run a quick probe to see if a simple-ipc server
339 * is active on this path.
340 *
341 * Returns 0 if the server is alive.
342 */
343 static int client__probe_server(void)
344 {
345 enum ipc_active_state s;
346
347 s = ipc_get_active_state(cl_args.path);
348 switch (s) {
349 case IPC_STATE__LISTENING:
350 return 0;
351
352 case IPC_STATE__NOT_LISTENING:
353 return error("no server listening at '%s'", cl_args.path);
354
355 case IPC_STATE__PATH_NOT_FOUND:
356 return error("path not found '%s'", cl_args.path);
357
358 case IPC_STATE__INVALID_PATH:
359 return error("invalid pipe/socket name '%s'", cl_args.path);
360
361 case IPC_STATE__OTHER_ERROR:
362 default:
363 return error("other error for '%s'", cl_args.path);
364 }
365 }
366
367 /*
368 * Send an IPC command token to an already-running server daemon and
369 * print the response.
370 *
371 * This is a simple 1 word command/token that `test_app_cb()` (in the
372 * daemon process) will understand.
373 */
374 static int client__send_ipc(void)
375 {
376 const char *command = "(no-command)";
377 struct strbuf buf = STRBUF_INIT;
378 struct ipc_client_connect_options options
379 = IPC_CLIENT_CONNECT_OPTIONS_INIT;
380
381 if (cl_args.token && *cl_args.token)
382 command = cl_args.token;
383
384 options.wait_if_busy = 1;
385 options.wait_if_not_found = 0;
386
387 if (!ipc_client_send_command(cl_args.path, &options,
388 command, strlen(command),
389 &buf)) {
390 if (buf.len) {
391 printf("%s\n", buf.buf);
392 fflush(stdout);
393 }
394 strbuf_release(&buf);
395
396 return 0;
397 }
398
399 return error("failed to send '%s' to '%s'", command, cl_args.path);
400 }
401
402 /*
403 * Send an IPC command to an already-running server and ask it to
404 * shutdown. "send quit" is an async request and queues a shutdown
405 * event in the server, so we spin and wait here for it to actually
406 * shutdown to make the unit tests a little easier to write.
407 */
408 static int client__stop_server(void)
409 {
410 int ret;
411 time_t time_limit, now;
412 enum ipc_active_state s;
413
414 time(&time_limit);
415 time_limit += cl_args.max_wait_sec;
416
417 cl_args.token = "quit";
418
419 ret = client__send_ipc();
420 if (ret)
421 return ret;
422
423 for (;;) {
424 sleep_millisec(100);
425
426 s = ipc_get_active_state(cl_args.path);
427
428 if (s != IPC_STATE__LISTENING) {
429 /*
430 * The socket/pipe is gone and/or has stopped
431 * responding. Lets assume that the daemon
432 * process has exited too.
433 */
434 return 0;
435 }
436
437 time(&now);
438 if (now > time_limit)
439 return error("daemon has not shutdown yet");
440 }
441 }
442
443 /*
444 * Send an IPC command followed by ballast to confirm that a large
445 * message can be sent and that the kernel or pkt-line layers will
446 * properly chunk it and that the daemon receives the entire message.
447 */
448 static int do_sendbytes(int bytecount, char byte, const char *path,
449 const struct ipc_client_connect_options *options)
450 {
451 struct strbuf buf_send = STRBUF_INIT;
452 struct strbuf buf_resp = STRBUF_INIT;
453
454 strbuf_addstr(&buf_send, "sendbytes ");
455 strbuf_addchars(&buf_send, byte, bytecount);
456
457 if (!ipc_client_send_command(path, options,
458 buf_send.buf, buf_send.len,
459 &buf_resp)) {
460 strbuf_rtrim(&buf_resp);
461 printf("sent:%c%08d %s\n", byte, bytecount, buf_resp.buf);
462 fflush(stdout);
463 strbuf_release(&buf_send);
464 strbuf_release(&buf_resp);
465
466 return 0;
467 }
468
469 return error("client failed to sendbytes(%d, '%c') to '%s'",
470 bytecount, byte, path);
471 }
472
473 /*
474 * Send an IPC command with ballast to an already-running server daemon.
475 */
476 static int client__sendbytes(void)
477 {
478 struct ipc_client_connect_options options
479 = IPC_CLIENT_CONNECT_OPTIONS_INIT;
480
481 options.wait_if_busy = 1;
482 options.wait_if_not_found = 0;
483 options.uds_disallow_chdir = 0;
484
485 return do_sendbytes(cl_args.bytecount, cl_args.bytevalue, cl_args.path,
486 &options);
487 }
488
489 struct multiple_thread_data {
490 pthread_t pthread_id;
491 struct multiple_thread_data *next;
492 const char *path;
493 int bytecount;
494 int batchsize;
495 int sum_errors;
496 int sum_good;
497 char letter;
498 };
499
500 static void *multiple_thread_proc(void *_multiple_thread_data)
501 {
502 struct multiple_thread_data *d = _multiple_thread_data;
503 int k;
504 struct ipc_client_connect_options options
505 = IPC_CLIENT_CONNECT_OPTIONS_INIT;
506
507 options.wait_if_busy = 1;
508 options.wait_if_not_found = 0;
509 /*
510 * A multi-threaded client should not be randomly calling chdir().
511 * The test will pass without this restriction because the test is
512 * not otherwise accessing the filesystem, but it makes us honest.
513 */
514 options.uds_disallow_chdir = 1;
515
516 trace2_thread_start("multiple");
517
518 for (k = 0; k < d->batchsize; k++) {
519 if (do_sendbytes(d->bytecount + k, d->letter, d->path, &options))
520 d->sum_errors++;
521 else
522 d->sum_good++;
523 }
524
525 trace2_thread_exit();
526 return NULL;
527 }
528
529 /*
530 * Start a client-side thread pool. Each thread sends a series of
531 * IPC requests. Each request is on a new connection to the server.
532 */
533 static int client__multiple(void)
534 {
535 struct multiple_thread_data *list = NULL;
536 int k;
537 int sum_join_errors = 0;
538 int sum_thread_errors = 0;
539 int sum_good = 0;
540
541 for (k = 0; k < cl_args.nr_threads; k++) {
542 struct multiple_thread_data *d = xcalloc(1, sizeof(*d));
543 d->next = list;
544 d->path = cl_args.path;
545 d->bytecount = cl_args.bytecount + cl_args.batchsize*(k/26);
546 d->batchsize = cl_args.batchsize;
547 d->sum_errors = 0;
548 d->sum_good = 0;
549 d->letter = 'A' + (k % 26);
550
551 if (pthread_create(&d->pthread_id, NULL, multiple_thread_proc, d)) {
552 warning("failed to create thread[%d] skipping remainder", k);
553 free(d);
554 break;
555 }
556
557 list = d;
558 }
559
560 while (list) {
561 struct multiple_thread_data *d = list;
562
563 if (pthread_join(d->pthread_id, NULL))
564 sum_join_errors++;
565
566 sum_thread_errors += d->sum_errors;
567 sum_good += d->sum_good;
568
569 list = d->next;
570 free(d);
571 }
572
573 printf("client (good %d) (join %d), (errors %d)\n",
574 sum_good, sum_join_errors, sum_thread_errors);
575
576 return (sum_join_errors + sum_thread_errors) ? 1 : 0;
577 }
578
579 int cmd__simple_ipc(int argc, const char **argv)
580 {
581 const char * const simple_ipc_usage[] = {
582 N_("test-helper simple-ipc is-active [<name>] [<options>]"),
583 N_("test-helper simple-ipc run-daemon [<name>] [<threads>]"),
584 N_("test-helper simple-ipc start-daemon [<name>] [<threads>] [<max-wait>]"),
585 N_("test-helper simple-ipc stop-daemon [<name>] [<max-wait>]"),
586 N_("test-helper simple-ipc send [<name>] [<token>]"),
587 N_("test-helper simple-ipc sendbytes [<name>] [<bytecount>] [<byte>]"),
588 N_("test-helper simple-ipc multiple [<name>] [<threads>] [<bytecount>] [<batchsize>]"),
589 NULL
590 };
591
592 const char *bytevalue = NULL;
593
594 struct option options[] = {
595 #ifndef GIT_WINDOWS_NATIVE
596 OPT_STRING(0, "name", &cl_args.path, N_("name"), N_("name or pathname of unix domain socket")),
597 #else
598 OPT_STRING(0, "name", &cl_args.path, N_("name"), N_("named-pipe name")),
599 #endif
600 OPT_INTEGER(0, "threads", &cl_args.nr_threads, N_("number of threads in server thread pool")),
601 OPT_INTEGER(0, "max-wait", &cl_args.max_wait_sec, N_("seconds to wait for daemon to start or stop")),
602
603 OPT_INTEGER(0, "bytecount", &cl_args.bytecount, N_("number of bytes")),
604 OPT_INTEGER(0, "batchsize", &cl_args.batchsize, N_("number of requests per thread")),
605
606 /*
607 * The "byte" string here is not marked for translation and
608 * instead relies on translation in strbuf.c:humanise_bytes() to
609 * avoid conflict with the plural form.
610 */
611 OPT_STRING(0, "byte", &bytevalue, "byte", N_("ballast character")),
612 OPT_STRING(0, "token", &cl_args.token, N_("token"), N_("command token to send to the server")),
613
614 OPT_END()
615 };
616
617 if (argc < 2)
618 usage_with_options(simple_ipc_usage, options);
619
620 show_usage_with_options_if_asked(argc, argv,
621 simple_ipc_usage, options);
622
623 if (argc == 2 && !strcmp(argv[1], "SUPPORTS_SIMPLE_IPC"))
624 return 0;
625
626 cl_args.subcommand = argv[1];
627
628 argc--;
629 argv++;
630
631 argc = parse_options(argc, argv, NULL, options, simple_ipc_usage, 0);
632
633 if (cl_args.nr_threads < 1)
634 cl_args.nr_threads = 1;
635 if (cl_args.max_wait_sec < 0)
636 cl_args.max_wait_sec = 0;
637 if (cl_args.bytecount < 1)
638 cl_args.bytecount = 1;
639 if (cl_args.batchsize < 1)
640 cl_args.batchsize = 1;
641
642 if (bytevalue && *bytevalue)
643 cl_args.bytevalue = bytevalue[0];
644
645 /*
646 * Use '!!' on all dispatch functions to map from `error()` style
647 * (returns -1) style to `test_must_fail` style (expects 1). This
648 * makes shell error messages less confusing.
649 */
650
651 if (!strcmp(cl_args.subcommand, "is-active"))
652 return !!client__probe_server();
653
654 if (!strcmp(cl_args.subcommand, "run-daemon"))
655 return !!daemon__run_server();
656
657 if (!strcmp(cl_args.subcommand, "start-daemon"))
658 return !!daemon__start_server();
659
660 /*
661 * Client commands follow. Ensure a server is running before
662 * sending any data. This might be overkill, but then again
663 * this is a test harness.
664 */
665
666 if (!strcmp(cl_args.subcommand, "stop-daemon")) {
667 if (client__probe_server())
668 return 1;
669 return !!client__stop_server();
670 }
671
672 if (!strcmp(cl_args.subcommand, "send")) {
673 if (client__probe_server())
674 return 1;
675 return !!client__send_ipc();
676 }
677
678 if (!strcmp(cl_args.subcommand, "sendbytes")) {
679 if (client__probe_server())
680 return 1;
681 return !!client__sendbytes();
682 }
683
684 if (!strcmp(cl_args.subcommand, "multiple")) {
685 if (client__probe_server())
686 return 1;
687 return !!client__multiple();
688 }
689
690 die("Unhandled subcommand: '%s'", cl_args.subcommand);
691 }
692 #endif