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

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

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

Hi Ken,

On Tue, 7 Sep 2021 12:26:31 +0900
Takashi Yano wrote:
> On Fri, 27 Aug 2021 20:24:40 +0900
> Takashi Yano wrote:
> > Hi Ken,
> > 
> > Thanks much! I tested topic/pipe branch.
> > 
> > [yano@cygwin-PC ~]$ scp test.dat yano@linux-server:.
> > yano@linux-server's password:
> > test.dat                                      100%  100MB  95.9MB/s   00:01
> > [yano@cygwin-PC ~]$ scp yano@linux-server:test.dat .
> > yano@linux-server's password:
> > test.dat                                      100%  100MB   8.0MB/s   00:12
> > 
> > yano@linux-server:~$ scp yano@cygwin-PC:test.dat .
> > yano@cygwin-PC's password:
> > test.dat                                      100%  100MB 109.7MB/s   00:00
> > yano@linux-server:~$ scp test.dat yano@cygwin-PC:.
> > yano@cygwin-PC's password:
> > test.dat                                      100%  100MB  31.4MB/s   00:03
> > 
> > As shown above, outgoing transfer-rate has been improved upto near
> > theoretical limit. However, incoming transfer-rate is not improved
> > much.
> > 
> > I digged further and found the first patch attached solves the issue
> > as follows.
> > 
> > [yano@cygwin-PC ~]$ scp yano@linux-server:test.dat .
> > yano@linux-server's password:
> > test.dat                                      100%  100MB 112.8MB/s   00:00
> > 
> > yano@linux-server2:~$ scp test.dat yano@cygwin-PC:.
> > yano@cygwin-PC's password:
> > test.dat                                      100%  100MB 102.5MB/s   00:00
> 
> With this patch (2e36ae2e), I found a problem that mintty gets into
> high load if several keys are typed quickly.
> 
> Therefore, I would like to propose a patch attached.

I found the fifo has same issue as pipe that throughput is
slowing down occasionally.

Please try simple test case attached (fifo_test.c).
Argument of fifo_test means:
0: with select, blocking I/O
1: with select, non-blocking I/O
2: without select, blocking I/O
3: without select, non-blocking I/O

In my environment, the results are as follows.

