Repository: streamlink/streamlink
Pull request #3883 (2021-07-24 07:04:45)

Refactor and improve HLS tests and HLS/Segmented stream classes
As mentioned in #3880, here are some improvements which I wrote while debugging #3868.
The changes are split into multiple logical commits, so please review them one by one.

There's one minor change in the HLSStreamReader in regards to `hls-timeout` and `stream-timeout`, where `stream-timeout` will be used when `hls-timeout` is falsy / zero, but that is also the case for the `hls-segment-attempts`, `hls-segment-threads` and `hls-segment-timeout` parameters in the HLSStreamWriter.

IMO, we don't need special `hls-` or `dash-` options/parameters when generic `stream-` ones already exist. The `dash-` ones which are read in the DASHStreamWriter aren't even defined as streamlink_cli arguments or on the Session's options.


Patch:
diff --git a/src/streamlink/buffers.py b/src/streamlink/buffers.py
index af56b9b4..f98d8fac 100644
--- a/src/streamlink/buffers.py
+++ b/src/streamlink/buffers.py
@@ -103,10 +103,7 @@ class RingBuffer(Buffer):
 
     def read(self, size=-1, block=True, timeout=None):
         if block and not self.closed:
-            self.event_used.wait(timeout)
-
-            # If the event is still not set it's a timeout
-            if not self.event_used.is_set() and self.length == 0:
+            if not self.event_used.wait(timeout) and self.length == 0:
                 raise OSError("Read timeout")
 
         return self._read(size)
