rebase: use child_process_clear() to clean

[thirdparty/git.git] / compat / simple-ipc / ipc-win32.c

#include "cache.h"
#include "simple-ipc.h"
#include "strbuf.h"
#include "pkt-line.h"
#include "thread-utils.h"
#include "accctrl.h"
#include "aclapi.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

static int initialize_pipe_name(const char *path, wchar_t *wpath, size_t alloc)
{
        int off = 0;
        struct strbuf realpath = STRBUF_INIT;

        if (!strbuf_realpath(&realpath, path, 0))
                return -1;

        off = swprintf(wpath, alloc, L"\\\\.\\pipe\\");
        if (xutftowcs(wpath + off, realpath.buf, alloc - off) < 0)
                return -1;

        /* Handle drive prefix */
        if (wpath[off] && wpath[off + 1] == L':') {
                wpath[off + 1] = L'_';
                off += 2;
        }

        for (; wpath[off]; off++)
                if (wpath[off] == L'/')
                        wpath[off] = L'\\';

        strbuf_release(&realpath);
        return 0;
}

static enum ipc_active_state get_active_state(wchar_t *pipe_path)
{
        if (WaitNamedPipeW(pipe_path, NMPWAIT_USE_DEFAULT_WAIT))
                return IPC_STATE__LISTENING;

        if (GetLastError() == ERROR_SEM_TIMEOUT)
                return IPC_STATE__NOT_LISTENING;

        if (GetLastError() == ERROR_FILE_NOT_FOUND)
                return IPC_STATE__PATH_NOT_FOUND;

        trace2_data_intmax("ipc-debug", NULL, "getstate/waitpipe/gle",
                           (intmax_t)GetLastError());

        return IPC_STATE__OTHER_ERROR;
}

enum ipc_active_state ipc_get_active_state(const char *path)
{
        wchar_t pipe_path[MAX_PATH];

        if (initialize_pipe_name(path, pipe_path, ARRAY_SIZE(pipe_path)) < 0)
                return IPC_STATE__INVALID_PATH;

        return get_active_state(pipe_path);
}

#define WAIT_STEP_MS (50)

static enum ipc_active_state connect_to_server(
        const wchar_t *wpath,
        DWORD timeout_ms,
        const struct ipc_client_connect_options *options,
        int *pfd)
{
        DWORD t_start_ms, t_waited_ms;
        DWORD step_ms;
        HANDLE hPipe = INVALID_HANDLE_VALUE;
        DWORD mode = PIPE_READMODE_BYTE;
        DWORD gle;

        *pfd = -1;

        for (;;) {
                hPipe = CreateFileW(wpath, GENERIC_READ | GENERIC_WRITE,
                                    0, NULL, OPEN_EXISTING, 0, NULL);
                if (hPipe != INVALID_HANDLE_VALUE)
                        break;

                gle = GetLastError();

                switch (gle) {
                case ERROR_FILE_NOT_FOUND:
                        if (!options->wait_if_not_found)
                                return IPC_STATE__PATH_NOT_FOUND;
                        if (!timeout_ms)
                                return IPC_STATE__PATH_NOT_FOUND;

                        step_ms = (timeout_ms < WAIT_STEP_MS) ?
                                timeout_ms : WAIT_STEP_MS;
                        sleep_millisec(step_ms);

                        timeout_ms -= step_ms;
                        break; /* try again */

                case ERROR_PIPE_BUSY:
                        if (!options->wait_if_busy)
                                return IPC_STATE__NOT_LISTENING;
                        if (!timeout_ms)
                                return IPC_STATE__NOT_LISTENING;

                        t_start_ms = (DWORD)(getnanotime() / 1000000);

                        if (!WaitNamedPipeW(wpath, timeout_ms)) {
                                DWORD gleWait = GetLastError();

                                if (gleWait == ERROR_SEM_TIMEOUT)
                                        return IPC_STATE__NOT_LISTENING;

                                trace2_data_intmax("ipc-debug", NULL,
                                                   "connect/waitpipe/gle",
                                                   (intmax_t)gleWait);

                                return IPC_STATE__OTHER_ERROR;
                        }

                        /*
                         * A pipe server instance became available.
                         * Race other client processes to connect to
                         * it.
                         *
                         * But first decrement our overall timeout so
                         * that we don't starve if we keep losing the
                         * race.  But also guard against special
                         * NPMWAIT_ values (0 and -1).
                         */
                        t_waited_ms = (DWORD)(getnanotime() / 1000000) - t_start_ms;
                        if (t_waited_ms < timeout_ms)
                                timeout_ms -= t_waited_ms;
                        else
                                timeout_ms = 1;
                        break; /* try again */

                default:
                        trace2_data_intmax("ipc-debug", NULL,
                                           "connect/createfile/gle",
                                           (intmax_t)gle);

                        return IPC_STATE__OTHER_ERROR;
                }
        }

        if (!SetNamedPipeHandleState(hPipe, &mode, NULL, NULL)) {
                gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL,
                                   "connect/setpipestate/gle",
                                   (intmax_t)gle);

                CloseHandle(hPipe);
                return IPC_STATE__OTHER_ERROR;
        }

        *pfd = _open_osfhandle((intptr_t)hPipe, O_RDWR|O_BINARY);
        if (*pfd < 0) {
                gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL,
                                   "connect/openosfhandle/gle",
                                   (intmax_t)gle);

                CloseHandle(hPipe);
                return IPC_STATE__OTHER_ERROR;
        }

        /* fd now owns hPipe */

        return IPC_STATE__LISTENING;
}

