From 261fe9fe39e3e0587df302960f6052ebc2dd514a Mon Sep 17 00:00:00 2001 From: Ken Brown Date: Sun, 13 Jan 2019 15:11:12 -0500 Subject: [PATCH FIFO, draft 2/4] Cygwin: fhandler_fifo: Allow multiple writers Introduce a 'pipe_instance' structure that can be used by a reader to communicate with a writer using an instance of the named pipe. An fhandler_fifo opened for reading creates a thread that does the following: - maintains a list of pipe_instances - listens for clients trying to connect - creates new pipe_instances as needed so that there's always at least one available for connecting. fhandler_fifo::raw_read now loops through the connected pipe_instances and reads from the first one that has data available. --- winsup/cygwin/fhandler.h | 17 ++ winsup/cygwin/fhandler_fifo.cc | 387 ++++++++++++++++++++++++++------- 2 files changed, 320 insertions(+), 84 deletions(-) diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h index 5ff00c4d7..311cf0958 100644 --- a/winsup/cygwin/fhandler.h +++ b/winsup/cygwin/fhandler.h @@ -1233,14 +1233,31 @@ public: } }; +#define MAX_INSTANCES PIPE_UNLIMITED_INSTANCES + class fhandler_fifo: public fhandler_base_overlapped { + struct pipe_instance + { + fhandler_base_overlapped *fh; + bool connected; + HANDLE dummy_evt; /* Never signaled. */ + }; HANDLE read_ready; HANDLE write_ready; + HANDLE listen_client_thr; + HANDLE lct_termination_evt; + pipe_instance inst[MAX_INSTANCES]; + int ninstances, nconnected; bool __reg2 wait (HANDLE); char __reg2 *fifo_name (char *, const char *); + int disconnect_and_reconnect (int); + int create_pipe_instance (); + bool listen_client (); + bool check_listen_client_thread () const; public: fhandler_fifo (); + DWORD listen_client_thread (); int open (int, mode_t); off_t lseek (off_t offset, int whence); int close (); diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc index b171719b1..13d87eab5 100644 --- a/winsup/cygwin/fhandler_fifo.cc +++ b/winsup/cygwin/fhandler_fifo.cc @@ -22,8 +22,9 @@ #include "cygwait.h" fhandler_fifo::fhandler_fifo (): - fhandler_base_overlapped (), - read_ready (NULL), write_ready (NULL) + fhandler_base_overlapped (), read_ready (NULL), write_ready (NULL), + listen_client_thr (NULL), lct_termination_evt (NULL), ninstances (0), + nconnected (0) { max_atomic_write = DEFAULT_PIPEBUFSIZE; need_fork_fixup (true); @@ -74,6 +75,198 @@ fhandler_fifo::arm (HANDLE h) return res; } +static int +connect_to_new_client (fhandler_base_overlapped *fh) +{ + bool connected; + bool res = ConnectNamedPipe (fh->get_handle (), fh->get_overlapped ()); + if (res) + return -1; + switch (GetLastError ()) + { + case ERROR_IO_PENDING: + connected = false; + break; + case ERROR_PIPE_CONNECTED: + connected = true; + ResetEvent (fh->get_overlapped ()->hEvent); + break; + default: + return -1; + } + return connected; +} + +int +fhandler_fifo::disconnect_and_reconnect (int i) +{ + int connected; + + inst[i].connected = false; + nconnected--; + if (!DisconnectNamedPipe (inst[i].fh->get_io_handle ())) + { + debug_printf ("DisconnectNamedPipe failed, %E"); + goto errout; + } + else if ((connected = connect_to_new_client (inst[i].fh)) < 0) + goto errout; + else + inst[i].connected = connected; + if (connected) + nconnected++; + return 0; +errout: + __seterrno (); + return -1; +} + +int +fhandler_fifo::create_pipe_instance () +{ + pipe_instance pi; + fhandler_base_overlapped *fh; + int connected; + bool first_instance = (ninstances == 0); + DWORD open_mode = FILE_FLAG_OVERLAPPED; + + if (ninstances == MAX_INSTANCES) + { + set_errno (EMFILE); + return -1; + } + if (!(pi.dummy_evt = CreateEvent (NULL, true, false, NULL))) + { + __seterrno (); + return -1; + } + if (!(fh = (fhandler_base_overlapped *) build_fh_dev (dev ()))) + { + set_errno (EMFILE); + return -1; + } + pi.fh = fh; + char char_sa_buf[1024]; + LPSECURITY_ATTRIBUTES sa_buf; + sa_buf = sec_user_cloexec (get_flags () & O_CLOEXEC, + (PSECURITY_ATTRIBUTES) char_sa_buf, + cygheap->user.sid()); + char npbuf[MAX_PATH]; + HANDLE h; + if (create_pipe (&h, NULL, first_instance)) + goto errout; + fh->set_io_handle (h); + /* To play it safe, make sure this fhandler_fifo has an io_handle. */ + if (first_instance) + set_io_handle (h); + fh->set_flags (get_flags ()); + if (fh->setup_overlapped ()) + goto errout; + if ((connected = connect_to_new_client (fh)) < 0) + goto errout; + if ((pi.connected = connected)) + nconnected++; + inst[ninstances++] = pi; + return 0; +errout: + __seterrno (); + delete fh; + return -1; +} + +/* Just hop to the listen_client_thread method. */ +DWORD WINAPI +listen_client_func (LPVOID param) +{ + fhandler_fifo *fh = (fhandler_fifo *) param; + return fh->listen_client_thread (); +} + +/* Start a thread that listens for client connections. Whenever a new + client connects, it creates a new pipe_instance if necessary. + (There may already be an available instance if a client has + disconnected.) */ +bool +fhandler_fifo::listen_client () +{ + if (!(lct_termination_evt = CreateEvent (NULL, true, false, NULL))) + { + __seterrno (); + return false; + } + + listen_client_thr = CreateThread (NULL, PREFERRED_IO_BLKSIZE, + listen_client_func, (PVOID) this, 0, NULL); + if (!listen_client_thr) + { + __seterrno (); + HANDLE evt = InterlockedExchangePointer (&lct_termination_evt, NULL); + if (evt) + CloseHandle (evt); + return false; + } + return true; +} + +DWORD +fhandler_fifo::listen_client_thread () +{ + while (1) + { + if (nconnected == ninstances && create_pipe_instance () < 0) + return -1; + + /* Wait for a client to connect. */ + HANDLE w[MAX_INSTANCES + 1]; + int i; + DWORD wait_ret; + for (i = 0; i < ninstances; i++) + w[i] = inst[i].connected ? inst[i].dummy_evt + : inst[i].fh->get_overlapped ()->hEvent; + w[ninstances] = lct_termination_evt; + if (!arm (read_ready)) + goto errout; + wait_ret = WaitForMultipleObjects (ninstances + 1, w, false, INFINITE); + i = wait_ret - WAIT_OBJECT_0; + if (i < 0 || i > ninstances) + goto errout; + else if (i == ninstances) /* Reader is closing. */ + return 0; + else + { + inst[i].connected = true; + nconnected++; + } + } +errout: + __seterrno (); + ResetEvent (read_ready); + return -1; +} + +/* Is the thread still running? */ +bool +fhandler_fifo::check_listen_client_thread () const +{ + bool ret = false; + switch (WaitForSingleObject (listen_client_thr, 0)) + { + case WAIT_OBJECT_0: + DWORD err; + GetExitCodeThread (listen_client_thr, &err); + __seterrno_from_win_error (err); + break; + case WAIT_TIMEOUT: + ret = true; + break; + case WAIT_FAILED: + default: + __seterrno (); + break; + } + return ret; +} + int fhandler_fifo::open (int flags, mode_t) { @@ -133,73 +326,81 @@ fhandler_fifo::open (int flags, mode_t) goto out; } - /* If we're reading, create the pipe, signal that we're ready and wait for - a writer. + /* If we're reading, start the listen_client thread (which should + signal read_ready), and wait for a writer. + FIXME: Probably need to special case O_RDWR case. */ - if (!reader) - /* We are not a reader */; - else if (create_pipe (&get_io_handle (), NULL, true)) - { - debug_printf ("create of reader failed"); - res = error_set_errno; - goto out; - } - else if (!arm (read_ready)) - { - res = error_set_errno; - goto out; - } - else if (!duplexer && !wait (write_ready)) + if (reader) { - res = error_errno_set; - goto out; + bool thr_ok = listen_client (); + if (!thr_ok) + { + debug_printf ("create of listen_client thread failed"); + res = error_errno_set; + goto out; + } + /* Wait for the listen_client thread to create the pipe and + signal read_ready. This should be quick. */ + HANDLE w[2] = { listen_client_thr, read_ready }; + switch (WaitForMultipleObjects (2, w, FALSE, INFINITE)) + { + case WAIT_OBJECT_0: + debug_printf ("listen_client_thread exited unexpectedly"); + DWORD err; + GetExitCodeThread (listen_client_thr, &err); + __seterrno_from_win_error (err); + res = error_errno_set; + goto out; + break; + case WAIT_OBJECT_0 + 1: + if (!arm (read_ready)) + { + res = error_set_errno; + goto out; + } + break; + default: + res = error_set_errno; + goto out; + break; + } + if (!duplexer && !wait (write_ready)) + { + res = error_errno_set; + goto out; + } + else + res = success; } - - /* If we're writing, it's a little tricky since it is possible that - we're attempting to open the other end of a pipe which is already - connected. In that case, we detect ERROR_PIPE_BUSY, reset the - read_ready event and wait for the reader to allow us to connect - by signalling read_ready. - - Once the pipe has been set up, we signal write_ready. */ + /* If we're writing, wait for read_ready and then connect to the + pipe. This should always succeed quickly if the reader's + listen_client thread is running. Then signal write_ready. */ if (writer) { - int err; - while (1) - if (!wait (read_ready)) - { - res = error_errno_set; - goto out; - } - else if ((err = create_pipe (NULL, &get_io_handle (), false)) == 0) - break; - else if (err == ERROR_PIPE_BUSY) - { - debug_only_printf ("pipe busy"); - ResetEvent (read_ready); - } - else - { - debug_printf ("create of writer failed"); - res = error_set_errno; - goto out; - } - if (!arm (write_ready)) + if (!wait (read_ready)) + { + res = error_errno_set; + goto out; + } + else if (create_pipe (NULL, &get_io_handle (), false)) { + debug_printf ("create of writer failed"); res = error_set_errno; goto out; } + else if (!arm (write_ready)) + { + res = error_set_errno; + goto out; + } + else if (setup_overlapped () == 0) + res = success; + else + { + debug_printf ("setup_overlapped failed, %E"); + res = error_set_errno; + } } - - /* If setup_overlapped() succeeds (and why wouldn't it?) we are all set. */ - if (setup_overlapped () == 0) - res = success; - else - { - debug_printf ("setup_overlapped failed, %E"); - res = error_set_errno; - } - out: if (res == error_set_errno) __seterrno (); @@ -217,6 +418,8 @@ out: } if (get_io_handle ()) CloseHandle (get_io_handle ()); + if (listen_client_thr) + CloseHandle (listen_client_thr); } debug_printf ("res %d", res); return res == success; @@ -284,37 +487,53 @@ void __reg3 fhandler_fifo::raw_read (void *in_ptr, size_t& len) { size_t orig_len = len; - for (int i = 0; i < 2; i++) + while (1) { - fhandler_base_overlapped::raw_read (in_ptr, len); - if (len || i || WaitForSingleObject (read_ready, 0) != WAIT_OBJECT_0) - break; - /* If we got here, then fhandler_base_overlapped::raw_read returned 0, - indicating "EOF" and something has set read_ready to zero. That means - we should have a client waiting to connect. - FIXME: If the client CTRL-C's the open during this time then this - could hang indefinitely. Maybe implement a timeout? */ - if (!DisconnectNamedPipe (get_io_handle ())) + if (nconnected == 0) /* Return EOF. */ { - debug_printf ("DisconnectNamedPipe failed, %E"); - goto errno_out; + len = 0; + return; } - else if (!ConnectNamedPipe (get_io_handle (), get_overlapped ()) - && GetLastError () != ERROR_IO_PENDING) + if (!check_listen_client_thread ()) + goto errout; + + for (int i = 0; i < ninstances; i++) + if (inst[i].connected) + { + ssize_t nread; + len = orig_len; + bool was_nonblocking = inst[i].fh->is_nonblocking (); + inst[i].fh->set_nonblocking (true); + inst[i].fh->fhandler_base_overlapped::raw_read (in_ptr, len); + inst[i].fh->set_nonblocking (was_nonblocking); + nread = (ssize_t) len; + if (nread > 0) + return; + else if (nread < 0 && errno != EAGAIN) + goto errout; + else if (nread == 0 && disconnect_and_reconnect (i) < 0) + goto errout; + } + if (is_nonblocking ()) { - debug_printf ("ConnectNamedPipe failed, %E"); - goto errno_out; + set_errno (EAGAIN); + goto errout; + } + else + { + /* Allow interruption. Copied from + fhandler_socket_unix::open_reparse_point. */ + pthread_testcancel (); + if (cygwait (NULL, cw_nowait, cw_sig_eintr) == WAIT_SIGNALED + && !_my_tls.call_signal_handler ()) + { + set_errno (EINTR); + goto errout; + } + /* Don't hog the CPU. */ + Sleep (50); } - else if (!arm (read_ready)) - goto errno_out; - else if (!wait (get_overlapped_buffer ()->hEvent)) - goto errout; /* If wait() fails, errno is set so no need to set it */ - len = orig_len; /* Reset since raw_read above set it to zero. */ } - return; - -errno_out: - __seterrno (); errout: len = -1; } -- 2.17.0