[PATCH 00/21] FIFO: Support multiple readers

classic Classic list List threaded Threaded
33 messages Options
12
Reply | Threaded
Open this post in threaded view
|

[PATCH 00/21] FIFO: Support multiple readers

cygwin-patches mailing list
This project began as a an attempt to allow a FIFO to be opened
multiple times for reading.  The initial motivation was that Midnight
Commander running under tcsh does this (unsuccessfully on Cygwin).
See

   https://sourceware.org/pipermail/cygwin/2019-December/243317.html

It quickly became apparent, however, that the current code doesn't
even properly handle the case where the FIFO is *explicitly* opened
only once for reading, but additional readers are created via
dup/fork/exec.

This explained some of the bugs reported by Kristian Ivarsson.  See,
for example, the thread starting here:

  https://sourceware.org/pipermail/cygwin/2020-March/000206.html

as well as later similar threads.

[The discussion continued in private email, with many bug reports and
test programs by Kristian.  I'm very grateful to him for his reports
and testing.]

The first 10 patches in this series make some improvements and bug
fixes that came up along the way and don't specifically relate to
multiple readers.  The next 10 patches, with the exception of "allow
fc_handler list to grow dynamically", add the support for multiple
readers.  The last one updates the commentary at the beginning of
fhandler_fifo.cc that tries to explain how it all works.

The key ideas in these patches are:

1. Use shared memory, so that all readers have the necessary
information about the writers that are open.

2. Designate one reader as the "owner".  This reader runs a thread
that listens for connections and keeps track of the writers.

3. Use a second shared memory block to be used for transfer of
ownership.  Ownership must be transferred when the owner closes or
execs.  And a reader that wants to read or run select must take
ownership in order to be able to poll the writers for input.

The patches in this series have been applied to the topic/fifo branch
in case it's easier to review/test them there.

Ken Brown (21):
  Cygwin: FIFO: minor change - use NtClose
  Cygwin: FIFO: simplify the fifo_client_handler structure
  Cygwin: FIFO: change the fifo_client_connect_state enum
  Cygwin: FIFO: simplify the listen_client_thread code
  Cygwin: FIFO: remove the arm method
  Cygwin: FIFO: honor the flags argument in dup
  Cygwin: FIFO: dup/fork/exec: make sure child starts unlocked
  Cygwin: FIFO: fix hit_eof
  Cygwin: FIFO: make opening a writer more robust
  Cygwin: FIFO: use a cygthread instead of a homemade thread
  Cygwin: FIFO: add shared memory
  Cygwin: FIFO: keep track of the number of readers
  Cygwin: FIFO: introduce a new type, fifo_reader_id_t
  Cygwin: FIFO: designate one reader as owner
  Cygwin: FIFO: allow fc_handler list to grow dynamically
  Cygwin: FIFO: add a shared fifo_client_handler list
  Cygwin: FIFO: take ownership on exec
  Cygwin: FIFO: find a new owner when closing
  Cygwin: FIFO: allow any reader to take ownership
  Cygwin: FIFO: support opening multiple readers
  Cygwin: FIFO: update commentary

 winsup/cygwin/fhandler.h       |  208 ++++-
 winsup/cygwin/fhandler_fifo.cc | 1564 ++++++++++++++++++++++----------
 winsup/cygwin/select.cc        |   48 +-
 3 files changed, 1311 insertions(+), 509 deletions(-)

--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 01/21] Cygwin: FIFO: minor change - use NtClose

cygwin-patches mailing list
Replace CloseHandle by NtClose since all handles are created by NT
functions.
---
 winsup/cygwin/fhandler_fifo.cc | 32 ++++++++++++++++----------------
 1 file changed, 16 insertions(+), 16 deletions(-)

diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 19cd0e507..c091b0add 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -319,7 +319,7 @@ fhandler_fifo::listen_client ()
       __seterrno ();
       HANDLE evt = InterlockedExchangePointer (&lct_termination_evt, NULL);
       if (evt)
- CloseHandle (evt);
+ NtClose (evt);
       return false;
     }
   return true;
@@ -441,7 +441,7 @@ fhandler_fifo::listen_client_thread ()
       ret = -1;
     }
   if (ph)
-    CloseHandle (ph);
+    NtClose (ph);
   fifo_client_unlock ();
   goto out;
  default:
@@ -462,7 +462,7 @@ fhandler_fifo::listen_client_thread ()
     }
 out:
   if (evt)
-    CloseHandle (evt);
+    NtClose (evt);
   ResetEvent (read_ready);
   if (ret < 0)
     debug_printf ("exiting with error, %E");
@@ -617,16 +617,16 @@ out:
     {
       if (read_ready)
  {
-  CloseHandle (read_ready);
+  NtClose (read_ready);
   read_ready = NULL;
  }
       if (write_ready)
  {
-  CloseHandle (write_ready);
+  NtClose (write_ready);
   write_ready = NULL;
  }
       if (get_handle ())
- CloseHandle (get_handle ());
+ NtClose (get_handle ());
       if (listen_client_thr)
  stop_listen_client ();
     }
@@ -775,7 +775,7 @@ fhandler_fifo::raw_write (const void *ptr, size_t len)
  ret = nbytes;
     }
   if (evt)
-    CloseHandle (evt);
+    NtClose (evt);
   if (status == STATUS_THREAD_SIGNALED && ret < 0)
     set_errno (EINTR);
   else if (status == STATUS_THREAD_CANCELED)
@@ -819,7 +819,7 @@ fhandler_fifo::check_listen_client_thread ()
       switch (waitret)
  {
  case WAIT_OBJECT_0:
-  CloseHandle (listen_client_thr);
+  NtClose (listen_client_thr);
   break;
  case WAIT_TIMEOUT:
   ret = 1;
@@ -828,7 +828,7 @@ fhandler_fifo::check_listen_client_thread ()
   debug_printf ("WaitForSingleObject failed, %E");
   ret = -1;
   __seterrno ();
-  CloseHandle (listen_client_thr);
+  NtClose (listen_client_thr);
   break;
  }
     }
@@ -1001,11 +1001,11 @@ fhandler_fifo::stop_listen_client ()
   ret = -1;
   debug_printf ("listen_client_thread exited with error");
  }
-      CloseHandle (thr);
+      NtClose (thr);
     }
   evt = InterlockedExchangePointer (&lct_termination_evt, NULL);
   if (evt)
-    CloseHandle (evt);
+    NtClose (evt);
   return ret;
 }
 
@@ -1017,9 +1017,9 @@ fhandler_fifo::close ()
   fifo_client_unlock ();
   int ret = stop_listen_client ();
   if (read_ready)
-    CloseHandle (read_ready);
+    NtClose (read_ready);
   if (write_ready)
-    CloseHandle (write_ready);
+    NtClose (write_ready);
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
     if (fc_handler[i].close () < 0)
@@ -1070,7 +1070,7 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  GetCurrentProcess (), &fhf->write_ready,
  0, true, DUPLICATE_SAME_ACCESS))
     {
-      CloseHandle (fhf->read_ready);
+      NtClose (fhf->read_ready);
       fhf->close ();
       __seterrno ();
       goto out;
@@ -1084,8 +1084,8 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
     0, true, DUPLICATE_SAME_ACCESS))
  {
   fifo_client_unlock ();
-  CloseHandle (fhf->read_ready);
-  CloseHandle (fhf->write_ready);
+  NtClose (fhf->read_ready);
+  NtClose (fhf->write_ready);
   fhf->close ();
   __seterrno ();
   goto out;
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 02/21] Cygwin: FIFO: simplify the fifo_client_handler structure

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Replace the 'fhandler_base *' member by a HANDLE to the server side of
the Windows named pipe instance.  Make the corresponding
simplifications throughout.
---
 winsup/cygwin/fhandler.h       | 19 +++-------
 winsup/cygwin/fhandler_fifo.cc | 65 ++++++++--------------------------
 2 files changed, 19 insertions(+), 65 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 1c7336370..e841f96ac 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1284,10 +1284,10 @@ enum
 
 struct fifo_client_handler
 {
-  fhandler_base *fh;
+  HANDLE h;
   fifo_client_connect_state state;
-  fifo_client_handler () : fh (NULL), state (fc_unknown) {}
-  int close ();
+  fifo_client_handler () : h (NULL), state (fc_unknown) {}
+  void close () { NtClose (h); }
 /* Returns FILE_PIPE_DISCONNECTED_STATE, FILE_PIPE_LISTENING_STATE,
    FILE_PIPE_CONNECTED_STATE, FILE_PIPE_CLOSING_STATE,
    FILE_PIPE_INPUT_AVAILABLE_STATE, or -1 on error. */
@@ -1312,7 +1312,7 @@ class fhandler_fifo: public fhandler_base
   HANDLE create_pipe_instance (bool);
   NTSTATUS open_pipe (HANDLE&);
   int add_client_handler ();
-  int delete_client_handler (int);
+  void delete_client_handler (int);
   bool listen_client ();
   int stop_listen_client ();
   int check_listen_client_thread ();
@@ -1321,8 +1321,7 @@ public:
   fhandler_fifo ();
   bool hit_eof ();
   int get_nhandlers () const { return nhandlers; }
-  HANDLE get_fc_handle (int i) const
-  { return fc_handler[i].fh->get_handle (); }
+  HANDLE get_fc_handle (int i) const { return fc_handler[i].h; }
   bool is_connected (int i) const
   { return fc_handler[i].state == fc_connected; }
   PUNICODE_STRING get_pipe_name ();
@@ -1345,12 +1344,6 @@ public:
   void fixup_after_fork (HANDLE);
   void fixup_after_exec ();
   int __reg2 fstatvfs (struct statvfs *buf);
-  void clear_readahead ()
-  {
-    fhandler_base::clear_readahead ();
-    for (int i = 0; i < nhandlers; i++)
-      fc_handler[i].fh->clear_readahead ();
-  }
   select_record *select_read (select_stuff *);
   select_record *select_write (select_stuff *);
   select_record *select_except (select_stuff *);
@@ -1374,8 +1367,6 @@ public:
     /* fhf->pipe_name_buf is a *copy* of this->pipe_name_buf, but
        fhf->pipe_name.Buffer == this->pipe_name_buf. */
     fhf->pipe_name.Buffer = fhf->pipe_name_buf;
-    for (int i = 0; i < nhandlers; i++)
-      fhf->fc_handler[i].fh = fc_handler[i].fh->clone ();
     return fhf;
   }
 };
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index c091b0add..6b71dd950 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -252,7 +252,6 @@ fhandler_fifo::add_client_handler ()
 {
   int ret = -1;
   fifo_client_handler fc;
-  fhandler_base *fh;
   HANDLE ph = NULL;
   bool first = (nhandlers == 0);
 
@@ -261,40 +260,26 @@ fhandler_fifo::add_client_handler ()
       set_errno (EMFILE);
       goto out;
     }
-  if (!(fh = build_fh_dev (dev ())))
-    {
-      set_errno (EMFILE);
-      goto out;
-    }
   ph = create_pipe_instance (first);
   if (!ph)
-    {
-      delete fh;
-      goto out;
-    }
+    goto out;
   else
     {
-      fh->set_handle (ph);
-      fh->set_flags ((openflags & ~O_ACCMODE) | O_RDONLY);
-      fh->set_nonblocking (false);
       ret = 0;
-      fc.fh = fh;
-      fifo_client_lock ();
+      fc.h = ph;
       fc_handler[nhandlers++] = fc;
-      fifo_client_unlock ();
     }
 out:
   return ret;
 }
 
-int
+void
 fhandler_fifo::delete_client_handler (int i)
 {
-  int ret = fc_handler[i].close ();
+  fc_handler[i].close ();
   if (i < --nhandlers)
     memmove (fc_handler + i, fc_handler + i + 1,
      (nhandlers - i) * sizeof (fc_handler[i]));
-  return ret;
 }
 
 /* Just hop to the listen_client_thread method. */
@@ -331,8 +316,7 @@ fhandler_fifo::record_connection (fifo_client_handler& fc)
   SetEvent (write_ready);
   fc.state = fc_connected;
   nconnected++;
-  fc.fh->set_nonblocking (true);
-  set_pipe_non_blocking (fc.fh->get_handle (), true);
+  set_pipe_non_blocking (fc.h, true);
 }
 
 DWORD
@@ -355,13 +339,7 @@ fhandler_fifo::listen_client_thread ()
       while (i < nhandlers)
  {
   if (fc_handler[i].state == fc_invalid)
-    {
-      if (delete_client_handler (i) < 0)
- {
-  fifo_client_unlock ();
-  goto out;
- }
-    }
+    delete_client_handler (i);
   else
     i++;
  }
@@ -383,7 +361,7 @@ fhandler_fifo::listen_client_thread ()
       NTSTATUS status;
       IO_STATUS_BLOCK io;
 