/*
 * The default connection timeout for Windows clients.
 *
 * This is not currently part of the ipc_ API (nor the config settings)
 * because of differences between Windows and other platforms.
 *
 * This value was chosen at random.
 */
#define WINDOWS_CONNECTION_TIMEOUT_MS (30000)

enum ipc_active_state ipc_client_try_connect(
        const char *path,
        const struct ipc_client_connect_options *options,
        struct ipc_client_connection **p_connection)
{
        wchar_t wpath[MAX_PATH];
        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);

        if (initialize_pipe_name(path, wpath, ARRAY_SIZE(wpath)) < 0)
                state = IPC_STATE__INVALID_PATH;
        else
                state = connect_to_server(wpath, WINDOWS_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;
        }

        FlushFileBuffers((HANDLE)_get_osfhandle(connection->fd));

        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 *response)
{
        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,
                                                    response);

        ipc_client_close_connection(connection);

        return ret;
}

/*
 * Duplicate the given pipe handle and wrap it in a file descriptor so
 * that we can use pkt-line on it.
 */
static int dup_fd_from_pipe(const HANDLE pipe)
{
        HANDLE process = GetCurrentProcess();
        HANDLE handle;
        int fd;

        if (!DuplicateHandle(process, pipe, process, &handle, 0, FALSE,
                             DUPLICATE_SAME_ACCESS)) {
                errno = err_win_to_posix(GetLastError());
                return -1;
        }

        fd = _open_osfhandle((intptr_t)handle, O_RDWR|O_BINARY);
        if (fd < 0) {
                errno = err_win_to_posix(GetLastError());
                CloseHandle(handle);
                return -1;
        }

        /*
         * `handle` is now owned by `fd` and will be automatically closed
         * when the descriptor is closed.
         */

        return fd;
}

/*
 * 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_SERVER_THREAD_DATA,
        MAGIC_SERVER_DATA,
};

struct ipc_server_reply_data {
        enum magic magic;
        int fd;
        struct ipc_server_thread_data *server_thread_data;
};

struct ipc_server_thread_data {
        enum magic magic;
        struct ipc_server_thread_data *next_thread;
        struct ipc_server_data *server_data;
        pthread_t pthread_id;
        HANDLE hPipe;
};

/*
 * On Windows, the conceptual "ipc-server" is implemented as a pool of
 * n idential/peer "server-thread" threads.  That is, there is no
 * hierarchy of threads; and therefore no controller thread managing
 * the pool.  Each thread has an independent handle to the named pipe,
 * receives incoming connections, processes the client, and re-uses
 * the pipe for the next client connection.
 *
 * Therefore, the "ipc-server" only needs to maintain a list of the
 * spawned threads for eventual "join" purposes.
 *
 * A single "stop-event" is visible to all of the server threads to
 * tell them to shutdown (when idle).
 */
