[thirdparty/git.git] / t / helper / test-simple-ipc.c

/*
 * test-simple-ipc.c: verify that the Inter-Process Communication works.
 */

#include "test-tool.h"
#include "gettext.h"
#include "strbuf.h"
#include "simple-ipc.h"
#include "parse-options.h"
#include "thread-utils.h"
#include "strvec.h"
#include "run-command.h"
#include "trace2.h"

#ifndef SUPPORTS_SIMPLE_IPC
int cmd__simple_ipc(int argc, const char **argv)
{
	die("simple IPC not available on this platform");
}
#else

/*
 * The test daemon defines an "application callback" that supports a
 * series of commands (see `test_app_cb()`).
 *
 * Unknown commands are caught here and we send an error message back
 * to the client process.
 */
static int app__unhandled_command(const char *command,
				  ipc_server_reply_cb *reply_cb,
				  struct ipc_server_reply_data *reply_data)
{
	struct strbuf buf = STRBUF_INIT;
	int ret;

	strbuf_addf(&buf, "unhandled command: %s", command);
	ret = reply_cb(reply_data, buf.buf, buf.len);
	strbuf_release(&buf);

	return ret;
}

/*
 * Reply with a single very large buffer.  This is to ensure that
 * long response are properly handled -- whether the chunking occurs
 * in the kernel or in the (probably pkt-line) layer.
 */
#define BIG_ROWS (10000)
static int app__big_command(ipc_server_reply_cb *reply_cb,
			    struct ipc_server_reply_data *reply_data)
{
	struct strbuf buf = STRBUF_INIT;
	int row;
	int ret;

	for (row = 0; row < BIG_ROWS; row++)
		strbuf_addf(&buf, "big: %.75d\n", row);

	ret = reply_cb(reply_data, buf.buf, buf.len);
	strbuf_release(&buf);

	return ret;
}

/*
 * Reply with a series of lines.  This is to ensure that we can incrementally
 * compute the response and chunk it to the client.
 */
#define CHUNK_ROWS (10000)
static int app__chunk_command(ipc_server_reply_cb *reply_cb,
			      struct ipc_server_reply_data *reply_data)
{
	struct strbuf buf = STRBUF_INIT;
	int row;
	int ret;

	for (row = 0; row < CHUNK_ROWS; row++) {
		strbuf_setlen(&buf, 0);
		strbuf_addf(&buf, "big: %.75d\n", row);
		ret = reply_cb(reply_data, buf.buf, buf.len);
	}

	strbuf_release(&buf);

	return ret;
}

/*
 * Slowly reply with a series of lines.  This is to model an expensive to
 * compute chunked response (which might happen if this callback is running
 * in a thread and is fighting for a lock with other threads).
 */
#define SLOW_ROWS     (1000)
#define SLOW_DELAY_MS (10)
static int app__slow_command(ipc_server_reply_cb *reply_cb,
			     struct ipc_server_reply_data *reply_data)
{
	struct strbuf buf = STRBUF_INIT;
	int row;
	int ret;

	for (row = 0; row < SLOW_ROWS; row++) {
		strbuf_setlen(&buf, 0);
		strbuf_addf(&buf, "big: %.75d\n", row);
		ret = reply_cb(reply_data, buf.buf, buf.len);
		sleep_millisec(SLOW_DELAY_MS);
	}

	strbuf_release(&buf);

	return ret;
}

/*
 * The client sent a command followed by a (possibly very) large buffer.
 */
static int app__sendbytes_command(const char *received, size_t received_len,
				  ipc_server_reply_cb *reply_cb,
				  struct ipc_server_reply_data *reply_data)
{
	struct strbuf buf_resp = STRBUF_INIT;
	const char *p = "?";
	int len_ballast = 0;
	int k;
	int errs = 0;
	int ret;

	/*
	 * The test is setup to send:
	 *     "sendbytes" SP <n * char>
	 */
	if (received_len < strlen("sendbytes "))
		BUG("received_len is short in app__sendbytes_command");

	if (skip_prefix(received, "sendbytes ", &p))
		len_ballast = strlen(p);

	/*
	 * Verify that the ballast is n copies of a single letter.
	 * And that the multi-threaded IO layer didn't cross the streams.
	 */
	for (k = 1; k < len_ballast; k++)
		if (p[k] != p[0])
			errs++;

	if (errs)
		strbuf_addf(&buf_resp, "errs:%d\n", errs);
	else
		strbuf_addf(&buf_resp, "rcvd:%c%08d\n", p[0], len_ballast);

