From 39b525b40af455e1ac1767359b7ff81d9f263a7c Mon Sep 17 00:00:00 2001 From: Ken Brown Date: Mon, 24 Dec 2018 10:36:57 -0500 Subject: [PATCH] Allow several writers to open a FIFO This is work in progress. - A FIFO reader opens a Windows named pipe that allows unlimited instances. - Each time the FIFO is opened for writing, a new instance of the pipe is created for the writer to connect to. - This is accomplished by having the reader start a thread that listens for connection attempts. - The reader maintains a list of connected writers. - When the reader wants to read, it polls the writers to see if there's any data to be read. --- winsup/cygwin/fhandler.h | 10 ++- winsup/cygwin/fhandler_fifo.cc | 143 ++++++++++++++++++++------------- winsup/cygwin/pipe.cc | 6 +- 3 files changed, 99 insertions(+), 60 deletions(-) diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h index 9e63867ab..042590883 100644 --- a/winsup/cygwin/fhandler.h +++ b/winsup/cygwin/fhandler.h @@ -1209,7 +1209,8 @@ public: int init (HANDLE, DWORD, mode_t, int64_t); static int create (fhandler_pipe *[2], unsigned, int); static DWORD create (LPSECURITY_ATTRIBUTES, HANDLE *, HANDLE *, DWORD, - const char *, DWORD, int64_t *unique_id = NULL); + const char *, DWORD, int64_t *unique_id = NULL, + DWORD max_instances = 1); fhandler_pipe (void *) {} void copyto (fhandler_base *x) @@ -1229,12 +1230,17 @@ public: } }; +#define MAX_CLIENTS PIPE_UNLIMITED_INSTANCES + class fhandler_fifo: public fhandler_base_overlapped { HANDLE read_ready; HANDLE write_ready; + fhandler_base_overlapped *client[MAX_CLIENTS]; + int nclients; bool __reg2 wait (HANDLE); char __reg2 *fifo_name (char *, const char *); + bool wait_client (); public: fhandler_fifo (); int open (int, mode_t); @@ -1250,6 +1256,8 @@ public: select_record *select_read (select_stuff *); select_record *select_write (select_stuff *); select_record *select_except (select_stuff *); + int add_client (fhandler_base_overlapped *); + void del_client (int); fhandler_fifo (void *) {} diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc index 5733ec778..13d0f80c3 100644 --- a/winsup/cygwin/fhandler_fifo.cc +++ b/winsup/cygwin/fhandler_fifo.cc @@ -23,7 +23,7 @@ fhandler_fifo::fhandler_fifo (): fhandler_base_overlapped (), - read_ready (NULL), write_ready (NULL) + read_ready (NULL), write_ready (NULL), nclients (0) { max_atomic_write = DEFAULT_PIPEBUFSIZE; need_fork_fixup (true); @@ -32,7 +32,8 @@ fhandler_fifo::fhandler_fifo (): #define fnevent(w) fifo_name (npbuf, w "-event") #define fnpipe() fifo_name (npbuf, "fifo") #define create_pipe(r, w) \ - fhandler_pipe::create (sa_buf, (r), (w), 0, fnpipe (), open_mode) + fhandler_pipe::create (sa_buf, (r), (w), 0, fnpipe (), open_mode, NULL, \ + PIPE_UNLIMITED_INSTANCES) char * fhandler_fifo::fifo_name (char *buf, const char *what) @@ -153,36 +154,25 @@ fhandler_fifo::open (int flags, mode_t) res = error_errno_set; goto out; } + else if (!wait_client ()) + { + res = error_errno_set; + goto out; + } - /* 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 (writer) { - int err; - while (1) - if (!wait (read_ready)) - { - res = error_errno_set; - goto out; - } - else if ((err = create_pipe (NULL, &get_io_handle ())) == 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 (!wait (read_ready)) + { + res = error_errno_set; + goto out; + } + else if (!create_pipe (NULL, &get_io_handle ())) + { + debug_printf ("create of writer failed"); + res = error_set_errno; + goto out; + } if (!arm (write_ready)) { res = error_set_errno; @@ -221,6 +211,45 @@ out: return res == success; } +int +fhandler_fifo::add_client (fhandler_base_overlapped *fh) +{ + if (nclients == MAX_CLIENTS) + { + set_errno (EMFILE); + return -1; + } + + client[nclients++] = fh; + return 0; +} + +void +fhandler_fifo::del_client (int i) +{ + client[i]->close (); /* Do we need to call DisconnectNamedPipe? */ + delete client[i]; /* Is that enough cleanup? */ + nclients--; + /* Fill in the hole. */ + if (i < nclients) + memmove (client + i, client + i + 1, + (nclients - i) * sizeof (client[i])); +} + +bool +fhandler_fifo::wait_client () +{ + /* Start a thread that loops forever and does the following: + - waits for a client to connect + - creates a new instance of the pipe + - sets the io_handle of the reader to the new instance + - creates a new fhandler_base_overlapped object whose io_handle + is the previous handle of the reader + - adds a pointer to that object to the client array. + - signals read_ready */ + return true; +} + off_t fhandler_fifo::lseek (off_t offset, int whence) { @@ -282,40 +311,42 @@ fhandler_fifo::wait (HANDLE h) void __reg3 fhandler_fifo::raw_read (void *in_ptr, size_t& len) { - size_t orig_len = len; - for (int i = 0; i < 2; i++) + bool keep_looping = true; + do { - 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 (nclients == 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) + int i = 0; + while (i < nclients) { - debug_printf ("ConnectNamedPipe failed, %E"); - goto errno_out; + size_t orig_len = len; + bool was_nonblocking = client[i]->is_nonblocking (); + client[i]->set_nonblocking (true); + client[i]->raw_read (in_ptr, len); + client[i]->set_nonblocking (was_nonblocking); + if (len > 0) /* Success. */ + { + keep_looping = false; + break; + } + else if (len < 0 && errno != EAGAIN) /* Unexpected error. */ + return; + else if (len == 0) /* Client no longer connected. */ + del_client (i); + else /* No data available to read. */ + i++; + len = orig_len; } - 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. */ + if (is_nonblocking ()) + keep_looping = false; + else + yield (); } + while (keep_looping); return; - -errno_out: - __seterrno (); -errout: - len = -1; } int __reg2 diff --git a/winsup/cygwin/pipe.cc b/winsup/cygwin/pipe.cc index f1eace6a6..94de5f2d0 100644 --- a/winsup/cygwin/pipe.cc +++ b/winsup/cygwin/pipe.cc @@ -209,7 +209,7 @@ fhandler_pipe::dup (fhandler_base *child, int flags) DWORD fhandler_pipe::create (LPSECURITY_ATTRIBUTES sa_ptr, PHANDLE r, PHANDLE w, DWORD psize, const char *name, DWORD open_mode, - int64_t *unique_id) + int64_t *unique_id, DWORD max_instances) { /* Default to error. */ if (r) @@ -274,8 +274,8 @@ fhandler_pipe::create (LPSECURITY_ATTRIBUTES sa_ptr, PHANDLE r, PHANDLE w, definitely required for pty handling since fhandler_pty_master writes to the pipe in chunks, terminated by newline when CANON mode is specified. */ - *r = CreateNamedPipe (pipename, open_mode, pipe_mode, 1, psize, - psize, NMPWAIT_USE_DEFAULT_WAIT, sa_ptr); + *r = CreateNamedPipe (pipename, open_mode, pipe_mode, max_instances, + psize, psize, NMPWAIT_USE_DEFAULT_WAIT, sa_ptr); if (*r != INVALID_HANDLE_VALUE) { -- 2.17.0