Merge branch 'rs/diff-parseopts-cleanup'

[git/gitster.git] / compat / simple-ipc / ipc-unix-socket.c

#include "git-compat-util.h"
#include "gettext.h"
#include "simple-ipc.h"
#include "strbuf.h"
#include "thread-utils.h"
#include "trace2.h"
#include "unix-socket.h"
#include "unix-stream-server.h"

#ifndef SUPPORTS_SIMPLE_IPC
/*
 * This source file should only be compiled when Simple IPC is supported.
 * See the top-level Makefile.
 */
#error SUPPORTS_SIMPLE_IPC not defined
#endif

enum ipc_active_state ipc_get_active_state(const char *path)
{
        enum ipc_active_state state = IPC_STATE__OTHER_ERROR;
        struct ipc_client_connect_options options
                = IPC_CLIENT_CONNECT_OPTIONS_INIT;
        struct stat st;
        struct ipc_client_connection *connection_test = NULL;

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

        if (lstat(path, &st) == -1) {
                switch (errno) {
                case ENOENT:
                case ENOTDIR:
                        return IPC_STATE__NOT_LISTENING;
                default:
                        return IPC_STATE__INVALID_PATH;
                }
        }

#ifdef __CYGWIN__
        /*
         * Cygwin emulates Unix sockets by writing special-crafted files whose
         * `system` bit is set.
         *
         * If we are too fast, Cygwin might still be in the process of marking
         * the underlying file as a system file. Until then, we will not see a
         * Unix socket here, but a plain file instead. Just in case that this
         * is happening, wait a little and try again.
         */
        {
                static const int delay[] = { 1, 10, 20, 40, -1 };
                int i;

                for (i = 0; S_ISREG(st.st_mode) && delay[i] > 0; i++) {
                        sleep_millisec(delay[i]);
                        if (lstat(path, &st) == -1)
                                return IPC_STATE__INVALID_PATH;
                }
        }
#endif

        /* also complain if a plain file is in the way */
        if ((st.st_mode & S_IFMT) != S_IFSOCK)
                return IPC_STATE__INVALID_PATH;

        /*
         * Just because the filesystem has a S_IFSOCK type inode
         * at `path`, doesn't mean it that there is a server listening.
         * Ping it to be sure.
         */
        state = ipc_client_try_connect(path, &options, &connection_test);
        ipc_client_close_connection(connection_test);

        return state;
}

/*
 * Retry frequency when trying to connect to a server.
 *
 * This value should be short enough that we don't seriously delay our
 * caller, but not fast enough that our spinning puts pressure on the
 * system.
 */
#define WAIT_STEP_MS (50)

/*
 * Try to connect to the server.  If the server is just starting up or
 * is very busy, we may not get a connection the first time.
 */
static enum ipc_active_state connect_to_server(
        const char *path,
        int timeout_ms,
        const struct ipc_client_connect_options *options,
        int *pfd)
{
        int k;

        *pfd = -1;

        for (k = 0; k < timeout_ms; k += WAIT_STEP_MS) {
                int fd = unix_stream_connect(path, options->uds_disallow_chdir);

                if (fd != -1) {
                        *pfd = fd;
                        return IPC_STATE__LISTENING;
                }

                if (errno == ENOENT) {
                        if (!options->wait_if_not_found)
                                return IPC_STATE__PATH_NOT_FOUND;

                        goto sleep_and_try_again;
                }

                if (errno == ETIMEDOUT) {
                        if (!options->wait_if_busy)
                                return IPC_STATE__NOT_LISTENING;

                        goto sleep_and_try_again;
                }

                if (errno == ECONNREFUSED) {
                        if (!options->wait_if_busy)
                                return IPC_STATE__NOT_LISTENING;

                        goto sleep_and_try_again;
                }

                return IPC_STATE__OTHER_ERROR;

        sleep_and_try_again:
                sleep_millisec(WAIT_STEP_MS);
        }

        return IPC_STATE__NOT_LISTENING;
}

/*
 * The total amount of time that we are willing to wait when trying to
 * connect to a server.
 *
 * When the server is first started, it might take a little while for
 * it to become ready to service requests.  Likewise, the server may
 * be very (temporarily) busy and not respond to our connections.
 *
 * We should gracefully and silently handle those conditions and try
 * again for a reasonable time period.
 *
 * The value chosen here should be long enough for the server
 * to reliably heal from the above conditions.
 */