struct ipc_server_data {
        enum magic magic;
        ipc_server_application_cb *application_cb;
        void *application_data;
        struct strbuf buf_path;
        wchar_t wpath[MAX_PATH];

        HANDLE hEventStopRequested;
        struct ipc_server_thread_data *thread_list;
        int is_stopped;
};

enum connect_result {
        CR_CONNECTED = 0,
        CR_CONNECT_PENDING,
        CR_CONNECT_ERROR,
        CR_WAIT_ERROR,
        CR_SHUTDOWN,
};

static enum connect_result queue_overlapped_connect(
        struct ipc_server_thread_data *server_thread_data,
        OVERLAPPED *lpo)
{
        if (ConnectNamedPipe(server_thread_data->hPipe, lpo))
                goto failed;

        switch (GetLastError()) {
        case ERROR_IO_PENDING:
                return CR_CONNECT_PENDING;

        case ERROR_PIPE_CONNECTED:
                SetEvent(lpo->hEvent);
                return CR_CONNECTED;

        default:
                break;
        }

failed:
        error(_("ConnectNamedPipe failed for '%s' (%lu)"),
              server_thread_data->server_data->buf_path.buf,
              GetLastError());
        return CR_CONNECT_ERROR;
}

/*
 * Use Windows Overlapped IO to wait for a connection or for our event
 * to be signalled.
 */
static enum connect_result wait_for_connection(
        struct ipc_server_thread_data *server_thread_data,
        OVERLAPPED *lpo)
{
        enum connect_result r;
        HANDLE waitHandles[2];
        DWORD dwWaitResult;

        r = queue_overlapped_connect(server_thread_data, lpo);
        if (r != CR_CONNECT_PENDING)
                return r;

        waitHandles[0] = server_thread_data->server_data->hEventStopRequested;
        waitHandles[1] = lpo->hEvent;

        dwWaitResult = WaitForMultipleObjects(2, waitHandles, FALSE, INFINITE);
        switch (dwWaitResult) {
        case WAIT_OBJECT_0 + 0:
                return CR_SHUTDOWN;

        case WAIT_OBJECT_0 + 1:
                ResetEvent(lpo->hEvent);
                return CR_CONNECTED;

        default:
                return CR_WAIT_ERROR;
        }
}

/*
 * 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);
}

/*
 * 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.
 *
 * Simple-IPC only contains one round trip, so we flush and close
 * here after the response.
 */