diff --git a/src/streamlink/plugins/tv4play.py b/src/streamlink/plugins/tv4play.py
index 6aa176d1..d53fc39f 100644
--- a/src/streamlink/plugins/tv4play.py
+++ b/src/streamlink/plugins/tv4play.py
@@ -12,7 +12,7 @@ log = logging.getLogger(__name__)
 @pluginmatcher(re.compile(r"""
     https?://(?:www\.)?
     (?:
-        tv4play\.se/program/[^?/]+
+        tv4play\.se/program/[^?/]+/[^?/]+
         |
         fotbollskanalen\.se/video
     )
diff --git a/src/streamlink/stream/hls.py b/src/streamlink/stream/hls.py
index a1ed1665..f1127c7f 100644
--- a/src/streamlink/stream/hls.py
+++ b/src/streamlink/stream/hls.py
@@ -123,11 +123,11 @@ class HLSStreamWriter(SegmentedStreamWriter):
                 if future is None or sequence.segment.discontinuity:
                     future = self.executor.submit(self.fetch_map, sequence)
                     self.map_cache.set(sequence.segment.map.uri, future)
-                self.queue(future, sequence, True)
+                self.queue(sequence, future, True)
 
             # regular segment request
             future = self.executor.submit(self.fetch, sequence)
-            self.queue(future, sequence, False)
+            self.queue(sequence, future, False)
 
     def fetch(self, sequence: Sequence) -> Optional[Response]:
         try:
@@ -231,25 +231,6 @@ class HLSStreamWorker(SegmentedStreamWorker):
         elif self.playlist_reload_time_override not in ["segment", "live-edge"]:
             self.playlist_reload_time_override = 0
 
-        self.reload_playlist()
-
-        if self.playlist_end is None:
-            if self.duration_offset_start > 0:
-                log.debug(f"Time offsets negative for live streams, skipping back {self.duration_offset_start} seconds")
-            # live playlist, force offset durations back to None
-            self.duration_offset_start = -self.duration_offset_start
-
-        if self.duration_offset_start != 0:
-            self.playlist_sequence = self.duration_to_sequence(self.duration_offset_start, self.playlist_sequences)
-
-        if self.playlist_sequences:
-            log.debug(f"First Sequence: {self.playlist_sequences[0].num}; "
-                      f"Last Sequence: {self.playlist_sequences[-1].num}")
-            log.debug(f"Start offset: {self.duration_offset_start}; "
-                      f"Duration: {self.duration_limit}; "
-                      f"Start Sequence: {self.playlist_sequence}; "
-                      f"End Sequence: {self.playlist_end}")
-
     def _reload_playlist(self, text, url):
         return hls_playlist.load(text, url)
 
@@ -341,6 +322,25 @@ class HLSStreamWorker(SegmentedStreamWorker):
         return default
 
     def iter_segments(self):
+        self.reload_playlist()
+
+        if self.playlist_end is None:
+            if self.duration_offset_start > 0:
+                log.debug(f"Time offsets negative for live streams, skipping back {self.duration_offset_start} seconds")
+            # live playlist, force offset durations back to None
+            self.duration_offset_start = -self.duration_offset_start
+
+        if self.duration_offset_start != 0:
+            self.playlist_sequence = self.duration_to_sequence(self.duration_offset_start, self.playlist_sequences)
+
+        if self.playlist_sequences:
+            log.debug(f"First Sequence: {self.playlist_sequences[0].num}; "
+                      f"Last Sequence: {self.playlist_sequences[-1].num}")
+            log.debug(f"Start offset: {self.duration_offset_start}; "
+                      f"Duration: {self.duration_limit}; "
+                      f"Start Sequence: {self.playlist_sequence}; "
+                      f"End Sequence: {self.playlist_end}")
+
         total_duration = 0
         while not self.closed:
             for sequence in filter(self.valid_sequence, self.playlist_sequences):
@@ -369,20 +369,21 @@ class HLSStreamReader(SegmentedStreamReader):
     __worker__ = HLSStreamWorker
     __writer__ = HLSStreamWriter
 
-    def __init__(self, stream, *args, **kwargs):
-        super().__init__(stream, *args, **kwargs)
+    def __init__(self, stream):
         self.request_params = dict(stream.args)
-        self.timeout = stream.session.options.get("hls-timeout")
-
-        self.filter_event = Event()
-        self.filter_event.set()
-
         # These params are reserved for internal use
         self.request_params.pop("exception", None)
         self.request_params.pop("stream", None)
         self.request_params.pop("timeout", None)
         self.request_params.pop("url", None)
 
+        self.filter_event = Event()
+        self.filter_event.set()
+
+        timeout = stream.session.options.get("hls-timeout")
+
+        super().__init__(stream, timeout)
+
     def read(self, size):
         while True:
             try:
diff --git a/src/streamlink/stream/segmented.py b/src/streamlink/stream/segmented.py
index 4812dd03..51e1de11 100644
--- a/src/streamlink/stream/segmented.py
+++ b/src/streamlink/stream/segmented.py
@@ -48,20 +48,21 @@ class SegmentedStreamWorker(Thread):
         self.reader = reader
         self.writer = reader.writer
         self.stream = reader.stream
-        self.session = reader.stream.session
+        self.session = reader.session
 
-        self._wait = None
+        self._wait = Event()
 
         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
 
     def close(self):
         """Shuts down the thread."""
-        if not self.closed:
-            log.debug("Closing worker thread")
+        if self.closed:  # pragma: no cover
+            return
+
+        log.debug("Closing worker thread")
 
         self.closed = True
-        if self._wait:
-            self._wait.set()
+        self._wait.set()
 
     def wait(self, time):
         """Pauses the thread for a specified time.
@@ -69,7 +70,6 @@ class SegmentedStreamWorker(Thread):
         Returns False if interrupted by another thread and True if the
         time runs out normally.
         """
-        self._wait = Event()
         return not self._wait.wait(time)
 
     def iter_segments(self):
@@ -82,7 +82,7 @@ class SegmentedStreamWorker(Thread):
 
     def run(self):
         for segment in self.iter_segments():
-            if self.closed:
+            if self.closed:  # pragma: no cover
                 break
             self.writer.put(segment)
 
@@ -102,7 +102,7 @@ class SegmentedStreamWriter(Thread):
         self.closed = False
         self.reader = reader
         self.stream = reader.stream
-        self.session = reader.stream.session
+        self.session = reader.session
 
         if not retries:
             retries = self.session.options.get("stream-segment-attempts")
@@ -121,17 +121,19 @@ class SegmentedStreamWriter(Thread):
         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
 
     def close(self):
-        """Shuts down the thread."""
-        if not self.closed:
-            log.debug("Closing writer thread")
+        """Shuts down the thread, its executor and closes the reader (worker thread and buffer)."""
+        if self.closed:  # pragma: no cover
+            return
+
+        log.debug("Closing writer thread")
 
         self.closed = True
-        self.reader.buffer.close()
+        self.reader.close()
         self.executor.shutdown(wait=True, cancel_futures=True)
 
     def put(self, segment):
         """Adds a segment to the download pool and write queue."""
-        if self.closed:
+        if self.closed:  # pragma: no cover
             return
 
         if segment is None:
@@ -139,17 +141,27 @@ class SegmentedStreamWriter(Thread):
         else:
             future = self.executor.submit(self.fetch, segment, retries=self.retries)
 
-        self.queue(future, segment)
+        self.queue(segment, future)
 
-    def queue(self, future: Optional[Future], segment: Any, *data):
+    def queue(self, segment: Any, future: Optional[Future], *data):
         """Puts values into a queue but aborts if this thread is closed."""
-        while not self.closed:
+        while not self.closed:  # pragma: no branch
             try:
-                self.futures.put((future, segment, *data), block=True, timeout=1)
+                self._futures_put((segment, future, *data))
                 return
             except queue.Full:  # pragma: no cover
                 continue
 
+    def _futures_put(self, item):
+        self.futures.put(item, block=True, timeout=1)
+
+    def _futures_get(self):
+        return self.futures.get(block=True, timeout=0.5)
+
+    @staticmethod
+    def _future_result(future: Future):
+        return future.result(timeout=0.5)
+
     def fetch(self, segment):
         """Fetches a segment.
 
@@ -167,7 +179,7 @@ class SegmentedStreamWriter(Thread):
     def run(self):
         while not self.closed:
             try:
-                future, segment, *data = self.futures.get(block=True, timeout=0.5)
+                segment, future, *data = self._futures_get()
             except queue.Empty:  # pragma: no cover
                 continue
 
@@ -177,7 +189,7 @@ class SegmentedStreamWriter(Thread):
 
             while not self.closed:  # pragma: no branch
                 try:
-                    result = future.result(timeout=0.5)
+                    result = self._future_result(future)
                 except futures.TimeoutError:  # pragma: no cover
                     continue
                 except futures.CancelledError:  # pragma: no cover
@@ -199,18 +211,14 @@ class SegmentedStreamReader(StreamIO):
         super().__init__()
         self.session = stream.session
         self.stream = stream
+        self.timeout = timeout or self.session.options.get("stream-timeout")
 
-        if not timeout:
-            timeout = self.session.options.get("stream-timeout")
-
-        self.timeout = timeout
-
-    def open(self):
         buffer_size = self.session.get_option("ringbuffer-size")
         self.buffer = RingBuffer(buffer_size)
         self.writer = self.__writer__(self)
         self.worker = self.__worker__(self)
 
+    def open(self):
         self.writer.start()
         self.worker.start()
 
@@ -220,8 +228,8 @@ class SegmentedStreamReader(StreamIO):
         self.buffer.close()
 
     def read(self, size):
-        if not self.buffer:
-            return b""
-
-        return self.buffer.read(size, block=self.writer.is_alive(),
-                                timeout=self.timeout)
+        return self.buffer.read(
+            size,
+            block=self.writer.is_alive(),
+            timeout=self.timeout
+        )


Test patch:
diff --git a/tests/mixins/stream_hls.py b/tests/mixins/stream_hls.py
index 55f29b26..c9b3eb18 100644
--- a/tests/mixins/stream_hls.py
+++ b/tests/mixins/stream_hls.py
@@ -10,6 +10,11 @@ from streamlink import Streamlink
 from streamlink.stream.hls import HLSStream, HLSStreamWriter as _HLSStreamWriter
 
 
+TIMEOUT_AWAIT_READ = 5
+TIMEOUT_AWAIT_READ_ONCE = 5
+TIMEOUT_AWAIT_WRITE = 5
+
+
 class HLSItemBase:
     path = ""
 
@@ -89,6 +94,16 @@ class EventedHLSStreamWriter(_HLSStreamWriter):
         self.write_done = Event()
         self.write_error = None
 
+    def _futures_put(self, item):
+        self.futures.put_nowait(item)
+
+    def _futures_get(self):
+        return self.futures.get_nowait()
+
+    @staticmethod
+    def _future_result(future):
+        return future.result(timeout=0)
+
     def write(self, *args, **kwargs):
         # only write once per step
         self.write_wait.wait()
@@ -110,13 +125,8 @@ class HLSStreamReadThread(Thread):
     """
     Run the reader on a separate thread, so that each read can be controlled from within the main thread
     """
-    def __init__(self, session, stream, *args, **kwargs):
-        """
-        :param Streamlink session:
-        :param HLSStream stream:
-        """
-        Thread.__init__(self, *args, **kwargs)
-        self.daemon = True
+    def __init__(self, session: Streamlink, stream: HLSStream, *args, **kwargs):
+        super().__init__(*args, **kwargs, daemon=True)
 
         self.read_wait = Event()
         self.read_once = Event()
@@ -132,7 +142,7 @@ class HLSStreamReadThread(Thread):
         # ensure that at least one read was attempted before closing the writer thread early
         # otherwise, the writer will close the reader's buffer, making it not block on read and yielding empty results
         def _await_read_then_close():
-            self.read_once.wait(timeout=5)
+            self.read_once.wait(timeout=TIMEOUT_AWAIT_READ_ONCE)
             return self.writer_close()
 
         self.writer_close = self.reader.writer.close
@@ -168,18 +178,24 @@ class HLSStreamReadThread(Thread):
         self.data[:] = []
         self.error = None
 
+    def close(self):
+        self.reader.buffer.close()
+        self.read_once.set()
+        self.read_wait.set()
+
 
 class TestMixinStreamHLS(unittest.TestCase):
     __stream__ = HLSStream
     __readthread__ = HLSStreamReadThread
 
+    session: Streamlink
+    stream: HLSStream
+    thread: HLSStreamReadThread
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.mocker = requests_mock.Mocker()
         self.mocks = {}
-        self.session = None
-        self.stream = None
-        self.thread = None
 
     def setUp(self):
         super().setUp()
@@ -187,12 +203,19 @@ class TestMixinStreamHLS(unittest.TestCase):
 
     def tearDown(self):
         super().tearDown()
-        self.close_thread()
+
+        # close read thread and make sure that all threads have terminated before moving on
+        thread = self.thread
+        thread.reader.close()
+        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
+            thread.reader.writer.write_wait.set()
+        thread.read_wait.set()
+        thread.reader.writer.join()
+        thread.reader.worker.join()
+        thread.join()
+
         self.mocker.stop()
         self.mocks.clear()
-        self.session = None
-        self.stream = None
-        self.thread = None
 
     def mock(self, method, url, *args, **kwargs):
         self.mocks[url] = self.mocker.request(method, url, *args, **kwargs)
@@ -213,40 +236,39 @@ class TestMixinStreamHLS(unittest.TestCase):
             segments = segments.values()
         return b"".join([getattr(segment, prop) for segment in segments if cond is None or cond(segment)])
 
-    # close read thread and make sure that all threads have terminated before moving on
-    def close_thread(self):
-        thread = self.thread
-        if isinstance(thread.reader.writer, EventedHLSStreamWriter):
-            thread.reader.writer.write_wait.set()
-        thread.reader.close()
-        thread.read_wait.set()
-        thread.reader.writer.join()
-        thread.reader.worker.join()
-        thread.join()
-
     # make one write call on the write thread and wait until it has finished
-    def await_write(self, write_calls=1, timeout=5):
+    def await_write(self, write_calls=1, timeout=TIMEOUT_AWAIT_WRITE):
         writer = self.thread.reader.writer
-        for _ in range(write_calls):
+        if not writer.is_alive():  # pragma: no cover
+            raise RuntimeError("Write thread is not alive")
+
+        for write_call in range(write_calls):
             writer.write_wait.set()
-            writer.write_done.wait(timeout)
-            writer.write_done.clear()
+            done = writer.write_done.wait(timeout)
             if writer.write_error:
                 raise writer.write_error
+            if not done:  # pragma: no cover
+                raise RuntimeError(f"Await write timeout: write_call={write_call + 1}")
+            writer.write_done.clear()
 
     # make one read call on the read thread and wait until it has finished
-    def await_read(self, read_all=False, timeout=5):
+    def await_read(self, read_all=False, timeout=TIMEOUT_AWAIT_READ):
         thread = self.thread
+        if not thread.is_alive():  # pragma: no cover
+            raise RuntimeError("Read thread is not alive")
+
         thread.read_all = read_all
         thread.read_wait.set()
-        thread.read_done.wait(timeout)
-        thread.read_done.clear()
+        done = thread.read_done.wait(timeout)
 
         try:
             if thread.error:
                 raise thread.error
+            if not done:  # pragma: no cover
+                raise RuntimeError(f"Await read timeout: read_all={read_all}")
             return b"".join(thread.data)
         finally:
+            thread.read_done.clear()
             thread.reset()
 
     def get_session(self, options=None, *args, **kwargs):
@@ -264,7 +286,7 @@ class TestMixinStreamHLS(unittest.TestCase):
 
         self.session = self.get_session(options, *args, **kwargs)
         self.stream = self.__stream__(self.session, self.url(playlists[0]), **(streamoptions or {}))
-        self.thread = self.__readthread__(self.session, self.stream, **(threadoptions or {}))
+        self.thread = self.__readthread__(self.session, self.stream, name=f"ReadThread-{self.id()}", **(threadoptions or {}))
         self.thread.start()
 
         return self.thread, segments
diff --git a/tests/plugins/test_tv4play.py b/tests/plugins/test_tv4play.py
index f1bd77a5..12f14018 100644
--- a/tests/plugins/test_tv4play.py
+++ b/tests/plugins/test_tv4play.py
@@ -6,8 +6,8 @@ class TestPluginCanHandleUrlTV4Play(PluginCanHandleUrl):
     __plugin__ = TV4Play
 
     should_match = [
-        'https://www.tv4play.se/program/fridas-vm-resa/10000884',
-        'https://www.tv4play.se/program/monk/3938989',
-        'https://www.tv4play.se/program/nyheterna/10378590',
+        'https://www.tv4play.se/program/robinson/del-26-sasong-2021/13299862',
+        'https://www.tv4play.se/program/sverige-mot-norge/del-1-sasong-1/12490380',
+        'https://www.tv4play.se/program/nyheterna/live/10378590',
         'https://www.fotbollskanalen.se/video/10395484/ghoddos-fullbordar-vandningen---ger-ofk-ledningen/',
     ]
diff --git a/tests/streams/test_hls.py b/tests/streams/test_hls.py
index ac3d4753..afd79ebb 100644
--- a/tests/streams/test_hls.py
+++ b/tests/streams/test_hls.py
@@ -248,6 +248,9 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
         ])
 