#define MY_CONNECTION_TIMEOUT_MS (1000)

enum ipc_active_state ipc_client_try_connect(
        const char *path,
        const struct ipc_client_connect_options *options,
        struct ipc_client_connection **p_connection)
{
        enum ipc_active_state state = IPC_STATE__OTHER_ERROR;
        int fd = -1;

        *p_connection = NULL;

        trace2_region_enter("ipc-client", "try-connect", NULL);
        trace2_data_string("ipc-client", NULL, "try-connect/path", path);

        state = connect_to_server(path, MY_CONNECTION_TIMEOUT_MS,
                                  options, &fd);

        trace2_data_intmax("ipc-client", NULL, "try-connect/state",
                           (intmax_t)state);
        trace2_region_leave("ipc-client", "try-connect", NULL);

        if (state == IPC_STATE__LISTENING) {
                (*p_connection) = xcalloc(1, sizeof(struct ipc_client_connection));
                (*p_connection)->fd = fd;
        }

        return state;
}

void ipc_client_close_connection(struct ipc_client_connection *connection)
{
        if (!connection)
                return;

        if (connection->fd != -1)
                close(connection->fd);

        free(connection);
}

int ipc_client_send_command_to_connection(
        struct ipc_client_connection *connection,
        const char *message, size_t message_len,
        struct strbuf *answer)
{
        int ret = 0;

        strbuf_setlen(answer, 0);

        trace2_region_enter("ipc-client", "send-command", NULL);

        if (write_packetized_from_buf_no_flush(message, message_len,
                                               connection->fd) < 0 ||
            packet_flush_gently(connection->fd) < 0) {
                ret = error(_("could not send IPC command"));
                goto done;
        }

        if (read_packetized_to_strbuf(
                    connection->fd, answer,
                    PACKET_READ_GENTLE_ON_EOF | PACKET_READ_GENTLE_ON_READ_ERROR) < 0) {
                ret = error(_("could not read IPC response"));
                goto done;
        }

done:
        trace2_region_leave("ipc-client", "send-command", NULL);
        return ret;
}

int ipc_client_send_command(const char *path,
                            const struct ipc_client_connect_options *options,
                            const char *message, size_t message_len,
                            struct strbuf *answer)
{
        int ret = -1;
        enum ipc_active_state state;
        struct ipc_client_connection *connection = NULL;

        state = ipc_client_try_connect(path, options, &connection);

        if (state != IPC_STATE__LISTENING)
                return ret;

        ret = ipc_client_send_command_to_connection(connection,
                                                    message, message_len,
                                                    answer);

        ipc_client_close_connection(connection);

        return ret;
}

static int set_socket_blocking_flag(int fd, int make_nonblocking)
{
        int flags;

        flags = fcntl(fd, F_GETFL, NULL);

        if (flags < 0)
                return -1;

        if (make_nonblocking)
                flags |= O_NONBLOCK;
        else
                flags &= ~O_NONBLOCK;

        return fcntl(fd, F_SETFL, flags);
}

/*
 * Magic numbers used to annotate callback instance data.
 * These are used to help guard against accidentally passing the
 * wrong instance data across multiple levels of callbacks (which
 * is easy to do if there are `void*` arguments).
 */
enum magic {
        MAGIC_SERVER_REPLY_DATA,
        MAGIC_WORKER_THREAD_DATA,
        MAGIC_ACCEPT_THREAD_DATA,
        MAGIC_SERVER_DATA,
};

struct ipc_server_reply_data {
        enum magic magic;
        int fd;
        struct ipc_worker_thread_data *worker_thread_data;
};

struct ipc_worker_thread_data {
        enum magic magic;
        struct ipc_worker_thread_data *next_thread;
        struct ipc_server_data *server_data;
        pthread_t pthread_id;
};

struct ipc_accept_thread_data {
        enum magic magic;
        struct ipc_server_data *server_data;

        struct unix_ss_socket *server_socket;

        int fd_send_shutdown;
        int fd_wait_shutdown;
        pthread_t pthread_id;
};

