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

[Ken Brown <kbrown@cornell.edu>]

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

On 9/3/2021 3:00 PM, Ken Brown wrote:
> On 9/3/2021 5:12 AM, Corinna Vinschen wrote:
>> I pushed my stuff to the topic/pipe branch split into hopefully useful
>> chunks.  Kick me if anything is wrong or not working.
> 
> Some of the bugs you fixed in the pipe code exist in the fifo code also.  I 
> started going through them and fixing them, but then I realized that 
> fhandler_pipe::raw_write and fhandler_fifo::raw_write are identical.  For ease 
> of maintenance, I'm thinking we should have a single function, say 
> fhandler_base::raw_write_pipe or fhandler_base::raw_write_pipe_fifo, which is 
> called by both of them.
> 
> WDYT?

Here's what that would look like (attached).

Ken

[-- Attachment #2: 0001-wip.patch --]
[-- Type: text/plain, Size: 13629 bytes --]

From 108a50006decade6dca0b91ca6f1c165bdfd20dd Mon Sep 17 00:00:00 2001
From: Ken Brown <kbrown@cornell.edu>
Date: Fri, 3 Sep 2021 15:43:17 -0400
Subject: [PATCH 1/2] wip

---
 winsup/cygwin/fhandler.cc      | 124 +++++++++++++++++++++++++++++++++
 winsup/cygwin/fhandler.h       |   6 +-
 winsup/cygwin/fhandler_fifo.cc |  98 +-------------------------
 winsup/cygwin/fhandler_pipe.cc | 121 +-------------------------------
 4 files changed, 130 insertions(+), 219 deletions(-)

diff --git a/winsup/cygwin/fhandler.cc b/winsup/cygwin/fhandler.cc
index f0c1b68f1..7d7aad216 100644
--- a/winsup/cygwin/fhandler.cc
+++ b/winsup/cygwin/fhandler.cc
@@ -280,6 +280,129 @@ fhandler_base::raw_write (const void *ptr, size_t len)
   return io.Information;
 }
 
