bpo-41586: Add pipesize parameter to subprocess & F_GETPIPE_SZ and F_SETPIPE_SZ to fcntl. (GH-21921)

* Add F_SETPIPE_SZ and F_GETPIPE_SZ to fcntl module
* Add pipesize parameter for subprocess.Popen class

This will allow the user to control the size of the pipes.
On linux the default is 64K. When a pipe is full it blocks for writing.
When a pipe is empty it blocks for reading. On processes that are
very fast this can lead to a lot of wasted CPU cycles. On a typical
Linux system the max pipe size is 1024K which is much better.
For high performance-oriented libraries such as xopen it is nice to
be able to set the pipe size.

The workaround without this feature is to use my_popen_process.stdout.fileno() in
conjuction with fcntl and 1031 (value of F_SETPIPE_SZ) to acquire this behavior.

3 files changed, 77 insertions, 2 deletions

diff --git a/Lib/subprocess.py b/Lib/subprocess.py
index 86fdf27..6a6c2fc 100644
--- a/Lib/subprocess.py
+++ b/Lib/subprocess.py
@@ -62,6 +62,11 @@ try:
     import grp
 except ImportError:
     grp = None
+try:
+    import fcntl
+except ImportError:
+    fcntl = None
+
 
 __all__ = ["Popen", "PIPE", "STDOUT", "call", "check_call", "getstatusoutput",
            "getoutput", "check_output", "run", "CalledProcessError", "DEVNULL",
@@ -756,7 +761,7 @@ class Popen(object):
                  startupinfo=None, creationflags=0,
                  restore_signals=True, start_new_session=False,
                  pass_fds=(), *, user=None, group=None, extra_groups=None,
-                 encoding=None, errors=None, text=None, umask=-1):
+                 encoding=None, errors=None, text=None, umask=-1, pipesize=-1):
         """Create new Popen instance."""
         _cleanup()
         # Held while anything is calling waitpid before returncode has been
@@ -773,6 +778,11 @@ class Popen(object):
         if not isinstance(bufsize, int):
             raise TypeError("bufsize must be an integer")
 
+        if pipesize is None:
+            pipesize = -1  # Restore default
+        if not isinstance(pipesize, int):
+            raise TypeError("pipesize must be an integer")
+
         if _mswindows:
             if preexec_fn is not None:
                 raise ValueError("preexec_fn is not supported on Windows "
@@ -797,6 +807,7 @@ class Popen(object):
         self.returncode = None
         self.encoding = encoding
         self.errors = errors
+        self.pipesize = pipesize
 
         # Validate the combinations of text and universal_newlines
         if (text is not None and universal_newlines is not None
@@ -1575,6 +1586,8 @@ class Popen(object):
                 pass
             elif stdin == PIPE:
                 p2cread, p2cwrite = os.pipe()
+                if self.pipesize > 0 and hasattr(fcntl, "F_SETPIPE_SZ"):
+                    fcntl.fcntl(p2cwrite, fcntl.F_SETPIPE_SZ, self.pipesize)
             elif stdin == DEVNULL:
                 p2cread = self._get_devnull()
             elif isinstance(stdin, int):
@@ -1587,6 +1600,8 @@ class Popen(object):
                 pass
             elif stdout == PIPE:
                 c2pread, c2pwrite = os.pipe()
+                if self.pipesize > 0 and hasattr(fcntl, "F_SETPIPE_SZ"):
+                    fcntl.fcntl(c2pwrite, fcntl.F_SETPIPE_SZ, self.pipesize)
             elif stdout == DEVNULL:
                 c2pwrite = self._get_devnull()
             elif isinstance(stdout, int):
@@ -1599,6 +1614,8 @@ class Popen(object):
                 pass
             elif stderr == PIPE:
                 errread, errwrite = os.pipe()
+                if self.pipesize > 0 and hasattr(fcntl, "F_SETPIPE_SZ"):
+                    fcntl.fcntl(errwrite, fcntl.F_SETPIPE_SZ, self.pipesize)
             elif stderr == STDOUT:
                 if c2pwrite != -1:
                     errwrite = c2pwrite
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index 7e10920..8d6e9ff 100644
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -190,6 +190,19 @@ class TestFcntl(unittest.TestCase):
         res = fcntl.fcntl(self.f.fileno(), fcntl.F_GETPATH, bytes(len(expected)))
         self.assertEqual(expected, res)
 
+    @unittest.skipIf(not (hasattr(fcntl, "F_SETPIPE_SZ") and hasattr(fcntl, "F_GETPIPE_SZ")),
+                     "F_SETPIPE_SZ and F_GETPIPE_SZ are not available on all unix platforms.")
+    def test_fcntl_f_pipesize(self):
+        test_pipe_r, test_pipe_w = os.pipe()
+        # Get the default pipesize with F_GETPIPE_SZ
+        pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ)
+        # Multiply the default with 2 to get a new value.
+        fcntl.fcntl(test_pipe_w, fcntl.F_SETPIPE_SZ, pipesize_default * 2)
+        self.assertEqual(fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ), pipesize_default * 2)
+        os.close(test_pipe_r)
+        os.close(test_pipe_w)
+
+
 def test_main():
     run_unittest(TestFcntl)
 
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index 434ba56..8b576c0 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -39,6 +39,11 @@ try:
 except ImportError:
     grp = None
 