[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 0

Total: 100MB in 4.593770 second, 21.768613MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 1

Total: 100MB in 0.564711 second, 177.081603MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 2

Total: 100MB in 0.514730 second, 194.276724MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 3
wwwwwwwwwwwwwwwwwwwwwwwwwwwwwrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr
Total: 100MB in 0.038880 second, 2572.003230MB/s

Therefore, I would like to propose patch attached, which utilizes
select_sem just as pipe.

With the patch attached, the throughput of the fifo is improved
much as follows.

[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 0

Total: 100MB in 0.076400 second, 1308.895384MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 1

Total: 100MB in 0.064198 second, 1557.683351MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 2

Total: 100MB in 0.021803 second, 4586.503754MB/s
[yano@Express5800-S70 ~/pipe_test]$ ./fifo_test 3
wwwwwwrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr
Total: 100MB in 0.021747 second, 4598.271969MB/s


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

[-- Attachment #2: fifo_test.c --]
[-- Type: text/x-csrc, Size: 2228 bytes --]

#include <unistd.h>
#include <stdlib.h>
#include <sys/wait.h>
#include <stdio.h>
#include <time.h>
#include <sys/select.h>
#include <fcntl.h>
#include <errno.h>

#define FIFO_NAME "/tmp/fifo1"
#define BLKSIZ 65536
#define NBLK (100*(1<<20)/BLKSIZ)


int main(int argc, char *argv[])
{
	pid_t pid;
	int nonblock = 0;
	int skip_select = 0;

	if (argc >= 2 && (atoi(argv[1]) & 1)) nonblock = 1;
	if (argc >= 2 && (atoi(argv[1]) & 2)) skip_select = 1;

	mkfifo("/tmp/fifo1", 0644);

	if (!(pid = fork ())) {
		int fifo;
		char buf[BLKSIZ] = {0,};
		int i;
		fifo = open(FIFO_NAME, O_WRONLY);
		if (nonblock) {
			int flags;
			flags = fcntl(fifo, F_GETFL);
			flags |= O_NONBLOCK;
			fcntl(fifo, F_SETFL, flags);
		}
		fd_set wfds;
		for (i=0; i<NBLK; i++) {
			int total = 0;
			while (total < sizeof(buf)) {
				FD_ZERO(&wfds);
				FD_SET(fifo, &wfds);
				if (skip_select || select(fifo+1, NULL, &wfds, NULL, NULL) > 0
						&& FD_ISSET(fifo, &wfds)) {
					errno = 0;
					ssize_t len = write(fifo, buf + total, sizeof(buf) - total);
					if (len <= 0 && errno == EAGAIN) printf("w", i);
					if (len > 0) total += len;
				}
			}
		}
		close(fifo);
	} else {
		int fifo;
		char buf[BLKSIZ] = {0,};
		int total = 0;
		fd_set rfds;
		struct timespec tv0, tv1;
		double elasped;
		fifo = open(FIFO_NAME, O_RDONLY);
		if (nonblock) {
			int flags;
			flags = fcntl(fifo, F_GETFL);
			flags |= O_NONBLOCK;
			fcntl(fifo, F_SETFL, flags);
		}
		clock_gettime(CLOCK_MONOTONIC, &tv0);
		for (;;) {
			FD_ZERO(&rfds);
			FD_SET(fifo, &rfds);
			if (skip_select || select(fifo+1, &rfds, NULL, NULL, NULL) > 0
					&& FD_ISSET(fifo, &rfds)) {
				errno = 0;
				ssize_t len = read(fifo, buf, sizeof(buf));
				if (len <= 0 && errno == EAGAIN) printf("r");
				else if(len <= 0) break;
				else total += len;
			}
		}
		clock_gettime(CLOCK_MONOTONIC, &tv1);
		close(fifo);
		elasped = (tv1.tv_sec - tv0.tv_sec) + (tv1.tv_nsec - tv0.tv_nsec)*1e-9;
		printf("\nTotal: %dMB in %f second, %fMB/s\n",
			total/(1<<20), elasped, total/(1<<20)/elasped);
		if (total != NBLK*BLKSIZ) {
			printf("Error: %d/%d (%f)\n",
				total, NBLK*BLKSIZ, (double)(total-NBLK*BLKSIZ)/BLKSIZ);
		}
		waitpid(pid, NULL, 0);
		unlink(FIFO_NAME);
	}

	return 0;
}


[-- Attachment #3: 0001-Cygwin-fifo-Utilize-select_sem-for-fifo-as-well-as-p.patch --]
[-- Type: application/octet-stream, Size: 4455 bytes --]

From beeae61f97145d795c4479b0d46cd3b75d6256c4 Mon Sep 17 00:00:00 2001
From: Takashi Yano <takashi.yano@nifty.ne.jp>
Date: Thu, 9 Sep 2021 11:53:11 +0900
Subject: [PATCH] Cygwin: fifo: Utilize select_sem for fifo as well as pipe.

---
 winsup/cygwin/fhandler_fifo.cc | 37 +++++++++++++++++++++++++++++++++-
 winsup/cygwin/select.cc        | 10 +++++++--
 2 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/winsup/cygwin/fhandler_fifo.cc b/winsup/cygwin/fhandler_fifo.cc
index 6709fb974..c40573783 100644
--- a/winsup/cygwin/fhandler_fifo.cc
+++ b/winsup/cygwin/fhandler_fifo.cc
@@ -1047,6 +1047,12 @@ writer_shmem:
   ResetEvent (writer_opening);
   nwriters_unlock ();
 success:
+  if (!select_sem)
+    {
+      char name[MAX_PATH];
+      __small_sprintf(name, "semaphore-%W", get_pipe_name ()->Buffer);
+      select_sem = CreateSemaphore (&sec_none, 0, INT32_MAX, name);
+    }
   return 1;
 err_close_reader:
   saved_errno = get_errno ();
@@ -1235,6 +1241,10 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 		  len = io.Information;
 		  fifo_client_unlock ();
 		  reading_unlock ();
+		  if (select_sem)
+		    ReleaseSemaphore (select_sem,
+				      get_obj_handle_count (select_sem),
+				      NULL);
 		  return;
 		}
 	      break;
@@ -1273,6 +1283,10 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 		    fc_handler[i].last_read = true;
 		    fifo_client_unlock ();
 		    reading_unlock ();
+		    if (select_sem)
+		      ReleaseSemaphore (select_sem,
+					get_obj_handle_count (select_sem),
+					NULL);
 		    return;
 		  }
 		break;
@@ -1312,6 +1326,10 @@ fhandler_fifo::raw_read (void *in_ptr, size_t& len)
 		    fc_handler[i].last_read = true;
 		    fifo_client_unlock ();
 		    reading_unlock ();
+		    if (select_sem)
+		      ReleaseSemaphore (select_sem,
+					get_obj_handle_count (select_sem),
+					NULL);
 		    return;
 		  }
 		break;
@@ -1345,7 +1363,7 @@ maybe_retry:
       else
 	{
 	  /* Allow interruption and don't hog the CPU. */
-	  DWORD waitret = cygwait (NULL, 1, cw_cancel | cw_sig_eintr);
+	  DWORD waitret = cygwait (select_sem, 1, cw_cancel | cw_sig_eintr);
 	  if (waitret == WAIT_CANCELED)
 	    pthread::static_cancel_self ();
 	  else if (waitret == WAIT_SIGNALED)
@@ -1569,6 +1587,11 @@ fhandler_fifo::close ()
     NtClose (write_ready);
   if (writer_opening)
     NtClose (writer_opening);
+  if (select_sem)
+    {
+      ReleaseSemaphore (select_sem, get_obj_handle_count (select_sem), NULL);
+      NtClose (select_sem);
+    }
   if (nohandle ())
     return 0;
   else
@@ -1683,7 +1706,17 @@ fhandler_fifo::dup (fhandler_base *child, int flags)
     }
   if (writer)
     inc_nwriters ();
+  if (select_sem &&
+      !DuplicateHandle (GetCurrentProcess (), select_sem,
+			GetCurrentProcess (), &fhf->select_sem,
+			0, !(flags & O_CLOEXEC), DUPLICATE_SAME_ACCESS))
+    {
+      __seterrno ();
+      goto err_close_thr_sync_evt;
+    }
   return 0;
+err_close_thr_sync_evt:
+  NtClose (fhf->thr_sync_evt);
 err_close_cancel_evt:
   NtClose (fhf->cancel_evt);
 err_close_update_needed_evt:
@@ -1743,6 +1776,8 @@ fhandler_fifo::fixup_after_fork (HANDLE parent)
       me.winpid = GetCurrentProcessId ();
       new cygthread (fifo_reader_thread, this, "fifo_reader", thr_sync_evt);
     }