+        # close read thread early
+        self.thread.close()
+
         with self.assertRaises(ValueError) as cm:
             self.await_write()
         self.assertEqual(str(cm.exception), "Padding is incorrect.", "Crypto.Util.Padding.unpad exception")
@@ -260,15 +263,24 @@ class TestHLSStreamEncrypted(TestMixinStreamHLS, unittest.TestCase):
             Playlist(0, [key, SegmentEnc(0, aesKey, aesIv, padding=padding)], end=True)
         ])
 
+        # close read thread early
+        self.thread.close()
+
         with self.assertRaises(ValueError) as cm:
             self.await_write()
         self.assertEqual(str(cm.exception), "PKCS#7 padding is incorrect.", "Crypto.Util.Padding.unpad exception")
 
 
 @patch("streamlink.stream.hls.HLSStreamWorker.wait", Mock(return_value=True))
-@patch("streamlink.stream.hls.HLSStreamWriter.run", Mock(return_value=True))
 class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
-    segments = [Segment(0, "", 11), Segment(1, "", 7), Segment(2, "", 5), Segment(3, "", 3)]
+    __stream__ = EventedHLSStream
+
+    segments = [
+        Segment(0, duration=11),
+        Segment(1, duration=7),
+        Segment(2, duration=5),
+        Segment(3, duration=3)
+    ]
 
     def get_session(self, options=None, reload_time=None, *args, **kwargs):
         return super().get_session(dict(options or {}, **{
@@ -277,8 +289,9 @@ class TestHlsPlaylistReloadTime(TestMixinStreamHLS, unittest.TestCase):
         }))
 
     def subject(self, *args, **kwargs):