static int do_io(struct ipc_server_thread_data *server_thread_data)
{
        struct strbuf buf = STRBUF_INIT;
        struct ipc_server_reply_data reply_data;
        int ret = 0;

        reply_data.magic = MAGIC_SERVER_REPLY_DATA;
        reply_data.server_thread_data = server_thread_data;

        reply_data.fd = dup_fd_from_pipe(server_thread_data->hPipe);
        if (reply_data.fd < 0)
                return error(_("could not create fd from pipe for '%s'"),
                             server_thread_data->server_data->buf_path.buf);

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

                packet_flush_gently(reply_data.fd);

                FlushFileBuffers((HANDLE)_get_osfhandle((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;
}

/*
 * Handle IPC request and response with this connected client.  And reset
 * the pipe to prepare for the next client.
 */
static int use_connection(struct ipc_server_thread_data *server_thread_data)
{
        int ret;

        ret = do_io(server_thread_data);

        FlushFileBuffers(server_thread_data->hPipe);
        DisconnectNamedPipe(server_thread_data->hPipe);

        return ret;
}

/*
 * Thread proc for an IPC server worker thread.  It handles a series of
 * connections from clients.  It cleans and reuses the hPipe between each
 * client.
 */
static void *server_thread_proc(void *_server_thread_data)
{
        struct ipc_server_thread_data *server_thread_data = _server_thread_data;
        HANDLE hEventConnected = INVALID_HANDLE_VALUE;
        OVERLAPPED oConnect;
        enum connect_result cr;
        int ret;

        assert(server_thread_data->hPipe != INVALID_HANDLE_VALUE);

        trace2_thread_start("ipc-server");
        trace2_data_string("ipc-server", NULL, "pipe",
                           server_thread_data->server_data->buf_path.buf);

        hEventConnected = CreateEventW(NULL, TRUE, FALSE, NULL);

        memset(&oConnect, 0, sizeof(oConnect));
        oConnect.hEvent = hEventConnected;

        for (;;) {
                cr = wait_for_connection(server_thread_data, &oConnect);

                switch (cr) {
                case CR_SHUTDOWN:
                        goto finished;

                case CR_CONNECTED:
                        ret = use_connection(server_thread_data);
                        if (ret == SIMPLE_IPC_QUIT) {
                                ipc_server_stop_async(
                                        server_thread_data->server_data);
                                goto finished;
                        }
                        if (ret > 0) {
                                /*
                                 * Ignore (transient) IO errors with this
                                 * client and reset for the next client.
                                 */
                        }
                        break;

                case CR_CONNECT_PENDING:
                        /* By construction, this should not happen. */
                        BUG("ipc-server[%s]: unexpeced CR_CONNECT_PENDING",
                            server_thread_data->server_data->buf_path.buf);

                case CR_CONNECT_ERROR:
                case CR_WAIT_ERROR:
                        /*
                         * Ignore these theoretical errors.
                         */
                        DisconnectNamedPipe(server_thread_data->hPipe);
                        break;

                default:
                        BUG("unandled case after wait_for_connection");
                }
        }

finished:
        CloseHandle(server_thread_data->hPipe);
        CloseHandle(hEventConnected);

        trace2_thread_exit();
        return NULL;
}

/*
 * We need to build a Windows "SECURITY_ATTRIBUTES" object and use it
 * to apply an ACL when we create the initial instance of the Named
 * Pipe.  The construction is somewhat involved and consists of
 * several sequential steps and intermediate objects.
 *
 * We use this structure to hold these intermediate pointers so that
 * we can free them as a group.  (It is unclear from the docs whether
 * some of these intermediate pointers can be freed before we are
 * finished using the "lpSA" member.)
 */
struct my_sa_data
{
        PSID pEveryoneSID;
        PACL pACL;
        PSECURITY_DESCRIPTOR pSD;
        LPSECURITY_ATTRIBUTES lpSA;
};

static void init_sa(struct my_sa_data *d)
{
        memset(d, 0, sizeof(*d));
}

static void release_sa(struct my_sa_data *d)
{
        if (d->pEveryoneSID)
                FreeSid(d->pEveryoneSID);
        if (d->pACL)
                LocalFree(d->pACL);
        if (d->pSD)
                LocalFree(d->pSD);
        if (d->lpSA)
                LocalFree(d->lpSA);

        memset(d, 0, sizeof(*d));
}

/*
 * Create SECURITY_ATTRIBUTES to apply to the initial named pipe.  The
 * creator of the first server instance gets to set the ACLs on it.
 *
 * We allow the well-known group `EVERYONE` to have read+write access
 * to the named pipe so that clients can send queries to the daemon
 * and receive the response.
 *
 * Normally, this is not necessary since the daemon is usually
 * automatically started by a foreground command like `git status`,
 * but in those cases where an elevated Git command started the daemon
 * (such that the daemon itself runs with elevation), we need to add
 * the ACL so that non-elevated commands can write to it.
 *
 * The following document was helpful:
 * https://docs.microsoft.com/en-us/windows/win32/secauthz/creating-a-security-descriptor-for-a-new-object-in-c--
 *
 * Returns d->lpSA set to a SA or NULL.
 */
static LPSECURITY_ATTRIBUTES get_sa(struct my_sa_data *d)
{
        SID_IDENTIFIER_AUTHORITY sid_auth_world = SECURITY_WORLD_SID_AUTHORITY;
#define NR_EA (1)
        EXPLICIT_ACCESS ea[NR_EA];
        DWORD dwResult;

        if (!AllocateAndInitializeSid(&sid_auth_world, 1,
                                      SECURITY_WORLD_RID, 0,0,0,0,0,0,0,
                                      &d->pEveryoneSID)) {
                DWORD gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL, "alloc-world-sid/gle",
                                   (intmax_t)gle);
                goto fail;
        }

        memset(ea, 0, NR_EA * sizeof(EXPLICIT_ACCESS));

        ea[0].grfAccessPermissions = GENERIC_READ | GENERIC_WRITE;
        ea[0].grfAccessMode = SET_ACCESS;
        ea[0].grfInheritance = NO_INHERITANCE;
        ea[0].Trustee.MultipleTrusteeOperation = NO_MULTIPLE_TRUSTEE;
        ea[0].Trustee.TrusteeForm = TRUSTEE_IS_SID;
        ea[0].Trustee.TrusteeType = TRUSTEE_IS_WELL_KNOWN_GROUP;
        ea[0].Trustee.ptstrName = (LPTSTR)d->pEveryoneSID;

        dwResult = SetEntriesInAcl(NR_EA, ea, NULL, &d->pACL);
        if (dwResult != ERROR_SUCCESS) {
                DWORD gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL, "set-acl-entry/gle",
                                   (intmax_t)gle);
                trace2_data_intmax("ipc-debug", NULL, "set-acl-entry/dw",
                                   (intmax_t)dwResult);
                goto fail;
        }

        d->pSD = (PSECURITY_DESCRIPTOR)LocalAlloc(
                LPTR, SECURITY_DESCRIPTOR_MIN_LENGTH);
        if (!InitializeSecurityDescriptor(d->pSD, SECURITY_DESCRIPTOR_REVISION)) {
                DWORD gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL, "init-sd/gle", (intmax_t)gle);
                goto fail;
        }

        if (!SetSecurityDescriptorDacl(d->pSD, TRUE, d->pACL, FALSE)) {
                DWORD gle = GetLastError();
                trace2_data_intmax("ipc-debug", NULL, "set-sd-dacl/gle", (intmax_t)gle);
                goto fail;
        }

        d->lpSA = (LPSECURITY_ATTRIBUTES)LocalAlloc(LPTR, sizeof(SECURITY_ATTRIBUTES));
        d->lpSA->nLength = sizeof(SECURITY_ATTRIBUTES);
        d->lpSA->lpSecurityDescriptor = d->pSD;
        d->lpSA->bInheritHandle = FALSE;

        return d->lpSA;

fail:
        release_sa(d);
        return NULL;
}

