Repository: streamlink/streamlink
Pull request #3709 (2021-04-27 18:33:28)

utils.named_pipe: rewrite named pipes
- split into two classes with an abstract base class
- remove name parameter from class and generate proper names
- remove mode parameter from open() as it's always binary mode
- change type of NamedPipe.path from str to pathlib.Path
- add tests

This is just a code refactor and doesn't fix the other issues mentioned here, as it would require refactoring the cli.output classes and stream.ffmpegmux in addition to the current changes:
https://github.com/streamlink/streamlink/issues/3568#issuecomment-783130671

There are also a couple of other issues I found while rewriting the Windows specific stuff and it seems like it doesn't close the named pipe gracefully and instead just swallows/supresses errors on the last (invalid) write attempt. I couldn't find a solution for this now, hence the lack of `TestNamedPipeWindows.test_write_before_open`, which would also cover this issue.

gravyboat: Looks great @bastimeyer! Sorry about the delay, been a busy week.

Patch:
diff --git a/src/streamlink/stream/ffmpegmux.py b/src/streamlink/stream/ffmpegmux.py
index aa319442..135effc9 100644
--- a/src/streamlink/stream/ffmpegmux.py
+++ b/src/streamlink/stream/ffmpegmux.py
@@ -1,6 +1,4 @@
 import logging
-import os
-import random
 import subprocess
 import sys
 import threading
@@ -62,7 +60,7 @@ class FFMPEGMuxer(StreamIO):
     @staticmethod
     def copy_to_pipe(self, stream, pipe):
         log.debug("Starting copy to pipe: {0}".format(pipe.path))
-        pipe.open("wb")
+        pipe.open()
         while not stream.closed:
             try:
                 data = stream.read(8192)
@@ -87,7 +85,7 @@ class FFMPEGMuxer(StreamIO):
         self.process = None
         self.streams = streams
 
-        self.pipes = [NamedPipe("ffmpeg-{0}-{1}".format(os.getpid(), random.randint(0, 1000))) for _ in self.streams]
+        self.pipes = [NamedPipe() for _ in self.streams]
         self.pipe_threads = [threading.Thread(target=self.copy_to_pipe, args=(self, stream, np))
                              for stream, np in
                              zip(self.streams, self.pipes)]
@@ -103,7 +101,7 @@ class FFMPEGMuxer(StreamIO):
 
         self._cmd = [self.command(session), '-nostats', '-y']
         for np in self.pipes:
-            self._cmd.extend(["-i", np.path])
+            self._cmd.extend(["-i", str(np.path)])
 
         self._cmd.extend(['-c:v', videocodec])
         self._cmd.extend(['-c:a', audiocodec])
diff --git a/src/streamlink/utils/named_pipe.py b/src/streamlink/utils/named_pipe.py
index f4b1b874..b102d3b5 100644
--- a/src/streamlink/utils/named_pipe.py
+++ b/src/streamlink/utils/named_pipe.py
@@ -1,69 +1,124 @@
+import abc
+import logging
 import os
+import random
 import tempfile
+import threading
+from pathlib import Path
 
 from streamlink.compat import is_win32
 
 if is_win32:
     from ctypes import windll, cast, c_ulong, c_void_p, byref
 
-    PIPE_ACCESS_OUTBOUND = 0x00000002
-    PIPE_TYPE_BYTE = 0x00000000
-    PIPE_READMODE_BYTE = 0x00000000
-    PIPE_WAIT = 0x00000000
-    PIPE_UNLIMITED_INSTANCES = 255
-    INVALID_HANDLE_VALUE = -1
+
+log = logging.getLogger(__name__)
+
+_lock = threading.Lock()
+_id = 0
 
 
-class NamedPipe:
-    def __init__(self, name):
-        self.fifo = None
-        self.pipe = None
+class NamedPipeBase(abc.ABC):
+    path: Path
 