+  if (select_sem)
+    fork_fixup (parent, select_sem, "select_sem");
   if (writer)
     inc_nwriters ();
 }
diff --git a/winsup/cygwin/select.cc b/winsup/cygwin/select.cc
index e9e71b269..5e583434c 100644
--- a/winsup/cygwin/select.cc
+++ b/winsup/cygwin/select.cc
@@ -963,7 +963,13 @@ start_thread_fifo (select_record *me, select_stuff *stuff)
     {
       pi->start = &stuff->start;
       pi->stop_thread = false;
-      pi->bye = CreateEvent (&sec_none_nih, TRUE, FALSE, NULL);
+      pi->bye = me->fh->get_select_sem ();
+      if (pi->bye)
+	DuplicateHandle (GetCurrentProcess (), pi->bye,
+			 GetCurrentProcess (), &pi->bye,
+			 0, 0, DUPLICATE_SAME_ACCESS);
+      else
+	pi->bye = CreateSemaphore (&sec_none_nih, 0, INT32_MAX, NULL);
       pi->thread = new cygthread (thread_fifo, pi, "fifosel");
       me->h = *pi->thread;
       if (!me->h)
@@ -981,7 +987,7 @@ fifo_cleanup (select_record *, select_stuff *stuff)
   if (pi->thread)
     {
       pi->stop_thread = true;
-      SetEvent (pi->bye);
+      ReleaseSemaphore (pi->bye, get_obj_handle_count (pi->bye), NULL);
       pi->thread->detach ();
       CloseHandle (pi->bye);
     }
-- 
2.33.0