static HANDLE create_new_pipe(wchar_t *wpath, int is_first)
{
        HANDLE hPipe;
        DWORD dwOpenMode, dwPipeMode;
        struct my_sa_data my_sa_data;

        init_sa(&my_sa_data);

        dwOpenMode = PIPE_ACCESS_INBOUND | PIPE_ACCESS_OUTBOUND |
                FILE_FLAG_OVERLAPPED;

        dwPipeMode = PIPE_TYPE_MESSAGE | PIPE_READMODE_BYTE | PIPE_WAIT |
                PIPE_REJECT_REMOTE_CLIENTS;

        if (is_first) {
                dwOpenMode |= FILE_FLAG_FIRST_PIPE_INSTANCE;

                /*
                 * On Windows, the first server pipe instance gets to
                 * set the ACL / Security Attributes on the named
                 * pipe; subsequent instances inherit and cannot
                 * change them.
                 */
                get_sa(&my_sa_data);
        }

        hPipe = CreateNamedPipeW(wpath, dwOpenMode, dwPipeMode,
                                 PIPE_UNLIMITED_INSTANCES, 1024, 1024, 0,
                                 my_sa_data.lpSA);

        release_sa(&my_sa_data);

        return hPipe;
}

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 ipc_server_data *server_data;
        wchar_t wpath[MAX_PATH];
        HANDLE hPipeFirst = INVALID_HANDLE_VALUE;
        int k;
        int ret = 0;
        int nr_threads = opts->nr_threads;

        *returned_server_data = NULL;

        ret = initialize_pipe_name(path, wpath, ARRAY_SIZE(wpath));
        if (ret < 0) {
                errno = EINVAL;
                return -1;
        }

        hPipeFirst = create_new_pipe(wpath, 1);
        if (hPipeFirst == INVALID_HANDLE_VALUE) {
                errno = EADDRINUSE;
                return -2;
        }

        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;
        server_data->hEventStopRequested = CreateEvent(NULL, TRUE, FALSE, NULL);
        strbuf_init(&server_data->buf_path, 0);
        strbuf_addstr(&server_data->buf_path, path);
        wcscpy(server_data->wpath, wpath);

        if (nr_threads < 1)
                nr_threads = 1;

        for (k = 0; k < nr_threads; k++) {
                struct ipc_server_thread_data *std;

                std = xcalloc(1, sizeof(*std));
                std->magic = MAGIC_SERVER_THREAD_DATA;
                std->server_data = server_data;
                std->hPipe = INVALID_HANDLE_VALUE;

                std->hPipe = (k == 0)
                        ? hPipeFirst
                        : create_new_pipe(server_data->wpath, 0);

                if (std->hPipe == INVALID_HANDLE_VALUE) {
                        /*
                         * If we've reached a pipe instance limit for
                         * this path, just use fewer threads.
                         */
                        free(std);
                        break;
                }

                if (pthread_create(&std->pthread_id, NULL,
                                   server_thread_proc, std)) {
                        /*
                         * Likewise, if we're out of threads, just use
                         * fewer threads than requested.
                         *
                         * However, we just give up if we can't even get
                         * one thread.  This should not happen.
                         */
                        if (k == 0)
                                die(_("could not start thread[0] for '%s'"),
                                    path);

                        CloseHandle(std->hPipe);
                        free(std);
                        break;
                }

                std->next_thread = server_data->thread_list;
                server_data->thread_list = std;
        }

        *returned_server_data = server_data;
        return 0;
}