/*
 * With unix-sockets, the conceptual "ipc-server" is implemented as a single
 * controller "accept-thread" thread and a pool of "worker-thread" threads.
 * The former does the usual `accept()` loop and dispatches connections
 * to an idle worker thread.  The worker threads wait in an idle loop for
 * a new connection, communicate with the client and relay data to/from
 * the `application_cb` and then wait for another connection from the
 * server thread.  This avoids the overhead of constantly creating and
 * destroying threads.
 */
struct ipc_server_data {
        enum magic magic;
        ipc_server_application_cb *application_cb;
        void *application_data;
        struct strbuf buf_path;

        struct ipc_accept_thread_data *accept_thread;
        struct ipc_worker_thread_data *worker_thread_list;

        pthread_mutex_t work_available_mutex;
        pthread_cond_t work_available_cond;

        /*
         * Accepted but not yet processed client connections are kept
         * in a circular buffer FIFO.  The queue is empty when the
         * positions are equal.
         */
        int *fifo_fds;
        int queue_size;
        int back_pos;
        int front_pos;

        int shutdown_requested;
        int is_stopped;
};

/*
 * Remove and return the oldest queued connection.
 *
 * Returns -1 if empty.
 */
static int fifo_dequeue(struct ipc_server_data *server_data)
{
        /* ASSERT holding mutex */

        int fd;

        if (server_data->back_pos == server_data->front_pos)
                return -1;

        fd = server_data->fifo_fds[server_data->front_pos];
        server_data->fifo_fds[server_data->front_pos] = -1;

        server_data->front_pos++;
        if (server_data->front_pos == server_data->queue_size)
                server_data->front_pos = 0;

        return fd;
}

/*
 * Push a new fd onto the back of the queue.
 *
 * Drop it and return -1 if queue is already full.
 */
static int fifo_enqueue(struct ipc_server_data *server_data, int fd)
{
        /* ASSERT holding mutex */

        int next_back_pos;

        next_back_pos = server_data->back_pos + 1;
        if (next_back_pos == server_data->queue_size)
                next_back_pos = 0;

        if (next_back_pos == server_data->front_pos) {
                /* Queue is full. Just drop it. */
                close(fd);
                return -1;
        }

        server_data->fifo_fds[server_data->back_pos] = fd;
        server_data->back_pos = next_back_pos;

        return fd;
}

/*
 * Wait for a connection to be queued to the FIFO and return it.
 *
 * Returns -1 if someone has already requested a shutdown.
 */
static int worker_thread__wait_for_connection(
        struct ipc_worker_thread_data *worker_thread_data)
{
        /* ASSERT NOT holding mutex */

        struct ipc_server_data *server_data = worker_thread_data->server_data;
        int fd = -1;

        pthread_mutex_lock(&server_data->work_available_mutex);
        for (;;) {
                if (server_data->shutdown_requested)
                        break;

                fd = fifo_dequeue(server_data);
                if (fd >= 0)
                        break;

                pthread_cond_wait(&server_data->work_available_cond,
                                  &server_data->work_available_mutex);
        }
        pthread_mutex_unlock(&server_data->work_available_mutex);

        return fd;
}

/*
 * Forward declare our reply callback function so that any compiler
 * errors are reported when we actually define the function (in addition
 * to any errors reported when we try to pass this callback function as
 * a parameter in a function call).  The former are easier to understand.
 */
static ipc_server_reply_cb do_io_reply_callback;

/*
 * Relay application's response message to the client process.
 * (We do not flush at this point because we allow the caller
 * to chunk data to the client thru us.)
 */
static int do_io_reply_callback(struct ipc_server_reply_data *reply_data,
                       const char *response, size_t response_len)
{
        if (reply_data->magic != MAGIC_SERVER_REPLY_DATA)
                BUG("reply_cb called with wrong instance data");

        return write_packetized_from_buf_no_flush(response, response_len,
                                                  reply_data->fd);
}

/* A randomly chosen value. */
#define MY_WAIT_POLL_TIMEOUT_MS (10)

/*
 * If the client hangs up without sending any data on the wire, just
 * quietly close the socket and ignore this client.
 *
 * This worker thread is committed to reading the IPC request data
 * from the client at the other end of this fd.  Wait here for the
 * client to actually put something on the wire -- because if the
 * client just does a ping (connect and hangup without sending any
 * data), our use of the pkt-line read routines will spew an error
 * message.
 *
 * Return -1 if the client hung up.
 * Return 0 if data (possibly incomplete) is ready.
 */