-      status = NtFsControlFile (fc.fh->get_handle (), evt, NULL, NULL, &io,
+      status = NtFsControlFile (fc.h, evt, NULL, NULL, &io,
  FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
       if (status == STATUS_PENDING)
  {
@@ -424,8 +402,7 @@ fhandler_fifo::listen_client_thread ()
       && (NT_SUCCESS (io.Status) || io.Status == STATUS_PIPE_CONNECTED))
     {
       debug_printf ("successfully connected bogus client");
-      if (delete_client_handler (nhandlers - 1) < 0)
- ret = -1;
+      delete_client_handler (nhandlers - 1);
     }
   else if ((ps = fc.pipe_state ()) == FILE_PIPE_CONNECTED_STATE
    || ps == FILE_PIPE_INPUT_AVAILABLE_STATE)
@@ -948,19 +925,6 @@ fhandler_fifo::fstatvfs (struct statvfs *sfs)
   return fh.fstatvfs (sfs);
 }
 
-int
-fifo_client_handler::close ()
-{
-  int res = 0;
-
-  if (fh)
-    {
-      res = fh->fhandler_base::close ();
-      delete fh;
-    }
-  return res;
-}
-
 int
 fifo_client_handler::pipe_state ()
 {
@@ -968,7 +932,7 @@ fifo_client_handler::pipe_state ()
   FILE_PIPE_LOCAL_INFORMATION fpli;
   NTSTATUS status;
 
-  status = NtQueryInformationFile (fh->get_handle (), &io, &fpli,
+  status = NtQueryInformationFile (h, &io, &fpli,
    sizeof (fpli), FilePipeLocalInformation);
   if (!NT_SUCCESS (status))
     {
@@ -1022,8 +986,7 @@ fhandler_fifo::close ()
     NtClose (write_ready);
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
-    if (fc_handler[i].close () < 0)
-      ret = -1;
+    fc_handler[i].close ();
   fifo_client_unlock ();
   return fhandler_base::close () || ret;
 }
@@ -1078,9 +1041,9 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
     {
-      if (!DuplicateHandle (GetCurrentProcess (), fc_handler[i].fh->get_handle (),
+      if (!DuplicateHandle (GetCurrentProcess (), fc_handler[i].h,
     GetCurrentProcess (),
-    &fhf->fc_handler[i].fh->get_handle (),
+    &fhf->fc_handler[i].h,
     0, true, DUPLICATE_SAME_ACCESS))
  {
   fifo_client_unlock ();
@@ -1114,7 +1077,7 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
   fork_fixup (parent, write_ready, "write_ready");
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
-    fc_handler[i].fh->fhandler_base::fixup_after_fork (parent);
+  fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
   fifo_client_unlock ();
   if (reader && !listen_client ())
     debug_printf ("failed to start lct, %E");
@@ -1136,6 +1099,6 @@ fhandler_fifo::set_close_on_exec (bool val)
   set_no_inheritance (write_ready, val);
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
-    fc_handler[i].fh->fhandler_base::set_close_on_exec (val);
+    set_no_inheritance (fc_handler[i].h, val);
   fifo_client_unlock ();
 }
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 03/21] Cygwin: FIFO: change the fifo_client_connect_state enum

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Make the values correspond to the possible return values of
fifo_client_handler::pipe_state().

When cleaning up the fc_handler list in listen_client_thread(), don't
delete handlers in the fc_closing state.  I think the pipe might still
have input to be read in that case.

Set the state to fc_closing later in the same function if a connection
is made and the status returned by NtFsControlFile is
STATUS_PIPE_CLOSING.

In raw_read, don't error out if NtReadFile returns an unexpected
status; just set the state of that handler to fc_error.  One writer in
a bad state doesn't justify giving up on reading.
---
 winsup/cygwin/fhandler.h       | 10 ++++++++--
 winsup/cygwin/fhandler_fifo.cc | 29 ++++++++++++++---------------
 2 files changed, 22 insertions(+), 17 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index e841f96ac..c1f47025a 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1270,11 +1270,16 @@ public:
 #define CYGWIN_FIFO_PIPE_NAME_LEN     47
 #define MAX_CLIENTS 64
 
+/* The last three are the ones we try to read from. */
 enum fifo_client_connect_state
 {
   fc_unknown,
+  fc_error,
+  fc_disconnected,
+  fc_listening,
   fc_connected,
-  fc_invalid
+  fc_closing,
+  fc_input_avail,
 };
 
 enum
@@ -1316,7 +1321,8 @@ class fhandler_fifo: public fhandler_base
   bool listen_client ();
   int stop_listen_client ();
   int check_listen_client_thread ();
-  void record_connection (fifo_client_handler&);
+  void record_connection (fifo_client_handler&,
+  fifo_client_connect_state = fc_connected);
 public:
   fhandler_fifo ();
   bool hit_eof ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 6b71dd950..ba3dbb124 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -267,6 +267,7 @@ fhandler_fifo::add_client_handler ()
     {
       ret = 0;
       fc.h = ph;
+      fc.state = fc_listening;
       fc_handler[nhandlers++] = fc;
     }
 out:
@@ -311,10 +312,11 @@ fhandler_fifo::listen_client ()
 }
 
 void
-fhandler_fifo::record_connection (fifo_client_handler& fc)
+fhandler_fifo::record_connection (fifo_client_handler& fc,
+  fifo_client_connect_state s)
 {
   SetEvent (write_ready);
-  fc.state = fc_connected;
+  fc.state = s;
   nconnected++;
   set_pipe_non_blocking (fc.h, true);
 }
@@ -330,15 +332,12 @@ fhandler_fifo::listen_client_thread ()
 
   while (1)
     {
-      /* At the beginning of the loop, all client handlers are
- in the fc_connected or fc_invalid state. */
-
-      /* Delete any invalid clients. */
+      /* Cleanup the fc_handler list. */
       fifo_client_lock ();
       int i = 0;
       while (i < nhandlers)
  {
-  if (fc_handler[i].state == fc_invalid)
+  if (fc_handler[i].state < fc_connected)
     delete_client_handler (i);
   else
     i++;
@@ -393,6 +392,10 @@ fhandler_fifo::listen_client_thread ()
   record_connection (fc);
   ResetEvent (evt);
   break;
+ case STATUS_PIPE_CLOSING:
+  record_connection (fc, fc_closing);
+  ResetEvent (evt);
+  break;
  case STATUS_THREAD_IS_TERMINATING:
   /* Force NtFsControlFile to complete.  Otherwise the next
      writer to connect might not be recorded in the client
@@ -835,7 +838,7 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
       /* Poll the connected clients for input. */
       fifo_client_lock ();
       for (int i = 0; i < nhandlers; i++)
- if (fc_handler[i].state == fc_connected)
+ if (fc_handler[i].state >= fc_connected)
   {
     NTSTATUS status;
     IO_STATUS_BLOCK io;
@@ -859,18 +862,14 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
       case STATUS_PIPE_EMPTY:
  break;
       case STATUS_PIPE_BROKEN:
- /* Client has disconnected.  Mark the client handler
-   to be deleted when it's safe to do that. */
- fc_handler[i].state = fc_invalid;
+ fc_handler[i].state = fc_disconnected;
  nconnected--;
  break;
       default:
  debug_printf ("NtReadFile status %y", status);
- __seterrno_from_nt_status (status);
- fc_handler[i].state = fc_invalid;
+ fc_handler[i].state = fc_error;
  nconnected--;
- fifo_client_unlock ();
- goto errout;
+ break;
       }
   }
       fifo_client_unlock ();
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 04/21] Cygwin: FIFO: simplify the listen_client_thread code

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Always return 0; no one is doing anything with the return value
anyway.

Remove the return value from stop_listen_client.

Make the connection event auto-reset, so that we don't have to reset
it later.

Simplify the process of connecting a bogus client when thread
termination is signaled.

Make some failures fatal.

Remove the unnecessary extra check for thread termination near the end
of listen_client_thread.
---
 winsup/cygwin/fhandler.h       |   4 +-
 winsup/cygwin/fhandler_fifo.cc | 117 +++++++++++++--------------------
 2 files changed, 47 insertions(+), 74 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index c1f47025a..c8f7a39a2 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1319,7 +1319,7 @@ class fhandler_fifo: public fhandler_base
   int add_client_handler ();
   void delete_client_handler (int);
   bool listen_client ();
-  int stop_listen_client ();
+  void stop_listen_client ();
   int check_listen_client_thread ();
   void record_connection (fifo_client_handler&,
   fifo_client_connect_state = fc_connected);
@@ -1345,7 +1345,7 @@ public:
   ssize_t __reg3 raw_write (const void *ptr, size_t ulen);
   bool arm (HANDLE h);
   bool need_fixup_before () const { return reader; }
-  int fixup_before_fork_exec (DWORD) { return stop_listen_client (); }
+  int fixup_before_fork_exec (DWORD) { stop_listen_client (); return 0; }
   void init_fixup_before ();
   void fixup_after_fork (HANDLE);
   void fixup_after_exec ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index ba3dbb124..fb20e5a7e 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -324,11 +324,10 @@ fhandler_fifo::record_connection (fifo_client_handler& fc,
 DWORD
 fhandler_fifo::listen_client_thread ()
 {
-  DWORD ret = -1;
-  HANDLE evt;
+  HANDLE conn_evt;
 
-  if (!(evt = create_event ()))
-    goto out;
+  if (!(conn_evt = CreateEvent (NULL, false, false, NULL)))
+    api_fatal ("Can't create connection event, %E");
 
   while (1)
     {
@@ -346,7 +345,7 @@ fhandler_fifo::listen_client_thread ()
 
       /* Create a new client handler. */
       if (add_client_handler () < 0)
- goto out;
+ api_fatal ("Can't add a client handler, %E");
 
       /* Allow a writer to open. */
       if (!arm (read_ready))
@@ -359,12 +358,13 @@ fhandler_fifo::listen_client_thread ()
       fifo_client_handler& fc = fc_handler[nhandlers - 1];
       NTSTATUS status;
       IO_STATUS_BLOCK io;
+      bool cancel = false;
 
-      status = NtFsControlFile (fc.h, evt, NULL, NULL, &io,
+      status = NtFsControlFile (fc.h, conn_evt, NULL, NULL, &io,
  FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
       if (status == STATUS_PENDING)
  {
-  HANDLE w[2] = { evt, lct_termination_evt };
+  HANDLE w[2] = { conn_evt, lct_termination_evt };
   DWORD waitret = WaitForMultipleObjects (2, w, false, INFINITE);
   switch (waitret)
     {
@@ -372,83 +372,65 @@ fhandler_fifo::listen_client_thread ()
       status = io.Status;
       break;
     case WAIT_OBJECT_0 + 1:
-      ret = 0;
       status = STATUS_THREAD_IS_TERMINATING;
+      cancel = true;
       break;
     default:
-      __seterrno ();
-      debug_printf ("WaitForMultipleObjects failed, %E");
-      status = STATUS_THREAD_IS_TERMINATING;
-      break;
+      api_fatal ("WFMO failed, %E");
     }
  }
       HANDLE ph = NULL;
-      int ps = -1;
+      NTSTATUS status1;
+
       fifo_client_lock ();
       switch (status)
  {
  case STATUS_SUCCESS:
  case STATUS_PIPE_CONNECTED:
   record_connection (fc);
-  ResetEvent (evt);
   break;
  case STATUS_PIPE_CLOSING:
   record_connection (fc, fc_closing);
-  ResetEvent (evt);
   break;
  case STATUS_THREAD_IS_TERMINATING:
-  /* Force NtFsControlFile to complete.  Otherwise the next
-     writer to connect might not be recorded in the client
-     handler list. */
-  status = open_pipe (ph);
-  if (NT_SUCCESS (status)
-      && (NT_SUCCESS (io.Status) || io.Status == STATUS_PIPE_CONNECTED))
-    {
-      debug_printf ("successfully connected bogus client");
-      delete_client_handler (nhandlers - 1);
-    }
-  else if ((ps = fc.pipe_state ()) == FILE_PIPE_CONNECTED_STATE
-   || ps == FILE_PIPE_INPUT_AVAILABLE_STATE)
-    {
-      /* A connection was made under our nose. */
-      debug_printf ("recording connection before terminating");
-      record_connection (fc);
-    }
+  /* Try to connect a bogus client.  Otherwise fc is still
+     listening, and the next connection might not get recorded. */
+  status1 = open_pipe (ph);
+  WaitForSingleObject (conn_evt, INFINITE);
+  if (NT_SUCCESS (status1))
+    /* Bogus cilent connected. */
+    delete_client_handler (nhandlers - 1);
   else
-    {
-      debug_printf ("failed to terminate NtFsControlFile cleanly");
-      delete_client_handler (nhandlers - 1);
-      ret = -1;
-    }
-  if (ph)
-    NtClose (ph);
-  fifo_client_unlock ();
-  goto out;
+    /* Did a real client connect? */
+    switch (io.Status)
+      {
+      case STATUS_SUCCESS:
+      case STATUS_PIPE_CONNECTED:
+ record_connection (fc);
+ break;
+      case STATUS_PIPE_CLOSING:
+ record_connection (fc, fc_closing);
+ break;
+      default:
+ debug_printf ("NtFsControlFile status %y after failing to connect bogus client or real client", io.Status);
+ fc.state = fc_unknown;
+ break;
+      }
+  break;
  default:
-  debug_printf ("NtFsControlFile status %y", status);
-  __seterrno_from_nt_status (status);
-  delete_client_handler (nhandlers - 1);
-  fifo_client_unlock ();
-  goto out;
+  break;
  }
       fifo_client_unlock ();
-      /* Check for thread termination in case WaitForMultipleObjects
- didn't get called above. */
-      if (IsEventSignalled (lct_termination_evt))
- {
-  ret = 0;
-  goto out;
- }
+      if (ph)
+ NtClose (ph);
+      if (cancel)
+ goto out;
     }
 out:
-  if (evt)
-    NtClose (evt);
+  if (conn_evt)
+    NtClose (conn_evt);
   ResetEvent (read_ready);
-  if (ret < 0)
-    debug_printf ("exiting with error, %E");
-  else
-    debug_printf ("exiting without error");
-  return ret;
+  return 0;
 }
 
 int
@@ -945,10 +927,9 @@ fifo_client_handler::pipe_state ()
     return fpli.NamedPipeState;
 }
 
-int
+void
 fhandler_fifo::stop_listen_client ()
 {
-  int ret = 0;
   HANDLE thr, evt;
 
   thr = InterlockedExchangePointer (&listen_client_thr, NULL);
@@ -957,19 +938,11 @@ fhandler_fifo::stop_listen_client ()
       if (lct_termination_evt)
  SetEvent (lct_termination_evt);
       WaitForSingleObject (thr, INFINITE);
-      DWORD err;
-      GetExitCodeThread (thr, &err);
-      if (err)
- {
-  ret = -1;
-  debug_printf ("listen_client_thread exited with error");
- }
       NtClose (thr);
     }
   evt = InterlockedExchangePointer (&lct_termination_evt, NULL);
   if (evt)
     NtClose (evt);
-  return ret;
 }
 
 int
@@ -978,7 +951,7 @@ fhandler_fifo::close ()
   /* Avoid deadlock with lct in case this is called from a signal
      handler or another thread. */
   fifo_client_unlock ();
-  int ret = stop_listen_client ();
+  stop_listen_client ();
   if (read_ready)
     NtClose (read_ready);
   if (write_ready)
@@ -987,7 +960,7 @@ fhandler_fifo::close ()
   for (int i = 0; i < nhandlers; i++)
     fc_handler[i].close ();
   fifo_client_unlock ();
-  return fhandler_base::close () || ret;
+  return fhandler_base::close ();
 }
 
 /* If we have a write handle (i.e., we're a duplexer or a writer),
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 05/21] Cygwin: FIFO: remove the arm method

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
There's no reason to check for errors when we set read_ready or
write_ready.  We don't do that for other events.
---
 winsup/cygwin/fhandler.h       |  1 -
 winsup/cygwin/fhandler_fifo.cc | 34 +++-------------------------------
 2 files changed, 3 insertions(+), 32 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index c8f7a39a2..4d691a0fc 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1343,7 +1343,6 @@ public:
   void set_close_on_exec (bool val);
   void __reg3 raw_read (void *ptr, size_t& ulen);
   ssize_t __reg3 raw_write (const void *ptr, size_t ulen);
-  bool arm (HANDLE h);
   bool need_fixup_before () const { return reader; }
   int fixup_before_fork_exec (DWORD) { stop_listen_client (); return 0; }
   void init_fixup_before ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index fb20e5a7e..44919c19e 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -93,28 +93,6 @@ sec_user_cloexec (bool cloexec, PSECURITY_ATTRIBUTES sa, PSID sid)
   return cloexec ? sec_user_nih (sa, sid) : sec_user (sa, sid);
 }
 
-bool inline
-fhandler_fifo::arm (HANDLE h)
-{
-#ifdef DEBUGGING
-  const char *what;
-  if (h == read_ready)
-    what = "reader";
-  else
-    what = "writer";
-  debug_only_printf ("arming %s", what);
-#endif
-
-  bool res = SetEvent (h);
-  if (!res)
-#ifdef DEBUGGING
-    debug_printf ("SetEvent for %s failed, %E", what);
-#else
-    debug_printf ("SetEvent failed, %E");
-#endif
-  return res;
-}
-
 static HANDLE
 create_event ()
 {
@@ -348,11 +326,7 @@ fhandler_fifo::listen_client_thread ()
  api_fatal ("Can't add a client handler, %E");
 
       /* Allow a writer to open. */
-      if (!arm (read_ready))
- {
-  __seterrno ();
-  goto out;
- }
+      SetEvent (read_ready);
 
       /* Listen for a writer to connect to the new client handler. */
       fifo_client_handler& fc = fc_handler[nhandlers - 1];
@@ -555,10 +529,8 @@ fhandler_fifo::open (int flags, mode_t)
   if (NT_SUCCESS (status))
     {
       set_pipe_non_blocking (get_handle (), flags & O_NONBLOCK);
-      if (!arm (write_ready))
- res = error_set_errno;
-      else
- res = success;
+      SetEvent (write_ready);
+      res = success;
       goto out;
     }
   else if (STATUS_PIPE_NO_INSTANCE_AVAILABLE (status))
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 06/21] Cygwin: FIFO: honor the flags argument in dup

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Also improve the error handling.
---
 winsup/cygwin/fhandler_fifo.cc | 60 +++++++++++++++++++---------------
 1 file changed, 33 insertions(+), 27 deletions(-)

diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 44919c19e..f61e2fe72 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -955,56 +955,62 @@ fhandler_fifo::fcntl (int cmd, intptr_t arg)
 int
 fhandler_fifo::dup (fhandler_base *child, int flags)
 {
-  int ret = -1;
+  int i = 0;
   fhandler_fifo *fhf = NULL;
 
   if (get_flags () & O_PATH)
     return fhandler_base::dup (child, flags);
 
   if (fhandler_base::dup (child, flags))
-    goto out;
+    goto err;
 
   fhf = (fhandler_fifo *) child;
   if (!DuplicateHandle (GetCurrentProcess (), read_ready,
  GetCurrentProcess (), &fhf->read_ready,
- 0, true, DUPLICATE_SAME_ACCESS))
+ 0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
     {
-      fhf->close ();
       __seterrno ();
-      goto out;
+      goto err;
     }
   if (!DuplicateHandle (GetCurrentProcess (), write_ready,
  GetCurrentProcess (), &fhf->write_ready,
- 0, true, DUPLICATE_SAME_ACCESS))
+ 0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
     {
-      NtClose (fhf->read_ready);
-      fhf->close ();
       __seterrno ();
-      goto out;
+      goto err_close_read_ready;
     }
-  fifo_client_lock ();
-  for (int i = 0; i < nhandlers; i++)
+  if (reader)
     {
-      if (!DuplicateHandle (GetCurrentProcess (), fc_handler[i].h,
-    GetCurrentProcess (),
-    &fhf->fc_handler[i].h,
-    0, true, DUPLICATE_SAME_ACCESS))
+      fifo_client_lock ();
+      for (i = 0; i < nhandlers; i++)
+ {
+  if (!DuplicateHandle (GetCurrentProcess (), fc_handler[i].h,
+ GetCurrentProcess (), &fhf->fc_handler[i].h,
+ 0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+    {
+      __seterrno ();
+      break;
+    }
+ }
+      if (i < nhandlers)
  {
   fifo_client_unlock ();
-  NtClose (fhf->read_ready);
-  NtClose (fhf->write_ready);
-  fhf->close ();
-  __seterrno ();
-  goto out;
+  goto err_close_handlers;
  }
+      fifo_client_unlock ();
+      if (!fhf->listen_client ())
+ goto err_close_handlers;
+      fhf->init_fixup_before ();
     }
-  fifo_client_unlock ();
-  if (!reader || fhf->listen_client ())
-    ret = 0;
-  if (reader)
-    fhf->init_fixup_before ();
-out:
-  return ret;
+  return 0;
+err_close_handlers:
+  for (int j = 0; j < i; j++)
+    fhf->fc_handler[j].close ();
+  NtClose (fhf->write_ready);
+err_close_read_ready:
+  NtClose (fhf->read_ready);
+err:
+  return -1;
 }
 
 void
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 07/21] Cygwin: FIFO: dup/fork/exec: make sure child starts unlocked

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
There can be deadlocks if the child starts with its fifo_client_lock
in the locked state.
---
 winsup/cygwin/fhandler_fifo.cc | 31 +++++++++++++++++++++++--------
 1 file changed, 23 insertions(+), 8 deletions(-)

diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index f61e2fe72..4904a535d 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -981,6 +981,9 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
     }
   if (reader)
     {
+      /* Make sure the child starts unlocked. */
+      fhf->fifo_client_unlock ();
+
       fifo_client_lock ();
       for (i = 0; i < nhandlers; i++)
  {
@@ -1025,20 +1028,32 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
   fhandler_base::fixup_after_fork (parent);
   fork_fixup (parent, read_ready, "read_ready");
   fork_fixup (parent, write_ready, "write_ready");
-  fifo_client_lock ();
-  for (int i = 0; i < nhandlers; i++)
-  fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
-  fifo_client_unlock ();
-  if (reader && !listen_client ())
-    debug_printf ("failed to start lct, %E");
+  if (reader)
+    {
+      /* Make sure the child starts unlocked. */
+      fifo_client_unlock ();
+
+      fifo_client_lock ();
+      for (int i = 0; i < nhandlers; i++)
+ fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
+      fifo_client_unlock ();
+      if (!listen_client ())
+ debug_printf ("failed to start lct, %E");
+    }
 }
 
 void
 fhandler_fifo::fixup_after_exec ()
 {
   fhandler_base::fixup_after_exec ();
-  if (reader && !listen_client ())
-    debug_printf ("failed to start lct, %E");
+  if (reader && !close_on_exec ())
+    {
+      /* Make sure the child starts unlocked. */
+      fifo_client_unlock ();
+
+      if (!listen_client ())
+ debug_printf ("failed to start lct, %E");
+    }
 }
 
 void
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 08/21] Cygwin: FIFO: fix hit_eof

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
According to Posix, a FIFO open for reading is at EOF if it is empty
and there are no writers open.

The only way to test this is to poll the fifo_client_handlers as in
raw_read and select.cc:peek_fifo.  The current hit_eof instead relies
on the value of nconnected, which can be out of date.  On the one
hand, it doesn't take into account writers that were connected but
have since closed.  On the other hand, it doesn't take into account
writers that are in the process of opening but haven't yet connected.

Fix this by introducing a maybe_eof method that tentatively assumes
EOF if there are no connected writers after polling.  Then check for
writers currently opening (via a new 'writer_opening' event), and wait
for the fifo_reader_thread to record any new connection that was made
while we were polling.

To handle the needs of peek_fifo, replace the get_fc_handle method
by a get_fc_handler method, and add a fifo_client_handler::get_state
method.

Remove the is_connected method, which was used only in peek_fifo and
is no longer needed.

Remove the nconnected data member, which was used only for the flawed
hit_eof.

Add some comments about events to fhandler.h.
---
 winsup/cygwin/fhandler.h       | 19 +++++---
 winsup/cygwin/fhandler_fifo.cc | 84 ++++++++++++++++++++++------------
 winsup/cygwin/select.cc        | 44 ++++++++++++------
 3 files changed, 98 insertions(+), 49 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 4d691a0fc..3bc04cf13 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1296,19 +1296,26 @@ struct fifo_client_handler
 /* Returns FILE_PIPE_DISCONNECTED_STATE, FILE_PIPE_LISTENING_STATE,
    FILE_PIPE_CONNECTED_STATE, FILE_PIPE_CLOSING_STATE,
    FILE_PIPE_INPUT_AVAILABLE_STATE, or -1 on error. */
+  fifo_client_connect_state &get_state () { return state; }
   int pipe_state ();
 };
 
 class fhandler_fifo: public fhandler_base
 {
-  HANDLE read_ready;
-  HANDLE write_ready;
+  /* Handles to named events shared by all fhandlers for a given FIFO. */
+  HANDLE read_ready;            /* A reader is open; OK for a writer to open. */
+  HANDLE write_ready;           /* A writer is open; OK for a reader to open. */
+  HANDLE writer_opening;        /* A writer is opening; no EOF. */
+
+  /* Non-shared handles needed for the listen_client_thread. */
   HANDLE listen_client_thr;
   HANDLE lct_termination_evt;
+
   UNICODE_STRING pipe_name;
   WCHAR pipe_name_buf[CYGWIN_FIFO_PIPE_NAME_LEN + 1];
+  bool _maybe_eof;
   fifo_client_handler fc_handler[MAX_CLIENTS];
-  int nhandlers, nconnected;
+  int nhandlers;
   af_unix_spinlock_t _fifo_client_lock;
   bool reader, writer, duplexer;
   size_t max_atomic_write;
@@ -1326,10 +1333,10 @@ class fhandler_fifo: public fhandler_base
 public:
   fhandler_fifo ();
   bool hit_eof ();
+  bool maybe_eof () const { return _maybe_eof; }
+  void maybe_eof (bool val) { _maybe_eof = val; }
   int get_nhandlers () const { return nhandlers; }
-  HANDLE get_fc_handle (int i) const { return fc_handler[i].h; }
-  bool is_connected (int i) const
-  { return fc_handler[i].state == fc_connected; }
+  fifo_client_handler get_fc_handler (int i) const { return fc_handler[i]; }
   PUNICODE_STRING get_pipe_name ();
   DWORD listen_client_thread ();
   void fifo_client_lock () { _fifo_client_lock.lock (); }
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 4904a535d..21faf4ec2 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -66,9 +66,10 @@ STATUS_PIPE_EMPTY simply means there's no data to be read. */
    || _s == STATUS_PIPE_BUSY; })
 
 fhandler_fifo::fhandler_fifo ():
-  fhandler_base (), read_ready (NULL), write_ready (NULL),
-  listen_client_thr (NULL), lct_termination_evt (NULL), nhandlers (0),
-  nconnected (0), reader (false), writer (false), duplexer (false),
+  fhandler_base (),
+  read_ready (NULL), write_ready (NULL), writer_opening (NULL),
+  listen_client_thr (NULL), lct_termination_evt (NULL), _maybe_eof (false), nhandlers (0),
+  reader (false), writer (false), duplexer (false),
   max_atomic_write (DEFAULT_PIPEBUFSIZE)
 {
   pipe_name_buf[0] = L'\0';
@@ -295,7 +296,8 @@ fhandler_fifo::record_connection (fifo_client_handler& fc,
 {
   SetEvent (write_ready);
   fc.state = s;
-  nconnected++;
+  maybe_eof (false);
+  ResetEvent (writer_opening);
   set_pipe_non_blocking (fc.h, true);
 }
 
@@ -465,6 +467,13 @@ fhandler_fifo::open (int flags, mode_t)
       res = error_set_errno;
       goto out;
     }
+  npbuf[0] = 'o';
+  if (!(writer_opening = CreateEvent (sa_buf, true, false, npbuf)))
+    {
+      debug_printf ("CreateEvent for %s failed, %E", npbuf);
+      res = error_set_errno;
+      goto out;
+    }
 
   /* If we're a duplexer, create the pipe and the first client handler. */
   if (duplexer)
@@ -518,10 +527,12 @@ fhandler_fifo::open (int flags, mode_t)
      listen_client thread is running.  Then signal write_ready.  */
   if (writer)
     {
+      SetEvent (writer_opening);
       while (1)
  {
   if (!wait (read_ready))
     {
+      ResetEvent (writer_opening);
       res = error_errno_set;
       goto out;
     }
@@ -540,6 +551,7 @@ fhandler_fifo::open (int flags, mode_t)
       debug_printf ("create of writer failed");
       __seterrno_from_nt_status (status);
       res = error_errno_set;
+      ResetEvent (writer_opening);
       goto out;
     }
  }
@@ -559,6 +571,11 @@ out:
   NtClose (write_ready);
   write_ready = NULL;
  }
+      if (writer_opening)
+ {
+  NtClose (writer_opening);
+  writer_opening = NULL;
+ }
       if (get_handle ())
  NtClose (get_handle ());
       if (listen_client_thr)
@@ -717,28 +734,23 @@ fhandler_fifo::raw_write (const void *ptr, size_t len)
   return ret;
 }
 