	ret = reply_cb(reply_data, buf_resp.buf, buf_resp.len);

	strbuf_release(&buf_resp);

	return ret;
}

/*
 * An arbitrary fixed address to verify that the application instance
 * data is handled properly.
 */
static int my_app_data = 42;

static ipc_server_application_cb test_app_cb;

/*
 * This is the "application callback" that sits on top of the
 * "ipc-server".  It completely defines the set of commands supported
 * by this application.
 */
static int test_app_cb(void *application_data,
		       const char *command, size_t command_len,
		       ipc_server_reply_cb *reply_cb,
		       struct ipc_server_reply_data *reply_data)
{
	/*
	 * Verify that we received the application-data that we passed
	 * when we started the ipc-server.  (We have several layers of
	 * callbacks calling callbacks and it's easy to get things mixed
	 * up (especially when some are "void*").)
	 */
	if (application_data != (void*)&my_app_data)
		BUG("application_cb: application_data pointer wrong");

	if (command_len == 4 && !strncmp(command, "quit", 4)) {
		/*
		 * The client sent a "quit" command.  This is an async
		 * request for the server to shutdown.
		 *
		 * We DO NOT send the client a response message
		 * (because we have nothing to say and the other
		 * server threads have not yet stopped).
		 *
		 * Tell the ipc-server layer to start shutting down.
		 * This includes: stop listening for new connections
		 * on the socket/pipe and telling all worker threads
		 * to finish/drain their outgoing responses to other
		 * clients.
		 *
		 * This DOES NOT force an immediate sync shutdown.
		 */
		return SIMPLE_IPC_QUIT;
	}

	if (command_len == 4 && !strncmp(command, "ping", 4)) {
		const char *answer = "pong";
		return reply_cb(reply_data, answer, strlen(answer));
	}

	if (command_len == 3 && !strncmp(command, "big", 3))
		return app__big_command(reply_cb, reply_data);

	if (command_len == 5 && !strncmp(command, "chunk", 5))
		return app__chunk_command(reply_cb, reply_data);

	if (command_len == 4 && !strncmp(command, "slow", 4))
		return app__slow_command(reply_cb, reply_data);

	if (command_len >= 10 && starts_with(command, "sendbytes "))
		return app__sendbytes_command(command, command_len,
					      reply_cb, reply_data);

	return app__unhandled_command(command, reply_cb, reply_data);
}

struct cl_args
{
	const char *subcommand;
	const char *path;
	const char *token;

	int nr_threads;
	int max_wait_sec;
	int bytecount;
	int batchsize;

	char bytevalue;
};

static struct cl_args cl_args = {
	.subcommand = NULL,
	.path = "ipc-test",
	.token = NULL,

	.nr_threads = 5,
	.max_wait_sec = 60,
	.bytecount = 1024,
	.batchsize = 10,

	.bytevalue = 'x',
};

/*
 * This process will run as a simple-ipc server and listen for IPC commands
 * from client processes.
 */
static int daemon__run_server(void)
{
	int ret;

	struct ipc_server_opts opts = {
		.nr_threads = cl_args.nr_threads,
	};

	/*
	 * Synchronously run the ipc-server.  We don't need any application
	 * instance data, so pass an arbitrary pointer (that we'll later
	 * verify made the round trip).
	 */
	ret = ipc_server_run(cl_args.path, &opts, test_app_cb, (void*)&my_app_data);
	if (ret == -2)
		error("socket/pipe already in use: '%s'", cl_args.path);
	else if (ret == -1)
		error_errno("could not start server on: '%s'", cl_args.path);

	return ret;
}

static start_bg_wait_cb bg_wait_cb;

static int bg_wait_cb(const struct child_process *cp, void *cb_data)
{
	int s = ipc_get_active_state(cl_args.path);

	switch (s) {
	case IPC_STATE__LISTENING:
		/* child is "ready" */
		return 0;

	case IPC_STATE__NOT_LISTENING:
	case IPC_STATE__PATH_NOT_FOUND:
		/* give child more time */
		return 1;

	default:
	case IPC_STATE__INVALID_PATH:
	case IPC_STATE__OTHER_ERROR:
		/* all the time in world won't help */
		return -1;
	}
}

static int daemon__start_server(void)
{
	struct child_process cp = CHILD_PROCESS_INIT;
	enum start_bg_result sbgr;

	strvec_push(&cp.args, "test-tool");
	strvec_push(&cp.args, "simple-ipc");
	strvec_push(&cp.args, "run-daemon");
	strvec_pushf(&cp.args, "--name=%s", cl_args.path);
	strvec_pushf(&cp.args, "--threads=%d", cl_args.nr_threads);

	cp.no_stdin = 1;
	cp.no_stdout = 1;
	cp.no_stderr = 1;

	sbgr = start_bg_command(&cp, bg_wait_cb, NULL, cl_args.max_wait_sec);

	switch (sbgr) {
	case SBGR_READY:
		return 0;

	default:
	case SBGR_ERROR:
	case SBGR_CB_ERROR:
		return error("daemon failed to start");

	case SBGR_TIMEOUT:
		return error("daemon not online yet");

	case SBGR_DIED:
		return error("daemon terminated");
	}
}

/*
 * This process will run a quick probe to see if a simple-ipc server
 * is active on this path.
 *
 * Returns 0 if the server is alive.
 */
static int client__probe_server(void)
{
	enum ipc_active_state s;

	s = ipc_get_active_state(cl_args.path);
	switch (s) {
	case IPC_STATE__LISTENING:
		return 0;

	case IPC_STATE__NOT_LISTENING:
		return error("no server listening at '%s'", cl_args.path);

	case IPC_STATE__PATH_NOT_FOUND:
		return error("path not found '%s'", cl_args.path);

	case IPC_STATE__INVALID_PATH:
		return error("invalid pipe/socket name '%s'", cl_args.path);

	case IPC_STATE__OTHER_ERROR:
	default:
		return error("other error for '%s'", cl_args.path);
	}
}

/*
 * Send an IPC command token to an already-running server daemon and
 * print the response.
 *
 * This is a simple 1 word command/token that `test_app_cb()` (in the
 * daemon process) will understand.
 */
static int client__send_ipc(void)
{
	const char *command = "(no-command)";
	struct strbuf buf = STRBUF_INIT;
	struct ipc_client_connect_options options
		= IPC_CLIENT_CONNECT_OPTIONS_INIT;

	if (cl_args.token && *cl_args.token)
		command = cl_args.token;

	options.wait_if_busy = 1;
	options.wait_if_not_found = 0;

	if (!ipc_client_send_command(cl_args.path, &options,
				     command, strlen(command),
				     &buf)) {
		if (buf.len) {
			printf("%s\n", buf.buf);
			fflush(stdout);
		}
		strbuf_release(&buf);

		return 0;
	}

	return error("failed to send '%s' to '%s'", command, cl_args.path);
}

/*
 * Send an IPC command to an already-running server and ask it to
 * shutdown.  "send quit" is an async request and queues a shutdown
 * event in the server, so we spin and wait here for it to actually
 * shutdown to make the unit tests a little easier to write.
 */
static int client__stop_server(void)
{
	int ret;
	time_t time_limit, now;
	enum ipc_active_state s;

	time(&time_limit);
	time_limit += cl_args.max_wait_sec;

	cl_args.token = "quit";

	ret = client__send_ipc();
	if (ret)
		return ret;

	for (;;) {
		sleep_millisec(100);

		s = ipc_get_active_state(cl_args.path);

		if (s != IPC_STATE__LISTENING) {
			/*
			 * The socket/pipe is gone and/or has stopped
			 * responding.  Lets assume that the daemon
			 * process has exited too.
			 */
			return 0;
		}

		time(&now);
		if (now > time_limit)
			return error("daemon has not shutdown yet");
	}
}

/*
 * Send an IPC command followed by ballast to confirm that a large
 * message can be sent and that the kernel or pkt-line layers will
 * properly chunk it and that the daemon receives the entire message.
 */
static int do_sendbytes(int bytecount, char byte, const char *path,
			const struct ipc_client_connect_options *options)
{
	struct strbuf buf_send = STRBUF_INIT;
	struct strbuf buf_resp = STRBUF_INIT;

	strbuf_addstr(&buf_send, "sendbytes ");
	strbuf_addchars(&buf_send, byte, bytecount);

	if (!ipc_client_send_command(path, options,
				     buf_send.buf, buf_send.len,
				     &buf_resp)) {
		strbuf_rtrim(&buf_resp);
		printf("sent:%c%08d %s\n", byte, bytecount, buf_resp.buf);
		fflush(stdout);
		strbuf_release(&buf_send);
		strbuf_release(&buf_resp);

		return 0;
	}

	return error("client failed to sendbytes(%d, '%c') to '%s'",
		     bytecount, byte, path);
}

/*
 * Send an IPC command with ballast to an already-running server daemon.
 */
static int client__sendbytes(void)
{
	struct ipc_client_connect_options options
		= IPC_CLIENT_CONNECT_OPTIONS_INIT;

	options.wait_if_busy = 1;
	options.wait_if_not_found = 0;
	options.uds_disallow_chdir = 0;

	return do_sendbytes(cl_args.bytecount, cl_args.bytevalue, cl_args.path,
			    &options);
}

struct multiple_thread_data {
	pthread_t pthread_id;
	struct multiple_thread_data *next;
	const char *path;
	int bytecount;
	int batchsize;
	int sum_errors;
	int sum_good;
	char letter;
};

static void *multiple_thread_proc(void *_multiple_thread_data)
{
	struct multiple_thread_data *d = _multiple_thread_data;
	int k;
	struct ipc_client_connect_options options
		= IPC_CLIENT_CONNECT_OPTIONS_INIT;

	options.wait_if_busy = 1;
	options.wait_if_not_found = 0;
	/*
	 * A multi-threaded client should not be randomly calling chdir().
	 * The test will pass without this restriction because the test is
	 * not otherwise accessing the filesystem, but it makes us honest.
	 */
	options.uds_disallow_chdir = 1;

	trace2_thread_start("multiple");

	for (k = 0; k < d->batchsize; k++) {
		if (do_sendbytes(d->bytecount + k, d->letter, d->path, &options))
			d->sum_errors++;
		else
			d->sum_good++;
	}

	trace2_thread_exit();
	return NULL;
}

/*
 * Start a client-side thread pool.  Each thread sends a series of
 * IPC requests.  Each request is on a new connection to the server.
 */
static int client__multiple(void)
{
	struct multiple_thread_data *list = NULL;
	int k;
	int sum_join_errors = 0;
	int sum_thread_errors = 0;
	int sum_good = 0;

	for (k = 0; k < cl_args.nr_threads; k++) {
		struct multiple_thread_data *d = xcalloc(1, sizeof(*d));
		d->next = list;
		d->path = cl_args.path;
		d->bytecount = cl_args.bytecount + cl_args.batchsize*(k/26);
		d->batchsize = cl_args.batchsize;
		d->sum_errors = 0;
		d->sum_good = 0;
		d->letter = 'A' + (k % 26);

		if (pthread_create(&d->pthread_id, NULL, multiple_thread_proc, d)) {
			warning("failed to create thread[%d] skipping remainder", k);
			free(d);
			break;
		}

		list = d;
	}

	while (list) {
		struct multiple_thread_data *d = list;

		if (pthread_join(d->pthread_id, NULL))
			sum_join_errors++;

		sum_thread_errors += d->sum_errors;
		sum_good += d->sum_good;

		list = d->next;
		free(d);
	}

	printf("client (good %d) (join %d), (errors %d)\n",
	       sum_good, sum_join_errors, sum_thread_errors);

	return (sum_join_errors + sum_thread_errors) ? 1 : 0;
}