static int worker_thread__wait_for_io_start(
        struct ipc_worker_thread_data *worker_thread_data,
        int fd)
{
        struct ipc_server_data *server_data = worker_thread_data->server_data;
        struct pollfd pollfd[1];
        int result;

        for (;;) {
                pollfd[0].fd = fd;
                pollfd[0].events = POLLIN;

                result = poll(pollfd, 1, MY_WAIT_POLL_TIMEOUT_MS);
                if (result < 0) {
                        if (errno == EINTR)
                                continue;
                        goto cleanup;
                }

                if (result == 0) {
                        /* a timeout */

                        int in_shutdown;

                        pthread_mutex_lock(&server_data->work_available_mutex);
                        in_shutdown = server_data->shutdown_requested;
                        pthread_mutex_unlock(&server_data->work_available_mutex);

                        /*
                         * If a shutdown is already in progress and this
                         * client has not started talking yet, just drop it.
                         */
                        if (in_shutdown)
                                goto cleanup;
                        continue;
                }

                if (pollfd[0].revents & POLLHUP)
                        goto cleanup;

                if (pollfd[0].revents & POLLIN)
                        return 0;

                goto cleanup;
        }

cleanup:
        close(fd);
        return -1;
}

/*
 * Receive the request/command from the client and pass it to the
 * registered request-callback.  The request-callback will compose
 * a response and call our reply-callback to send it to the client.
 */
static int worker_thread__do_io(
        struct ipc_worker_thread_data *worker_thread_data,
        int fd)
{
        /* ASSERT NOT holding lock */

        struct strbuf buf = STRBUF_INIT;
        struct ipc_server_reply_data reply_data;
        int ret = 0;

        reply_data.magic = MAGIC_SERVER_REPLY_DATA;
        reply_data.worker_thread_data = worker_thread_data;

        reply_data.fd = fd;

        ret = read_packetized_to_strbuf(
                reply_data.fd, &buf,
                PACKET_READ_GENTLE_ON_EOF | PACKET_READ_GENTLE_ON_READ_ERROR);
        if (ret >= 0) {
                ret = worker_thread_data->server_data->application_cb(
                        worker_thread_data->server_data->application_data,
                        buf.buf, buf.len, do_io_reply_callback, &reply_data);

                packet_flush_gently(reply_data.fd);
        }
        else {
                /*
                 * The client probably disconnected/shutdown before it
                 * could send a well-formed message.  Ignore it.
                 */
        }

        strbuf_release(&buf);
        close(reply_data.fd);

        return ret;
}

/*
 * Block SIGPIPE on the current thread (so that we get EPIPE from
 * write() rather than an actual signal).
 *
 * Note that using sigchain_push() and _pop() to control SIGPIPE
 * around our IO calls is not thread safe:
 * [] It uses a global stack of handler frames.
 * [] It uses ALLOC_GROW() to resize it.
 * [] Finally, according to the `signal(2)` man-page:
 *    "The effects of `signal()` in a multithreaded process are unspecified."
 */
static void thread_block_sigpipe(sigset_t *old_set)
{
        sigset_t new_set;

        sigemptyset(&new_set);
        sigaddset(&new_set, SIGPIPE);

        sigemptyset(old_set);
        pthread_sigmask(SIG_BLOCK, &new_set, old_set);
}

/*
 * Thread proc for an IPC worker thread.  It handles a series of
 * connections from clients.  It pulls the next fd from the queue
 * processes it, and then waits for the next client.
 *
 * Block SIGPIPE in this worker thread for the life of the thread.
 * This avoids stray (and sometimes delayed) SIGPIPE signals caused
 * by client errors and/or when we are under extremely heavy IO load.
 *
 * This means that the application callback will have SIGPIPE blocked.
 * The callback should not change it.
 */