-        if is_win32:
-            self.path = os.path.join("\\\\.\\pipe", name)
-            self.pipe = self._create_named_pipe(self.path)
-        else:
-            self.path = os.path.join(tempfile.gettempdir(), name)
-            self._create_fifo(self.path)
+    def __init__(self):
+        global _id
+        with _lock:
+            _id += 1
+            self.name = f"streamlinkpipe-{os.getpid()}-{_id}-{random.randint(0, 9999)}"
+        log.info(f"Creating pipe {self.name}")
+        self._create()
 
-    def _create_fifo(self, name):
-        os.mkfifo(name, 0o660)
+    @abc.abstractmethod
+    def _create(self) -> None:
+        raise NotImplementedError
 
-    def _create_named_pipe(self, path):
-        bufsize = 8192
+    @abc.abstractmethod
+    def open(self) -> None:
+        raise NotImplementedError
 
-        create_named_pipe = windll.kernel32.CreateNamedPipeW
+    @abc.abstractmethod
+    def write(self, data) -> int:
+        raise NotImplementedError
 
-        pipe = create_named_pipe(path, PIPE_ACCESS_OUTBOUND,
-                                 PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
-                                 PIPE_UNLIMITED_INSTANCES,
-                                 bufsize, bufsize,
-                                 0, None)
+    @abc.abstractmethod
+    def close(self) -> None:
+        raise NotImplementedError
 
-        if pipe == INVALID_HANDLE_VALUE:
-            error_code = windll.kernel32.GetLastError()
-            raise OSError("Error code 0x{0:08X}".format(error_code))
 
-        return pipe
+class NamedPipePosix(NamedPipeBase):
+    mode = "wb"
+    permissions = 0o660
+    fifo = None
 
-    def open(self, mode):
-        if not self.pipe:
-            self.fifo = open(self.path, mode)
+    def _create(self):
+        self.path = Path(tempfile.gettempdir(), self.name)
+        os.mkfifo(self.path, self.permissions)
+
+    def open(self):
+        self.fifo = open(self.path, self.mode)
 
     def write(self, data):
-        if self.pipe:
-            windll.kernel32.ConnectNamedPipe(self.pipe, None)
-            written = c_ulong(0)
-            windll.kernel32.WriteFile(self.pipe, cast(data, c_void_p),
-                                      len(data), byref(written),
-                                      None)
-            return written
-        else:
-            return self.fifo.write(data)
+        return self.fifo.write(data)
 
     def close(self):
-        if self.pipe:
-            windll.kernel32.DisconnectNamedPipe(self.pipe)
-        else:
+        if self.fifo:
             self.fifo.close()
+            self.fifo = None
+        if self.path.is_fifo():
             os.unlink(self.path)
+
+
+class NamedPipeWindows(NamedPipeBase):
+    bufsize = 8192
+    pipe = None
+
+    PIPE_ACCESS_OUTBOUND = 0x00000002
+    PIPE_TYPE_BYTE = 0x00000000
+    PIPE_READMODE_BYTE = 0x00000000
+    PIPE_WAIT = 0x00000000
+    PIPE_UNLIMITED_INSTANCES = 255
+    INVALID_HANDLE_VALUE = -1
+
+    @staticmethod
+    def _get_last_error():
+        error_code = windll.kernel32.GetLastError()
+        raise OSError(f"Named pipe error code 0x{error_code:08X}")
+
+    def _create(self):
+        self.path = Path("\\\\.\\pipe", self.name)
+        self.pipe = windll.kernel32.CreateNamedPipeW(
+            str(self.path),
+            self.PIPE_ACCESS_OUTBOUND,
+            self.PIPE_TYPE_BYTE | self.PIPE_READMODE_BYTE | self.PIPE_WAIT,
+            self.PIPE_UNLIMITED_INSTANCES,
+            self.bufsize,
+            self.bufsize,
+            0,
+            None
+        )
+        if self.pipe == self.INVALID_HANDLE_VALUE:
+            self._get_last_error()
+
+    def open(self):
+        windll.kernel32.ConnectNamedPipe(self.pipe, None)
+
+    def write(self, data):
+        written = c_ulong(0)
+        windll.kernel32.WriteFile(
+            self.pipe,
+            cast(data, c_void_p),
+            len(data),
+            byref(written),
+            None
+        )
+        return written.value
+
+    def close(self):
+        if self.pipe is not None:
+            windll.kernel32.DisconnectNamedPipe(self.pipe)
+            windll.kernel32.CloseHandle(self.pipe)
+            self.pipe = None
+
+
+NamedPipe = NamedPipePosix if not is_win32 else NamedPipeWindows
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index c48868d2..2db30ac7 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -96,11 +96,8 @@ def create_output(plugin):
                          "executable with --player.")
 
         if args.player_fifo:
-            pipename = "streamlinkpipe-{0}".format(os.getpid())
-            log.info("Creating pipe {0}".format(pipename))
-
             try:
-                namedpipe = NamedPipe(pipename)
+                namedpipe = NamedPipe()
             except OSError as err:
                 console.exit("Failed to create pipe: {0}", err)
         elif args.player_http:
diff --git a/src/streamlink_cli/output.py b/src/streamlink_cli/output.py
index f684625e..24169014 100644
--- a/src/streamlink_cli/output.py
+++ b/src/streamlink_cli/output.py
@@ -270,7 +270,7 @@ class PlayerOutput(Output):
             raise OSError("Process exited prematurely")
 
         if self.namedpipe:
-            self.namedpipe.open("wb")
+            self.namedpipe.open()
         elif self.http:
             self.http.open()
 


Test patch:
diff --git a/tests/test_utils_named_pipe.py b/tests/test_utils_named_pipe.py
new file mode 100644
index 00000000..7b78bd38
--- /dev/null
+++ b/tests/test_utils_named_pipe.py
@@ -0,0 +1,164 @@
+import threading
+import unittest
+from unittest.mock import Mock, call, patch
+
+from streamlink.compat import is_win32
+from streamlink.utils.named_pipe import NamedPipe, NamedPipePosix, NamedPipeWindows
+
+if is_win32:
+    from ctypes import windll, create_string_buffer, c_ulong, byref
+
+
+GENERIC_READ = 0x80000000
+OPEN_EXISTING = 3
+
+
+class ReadNamedPipeThread(threading.Thread):
+    def __init__(self, pipe: NamedPipe):
+        super().__init__(daemon=True)
+        self.path = str(pipe.path)
+        self.error = None
+        self.data = b""
+        self.done = threading.Event()
+
+    def run(self):
+        try:
+            self.read()
+        except OSError as err:  # pragma: no cover
+            self.error = err
+        self.done.set()
+
+    def read(self):
+        raise NotImplementedError
+
+
+class ReadNamedPipeThreadPosix(ReadNamedPipeThread):
+    def read(self):
+        with open(self.path, "rb") as file:
+            while True:
+                data = file.read(-1)
+                if len(data) == 0:
+                    break
+                self.data += data
+
+
+class ReadNamedPipeThreadWindows(ReadNamedPipeThread):
+    def read(self):
+        handle = windll.kernel32.CreateFileW(self.path, GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        try:
+            while True:
+                data = create_string_buffer(NamedPipeWindows.bufsize)
+                read = c_ulong(0)
+                if not windll.kernel32.ReadFile(handle, data, NamedPipeWindows.bufsize, byref(read), None):  # pragma: no cover
+                    raise OSError(f"Failed reading pipe: {windll.kernel32.GetLastError()}")
+                self.data += data.value
+                if read.value != len(data.value):
+                    break
+        finally:
+            windll.kernel32.CloseHandle(handle)
+
+
+class TestNamedPipe(unittest.TestCase):
+    @patch("streamlink.utils.named_pipe._id", 0)
+    @patch("streamlink.utils.named_pipe.os.getpid", Mock(return_value=12345))
+    @patch("streamlink.utils.named_pipe.random.randint", Mock(return_value=67890))
+    @patch("streamlink.utils.named_pipe.NamedPipe._create", Mock(return_value=None))
+    @patch("streamlink.utils.named_pipe.log")
+    def test_name(self, mock_log):
+        NamedPipe()
+        NamedPipe()
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("Creating pipe streamlinkpipe-12345-1-67890"),
+            call("Creating pipe streamlinkpipe-12345-2-67890")
+        ])
+
+
+@unittest.skipIf(is_win32, "test only applicable on a POSIX OS")
+class TestNamedPipePosix(unittest.TestCase):
+    def test_export(self):
+        self.assertEqual(NamedPipe, NamedPipePosix)
+
+    @patch("streamlink.utils.named_pipe.os.mkfifo")
+    def test_create(self, mock_mkfifo):
+        mock_mkfifo.side_effect = OSError()
+        with self.assertRaises(OSError):
+            NamedPipePosix()
+        self.assertEqual(mock_mkfifo.call_args[0][1:], (0o660,))
+
+    def test_close_before_open(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        pipe.close()
+        self.assertFalse(pipe.path.is_fifo())
+        # closing twice doesn't raise
+        pipe.close()
+
+    def test_write_before_open(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        with self.assertRaises(Exception):
+            pipe.write(b"foo")
+        pipe.close()
+
+    def test_named_pipe(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        reader = ReadNamedPipeThreadPosix(pipe)
+        reader.start()
+        pipe.open()
+        self.assertEqual(pipe.write(b"foo"), 3)
+        self.assertEqual(pipe.write(b"bar"), 3)
+        pipe.close()
+        self.assertFalse(pipe.path.is_fifo())
+        reader.done.wait(4000)
+        self.assertEqual(reader.error, None)
+        self.assertEqual(reader.data, b"foobar")
+        self.assertFalse(reader.is_alive())
+
+
+@unittest.skipIf(not is_win32, "test only applicable on Windows")
+class TestNamedPipeWindows(unittest.TestCase):
+    def test_export(self):
+        self.assertEqual(NamedPipe, NamedPipeWindows)
+
+    @patch("streamlink.utils.named_pipe.windll.kernel32")
+    def test_create(self, mock_kernel32):
+        mock_kernel32.CreateNamedPipeW.return_value = NamedPipeWindows.INVALID_HANDLE_VALUE
+        mock_kernel32.GetLastError.return_value = 12345
+        with self.assertRaises(OSError) as cm:
+            NamedPipeWindows()
+        self.assertEqual(str(cm.exception), "Named pipe error code 0x00003039")
+        self.assertEqual(mock_kernel32.CreateNamedPipeW.call_args[0][1:], (
+            0x00000002,
+            0x00000000,
+            255,
+            8192,
+            8192,
+            0,
+            None
+        ))
+
+    def test_close_before_open(self):
+        pipe = NamedPipeWindows()
+        handle = windll.kernel32.CreateFileW(str(pipe.path), GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        self.assertNotEqual(handle, NamedPipeWindows.INVALID_HANDLE_VALUE)
+        windll.kernel32.CloseHandle(handle)
+        pipe.close()
+        handle = windll.kernel32.CreateFileW(str(pipe.path), GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        self.assertEqual(handle, NamedPipeWindows.INVALID_HANDLE_VALUE)
+        # closing twice doesn't raise
+        pipe.close()
+
+    def test_named_pipe(self):
+        pipe = NamedPipeWindows()
+        reader = ReadNamedPipeThreadWindows(pipe)
+        reader.start()
+        pipe.open()
+        self.assertEqual(pipe.write(b"foo"), 3)
+        self.assertEqual(pipe.write(b"bar"), 3)
+        self.assertEqual(pipe.write(b"\0"), 1)
+        reader.done.wait(4000)
+        self.assertEqual(reader.error, None)
+        self.assertEqual(reader.data, b"foobar")
+        self.assertFalse(reader.is_alive())
+        pipe.close()