-/* A FIFO open for reading is at EOF if no process has it open for
-   writing.  We test this by checking nconnected.  But we must take
-   account of the possible delay from the time of connection to the
-   time the connection is recorded by the listen_client thread. */
+/* A reader is at EOF if the pipe is empty and no writers are open.
+   hit_eof is called by raw_read and select.cc:peek_fifo if it appears
+   that we are at EOF after polling the fc_handlers.  We recheck this
+   in case a writer opened while we were polling.  */
 bool
 fhandler_fifo::hit_eof ()
 {
-  bool eof;
-  bool retry = true;
-
-repeat:
+  bool ret = maybe_eof () && !IsEventSignalled (writer_opening);
+  if (ret)
+    {
+      yield ();
+      /* Wait for the reader thread to finish recording any connection. */
       fifo_client_lock ();
-      eof = (nconnected == 0);
       fifo_client_unlock ();
-      if (eof && retry)
- {
-  retry = false;
-  /* Give the listen_client thread time to catch up. */
-  Sleep (1);
-  goto repeat;
- }
-  return eof;
+      ret = maybe_eof ();
+    }
+  return ret;
 }
 
 /* Is the lct running? */
@@ -783,13 +795,8 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 
   while (1)
     {
-      if (hit_eof ())
- {
-  len = 0;
-  return;
- }
-
       /* Poll the connected clients for input. */
+      int nconnected = 0;
       fifo_client_lock ();
       for (int i = 0; i < nhandlers; i++)
  if (fc_handler[i].state >= fc_connected)
@@ -798,7 +805,8 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
     IO_STATUS_BLOCK io;
     size_t nbytes = 0;
 
-    status = NtReadFile (get_fc_handle (i), NULL, NULL, NULL,
+    nconnected++;
+    status = NtReadFile (fc_handler[i].h, NULL, NULL, NULL,
  &io, in_ptr, len, NULL, NULL);
     switch (status)
       {
@@ -826,7 +834,13 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
  break;
       }
   }
+      maybe_eof (!nconnected && !IsEventSignalled (writer_opening));
       fifo_client_unlock ();
+      if (maybe_eof () && hit_eof ())
+ {
+  len = 0;
+  return;
+ }
       if (is_nonblocking ())
  {
   set_errno (EAGAIN);
@@ -928,6 +942,8 @@ fhandler_fifo::close ()
     NtClose (read_ready);
   if (write_ready)
     NtClose (write_ready);
+  if (writer_opening)
+    NtClose (writer_opening);
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
     fc_handler[i].close ();
@@ -979,6 +995,13 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
       __seterrno ();
       goto err_close_read_ready;
     }
+  if (!DuplicateHandle (GetCurrentProcess (), writer_opening,
+ GetCurrentProcess (), &fhf->writer_opening,
+ 0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+    {
+      __seterrno ();
+      goto err_close_write_ready;
+    }
   if (reader)
     {
       /* Make sure the child starts unlocked. */
@@ -1009,6 +1032,9 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
 err_close_handlers:
   for (int j = 0; j < i; j++)
     fhf->fc_handler[j].close ();
+/* err_close_writer_opening: */
+  NtClose (fhf->writer_opening);
+err_close_write_ready:
   NtClose (fhf->write_ready);
 err_close_read_ready:
   NtClose (fhf->read_ready);
@@ -1028,6 +1054,7 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
   fhandler_base::fixup_after_fork (parent);
   fork_fixup (parent, read_ready, "read_ready");
   fork_fixup (parent, write_ready, "write_ready");
+  fork_fixup (parent, writer_opening, "writer_opening");
   if (reader)
     {
       /* Make sure the child starts unlocked. */
@@ -1062,6 +1089,7 @@ fhandler_fifo::set_close_on_exec (bool val)
   fhandler_base::set_close_on_exec (val);
   set_no_inheritance (read_ready, val);
   set_no_inheritance (write_ready, val);
+  set_no_inheritance (writer_opening, val);
   fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
     set_no_inheritance (fc_handler[i].h, val);
diff --git a/winsup/cygwin/select.cc b/winsup/cygwin/select.cc
index b5d19cf31..9323c423f 100644
--- a/winsup/cygwin/select.cc
+++ b/winsup/cygwin/select.cc
@@ -866,31 +866,45 @@ peek_fifo (select_record *s, bool from_select)
   goto out;
  }
 
-      if (fh->hit_eof ())
- {
-  select_printf ("read: %s, saw EOF", fh->get_name ());
-  gotone = s->read_ready = true;
-  if (s->except_selected)
-    gotone += s->except_ready = true;
-  goto out;
- }
-
       fh->fifo_client_lock ();
+      int nconnected = 0;
       for (int i = 0; i < fh->get_nhandlers (); i++)
- if (fh->is_connected (i))
+ if (fh->get_fc_handler (i).get_state () >= fc_connected)
   {
-    int n = pipe_data_available (s->fd, fh, fh->get_fc_handle (i),
- false);
-    if (n > 0)
+    nconnected++;
+    switch (fh->get_fc_handler (i).pipe_state ())
       {
- select_printf ("read: %s, ready for read: avail %d, client %d",
-       fh->get_name (), n, i);
+      case FILE_PIPE_CONNECTED_STATE:
+ fh->get_fc_handler (i).get_state () = fc_connected;
+ break;
+      case FILE_PIPE_DISCONNECTED_STATE:
+ fh->get_fc_handler (i).get_state () = fc_disconnected;
+ nconnected--;
+ break;
+      case FILE_PIPE_CLOSING_STATE:
+ fh->get_fc_handler (i).get_state () = fc_closing;
+ break;
+      case FILE_PIPE_INPUT_AVAILABLE_STATE:
+ fh->get_fc_handler (i).get_state () = fc_input_avail;
+ select_printf ("read: %s, ready for read", fh->get_name ());
  fh->fifo_client_unlock ();
  gotone += s->read_ready = true;
  goto out;
+      default:
+ fh->get_fc_handler (i).get_state () = fc_error;
+ nconnected--;
+ break;
       }
   }
+      fh->maybe_eof (!nconnected);
       fh->fifo_client_unlock ();
+      if (fh->maybe_eof () && fh->hit_eof ())
+ {
+  select_printf ("read: %s, saw EOF", fh->get_name ());
+  gotone += s->read_ready = true;
+  if (s->except_selected)
+    gotone += s->except_ready = true;
+ }
     }
 out:
   if (s->write_selected)
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 09/21] Cygwin: FIFO: make opening a writer more robust

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
- Make read_ready a manual-reset event.

- Signal read_ready in open instead of in the listen_client_thread.

- Don't reset read_ready when the listen_client thread terminates;
  instead do it in close().

- Rearrange open and change its error handling.

- Add a wait_open_pipe method that waits for a pipe instance to be
  available and then calls open_pipe.  Use it when opening a writer if
  we can't connect immediately.  This can happen if the system is
  heavily loaded and/or if many writers are trying to open
  simultaneously.
---
 winsup/cygwin/fhandler.h       |   1 +
 winsup/cygwin/fhandler_fifo.cc | 267 +++++++++++++++++++++------------
 2 files changed, 168 insertions(+), 100 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 3bc04cf13..2516c93b4 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1323,6 +1323,7 @@ class fhandler_fifo: public fhandler_base
   static NTSTATUS npfs_handle (HANDLE &);
   HANDLE create_pipe_instance (bool);
   NTSTATUS open_pipe (HANDLE&);
+  NTSTATUS wait_open_pipe (HANDLE&);
   int add_client_handler ();
   void delete_client_handler (int);
   bool listen_client ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 21faf4ec2..5c3df5497 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -222,7 +222,64 @@ fhandler_fifo::open_pipe (HANDLE& ph)
       openflags & O_CLOEXEC ? 0 : OBJ_INHERIT,
       npfsh, NULL);
   sharing = FILE_SHARE_READ | FILE_SHARE_WRITE;
-  status = NtOpenFile (&ph, access, &attr, &io, sharing, 0);
+  return NtOpenFile (&ph, access, &attr, &io, sharing, 0);
+}
+
+/* Wait up to 100ms for a pipe instance to be available, then connect. */
+NTSTATUS
+fhandler_fifo::wait_open_pipe (HANDLE& ph)
+{
+  HANDLE npfsh;
+  HANDLE evt;
+  NTSTATUS status;
+  IO_STATUS_BLOCK io;
+  ULONG pwbuf_size;
+  PFILE_PIPE_WAIT_FOR_BUFFER pwbuf;
+  LONGLONG stamp;
+  LONGLONG orig_timeout = -100 * NS100PERSEC / MSPERSEC;   /* 100ms */
+
+  status = npfs_handle (npfsh);
+  if (!NT_SUCCESS (status))
+    return status;
+  if (!(evt = create_event ()))
+    api_fatal ("Can't create event, %E");
+  pwbuf_size
+    = offsetof (FILE_PIPE_WAIT_FOR_BUFFER, Name) + get_pipe_name ()->Length;
+  pwbuf = (PFILE_PIPE_WAIT_FOR_BUFFER) alloca (pwbuf_size);
+  pwbuf->Timeout.QuadPart = orig_timeout;
+  pwbuf->NameLength = get_pipe_name ()->Length;
+  pwbuf->TimeoutSpecified = TRUE;
+  memcpy (pwbuf->Name, get_pipe_name ()->Buffer, get_pipe_name ()->Length);
+  stamp = get_clock (CLOCK_MONOTONIC)->n100secs ();
+  bool retry;
+  do
+    {
+      retry = false;
+      status = NtFsControlFile (npfsh, evt, NULL, NULL, &io, FSCTL_PIPE_WAIT,
+ pwbuf, pwbuf_size, NULL, 0);
+      if (status == STATUS_PENDING)
+ {
+  if (WaitForSingleObject (evt, INFINITE) == WAIT_OBJECT_0)
+    status = io.Status;
+  else
+    api_fatal ("WFSO failed, %E");
+ }
+      if (NT_SUCCESS (status))
+ status = open_pipe (ph);
+      if (STATUS_PIPE_NO_INSTANCE_AVAILABLE (status))
+ {
+  /* Another writer has grabbed the pipe instance.  Adjust
+     the timeout and keep waiting if there's time left. */
+  pwbuf->Timeout.QuadPart = orig_timeout
+    + get_clock (CLOCK_MONOTONIC)->n100secs () - stamp;
+  if (pwbuf->Timeout.QuadPart < 0)
+    retry = true;
+  else
+    status = STATUS_IO_TIMEOUT;
+ }
+    }
+  while (retry);
+  NtClose (evt);
   return status;
 }
 
@@ -294,7 +351,6 @@ void
 fhandler_fifo::record_connection (fifo_client_handler& fc,
   fifo_client_connect_state s)
 {
-  SetEvent (write_ready);
   fc.state = s;
   maybe_eof (false);
   ResetEvent (writer_opening);
@@ -327,9 +383,6 @@ fhandler_fifo::listen_client_thread ()
       if (add_client_handler () < 0)
  api_fatal ("Can't add a client handler, %E");
 
-      /* Allow a writer to open. */
-      SetEvent (read_ready);
-
       /* Listen for a writer to connect to the new client handler. */
       fifo_client_handler& fc = fc_handler[nhandlers - 1];
       NTSTATUS status;
@@ -405,19 +458,13 @@ fhandler_fifo::listen_client_thread ()
 out:
   if (conn_evt)
     NtClose (conn_evt);
-  ResetEvent (read_ready);
   return 0;
 }
 
 int
 fhandler_fifo::open (int flags, mode_t)
 {
-  enum
-  {
-   success,
-   error_errno_set,
-   error_set_errno
-  } res;
+  int saved_errno = 0;
 
   if (flags & O_PATH)
     return open_fs (flags);
@@ -437,8 +484,7 @@ fhandler_fifo::open (int flags, mode_t)
       break;
     default:
       set_errno (EINVAL);
-      res = error_errno_set;
-      goto out;
+      goto err;
     }
 
   debug_only_printf ("reader %d, writer %d, duplexer %d", reader, writer, duplexer);
@@ -454,135 +500,151 @@ fhandler_fifo::open (int flags, mode_t)
 
   char npbuf[MAX_PATH];
   __small_sprintf (npbuf, "r-event.%08x.%016X", get_dev (), get_ino ());
-  if (!(read_ready = CreateEvent (sa_buf, false, false, npbuf)))
+  if (!(read_ready = CreateEvent (sa_buf, true, false, npbuf)))
     {
       debug_printf ("CreateEvent for %s failed, %E", npbuf);
-      res = error_set_errno;
-      goto out;
+      __seterrno ();
+      goto err;
     }
   npbuf[0] = 'w';
   if (!(write_ready = CreateEvent (sa_buf, true, false, npbuf)))
     {
       debug_printf ("CreateEvent for %s failed, %E", npbuf);
-      res = error_set_errno;
-      goto out;
+      __seterrno ();
+      goto err_close_read_ready;
     }
   npbuf[0] = 'o';
   if (!(writer_opening = CreateEvent (sa_buf, true, false, npbuf)))
     {
       debug_printf ("CreateEvent for %s failed, %E", npbuf);
-      res = error_set_errno;
-      goto out;
-    }
-
-  /* If we're a duplexer, create the pipe and the first client handler. */
-  if (duplexer)
-    {
-      HANDLE ph = NULL;
-
-      if (add_client_handler () < 0)
- {
-  res = error_errno_set;
-  goto out;
- }
-      NTSTATUS status = open_pipe (ph);
-      if (NT_SUCCESS (status))
- {
-  record_connection (fc_handler[0]);
-  set_handle (ph);
-  set_pipe_non_blocking (ph, flags & O_NONBLOCK);
- }
-      else
- {
-  __seterrno_from_nt_status (status);
-  res = error_errno_set;
-  goto out;
- }
+      __seterrno ();
+      goto err_close_write_ready;
     }
 
-  /* If we're reading, start the listen_client thread (which should
-     signal read_ready), and wait for a writer. */
+  /* If we're reading, signal read_ready and start the listen_client
+     thread. */
   if (reader)
     {
       if (!listen_client ())
  {
   debug_printf ("create of listen_client thread failed");
-  res = error_errno_set;
-  goto out;
+  goto err_close_writer_opening;
  }
-      else if (!duplexer && !wait (write_ready))
- {
-  res = error_errno_set;
-  goto out;
- }
-      else
+      SetEvent (read_ready);
+
+      /* If we're a duplexer, we need a handle for writing. */
+      if (duplexer)
  {
-  init_fixup_before ();
-  res = success;
+  HANDLE ph = NULL;
+  NTSTATUS status;
+
+  while (1)
+    {
+      status = open_pipe (ph);
+      if (NT_SUCCESS (status))
+ {
+  set_handle (ph);
+  set_pipe_non_blocking (ph, flags & O_NONBLOCK);
+  break;
+ }
+      else if (status == STATUS_OBJECT_NAME_NOT_FOUND)
+ {
+  /* The pipe hasn't been created yet. */
+  yield ();
+  continue;
+ }
+      else
+ {
+  __seterrno_from_nt_status (status);
+  goto err_close_reader;
+ }
+    }
  }
+      /* Not a duplexer; wait for a writer to connect. */
+      else if (!wait (write_ready))
+ goto err_close_reader;
+      init_fixup_before ();
+      goto success;
     }
 
-  /* 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 we're writing, wait for read_ready, connect to the pipe, and
+     signal write_ready.  */
   if (writer)
     {
+      NTSTATUS status;
+
       SetEvent (writer_opening);
+      if (!wait (read_ready))
+ {
+  ResetEvent (writer_opening);
+  goto err_close_writer_opening;
+ }
       while (1)
  {
-  if (!wait (read_ready))
-    {
-      ResetEvent (writer_opening);
-      res = error_errno_set;
-      goto out;
-    }
-  NTSTATUS status = open_pipe (get_handle ());
+  status = open_pipe (get_handle ());
   if (NT_SUCCESS (status))
+    goto writer_success;
+  else if (status == STATUS_OBJECT_NAME_NOT_FOUND)
     {
-      set_pipe_non_blocking (get_handle (), flags & O_NONBLOCK);
-      SetEvent (write_ready);
-      res = success;
-      goto out;
+      /* The pipe hasn't been created yet. */
+      yield ();
+      continue;
     }
   else if (STATUS_PIPE_NO_INSTANCE_AVAILABLE (status))
-    Sleep (1);
+    break;
   else
     {
       debug_printf ("create of writer failed");
       __seterrno_from_nt_status (status);
-      res = error_errno_set;
       ResetEvent (writer_opening);
-      goto out;
+      goto err_close_writer_opening;
     }
  }
-    }
-out:
-  if (res == error_set_errno)
-    __seterrno ();
-  if (res != success)
-    {
-      if (read_ready)
- {
-  NtClose (read_ready);
-  read_ready = NULL;
- }
-      if (write_ready)
- {
-  NtClose (write_ready);
-  write_ready = NULL;
- }
-      if (writer_opening)
+
+      /* We should get here only if the system is heavily loaded
+ and/or many writers are trying to connect simultaneously */
+      while (1)
  {
-  NtClose (writer_opening);
-  writer_opening = NULL;
+  SetEvent (writer_opening);
+  if (!wait (read_ready))
+    {
+      ResetEvent (writer_opening);
+      goto err_close_writer_opening;
+    }
+  status = wait_open_pipe (get_handle ());
+  if (NT_SUCCESS (status))
+    goto writer_success;
+  else if (status == STATUS_IO_TIMEOUT)
+    continue;
+  else
+    {
+      debug_printf ("create of writer failed");
+      __seterrno_from_nt_status (status);
+      ResetEvent (writer_opening);
+      goto err_close_writer_opening;
+    }
  }
-      if (get_handle ())
- NtClose (get_handle ());
-      if (listen_client_thr)
- stop_listen_client ();
     }
-  debug_printf ("res %d", res);
-  return res == success;
+writer_success:
+  set_pipe_non_blocking (get_handle (), flags & O_NONBLOCK);
+  SetEvent (write_ready);
+success:
+  return 1;
+err_close_reader:
+  saved_errno = get_errno ();
+  close ();
+  set_errno (saved_errno);
+  return 0;
+err_close_writer_opening:
+  NtClose (writer_opening);
+err_close_write_ready:
+  NtClose (write_ready);
+err_close_read_ready:
+  NtClose (read_ready);
+err:
+  if (get_handle ())
+    NtClose (get_handle ());
+  return 0;
 }
 
 off_t
@@ -938,6 +1000,11 @@ fhandler_fifo::close ()
      handler or another thread. */
   fifo_client_unlock ();
   stop_listen_client ();
+  if (reader)
+    /* FIXME: There could be several readers open because of
+       dup/fork/exec; we should only reset read_ready when the last
+       one closes. */
+    ResetEvent (read_ready);
   if (read_ready)
     NtClose (read_ready);
   if (write_ready)
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 10/21] Cygwin: FIFO: use a cygthread instead of a homemade thread

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
This will simplify future work.

Rename the thread from "listen_client_thread" to "fifo_reader_thread"
because it will be used for more than just listening.

Remove the fixup_before stuff, which won't be needed after future
changes to fixup_after_fork and fixup_after_exec.
---
 winsup/cygwin/fhandler.h       |  17 ++--
 winsup/cygwin/fhandler_fifo.cc | 173 +++++++++++----------------------
 2 files changed, 65 insertions(+), 125 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 2516c93b4..5e6a1d1db 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1307,9 +1307,9 @@ class fhandler_fifo: public fhandler_base
   HANDLE write_ready;           /* A writer is open; OK for a reader to open. */
   HANDLE writer_opening;        /* A writer is opening; no EOF. */
 