int ipc_server_stop_async(struct ipc_server_data *server_data)
{
        if (!server_data)
                return 0;

        /*
         * Gently tell all of the ipc_server threads to shutdown.
         * This will be seen the next time they are idle (and waiting
         * for a connection).
         *
         * We DO NOT attempt to force them to drop an active connection.
         */
        SetEvent(server_data->hEventStopRequested);
        return 0;
}

int ipc_server_await(struct ipc_server_data *server_data)
{
        DWORD dwWaitResult;

        if (!server_data)
                return 0;

        dwWaitResult = WaitForSingleObject(server_data->hEventStopRequested, INFINITE);
        if (dwWaitResult != WAIT_OBJECT_0)
                return error(_("wait for hEvent failed for '%s'"),
                             server_data->buf_path.buf);

        while (server_data->thread_list) {
                struct ipc_server_thread_data *std = server_data->thread_list;

                pthread_join(std->pthread_id, NULL);

                server_data->thread_list = std->next_thread;
                free(std);
        }

        server_data->is_stopped = 1;

        return 0;
}

void ipc_server_free(struct ipc_server_data *server_data)
{
        if (!server_data)
                return;

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

        strbuf_release(&server_data->buf_path);

        if (server_data->hEventStopRequested != INVALID_HANDLE_VALUE)
                CloseHandle(server_data->hEventStopRequested);

        while (server_data->thread_list) {
                struct ipc_server_thread_data *std = server_data->thread_list;

                server_data->thread_list = std->next_thread;
                free(std);
        }

        free(server_data);
}