Re: Unexpectedly low read/write performance of open-pipe

[Mark H Weaver]

Hi Rob,

Rob Browning <address@hidden> writes:
> For a first quick test of your patch using the original program I was
> working on, I see about ~1.4MiB/s without the patch, and about 150MiB/s
> with it, measured by pv.
>
> (If the patch holds up, it'd be nice to have in 2.2, but I suppose that
>  might not be appropriate.)

I've made more improvements, and now your attached test is able to
discard /dev/zero via OPEN_BOTH at ~1267 mb/s on my Thinkpad X200.
That's only a little bit slower than the same test with OPEN_READ, which
discards at ~1462 mb/s on my machine.

  scheme@(guile-user)> (display "discarding dev-zero via OPEN_READ: " 
(current-error-port))
  (time-cat-mb 10000 (cat-zero OPEN_READ) *discard*)
  discarding dev-zero via OPEN_READ: 1461.99 mb/s
  $9 = #t
  scheme@(guile-user)> (display "discarding dev-zero via OPEN_BOTH: " 
(current-error-port))
  (time-cat-mb 10000 (cat-zero OPEN_BOTH) *discard*)
  discarding dev-zero via OPEN_BOTH: 1267.43 mb/s
  $10 = #t

The key improvement over my last version is that I added a new primitive
'get-bytevector-some!', which (1) copies the data into an existing
bytevector, and (2) allows you to specify a maximum read size, which is
of course now necessary.

Crucially, this allows the data to pass through without any heap
allocation.  Previously, every call to the custom 'read!' procedure
allocated a fresh bytevector.  Also, it radically simplifies the 'read!'
procedure:

  (define (read! bv start count)
    (get-bytevector-some! read-port bv start count))

I've attached four patches.  Only the first two should be needed for
what I have described above.

The last two patches will only be of interest if you need suspendable
I/O.  They add suspendable I/O support for custom binary ports, and in
particular for these OPEN_BOTH pipe ports.

Note that I've not yet done much testing of these patches, besides
running the Guile test suite.

       Mark


>From 14dec723707ee766642397962fa93124d9c86811 Mon Sep 17 00:00:00 2001
From: Mark H Weaver <address@hidden>
Date: Tue, 16 Apr 2019 23:14:27 -0400
Subject: [PATCH 1/4] DRAFT: Add get-bytevector-some!.

---
 libguile/r6rs-ports.c         | 37 +++++++++++++++++++++++++++++++++++
 libguile/r6rs-ports.h         |  5 ++++-
 module/ice-9/binary-ports.scm |  1 +
 3 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/libguile/r6rs-ports.c b/libguile/r6rs-ports.c
index c1cbbdf30..9b64696a6 100644
--- a/libguile/r6rs-ports.c
+++ b/libguile/r6rs-ports.c
@@ -502,6 +502,43 @@ SCM_DEFINE (scm_get_bytevector_some, 
"get-bytevector-some", 1, 0, 0,
 }
 #undef FUNC_NAME
 