-  /* Non-shared handles needed for the listen_client_thread. */
-  HANDLE listen_client_thr;
-  HANDLE lct_termination_evt;
+  /* Handles to non-shared events needed for fifo_reader_threads. */
+  HANDLE cancel_evt;            /* Signal thread to terminate. */
+  HANDLE thr_sync_evt;          /* The thread has terminated. */
 
   UNICODE_STRING pipe_name;
   WCHAR pipe_name_buf[CYGWIN_FIFO_PIPE_NAME_LEN + 1];
@@ -1326,11 +1326,10 @@ class fhandler_fifo: public fhandler_base
   NTSTATUS wait_open_pipe (HANDLE&);
   int add_client_handler ();
   void delete_client_handler (int);
-  bool listen_client ();
-  void stop_listen_client ();
-  int check_listen_client_thread ();
+  void cancel_reader_thread ();
   void record_connection (fifo_client_handler&,
   fifo_client_connect_state = fc_connected);
+
 public:
   fhandler_fifo ();
   bool hit_eof ();
@@ -1339,7 +1338,7 @@ public:
   int get_nhandlers () const { return nhandlers; }
   fifo_client_handler get_fc_handler (int i) const { return fc_handler[i]; }
   PUNICODE_STRING get_pipe_name ();
-  DWORD listen_client_thread ();
+  DWORD fifo_reader_thread_func ();
   void fifo_client_lock () { _fifo_client_lock.lock (); }
   void fifo_client_unlock () { _fifo_client_lock.unlock (); }
   int open (int, mode_t);
@@ -1351,9 +1350,6 @@ public:
   void set_close_on_exec (bool val);
   void __reg3 raw_read (void *ptr, size_t& ulen);
   ssize_t __reg3 raw_write (const void *ptr, size_t ulen);
-  bool need_fixup_before () const { return reader; }
-  int fixup_before_fork_exec (DWORD) { stop_listen_client (); return 0; }
-  void init_fixup_before ();
   void fixup_after_fork (HANDLE);
   void fixup_after_exec ();
   int __reg2 fstatvfs (struct statvfs *buf);
@@ -1375,7 +1371,6 @@ public:
     void *ptr = (void *) ccalloc (malloc_type, 1, sizeof (fhandler_fifo));
     fhandler_fifo *fhf = new (ptr) fhandler_fifo (ptr);
     /* We don't want our client list to change any more. */
-    stop_listen_client ();
     copyto (fhf);
     /* fhf->pipe_name_buf is a *copy* of this->pipe_name_buf, but
        fhf->pipe_name.Buffer == this->pipe_name_buf. */
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 5c3df5497..09a7eb321 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -32,11 +32,11 @@
      When a FIFO is opened for reading,
      fhandler_fifo::create_pipe_instance is called to create the first
      instance of a Windows named pipe server (Windows terminology).  A
-     "listen_client" thread is also started; it waits for pipe clients
+     "fifo_reader" thread is also started; it waits for pipe clients
      (Windows terminology again) to connect.  This happens every time
      a process opens the FIFO for writing.
 
-     The listen_client thread creates new instances of the pipe server
+     The fifo_reader thread creates new instances of the pipe server
      as needed, so that there is always an instance available for a
      writer to connect to.
 
@@ -68,7 +68,7 @@ STATUS_PIPE_EMPTY simply means there's no data to be read. */
 fhandler_fifo::fhandler_fifo ():
   fhandler_base (),
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
-  listen_client_thr (NULL), lct_termination_evt (NULL), _maybe_eof (false), nhandlers (0),
+  cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false), nhandlers (0),
   reader (false), writer (false), duplexer (false),
   max_atomic_write (DEFAULT_PIPEBUFSIZE)
 {
@@ -319,34 +319,6 @@ fhandler_fifo::delete_client_handler (int i)
      (nhandlers - i) * sizeof (fc_handler[i]));
 }
 
-/* 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. */
-bool
-fhandler_fifo::listen_client ()
-{
-  if (!(lct_termination_evt = create_event ()))
-    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)
- NtClose (evt);
-      return false;
-    }
-  return true;
-}
-
 void
 fhandler_fifo::record_connection (fifo_client_handler& fc,
   fifo_client_connect_state s)
@@ -357,8 +329,15 @@ fhandler_fifo::record_connection (fifo_client_handler& fc,
   set_pipe_non_blocking (fc.h, true);
 }
 
+static DWORD WINAPI
+fifo_reader_thread (LPVOID param)
+{
+  fhandler_fifo *fh = (fhandler_fifo *) param;
+  return fh->fifo_reader_thread_func ();
+}
+
 DWORD
-fhandler_fifo::listen_client_thread ()
+fhandler_fifo::fifo_reader_thread_func ()
 {
   HANDLE conn_evt;
 
@@ -377,7 +356,6 @@ fhandler_fifo::listen_client_thread ()
   else
     i++;
  }
-      fifo_client_unlock ();
 
       /* Create a new client handler. */
       if (add_client_handler () < 0)
@@ -385,6 +363,7 @@ fhandler_fifo::listen_client_thread ()
 
       /* Listen for a writer to connect to the new client handler. */
       fifo_client_handler& fc = fc_handler[nhandlers - 1];
+      fifo_client_unlock ();
       NTSTATUS status;
       IO_STATUS_BLOCK io;
       bool cancel = false;
@@ -393,9 +372,8 @@ fhandler_fifo::listen_client_thread ()
  FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
       if (status == STATUS_PENDING)
  {
-  HANDLE w[2] = { conn_evt, lct_termination_evt };
-  DWORD waitret = WaitForMultipleObjects (2, w, false, INFINITE);
-  switch (waitret)
+  HANDLE w[2] = { conn_evt, cancel_evt };
+  switch (WaitForMultipleObjects (2, w, false, INFINITE))
     {
     case WAIT_OBJECT_0:
       status = io.Status;
@@ -453,11 +431,13 @@ fhandler_fifo::listen_client_thread ()
       if (ph)
  NtClose (ph);
       if (cancel)
- goto out;
+ goto canceled;
     }
-out:
+canceled:
   if (conn_evt)
     NtClose (conn_evt);
+  /* automatically return the cygthread to the cygthread pool */
+  _my_tls._ctinfo->auto_release ();
   return 0;
 }
 
@@ -521,16 +501,15 @@ fhandler_fifo::open (int flags, mode_t)
       goto err_close_write_ready;
     }
 
-  /* If we're reading, signal read_ready and start the listen_client
-     thread. */
+  /* If we're reading, signal read_ready and start the fifo_reader_thread. */
   if (reader)
     {
-      if (!listen_client ())
- {
-  debug_printf ("create of listen_client thread failed");
-  goto err_close_writer_opening;
- }
       SetEvent (read_ready);
+      if (!(cancel_evt = create_event ()))
+ goto err_close_writer_opening;
+      if (!(thr_sync_evt = create_event ()))
+ goto err_close_cancel_evt;
+      new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
 
       /* If we're a duplexer, we need a handle for writing. */
       if (duplexer)
@@ -563,7 +542,6 @@ fhandler_fifo::open (int flags, mode_t)
       /* Not a duplexer; wait for a writer to connect. */
       else if (!wait (write_ready))
  goto err_close_reader;
-      init_fixup_before ();
       goto success;
     }
 
@@ -635,6 +613,8 @@ err_close_reader:
   close ();
   set_errno (saved_errno);
   return 0;
+err_close_cancel_evt:
+  NtClose (cancel_evt);
 err_close_writer_opening:
   NtClose (writer_opening);
 err_close_write_ready:
@@ -815,43 +795,9 @@ fhandler_fifo::hit_eof ()
   return ret;
 }
 
-/* Is the lct running? */
-int
-fhandler_fifo::check_listen_client_thread ()
-{
-  int ret = 0;
-
-  if (listen_client_thr)
-    {
-      DWORD waitret = WaitForSingleObject (listen_client_thr, 0);
-      switch (waitret)
- {
- case WAIT_OBJECT_0:
-  NtClose (listen_client_thr);
-  break;
- case WAIT_TIMEOUT:
-  ret = 1;
-  break;
- default:
-  debug_printf ("WaitForSingleObject failed, %E");
-  ret = -1;
-  __seterrno ();
-  NtClose (listen_client_thr);
-  break;
- }
-    }
-  return ret;
-}
-
 void __reg3
 fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 {
-  /* Make sure the lct is running. */
-  int res = check_listen_client_thread ();
-  debug_printf ("lct status %d", res);
-  if (res < 0 || (res == 0 && !listen_client ()))
-    goto errout;
-
   if (!len)
     return;
 
@@ -976,35 +922,29 @@ fifo_client_handler::pipe_state ()
 }
 
 void
-fhandler_fifo::stop_listen_client ()
+fhandler_fifo::cancel_reader_thread ()
 {
-  HANDLE thr, evt;
-
-  thr = InterlockedExchangePointer (&listen_client_thr, NULL);
-  if (thr)
-    {
-      if (lct_termination_evt)
- SetEvent (lct_termination_evt);
-      WaitForSingleObject (thr, INFINITE);
-      NtClose (thr);
-    }
-  evt = InterlockedExchangePointer (&lct_termination_evt, NULL);
-  if (evt)
-    NtClose (evt);
+  if (cancel_evt)
+    SetEvent (cancel_evt);
+  if (thr_sync_evt)
+    WaitForSingleObject (thr_sync_evt, INFINITE);
 }
 
 int
 fhandler_fifo::close ()
 {
-  /* Avoid deadlock with lct in case this is called from a signal
-     handler or another thread. */
-  fifo_client_unlock ();
-  stop_listen_client ();
   if (reader)
-    /* FIXME: There could be several readers open because of
-       dup/fork/exec; we should only reset read_ready when the last
-       one closes. */
-    ResetEvent (read_ready);
+    {
+      cancel_reader_thread ();
+      if (cancel_evt)
+ NtClose (cancel_evt);
+      if (thr_sync_evt)
+ NtClose (thr_sync_evt);
+      /* FIXME: There could be several readers open because of
+ dup/fork/exec; we should only reset read_ready when the last
+ one closes. */
+      ResetEvent (read_ready);
+    }
   if (read_ready)
     NtClose (read_ready);
   if (write_ready)
@@ -1091,11 +1031,16 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   goto err_close_handlers;
  }
       fifo_client_unlock ();
-      if (!fhf->listen_client ())
+      if (!(fhf->cancel_evt = create_event ()))
  goto err_close_handlers;
-      fhf->init_fixup_before ();
+      if (!(fhf->thr_sync_evt = create_event ()))
+ goto err_close_cancel_evt;
+      new cygthread (fifo_reader_thread, fhf, "fifo_reader",
+     fhf->thr_sync_evt);
     }
   return 0;
+err_close_cancel_evt:
+  NtClose (fhf->cancel_evt);
 err_close_handlers:
   for (int j = 0; j < i; j++)
     fhf->fc_handler[j].close ();
@@ -1109,12 +1054,6 @@ err:
   return -1;
 }
 
-void
-fhandler_fifo::init_fixup_before ()
-{
-  cygheap->fdtab.inc_need_fixup_before ();
-}
-
 void
 fhandler_fifo::fixup_after_fork (HANDLE parent)
 {
@@ -1131,8 +1070,11 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       for (int i = 0; i < nhandlers; i++)
  fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
       fifo_client_unlock ();
-      if (!listen_client ())
- debug_printf ("failed to start lct, %E");
+      if (!(cancel_evt = create_event ()))
+ api_fatal ("Can't create reader thread cancel event during fork, %E");
+      if (!(thr_sync_evt = create_event ()))
+ api_fatal ("Can't create reader thread sync event during fork, %E");
+      new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
 }
 
@@ -1145,8 +1087,11 @@ fhandler_fifo::fixup_after_exec ()
       /* Make sure the child starts unlocked. */
       fifo_client_unlock ();
 
-      if (!listen_client ())
- debug_printf ("failed to start lct, %E");
+      if (!(cancel_evt = create_event ()))
+ api_fatal ("Can't create reader thread cancel event during exec, %E");
+      if (!(thr_sync_evt = create_event ()))
+ api_fatal ("Can't create reader thread sync event during exec, %E");
+      new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
 }
 
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 11/21] Cygwin: FIFO: add shared memory

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Even though we currently allow a FIFO to be opened for reading only
once, we can still have more than one reader open because of dup and
fork.  Add a named shared memory section accessible to all readers of
a given FIFO.  In future commits we will add information needed by all
readers to this section

Add a class fifo_shmem_t that lets us access this information.

Add a method create_shmem that is called when a reader opens, and add
a method reopen_shmem that is called by dup, fork, and exec.  (Each
new reader needs its own view of the shared memory.)
---
 winsup/cygwin/fhandler.h       | 13 +++++
 winsup/cygwin/fhandler_fifo.cc | 97 ++++++++++++++++++++++++++++++++--
 2 files changed, 106 insertions(+), 4 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 5e6a1d1db..8d6b94021 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1300,6 +1300,11 @@ struct fifo_client_handler
   int pipe_state ();
 };
 
+/* Info needed by all readers of a FIFO, stored in named shared memory. */
+class fifo_shmem_t
+{
+};
+
 class fhandler_fifo: public fhandler_base
 {
   /* Handles to named events shared by all fhandlers for a given FIFO. */
@@ -1319,6 +1324,10 @@ class fhandler_fifo: public fhandler_base
   af_unix_spinlock_t _fifo_client_lock;
   bool reader, writer, duplexer;
   size_t max_atomic_write;
+
+  HANDLE shmem_handle;
+  fifo_shmem_t *shmem;
+
   bool __reg2 wait (HANDLE);
   static NTSTATUS npfs_handle (HANDLE &);
   HANDLE create_pipe_instance (bool);
@@ -1330,6 +1339,9 @@ class fhandler_fifo: public fhandler_base
   void record_connection (fifo_client_handler&,
   fifo_client_connect_state = fc_connected);
 
+  int create_shmem ();
+  int reopen_shmem ();
+
 public:
   fhandler_fifo ();
   bool hit_eof ();
@@ -1341,6 +1353,7 @@ public:
   DWORD fifo_reader_thread_func ();
   void fifo_client_lock () { _fifo_client_lock.lock (); }
   void fifo_client_unlock () { _fifo_client_lock.unlock (); }
+
   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 09a7eb321..9a0db3f33 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -70,7 +70,8 @@ fhandler_fifo::fhandler_fifo ():
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
   cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false), nhandlers (0),
   reader (false), writer (false), duplexer (false),
-  max_atomic_write (DEFAULT_PIPEBUFSIZE)
+  max_atomic_write (DEFAULT_PIPEBUFSIZE),
+  shmem_handle (NULL), shmem (NULL)
 {
   pipe_name_buf[0] = L'\0';
   need_fork_fixup (true);
@@ -441,6 +442,67 @@ canceled:
   return 0;
 }
 