+try:
+    import fcntl
+except:
+    fcntl = None
+
 if support.PGO:
     raise unittest.SkipTest("test is not helpful for PGO")
 
@@ -661,6 +666,46 @@ class ProcessTestCase(BaseTestCase):
         p.wait()
         self.assertEqual(p.stdin, None)
 
+    def test_pipesizes(self):
+        # stdin redirection
+        pipesize = 16 * 1024
+        p = subprocess.Popen([sys.executable, "-c",
+                         'import sys; sys.stdin.read(); sys.stdout.write("out"); sys.stderr.write("error!")'],
+                         stdin=subprocess.PIPE,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE,
+                         pipesize=pipesize)
+        # We only assert pipe size has changed on platforms that support it.
+        if sys.platform != "win32" and hasattr(fcntl, "F_GETPIPE_SZ"):
+            for fifo in [p.stdin, p.stdout, p.stderr]:
+                self.assertEqual(fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), pipesize)
+        # Windows pipe size can be acquired with the GetNamedPipeInfoFunction
+        # https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo
+        # However, this function is not yet in _winapi.
+        p.stdin.write(b"pear")
+        p.stdin.close()
+        p.wait()
+
+    def test_pipesize_default(self):
+        p = subprocess.Popen([sys.executable, "-c",
+                         'import sys; sys.stdin.read(); sys.stdout.write("out");'
+                         ' sys.stderr.write("error!")'],
+                         stdin=subprocess.PIPE,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE,
+                         pipesize=-1)
+        # UNIX tests using fcntl
+        if sys.platform != "win32" and hasattr(fcntl, "F_GETPIPE_SZ"):
+            fp_r, fp_w = os.pipe()
+            default_pipesize = fcntl.fcntl(fp_w, fcntl.F_GETPIPE_SZ)
+            for fifo in [p.stdin, p.stdout, p.stderr]:
+                self.assertEqual(
+                    fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), default_pipesize)
+        # On other platforms we cannot test the pipe size (yet). But above code
+        # using pipesize=-1 should not crash.
+        p.stdin.close()
+        p.wait()
+
     def test_env(self):
         newenv = os.environ.copy()
         newenv["FRUIT"] = "orange"
@@ -3503,7 +3548,7 @@ class MiscTests(unittest.TestCase):
 
     def test__all__(self):
         """Ensure that __all__ is populated properly."""
-        intentionally_excluded = {"list2cmdline", "Handle", "pwd", "grp"}
+        intentionally_excluded = {"list2cmdline", "Handle", "pwd", "grp", "fcntl"}
         exported = set(subprocess.__all__)
         possible_exports = set()
         import types