-        thread, _ = super().subject(*args, **kwargs)
-        self.await_read(read_all=True)
+        thread, segments = super().subject(*args, **kwargs)
+        # wait for the segments to be written, so that we're sure the playlist was parsed
+        self.await_write(len(segments))
 
         return thread.reader.worker.playlist_reload_time
 
diff --git a/tests/streams/test_hls_filtered.py b/tests/streams/test_hls_filtered.py
index 73b487db..5d79a1d3 100644
--- a/tests/streams/test_hls_filtered.py
+++ b/tests/streams/test_hls_filtered.py
@@ -133,7 +133,7 @@ class TestFilteredHLSStream(TestMixinStreamHLS, unittest.TestCase):
     @patch("streamlink.stream.hls.HLSStreamWriter.should_filter_sequence", new=filter_sequence)
     def test_filtered_closed(self):
         thread, reader, writer, segments = self.subject([
-            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)])
+            Playlist(0, [SegmentFiltered(0), SegmentFiltered(1)], end=True)
         ])
 
         # mock the reader thread's filter_event.wait method, so that the main thread can wait on its call
@@ -144,27 +144,24 @@ class TestFilteredHLSStream(TestMixinStreamHLS, unittest.TestCase):
             filter_event_wait_called.set()
             return orig_wait(*args, **kwargs)
 
