Address further comments.

pyperf/_manager.py

@@ -88,7 +88,7 @@ def spawn_worker(self, calibrate_loops, calibrate_warmups):
                              self.args.locale,
                              self.args.copy_env)
 
-        rpipe, wpipe = create_pipe(timeout=self.args.timeout)
+        rpipe, wpipe = create_pipe()
         with rpipe:
             with wpipe:
                 warg = wpipe.to_subprocess()
@@ -108,7 +108,7 @@ def spawn_worker(self, calibrate_loops, calibrate_warmups):
 
             with popen_killer(proc):
                 try:
-                    bench_json = rpipe.read_text()
+                    bench_json = rpipe.read_text(timeout=self.args.timeout)
                     exitcode = proc.wait(timeout=EXIT_TIMEOUT)
                 except TimeoutError as exc:
                     print(exc)

pyperf/_utils.py

@@ -288,9 +288,8 @@ def create_environ(inherit_environ, locale, copy_all):
 class _Pipe:
     _OPEN_MODE = "r"
 
-    def __init__(self, fd, timeout=None):
+    def __init__(self, fd):
         self._fd = fd
-        self._timeout = timeout
         self._file = None
         if MS_WINDOWS:
             self._handle = msvcrt.get_osfhandle(fd)
@@ -320,32 +319,38 @@ def __exit__(self, *args):
 class ReadPipe(_Pipe):
     def open_text(self):
         file = open(self._fd, "r", encoding="utf8")
-        if self._timeout:
-            os.set_blocking(file.fileno(), False)
         self._file = file
         return file
 
-    def read_text(self):
-        with self.open_text() as rfile:
-            if self._timeout is not None:
-                return self._read_text_timeout(rfile, self._timeout)
-            else:
+    def read_text(self, timeout=None):
+        if timeout is not None:
+            return self._read_text_timeout(timeout)
+        else:
+            with self.open_text() as rfile:
                 return rfile.read()
 
-    def _read_text_timeout(self, rfile, timeout):
+    def _read_text_timeout(self, timeout):
+        fd = self.fd
+        os.set_blocking(fd, False)
+
         start_time = time.monotonic()
         output = []
         while True:
             if time.monotonic() - start_time > timeout:
                 raise TimeoutError(f"Timed out after {timeout} seconds")
-            ready, _, _ = select.select([rfile], [], [], timeout)
+            ready, _, _ = select.select([fd], [], [], timeout)
             if not ready:
                 continue
-            data = rfile.read(1024)
+            try:
+                data = os.read(fd, 1024)
+            except BlockingIOError:
+                continue
             if not data:
                 break
             output.append(data)
-        return "".join(output)
+
+        data = b"".join(output)
+        return data.decode("utf8")
 
 
 class WritePipe(_Pipe):
@@ -373,9 +378,9 @@ def open_text(self):
         return file
 
 
-def create_pipe(timeout=None):
+def create_pipe():
     rfd, wfd = os.pipe()
-    rpipe = ReadPipe(rfd, timeout)
+    rpipe = ReadPipe(rfd)
     wpipe = WritePipe(wfd)
     return (rpipe, wpipe)
 

pyperf/tests/test_runner.py

@@ -150,23 +150,32 @@ def test_pipe(self):
                          tests.benchmark_as_json(result.bench))
 
     def test_pipe_with_timeout(self):
-        rpipe, wpipe = create_pipe(timeout=0.1)
+        rpipe, wpipe = create_pipe()
         with rpipe:
             with wpipe:
                 arg = wpipe.to_subprocess()
                 # Don't close the file descriptor, it is closed by
                 # the Runner class
                 wpipe._fd = None
 
-                self.exec_runner('--pipe', str(arg), '--worker', '-l1', '-w1')
+                result = self.exec_runner('--pipe', str(arg),
+                                          '--worker', '-l1', '-w1')
 
             # Mock the select to make the read pipeline not ready
-            with mock.patch('pyperf._utils.select.select', return_value=(False, False, False)):
+            with mock.patch('pyperf._utils.select.select',
+                            return_value=(False, False, False)):
                 with self.assertRaises(TimeoutError) as cm:
-                    rpipe.read_text()
+                    rpipe.read_text(timeout=0.1)
                 self.assertEqual(str(cm.exception),
                          'Timed out after 0.1 seconds')
 
+            # Mock the select to make the read pipeline ready
+            with mock.patch('pyperf._utils.select.select',
+                            return_value=(True, False, False)):
+                bench_json = rpipe.read_text(timeout=0.1)
+                self.assertEqual(bench_json,
+                                tests.benchmark_as_json(result.bench))
+
     def test_json_exists(self):
         with tempfile.NamedTemporaryFile('wb+') as tmp: