@matheusbsbs / matheus-bagliano / commits / 766fce69e9

simple-ipc: split async server initialization and running

To start an async ipc server, you call ipc_server_run_async(). That initializes the ipc_server_data object, and starts all of the threads running, which may immediately start serving clients. This can create some awkward timing problems, though. In the fsmonitor daemon (the sole user of the simple-ipc system), we want to create the ipc server early in the process, which means we may start serving clients before the rest of the daemon is fully initialized. To solve this, let's break run_async() into two parts: an initialization which allocates all data and spawns the threads (without letting them run), and a start function which actually lets them begin work. Since we have two simple-ipc implementations, we have to handle this twice: - in ipc-unix-socket.c, we have a central listener thread which hands connections off to worker threads using a work_available mutex. We can hold that mutex after init, and release it when we're ready to start. We do need an extra "started" flag so that we know whether the main thread is holding the mutex or not (e.g., if we prematurely stop the server, we want to make sure all of the worker threads are released to hear about the shutdown). - in ipc-win32.c, we don't have a central mutex. So we'll introduce a new startup_barrier mutex, which we'll similarly hold until we're ready to let the threads proceed. We again need a "started" flag here to make sure that we release the barrier mutex when shutting down, so that the sub-threads can proceed to the finish. I've renamed the run_async() function to init_async() to make sure we catch all callers, since they'll now need to call the matching start_async(). We could leave run_async() as a wrapper that does both, but there's not much point. There are only two callers, one of which is fsmonitor, which will want to actually do work between the two calls. And the other is just a test-tool wrapper. For now I've added the start_async() calls in fsmonitor where they would otherwise have happened, so there should be no behavior change with this patch. Signed-off-by: Jeff King <peff@peff.net> Acked-by: Koji Nakamaru <koji.nakamaru@gree.net> Signed-off-by: Junio C Hamano <gitster@pobox.com>

Jeff King committed Oct 8, 2024 at 04:33 UTC 766fce69e947de20e2ed99b7e298b15338df5534

5 files changed +88 -18

builtin/fsmonitor--daemon.c

+5 -3

@@ -1208,13 +1208,15 @@ static int fsmonitor_run_daemon_1(struct fsmonitor_daemon_state *state)

1208 * system event listener thread so that we have the IPC handle

1209 * before we need it.

1210 */

1211 - if (ipc_server_run_async(&state->ipc_server_data,

1212 - state->path_ipc.buf, &ipc_opts,

1213 - handle_client, state))

1211 + if (ipc_server_init_async(&state->ipc_server_data,

1212 + state->path_ipc.buf, &ipc_opts,

1213 + handle_client, state))

1214 return error_errno(

1215 _("could not start IPC thread pool on '%s'"),

1216 state->path_ipc.buf);

1217

1218 + ipc_server_start_async(&state->ipc_server_data);

1219 +

1220 /*

1221 * Start the fsmonitor listener thread to collect filesystem

1222 * events.

compat/simple-ipc/ipc-shared.c

+3 -2

@@ -16,11 +16,12 @@ int ipc_server_run(const char *path, const struct ipc_server_opts *opts,

16 struct ipc_server_data *server_data = NULL;

17 int ret;

19 - ret = ipc_server_run_async(&server_data, path, opts,

20 - application_cb, application_data);

19 + ret = ipc_server_init_async(&server_data, path, opts,

20 + application_cb, application_data);

21 if (ret)

22 return ret;

24 + ipc_server_start_async(server_data);

25 ret = ipc_server_await(server_data);

27 ipc_server_free(server_data);

compat/simple-ipc/ipc-unix-socket.c

+23 -5

@@ -328,6 +328,7 @@ struct ipc_server_data {

328 int back_pos;

329 int front_pos;

330

331 + int started;

332 int shutdown_requested;

333 int is_stopped;

334 };

@@ -824,10 +825,10 @@ static int setup_listener_socket(

825 /*

826 * Start IPC server in a pool of background threads.

827 */

827 -int ipc_server_run_async(struct ipc_server_data **returned_server_data,

828 - const char *path, const struct ipc_server_opts *opts,

829 - ipc_server_application_cb *application_cb,

830 - void *application_data)

828 +int ipc_server_init_async(struct ipc_server_data **returned_server_data,

829 + const char *path, const struct ipc_server_opts *opts,

830 + ipc_server_application_cb *application_cb,

831 + void *application_data)

832 {

833 struct unix_ss_socket *server_socket = NULL;

834 struct ipc_server_data *server_data;

@@ -888,6 +889,12 @@ int ipc_server_run_async(struct ipc_server_data **returned_server_data,

889 server_data->accept_thread->fd_send_shutdown = sv[0];

890 server_data->accept_thread->fd_wait_shutdown = sv[1];

891

892 + /*

893 + * Hold work-available mutex so that no work can start until

894 + * we unlock it.

895 + */

896 + pthread_mutex_lock(&server_data->work_available_mutex);

897 +

898 if (pthread_create(&server_data->accept_thread->pthread_id, NULL,

899 accept_thread_proc, server_data->accept_thread))

900 die_errno(_("could not start accept_thread '%s'"), path);

@@ -918,6 +925,15 @@ int ipc_server_run_async(struct ipc_server_data **returned_server_data,

925 return 0;

926 }

927

928 +void ipc_server_start_async(struct ipc_server_data *server_data)

929 +{

930 + if (!server_data || server_data->started)

931 + return;

932 +

933 + server_data->started = 1;

934 + pthread_mutex_unlock(&server_data->work_available_mutex);

935 +}

936 +

937 /*

938 * Gently tell the IPC server treads to shutdown.

939 * Can be run on any thread.

@@ -933,7 +949,9 @@ int ipc_server_stop_async(struct ipc_server_data *server_data)

949

950 trace2_region_enter("ipc-server", "server-stop-async", NULL);

951

936 - pthread_mutex_lock(&server_data->work_available_mutex);

952 + /* If we haven't started yet, we are already holding lock. */

953 + if (server_data->started)

954 + pthread_mutex_lock(&server_data->work_available_mutex);

955

956 server_data->shutdown_requested = 1;

957

compat/simple-ipc/ipc-win32.c

+44 -4

@@ -371,6 +371,9 @@ struct ipc_server_data {

371 HANDLE hEventStopRequested;

372 struct ipc_server_thread_data *thread_list;

373 int is_stopped;

374 +

375 + pthread_mutex_t startup_barrier;

376 + int started;

377 };

378

379 enum connect_result {

@@ -526,6 +529,16 @@ static int use_connection(struct ipc_server_thread_data *server_thread_data)

529 return ret;

530 }

531

532 +static void wait_for_startup_barrier(struct ipc_server_data *server_data)

533 +{

534 + /*

535 + * Temporarily hold the startup_barrier mutex before starting,

536 + * which lets us know that it's OK to start serving requests.

537 + */

538 + pthread_mutex_lock(&server_data->startup_barrier);

539 + pthread_mutex_unlock(&server_data->startup_barrier);

540 +}

541 +

542 /*

543 * Thread proc for an IPC server worker thread. It handles a series of

544 * connections from clients. It cleans and reuses the hPipe between each

@@ -550,6 +563,8 @@ static void *server_thread_proc(void *_server_thread_data)

563 memset(&oConnect, 0, sizeof(oConnect));

564 oConnect.hEvent = hEventConnected;

565

566 + wait_for_startup_barrier(server_thread_data->server_data);

567 +

568 for (;;) {

569 cr = wait_for_connection(server_thread_data, &oConnect);

570

@@ -752,10 +767,10 @@ static HANDLE create_new_pipe(wchar_t *wpath, int is_first)

767 return hPipe;

768 }

769

755 -int ipc_server_run_async(struct ipc_server_data **returned_server_data,

756 - const char *path, const struct ipc_server_opts *opts,

757 - ipc_server_application_cb *application_cb,

758 - void *application_data)

770 +int ipc_server_init_async(struct ipc_server_data **returned_server_data,

771 + const char *path, const struct ipc_server_opts *opts,

772 + ipc_server_application_cb *application_cb,

773 + void *application_data)

774 {

775 struct ipc_server_data *server_data;

776 wchar_t wpath[MAX_PATH];

@@ -787,6 +802,13 @@ int ipc_server_run_async(struct ipc_server_data **returned_server_data,

802 strbuf_addstr(&server_data->buf_path, path);

803 wcscpy(server_data->wpath, wpath);

804

805 + /*

806 + * Hold the startup_barrier lock so that no threads will progress

807 + * until ipc_server_start_async() is called.

808 + */

809 + pthread_mutex_init(&server_data->startup_barrier, NULL);

810 + pthread_mutex_lock(&server_data->startup_barrier);

811 +

812 if (nr_threads < 1)

813 nr_threads = 1;

814

@@ -837,6 +859,15 @@ int ipc_server_run_async(struct ipc_server_data **returned_server_data,

859 return 0;

860 }

861

862 +void ipc_server_start_async(struct ipc_server_data *server_data)

863 +{

864 + if (!server_data || server_data->started)

865 + return;

866 +

867 + server_data->started = 1;

868 + pthread_mutex_unlock(&server_data->startup_barrier);

869 +}

870 +

871 int ipc_server_stop_async(struct ipc_server_data *server_data)

872 {

873 if (!server_data)

@@ -850,6 +881,13 @@ int ipc_server_stop_async(struct ipc_server_data *server_data)

881 * We DO NOT attempt to force them to drop an active connection.

882 */

883 SetEvent(server_data->hEventStopRequested);

884 +

885 + /*

886 + * If we haven't yet told the threads they are allowed to run,

887 + * do so now, so they can receive the shutdown event.

888 + */

889 + ipc_server_start_async(server_data);

890 +

891 return 0;

892 }

893

@@ -900,5 +938,7 @@ void ipc_server_free(struct ipc_server_data *server_data)

938 free(std);

939 }

940

941 + pthread_mutex_destroy(&server_data->startup_barrier);

942 +

943 free(server_data);

944 }

simple-ipc.h

+13 -4

@@ -179,11 +179,20 @@ struct ipc_server_opts

179 * When a client IPC message is received, the `application_cb` will be

180 * called (possibly on a random thread) to handle the message and

181 * optionally compose a reply message.

182 + *

183 + * This initializes all threads but no actual work will be done until

184 + * ipc_server_start_async() is called.

185 + */

186 +int ipc_server_init_async(struct ipc_server_data **returned_server_data,

187 + const char *path, const struct ipc_server_opts *opts,

188 + ipc_server_application_cb *application_cb,

189 + void *application_data);

190 +

191 +/*

192 + * Let an async server start running. This needs to be called only once

193 + * after initialization.

194 */

183 -int ipc_server_run_async(struct ipc_server_data **returned_server_data,

184 - const char *path, const struct ipc_server_opts *opts,

185 - ipc_server_application_cb *application_cb,

186 - void *application_data);

195 +void ipc_server_start_async(struct ipc_server_data *server_data);

196

197 /*

198 * Gently signal the IPC server pool to shutdown. No new client