+SCM_DEFINE (scm_get_bytevector_some_x, "get-bytevector-some!", 4, 0, 0,
+           (SCM port, SCM bv, SCM start, SCM count),
+            "Read up to @var{count} bytes from @var{port}, blocking "
+            "as necessary until at least one byte is available or an "
+            "end-of-file is reached.  Store them in @var{bv} starting "
+            "at index @var{start}.  Return the number of bytes actually "
+            "read, or 0 if an end-of-file was reached.")
+#define FUNC_NAME s_scm_get_bytevector_some_x
+{
+  SCM buf;
+  size_t c_start, c_count, c_len;
+  size_t cur, avail, transfer_size;
+
+  SCM_VALIDATE_BINARY_INPUT_PORT (1, port);
+  SCM_VALIDATE_BYTEVECTOR (2, bv);
+  c_start = scm_to_size_t (start);
+  c_count = scm_to_size_t (count);
+  c_len = SCM_BYTEVECTOR_LENGTH (bv);
+
+  if (SCM_UNLIKELY (c_start + c_count > c_len ||
+                    c_count == 0))
+    scm_out_of_range (FUNC_NAME, count);
+
+  buf = scm_fill_input (port, 0, &cur, &avail);
+  transfer_size = min (avail, c_count);
+
+  if (transfer_size == 0)
+    scm_port_buffer_set_has_eof_p (buf, SCM_BOOL_F);
+  else
+    scm_port_buffer_take
+      (buf, ((scm_t_uint8 *) SCM_BYTEVECTOR_CONTENTS (bv)) + c_start,
+       transfer_size, cur, avail);
+  
+  return scm_from_size_t (transfer_size);
+}
+#undef FUNC_NAME
+
 SCM_DEFINE (scm_get_bytevector_all, "get-bytevector-all", 1, 0, 0,
            (SCM port),
            "Read from @var{port}, blocking as necessary, until "
diff --git a/libguile/r6rs-ports.h b/libguile/r6rs-ports.h
index a2c63c7f4..7dfa382ef 100644
--- a/libguile/r6rs-ports.h
+++ b/libguile/r6rs-ports.h
@@ -1,7 +1,7 @@
 #ifndef SCM_R6RS_PORTS_H
 #define SCM_R6RS_PORTS_H
 
-/* Copyright (C) 2009, 2010, 2011, 2013 Free Software Foundation, Inc.
+/* Copyright (C) 2009-2011, 2013, 2019 Free Software Foundation, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public License
@@ -46,4 +46,7 @@ SCM_API SCM scm_get_string_n_x (SCM, SCM, SCM, SCM);
 SCM_API void scm_init_r6rs_ports (void);
 SCM_INTERNAL void scm_register_r6rs_ports (void);
 
+/* Guile extensions, not in R6RS.  */
+SCM_API SCM scm_get_bytevector_some_x (SCM, SCM, SCM, SCM);
+
 #endif /* SCM_R6RS_PORTS_H */
diff --git a/module/ice-9/binary-ports.scm b/module/ice-9/binary-ports.scm
index e0da3df1a..62fd9786f 100644
--- a/module/ice-9/binary-ports.scm
+++ b/module/ice-9/binary-ports.scm
@@ -36,6 +36,7 @@
             get-bytevector-n
             get-bytevector-n!
             get-bytevector-some
+            get-bytevector-some!  ; Guile extension, not in R6RS
             get-bytevector-all
             get-string-n!
             put-u8
-- 
2.21.0

>From 0dbb4989245305f79e053e49330434230fd14cee Mon Sep 17 00:00:00 2001
From: Mark H Weaver <address@hidden>
Date: Mon, 8 Apr 2019 06:22:41 -0400
Subject: [PATCH 2/4] DRAFT: open-pipe*: Improve performance of OPEN_BOTH mode.

* module/ice-9/popen.scm (make-rw-port): Re-implement using R6RS custom
binary input/output ports.
---
 module/ice-9/popen.scm | 56 +++++++++++++++++++++++++++++++-----------
 1 file changed, 42 insertions(+), 14 deletions(-)

diff --git a/module/ice-9/popen.scm b/module/ice-9/popen.scm
index b166e9d0f..1724b5a7e 100644
--- a/module/ice-9/popen.scm
+++ b/module/ice-9/popen.scm
@@ -1,7 +1,7 @@
 ;; popen emulation, for non-stdio based ports.
 
-;;;; Copyright (C) 1998, 1999, 2000, 2001, 2003, 2006, 2010, 2011, 2012,
-;;;;   2013 Free Software Foundation, Inc.
+;;;; Copyright (C) 1998-2001, 2003, 2006, 2010-2013, 2019
+;;;;   Free Software Foundation, Inc.
 ;;;; 
 ;;;; This library is free software; you can redistribute it and/or
 ;;;; modify it under the terms of the GNU Lesser General Public
@@ -19,10 +19,12 @@
 ;;;; 
 
 (define-module (ice-9 popen)
-  :use-module (ice-9 threads)
-  :use-module (srfi srfi-9)
-  :export (port/pid-table open-pipe* open-pipe close-pipe open-input-pipe
-          open-output-pipe open-input-output-pipe))
+  #:use-module (rnrs bytevectors)
+  #:use-module (ice-9 binary-ports)
+  #:use-module (ice-9 threads)
+  #:use-module (srfi srfi-9)
+  #:export (port/pid-table open-pipe* open-pipe close-pipe open-input-pipe
+            open-output-pipe open-input-output-pipe))
 
 (eval-when (expand load eval)
   (load-extension (string-append "libguile-" (effective-version))
@@ -34,14 +36,40 @@
   (pid pipe-info-pid set-pipe-info-pid!))
 
 (define (make-rw-port read-port write-port)
-  (make-soft-port
-   (vector
-    (lambda (c) (write-char c write-port))
-    (lambda (s) (display s write-port))
-    (lambda () (force-output write-port))
-    (lambda () (read-char read-port))
-    (lambda () (close-port read-port) (close-port write-port)))
-   "r+"))
+  (define (read! bv start count)
+    (get-bytevector-some! read-port bv start count))
+
+  (define (write! bv start count)
+    (put-bytevector write-port bv start count)
+    count)
+
+  (define (close)
+    (close-port read-port)
+    (close-port write-port))
+
+  (define rw-port
+    (make-custom-binary-input/output-port "ice-9-popen-rw-port"
+                                          read!
+                                          write!
+                                          #f ;get-position
+                                          #f ;set-position!
+                                          close))
+  ;; Enable buffering on 'read-port' so that 'get-bytevector-some' will
+  ;; return non-trivial blocks.
+  (setvbuf read-port 'block 65536)
+
+  ;; Inherit the port-encoding from the read-port.
+  (set-port-encoding! rw-port (port-encoding read-port))
+
+  ;; Reset the port encoding on the underlying ports to inhibit BOM
+  ;; handling there.  Instead, the BOM handling (if any) will be handled
+  ;; in the rw-port.  In the current implementation of Guile ports,
+  ;; using binary I/O primitives alone is not enough to reliably inhibit
+  ;; BOM handling, if the port encoding is set to UTF-{8,16,32}.
+  (set-port-encoding! read-port "ISO-8859-1")
+  (set-port-encoding! write-port "ISO-8859-1")
+
+  rw-port)
 
 ;; a guardian to ensure the cleanup is done correctly when
 ;; an open pipe is gc'd or a close-port is used.
-- 
2.21.0

>From be411941ddf79561b168ae15f1f1bec96d1305ed Mon Sep 17 00:00:00 2001
From: Mark H Weaver <address@hidden>
Date: Tue, 16 Apr 2019 23:14:58 -0400
Subject: [PATCH 3/4] DRAFT: Make 'get-bytevector-some' and
 'get-bytevector-some!' suspendable.

---
 module/ice-9/suspendable-ports.scm | 29 ++++++++++++++++++++++++++++-
 1 file changed, 28 insertions(+), 1 deletion(-)

diff --git a/module/ice-9/suspendable-ports.scm 
b/module/ice-9/suspendable-ports.scm
index a366c8b9c..96107c43d 100644
--- a/module/ice-9/suspendable-ports.scm
+++ b/module/ice-9/suspendable-ports.scm
@@ -1,5 +1,5 @@
 ;;; Ports, implemented in Scheme
-;;; Copyright (C) 2016 Free Software Foundation, Inc.
+;;; Copyright (C) 2016, 2019 Free Software Foundation, Inc.
 ;;;
 ;;; This library is free software: you can redistribute it and/or modify
 ;;; it under the terms of the GNU Lesser General Public License as
@@ -292,6 +292,32 @@
        ((< (- count pos) (port-read-buffering port)) (buffer-and-fill pos))
        (else (fill-directly pos))))))
 