int cmd__simple_ipc(int argc, const char **argv)
{
	const char * const simple_ipc_usage[] = {
		N_("test-helper simple-ipc is-active    [<name>] [<options>]"),
		N_("test-helper simple-ipc run-daemon   [<name>] [<threads>]"),
		N_("test-helper simple-ipc start-daemon [<name>] [<threads>] [<max-wait>]"),
		N_("test-helper simple-ipc stop-daemon  [<name>] [<max-wait>]"),
		N_("test-helper simple-ipc send         [<name>] [<token>]"),
		N_("test-helper simple-ipc sendbytes    [<name>] [<bytecount>] [<byte>]"),
		N_("test-helper simple-ipc multiple     [<name>] [<threads>] [<bytecount>] [<batchsize>]"),
		NULL
	};

	const char *bytevalue = NULL;

	struct option options[] = {
#ifndef GIT_WINDOWS_NATIVE
		OPT_STRING(0, "name", &cl_args.path, N_("name"), N_("name or pathname of unix domain socket")),
#else
		OPT_STRING(0, "name", &cl_args.path, N_("name"), N_("named-pipe name")),
#endif
		OPT_INTEGER(0, "threads", &cl_args.nr_threads, N_("number of threads in server thread pool")),
		OPT_INTEGER(0, "max-wait", &cl_args.max_wait_sec, N_("seconds to wait for daemon to start or stop")),

		OPT_INTEGER(0, "bytecount", &cl_args.bytecount, N_("number of bytes")),
		OPT_INTEGER(0, "batchsize", &cl_args.batchsize, N_("number of requests per thread")),

		OPT_STRING(0, "byte", &bytevalue, N_("byte"), N_("ballast character")),
		OPT_STRING(0, "token", &cl_args.token, N_("token"), N_("command token to send to the server")),

		OPT_END()
	};

	if (argc < 2)
		usage_with_options(simple_ipc_usage, options);

	if (argc == 2 && !strcmp(argv[1], "-h"))
		usage_with_options(simple_ipc_usage, options);

	if (argc == 2 && !strcmp(argv[1], "SUPPORTS_SIMPLE_IPC"))
		return 0;

	cl_args.subcommand = argv[1];

	argc--;
	argv++;

	argc = parse_options(argc, argv, NULL, options, simple_ipc_usage, 0);

	if (cl_args.nr_threads < 1)
		cl_args.nr_threads = 1;
	if (cl_args.max_wait_sec < 0)
		cl_args.max_wait_sec = 0;
	if (cl_args.bytecount < 1)
		cl_args.bytecount = 1;
	if (cl_args.batchsize < 1)
		cl_args.batchsize = 1;

	if (bytevalue && *bytevalue)
		cl_args.bytevalue = bytevalue[0];

	/*
	 * Use '!!' on all dispatch functions to map from `error()` style
	 * (returns -1) style to `test_must_fail` style (expects 1).  This
	 * makes shell error messages less confusing.
	 */

	if (!strcmp(cl_args.subcommand, "is-active"))
		return !!client__probe_server();

	if (!strcmp(cl_args.subcommand, "run-daemon"))
		return !!daemon__run_server();

	if (!strcmp(cl_args.subcommand, "start-daemon"))
		return !!daemon__start_server();

	/*
	 * Client commands follow.  Ensure a server is running before
	 * sending any data.  This might be overkill, but then again
	 * this is a test harness.
	 */

	if (!strcmp(cl_args.subcommand, "stop-daemon")) {
		if (client__probe_server())
			return 1;
		return !!client__stop_server();
	}

	if (!strcmp(cl_args.subcommand, "send")) {
		if (client__probe_server())
			return 1;
		return !!client__send_ipc();
	}

	if (!strcmp(cl_args.subcommand, "sendbytes")) {
		if (client__probe_server())
			return 1;
		return !!client__sendbytes();
	}

	if (!strcmp(cl_args.subcommand, "multiple")) {
		if (client__probe_server())
			return 1;
		return !!client__multiple();
	}

	die("Unhandled subcommand: '%s'", cl_args.subcommand);
}
#endif
Commit	Line	Data
36a7eb68 JH	1	/*
	2	* test-simple-ipc.c: verify that the Inter-Process Communication works.
	3	*/
	4
	5	#include "test-tool.h"
15db4e7f	6	#include "gettext.h"
36a7eb68 JH	7	#include "strbuf.h"
	8	#include "simple-ipc.h"
	9	#include "parse-options.h"
	10	#include "thread-utils.h"
	11	#include "strvec.h"
