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

[Takashi Yano <takashi.yano@nifty.ne.jp>]

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

On Mon, 13 Sep 2021 11:07:42 +0200
Corinna Vinschen wrote:
> On Sep 11 11:35, Takashi Yano wrote:
> > Keeping read handle in write pipe (Corinna's query_hdl) causes problem
> > that write side cannot detect close on read side.
> > Is it possible to open read handle temporally when pipe_data_available()
> > is called?
> 
> 1. You would have to know which process keeps the other side of the pipe.
> 2. You would have to have the permission to open the other process to
>    duplicate the pipe into your own process
> 3. You would have to know the HANDLE value of the read side of your pipe
>    in that other process.
> 
> Point 1 is (kind of) doable using GetNamedPipeClientProcessId or
> GetNamedPipeServerProcessId.  ZIt's not clear how reliable these
> functions are, given that both pipe sides are created by the same
> process and then usually inherited by two child processes communicating
> over that pipe.
> 
> Point 2 is most of the time the case, especially when talking with
> native processes.
> 
> Point 3 requires some sort of IPC.
> 
> Having said that, I think this is too complicated.

In the current implementation, there is a potential risk that
write side fails to detect the closure of read side if more than
one writers exist and one of them is non-cygwin process.

Therefore, I had tried to implement the above idea (tentative
query handle). Finally, I could convince myself to some extent,
so I attach a patch for that.

As for point 1 and 3, I used NtQuerySystemInformation() with
SystemHandleInformation. I also used NtQueryObject() with
ObjectNameInformation in order to know which handle is the
other side of the pipe. As for point 2, it is not possible to
open the process which is running as a service, so the current
query_hdl is used for such process as an exception.

Ken, WDYT of this implementation?

-- 
Takashi Yano <takashi.yano@nifty.ne.jp>

[-- Attachment #2: v2-0001-Cygwin-pipe-Introduce-tentative-query_hdl.patch --]
[-- Type: application/octet-stream, Size: 9164 bytes --]

From 98b0e3f0dba382751cbe85c1a8ae44327c592d4c Mon Sep 17 00:00:00 2001
From: Takashi Yano <takashi.yano@nifty.ne.jp>
Date: Mon, 20 Sep 2021 16:30:20 +0900
Subject: [PATCH v2] Cygwin: pipe: Introduce tentative query_hdl.

- The commit f79a4611 introduced query_hdl which is the read pipe
  handle kept in the write pipe instance in order to determine if
  the pipe is ready to write in select(). This implementation has
  a potential risk that write side fails to detect the closure of
  read side if more than one writers exist and one of them is non-
  cygwin process.

  With this patch, the strategy of commit f79a4611 is used only if
  the process is running as a service. For normal process, instead
  of keeping query_hdl in the write pipe instance, it is retrieved
  tentatively when select() is called. Actually, we want to use
  tentative query_hdl for all processes, however, it does not work
  for service processes due to OpenProcess() failure.
---
 winsup/cygwin/fhandler.h       |   3 +
 winsup/cygwin/fhandler_pipe.cc | 134 +++++++++++++++++++++++++++++++--
 winsup/cygwin/select.cc        |   8 +-
 3 files changed, 139 insertions(+), 6 deletions(-)

diff --git a/winsup/cygwin/fhandler.h b/winsup/cygwin/fhandler.h
index 3471e95b9..d1149d301 100644
--- a/winsup/cygwin/fhandler.h
+++ b/winsup/cygwin/fhandler.h
@@ -1191,6 +1191,8 @@ private:
   pid_t popen_pid;
   HANDLE query_hdl;
   HANDLE hdl_cnt_mtx;
+  HANDLE query_hdl_proc;
+  HANDLE query_hdl_value;
   void release_select_sem (const char *);
 public:
   fhandler_pipe ();
@@ -1249,6 +1251,7 @@ public:
       }
   }
   bool reader_closed ();
+  HANDLE tentative_query_hdl ();
 };
 
 #define CYGWIN_FIFO_PIPE_NAME_LEN     47
diff --git a/winsup/cygwin/fhandler_pipe.cc b/winsup/cygwin/fhandler_pipe.cc
index 6b41a755f..fb2d06bb8 100644
--- a/winsup/cygwin/fhandler_pipe.cc
+++ b/winsup/cygwin/fhandler_pipe.cc
@@ -19,6 +19,7 @@ details. */
 #include "cygheap.h"
 #include "pinfo.h"
 #include "shared_info.h"
+#include "tls_pbuf.h"
 
 /* This is only to be used for writing.  When reading,
 STATUS_PIPE_EMPTY simply means there's no data to be read. */
@@ -220,8 +221,6 @@ fhandler_pipe::open_setup (int flags)
 	  goto err_close_read_mtx;
 	}
     }
-  if (get_dev () == FH_PIPEW && !query_hdl)
-    set_pipe_non_blocking (is_nonblocking ());
   return true;
 
 err_close_read_mtx:
@@ -267,7 +266,7 @@ fhandler_pipe::release_select_sem (const char *from)
       - get_obj_handle_count (read_mtx);
   else /* Number of select() call */
     n_release = get_obj_handle_count (select_sem)
-      - get_obj_handle_count (query_hdl);
+      - get_obj_handle_count (hdl_cnt_mtx);
   debug_printf("%s(%s) release %d", from,
 	       get_dev () == FH_PIPER ? "PIPER" : "PIPEW", n_release);
   if (n_release)
@@ -667,6 +666,8 @@ fhandler_pipe::close ()
   int ret = fhandler_base::close ();
   ReleaseMutex (hdl_cnt_mtx);
   CloseHandle (hdl_cnt_mtx);
+  if (query_hdl_proc)
+    CloseHandle (query_hdl_proc);
   return ret;
 }
 
@@ -820,6 +821,13 @@ fhandler_pipe::create (LPSECURITY_ATTRIBUTES sa_ptr, PHANDLE r, PHANDLE w,
   return 0;
 }
 
+inline static bool
+is_running_as_service (void)
+{
+  return check_token_membership (well_known_service_sid)
+    || cygheap->user.saved_sid () == well_known_system_sid;
+}
+
 /* The next version of fhandler_pipe::create used to call the previous
    version.  But the read handle created by the latter doesn't have
    FILE_WRITE_ATTRIBUTES access unless the pipe is created with
@@ -874,7 +882,8 @@ fhandler_pipe::create (fhandler_pipe *fhs[2], unsigned psize, int mode)
 			0, sa->bInheritHandle, DUPLICATE_SAME_ACCESS))
     goto err_close_select_sem0;
 
-  if (!DuplicateHandle (GetCurrentProcess (), r,
+  if (is_running_as_service () &&
+      !DuplicateHandle (GetCurrentProcess (), r,
 			GetCurrentProcess (), &fhs[1]->query_hdl,
 			FILE_READ_DATA, sa->bInheritHandle, 0))
     goto err_close_select_sem1;
@@ -893,7 +902,8 @@ fhandler_pipe::create (fhandler_pipe *fhs[2], unsigned psize, int mode)
 err_close_hdl_cnt_mtx0:
   CloseHandle (fhs[0]->hdl_cnt_mtx);
 err_close_query_hdl:
-  CloseHandle (fhs[1]->query_hdl);
+  if (fhs[1]->query_hdl)
+    CloseHandle (fhs[1]->query_hdl);
 err_close_select_sem1:
   CloseHandle (fhs[1]->select_sem);
 err_close_select_sem0:
@@ -946,6 +956,7 @@ nt_create (LPSECURITY_ATTRIBUTES sa_ptr, HANDLE &r, HANDLE &w,
 				 GetCurrentProcessId ());
 
   access = GENERIC_READ | FILE_WRITE_ATTRIBUTES | SYNCHRONIZE;
+  access |= FILE_WRITE_EA; /* Add this right as a marker of cygwin read pipe */
 
   ULONG pipe_type = pipe_byte ? FILE_PIPE_BYTE_STREAM_TYPE
     : FILE_PIPE_MESSAGE_TYPE;
@@ -1112,3 +1123,116 @@ fhandler_pipe::fstatvfs (struct statvfs *sfs)
   set_errno (EBADF);
   return -1;
 }