+(define (get-bytevector-some port)
+  (call-with-values (lambda () (fill-input port 1 'binary))
+    (lambda (buf cur buffered)
+      (if (zero? buffered)
+          (begin
+            (set-port-buffer-has-eof?! buf #f)
+            the-eof-object)
+          (let ((result (make-bytevector buffered)))
+            (bytevector-copy! (port-buffer-bytevector buf) cur
+                              result 0 buffered)
+            (set-port-buffer-cur! buf (+ cur buffered))
+            result)))))
+
+(define (get-bytevector-some! port bv start count)
+  (call-with-values (lambda () (fill-input port 1 'binary))
+    (lambda (buf cur buffered)
+      (if (zero? buffered)
+          (begin
+            (set-port-buffer-has-eof?! buf #f)
+            0)
+          (let ((transfer-size (min count buffered)))
+            (bytevector-copy! (port-buffer-bytevector buf) cur
+                              transfer-size start buffered)
+            (set-port-buffer-cur! buf (+ cur transfer-size))
+            transfer-size)))))
+
 (define (put-u8 port byte)
   (let* ((buf (port-write-buffer port))
          (bv (port-buffer-bytevector buf))
@@ -703,6 +729,7 @@
      accept connect)
     ((ice-9 binary-ports)
      get-u8 lookahead-u8 get-bytevector-n
+     get-bytevector-some get-bytevector-some!
      put-u8 put-bytevector)
     ((ice-9 textual-ports)
      put-char put-string)
-- 
2.21.0

>From 7cbf017e1656a450c5925414df2f0a751a6ba6c2 Mon Sep 17 00:00:00 2001
From: Mark H Weaver <address@hidden>
Date: Sun, 14 Apr 2019 17:30:40 -0400
Subject: [PATCH 4/4] DRAFT: Make custom binary ports suspendable.

---
 libguile/r6rs-ports.c         | 136 +++++++++++++++++++++-------------
 module/ice-9/binary-ports.scm |  24 +++++-
 2 files changed, 107 insertions(+), 53 deletions(-)

diff --git a/libguile/r6rs-ports.c b/libguile/r6rs-ports.c
index 9b64696a6..9c4f2c8da 100644
--- a/libguile/r6rs-ports.c
+++ b/libguile/r6rs-ports.c
@@ -1,4 +1,5 @@
-/* Copyright (C) 2009, 2010, 2011, 2013-2015, 2018 Free Software Foundation, 
Inc.
+/* Copyright (C) 2009-2011, 2013-2015, 2018, 2019
+ *   Free Software Foundation, Inc.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public License
@@ -289,24 +290,6 @@ make_custom_binary_input_port (SCM read_proc, SCM 
get_position_proc,
                                         (scm_t_bits) stream);
 }
 
-static size_t
-custom_binary_input_port_read (SCM port, SCM dst, size_t start, size_t count)
-#define FUNC_NAME "custom_binary_input_port_read"
-{
-  struct custom_binary_port *stream = (void *) SCM_STREAM (port);
-  SCM octets;
-  size_t c_octets;
-
-  octets = scm_call_3 (stream->read, dst, scm_from_size_t (start),
-                       scm_from_size_t (count));
-  c_octets = scm_to_size_t (octets);
-  if (c_octets > count)
-    scm_out_of_range (FUNC_NAME, octets);
-
-  return c_octets;
-}
-#undef FUNC_NAME
-
 
 SCM_DEFINE (scm_make_custom_binary_input_port,
            "make-custom-binary-input-port", 5, 0, 0,
@@ -317,6 +300,9 @@ SCM_DEFINE (scm_make_custom_binary_input_port,
            "index where octets should be written, and an octet count.")
 #define FUNC_NAME s_scm_make_custom_binary_input_port
 {
+  /* Ensure that custom binary ports are initialized. */
+  scm_c_resolve_module ("ice-9 binary-ports");
+  
   SCM_VALIDATE_STRING (1, id);
   SCM_VALIDATE_PROC (2, read_proc);
 
@@ -340,9 +326,11 @@ static inline void
 initialize_custom_binary_input_ports (void)
 {
   custom_binary_input_port_type =
-    scm_make_port_type ("r6rs-custom-binary-input-port",
-                       custom_binary_input_port_read, NULL);
+    scm_make_port_type ("r6rs-custom-binary-input-port", NULL, NULL);
 
+  scm_set_port_scm_read (custom_binary_input_port_type,
+                         scm_c_private_ref ("ice-9 binary-ports",
+                                            "custom-binary-port-read!"));
   scm_set_port_seek (custom_binary_input_port_type, custom_binary_port_seek);
   scm_set_port_random_access_p (custom_binary_input_port_type,
                                 custom_binary_port_random_access_p);
@@ -929,28 +917,6 @@ make_custom_binary_output_port (SCM write_proc, SCM 
get_position_proc,
                                         (scm_t_bits) stream);
 }
 
-/* Flush octets from BUF to the backing store.  */
-static size_t
-custom_binary_output_port_write (SCM port, SCM src, size_t start, size_t count)
-#define FUNC_NAME "custom_binary_output_port_write"
-{
-  struct custom_binary_port *stream = (void *) SCM_STREAM (port);
-  size_t written;
-  SCM result;
-
-  result = scm_call_3 (stream->write, src, scm_from_size_t (start),
-                       scm_from_size_t (count));
-
-  written = scm_to_size_t (result);
-  if (written > count)
-    scm_wrong_type_arg_msg (FUNC_NAME, 0, result,
-                            "R6RS custom binary output port `write!' "
-                            "returned a incorrect integer");
-
-  return written;
-}
-#undef FUNC_NAME
-
 
 SCM_DEFINE (scm_make_custom_binary_output_port,
            "make-custom-binary-output-port", 5, 0, 0,
@@ -961,6 +927,9 @@ SCM_DEFINE (scm_make_custom_binary_output_port,
            "index where octets should be written, and an octet count.")
 #define FUNC_NAME s_scm_make_custom_binary_output_port
 {
+  /* Ensure that custom binary ports are initialized. */
+  scm_c_resolve_module ("ice-9 binary-ports");
+  
   SCM_VALIDATE_STRING (1, id);
   SCM_VALIDATE_PROC (2, write_proc);
 
@@ -984,9 +953,11 @@ static inline void
 initialize_custom_binary_output_ports (void)
 {
   custom_binary_output_port_type =
-    scm_make_port_type ("r6rs-custom-binary-output-port",
-                       NULL, custom_binary_output_port_write);
+    scm_make_port_type ("r6rs-custom-binary-output-port", NULL, NULL);
 
+  scm_set_port_scm_write (custom_binary_output_port_type,
+                          scm_c_private_ref ("ice-9 binary-ports",
+                                             "custom-binary-port-write!"));
   scm_set_port_seek (custom_binary_output_port_type, custom_binary_port_seek);
   scm_set_port_random_access_p (custom_binary_output_port_type,
                                 custom_binary_port_random_access_p);
@@ -1033,6 +1004,9 @@ SCM_DEFINE (scm_make_custom_binary_input_output_port,
             "written, and an octet count.")
 #define FUNC_NAME s_scm_make_custom_binary_input_output_port
 {
+  /* Ensure that custom binary ports are initialized. */
+  scm_c_resolve_module ("ice-9 binary-ports");
+  
   SCM_VALIDATE_STRING (1, id);
   SCM_VALIDATE_PROC (2, read_proc);
   SCM_VALIDATE_PROC (3, write_proc);
@@ -1057,10 +1031,14 @@ static inline void
 initialize_custom_binary_input_output_ports (void)
 {
   custom_binary_input_output_port_type =
-    scm_make_port_type ("r6rs-custom-binary-input/output-port",
-                       custom_binary_input_port_read,
-                       custom_binary_output_port_write);
-
+    scm_make_port_type ("r6rs-custom-binary-input/output-port", NULL, NULL);
+
+  scm_set_port_scm_read (custom_binary_input_output_port_type,
+                         scm_c_private_ref ("ice-9 binary-ports",
+                                            "custom-binary-port-read!"));
+  scm_set_port_scm_write (custom_binary_input_output_port_type,
+                          scm_c_private_ref ("ice-9 binary-ports",
+                                             "custom-binary-port-write!"));
   scm_set_port_seek (custom_binary_input_output_port_type,
                      custom_binary_port_seek);
   scm_set_port_random_access_p (custom_binary_input_output_port_type,
@@ -1072,6 +1050,56 @@ initialize_custom_binary_input_output_ports (void)
 
 
 
+/* Internal accessors needed by 'custom-binary-port-read!' and
+   'custom-binary-port-write!'.  */
+
+SCM_INTERNAL SCM scm_i_custom_binary_port_reader (SCM);
+SCM_DEFINE (scm_i_custom_binary_port_reader,
+            "custom-binary-port-reader", 1, 0, 0,
+            (SCM port),
+            "Return the 'read!' procedure associated with PORT, "
+            "which must be custom binary input or input/output port.")
+#define FUNC_NAME s_scm_i_custom_binary_port_reader
+{
+  SCM_VALIDATE_BINARY_INPUT_PORT (1, port);
+
+  if (SCM_PORT_TYPE (port) == custom_binary_input_port_type ||
+      SCM_PORT_TYPE (port) == custom_binary_input_output_port_type)
+    {
+      struct custom_binary_port *stream = (void *) SCM_STREAM (port);
+      return stream->read;
+    }
+  else
+    scm_wrong_type_arg_msg (FUNC_NAME, 1, port,
+                            "custom binary input or input/output port");
+}
+#undef FUNC_NAME
+
+SCM_INTERNAL SCM scm_i_custom_binary_port_writer (SCM);
+SCM_DEFINE (scm_i_custom_binary_port_writer,
+            "custom-binary-port-writer", 1, 0, 0,
+            (SCM port),
+            "Return the 'write!' procedure associated with PORT, "
+            "which must be custom binary output or input/output port.")
+#define FUNC_NAME s_scm_i_custom_binary_port_writer
+{
+  SCM_VALIDATE_BINARY_OUTPUT_PORT (1, port);
+
+  if (SCM_PORT_TYPE (port) == custom_binary_output_port_type ||
+      SCM_PORT_TYPE (port) == custom_binary_input_output_port_type)
+    {
+      struct custom_binary_port *stream = (void *) SCM_STREAM (port);
+      return stream->write;
+    }
+  else
+    scm_wrong_type_arg_msg (FUNC_NAME, 1, port,
+                            "custom binary output or input/output port");
+}
+#undef FUNC_NAME
+
+
+
+
 /* Transcoded ports.  */
 
 static scm_t_port_type *transcoded_port_type = 0;
@@ -1197,15 +1225,19 @@ scm_register_r6rs_ports (void)
                            NULL);
 
   initialize_bytevector_input_ports ();
-  initialize_custom_binary_input_ports ();
   initialize_bytevector_output_ports ();
-  initialize_custom_binary_output_ports ();
-  initialize_custom_binary_input_output_ports ();
   initialize_transcoded_ports ();
 }
 
 void
 scm_init_r6rs_ports (void)
 {
+  /* We postpone registering custom binary ports until (ice-9 binary-ports)
+   * is loaded, because these custom port types depend on Scheme procedures
+   * defined there.  */
+  initialize_custom_binary_input_ports ();
+  initialize_custom_binary_output_ports ();
+  initialize_custom_binary_input_output_ports ();
+
 #include "libguile/r6rs-ports.x"
 }
diff --git a/module/ice-9/binary-ports.scm b/module/ice-9/binary-ports.scm
index 62fd9786f..70c6577dd 100644
--- a/module/ice-9/binary-ports.scm
+++ b/module/ice-9/binary-ports.scm
@@ -1,6 +1,6 @@
 ;;;; binary-ports.scm --- Binary IO on ports
 
-;;;;   Copyright (C) 2009, 2010, 2011, 2013 Free Software Foundation, Inc.
+;;;; Copyright (C) 2009-2011, 2013, 2019 Free Software Foundation, Inc.
 ;;;;
 ;;;; This library is free software; you can redistribute it and/or
 ;;;; modify it under the terms of the GNU Lesser General Public
@@ -46,6 +46,28 @@
             make-custom-binary-output-port
             make-custom-binary-input/output-port))
 
+(define (custom-binary-port-read! port bv start count)
+  (let* ((read! (custom-binary-port-reader port))
+         (result (read! bv start count)))
+    (unless (and (exact-integer? result)
+                 (<= 0 result count))
+      (scm-error 'out-of-range #f
+                 "custom port 'read!' (~S) returned value out of range; 
expected an exact integer between 0 and ~A, got ~A"
+                 (list read! count result)
+                 (list result)))
+    result))
+
+(define (custom-binary-port-write! port bv start count)
+  (let* ((write! (custom-binary-port-writer port))
+         (result (write! bv start count)))
+    (unless (and (exact-integer? result)
+                 (<= 0 result count))
+      (scm-error 'out-of-range #f
+                 "custom port 'write!' (~S) returned value out of range; 
expected an exact integer between 0 and ~A, got ~A"
+                 (list write! count result)
+                 (list result)))
+    result))
+
 ;; Note that this extension also defines %make-transcoded-port, which is
 ;; not exported but is used by (rnrs io ports).
 
-- 
2.21.0