05881a6f	12	#include "run-command.h"
15db4e7f	13	#include "trace2.h"
36a7eb68 JH	14
	15	#ifndef SUPPORTS_SIMPLE_IPC
	16	int cmd__simple_ipc(int argc, const char **argv)
	17	{
	18	die("simple IPC not available on this platform");
	19	}
	20	#else
	21
	22	/*
	23	* The test daemon defines an "application callback" that supports a
	24	* series of commands (see `test_app_cb()`).
	25	*
	26	* Unknown commands are caught here and we send an error message back
	27	* to the client process.
	28	*/
	29	static int app__unhandled_command(const char *command,
	30	ipc_server_reply_cb *reply_cb,
	31	struct ipc_server_reply_data *reply_data)
	32	{
	33	struct strbuf buf = STRBUF_INIT;
	34	int ret;
	35
	36	strbuf_addf(&buf, "unhandled command: %s", command);
	37	ret = reply_cb(reply_data, buf.buf, buf.len);
	38	strbuf_release(&buf);
	39
	40	return ret;
	41	}
	42
	43	/*
	44	* Reply with a single very large buffer. This is to ensure that
	45	* long response are properly handled -- whether the chunking occurs
	46	* in the kernel or in the (probably pkt-line) layer.
	47	*/
	48	#define BIG_ROWS (10000)
	49	static int app__big_command(ipc_server_reply_cb *reply_cb,
	50	struct ipc_server_reply_data *reply_data)
	51	{
	52	struct strbuf buf = STRBUF_INIT;
	53	int row;
	54	int ret;
	55
	56	for (row = 0; row < BIG_ROWS; row++)
	57	strbuf_addf(&buf, "big: %.75d\n", row);
	58
	59	ret = reply_cb(reply_data, buf.buf, buf.len);
	60	strbuf_release(&buf);
	61
	62	return ret;
	63	}
	64
	65	/*
	66	* Reply with a series of lines. This is to ensure that we can incrementally
	67	* compute the response and chunk it to the client.
	68	*/
	69	#define CHUNK_ROWS (10000)
	70	static int app__chunk_command(ipc_server_reply_cb *reply_cb,
	71	struct ipc_server_reply_data *reply_data)
	72	{
	73	struct strbuf buf = STRBUF_INIT;
	74	int row;
	75	int ret;
	76
	77	for (row = 0; row < CHUNK_ROWS; row++) {
78	strbuf_setlen(&buf, 0);
79	strbuf_addf(&buf, "big: %.75d\n", row);
80	ret = reply_cb(reply_data, buf.buf, buf.len);
81	}
82
83	strbuf_release(&buf);
84
85	return ret;
86	}
87
88	/*
89	* Slowly reply with a series of lines. This is to model an expensive to
90	* compute chunked response (which might happen if this callback is running
91	* in a thread and is fighting for a lock with other threads).
92	*/
93	#define SLOW_ROWS (1000)
94	#define SLOW_DELAY_MS (10)
95	static int app__slow_command(ipc_server_reply_cb *reply_cb,
96	struct ipc_server_reply_data *reply_data)
97	{
98	struct strbuf buf = STRBUF_INIT;
99	int row;
100	int ret;
101
102	for (row = 0; row < SLOW_ROWS; row++) {
103	strbuf_setlen(&buf, 0);
104	strbuf_addf(&buf, "big: %.75d\n", row);
105	ret = reply_cb(reply_data, buf.buf, buf.len);
106	sleep_millisec(SLOW_DELAY_MS);
107	}
108
109	strbuf_release(&buf);
110
111	return ret;
112	}
113
114	/*
115	* The client sent a command followed by a (possibly very) large buffer.
116	*/
a3e2033e	117	static int app__sendbytes_command(const char *received, size_t received_len,
36a7eb68 JH	118	ipc_server_reply_cb *reply_cb,
	119	struct ipc_server_reply_data *reply_data)
	120	{
	121	struct strbuf buf_resp = STRBUF_INIT;
	122	const char *p = "?";
	123	int len_ballast = 0;
	124	int k;
	125	int errs = 0;
	126	int ret;
	127
a3e2033e JH	128	/*
	129	* The test is setup to send:
	130	* "sendbytes" SP <n * char>
	131	*/
	132	if (received_len < strlen("sendbytes "))
	133	BUG("received_len is short in app__sendbytes_command");
	134
36a7eb68 JH	135	if (skip_prefix(received, "sendbytes ", &p))
	136	len_ballast = strlen(p);
	137
	138	/*
	139	* Verify that the ballast is n copies of a single letter.
	140	* And that the multi-threaded IO layer didn't cross the streams.
	141	*/
	142	for (k = 1; k < len_ballast; k++)
	143	if (p[k] != p[0])
	144	errs++;
	145
	146	if (errs)
	147	strbuf_addf(&buf_resp, "errs:%d\n", errs);
	148	else
	149	strbuf_addf(&buf_resp, "rcvd:%c%08d\n", p[0], len_ballast);
	150
	151	ret = reply_cb(reply_data, buf_resp.buf, buf_resp.len);
	152
	153	strbuf_release(&buf_resp);
	154
	155	return ret;
	156	}
	157
	158	/*
	159	* An arbitrary fixed address to verify that the application instance
	160	* data is handled properly.
	161	*/
	162	static int my_app_data = 42;
	163
	164	static ipc_server_application_cb test_app_cb;
	165
	166	/*
	167	* This is the "application callback" that sits on top of the
	168	* "ipc-server". It completely defines the set of commands supported
	169	* by this application.
	170	*/
	171	static int test_app_cb(void *application_data,
a3e2033e	172	const char *command, size_t command_len,
36a7eb68 JH	173	ipc_server_reply_cb *reply_cb,
	174	struct ipc_server_reply_data *reply_data)
	175	{
	176	/*
	177	* Verify that we received the application-data that we passed
	178	* when we started the ipc-server. (We have several layers of
	179	* callbacks calling callbacks and it's easy to get things mixed
	180	* up (especially when some are "void*").)
	181	*/
	182	if (application_data != (void*)&my_app_data)
	183	BUG("application_cb: application_data pointer wrong");
	184
a3e2033e	185	if (command_len == 4 && !strncmp(command, "quit", 4)) {
36a7eb68 JH	186	/*
	187	* The client sent a "quit" command. This is an async
	188	* request for the server to shutdown.
	189	*
	190	* We DO NOT send the client a response message
	191	* (because we have nothing to say and the other
	192	* server threads have not yet stopped).
	193	*
	194	* Tell the ipc-server layer to start shutting down.
	195	* This includes: stop listening for new connections
	196	* on the socket/pipe and telling all worker threads
	197	* to finish/drain their outgoing responses to other
	198	* clients.
	199	*
	200	* This DOES NOT force an immediate sync shutdown.
	201	*/
	202	return SIMPLE_IPC_QUIT;
	203	}
	204
a3e2033e	205	if (command_len == 4 && !strncmp(command, "ping", 4)) {
36a7eb68 JH	206	const char *answer = "pong";
	207	return reply_cb(reply_data, answer, strlen(answer));
	208	}
	209
a3e2033e	210	if (command_len == 3 && !strncmp(command, "big", 3))
36a7eb68 JH	211	return app__big_command(reply_cb, reply_data);
36a7eb68 JH	212
a3e2033e	213	if (command_len == 5 && !strncmp(command, "chunk", 5))
36a7eb68 JH	214	return app__chunk_command(reply_cb, reply_data);
36a7eb68 JH	215
a3e2033e	216	if (command_len == 4 && !strncmp(command, "slow", 4))
36a7eb68 JH	217	return app__slow_command(reply_cb, reply_data);
36a7eb68 JH	218
a3e2033e JH	219	if (command_len >= 10 && starts_with(command, "sendbytes "))
	220	return app__sendbytes_command(command, command_len,
	221	reply_cb, reply_data);
36a7eb68 JH	222
	223	return app__unhandled_command(command, reply_cb, reply_data);
	224	}
	225
	226	struct cl_args
	227	{
	228	const char *subcommand;
	229	const char *path;
	230	const char *token;
	231
	232	int nr_threads;
	233	int max_wait_sec;
	234	int bytecount;
	235	int batchsize;
	236
	237	char bytevalue;
	238	};
	239
	240	static struct cl_args cl_args = {
	241	.subcommand = NULL,
	242	.path = "ipc-test",
	243	.token = NULL,
	244
	245	.nr_threads = 5,
	246	.max_wait_sec = 60,
	247	.bytecount = 1024,
	248	.batchsize = 10,
	249
	250	.bytevalue = 'x',
	251	};
	252
	253	/*
	254	* This process will run as a simple-ipc server and listen for IPC commands
	255	* from client processes.
	256	*/
	257	static int daemon__run_server(void)
	258	{
	259	int ret;
	260
	261	struct ipc_server_opts opts = {
	262	.nr_threads = cl_args.nr_threads,
	263	};
	264
	265	/*
	266	* Synchronously run the ipc-server. We don't need any application
	267	* instance data, so pass an arbitrary pointer (that we'll later
	268	* verify made the round trip).
	269	*/
	270	ret = ipc_server_run(cl_args.path, &opts, test_app_cb, (void*)&my_app_data);
	271	if (ret == -2)
05881a6f	272	error("socket/pipe already in use: '%s'", cl_args.path);
36a7eb68	273	else if (ret == -1)
05881a6f	274	error_errno("could not start server on: '%s'", cl_args.path);
36a7eb68 JH	275
	276	return ret;
	277	}
	278
05881a6f	279	static start_bg_wait_cb bg_wait_cb;
36a7eb68	280
05881a6f JH	281	static int bg_wait_cb(const struct child_process cp, void cb_data)
	282	{
	283	int s = ipc_get_active_state(cl_args.path);
36a7eb68	284
05881a6f JH	285	switch (s) {
	286	case IPC_STATE__LISTENING:
	287	/* child is "ready" */
	288	return 0;
36a7eb68	289
05881a6f JH	290	case IPC_STATE__NOT_LISTENING:
	291	case IPC_STATE__PATH_NOT_FOUND:
	292	/* give child more time */
	293	return 1;
36a7eb68 JH	294
36a7eb68 JH	295	default:
05881a6f JH	296	case IPC_STATE__INVALID_PATH:
	297	case IPC_STATE__OTHER_ERROR:
	298	/* all the time in world won't help */
	299	return -1;
36a7eb68 JH	300	}
36a7eb68 JH	301	}
36a7eb68	302
05881a6f	303	static int daemon__start_server(void)
36a7eb68	304	{
05881a6f JH	305	struct child_process cp = CHILD_PROCESS_INIT;
05881a6f JH	306	enum start_bg_result sbgr;
36a7eb68	307
05881a6f JH	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);
36a7eb68	313
05881a6f JH	314	cp.no_stdin = 1;
	315	cp.no_stdout = 1;
	316	cp.no_stderr = 1;