+int
+fhandler_fifo::create_shmem ()
+{
+  HANDLE sect;
+  OBJECT_ATTRIBUTES attr;
+  NTSTATUS status;
+  LARGE_INTEGER size = { .QuadPart = sizeof (fifo_shmem_t) };
+  SIZE_T viewsize = sizeof (fifo_shmem_t);
+  PVOID addr = NULL;
+  UNICODE_STRING uname;
+  WCHAR shmem_name[MAX_PATH];
+
+  __small_swprintf (shmem_name, L"fifo-shmem.%08x.%016X", get_dev (),
+    get_ino ());
+  RtlInitUnicodeString (&uname, shmem_name);
+  InitializeObjectAttributes (&attr, &uname, OBJ_INHERIT,
+      get_shared_parent_dir (), NULL);
+  status = NtCreateSection (&sect, STANDARD_RIGHTS_REQUIRED | SECTION_QUERY
+    | SECTION_MAP_READ | SECTION_MAP_WRITE,
+    &attr, &size, PAGE_READWRITE, SEC_COMMIT, NULL);
+  if (status == STATUS_OBJECT_NAME_COLLISION)
+    status = NtOpenSection (&sect, STANDARD_RIGHTS_REQUIRED | SECTION_QUERY
+    | SECTION_MAP_READ | SECTION_MAP_WRITE, &attr);
+  if (!NT_SUCCESS (status))
+    {
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  status = NtMapViewOfSection (sect, NtCurrentProcess (), &addr, 0, viewsize,
+       NULL, &viewsize, ViewShare, 0, PAGE_READWRITE);
+  if (!NT_SUCCESS (status))
+    {
+      NtClose (sect);
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  shmem_handle = sect;
+  shmem = (fifo_shmem_t *) addr;
+  return 0;
+}
+
+/* shmem_handle must be valid when this is called. */
+int
+fhandler_fifo::reopen_shmem ()
+{
+  NTSTATUS status;
+  SIZE_T viewsize = sizeof (fifo_shmem_t);
+  PVOID addr = NULL;
+
+  status = NtMapViewOfSection (shmem_handle, NtCurrentProcess (), &addr,
+       0, viewsize, NULL, &viewsize, ViewShare,
+       0, PAGE_READWRITE);
+  if (!NT_SUCCESS (status))
+    {
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  shmem = (fifo_shmem_t *) addr;
+  return 0;
+}
+
 int
 fhandler_fifo::open (int flags, mode_t)
 {
@@ -501,12 +563,15 @@ fhandler_fifo::open (int flags, mode_t)
       goto err_close_write_ready;
     }
 
-  /* If we're reading, signal read_ready and start the fifo_reader_thread. */
+  /* If we're reading, signal read_ready, create the shared memory,
+     and start the fifo_reader_thread. */
   if (reader)
     {
       SetEvent (read_ready);
-      if (!(cancel_evt = create_event ()))
+      if (create_shmem () < 0)
  goto err_close_writer_opening;
+      if (!(cancel_evt = create_event ()))
+ goto err_close_shmem;
       if (!(thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
@@ -615,6 +680,9 @@ err_close_reader:
   return 0;
 err_close_cancel_evt:
   NtClose (cancel_evt);
+err_close_shmem:
+  NtUnmapViewOfSection (NtCurrentProcess (), shmem);
+  NtClose (shmem_handle);
 err_close_writer_opening:
   NtClose (writer_opening);
 err_close_write_ready:
@@ -944,6 +1012,10 @@ fhandler_fifo::close ()
  dup/fork/exec; we should only reset read_ready when the last
  one closes. */
       ResetEvent (read_ready);
+      if (shmem)
+ NtUnmapViewOfSection (NtCurrentProcess (), shmem);
+      if (shmem_handle)
+ NtClose (shmem_handle);
     }
   if (read_ready)
     NtClose (read_ready);
@@ -1014,6 +1086,15 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
       /* Make sure the child starts unlocked. */
       fhf->fifo_client_unlock ();
 
+      if (!DuplicateHandle (GetCurrentProcess (), shmem_handle,
+    GetCurrentProcess (), &fhf->shmem_handle,
+    0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+ {
+  __seterrno ();
+  goto err_close_writer_opening;
+ }
+      if (fhf->reopen_shmem () < 0)
+ goto err_close_shmem_handle;
       fifo_client_lock ();
       for (i = 0; i < nhandlers; i++)
  {
@@ -1044,7 +1125,10 @@ err_close_cancel_evt:
 err_close_handlers:
   for (int j = 0; j < i; j++)
     fhf->fc_handler[j].close ();
-/* err_close_writer_opening: */
+  NtUnmapViewOfSection (GetCurrentProcess (), fhf->shmem);
+err_close_shmem_handle:
+  NtClose (fhf->shmem_handle);
+err_close_writer_opening:
   NtClose (fhf->writer_opening);
 err_close_write_ready:
   NtClose (fhf->write_ready);
@@ -1066,6 +1150,9 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       /* Make sure the child starts unlocked. */
       fifo_client_unlock ();
 
+      fork_fixup (parent, shmem_handle, "shmem_handle");
+      if (reopen_shmem () < 0)
+ api_fatal ("Can't reopen shared memory during fork, %E");
       fifo_client_lock ();
       for (int i = 0; i < nhandlers; i++)
  fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
@@ -1087,6 +1174,8 @@ fhandler_fifo::fixup_after_exec ()
       /* Make sure the child starts unlocked. */
       fifo_client_unlock ();
 
+      if (reopen_shmem () < 0)
+ api_fatal ("Can't reopen shared memory during exec, %E");
       if (!(cancel_evt = create_event ()))
  api_fatal ("Can't create reader thread cancel event during exec, %E");
       if (!(thr_sync_evt = create_event ()))
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 12/21] Cygwin: FIFO: keep track of the number of readers

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Add data and methods to the shared memory that keep track of the
number of open readers.

Increment this number in open, dup, fork, and exec.  Decrement it in
close.  Reset read_ready if there are no readers left.
---
 winsup/cygwin/fhandler.h       |  8 ++++++++
 winsup/cygwin/fhandler_fifo.cc | 22 ++++++++++++++--------
 2 files changed, 22 insertions(+), 8 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 8d6b94021..b2ee7e6b6 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1303,6 +1303,11 @@ struct fifo_client_handler
 /* Info needed by all readers of a FIFO, stored in named shared memory. */
 class fifo_shmem_t
 {
+  LONG _nreaders;
+
+public:
+  int inc_nreaders () { return (int) InterlockedIncrement (&_nreaders); }
+  int dec_nreaders () { return (int) InterlockedDecrement (&_nreaders); }
 };
 
 class fhandler_fifo: public fhandler_base
@@ -1342,6 +1347,9 @@ class fhandler_fifo: public fhandler_base
   int create_shmem ();
   int reopen_shmem ();
 
+  int inc_nreaders () { return shmem->inc_nreaders (); }
+  int dec_nreaders () { return shmem->dec_nreaders (); }
+
 public:
   fhandler_fifo ();
   bool hit_eof ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 9a0db3f33..d87070ac7 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -570,8 +570,9 @@ fhandler_fifo::open (int flags, mode_t)
       SetEvent (read_ready);
       if (create_shmem () < 0)
  goto err_close_writer_opening;
+      inc_nreaders ();
       if (!(cancel_evt = create_event ()))
- goto err_close_shmem;
+ goto err_dec_nreaders;
       if (!(thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
@@ -680,7 +681,10 @@ err_close_reader:
   return 0;
 err_close_cancel_evt:
   NtClose (cancel_evt);
-err_close_shmem:
+err_dec_nreaders:
+  if (dec_nreaders () == 0)
+    ResetEvent (read_ready);
+/* err_close_shmem: */
   NtUnmapViewOfSection (NtCurrentProcess (), shmem);
   NtClose (shmem_handle);
 err_close_writer_opening:
@@ -1003,15 +1007,13 @@ fhandler_fifo::close ()
 {
   if (reader)
     {
+      if (dec_nreaders () == 0)
+ ResetEvent (read_ready);
       cancel_reader_thread ();
       if (cancel_evt)
  NtClose (cancel_evt);
       if (thr_sync_evt)
  NtClose (thr_sync_evt);
-      /* FIXME: There could be several readers open because of
- dup/fork/exec; we should only reset read_ready when the last
- one closes. */
-      ResetEvent (read_ready);
       if (shmem)
  NtUnmapViewOfSection (NtCurrentProcess (), shmem);
       if (shmem_handle)
@@ -1116,8 +1118,8 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  goto err_close_handlers;
       if (!(fhf->thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
-      new cygthread (fifo_reader_thread, fhf, "fifo_reader",
-     fhf->thr_sync_evt);
+      inc_nreaders ();
+      new cygthread (fifo_reader_thread, fhf, "fifo_reader", fhf->thr_sync_evt);
     }
   return 0;
 err_close_cancel_evt:
@@ -1161,6 +1163,7 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
  api_fatal ("Can't create reader thread cancel event during fork, %E");
       if (!(thr_sync_evt = create_event ()))
  api_fatal ("Can't create reader thread sync event during fork, %E");
+      inc_nreaders ();
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
 }
@@ -1180,6 +1183,9 @@ fhandler_fifo::fixup_after_exec ()
  api_fatal ("Can't create reader thread cancel event during exec, %E");
       if (!(thr_sync_evt = create_event ()))
  api_fatal ("Can't create reader thread sync event during exec, %E");
+      /* At this moment we're a new reader.  The count will be
+ decremented when the parent closes. */
+      inc_nreaders ();
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
 }
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 13/21] Cygwin: FIFO: introduce a new type, fifo_reader_id_t

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
This uniquely identifies an fhandler_fifo open for reading in any
process.

Add a new data member 'me' of this type, which is set in open, dup,
fork, and exec.
---
 winsup/cygwin/fhandler.h       | 23 +++++++++++++++++++++++
 winsup/cygwin/fhandler_fifo.cc |  9 ++++++++-
 2 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index b2ee7e6b6..65aab4da3 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1300,6 +1300,26 @@ struct fifo_client_handler
   int pipe_state ();
 };
 
+class fhandler_fifo;
+
+struct fifo_reader_id_t
+{
+  DWORD winpid;
+  fhandler_fifo *fh;
+
+  operator bool () const { return winpid != 0 || fh != NULL; }
+
+  friend operator == (const fifo_reader_id_t &l, const fifo_reader_id_t &r)
+  {
+    return l.winpid == r.winpid && l.fh == r.fh;
+  }
+
+  friend operator != (const fifo_reader_id_t &l, const fifo_reader_id_t &r)
+  {
+    return l.winpid != r.winpid || l.fh != r.fh;
+  }
+};
+
 /* Info needed by all readers of a FIFO, stored in named shared memory. */
 class fifo_shmem_t
 {
@@ -1329,6 +1349,7 @@ class fhandler_fifo: public fhandler_base
   af_unix_spinlock_t _fifo_client_lock;
   bool reader, writer, duplexer;
   size_t max_atomic_write;
+  fifo_reader_id_t me;
 
   HANDLE shmem_handle;
   fifo_shmem_t *shmem;
@@ -1362,6 +1383,8 @@ public:
   void fifo_client_lock () { _fifo_client_lock.lock (); }
   void fifo_client_unlock () { _fifo_client_lock.unlock (); }
 
+  fifo_reader_id_t get_me () const { return me; }
+
   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 d87070ac7..5676a2c97 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -65,13 +65,15 @@ STATUS_PIPE_EMPTY simply means there's no data to be read. */
    || _s == STATUS_PIPE_NOT_AVAILABLE \
    || _s == STATUS_PIPE_BUSY; })
 
+static NO_COPY fifo_reader_id_t null_fr_id = { .winpid = 0, .fh = NULL };
+
 fhandler_fifo::fhandler_fifo ():
   fhandler_base (),
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
   cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false), nhandlers (0),
   reader (false), writer (false), duplexer (false),
   max_atomic_write (DEFAULT_PIPEBUFSIZE),
-  shmem_handle (NULL), shmem (NULL)
+  me (null_fr_id), shmem_handle (NULL), shmem (NULL)
 {
   pipe_name_buf[0] = L'\0';
   need_fork_fixup (true);
@@ -575,6 +577,8 @@ fhandler_fifo::open (int flags, mode_t)
  goto err_dec_nreaders;
       if (!(thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
+      me.winpid = GetCurrentProcessId ();
+      me.fh = this;
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
 
       /* If we're a duplexer, we need a handle for writing. */
@@ -1119,6 +1123,7 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
       if (!(fhf->thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
       inc_nreaders ();
+      fhf->me.fh = fhf;
       new cygthread (fifo_reader_thread, fhf, "fifo_reader", fhf->thr_sync_evt);
     }
   return 0;
@@ -1164,6 +1169,7 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       if (!(thr_sync_evt = create_event ()))
  api_fatal ("Can't create reader thread sync event during fork, %E");
       inc_nreaders ();
+      me.winpid = GetCurrentProcessId ();
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
 }
@@ -1179,6 +1185,7 @@ fhandler_fifo::fixup_after_exec ()
 
       if (reopen_shmem () < 0)
  api_fatal ("Can't reopen shared memory during exec, %E");
+      me.winpid = GetCurrentProcessId ();
       if (!(cancel_evt = create_event ()))
  api_fatal ("Can't create reader thread cancel event during exec, %E");
       if (!(thr_sync_evt = create_event ()))
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 14/21] Cygwin: FIFO: designate one reader as owner

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Among all the open readers of a FIFO, one is declared to be the owner.
This is the only reader that listens for client connections, and it is
the only one that has an accurate fc_handler list.

Add shared data and methods for getting and setting the owner, as well
as a lock to prevent more than one reader from accessing these data
simultaneously.

Modify the fifo_reader_thread so that it checks the owner at the
beginning of its loop.  If there is no owner, it takes ownership.  If
there is an owner but it is a different reader, the thread just waits
to be canceled.  Otherwise, it listens for client connections as
before.

Remove the 'first' argument from create_pipe_instance.  It is not
needed, and it may be confusing in the future since only the owner
knows whether a pipe instance is the first.

When opening a reader, don't return until the fifo_reader_thread has
time to set an owner.

If the owner closes, indicate that there is no longer an owner.

Clear the child's fc_handler list in dup, and don't bother duplicating
the handles.  The child never starts out as owner, so it can't use
those handles.

Do the same thing in fixup_after_fork in the close-on-exec case.  In
the non-close-on-exec case, the child inherits an fc_handler list that
it can't use, but we can just leave it alone; the handles will be
closed when the child is closed.
---
 winsup/cygwin/fhandler.h       |  13 +-
 winsup/cygwin/fhandler_fifo.cc | 237 ++++++++++++++++++---------------
 2 files changed, 141 insertions(+), 109 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 65aab4da3..bd44da5cd 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1324,10 +1324,17 @@ struct fifo_reader_id_t
 class fifo_shmem_t
 {
   LONG _nreaders;
+  fifo_reader_id_t _owner;
+  af_unix_spinlock_t _owner_lock;
 
 public:
   int inc_nreaders () { return (int) InterlockedIncrement (&_nreaders); }
   int dec_nreaders () { return (int) InterlockedDecrement (&_nreaders); }
+
+  fifo_reader_id_t get_owner () const { return _owner; }
+  void set_owner (fifo_reader_id_t fr_id) { _owner = fr_id; }
+  void owner_lock () { _owner_lock.lock (); }
+  void owner_unlock () { _owner_lock.unlock (); }
 };
 
 class fhandler_fifo: public fhandler_base
@@ -1356,7 +1363,7 @@ class fhandler_fifo: public fhandler_base
 
   bool __reg2 wait (HANDLE);
   static NTSTATUS npfs_handle (HANDLE &);
-  HANDLE create_pipe_instance (bool);
+  HANDLE create_pipe_instance ();
   NTSTATUS open_pipe (HANDLE&);
   NTSTATUS wait_open_pipe (HANDLE&);
   int add_client_handler ();
@@ -1384,6 +1391,10 @@ public:
   void fifo_client_unlock () { _fifo_client_lock.unlock (); }
 
   fifo_reader_id_t get_me () const { return me; }
+  fifo_reader_id_t get_owner () const { return shmem->get_owner (); }
+  void set_owner (fifo_reader_id_t fr_id) { shmem->set_owner (fr_id); }
+  void owner_lock () { shmem->owner_lock (); }
+  void owner_unlock () { shmem->owner_unlock (); }
 
   int open (int, mode_t);
   off_t lseek (off_t offset, int whence);
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 5676a2c97..0b9b33785 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -164,7 +164,7 @@ fhandler_fifo::npfs_handle (HANDLE &nph)
    blocking mode so that we can easily wait for a connection.  After
    it is connected, it is put in nonblocking mode. */
 HANDLE
-fhandler_fifo::create_pipe_instance (bool first)
+fhandler_fifo::create_pipe_instance ()
 {
   NTSTATUS status;
   HANDLE npfsh;
@@ -187,14 +187,12 @@ fhandler_fifo::create_pipe_instance (bool first)
   access = GENERIC_READ | FILE_READ_ATTRIBUTES | FILE_WRITE_ATTRIBUTES
     | SYNCHRONIZE;
   sharing = FILE_SHARE_READ | FILE_SHARE_WRITE;
-  hattr = openflags & O_CLOEXEC ? 0 : OBJ_INHERIT;
-  if (first)
-    hattr |= OBJ_CASE_INSENSITIVE;
+  hattr = (openflags & O_CLOEXEC ? 0 : OBJ_INHERIT) | OBJ_CASE_INSENSITIVE;
   InitializeObjectAttributes (&attr, get_pipe_name (),
       hattr, npfsh, NULL);
   timeout.QuadPart = -500000;
   status = NtCreateNamedPipeFile (&ph, access, &attr, &io, sharing,
-  first ? FILE_CREATE : FILE_OPEN, 0,
+  FILE_OPEN_IF, 0,
   FILE_PIPE_MESSAGE_TYPE
     | FILE_PIPE_REJECT_REMOTE_CLIENTS,
   FILE_PIPE_MESSAGE_MODE,
@@ -292,14 +290,13 @@ fhandler_fifo::add_client_handler ()
   int ret = -1;
   fifo_client_handler fc;
   HANDLE ph = NULL;
-  bool first = (nhandlers == 0);
 
   if (nhandlers == MAX_CLIENTS)
     {
       set_errno (EMFILE);
       goto out;
     }
-  ph = create_pipe_instance (first);
+  ph = create_pipe_instance ();
   if (!ph)
     goto out;
   else
@@ -349,92 +346,120 @@ fhandler_fifo::fifo_reader_thread_func ()
 
   while (1)
     {
-      /* Cleanup the fc_handler list. */
-      fifo_client_lock ();
-      int i = 0;
-      while (i < nhandlers)
+      fifo_reader_id_t cur_owner;
+
+      owner_lock ();
+      cur_owner = get_owner ();
+      if (!cur_owner)
  {
-  if (fc_handler[i].state < fc_connected)
-    delete_client_handler (i);
-  else
-    i++;
+  set_owner (me);
+  owner_unlock ();
+  continue;
+ }
+      else if (cur_owner != me)
+ {
+  owner_unlock ();
+  WaitForSingleObject (cancel_evt, INFINITE);
+  goto canceled;
  }
+      else
+ {
+  /* I'm the owner */
+  fifo_client_lock ();
 
-      /* Create a new client handler. */
-      if (add_client_handler () < 0)
- api_fatal ("Can't add a client handler, %E");
+  /* Cleanup the fc_handler list. */
+  fifo_client_lock ();
+  int i = 0;
+  while (i < nhandlers)
+    {
+      if (fc_handler[i].state < fc_connected)
+ delete_client_handler (i);
+      else
+ i++;
+    }
 
-      /* Listen for a writer to connect to the new client handler. */
-      fifo_client_handler& fc = fc_handler[nhandlers - 1];
-      fifo_client_unlock ();
-      NTSTATUS status;
-      IO_STATUS_BLOCK io;
-      bool cancel = false;
+  /* Create a new client handler. */
+  if (add_client_handler () < 0)
+    api_fatal ("Can't add a client handler, %E");
 
-      status = NtFsControlFile (fc.h, conn_evt, NULL, NULL, &io,
- FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
-      if (status == STATUS_PENDING)
- {
-  HANDLE w[2] = { conn_evt, cancel_evt };
-  switch (WaitForMultipleObjects (2, w, false, INFINITE))
+  /* Listen for a writer to connect to the new client handler. */
+  fifo_client_handler& fc = fc_handler[nhandlers - 1];
+  fifo_client_unlock ();
+  owner_unlock ();
+  NTSTATUS status;
+  IO_STATUS_BLOCK io;
+  bool cancel = false;
+
+  status = NtFsControlFile (fc.h, conn_evt, NULL, NULL, &io,
+    FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
+  if (status == STATUS_PENDING)
     {
-    case WAIT_OBJECT_0:
-      status = io.Status;
+      HANDLE w[2] = { conn_evt, cancel_evt };
+      switch (WaitForMultipleObjects (2, w, false, INFINITE))
+ {
+ case WAIT_OBJECT_0:
+  status = io.Status;
+  debug_printf ("NtFsControlFile STATUS_PENDING, then %y",
+ status);
+  break;
+ case WAIT_OBJECT_0 + 1:
+  status = STATUS_THREAD_IS_TERMINATING;
+  cancel = true;
+  break;
+ default:
+  api_fatal ("WFMO failed, %E");
+ }
+    }
+  else
+    debug_printf ("NtFsControlFile status %y, no STATUS_PENDING",
+  status);
+  HANDLE ph = NULL;
+  NTSTATUS status1;
+
+  fifo_client_lock ();
+  switch (status)
+    {
+    case STATUS_SUCCESS:
+    case STATUS_PIPE_CONNECTED:
+      record_connection (fc);
       break;
-    case WAIT_OBJECT_0 + 1:
-      status = STATUS_THREAD_IS_TERMINATING;
-      cancel = true;
+    case STATUS_PIPE_CLOSING:
+      record_connection (fc, fc_closing);
+      break;
+    case STATUS_THREAD_IS_TERMINATING:
+      /* Try to connect a bogus client.  Otherwise fc is still
+ listening, and the next connection might not get recorded. */
+      status1 = open_pipe (ph);
+      WaitForSingleObject (conn_evt, INFINITE);
+      if (NT_SUCCESS (status1))
+ /* Bogus cilent connected. */
+ delete_client_handler (nhandlers - 1);
+      else
+ /* Did a real client connect? */
+ switch (io.Status)
+  {
+  case STATUS_SUCCESS:
+  case STATUS_PIPE_CONNECTED:
+    record_connection (fc);
+    break;
+  case STATUS_PIPE_CLOSING:
+    record_connection (fc, fc_closing);
+    break;
+  default:
+    debug_printf ("NtFsControlFile status %y after failing to connect bogus client or real client", io.Status);
+    fc.state = fc_unknown;
+    break;
+  }
       break;
     default:
-      api_fatal ("WFMO failed, %E");
+      break;
     }
+  fifo_client_unlock ();
+  if (ph)
+    NtClose (ph);
+  if (cancel)
+    goto canceled;
  }
-      HANDLE ph = NULL;
-      NTSTATUS status1;
-
-      fifo_client_lock ();
-      switch (status)
- {
- case STATUS_SUCCESS:
- case STATUS_PIPE_CONNECTED:
-  record_connection (fc);
-  break;
- case STATUS_PIPE_CLOSING:
-  record_connection (fc, fc_closing);
-  break;
- case STATUS_THREAD_IS_TERMINATING:
-  /* Try to connect a bogus client.  Otherwise fc is still
-     listening, and the next connection might not get recorded. */
-  status1 = open_pipe (ph);
-  WaitForSingleObject (conn_evt, INFINITE);
-  if (NT_SUCCESS (status1))
-    /* Bogus cilent connected. */
-    delete_client_handler (nhandlers - 1);
-  else
-    /* Did a real client connect? */
-    switch (io.Status)
-      {
-      case STATUS_SUCCESS:
-      case STATUS_PIPE_CONNECTED:
- record_connection (fc);
- break;
-      case STATUS_PIPE_CLOSING:
- record_connection (fc, fc_closing);
- break;
-      default:
- debug_printf ("NtFsControlFile status %y after failing to connect bogus client or real client", io.Status);
- fc.state = fc_unknown;
- break;
-      }
-  break;
- default:
-  break;
- }
-      fifo_client_unlock ();
-      if (ph)
- NtClose (ph);
-      if (cancel)
- goto canceled;
     }
 canceled:
   if (conn_evt)
@@ -580,6 +605,15 @@ fhandler_fifo::open (int flags, mode_t)
       me.winpid = GetCurrentProcessId ();
       me.fh = this;
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
+      /* Wait until there's an owner. */
+      owner_lock ();
+      while (!get_owner ())
+ {
+  owner_unlock ();
+  yield ();
+  owner_lock ();
+ }
+      owner_unlock ();
 
       /* If we're a duplexer, we need a handle for writing. */
       if (duplexer)
@@ -1014,6 +1048,10 @@ fhandler_fifo::close ()
       if (dec_nreaders () == 0)
  ResetEvent (read_ready);
       cancel_reader_thread ();
+      owner_lock ();
+      if (get_owner () == me)
+ set_owner (null_fr_id);
+      owner_unlock ();
       if (cancel_evt)
  NtClose (cancel_evt);
       if (thr_sync_evt)
@@ -1056,7 +1094,6 @@ fhandler_fifo::fcntl (int cmd, intptr_t arg)
 int
 fhandler_fifo::dup (fhandler_base *child, int flags)
 {
-  int i = 0;
   fhandler_fifo *fhf = NULL;
 
   if (get_flags () & O_PATH)
@@ -1092,6 +1129,9 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
       /* Make sure the child starts unlocked. */
       fhf->fifo_client_unlock ();
 
+      /* Clear fc_handler list; the child never starts as owner. */
+      fhf->nhandlers = 0;
+
       if (!DuplicateHandle (GetCurrentProcess (), shmem_handle,
     GetCurrentProcess (), &fhf->shmem_handle,
     0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
@@ -1101,25 +1141,8 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  }
       if (fhf->reopen_shmem () < 0)
  goto err_close_shmem_handle;
-      fifo_client_lock ();
-      for (i = 0; i < nhandlers; i++)
- {
-  if (!DuplicateHandle (GetCurrentProcess (), fc_handler[i].h,
- GetCurrentProcess (), &fhf->fc_handler[i].h,
- 0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
-    {
-      __seterrno ();
-      break;
-    }
- }
-      if (i < nhandlers)
- {
-  fifo_client_unlock ();
-  goto err_close_handlers;
- }
-      fifo_client_unlock ();
       if (!(fhf->cancel_evt = create_event ()))
- goto err_close_handlers;
+ goto err_close_shmem;
       if (!(fhf->thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
       inc_nreaders ();
@@ -1129,9 +1152,7 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   return 0;
 err_close_cancel_evt:
   NtClose (fhf->cancel_evt);
-err_close_handlers:
-  for (int j = 0; j < i; j++)
-    fhf->fc_handler[j].close ();
+err_close_shmem:
   NtUnmapViewOfSection (GetCurrentProcess (), fhf->shmem);
 err_close_shmem_handle:
   NtClose (fhf->shmem_handle);
@@ -1160,10 +1181,10 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       fork_fixup (parent, shmem_handle, "shmem_handle");
       if (reopen_shmem () < 0)
  api_fatal ("Can't reopen shared memory during fork, %E");
-      fifo_client_lock ();
-      for (int i = 0; i < nhandlers; i++)
- fork_fixup (parent, fc_handler[i].h, "fc_handler[].h");
-      fifo_client_unlock ();
+      if (close_on_exec ())
+ /* Prevent a later attempt to close the non-inherited
+   pipe-instance handles copied from the parent. */
+ nhandlers = 0;
       if (!(cancel_evt = create_event ()))
  api_fatal ("Can't create reader thread cancel event during fork, %E");
       if (!(thr_sync_evt = create_event ()))
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 15/21] Cygwin: FIFO: allow fc_handler list to grow dynamically

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Make fc_handler a pointer to malloc'd memory instead of a fixed-size
array.  The size is now a new data member 'shandlers'.  Call realloc
in add_client_handler if we need to grow the array.

free fc_handler in close.  As long as we're touching that code, also
remove an unneeded lock.
---
 winsup/cygwin/fhandler.h       |  6 ++---
 winsup/cygwin/fhandler_fifo.cc | 41 +++++++++++++++++++---------------
 2 files changed, 26 insertions(+), 21 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index bd44da5cd..4f42cf1b8 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1268,7 +1268,6 @@ public:
 };
 
 #define CYGWIN_FIFO_PIPE_NAME_LEN     47
-#define MAX_CLIENTS 64
 
 /* The last three are the ones we try to read from. */
 enum fifo_client_connect_state
@@ -1351,8 +1350,9 @@ class fhandler_fifo: public fhandler_base
   UNICODE_STRING pipe_name;
   WCHAR pipe_name_buf[CYGWIN_FIFO_PIPE_NAME_LEN + 1];
   bool _maybe_eof;
-  fifo_client_handler fc_handler[MAX_CLIENTS];
-  int nhandlers;
+  fifo_client_handler *fc_handler;     /* Dynamically growing array. */
+  int shandlers;                       /* Size (capacity) of the array. */
+  int nhandlers;                       /* Number of elements in the array. */
   af_unix_spinlock_t _fifo_client_lock;
   bool reader, writer, duplexer;
   size_t max_atomic_write;
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 0b9b33785..595e55ad9 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -70,7 +70,8 @@ static NO_COPY fifo_reader_id_t null_fr_id = { .winpid = 0, .fh = NULL };
 fhandler_fifo::fhandler_fifo ():
   fhandler_base (),
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
-  cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false), nhandlers (0),
+  cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false),
+  fc_handler (NULL), shandlers (0), nhandlers (0),
   reader (false), writer (false), duplexer (false),
   max_atomic_write (DEFAULT_PIPEBUFSIZE),
   me (null_fr_id), shmem_handle (NULL), shmem (NULL)
@@ -287,27 +288,28 @@ fhandler_fifo::wait_open_pipe (HANDLE& ph)
 int
 fhandler_fifo::add_client_handler ()
 {
-  int ret = -1;
   fifo_client_handler fc;
   HANDLE ph = NULL;
 
-  if (nhandlers == MAX_CLIENTS)
+  if (nhandlers >= shandlers)
     {
-      set_errno (EMFILE);
-      goto out;
+      void *temp = realloc (fc_handler,
+    (shandlers += 64) * sizeof (fc_handler[0]));
+      if (!temp)
+ {
+  shandlers -= 64;
+  set_errno (ENOMEM);
+  return -1;
+ }
+      fc_handler = (fifo_client_handler *) temp;
     }
   ph = create_pipe_instance ();
   if (!ph)
-    goto out;
-  else
-    {
-      ret = 0;
-      fc.h = ph;
-      fc.state = fc_listening;
-      fc_handler[nhandlers++] = fc;
-    }
-out:
-  return ret;
+    return -1;
+  fc.h = ph;
+  fc.state = fc_listening;
+  fc_handler[nhandlers++] = fc;
+  return 0;
 }
 
 void
@@ -1067,10 +1069,10 @@ fhandler_fifo::close ()
     NtClose (write_ready);
   if (writer_opening)
     NtClose (writer_opening);
-  fifo_client_lock ();
   for (int i = 0; i < nhandlers; i++)
     fc_handler[i].close ();
-  fifo_client_unlock ();
+  if (fc_handler)
+    free (fc_handler);
   return fhandler_base::close ();
 }
 
@@ -1130,7 +1132,8 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
       fhf->fifo_client_unlock ();
 
       /* Clear fc_handler list; the child never starts as owner. */
-      fhf->nhandlers = 0;
+      fhf->nhandlers = fhf->shandlers = 0;
+      fhf->fc_handler = NULL;
 
       if (!DuplicateHandle (GetCurrentProcess (), shmem_handle,
     GetCurrentProcess (), &fhf->shmem_handle,
@@ -1206,6 +1209,8 @@ fhandler_fifo::fixup_after_exec ()
 
       if (reopen_shmem () < 0)
  api_fatal ("Can't reopen shared memory during exec, %E");
+      fc_handler = NULL;
+      nhandlers = shandlers = 0;
       me.winpid = GetCurrentProcessId ();
       if (!(cancel_evt = create_event ()))
  api_fatal ("Can't create reader thread cancel event during exec, %E");
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 16/21] Cygwin: FIFO: add a shared fifo_client_handler list

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
This is in a new shared memory section.  We will use it for temporary
storage of the owner's fc_handler list when we need to change owner.
The new owner can then duplicate the pipe handles from that list
before taking ownership.

Add several shared data members and methods that are needed for the
duplication process

Add methods update_my_handlers and update_shared_handlers that carry
out the duplication.

Allow the shared list to grow dynamically, up to a point.  Do this by
initially reserving a block of memory (currently 100 pages) and only
committing pages as needed.

Add methods create_shared_fc_handler, reopen_shared_fc_handler, and
remap_shared_fc_handler to create the new shared memory section,
reopen it, and commit new pages.  The first is called in open, the
second is called in dup/fork/exec, and the third is called in
update_shared_handlers if more shared memory is needed.

Modify the fifo_reader_thread function to call update_my_handlers when
it finds that there is no owner.  Also make it call
update_shared_handlers when the owner's thread terminates, so that the
new owner will have an accurate shared fc_handler list from which to
duplicate.

For convenience, add new methods cleanup_handlers and
close_all_handlers.  And add an optional arg to add_client_handler
that allows it to create a new fifo_client_handler without creating a
new pipe instance.
---
 winsup/cygwin/fhandler.h       |  43 +++++-
 winsup/cygwin/fhandler_fifo.cc | 253 +++++++++++++++++++++++++++++----
 2 files changed, 269 insertions(+), 27 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 4f42cf1b8..34b209f5d 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1323,17 +1323,33 @@ struct fifo_reader_id_t
 class fifo_shmem_t
 {
   LONG _nreaders;
-  fifo_reader_id_t _owner;
+  fifo_reader_id_t _owner, _prev_owner;
   af_unix_spinlock_t _owner_lock;
 
+  /* Info about shared memory block used for temporary storage of the
+     owner's fc_handler list. */
+  LONG _sh_nhandlers, _sh_shandlers, _sh_fc_handler_committed;
+
 public:
   int inc_nreaders () { return (int) InterlockedIncrement (&_nreaders); }
   int dec_nreaders () { return (int) InterlockedDecrement (&_nreaders); }
 
   fifo_reader_id_t get_owner () const { return _owner; }
   void set_owner (fifo_reader_id_t fr_id) { _owner = fr_id; }
+  fifo_reader_id_t get_prev_owner () const { return _prev_owner; }
+  void set_prev_owner (fifo_reader_id_t fr_id) { _prev_owner = fr_id; }
+
   void owner_lock () { _owner_lock.lock (); }
   void owner_unlock () { _owner_lock.unlock (); }
+
+  int get_shared_nhandlers () const { return (int) _sh_nhandlers; }
+  void set_shared_nhandlers (int n) { InterlockedExchange (&_sh_nhandlers, n); }
+  int get_shared_shandlers () const { return (int) _sh_shandlers; }
+  void set_shared_shandlers (int n) { InterlockedExchange (&_sh_shandlers, n); }
+  size_t get_shared_fc_handler_committed () const
+  { return (size_t) _sh_fc_handler_committed; }
+  void set_shared_fc_handler_committed (size_t n)
+  { InterlockedExchange (&_sh_fc_handler_committed, (LONG) n); }
 };
 
 class fhandler_fifo: public fhandler_base
@@ -1360,24 +1376,47 @@ class fhandler_fifo: public fhandler_base
 
   HANDLE shmem_handle;
   fifo_shmem_t *shmem;
+  HANDLE shared_fc_hdl;
+  /* Dynamically growing array in shared memory. */
+  fifo_client_handler *shared_fc_handler;
 
   bool __reg2 wait (HANDLE);
   static NTSTATUS npfs_handle (HANDLE &);
   HANDLE create_pipe_instance ();
   NTSTATUS open_pipe (HANDLE&);
   NTSTATUS wait_open_pipe (HANDLE&);
-  int add_client_handler ();
+  int add_client_handler (bool new_pipe_instance = true);
   void delete_client_handler (int);
+  void cleanup_handlers ();
+  void close_all_handlers ();
   void cancel_reader_thread ();
   void record_connection (fifo_client_handler&,
   fifo_client_connect_state = fc_connected);
 
   int create_shmem ();
   int reopen_shmem ();
+  int create_shared_fc_handler ();
+  int reopen_shared_fc_handler ();
+  int remap_shared_fc_handler (size_t);
 
   int inc_nreaders () { return shmem->inc_nreaders (); }
   int dec_nreaders () { return shmem->dec_nreaders (); }
 
+  fifo_reader_id_t get_prev_owner () const { return shmem->get_prev_owner (); }
+  void set_prev_owner (fifo_reader_id_t fr_id)
+  { shmem->set_prev_owner (fr_id); }
+
+  int get_shared_nhandlers () { return shmem->get_shared_nhandlers (); }
+  void set_shared_nhandlers (int n) { shmem->set_shared_nhandlers (n); }
+  int get_shared_shandlers () { return shmem->get_shared_shandlers (); }
+  void set_shared_shandlers (int n) { shmem->set_shared_shandlers (n); }
+  size_t get_shared_fc_handler_committed () const
+  { return shmem->get_shared_fc_handler_committed (); }
+  void set_shared_fc_handler_committed (size_t n)
+  { shmem->set_shared_fc_handler_committed (n); }
+  int update_my_handlers ();
+  int update_shared_handlers ();
+
 public:
   fhandler_fifo ();
   bool hit_eof ();
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 595e55ad9..846115ad4 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -21,6 +21,7 @@
 #include "shared_info.h"
 #include "ntdll.h"
 #include "cygwait.h"
+#include <sys/param.h>
 
 /*
    Overview:
@@ -65,6 +66,9 @@ STATUS_PIPE_EMPTY simply means there's no data to be read. */
    || _s == STATUS_PIPE_NOT_AVAILABLE \
    || _s == STATUS_PIPE_BUSY; })
 
+/* Number of pages reserved for shared_fc_handler. */
+#define SH_FC_HANDLER_PAGES 100
+
 static NO_COPY fifo_reader_id_t null_fr_id = { .winpid = 0, .fh = NULL };
 
 fhandler_fifo::fhandler_fifo ():
@@ -74,7 +78,8 @@ fhandler_fifo::fhandler_fifo ():
   fc_handler (NULL), shandlers (0), nhandlers (0),
   reader (false), writer (false), duplexer (false),
   max_atomic_write (DEFAULT_PIPEBUFSIZE),
-  me (null_fr_id), shmem_handle (NULL), shmem (NULL)
+  me (null_fr_id), shmem_handle (NULL), shmem (NULL),
+  shared_fc_hdl (NULL), shared_fc_handler (NULL)
 {
   pipe_name_buf[0] = L'\0';
   need_fork_fixup (true);
@@ -286,10 +291,9 @@ fhandler_fifo::wait_open_pipe (HANDLE& ph)
 }
 
 int
-fhandler_fifo::add_client_handler ()
+fhandler_fifo::add_client_handler (bool new_pipe_instance)
 {
   fifo_client_handler fc;
-  HANDLE ph = NULL;
 
   if (nhandlers >= shandlers)
     {
@@ -303,11 +307,14 @@ fhandler_fifo::add_client_handler ()
  }
       fc_handler = (fifo_client_handler *) temp;
     }
-  ph = create_pipe_instance ();
-  if (!ph)
-    return -1;
-  fc.h = ph;
-  fc.state = fc_listening;
+  if (new_pipe_instance)
+    {
+      HANDLE ph = create_pipe_instance ();
+      if (!ph)
+ return -1;
+      fc.h = ph;
+      fc.state = fc_listening;
+    }
   fc_handler[nhandlers++] = fc;
   return 0;
 }
@@ -321,6 +328,21 @@ fhandler_fifo::delete_client_handler (int i)
      (nhandlers - i) * sizeof (fc_handler[i]));
 }
 
+/* Delete invalid handlers. */
+void
+fhandler_fifo::cleanup_handlers ()
+{
+  int i = 0;
+
+  while (i < nhandlers)
+    {
+      if (fc_handler[i].state < fc_connected)
+ delete_client_handler (i);
+      else
+ i++;
+    }
+}
+
 void
 fhandler_fifo::record_connection (fifo_client_handler& fc,
   fifo_client_connect_state s)
@@ -331,6 +353,65 @@ fhandler_fifo::record_connection (fifo_client_handler& fc,
   set_pipe_non_blocking (fc.h, true);
 }
 
+/* Called from fifo_reader_thread_func with owner_lock in place. */
+int
+fhandler_fifo::update_my_handlers ()
+{
+  close_all_handlers ();
+  fifo_reader_id_t prev = get_prev_owner ();
+  if (!prev)
+    {
+      debug_printf ("No previous owner to copy handles from");
+      return 0;
+    }
+  HANDLE prev_proc;
+  if (prev.winpid == me.winpid)
+    prev_proc =  GetCurrentProcess ();
+  else
+    prev_proc = OpenProcess (PROCESS_DUP_HANDLE, false, prev.winpid);
+  if (!prev_proc)
+    {
+      debug_printf ("Can't open process of previous owner, %E");
+      __seterrno ();
+      return -1;
+    }
+
+  for (int i = 0; i < get_shared_nhandlers (); i++)
+    {
+      /* Should never happen. */
+      if (shared_fc_handler[i].state != fc_connected)
+ continue;
+      if (add_client_handler (false) < 0)
+ api_fatal ("Can't add client handler, %E");
+      fifo_client_handler &fc = fc_handler[nhandlers - 1];
+      if (!DuplicateHandle (prev_proc, shared_fc_handler[i].h,
+    GetCurrentProcess (), &fc.h, 0,
+    !close_on_exec (), DUPLICATE_SAME_ACCESS))
+ {
+  debug_printf ("Can't duplicate handle of previous owner, %E");
+  --nhandlers;
+  __seterrno ();
+  return -1;
+ }
+      fc.state = fc_connected;
+    }
+  return 0;
+}
+
+int
+fhandler_fifo::update_shared_handlers ()
+{
+  cleanup_handlers ();
+  if (nhandlers > get_shared_shandlers ())
+    {
+      if (remap_shared_fc_handler (nhandlers * sizeof (fc_handler[0])) < 0)
+ return -1;
+    }
+  set_shared_nhandlers (nhandlers);
+  memcpy (shared_fc_handler, fc_handler, nhandlers * sizeof (fc_handler[0]));
+  return 0;
+}
+
 static DWORD WINAPI
 fifo_reader_thread (LPVOID param)
 {
@@ -355,6 +436,8 @@ fhandler_fifo::fifo_reader_thread_func ()
       if (!cur_owner)
  {
   set_owner (me);
+  if (update_my_handlers () < 0)
+    api_fatal ("Can't update my handlers, %E");
   owner_unlock ();
   continue;
  }
@@ -368,19 +451,7 @@ fhandler_fifo::fifo_reader_thread_func ()
  {
   /* I'm the owner */
   fifo_client_lock ();
-
-  /* Cleanup the fc_handler list. */
-  fifo_client_lock ();
-  int i = 0;
-  while (i < nhandlers)
-    {
-      if (fc_handler[i].state < fc_connected)
- delete_client_handler (i);
-      else
- i++;
-    }
-
-  /* Create a new client handler. */
+  cleanup_handlers ();
   if (add_client_handler () < 0)
     api_fatal ("Can't add a client handler, %E");
 
@@ -391,6 +462,7 @@ fhandler_fifo::fifo_reader_thread_func ()
   NTSTATUS status;
   IO_STATUS_BLOCK io;
   bool cancel = false;
+  bool update = false;
 
   status = NtFsControlFile (fc.h, conn_evt, NULL, NULL, &io,
     FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
@@ -407,6 +479,7 @@ fhandler_fifo::fifo_reader_thread_func ()
  case WAIT_OBJECT_0 + 1:
   status = STATUS_THREAD_IS_TERMINATING;
   cancel = true;
+  update = true;
   break;
  default:
   api_fatal ("WFMO failed, %E");
@@ -459,6 +532,8 @@ fhandler_fifo::fifo_reader_thread_func ()
   fifo_client_unlock ();
   if (ph)
     NtClose (ph);
+  if (update && update_shared_handlers () < 0)
+    api_fatal ("Can't update shared handlers, %E");
   if (cancel)
     goto canceled;
  }
@@ -532,6 +607,100 @@ fhandler_fifo::reopen_shmem ()
   return 0;
 }
 
+/* On first creation, map and commit one page of memory. */
+int
+fhandler_fifo::create_shared_fc_handler ()
+{
+  HANDLE sect;
+  OBJECT_ATTRIBUTES attr;
+  NTSTATUS status;
+  LARGE_INTEGER size
+    = { .QuadPart = (LONGLONG) (SH_FC_HANDLER_PAGES * wincap.page_size ()) };
+  SIZE_T viewsize = get_shared_fc_handler_committed () ?: wincap.page_size ();
+  PVOID addr = NULL;
+  UNICODE_STRING uname;
+  WCHAR shared_fc_name[MAX_PATH];
+
+  __small_swprintf (shared_fc_name, L"fifo-shared-fc.%08x.%016X", get_dev (),
+    get_ino ());
+  RtlInitUnicodeString (&uname, shared_fc_name);
+  InitializeObjectAttributes (&attr, &uname, OBJ_INHERIT,
+      get_shared_parent_dir (), NULL);
+  status = NtCreateSection (&sect, STANDARD_RIGHTS_REQUIRED | SECTION_QUERY
+    | SECTION_MAP_READ | SECTION_MAP_WRITE, &attr,
+    &size, PAGE_READWRITE, SEC_RESERVE, NULL);
+  if (status == STATUS_OBJECT_NAME_COLLISION)
+    status = NtOpenSection (&sect, STANDARD_RIGHTS_REQUIRED | SECTION_QUERY
+    | SECTION_MAP_READ | SECTION_MAP_WRITE, &attr);
+  if (!NT_SUCCESS (status))
+    {
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  status = NtMapViewOfSection (sect, NtCurrentProcess (), &addr, 0, viewsize,
+       NULL, &viewsize, ViewShare, 0, PAGE_READWRITE);
+  if (!NT_SUCCESS (status))
+    {
+      NtClose (sect);
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  shared_fc_hdl = sect;
+  shared_fc_handler = (fifo_client_handler *) addr;
+  if (!get_shared_fc_handler_committed ())
+    set_shared_fc_handler_committed (viewsize);
+  set_shared_shandlers (viewsize / sizeof (fifo_client_handler));
+  return 0;
+}
+
+/* shared_fc_hdl must be valid when this is called. */
+int
+fhandler_fifo::reopen_shared_fc_handler ()
+{
+  NTSTATUS status;
+  SIZE_T viewsize = get_shared_fc_handler_committed ();
+  PVOID addr = NULL;
+
+  status = NtMapViewOfSection (shared_fc_hdl, NtCurrentProcess (),
+       &addr, 0, viewsize, NULL, &viewsize,
+       ViewShare, 0, PAGE_READWRITE);
+  if (!NT_SUCCESS (status))
+    {
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  shared_fc_handler = (fifo_client_handler *) addr;
+  return 0;
+}
+
+int
+fhandler_fifo::remap_shared_fc_handler (size_t nbytes)
+{
+  NTSTATUS status;
+  SIZE_T viewsize = roundup2 (nbytes, wincap.page_size ());
+  PVOID addr = NULL;
+
+  if (viewsize > SH_FC_HANDLER_PAGES * wincap.page_size ())
+    {
+      set_errno (ENOMEM);
+      return -1;
+    }
+
+  NtUnmapViewOfSection (NtCurrentProcess (), shared_fc_handler);
+  status = NtMapViewOfSection (shared_fc_hdl, NtCurrentProcess (),
+       &addr, 0, viewsize, NULL, &viewsize,
+       ViewShare, 0, PAGE_READWRITE);
+  if (!NT_SUCCESS (status))
+    {
+      __seterrno_from_nt_status (status);
+      return -1;
+    }
+  shared_fc_handler = (fifo_client_handler *) addr;
+  set_shared_fc_handler_committed (viewsize);
+  set_shared_shandlers (viewsize / sizeof (fc_handler[0]));
+  return 0;
+}
+
 int
 fhandler_fifo::open (int flags, mode_t)
 {
@@ -599,6 +768,8 @@ fhandler_fifo::open (int flags, mode_t)
       SetEvent (read_ready);
       if (create_shmem () < 0)
  goto err_close_writer_opening;
+      if (create_shared_fc_handler () < 0)
+ goto err_close_shmem;
       inc_nreaders ();
       if (!(cancel_evt = create_event ()))
  goto err_dec_nreaders;
@@ -724,7 +895,10 @@ err_close_cancel_evt:
 err_dec_nreaders:
   if (dec_nreaders () == 0)
     ResetEvent (read_ready);
-/* err_close_shmem: */
+/* err_close_shared_fc_handler: */
+  NtUnmapViewOfSection (NtCurrentProcess (), shared_fc_handler);
+  NtClose (shared_fc_hdl);
+err_close_shmem:
   NtUnmapViewOfSection (NtCurrentProcess (), shmem);
   NtClose (shmem_handle);
 err_close_writer_opening:
@@ -1012,6 +1186,14 @@ fhandler_fifo::fstatvfs (struct statvfs *sfs)
   return fh.fstatvfs (sfs);
 }
 
+void
+fhandler_fifo::close_all_handlers ()
+{
+  for (int i = 0; i < nhandlers; i++)
+    fc_handler[i].close ();
+  nhandlers = 0;
+}
+
 int
 fifo_client_handler::pipe_state ()
 {
@@ -1062,6 +1244,10 @@ fhandler_fifo::close ()
  NtUnmapViewOfSection (NtCurrentProcess (), shmem);
       if (shmem_handle)
  NtClose (shmem_handle);
+      if (shared_fc_handler)
+ NtUnmapViewOfSection (NtCurrentProcess (), shared_fc_handler);
+      if (shared_fc_hdl)
+ NtClose (shared_fc_hdl);
     }
   if (read_ready)
     NtClose (read_ready);
@@ -1069,8 +1255,7 @@ fhandler_fifo::close ()
     NtClose (write_ready);
   if (writer_opening)
     NtClose (writer_opening);
-  for (int i = 0; i < nhandlers; i++)
-    fc_handler[i].close ();
+  close_all_handlers ();
   if (fc_handler)
     free (fc_handler);
   return fhandler_base::close ();
@@ -1144,8 +1329,17 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  }
       if (fhf->reopen_shmem () < 0)
  goto err_close_shmem_handle;
+      if (!DuplicateHandle (GetCurrentProcess (), shared_fc_hdl,
+    GetCurrentProcess (), &fhf->shared_fc_hdl,
+    0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+ {
+  __seterrno ();
+  goto err_close_shmem;
+ }
+      if (fhf->reopen_shared_fc_handler () < 0)
+ goto err_close_shared_fc_hdl;
       if (!(fhf->cancel_evt = create_event ()))
- goto err_close_shmem;
+ goto err_close_shared_fc_handler;
       if (!(fhf->thr_sync_evt = create_event ()))
  goto err_close_cancel_evt;
       inc_nreaders ();
@@ -1155,6 +1349,10 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   return 0;
 err_close_cancel_evt:
   NtClose (fhf->cancel_evt);
+err_close_shared_fc_handler:
+  NtUnmapViewOfSection (GetCurrentProcess (), fhf->shared_fc_handler);
+err_close_shared_fc_hdl:
+  NtClose (fhf->shared_fc_hdl);
 err_close_shmem:
   NtUnmapViewOfSection (GetCurrentProcess (), fhf->shmem);
 err_close_shmem_handle:
@@ -1184,6 +1382,9 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       fork_fixup (parent, shmem_handle, "shmem_handle");
       if (reopen_shmem () < 0)
  api_fatal ("Can't reopen shared memory during fork, %E");
+      fork_fixup (parent, shared_fc_hdl, "shared_fc_hdl");
+      if (reopen_shared_fc_handler () < 0)
+ api_fatal ("Can't reopen shared fc_handler memory during fork, %E");
       if (close_on_exec ())
  /* Prevent a later attempt to close the non-inherited
    pipe-instance handles copied from the parent. */
@@ -1209,6 +1410,8 @@ fhandler_fifo::fixup_after_exec ()
 
       if (reopen_shmem () < 0)
  api_fatal ("Can't reopen shared memory during exec, %E");
+      if (reopen_shared_fc_handler () < 0)
+ api_fatal ("Can't reopen shared fc_handler memory during exec, %E");
       fc_handler = NULL;
       nhandlers = shandlers = 0;
       me.winpid = GetCurrentProcessId ();
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 17/21] Cygwin: FIFO: take ownership on exec

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
If fixup_after_exec is called on a non-close-on-exec reader whose
parent is the owner, transfer ownership to the child.  Otherwise the
parent's pipe handles will be closed before any other reader can
duplicate them.

To help with this, make the cancel_evt and thr_sync_evt handles
inheritable, so that the child can terminate the parent's
fifo_reader_thread (and the parent will update the shared fc_handler
list).

Add an optional argument 'from_exec' to update_my_handlers to simplify
its use in this case; no handle duplication is required.
---
 winsup/cygwin/fhandler.h       |   2 +-
 winsup/cygwin/fhandler_fifo.cc | 151 +++++++++++++++++++++++----------
 2 files changed, 107 insertions(+), 46 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 34b209f5d..1cd7d2b11 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1414,7 +1414,7 @@ class fhandler_fifo: public fhandler_base
   { return shmem->get_shared_fc_handler_committed (); }
   void set_shared_fc_handler_committed (size_t n)
   { shmem->set_shared_fc_handler_committed (n); }
-  int update_my_handlers ();
+  int update_my_handlers (bool from_exec = false);
   int update_shared_handlers ();
 
 public:
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 846115ad4..1c59bb3f4 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -104,13 +104,14 @@ sec_user_cloexec (bool cloexec, PSECURITY_ATTRIBUTES sa, PSID sid)
 }
 
 static HANDLE
-create_event ()
+create_event (bool inherit = false)
 {
   NTSTATUS status;
   OBJECT_ATTRIBUTES attr;
   HANDLE evt = NULL;
 
-  InitializeObjectAttributes (&attr, NULL, 0, NULL, NULL);
+  InitializeObjectAttributes (&attr, NULL, inherit ? OBJ_INHERIT : 0,
+      NULL, NULL);
   status = NtCreateEvent (&evt, EVENT_ALL_ACCESS, &attr,
   NotificationEvent, FALSE);
   if (!NT_SUCCESS (status))
@@ -353,47 +354,72 @@ fhandler_fifo::record_connection (fifo_client_handler& fc,
   set_pipe_non_blocking (fc.h, true);
 }
 
-/* Called from fifo_reader_thread_func with owner_lock in place. */
+/* Called from fifo_reader_thread_func with owner_lock in place, also
+   from fixup_after_exec with shared handles useable as they are. */
 int
-fhandler_fifo::update_my_handlers ()
+fhandler_fifo::update_my_handlers (bool from_exec)
 {
-  close_all_handlers ();
-  fifo_reader_id_t prev = get_prev_owner ();
-  if (!prev)
+  if (from_exec)
     {
-      debug_printf ("No previous owner to copy handles from");
-      return 0;
+      nhandlers = get_shared_nhandlers ();
+      if (nhandlers > shandlers)
+ {
+  int save = shandlers;
+  shandlers = nhandlers + 64;
+  void *temp = realloc (fc_handler,
+ shandlers * sizeof (fc_handler[0]));
+  if (!temp)
+    {
+      shandlers = save;
+      nhandlers = 0;
+      set_errno (ENOMEM);
+      return -1;
+    }
+  fc_handler = (fifo_client_handler *) temp;
+ }
+      memcpy (fc_handler, shared_fc_handler,
+      nhandlers * sizeof (fc_handler[0]));
     }
-  HANDLE prev_proc;
-  if (prev.winpid == me.winpid)
-    prev_proc =  GetCurrentProcess ();
   else
-    prev_proc = OpenProcess (PROCESS_DUP_HANDLE, false, prev.winpid);
-  if (!prev_proc)
     {
-      debug_printf ("Can't open process of previous owner, %E");
-      __seterrno ();
-      return -1;
-    }
-
-  for (int i = 0; i < get_shared_nhandlers (); i++)
-    {
-      /* Should never happen. */
-      if (shared_fc_handler[i].state != fc_connected)
- continue;
-      if (add_client_handler (false) < 0)
- api_fatal ("Can't add client handler, %E");
-      fifo_client_handler &fc = fc_handler[nhandlers - 1];
-      if (!DuplicateHandle (prev_proc, shared_fc_handler[i].h,
-    GetCurrentProcess (), &fc.h, 0,
-    !close_on_exec (), DUPLICATE_SAME_ACCESS))
+      close_all_handlers ();
+      fifo_reader_id_t prev = get_prev_owner ();
+      if (!prev)
+ {
+  debug_printf ("No previous owner to copy handles from");
+  return 0;
+ }
+      HANDLE prev_proc;
+      if (prev.winpid == me.winpid)
+ prev_proc =  GetCurrentProcess ();
+      else
+ prev_proc = OpenProcess (PROCESS_DUP_HANDLE, false, prev.winpid);
+      if (!prev_proc)
  {
-  debug_printf ("Can't duplicate handle of previous owner, %E");
-  --nhandlers;
+  debug_printf ("Can't open process of previous owner, %E");
   __seterrno ();
   return -1;
  }
-      fc.state = fc_connected;
+
+      for (int i = 0; i < get_shared_nhandlers (); i++)
+ {
+  /* Should never happen. */
+  if (shared_fc_handler[i].state != fc_connected)
+    continue;
+  if (add_client_handler (false) < 0)
+    api_fatal ("Can't add client handler, %E");
+  fifo_client_handler &fc = fc_handler[nhandlers - 1];
+  if (!DuplicateHandle (prev_proc, shared_fc_handler[i].h,
+ GetCurrentProcess (), &fc.h, 0,
+ !close_on_exec (), DUPLICATE_SAME_ACCESS))
+    {
+      debug_printf ("Can't duplicate handle of previous owner, %E");
+      --nhandlers;
+      __seterrno ();
+      return -1;
+    }
+  fc.state = fc_connected;
+ }
     }
   return 0;
 }
@@ -771,9 +797,9 @@ fhandler_fifo::open (int flags, mode_t)
       if (create_shared_fc_handler () < 0)
  goto err_close_shmem;
       inc_nreaders ();
-      if (!(cancel_evt = create_event ()))
+      if (!(cancel_evt = create_event (true)))
  goto err_dec_nreaders;
-      if (!(thr_sync_evt = create_event ()))
+      if (!(thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       me.winpid = GetCurrentProcessId ();
       me.fh = this;
@@ -1338,9 +1364,9 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  }
       if (fhf->reopen_shared_fc_handler () < 0)
  goto err_close_shared_fc_hdl;
-      if (!(fhf->cancel_evt = create_event ()))
+      if (!(fhf->cancel_evt = create_event (true)))
  goto err_close_shared_fc_handler;
-      if (!(fhf->thr_sync_evt = create_event ()))
+      if (!(fhf->thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       inc_nreaders ();
       fhf->me.fh = fhf;
@@ -1389,9 +1415,17 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
  /* Prevent a later attempt to close the non-inherited
    pipe-instance handles copied from the parent. */
  nhandlers = 0;
-      if (!(cancel_evt = create_event ()))
+      else
+ {
+  /* Close inherited handles needed only by exec. */
+  if (cancel_evt)
+    NtClose (cancel_evt);
+  if (thr_sync_evt)
+    NtClose (thr_sync_evt);
+ }
+      if (!(cancel_evt = create_event (true)))
  api_fatal ("Can't create reader thread cancel event during fork, %E");
-      if (!(thr_sync_evt = create_event ()))
+      if (!(thr_sync_evt = create_event (true)))
  api_fatal ("Can't create reader thread sync event during fork, %E");
       inc_nreaders ();
       me.winpid = GetCurrentProcessId ();
@@ -1414,10 +1448,32 @@ fhandler_fifo::fixup_after_exec ()
  api_fatal ("Can't reopen shared fc_handler memory during exec, %E");
       fc_handler = NULL;
       nhandlers = shandlers = 0;
+
+      /* Cancel parent's reader thread */
+      if (cancel_evt)
+ SetEvent (cancel_evt);
+      if (thr_sync_evt)
+ WaitForSingleObject (thr_sync_evt, INFINITE);
+
+      /* Take ownership if parent is owner. */
+      fifo_reader_id_t parent_fr = me;
       me.winpid = GetCurrentProcessId ();
-      if (!(cancel_evt = create_event ()))
+      owner_lock ();
+      if (get_owner () == parent_fr)
+ {
+  set_owner (me);
+  if (update_my_handlers (true) < 0)
+    api_fatal ("Can't update my handlers, %E");
+ }
+      owner_unlock ();
+      /* Close inherited cancel_evt and thr_sync_evt. */
+      if (cancel_evt)
+ NtClose (cancel_evt);
+      if (thr_sync_evt)
+ NtClose (thr_sync_evt);
+      if (!(cancel_evt = create_event (true)))
  api_fatal ("Can't create reader thread cancel event during exec, %E");
-      if (!(thr_sync_evt = create_event ()))
+      if (!(thr_sync_evt = create_event (true)))
  api_fatal ("Can't create reader thread sync event during exec, %E");
       /* At this moment we're a new reader.  The count will be
  decremented when the parent closes. */
@@ -1433,8 +1489,13 @@ fhandler_fifo::set_close_on_exec (bool val)
   set_no_inheritance (read_ready, val);
   set_no_inheritance (write_ready, val);
   set_no_inheritance (writer_opening, val);
-  fifo_client_lock ();
-  for (int i = 0; i < nhandlers; i++)
-    set_no_inheritance (fc_handler[i].h, val);
-  fifo_client_unlock ();
+  if (reader)
+    {
+      set_no_inheritance (cancel_evt, val);
+      set_no_inheritance (thr_sync_evt, val);
+      fifo_client_lock ();
+      for (int i = 0; i < nhandlers; i++)
+ set_no_inheritance (fc_handler[i].h, val);
+      fifo_client_unlock ();
+    }
 }
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 18/21] Cygwin: FIFO: find a new owner when closing

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
If the owning reader is closing, wait for another reader (if there is
one) to take ownership before closing the owner's pipe handles.

To synchronize the ownership transfer, add events owner_needed_evt and
owner_found_evt, and add methods owner_needed and owner_found to
set/reset them.

Modify the fifo_reader_thread function to wake up all non-owners when
a new owner is needed.

Make a cosmetic change in close so that fhandler_base::close is called
only if we have a write handle.  This prevents strace output from
being littered with statements that the null handle is being closed.
---
 winsup/cygwin/fhandler.h       |  14 +++++
 winsup/cygwin/fhandler_fifo.cc | 109 +++++++++++++++++++++++++++++----
 2 files changed, 112 insertions(+), 11 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 1cd7d2b11..f8c1b52a4 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1359,6 +1359,10 @@ class fhandler_fifo: public fhandler_base
   HANDLE write_ready;           /* A writer is open; OK for a reader to open. */
   HANDLE writer_opening;        /* A writer is opening; no EOF. */
 
+  /* Handles to named events needed by all readers of a given FIFO. */
+  HANDLE owner_needed_evt;      /* The owner is closing. */
+  HANDLE owner_found_evt;       /* A new owner has taken over. */
+
   /* Handles to non-shared events needed for fifo_reader_threads. */
   HANDLE cancel_evt;            /* Signal thread to terminate. */
   HANDLE thr_sync_evt;          /* The thread has terminated. */
@@ -1405,6 +1409,16 @@ class fhandler_fifo: public fhandler_base
   fifo_reader_id_t get_prev_owner () const { return shmem->get_prev_owner (); }
   void set_prev_owner (fifo_reader_id_t fr_id)
   { shmem->set_prev_owner (fr_id); }
+  void owner_needed ()
+  {
+    ResetEvent (owner_found_evt);
+    SetEvent (owner_needed_evt);
+  }
+  void owner_found ()
+  {
+    ResetEvent (owner_needed_evt);
+    SetEvent (owner_found_evt);
+  }
 
   int get_shared_nhandlers () { return shmem->get_shared_nhandlers (); }
   void set_shared_nhandlers (int n) { shmem->set_shared_nhandlers (n); }
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 1c59bb3f4..bf33a52d6 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -74,6 +74,7 @@ static NO_COPY fifo_reader_id_t null_fr_id = { .winpid = 0, .fh = NULL };
 fhandler_fifo::fhandler_fifo ():
   fhandler_base (),
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
+  owner_needed_evt (NULL), owner_found_evt (NULL),
   cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false),
   fc_handler (NULL), shandlers (0), nhandlers (0),
   reader (false), writer (false), duplexer (false),
@@ -464,14 +465,23 @@ fhandler_fifo::fifo_reader_thread_func ()
   set_owner (me);
   if (update_my_handlers () < 0)
     api_fatal ("Can't update my handlers, %E");
+  owner_found ();
   owner_unlock ();
   continue;
  }
       else if (cur_owner != me)
  {
   owner_unlock ();
-  WaitForSingleObject (cancel_evt, INFINITE);
-  goto canceled;
+  HANDLE w[2] = { owner_needed_evt, cancel_evt };
+  switch (WaitForMultipleObjects (2, w, false, INFINITE))
+    {
+    case WAIT_OBJECT_0:
+      continue;
+    case WAIT_OBJECT_0 + 1:
+      goto canceled;
+    default:
+      api_fatal ("WFMO failed, %E");
+    }
  }
       else
  {
@@ -797,8 +807,23 @@ fhandler_fifo::open (int flags, mode_t)
       if (create_shared_fc_handler () < 0)
  goto err_close_shmem;
       inc_nreaders ();
+      npbuf[0] = 'n';
+      if (!(owner_needed_evt = CreateEvent (sa_buf, true, false, npbuf)))
+ {
+  debug_printf ("CreateEvent for %s failed, %E", npbuf);
+  __seterrno ();
+  goto err_dec_nreaders;
+ }
+      npbuf[0] = 'f';
+      if (!(owner_found_evt = CreateEvent (sa_buf, true, false, npbuf)))
+ {
+  debug_printf ("CreateEvent for %s failed, %E", npbuf);
+  __seterrno ();
+  goto err_close_owner_needed_evt;
+ }
+      /* Make cancel and sync inheritable for exec. */
       if (!(cancel_evt = create_event (true)))
- goto err_dec_nreaders;
+ goto err_close_owner_found_evt;
       if (!(thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       me.winpid = GetCurrentProcessId ();
@@ -918,6 +943,10 @@ err_close_reader:
   return 0;
 err_close_cancel_evt:
   NtClose (cancel_evt);
+err_close_owner_found_evt:
+  NtClose (owner_found_evt);
+err_close_owner_needed_evt:
+  NtClose (owner_needed_evt);
 err_dec_nreaders:
   if (dec_nreaders () == 0)
     ResetEvent (read_ready);
@@ -1255,13 +1284,49 @@ fhandler_fifo::close ()
 {
   if (reader)
     {
-      if (dec_nreaders () == 0)
- ResetEvent (read_ready);
+      /* If we're the owner, try to find a new owner. */
+      bool find_new_owner = false;
+
       cancel_reader_thread ();
       owner_lock ();
       if (get_owner () == me)
- set_owner (null_fr_id);
+ {
+  find_new_owner = true;
+  set_owner (null_fr_id);
+  set_prev_owner (me);
+  owner_needed ();
+ }
       owner_unlock ();
+      if (dec_nreaders () == 0)
+ ResetEvent (read_ready);
+      else if (find_new_owner && !IsEventSignalled (owner_found_evt))
+ {
+  bool found = false;
+  do
+    if (dec_nreaders () >= 0)
+      {
+ /* There's still another reader open. */
+ if (WaitForSingleObject (owner_found_evt, 1) == WAIT_OBJECT_0)
+  found = true;
+ else
+  {
+    owner_lock ();
+    if (get_owner ()) /* We missed owner_found_evt? */
+      found = true;
+    else
+      owner_needed ();
+    owner_unlock ();
+  }
+      }
+  while (inc_nreaders () > 0 && !found);
+ }
+      close_all_handlers ();
+      if (fc_handler)
+ free (fc_handler);
+      if (owner_needed_evt)
+ NtClose (owner_needed_evt);
+      if (owner_found_evt)
+ NtClose (owner_found_evt);
       if (cancel_evt)
  NtClose (cancel_evt);
       if (thr_sync_evt)
@@ -1281,10 +1346,10 @@ fhandler_fifo::close ()
     NtClose (write_ready);
   if (writer_opening)
     NtClose (writer_opening);
-  close_all_handlers ();
-  if (fc_handler)
-    free (fc_handler);
-  return fhandler_base::close ();
+  if (nohandle ())
+    return 0;
+  else
+    return fhandler_base::close ();
 }
 
 /* If we have a write handle (i.e., we're a duplexer or a writer),
@@ -1364,8 +1429,22 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
  }
       if (fhf->reopen_shared_fc_handler () < 0)
  goto err_close_shared_fc_hdl;
+      if (!DuplicateHandle (GetCurrentProcess (), owner_needed_evt,
+    GetCurrentProcess (), &fhf->owner_needed_evt,
+    0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+ {
+  __seterrno ();
+  goto err_close_shared_fc_handler;
+ }
+      if (!DuplicateHandle (GetCurrentProcess (), owner_found_evt,
+    GetCurrentProcess (), &fhf->owner_found_evt,
+    0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+ {
+  __seterrno ();
+  goto err_close_owner_needed_evt;
+ }
       if (!(fhf->cancel_evt = create_event (true)))
- goto err_close_shared_fc_handler;
+ goto err_close_owner_found_evt;
       if (!(fhf->thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       inc_nreaders ();
@@ -1375,6 +1454,10 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   return 0;
 err_close_cancel_evt:
   NtClose (fhf->cancel_evt);
+err_close_owner_found_evt:
+  NtClose (fhf->owner_found_evt);
+err_close_owner_needed_evt:
+  NtClose (fhf->owner_needed_evt);
 err_close_shared_fc_handler:
   NtUnmapViewOfSection (GetCurrentProcess (), fhf->shared_fc_handler);
 err_close_shared_fc_hdl:
@@ -1411,6 +1494,8 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       fork_fixup (parent, shared_fc_hdl, "shared_fc_hdl");
       if (reopen_shared_fc_handler () < 0)
  api_fatal ("Can't reopen shared fc_handler memory during fork, %E");
+      fork_fixup (parent, owner_needed_evt, "owner_needed_evt");
+      fork_fixup (parent, owner_found_evt, "owner_found_evt");
       if (close_on_exec ())
  /* Prevent a later attempt to close the non-inherited
    pipe-instance handles copied from the parent. */
@@ -1491,6 +1576,8 @@ fhandler_fifo::set_close_on_exec (bool val)
   set_no_inheritance (writer_opening, val);
   if (reader)
     {
+      set_no_inheritance (owner_needed_evt, val);
+      set_no_inheritance (owner_found_evt, val);
       set_no_inheritance (cancel_evt, val);
       set_no_inheritance (thr_sync_evt, val);
       fifo_client_lock ();
--
2.21.0

Reply | Threaded
Open this post in threaded view
|

[PATCH 19/21] Cygwin: FIFO: allow any reader to take ownership

cygwin-patches mailing list
In reply to this post by cygwin-patches mailing list
Add a take_ownership method, used by raw_read and select.cc:peek_fifo.
It wakes up all fifo_reader_threads and allows the caller to become
owner.  The work is done by the fifo_reader_threads.

For synchronization we introduce several new fhandler_fifo data
members and methods:

- update_needed_evt signals the current owner to stop listening for
  writer connections and update its fc_handler list.

- shared_fc_handler() gets and sets the status of the fc_handler
  update process.

- get_pending_owner() and set_pending_owner() get and set the reader
  that is requesting ownership.

Finally, a new 'reading_lock' prevents two readers from trying to take
ownership simultaneously.
---
 winsup/cygwin/fhandler.h       |  28 ++++++++-
 winsup/cygwin/fhandler_fifo.cc | 106 +++++++++++++++++++++++++++++----
 winsup/cygwin/select.cc        |   4 ++
 3 files changed, 122 insertions(+), 16 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index f8c1b52a4..31c65866e 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1323,12 +1323,13 @@ struct fifo_reader_id_t
 class fifo_shmem_t
 {
   LONG _nreaders;
-  fifo_reader_id_t _owner, _prev_owner;
-  af_unix_spinlock_t _owner_lock;
+  fifo_reader_id_t _owner, _prev_owner, _pending_owner;
+  af_unix_spinlock_t _owner_lock, _reading_lock;
 
   /* Info about shared memory block used for temporary storage of the
      owner's fc_handler list. */
-  LONG _sh_nhandlers, _sh_shandlers, _sh_fc_handler_committed;
+  LONG _sh_nhandlers, _sh_shandlers, _sh_fc_handler_committed,
+    _sh_fc_handler_updated;
 
 public:
   int inc_nreaders () { return (int) InterlockedIncrement (&_nreaders); }
@@ -1338,9 +1339,13 @@ public:
   void set_owner (fifo_reader_id_t fr_id) { _owner = fr_id; }
   fifo_reader_id_t get_prev_owner () const { return _prev_owner; }
   void set_prev_owner (fifo_reader_id_t fr_id) { _prev_owner = fr_id; }
+  fifo_reader_id_t get_pending_owner () const { return _pending_owner; }
+  void set_pending_owner (fifo_reader_id_t fr_id) { _pending_owner = fr_id; }
 
   void owner_lock () { _owner_lock.lock (); }
   void owner_unlock () { _owner_lock.unlock (); }
+  void reading_lock () { _reading_lock.lock (); }
+  void reading_unlock () { _reading_lock.unlock (); }
 
   int get_shared_nhandlers () const { return (int) _sh_nhandlers; }
   void set_shared_nhandlers (int n) { InterlockedExchange (&_sh_nhandlers, n); }
@@ -1350,6 +1355,9 @@ public:
   { return (size_t) _sh_fc_handler_committed; }
   void set_shared_fc_handler_committed (size_t n)
   { InterlockedExchange (&_sh_fc_handler_committed, (LONG) n); }
+  bool shared_fc_handler_updated () const { return _sh_fc_handler_updated; }
+  void shared_fc_handler_updated (bool val)
+  { InterlockedExchange (&_sh_fc_handler_updated, val); }
 };
 
 class fhandler_fifo: public fhandler_base
@@ -1362,6 +1370,7 @@ class fhandler_fifo: public fhandler_base
   /* Handles to named events needed by all readers of a given FIFO. */
   HANDLE owner_needed_evt;      /* The owner is closing. */
   HANDLE owner_found_evt;       /* A new owner has taken over. */
+  HANDLE update_needed_evt;     /* shared_fc_handler needs updating. */
 
   /* Handles to non-shared events needed for fifo_reader_threads. */
   HANDLE cancel_evt;            /* Signal thread to terminate. */
@@ -1409,6 +1418,11 @@ class fhandler_fifo: public fhandler_base
   fifo_reader_id_t get_prev_owner () const { return shmem->get_prev_owner (); }
   void set_prev_owner (fifo_reader_id_t fr_id)
   { shmem->set_prev_owner (fr_id); }
+  fifo_reader_id_t get_pending_owner () const
+  { return shmem->get_pending_owner (); }
+  void set_pending_owner (fifo_reader_id_t fr_id)
+  { shmem->set_pending_owner (fr_id); }
+
   void owner_needed ()
   {
     ResetEvent (owner_found_evt);
@@ -1430,6 +1444,10 @@ class fhandler_fifo: public fhandler_base
   { shmem->set_shared_fc_handler_committed (n); }
   int update_my_handlers (bool from_exec = false);
   int update_shared_handlers ();
+  bool shared_fc_handler_updated () const
+  { return shmem->shared_fc_handler_updated (); }
+  void shared_fc_handler_updated (bool val)
+  { shmem->shared_fc_handler_updated (val); }
 
 public:
   fhandler_fifo ();
@@ -1449,6 +1467,10 @@ public:
   void owner_lock () { shmem->owner_lock (); }
   void owner_unlock () { shmem->owner_unlock (); }
 
+  void take_ownership ();
+  void reading_lock () { shmem->reading_lock (); }
+  void reading_unlock () { shmem->reading_unlock (); }
+
   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 bf33a52d6..81473015e 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -74,7 +74,7 @@ static NO_COPY fifo_reader_id_t null_fr_id = { .winpid = 0, .fh = NULL };
 fhandler_fifo::fhandler_fifo ():
   fhandler_base (),
   read_ready (NULL), write_ready (NULL), writer_opening (NULL),
-  owner_needed_evt (NULL), owner_found_evt (NULL),
+  owner_needed_evt (NULL), owner_found_evt (NULL), update_needed_evt (NULL),
   cancel_evt (NULL), thr_sync_evt (NULL), _maybe_eof (false),
   fc_handler (NULL), shandlers (0), nhandlers (0),
   reader (false), writer (false), duplexer (false),
@@ -436,6 +436,8 @@ fhandler_fifo::update_shared_handlers ()
     }
   set_shared_nhandlers (nhandlers);
   memcpy (shared_fc_handler, fc_handler, nhandlers * sizeof (fc_handler[0]));
+  shared_fc_handler_updated (true);
+  set_prev_owner (me);
   return 0;
 }
 
@@ -456,20 +458,44 @@ fhandler_fifo::fifo_reader_thread_func ()
 
   while (1)
     {
-      fifo_reader_id_t cur_owner;
+      fifo_reader_id_t cur_owner, pending_owner;
+      bool idle = false, take_ownership = false;
 
       owner_lock ();
       cur_owner = get_owner ();
-      if (!cur_owner)
+      pending_owner = get_pending_owner ();
+
+      if (pending_owner)
  {
-  set_owner (me);
-  if (update_my_handlers () < 0)
-    api_fatal ("Can't update my handlers, %E");
-  owner_found ();
-  owner_unlock ();
-  continue;
+  if (pending_owner != me)
+    idle = true;
+  else
+    take_ownership = true;
  }
+      else if (!cur_owner)
+ take_ownership = true;
       else if (cur_owner != me)
+ idle = true;
+      if (take_ownership)
+ {
+  if (!shared_fc_handler_updated ())
+    {
+      owner_unlock ();
+      yield ();
+      continue;
+    }
+  else
+    {
+      set_owner (me);
+      set_pending_owner (null_fr_id);
+      if (update_my_handlers () < 0)
+ api_fatal ("Can't update my handlers, %E");
+      owner_found ();
+      owner_unlock ();
+      continue;
+    }
+ }
+      else if (idle)
  {
   owner_unlock ();
   HANDLE w[2] = { owner_needed_evt, cancel_evt };
@@ -494,6 +520,7 @@ fhandler_fifo::fifo_reader_thread_func ()
   /* Listen for a writer to connect to the new client handler. */
   fifo_client_handler& fc = fc_handler[nhandlers - 1];
   fifo_client_unlock ();
+  shared_fc_handler_updated (false);
   owner_unlock ();
   NTSTATUS status;
   IO_STATUS_BLOCK io;
@@ -504,8 +531,8 @@ fhandler_fifo::fifo_reader_thread_func ()
     FSCTL_PIPE_LISTEN, NULL, 0, NULL, 0);
   if (status == STATUS_PENDING)
     {
-      HANDLE w[2] = { conn_evt, cancel_evt };
-      switch (WaitForMultipleObjects (2, w, false, INFINITE))
+      HANDLE w[3] = { conn_evt, update_needed_evt, cancel_evt };
+      switch (WaitForMultipleObjects (3, w, false, INFINITE))
  {
  case WAIT_OBJECT_0:
   status = io.Status;
@@ -513,6 +540,10 @@ fhandler_fifo::fifo_reader_thread_func ()
  status);
   break;
  case WAIT_OBJECT_0 + 1:
+  status = STATUS_WAIT_1;
+  update = true;
+  break;
+ case WAIT_OBJECT_0 + 2:
   status = STATUS_THREAD_IS_TERMINATING;
   cancel = true;
   update = true;
@@ -538,6 +569,7 @@ fhandler_fifo::fifo_reader_thread_func ()
       record_connection (fc, fc_closing);
       break;
     case STATUS_THREAD_IS_TERMINATING:
+    case STATUS_WAIT_1:
       /* Try to connect a bogus client.  Otherwise fc is still
  listening, and the next connection might not get recorded. */
       status1 = open_pipe (ph);
@@ -807,6 +839,8 @@ fhandler_fifo::open (int flags, mode_t)
       if (create_shared_fc_handler () < 0)
  goto err_close_shmem;
       inc_nreaders ();
+      /* Reinitialize _sh_fc_handler_updated, which starts as 0. */
+      shared_fc_handler_updated (true);
       npbuf[0] = 'n';
       if (!(owner_needed_evt = CreateEvent (sa_buf, true, false, npbuf)))
  {
@@ -821,9 +855,16 @@ fhandler_fifo::open (int flags, mode_t)
   __seterrno ();
   goto err_close_owner_needed_evt;
  }
+      npbuf[0] = 'u';
+      if (!(update_needed_evt = CreateEvent (sa_buf, false, false, npbuf)))
+ {
+  debug_printf ("CreateEvent for %s failed, %E", npbuf);
+  __seterrno ();
+  goto err_close_owner_found_evt;
+ }
       /* Make cancel and sync inheritable for exec. */
       if (!(cancel_evt = create_event (true)))
- goto err_close_owner_found_evt;
+ goto err_close_update_needed_evt;
       if (!(thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       me.winpid = GetCurrentProcessId ();
@@ -943,6 +984,8 @@ err_close_reader:
   return 0;
 err_close_cancel_evt:
   NtClose (cancel_evt);
+err_close_update_needed_evt:
+  NtClose (update_needed_evt);
 err_close_owner_found_evt:
   NtClose (owner_found_evt);
 err_close_owner_needed_evt:
@@ -1136,6 +1179,24 @@ fhandler_fifo::hit_eof ()
   return ret;
 }
 
+/* Called from raw_read and select.cc:peek_fifo. */
+void
+fhandler_fifo::take_ownership ()
+{
+  owner_lock ();
+  if (get_owner () == me)
+    {
+      owner_unlock ();
+      return;
+    }
+  set_pending_owner (me);
+  owner_needed ();
+  SetEvent (update_needed_evt);
+  owner_unlock ();
+  /* The reader threads should now do the transfer.  */
+  WaitForSingleObject (owner_found_evt, INFINITE);
+}
+
 void __reg3
 fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 {
@@ -1144,6 +1205,9 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 
   while (1)
     {
+      /* No one else can take ownership while we hold the reading_lock. */
+      reading_lock ();
+      take_ownership ();
       /* Poll the connected clients for input. */
       int nconnected = 0;
       fifo_client_lock ();
@@ -1167,6 +1231,7 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
   {
     len = nbytes;
     fifo_client_unlock ();
+    reading_unlock ();
     return;
   }
  break;
@@ -1187,9 +1252,11 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
       fifo_client_unlock ();
       if (maybe_eof () && hit_eof ())
  {
+  reading_unlock ();
   len = 0;
   return;
  }
+      reading_unlock ();
       if (is_nonblocking ())
  {
   set_errno (EAGAIN);
@@ -1327,6 +1394,8 @@ fhandler_fifo::close ()
  NtClose (owner_needed_evt);
       if (owner_found_evt)
  NtClose (owner_found_evt);
+      if (update_needed_evt)
+ NtClose (update_needed_evt);
       if (cancel_evt)
  NtClose (cancel_evt);
       if (thr_sync_evt)
@@ -1443,8 +1512,15 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   __seterrno ();
   goto err_close_owner_needed_evt;
  }
+      if (!DuplicateHandle (GetCurrentProcess (), update_needed_evt,
+    GetCurrentProcess (), &fhf->update_needed_evt,
+    0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+ {
+  __seterrno ();
+  goto err_close_owner_found_evt;
+ }
       if (!(fhf->cancel_evt = create_event (true)))
- goto err_close_owner_found_evt;
+ goto err_close_update_needed_evt;
       if (!(fhf->thr_sync_evt = create_event (true)))
  goto err_close_cancel_evt;
       inc_nreaders ();
@@ -1454,6 +1530,8 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
   return 0;
 err_close_cancel_evt:
   NtClose (fhf->cancel_evt);
+err_close_update_needed_evt:
+  NtClose (fhf->update_needed_evt);
 err_close_owner_found_evt:
   NtClose (fhf->owner_found_evt);
 err_close_owner_needed_evt:
@@ -1496,6 +1574,7 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
  api_fatal ("Can't reopen shared fc_handler memory during fork, %E");
       fork_fixup (parent, owner_needed_evt, "owner_needed_evt");
       fork_fixup (parent, owner_found_evt, "owner_found_evt");
+      fork_fixup (parent, update_needed_evt, "update_needed_evt");
       if (close_on_exec ())
  /* Prevent a later attempt to close the non-inherited
    pipe-instance handles copied from the parent. */
@@ -1578,6 +1657,7 @@ fhandler_fifo::set_close_on_exec (bool val)
     {
       set_no_inheritance (owner_needed_evt, val);
       set_no_inheritance (owner_found_evt, val);
+      set_no_inheritance (update_needed_evt, val);
       set_no_inheritance (cancel_evt, val);
       set_no_inheritance (thr_sync_evt, val);
       fifo_client_lock ();
diff --git a/winsup/cygwin/select.cc b/winsup/cygwin/select.cc
index 9323c423f..2c299acf7 100644
--- a/winsup/cygwin/select.cc
+++ b/winsup/cygwin/select.cc
@@ -866,6 +866,8 @@ peek_fifo (select_record *s, bool from_select)
   goto out;
  }
 
+      fh->reading_lock ();
+      fh->take_ownership ();
       fh->fifo_client_lock ();
       int nconnected = 0;
       for (int i = 0; i < fh->get_nhandlers (); i++)
@@ -888,6 +890,7 @@ peek_fifo (select_record *s, bool from_select)
  fh->get_fc_handler (i).get_state () = fc_input_avail;
  select_printf ("read: %s, ready for read", fh->get_name ());
  fh->fifo_client_unlock ();
+ fh->reading_unlock ();
  gotone += s->read_ready = true;
  goto out;
       default:
@@ -905,6 +908,7 @@ peek_fifo (select_record *s, bool from_select)
   if (s->except_selected)
     gotone += s->except_ready = true;
  }
+      fh->reading_unlock ();
     }
 out:
   if (s->write_selected)
--
2.21.0

12