-        mock = patch.object(reader.filter_event, "wait", side_effect=mocked_wait)
-        mock.start()
-
-        # write first filtered segment and trigger the filter_event's lock
-        self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
-        self.await_write()
-        self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
-
-        # make reader read (no data available yet)
-        thread.read_wait.set()
-        # before calling reader.close(), wait until reader thread's filter_event.wait was called
-        filter_event_wait_called.wait()
-
-        # close stream while reader is waiting for filtering to end
-        thread.reader.close()
-        thread.read_done.wait()
-        thread.read_done.clear()
-        self.assertEqual(thread.data, [b""], "Stops reading on stream close")
-        self.assertFalse(thread.error, "Is not a read timeout on stream close")
+        with patch.object(reader.filter_event, "wait", side_effect=mocked_wait):
+            # write first filtered segment and trigger the filter_event's lock
+            self.assertTrue(reader.filter_event.is_set(), "Doesn't let the reader wait if not filtering")
+            self.await_write()
+            self.assertFalse(reader.filter_event.is_set(), "Lets the reader wait if filtering")
 
-        mock.stop()
+            # make reader read (no data available yet)
+            thread.read_wait.set()
+            # before calling reader.close(), wait until reader thread's filter_event.wait was called
+            if not filter_event_wait_called.wait(timeout=5):  # pragma: no cover
+                raise RuntimeError("Missing filter_event.wait() call")
+
+            # close stream while reader is waiting for filtering to end
+            thread.reader.close()
+            thread.read_done.wait()
+            thread.read_done.clear()
+            self.assertEqual(thread.data, [b""], "Stops reading on stream close")
+            self.assertFalse(thread.error, "Is not a read timeout on stream close")
 
     def test_hls_segment_ignore_names(self):
         thread, reader, writer, segments = self.subject([