36a7eb68	317
05881a6f	318	sbgr = start_bg_command(&cp, bg_wait_cb, NULL, cl_args.max_wait_sec);
36a7eb68	319
05881a6f JH	320	switch (sbgr) {
	321	case SBGR_READY:
	322	return 0;
36a7eb68	323
05881a6f JH	324	default:
	325	case SBGR_ERROR:
	326	case SBGR_CB_ERROR:
	327	return error("daemon failed to start");
36a7eb68	328
05881a6f JH	329	case SBGR_TIMEOUT:
05881a6f JH	330	return error("daemon not online yet");
36a7eb68	331
05881a6f JH	332	case SBGR_DIED:
05881a6f JH	333	return error("daemon terminated");
36a7eb68 JH	334	}
	335	}
	336
36a7eb68 JH	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
a3e2033e JH	387	if (!ipc_client_send_command(cl_args.path, &options,
	388	command, strlen(command),
	389	&buf)) {
36a7eb68 JH	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)
05881a6f	439	return error("daemon has not shutdown yet");
36a7eb68 JH	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
a3e2033e JH	457	if (!ipc_client_send_command(path, options,
	458	buf_send.buf, buf_send.len,
	459	&buf_resp)) {
36a7eb68 JH	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	OPT_STRING(0, "byte", &bytevalue, N_("byte"), N_("ballast character")),
607	OPT_STRING(0, "token", &cl_args.token, N_("token"), N_("command token to send to the server")),
608
609	OPT_END()
610	};
611
612	if (argc < 2)
613	usage_with_options(simple_ipc_usage, options);
614
615	if (argc == 2 && !strcmp(argv[1], "-h"))
616	usage_with_options(simple_ipc_usage, options);
617
618	if (argc == 2 && !strcmp(argv[1], "SUPPORTS_SIMPLE_IPC"))
619	return 0;
620
621	cl_args.subcommand = argv[1];
622
623	argc--;
624	argv++;
625
626	argc = parse_options(argc, argv, NULL, options, simple_ipc_usage, 0);
627
628	if (cl_args.nr_threads < 1)
629	cl_args.nr_threads = 1;
630	if (cl_args.max_wait_sec < 0)
631	cl_args.max_wait_sec = 0;
632	if (cl_args.bytecount < 1)
633	cl_args.bytecount = 1;
634	if (cl_args.batchsize < 1)
635	cl_args.batchsize = 1;
636
637	if (bytevalue && *bytevalue)
638	cl_args.bytevalue = bytevalue[0];
639
640	/*
641	* Use '!!' on all dispatch functions to map from `error()` style
642	* (returns -1) style to `test_must_fail` style (expects 1). This
643	* makes shell error messages less confusing.
644	*/
645
646	if (!strcmp(cl_args.subcommand, "is-active"))
647	return !!client__probe_server();
648
649	if (!strcmp(cl_args.subcommand, "run-daemon"))
650	return !!daemon__run_server();
651
652	if (!strcmp(cl_args.subcommand, "start-daemon"))
653	return !!daemon__start_server();
654
655	/*
656	* Client commands follow. Ensure a server is running before
657	* sending any data. This might be overkill, but then again
658	* this is a test harness.
659	*/
660
661	if (!strcmp(cl_args.subcommand, "stop-daemon")) {
662	if (client__probe_server())
663	return 1;
664	return !!client__stop_server();
665	}
666
667	if (!strcmp(cl_args.subcommand, "send")) {
668	if (client__probe_server())
669	return 1;
670	return !!client__send_ipc();
671	}
672
673	if (!strcmp(cl_args.subcommand, "sendbytes")) {
674	if (client__probe_server())
675	return 1;
676	return !!client__sendbytes();
677	}
678
679	if (!strcmp(cl_args.subcommand, "multiple")) {
680	if (client__probe_server())
681	return 1;
682	return !!client__multiple();
683	}
684
685	die("Unhandled subcommand: '%s'", cl_args.subcommand);
686	}
687	#endif