Re: cygrunsrv + sshd + rsync = 20 times too slow -- throttled?

[Corinna Vinschen <corinna-cygwin@cygwin.com>]

[-- Attachment #1: Type: text/plain, Size: 1933 bytes --]

On Aug 31 17:27, Corinna Vinschen wrote:
> On Aug 31 17:18, Corinna Vinschen wrote:
> > On Aug 31 08:33, Ken Brown wrote:
> > > On 8/31/2021 7:45 AM, Takashi Yano wrote:
> > > > On Tue, 31 Aug 2021 12:18:57 +0200
> > > > Corinna Vinschen wrote:
> > > > > Please try the attached patch on top of topic/pipe.
> > > > 
> > > > Thanks for the new patch. I have confirmed that above issue
> > > > is fixed and select() for write pipe seems to work as expected.
> > > > 
> > > > 
> > > > BTW, I found one minor difference between Linux and this pipe
> > > > implementation.
> > > > [...]
> > > > Is this difficult to be fixed?
> > > Two other remarks:
> > > 
> > > 1. I think query_hdl needs to be initialized in the fhandler_pipe constructor.
> > 
> > No, that's not necessary.  The fhandlers are always ccalloc'ed so they
> > are all 0 anyway.
> > 
> > > 2. When the read side of the pipe is non-blocking, there can be no pending
> > > reads, so shouldn't we be able to use WriteQuotaAvailable reliably on the
> > > write side?  (I can't test this at the moment.)
> > 
> > In theory, yes, but is it a safe bet that non-blocking reads won't change
> > WriteQuotaAvailable on the write side, at least for a very short time?
> > The question is, of course, if that really makes much of a difference.
> 
> Oh, btw... why do you want to use WriteQuotaAvailable for normal
> pipes, even though the read side information is available anyway?
> 
> We can do that for fifos, no problem, but it doesn't make much sense
> to differ between blocking and non-blocking pipes, the code flow is the
> same.

So for the time being I suggest the below patch on top of topic/pipe.
It contains everything we discussed so far.

One question left is, do we want to switch to FILE_PIPE_BYTE_STREAM_TYPE
entirely for pipes?  I don't see that it's still necessary to use
FILE_PIPE_MESSAGE_TYPE for pipes.  Everything seems to work normally
with byte-type pipes.

[-- Attachment #2: pipe.diff --]
[-- Type: text/plain, Size: 10392 bytes --]

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 132e6002133b..1f0f28077a7c 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1171,6 +1171,7 @@ class fhandler_socket_unix : public fhandler_socket
 class fhandler_pipe: public fhandler_base
 {
 private:
+  HANDLE query_hdl;
   pid_t popen_pid;
   size_t max_atomic_write;
   void set_pipe_non_blocking (bool nonblocking);
@@ -1179,6 +1180,8 @@ public:
 
   bool ispipe() const { return true; }
 
+  HANDLE get_query_handle () const { return query_hdl; }
+
   void set_popen_pid (pid_t pid) {popen_pid = pid;}
   pid_t get_popen_pid () const {return popen_pid;}
   off_t lseek (off_t offset, int whence);
@@ -1187,7 +1190,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 *);
diff --git a/winsup/cygwin/fhandler_pipe.cc b/winsup/cygwin/fhandler_pipe.cc
index 2dec0a84817c..2d9e87bb3450 100644
--- a/winsup/cygwin/fhandler_pipe.cc
+++ b/winsup/cygwin/fhandler_pipe.cc
@@ -352,8 +352,30 @@ fhandler_pipe::raw_write (const void *ptr, size_t len)
       else
 	len1 = (ULONG) left;
       nbytes_now = 0;
-      status = NtWriteFile (get_handle (), evt, NULL, NULL, &io,
-			    (PVOID) ptr, len1, NULL, NULL);
+      while (true)
+	{
+	  status = NtWriteFile (get_handle (), evt, NULL, NULL, &io,
+				(PVOID) ptr, len1, NULL, NULL);
+	  if (evt || !NT_SUCCESS (status) || io.Information > 0)
+	    break;
+
+	  FILE_PIPE_LOCAL_INFORMATION fpli;
+	  IO_STATUS_BLOCK qio;
+
+	  if (!NT_SUCCESS (NtQueryInformationFile (query_hdl, &qio, &fpli,
+			   sizeof (fpli), FilePipeLocalInformation)))
+	    len1 >>= 1;
+	  else
+	    {
+	      fpli.WriteQuotaAvailable = fpli.InboundQuota
+					 - fpli.ReadDataAvailable;
+	      if (len1 > fpli.WriteQuotaAvailable
+		  && fpli.WriteQuotaAvailable > 0)
+		len1 = fpli.InboundQuota - fpli.ReadDataAvailable;
+	      else
+		break;
+	    }
+	}
       if (evt && status == STATUS_PENDING)
 	{
 	  waitret = cygwait (evt);
@@ -378,7 +400,7 @@ fhandler_pipe::raw_write (const void *ptr, size_t len)
 	  /* NtWriteFile returns success with # of bytes written == 0
 	     if writing on a non-blocking pipe fails because the pipe
 	     buffer doesn't have sufficient space. */
-	  if (nbytes_now == 0)
+	  if (nbytes_now == 0 && nbytes == 0)
 	    set_errno (EAGAIN);
 	  ptr = ((char *) ptr) + chunk;
 	  nbytes += nbytes_now;
@@ -405,22 +427,45 @@ fhandler_pipe::raw_write (const void *ptr, size_t len)
   return ret;
 }
 
+void
+fhandler_pipe::fixup_after_fork (HANDLE parent)
+{
+  if (query_hdl)
+    fork_fixup (parent, query_hdl, "query_hdl");
+  fhandler_base::fixup_after_fork (parent);
+}
+
 int
 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 (query_hdl &&
+	   !DuplicateHandle (GetCurrentProcess (), query_hdl,
+			     GetCurrentProcess (), &ftp->query_hdl,
+			     0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+    {
+      __seterrno ();
+      ftp->close ();
+      res = -1;
+    }
 
   debug_printf ("res %d", res);
   return res;
 }
 
+int
+fhandler_pipe::close ()
+{
+  if (query_hdl)
+    NtClose (query_hdl);
+  return fhandler_base::close ();
+}
+
 #define PIPE_INTRO "\\\\.\\pipe\\cygwin-"
 
 /* Create a pipe, and return handles to the read and write ends,
@@ -608,6 +653,7 @@ fhandler_pipe::create (fhandler_pipe *fhs[2], unsigned psize, int mode)
   else if ((fhs[1] = (fhandler_pipe *) build_fh_dev (*pipew_dev)) == NULL)
     {
       delete fhs[0];
+      CloseHandle (r);
       CloseHandle (w);
     }
   else
@@ -617,10 +663,23 @@ 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 write side of the pipe, duplicate the handle to the read side
+	 into query_hdl just for calling NtQueryInformationFile.  See longish
+	 comment in select.cc, pipe_data_available() for the reasoning. */
+      if (!DuplicateHandle (GetCurrentProcess (), r, GetCurrentProcess (),
+			    &fhs[1]->query_hdl, GENERIC_READ,
+			    !(mode & O_CLOEXEC), 0))
+	{
+	  delete fhs[0];
+	  CloseHandle (r);
+	  delete fhs[1];
+	  CloseHandle (w);
+	}
+      else
+	res = 0;
     }
 
-  debug_printf ("%R = pipe([%p, %p], %d, %y)", res, fhs[0], fhs[1], psize, mode);
+  debug_printf ("%R = pipe(%d, %y)", res, psize, mode);
   return res;
 }
 
@@ -658,10 +717,10 @@ nt_create (LPSECURITY_ATTRIBUTES sa_ptr, PHANDLE r, PHANDLE w,
 				 &cygheap->installation_key,
 				 GetCurrentProcessId ());
 
-  access = GENERIC_READ | FILE_WRITE_ATTRIBUTES;
+  access = GENERIC_READ | FILE_WRITE_ATTRIBUTES | SYNCHRONIZE;
 
   ULONG pipe_type = pipe_byte ? FILE_PIPE_BYTE_STREAM_TYPE
-    : FILE_PIPE_MESSAGE_TYPE;
+			      : FILE_PIPE_MESSAGE_TYPE;
 
   /* Retry NtCreateNamedPipeFile as long as the pipe name is in use.
      Retrying will probably never be necessary, but we want
@@ -737,7 +796,7 @@ nt_create (LPSECURITY_ATTRIBUTES sa_ptr, PHANDLE r, PHANDLE w,
     {
       debug_printf ("NtOpenFile: name %S", &pipename);
 
-      access = GENERIC_WRITE | FILE_READ_ATTRIBUTES;
+      access = GENERIC_WRITE | FILE_READ_ATTRIBUTES | SYNCHRONIZE;
       status = NtOpenFile (w, access, &attr, &io, 0, 0);
       if (!NT_SUCCESS (status))
 	{
diff --git a/winsup/cygwin/select.cc b/winsup/cygwin/select.cc
index 83e1c00e0ac7..dc1f7961351b 100644
--- a/winsup/cygwin/select.cc
+++ b/winsup/cygwin/select.cc
@@ -585,7 +585,8 @@ no_verify (select_record *, fd_set *, fd_set *, fd_set *)
 }
 
 static int
-pipe_data_available (int fd, fhandler_base *fh, HANDLE h, bool writing)
+pipe_data_available (int fd, fhandler_base *fh, HANDLE h, bool writing,
+		     bool use_readside)
 {
   IO_STATUS_BLOCK iosb = {{0}, 0};
   FILE_PIPE_LOCAL_INFORMATION fpli = {0};
@@ -608,15 +609,34 @@ pipe_data_available (int fd, fhandler_base *fh, HANDLE h, bool writing)
     }
   if (writing)
     {
-	/* If there is anything available in the pipe buffer then signal
-	   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 there is anything available in the pipe buffer then signal
+	 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().
+
+	 Note that 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.
+
+	 Consequentially, the only reliable information is available on the
+	 read side, so fetch info from the read side via the pipe-specific
+	 query handle.  Use fpli.WriteQuotaAvailable as storage for the actual
+	 interesting value, which is the InboundQuote on the read side,
+	 decremented by the number of bytes of data in that buffer. */
+      if (use_readside)
+	fpli.WriteQuotaAvailable = fpli.InboundQuota - fpli.ReadDataAvailable;
       if (fpli.WriteQuotaAvailable > 0)
 	{
 	  paranoid_printf ("fd %d, %s, write: size %u, avail %u", fd,
-			   fh->get_name (), fpli.OutboundQuota,
+			   fh->get_name (), fpli.InboundQuota,
 			   fpli.WriteQuotaAvailable);
 	  return 1;
 	}
@@ -684,10 +704,11 @@ peek_pipe (select_record *s, bool from_select)
 	  gotone = s->read_ready = true;
 	  goto out;
 	}
-      int n = pipe_data_available (s->fd, fh, h, false);
+      int n = pipe_data_available (s->fd, fh, h, false, false);
       /* On PTY masters, check if input from the echo pipe is available. */
       if (n == 0 && fh->get_echo_handle ())
-	n = pipe_data_available (s->fd, fh, fh->get_echo_handle (), false);
+	n = pipe_data_available (s->fd, fh, fh->get_echo_handle (), false,
+				 false);
 
       if (n < 0)
 	{
@@ -718,10 +739,16 @@ out:
       fhandler_pty_master *fhm = (fhandler_pty_master *) fh;
       fhm->set_mask_flusho (s->read_ready);
     }
-  h = fh->get_output_handle ();
   if (s->write_selected && dev != FH_PIPER)
     {
-      gotone += s->write_ready =  pipe_data_available (s->fd, fh, h, true);
+      /* For the write side of a pipe, fetch the handle to the read side.
+	 See the longish comment in pipe_data_available for the reasoning. */
+      if (dev == FH_PIPEW)
+	h = ((fhandler_pipe *) fh)->get_query_handle ();
+      else
+	h = fh->get_output_handle ();
+      gotone += s->write_ready = pipe_data_available (s->fd, fh, h, true,
+						      dev == FH_PIPEW);
       select_printf ("write: %s, gotone %d", fh->get_name (), gotone);
     }
   return gotone;
@@ -922,7 +949,7 @@ out:
   if (s->write_selected)
     {
       gotone += s->write_ready
-	= pipe_data_available (s->fd, fh, fh->get_handle (), true);
+	= pipe_data_available (s->fd, fh, fh->get_handle (), true, false);
       select_printf ("write: %s, gotone %d", fh->get_name (), gotone);
     }
   return gotone;
@@ -1368,7 +1395,8 @@ out:
   HANDLE h = ptys->get_output_handle ();
   if (s->write_selected)
     {
-      gotone += s->write_ready =  pipe_data_available (s->fd, fh, h, true);
+      gotone += s->write_ready =  pipe_data_available (s->fd, fh, h, true,
+						       false);
       select_printf ("write: %s, gotone %d", fh->get_name (), gotone);
     }
   return gotone;