+#define STATUS_PIPE_IS_CLOSED(status)	\
+		({ NTSTATUS _s = (status); \
+		   _s == STATUS_PIPE_CLOSING \
+		   || _s == STATUS_PIPE_BROKEN \
+		   || _s == STATUS_PIPE_EMPTY; })
+
+/* Used by fhandler_pipe::raw_write and fhandler_fifo::raw_write. */
+ssize_t __reg3
+fhandler_base::raw_write_pipe_fifo (const void *ptr, size_t len)
+{
+  size_t nbytes = 0;
+  ULONG chunk;
+  NTSTATUS status = STATUS_SUCCESS;
+  IO_STATUS_BLOCK io;
+  HANDLE evt = NULL;
+
+  if (!len)
+    return 0;
+
+  if (len <= max_atomic_write)
+    chunk = len;
+  else if (is_nonblocking ())
+    chunk = len = max_atomic_write;
+  else
+    chunk = max_atomic_write;
+
+  /* Create a wait event if the pipe or fifo is in blocking mode. */
+  if (!is_nonblocking () && !(evt = CreateEvent (NULL, false, false, NULL)))
+    {
+      __seterrno ();
+      return -1;
+    }
+
+  /* Write in chunks, accumulating a total.  If there's an error, just
+     return the accumulated total unless the first write fails, in
+     which case return -1. */
+  while (nbytes < len)
+    {
+      ULONG_PTR nbytes_now = 0;
+      size_t left = len - nbytes;
+      ULONG len1;
+      DWORD waitret = WAIT_OBJECT_0;
+
+      if (left > chunk)
+	len1 = chunk;
+      else
+	len1 = (ULONG) left;
+      /* 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.
+
+	 POSIX requires
+	 - A write request for {PIPE_BUF} or fewer bytes shall have the
+	   following effect: if there is sufficient space available in the
+	   pipe, write() shall transfer all the data and return the number
+	   of bytes requested. Otherwise, write() shall transfer no data and
+	   return -1 with errno set to [EAGAIN].
+
+	 - A write request for more than {PIPE_BUF} bytes shall cause one
+	   of the following:
+
+	  - When at least one byte can be written, transfer what it can and
+	    return the number of bytes written. When all data previously
+	    written to the pipe is read, it shall transfer at least {PIPE_BUF}
+	    bytes.
+
+	  - When no data can be written, transfer no data, and return -1 with
+	    errno set to [EAGAIN]. */
+      while (len1 > 0)
+	{
+	  status = NtWriteFile (get_handle (), evt, NULL, NULL, &io,
+				(PVOID) ptr, len1, NULL, NULL);
+	  if (evt || !NT_SUCCESS (status) || io.Information > 0
+	      || len <= PIPE_BUF)
+	    break;
+	  len1 >>= 1;
+	}
+      if (evt && status == STATUS_PENDING)
+	{
+	  waitret = cygwait (evt);
+	  if (waitret == WAIT_OBJECT_0)
+	    status = io.Status;
+	}
+      if (waitret == WAIT_CANCELED)
+	status = STATUS_THREAD_CANCELED;
+      else if (waitret == WAIT_SIGNALED)
+	status = STATUS_THREAD_SIGNALED;
+      else if (isclosed ())  /* A signal handler might have closed the fd. */
+	{
+	  if (waitret == WAIT_OBJECT_0)
+	    set_errno (EBADF);
+	  else
+	    __seterrno ();
+	}
+      else if (NT_SUCCESS (status))
+	{
+	  nbytes_now = io.Information;
+	  ptr = ((char *) ptr) + nbytes_now;
+	  nbytes += nbytes_now;
+	  /* 0 bytes returned?  EAGAIN.  See above. */
+	  if (nbytes == 0)
+	    set_errno (EAGAIN);
+	}
+      else if (STATUS_PIPE_IS_CLOSED (status))
+	{
+	  set_errno (EPIPE);
+	  raise (SIGPIPE);
+	}
+      else
+	__seterrno_from_nt_status (status);
+
+      if (nbytes_now == 0)
+	break;
+    }
+  if (evt)
+    CloseHandle (evt);
+  if (status == STATUS_THREAD_SIGNALED && nbytes == 0)
+    set_errno (EINTR);
+  else if (status == STATUS_THREAD_CANCELED)
+    pthread::static_cancel_self ();
+  return nbytes ?: -1;
+}
+
 int
 fhandler_base::get_default_fmode (int flags)
 {
@@ -1464,6 +1587,7 @@ fhandler_base::fhandler_base () :
   _refcnt (0),
   openflags (0),
   unique_id (0),
+  max_atomic_write (DEFAULT_PIPEBUFSIZE),
   archetype (NULL),
   usecount (0)
 {
diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 032ab5fb0..ebddcca88 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -218,6 +218,9 @@ class fhandler_base
 
   HANDLE read_state;
 
+  /* Used by fhandler_pipe and fhandler_fifo. */
+  size_t max_atomic_write;
+
  public:
   LONG inc_refcnt () {return InterlockedIncrement (&_refcnt);}
   LONG dec_refcnt () {return InterlockedDecrement (&_refcnt);}
@@ -453,6 +456,7 @@ public:
 
   virtual void __reg3 raw_read (void *ptr, size_t& ulen);
   virtual ssize_t __reg3 raw_write (const void *ptr, size_t ulen);
+  ssize_t __reg3 raw_write_pipe_fifo (const void *ptr, size_t ulen);
 
   /* Virtual accessor functions to hide the fact
      that some fd's have two handles. */
@@ -1173,7 +1177,6 @@ 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);
 public:
   fhandler_pipe ();
@@ -1342,7 +1345,6 @@ class fhandler_fifo: public fhandler_base
   int nhandlers;                       /* Number of elements in the array. */
   af_unix_spinlock_t _fifo_client_lock;
   bool reader, writer, duplexer;
-  size_t max_atomic_write;
   fifo_reader_id_t me;
 
   HANDLE shmem_handle;
diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index b55ba95e7..11b06ed08 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -108,14 +108,6 @@
 */
 
 
-/* This is only to be used for writers.  When reading,
-STATUS_PIPE_EMPTY simply means there's no data to be read. */
-#define STATUS_PIPE_IS_CLOSED(status)	\
-		({ NTSTATUS _s = (status); \
-		   _s == STATUS_PIPE_CLOSING \
-		   || _s == STATUS_PIPE_BROKEN \
-		   || _s == STATUS_PIPE_EMPTY; })
-
 #define STATUS_PIPE_NO_INSTANCE_AVAILABLE(status)	\
 		({ NTSTATUS _s = (status); \
 		   _s == STATUS_INSTANCE_NOT_AVAILABLE \
@@ -134,7 +126,6 @@ fhandler_fifo::fhandler_fifo ():
   cancel_evt (NULL), thr_sync_evt (NULL), pipe_name_buf (NULL),
   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),
   shared_fc_hdl (NULL), shared_fc_handler (NULL)
 {
@@ -1141,94 +1132,7 @@ fhandler_fifo::wait (HANDLE h)
 ssize_t __reg3
 fhandler_fifo::raw_write (const void *ptr, size_t len)
 {
-  ssize_t ret = -1;
-  size_t nbytes = 0;
-  ULONG chunk;
-  NTSTATUS status = STATUS_SUCCESS;
-  IO_STATUS_BLOCK io;
-  HANDLE evt = NULL;
-
-  if (!len)
-    return 0;
-
-  if (len <= max_atomic_write)
-    chunk = len;
-  else if (is_nonblocking ())
-    chunk = len = max_atomic_write;
-  else
-    chunk = max_atomic_write;
-
-  /* Create a wait event if the FIFO is in blocking mode. */
-  if (!is_nonblocking () && !(evt = CreateEvent (NULL, false, false, NULL)))
-    {
-      __seterrno ();
-      return -1;
-    }
-
-  /* Write in chunks, accumulating a total.  If there's an error, just
-     return the accumulated total unless the first write fails, in
-     which case return -1. */
-  while (nbytes < len)
-    {
-      ULONG_PTR nbytes_now = 0;
-      size_t left = len - nbytes;
-      ULONG len1;
-      DWORD waitret = WAIT_OBJECT_0;
-
-      if (left > chunk)
-	len1 = chunk;
-      else
-	len1 = (ULONG) left;
-      nbytes_now = 0;
-      status = NtWriteFile (get_handle (), evt, NULL, NULL, &io,
-			    (PVOID) ptr, len1, NULL, NULL);
-      if (evt && status == STATUS_PENDING)
-	{
-	  waitret = cygwait (evt);
-	  if (waitret == WAIT_OBJECT_0)
-	    status = io.Status;
-	}
-      if (waitret == WAIT_CANCELED)
-	status = STATUS_THREAD_CANCELED;
-      else if (waitret == WAIT_SIGNALED)
-	status = STATUS_THREAD_SIGNALED;
-      else if (isclosed ())  /* A signal handler might have closed the fd. */
-	{
-	  if (waitret == WAIT_OBJECT_0)
-	    set_errno (EBADF);
-	  else
-	    __seterrno ();
-	}
-      else if (NT_SUCCESS (status))
-	{
-	  nbytes_now = io.Information;
-	  /* 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)
-	    set_errno (EAGAIN);
-	  ptr = ((char *) ptr) + chunk;
-	  nbytes += nbytes_now;
-	}
-      else if (STATUS_PIPE_IS_CLOSED (status))
-	{
-	  set_errno (EPIPE);
-	  raise (SIGPIPE);
-	}
-      else
-	__seterrno_from_nt_status (status);
-      if (nbytes_now == 0)
-	len = 0;		/* Terminate loop. */
-      if (nbytes > 0)
-	ret = nbytes;
-    }
-  if (evt)
-    NtClose (evt);
-  if (status == STATUS_THREAD_SIGNALED && ret < 0)
-    set_errno (EINTR);
-  else if (status == STATUS_THREAD_CANCELED)
-    pthread::static_cancel_self ();
-  return ret;
+  return fhandler_base::raw_write_pipe_fifo (ptr, len);
 }
 
 /* Called from raw_read and select.cc:peek_fifo. */
diff --git a/winsup/cygwin/fhandler_pipe.cc b/winsup/cygwin/fhandler_pipe.cc
index 81b1aed5e..88c98d41b 100644
--- a/winsup/cygwin/fhandler_pipe.cc
+++ b/winsup/cygwin/fhandler_pipe.cc
@@ -20,18 +20,9 @@ details. */
 #include "pinfo.h"
 #include "shared_info.h"
 
-/* This is only to be used for writing.  When reading,
-STATUS_PIPE_EMPTY simply means there's no data to be read. */
-#define STATUS_PIPE_IS_CLOSED(status)	\
-		({ NTSTATUS _s = (status); \
-		   _s == STATUS_PIPE_CLOSING \
-		   || _s == STATUS_PIPE_BROKEN \
-		   || _s == STATUS_PIPE_EMPTY; })
-
 fhandler_pipe::fhandler_pipe ()
   : fhandler_base (), popen_pid (0)
 {
-  max_atomic_write = DEFAULT_PIPEBUFSIZE;
   need_fork_fixup (true);
 }
 
@@ -342,117 +333,7 @@ fhandler_pipe::raw_read (void *ptr, size_t& len)
 ssize_t __reg3
 fhandler_pipe::raw_write (const void *ptr, size_t len)
 {
-  size_t nbytes = 0;
-  ULONG chunk;
-  NTSTATUS status = STATUS_SUCCESS;
-  IO_STATUS_BLOCK io;
-  HANDLE evt = NULL;
-
-  if (!len)
-    return 0;
-
-  if (len <= max_atomic_write)
-    chunk = len;
-  else if (is_nonblocking ())
-    chunk = len = max_atomic_write;
-  else
-    chunk = max_atomic_write;
-
-  /* Create a wait event if the pipe is in blocking mode. */
-  if (!is_nonblocking () && !(evt = CreateEvent (NULL, false, false, NULL)))
-    {
-      __seterrno ();
-      return -1;
-    }
-
-  /* Write in chunks, accumulating a total.  If there's an error, just
-     return the accumulated total unless the first write fails, in
-     which case return -1. */
-  while (nbytes < len)
-    {
-      ULONG_PTR nbytes_now = 0;
-      size_t left = len - nbytes;
-      ULONG len1;
-      DWORD waitret = WAIT_OBJECT_0;
-
-      if (left > chunk)
-	len1 = chunk;
-      else
-	len1 = (ULONG) left;
-      /* 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.
-
-	 POSIX requires
-	 - A write request for {PIPE_BUF} or fewer bytes shall have the
-	   following effect: if there is sufficient space available in the
-	   pipe, write() shall transfer all the data and return the number
-	   of bytes requested. Otherwise, write() shall transfer no data and
-	   return -1 with errno set to [EAGAIN].
-
-	 - A write request for more than {PIPE_BUF} bytes shall cause one
-	   of the following:
-
-	  - When at least one byte can be written, transfer what it can and
-	    return the number of bytes written. When all data previously
-	    written to the pipe is read, it shall transfer at least {PIPE_BUF}
-	    bytes.
-
-	  - When no data can be written, transfer no data, and return -1 with
-	    errno set to [EAGAIN]. */
-      while (len1 > 0)
-	{
-	  status = NtWriteFile (get_handle (), evt, NULL, NULL, &io,
-				(PVOID) ptr, len1, NULL, NULL);
-	  if (evt || !NT_SUCCESS (status) || io.Information > 0
-	      || len <= PIPE_BUF)
-	    break;
-	  len1 >>= 1;
-	}
-      if (evt && status == STATUS_PENDING)
-	{
-	  waitret = cygwait (evt);
-	  if (waitret == WAIT_OBJECT_0)
-	    status = io.Status;
-	}
-      if (waitret == WAIT_CANCELED)
-	status = STATUS_THREAD_CANCELED;
-      else if (waitret == WAIT_SIGNALED)
-	status = STATUS_THREAD_SIGNALED;
-      else if (isclosed ())  /* A signal handler might have closed the fd. */
-	{
-	  if (waitret == WAIT_OBJECT_0)
-	    set_errno (EBADF);
-	  else
-	    __seterrno ();
-	}
-      else if (NT_SUCCESS (status))
-	{
-	  nbytes_now = io.Information;
-	  ptr = ((char *) ptr) + nbytes_now;
-	  nbytes += nbytes_now;
-	  /* 0 bytes returned?  EAGAIN.  See above. */
-	  if (nbytes == 0)
-	    set_errno (EAGAIN);
-	}
-      else if (STATUS_PIPE_IS_CLOSED (status))
-	{
-	  set_errno (EPIPE);
-	  raise (SIGPIPE);
-	}
-      else
-	__seterrno_from_nt_status (status);
-
-      if (nbytes_now == 0)
-	break;
-    }
-  if (evt)
-    CloseHandle (evt);
-  if (status == STATUS_THREAD_SIGNALED && nbytes == 0)
-    set_errno (EINTR);
-  else if (status == STATUS_THREAD_CANCELED)
-    pthread::static_cancel_self ();
-  return nbytes ?: -1;
+  return fhandler_base::raw_write_pipe_fifo (ptr, len);
 }
 
 void
-- 
2.33.0