static void *worker_thread_proc(void *_worker_thread_data)
{
        struct ipc_worker_thread_data *worker_thread_data = _worker_thread_data;
        struct ipc_server_data *server_data = worker_thread_data->server_data;
        sigset_t old_set;
        int fd, io;
        int ret;

        trace2_thread_start("ipc-worker");

        thread_block_sigpipe(&old_set);

        for (;;) {
                fd = worker_thread__wait_for_connection(worker_thread_data);
                if (fd == -1)
                        break; /* in shutdown */

                io = worker_thread__wait_for_io_start(worker_thread_data, fd);
                if (io == -1)
                        continue; /* client hung up without sending anything */

                ret = worker_thread__do_io(worker_thread_data, fd);

                if (ret == SIMPLE_IPC_QUIT) {
                        trace2_data_string("ipc-worker", NULL, "queue_stop_async",
                                           "application_quit");
                        /*
                         * The application layer is telling the ipc-server
                         * layer to shutdown.
                         *
                         * We DO NOT have a response to send to the client.
                         *
                         * Queue an async stop (to stop the other threads) and
                         * allow this worker thread to exit now (no sense waiting
                         * for the thread-pool shutdown signal).
                         *
                         * Other non-idle worker threads are allowed to finish
                         * responding to their current clients.
                         */
                        ipc_server_stop_async(server_data);
                        break;
                }
        }

        trace2_thread_exit();
        return NULL;
}

/* A randomly chosen value. */
#define MY_ACCEPT_POLL_TIMEOUT_MS (60 * 1000)

/*
 * Accept a new client connection on our socket.  This uses non-blocking
 * IO so that we can also wait for shutdown requests on our socket-pair
 * without actually spinning on a fast timeout.
 */
static int accept_thread__wait_for_connection(
        struct ipc_accept_thread_data *accept_thread_data)
{
        struct pollfd pollfd[2];
        int result;

        for (;;) {
                pollfd[0].fd = accept_thread_data->fd_wait_shutdown;
                pollfd[0].events = POLLIN;

                pollfd[1].fd = accept_thread_data->server_socket->fd_socket;
                pollfd[1].events = POLLIN;

                result = poll(pollfd, 2, MY_ACCEPT_POLL_TIMEOUT_MS);
                if (result < 0) {
                        if (errno == EINTR)
                                continue;
                        return result;
                }

                if (result == 0) {
                        /* a timeout */

                        /*
                         * If someone deletes or force-creates a new unix
                         * domain socket at our path, all future clients
                         * will be routed elsewhere and we silently starve.
                         * If that happens, just queue a shutdown.
                         */
                        if (unix_ss_was_stolen(
                                    accept_thread_data->server_socket)) {
                                trace2_data_string("ipc-accept", NULL,
                                                   "queue_stop_async",
                                                   "socket_stolen");
                                ipc_server_stop_async(
                                        accept_thread_data->server_data);
                        }
                        continue;
                }

                if (pollfd[0].revents & POLLIN) {
                        /* shutdown message queued to socketpair */
                        return -1;
                }

                if (pollfd[1].revents & POLLIN) {
                        /* a connection is available on server_socket */

                        int client_fd =
                                accept(accept_thread_data->server_socket->fd_socket,
                                       NULL, NULL);
                        if (client_fd >= 0)
                                return client_fd;

                        /*
                         * An error here is unlikely -- it probably
                         * indicates that the connecting process has
                         * already dropped the connection.
                         */
                        continue;
                }

                BUG("unandled poll result errno=%d r[0]=%d r[1]=%d",
                    errno, pollfd[0].revents, pollfd[1].revents);
        }
}

/*
 * Thread proc for the IPC server "accept thread".  This waits for
 * an incoming socket connection, appends it to the queue of available
 * connections, and notifies a worker thread to process it.
 *
 * Block SIGPIPE in this thread for the life of the thread.  This
 * avoids any stray SIGPIPE signals when closing pipe fds under
 * extremely heavy loads (such as when the fifo queue is full and we
 * drop incomming connections).
 */
static void *accept_thread_proc(void *_accept_thread_data)
{
        struct ipc_accept_thread_data *accept_thread_data = _accept_thread_data;
        struct ipc_server_data *server_data = accept_thread_data->server_data;
        sigset_t old_set;

        trace2_thread_start("ipc-accept");

        thread_block_sigpipe(&old_set);

        for (;;) {
                int client_fd = accept_thread__wait_for_connection(
                        accept_thread_data);

                pthread_mutex_lock(&server_data->work_available_mutex);
                if (server_data->shutdown_requested) {
                        pthread_mutex_unlock(&server_data->work_available_mutex);
                        if (client_fd >= 0)
                                close(client_fd);
                        break;
                }

                if (client_fd < 0) {
                        /* ignore transient accept() errors */
                }
                else {
                        fifo_enqueue(server_data, client_fd);
                        pthread_cond_broadcast(&server_data->work_available_cond);
                }
                pthread_mutex_unlock(&server_data->work_available_mutex);
        }

        trace2_thread_exit();
        return NULL;
}