+
+HANDLE
+fhandler_pipe::tentative_query_hdl ()
+{
+  if (get_dev () != FH_PIPEW)
+    return NULL;
+  /* Try process handle opened and pipe handle value cached first
+     in order to reduce overhead. */
+  if (query_hdl_proc && query_hdl_value)
+    {
+      HANDLE h;
+      if (DuplicateHandle (query_hdl_proc, query_hdl_value,
+			   GetCurrentProcess (), &h, FILE_READ_DATA, 0, 0))
+	return h;
+      CloseHandle (query_hdl_proc);
+      query_hdl_proc = NULL;
+      query_hdl_value = NULL;
+    }
+
+  ULONG len;
+  NTSTATUS status;
+  tmp_pathbuf tp;
+  OBJECT_NAME_INFORMATION *ntfn =(OBJECT_NAME_INFORMATION *) tp.w_get ();
+  status = NtQueryObject (get_handle (), ObjectNameInformation, ntfn,
+			  65536, &len);
+  if (!NT_SUCCESS (status))
+    return NULL; /* Non cygwin pipe? */
+  WCHAR name[MAX_PATH];
+  int namelen = min (ntfn->Name.Length / sizeof (WCHAR), MAX_PATH-1);
+  memcpy (name, ntfn->Name.Buffer, namelen * sizeof (WCHAR));
+  name[namelen] = L'\0';
+  uint64_t key;
+  DWORD pid;
+  LONG id;
+  if (swscanf (name, L"\\Device\\NamedPipe\\%llx-%u-pipe-nt-0x%x",
+	       &key, &pid, &id) != 3)
+    return NULL; /* Non cygwin pipe? */
+
+  SIZE_T n_handle = 65536;
+  typedef struct
+    {
+      USHORT UniqueProcessId;
+      USHORT CreatorBackTraceIndex;
+      UCHAR ObjectTypeIndex;
+      UCHAR HandleAttributes;
+      USHORT HandleValue;
+      PVOID Object;
+      ULONG GrantedAccess;
+    } SYSTEM_HANDLE_TABLE_ENTRY_INFO;
+  typedef struct
+    {
+      ULONG NumberOfHandles;
+      SYSTEM_HANDLE_TABLE_ENTRY_INFO Handles[1];
+    } SYSTEM_HANDLE_INFORMATION;
+  SYSTEM_HANDLE_INFORMATION *shi;
+  do
+    {
+      SIZE_T nbytes =
+	sizeof (ULONG) + n_handle * sizeof (SYSTEM_HANDLE_TABLE_ENTRY_INFO);
+      shi = (SYSTEM_HANDLE_INFORMATION *) HeapAlloc (GetProcessHeap (),
+						     0, nbytes);
+      status = NtQuerySystemInformation (SystemHandleInformation,
+					 shi, nbytes, NULL);
+      if (NT_SUCCESS (status))
+	break;
+      HeapFree (GetProcessHeap (), 0, shi);
+      n_handle *= 2;
+    }
+  while (n_handle < (1L<<20));
+  if (!NT_SUCCESS (status))
+    return NULL;
+
+  HANDLE qh = NULL;
+  for (LONG i = (LONG) shi->NumberOfHandles - 1; i >= 0; i--)
+    {
+      /* Check for the peculiarity of cygwin read pipe */
+      DWORD access = FILE_READ_DATA | FILE_READ_EA | FILE_WRITE_EA /* marker */
+	| FILE_READ_ATTRIBUTES | FILE_WRITE_ATTRIBUTES
+	| READ_CONTROL | SYNCHRONIZE;
+      if (shi->Handles[i].GrantedAccess != access)
+	continue;
+
+      /* Retrieve handle */
+      HANDLE proc = OpenProcess (PROCESS_DUP_HANDLE, 0,
+				 shi->Handles[i].UniqueProcessId);
+      if (!proc)
+	continue;
+      HANDLE h = (HANDLE)(intptr_t) shi->Handles[i].HandleValue;
+      BOOL res  = DuplicateHandle (proc, h, GetCurrentProcess (), &h,
+				   FILE_READ_DATA, 0, 0);
+      if (!res)
+	goto close_proc;
+
+      /* Check object name */
+      status = NtQueryObject (h, ObjectNameInformation, ntfn, 65536, &len);
+      if (!NT_SUCCESS (status))
+	goto close_handle;
+      ntfn->Name.Buffer[ntfn->Name.Length / sizeof (WCHAR)] = L'\0';
+      if (wcscmp (name, ntfn->Name.Buffer) == 0)
+	{
+	  query_hdl_proc = proc;
+	  query_hdl_value = (HANDLE)(intptr_t) shi->Handles[i].HandleValue;
+	  qh = h;
+	  break;
+	}
+close_handle:
+      CloseHandle (h);
+close_proc:
+      CloseHandle (proc);
+    }
+  HeapFree (GetProcessHeap (), 0, shi);
+  return qh;
+}
diff --git a/winsup/cygwin/select.cc b/winsup/cygwin/select.cc
index 33c0c0bb0..15589b711 100644
--- a/winsup/cygwin/select.cc
+++ b/winsup/cygwin/select.cc
@@ -641,10 +641,16 @@ pipe_data_available (int fd, fhandler_base *fh, HANDLE h, bool writing)
       if (fh->get_device () == FH_PIPEW && fpli.WriteQuotaAvailable == 0)
 	{
 	  HANDLE query_hdl = ((fhandler_pipe *) fh)->get_query_handle ();
+	  if (!query_hdl)
+	    query_hdl = ((fhandler_pipe *) fh)->tentative_query_hdl ();
 	  if (!query_hdl)
 	    return 1; /* We cannot know actual write pipe space. */
 	  DWORD nbytes_in_pipe;
-	  if (!PeekNamedPipe (query_hdl, NULL, 0, NULL, &nbytes_in_pipe, NULL))
+	  BOOL res =
+	    PeekNamedPipe (query_hdl, NULL, 0, NULL, &nbytes_in_pipe, NULL);
+	  if (!((fhandler_pipe *) fh)->get_query_handle ())
+	    CloseHandle (query_hdl); /* Close tentative query_hdl */
+	  if (!res)
 	    return 1;
 	  fpli.WriteQuotaAvailable = fpli.InboundQuota - nbytes_in_pipe;
 	}
-- 
2.33.0