Cygwin: pipes: workaround unrelibale system info

FILE_PIPE_LOCAL_INFORMATION::WriteQuotaAvailable is unreliable.

Usually WriteQuotaAvailable on the write side reflects the space
available in the inbound buffer on the read side.  However, if a
pipe read is currently pending, WriteQuotaAvailable on the write side
is decremented by the number of bytes the read side is requesting.
So it's possible (even likely) that WriteQuotaAvailable is 0, even
if the inbound buffer on the read side is not full.  This can lead to
a deadlock situation: The reader is waiting for data, but select
on the writer side assumes that no space is available in the read
side inbound buffer.

This patch implements a workaround by never trying to read more than
half the buffer size blocking if the read buffer is empty.  This first
cut tries to take the number of open readers into account by reducing
the amount of requested bytes accordingly.

Signed-off-by: Corinna Vinschen <corinna@vinschen.de>

winsup/cygwin/fhandler.h

@@ -1171,6 +1171,7 @@ class fhandler_socket_unix : public fhandler_socket
 class fhandler_pipe: public fhandler_base
 {
 private:
+  HANDLE read_mtx;
   pid_t popen_pid;
   size_t max_atomic_write;
   void set_pipe_non_blocking (bool nonblocking);
@@ -1178,6 +1179,7 @@ public:
   fhandler_pipe ();
 
   bool ispipe() const { return true; }
+  void set_read_mutex (HANDLE mtx) { read_mtx = mtx; }
 
   void set_popen_pid (pid_t pid) {popen_pid = pid;}
   pid_t get_popen_pid () const {return popen_pid;}
@@ -1187,7 +1189,9 @@ public:
   select_record *select_except (select_stuff *);
   char *get_proc_fd_name (char *buf);
   int open (int flags, mode_t mode = 0);
+  void fixup_after_fork (HANDLE);
   int dup (fhandler_base *child, int);
+  int close ();
   void __reg3 raw_read (void *ptr, size_t& len);
   ssize_t __reg3 raw_write (const void *ptr, size_t len);
   int ioctl (unsigned int cmd, void *);

winsup/cygwin/fhandler_pipe.cc

@@ -240,8 +240,37 @@ fhandler_pipe::raw_read (void *ptr, size_t& len)
       keep_looping = false;
       if (evt)
 	ResetEvent (evt);
+      if (!is_nonblocking ())
+	{
+	  FILE_PIPE_LOCAL_INFORMATION fpli;
+	  ULONG reader_count;
+	  ULONG max_len = 64;
+
+	  WaitForSingleObject (read_mtx, INFINITE);
+
+	  /* Make sure never to request more bytes than half the pipe
+	     buffer size.  Every pending read lowers WriteQuotaAvailable
+	     on the write side and thus affects select's ability to return
+	     more or less reliable info whether a write succeeds or not.
+
+	     Let the size of the request depend on the number of readers
+	     at the time. */
+	  status = NtQueryInformationFile (get_handle (), &io,
+					   &fpli, sizeof (fpli),
+					   FilePipeLocalInformation);
+	  if (NT_SUCCESS (status) && fpli.ReadDataAvailable == 0)
+	    {
+	      reader_count = get_obj_handle_count (get_handle ());
+	      if (reader_count < 10)
+		max_len = fpli.InboundQuota / (2 * reader_count);
+	      if (len > max_len)
+		len = max_len;
+	    }
+	}
       status = NtReadFile (get_handle (), evt, NULL, NULL, &io, ptr,
 			   len, NULL, NULL);
+      if (!is_nonblocking ())
+	ReleaseMutex (read_mtx);
       if (evt && status == STATUS_PENDING)
 	{
 	  waitret = cygwait (evt);
@@ -426,6 +455,13 @@ fhandler_pipe::raw_write (const void *ptr, size_t len)
     pthread::static_cancel_self ();
   return nbytes ?: -1;
 }
+
+void
+fhandler_pipe::fixup_after_fork (HANDLE parent)
+{
+  if (read_mtx)
+    fork_fixup (parent, read_mtx, "read_mtx");
+  fhandler_base::fixup_after_fork (parent);
 }
 
 int
@@ -434,16 +470,31 @@ fhandler_pipe::dup (fhandler_base *child, int flags)
   fhandler_pipe *ftp = (fhandler_pipe *) child;
   ftp->set_popen_pid (0);
 
-  int res;
-  if (get_handle () && fhandler_base::dup (child, flags))
+  int res = 0;
+  if (fhandler_base::dup (child, flags))
     res = -1;
-  else
-    res = 0;
+  else if (read_mtx &&
+	   !DuplicateHandle (GetCurrentProcess (), read_mtx,
+			     GetCurrentProcess (), &ftp->read_mtx,
+			     0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+    {
+      __seterrno ();
+      ftp->close ();
+      res = -1;
+    }
 
   debug_printf ("res %d", res);
   return res;
 }
 
+int
+fhandler_pipe::close ()
+{
+  if (read_mtx)
+    CloseHandle (read_mtx);
+  return fhandler_base::close ();
+}
+
 #define PIPE_INTRO "\\\\.\\pipe\\cygwin-"
 
 /* Create a pipe, and return handles to the read and write ends,
@@ -641,7 +692,25 @@ fhandler_pipe::create (fhandler_pipe *fhs[2], unsigned psize, int mode)
 		    unique_id);
       fhs[1]->init (w, FILE_CREATE_PIPE_INSTANCE | GENERIC_WRITE, mode,
 		    unique_id);
-      res = 0;
+      /* For the read side of the pipe, add a mutex.  See raw_read for the
+	 usage. */
+      SECURITY_ATTRIBUTES sa = { .nLength = sizeof (SECURITY_ATTRIBUTES),
+				 .lpSecurityDescriptor = NULL,
+				 .bInheritHandle = !(mode & O_CLOEXEC)
+			       };
+      HANDLE mtx = CreateMutexW (&sa, FALSE, NULL);
+      if (!mtx)
+	{
+	  delete fhs[0];
+	  NtClose (r);
+	  delete fhs[1];
+	  NtClose (w);
+	}
+      else
+	{
+	  fhs[0]->set_read_mutex (mtx);
+	  res = 0;
+	}
     }
 
   debug_printf ("%R = pipe([%p, %p], %d, %y)", res, fhs[0], fhs[1], psize, mode);

winsup/cygwin/select.cc

@@ -612,7 +612,6 @@ pipe_data_available (int fd, fhandler_base *fh, HANDLE h, bool writing)
 	   that.  This means that a pipe could still block since you could
 	   be trying to write more to the pipe than is available in the
 	   buffer but that is the hazard of select().  */
-      fpli.WriteQuotaAvailable = fpli.OutboundQuota - fpli.ReadDataAvailable;
       if (fpli.WriteQuotaAvailable > 0)
 	{
 	  paranoid_printf ("fd %d, %s, write: size %u, avail %u", fd,