/*
 * We can't predict the connection arrival rate relative to the worker
 * processing rate, therefore we allow the "accept-thread" to queue up
 * a generous number of connections, since we'd rather have the client
 * not unnecessarily timeout if we can avoid it.  (The assumption is
 * that this will be used for FSMonitor and a few second wait on a
 * connection is better than having the client timeout and do the full
 * computation itself.)
 *
 * The FIFO queue size is set to a multiple of the worker pool size.
 * This value chosen at random.
 */
#define FIFO_SCALE (100)

/*
 * The backlog value for `listen(2)`.  This doesn't need to huge,
 * rather just large enough for our "accept-thread" to wake up and
 * queue incoming connections onto the FIFO without the kernel
 * dropping any.
 *
 * This value chosen at random.
 */
#define LISTEN_BACKLOG (50)

static int create_listener_socket(
        const char *path,
        const struct ipc_server_opts *ipc_opts,
        struct unix_ss_socket **new_server_socket)
{
        struct unix_ss_socket *server_socket = NULL;
        struct unix_stream_listen_opts uslg_opts = UNIX_STREAM_LISTEN_OPTS_INIT;
        int ret;

        uslg_opts.listen_backlog_size = LISTEN_BACKLOG;
        uslg_opts.disallow_chdir = ipc_opts->uds_disallow_chdir;

        ret = unix_ss_create(path, &uslg_opts, -1, &server_socket);
        if (ret)
                return ret;

        if (set_socket_blocking_flag(server_socket->fd_socket, 1)) {
                int saved_errno = errno;
                unix_ss_free(server_socket);
                errno = saved_errno;
                return -1;
        }

        *new_server_socket = server_socket;

        trace2_data_string("ipc-server", NULL, "listen-with-lock", path);
        return 0;
}

static int setup_listener_socket(
        const char *path,
        const struct ipc_server_opts *ipc_opts,
        struct unix_ss_socket **new_server_socket)
{
        int ret, saved_errno;

        trace2_region_enter("ipc-server", "create-listener_socket", NULL);

        ret = create_listener_socket(path, ipc_opts, new_server_socket);

        saved_errno = errno;
        trace2_region_leave("ipc-server", "create-listener_socket", NULL);
        errno = saved_errno;

        return ret;
}

/*
 * Start IPC server in a pool of background threads.
 */
int ipc_server_run_async(struct ipc_server_data **returned_server_data,
                         const char *path, const struct ipc_server_opts *opts,
                         ipc_server_application_cb *application_cb,
                         void *application_data)
{
        struct unix_ss_socket *server_socket = NULL;
        struct ipc_server_data *server_data;
        int sv[2];
        int k;
        int ret;
        int nr_threads = opts->nr_threads;

        *returned_server_data = NULL;

        /*
         * Create a socketpair and set sv[1] to non-blocking.  This
         * will used to send a shutdown message to the accept-thread
         * and allows the accept-thread to wait on EITHER a client
         * connection or a shutdown request without spinning.
         */
        if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv) < 0)
                return -1;

        if (set_socket_blocking_flag(sv[1], 1)) {
                int saved_errno = errno;
                close(sv[0]);
                close(sv[1]);
                errno = saved_errno;
                return -1;
        }

        ret = setup_listener_socket(path, opts, &server_socket);
        if (ret) {
                int saved_errno = errno;
                close(sv[0]);
                close(sv[1]);
                errno = saved_errno;
                return ret;
        }

        server_data = xcalloc(1, sizeof(*server_data));
        server_data->magic = MAGIC_SERVER_DATA;
        server_data->application_cb = application_cb;
        server_data->application_data = application_data;
        strbuf_init(&server_data->buf_path, 0);
        strbuf_addstr(&server_data->buf_path, path);

        if (nr_threads < 1)
                nr_threads = 1;

        pthread_mutex_init(&server_data->work_available_mutex, NULL);
        pthread_cond_init(&server_data->work_available_cond, NULL);

        server_data->queue_size = nr_threads * FIFO_SCALE;
        CALLOC_ARRAY(server_data->fifo_fds, server_data->queue_size);

        server_data->accept_thread =
                xcalloc(1, sizeof(*server_data->accept_thread));
        server_data->accept_thread->magic = MAGIC_ACCEPT_THREAD_DATA;
        server_data->accept_thread->server_data = server_data;
        server_data->accept_thread->server_socket = server_socket;
        server_data->accept_thread->fd_send_shutdown = sv[0];
        server_data->accept_thread->fd_wait_shutdown = sv[1];

        if (pthread_create(&server_data->accept_thread->pthread_id, NULL,
                           accept_thread_proc, server_data->accept_thread))
                die_errno(_("could not start accept_thread '%s'"), path);

        for (k = 0; k < nr_threads; k++) {
                struct ipc_worker_thread_data *wtd;

                wtd = xcalloc(1, sizeof(*wtd));
                wtd->magic = MAGIC_WORKER_THREAD_DATA;
                wtd->server_data = server_data;

                if (pthread_create(&wtd->pthread_id, NULL, worker_thread_proc,
                                   wtd)) {
                        if (k == 0)
                                die(_("could not start worker[0] for '%s'"),
                                    path);
                        /*
                         * Limp along with the thread pool that we have.
                         */
                        break;
                }

                wtd->next_thread = server_data->worker_thread_list;
                server_data->worker_thread_list = wtd;
        }

        *returned_server_data = server_data;
        return 0;
}

/*
 * Gently tell the IPC server treads to shutdown.
 * Can be run on any thread.
 */
int ipc_server_stop_async(struct ipc_server_data *server_data)
{
        /* ASSERT NOT holding mutex */

        int fd;

        if (!server_data)
                return 0;

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

        pthread_mutex_lock(&server_data->work_available_mutex);

        server_data->shutdown_requested = 1;

        /*
         * Write a byte to the shutdown socket pair to wake up the
         * accept-thread.
         */
        if (write(server_data->accept_thread->fd_send_shutdown, "Q", 1) < 0)
                error_errno("could not write to fd_send_shutdown");

        /*
         * Drain the queue of existing connections.
         */
        while ((fd = fifo_dequeue(server_data)) != -1)
                close(fd);

        /*
         * Gently tell worker threads to stop processing new connections
         * and exit.  (This does not abort in-process conversations.)
         */
        pthread_cond_broadcast(&server_data->work_available_cond);

        pthread_mutex_unlock(&server_data->work_available_mutex);

        trace2_region_leave("ipc-server", "server-stop-async", NULL);

        return 0;
}

/*
 * Wait for all IPC server threads to stop.
 */
int ipc_server_await(struct ipc_server_data *server_data)
{
        pthread_join(server_data->accept_thread->pthread_id, NULL);

        if (!server_data->shutdown_requested)
                BUG("ipc-server: accept-thread stopped for '%s'",
                    server_data->buf_path.buf);

        while (server_data->worker_thread_list) {
                struct ipc_worker_thread_data *wtd =
                        server_data->worker_thread_list;

                pthread_join(wtd->pthread_id, NULL);

                server_data->worker_thread_list = wtd->next_thread;
                free(wtd);
        }

        server_data->is_stopped = 1;

        return 0;
}

void ipc_server_free(struct ipc_server_data *server_data)
{
        struct ipc_accept_thread_data * accept_thread_data;

        if (!server_data)
                return;

        if (!server_data->is_stopped)
                BUG("cannot free ipc-server while running for '%s'",
                    server_data->buf_path.buf);

        accept_thread_data = server_data->accept_thread;
        if (accept_thread_data) {
                unix_ss_free(accept_thread_data->server_socket);

                if (accept_thread_data->fd_send_shutdown != -1)
                        close(accept_thread_data->fd_send_shutdown);
                if (accept_thread_data->fd_wait_shutdown != -1)
                        close(accept_thread_data->fd_wait_shutdown);

                free(server_data->accept_thread);
        }

        while (server_data->worker_thread_list) {
                struct ipc_worker_thread_data *wtd =
                        server_data->worker_thread_list;

                server_data->worker_thread_list = wtd->next_thread;
                free(wtd);
        }

        pthread_cond_destroy(&server_data->work_available_cond);
        pthread_mutex_destroy(&server_data->work_available_mutex);

        strbuf_release(&server_data->buf_path);

        free(server_data->fifo_fds);
        free(server_data);
}