We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
SegmentedStreamWriter.close() does not reliably finish before CLI exits (race condition)
### Checklist

- [X] This is a bug report and not a different kind of issue
- [X] [I have read the contribution guidelines](https://github.com/streamlink/streamlink/blob/master/CONTRIBUTING.md#contributing-to-streamlink)
- [X] [I have checked the list of open and recently closed bug reports](https://github.com/streamlink/streamlink/issues?q=is%3Aissue+label%3A%22bug%22)
- [X] [I have checked the commit log of the master branch](https://github.com/streamlink/streamlink/commits/master)

### Streamlink version

Latest build from the master branch

### Description

### Background
I was doing some works on latest `master` branch, adding some feature on Segmented/HLS streams, and it requires some cleanup, so I added them in the end of `SegmentedStreamWriter.close()`, like

```python
        self.closed = True
        self.reader.close()
        self.executor.shutdown(wait=True, cancel_futures=True)

        __my_extra_cleanup()
```

And I found my cleanup code does not always being executed, when HLS streams end normally.

### Problem
When an HLS stream ends normally, the whole shutdown process is triggered by the last line of `SegmentedStreamWriter.run()`, `self.close()`. Then `SegmentedStreamWriter.close()` calls `SegmentedStreamReader.close()`, then the iteration loop of `stream_cli.main:read_stream()` will reach its end, then `main()` exits.

The problem is, `SegmentedStreamWriter.run() -> SegmentedStreamWriter.close()` was run in a separated thread, which means `SegmentedStreamWriter.close()` cannot reliably finish its work before main thread exits.

### To reproduce
To reliably trigger the problem, adding a sleep to original `SegmentedStreamWriter.close()`, like

```python
        self.closed = True
        self.reader.close()
        self.executor.shutdown(wait=True, cancel_futures=True)

        time.sleep(3)
        log.debug("SegmentedStreamWriter.close() ends")
```

Then run the CLI with a short HLS stream, the `SegmentedStreamWriter.close() ends` message never appears.


### Debug log

```text
......

[cli][info] Stream ended
[cli][info] Closing currently open stream...
```

### Expected result
```text
......

[cli][info] Stream ended
[cli][info] Closing currently open stream...
[stream.segmented][debug] SegmentedStreamWriter.close() ends
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/streamlink/stream/segmented.py`
Content:
```
1 import logging
2 import queue
3 from concurrent import futures
4 from concurrent.futures import Future, ThreadPoolExecutor
5 from sys import version_info
6 from threading import Event, Thread
7 from typing import Any, Optional
8 
9 from streamlink.buffers import RingBuffer
10 from streamlink.stream.stream import StreamIO
11 
12 log = logging.getLogger(__name__)
13 
14 
15 class CompatThreadPoolExecutor(ThreadPoolExecutor):
16     if version_info < (3, 9):
17         def shutdown(self, wait=True, cancel_futures=False):  # pragma: no cover
18             with self._shutdown_lock:
19                 self._shutdown = True
20                 if cancel_futures:
21                     # Drain all work items from the queue, and then cancel their
22                     # associated futures.
23                     while True:
24                         try:
25                             work_item = self._work_queue.get_nowait()
26                         except queue.Empty:
27                             break
28                         if work_item is not None:
29                             work_item.future.cancel()
30 
31                 # Send a wake-up to prevent threads calling
32                 # _work_queue.get(block=True) from permanently blocking.
33                 self._work_queue.put(None)
34             if wait:
35                 for t in self._threads:
36                     t.join()
37 
38 
39 class SegmentedStreamWorker(Thread):
40     """The general worker thread.
41 
42     This thread is responsible for queueing up segments in the
43     writer thread.
44     """
45 
46     def __init__(self, reader, **kwargs):
47         self.closed = False
48         self.reader = reader
49         self.writer = reader.writer
50         self.stream = reader.stream
51         self.session = reader.session
52 
53         self._wait = Event()
54 
55         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
56 
57     def close(self):
58         """Shuts down the thread."""
59         if self.closed:  # pragma: no cover
60             return
61 
62         log.debug("Closing worker thread")
63 
64         self.closed = True
65         self._wait.set()
66 
67     def wait(self, time):
68         """Pauses the thread for a specified time.
69 
70         Returns False if interrupted by another thread and True if the
71         time runs out normally.
72         """
73         return not self._wait.wait(time)
74 
75     def iter_segments(self):
76         """The iterator that generates segments for the worker thread.
77 
78         Should be overridden by the inheriting class.
79         """
80         return
81         yield
82 
83     def run(self):
84         for segment in self.iter_segments():
85             if self.closed:  # pragma: no cover
86                 break
87             self.writer.put(segment)
88 
89         # End of stream, tells the writer to exit
90         self.writer.put(None)
91         self.close()
92 
93 
94 class SegmentedStreamWriter(Thread):
95     """The writer thread.
96 
97     This thread is responsible for fetching segments, processing them
98     and finally writing the data to the buffer.
99     """
100 
101     def __init__(self, reader, size=20, retries=None, threads=None, timeout=None):
102         self.closed = False
103         self.reader = reader
104         self.stream = reader.stream
105         self.session = reader.session
106 
107         if not retries:
108             retries = self.session.options.get("stream-segment-attempts")
109 
110         if not threads:
111             threads = self.session.options.get("stream-segment-threads")
112 
113         if not timeout:
114             timeout = self.session.options.get("stream-segment-timeout")
115 
116         self.retries = retries
117         self.timeout = timeout
118         self.threads = threads
119         self.executor = CompatThreadPoolExecutor(max_workers=self.threads)
120         self.futures = queue.Queue(size)
121 
122         super().__init__(daemon=True, name=f"Thread-{self.__class__.__name__}")
123 
124     def close(self):
125         """Shuts down the thread, its executor and closes the reader (worker thread and buffer)."""
126         if self.closed:  # pragma: no cover
127             return
128 
129         log.debug("Closing writer thread")
130 
131         self.closed = True
132         self.reader.close()
133         self.executor.shutdown(wait=True, cancel_futures=True)
134 
135     def put(self, segment):
136         """Adds a segment to the download pool and write queue."""
137         if self.closed:  # pragma: no cover
138             return
139 
140         if segment is None:
141             future = None
142         else:
143             future = self.executor.submit(self.fetch, segment, retries=self.retries)
144 
145         self.queue(segment, future)
146 
147     def queue(self, segment: Any, future: Optional[Future], *data):
148         """Puts values into a queue but aborts if this thread is closed."""
149         while not self.closed:  # pragma: no branch
150             try:
151                 self._futures_put((segment, future, *data))
152                 return
153             except queue.Full:  # pragma: no cover
154                 continue
155 
156     def _futures_put(self, item):
157         self.futures.put(item, block=True, timeout=1)
158 
159     def _futures_get(self):
160         return self.futures.get(block=True, timeout=0.5)
161 
162     @staticmethod
163     def _future_result(future: Future):
164         return future.result(timeout=0.5)
165 
166     def fetch(self, segment):
167         """Fetches a segment.
168 
169         Should be overridden by the inheriting class.
170         """
171         pass
172 
173     def write(self, segment, result, *data):
174         """Writes a segment to the buffer.
175 
176         Should be overridden by the inheriting class.
177         """
178         pass
179 
180     def run(self):
181         while not self.closed:
182             try:
183                 segment, future, *data = self._futures_get()
184             except queue.Empty:  # pragma: no cover
185                 continue
186 
187             # End of stream
188             if future is None:
189                 break
190 
191             while not self.closed:  # pragma: no branch
192                 try:
193                     result = self._future_result(future)
194                 except futures.TimeoutError:  # pragma: no cover
195                     continue
196                 except futures.CancelledError:  # pragma: no cover
197                     break
198 
199                 if result is not None:  # pragma: no branch
200                     self.write(segment, result, *data)
201 
202                 break
203 
204         self.close()
205 
206 
207 class SegmentedStreamReader(StreamIO):
208     __worker__ = SegmentedStreamWorker
209     __writer__ = SegmentedStreamWriter
210 
211     def __init__(self, stream, timeout=None):
212         super().__init__()
213         self.session = stream.session
214         self.stream = stream
215         self.timeout = timeout or self.session.options.get("stream-timeout")
216 
217         buffer_size = self.session.get_option("ringbuffer-size")
218         self.buffer = RingBuffer(buffer_size)
219         self.writer = self.__writer__(self)
220         self.worker = self.__worker__(self)
221 
222     def open(self):
223         self.writer.start()
224         self.worker.start()
225 
226     def close(self):
227         self.worker.close()
228         self.writer.close()
229         self.buffer.close()
230 
231     def read(self, size):
232         return self.buffer.read(
233             size,
234             block=self.writer.is_alive(),
235             timeout=self.timeout
236         )
237
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/streamlink/stream/segmented.py b/src/streamlink/stream/segmented.py
--- a/src/streamlink/stream/segmented.py
+++ b/src/streamlink/stream/segmented.py
@@ -3,7 +3,7 @@
 from concurrent import futures
 from concurrent.futures import Future, ThreadPoolExecutor
 from sys import version_info
-from threading import Event, Thread
+from threading import Event, Thread, current_thread
 from typing import Any, Optional
 
 from streamlink.buffers import RingBuffer
@@ -228,6 +228,12 @@
         self.writer.close()
         self.buffer.close()
 
+        current = current_thread()
+        if current is not self.worker:  # pragma: no branch
+            self.worker.join(timeout=self.timeout)
+        if current is not self.writer:  # pragma: no branch
+            self.writer.join(timeout=self.timeout)
+
     def read(self, size):
         return self.buffer.read(
             size,

{"golden_diff": "diff --git a/src/streamlink/stream/segmented.py b/src/streamlink/stream/segmented.py\n--- a/src/streamlink/stream/segmented.py\n+++ b/src/streamlink/stream/segmented.py\n@@ -3,7 +3,7 @@\n from concurrent import futures\n from concurrent.futures import Future, ThreadPoolExecutor\n from sys import version_info\n-from threading import Event, Thread\n+from threading import Event, Thread, current_thread\n from typing import Any, Optional\n \n from streamlink.buffers import RingBuffer\n@@ -228,6 +228,12 @@\n         self.writer.close()\n         self.buffer.close()\n \n+        current = current_thread()\n+        if current is not self.worker:  # pragma: no branch\n+            self.worker.join(timeout=self.timeout)\n+        if current is not self.writer:  # pragma: no branch\n+            self.writer.join(timeout=self.timeout)\n+\n     def read(self, size):\n         return self.buffer.read(\n             size,\n", "issue": "SegmentedStreamWriter.close() does not reliably finish before CLI exits (race condition)\n### Checklist\r\n\r\n- [X] This is a bug report and not a different kind of issue\r\n- [X] [I have read the contribution guidelines](https://github.com/streamlink/streamlink/blob/master/CONTRIBUTING.md#contributing-to-streamlink)\r\n- [X] [I have checked the list of open and recently closed bug reports](https://github.com/streamlink/streamlink/issues?q=is%3Aissue+label%3A%22bug%22)\r\n- [X] [I have checked the commit log of the master branch](https://github.com/streamlink/streamlink/commits/master)\r\n\r\n### Streamlink version\r\n\r\nLatest build from the master branch\r\n\r\n### Description\r\n\r\n### Background\r\nI was doing some works on latest `master` branch, adding some feature on Segmented/HLS streams, and it requires some cleanup, so I added them in the end of `SegmentedStreamWriter.close()`, like\r\n\r\n```python\r\n        self.closed = True\r\n        self.reader.close()\r\n        self.executor.shutdown(wait=True, cancel_futures=True)\r\n\r\n        __my_extra_cleanup()\r\n```\r\n\r\nAnd I found my cleanup code does not always being executed, when HLS streams end normally.\r\n\r\n### Problem\r\nWhen an HLS stream ends normally, the whole shutdown process is triggered by the last line of `SegmentedStreamWriter.run()`, `self.close()`. Then `SegmentedStreamWriter.close()` calls `SegmentedStreamReader.close()`, then the iteration loop of `stream_cli.main:read_stream()` will reach its end, then `main()` exits.\r\n\r\nThe problem is, `SegmentedStreamWriter.run() -> SegmentedStreamWriter.close()` was run in a separated thread, which means `SegmentedStreamWriter.close()` cannot reliably finish its work before main thread exits.\r\n\r\n### To reproduce\r\nTo reliably trigger the problem, adding a sleep to original `SegmentedStreamWriter.close()`, like\r\n\r\n```python\r\n        self.closed = True\r\n        self.reader.close()\r\n        self.executor.shutdown(wait=True, cancel_futures=True)\r\n\r\n        time.sleep(3)\r\n        log.debug(\"SegmentedStreamWriter.close() ends\")\r\n```\r\n\r\nThen run the CLI with a short HLS stream, the `SegmentedStreamWriter.close() ends` message never appears.\r\n\r\n\r\n### Debug log\r\n\r\n```text\r\n......\r\n\r\n[cli][info] Stream ended\r\n[cli][info] Closing currently open stream...\r\n```\r\n\r\n### Expected result\r\n```text\r\n......\r\n\r\n[cli][info] Stream ended\r\n[cli][info] Closing currently open stream...\r\n[stream.segmented][debug] SegmentedStreamWriter.close() ends\r\n```\r\n\n", "before_files": [{"content": "import logging\nimport queue\nfrom concurrent import futures\nfrom concurrent.futures import Future, ThreadPoolExecutor\nfrom sys import version_info\nfrom threading import Event, Thread\nfrom typing import Any, Optional\n\nfrom streamlink.buffers import RingBuffer\nfrom streamlink.stream.stream import StreamIO\n\nlog = logging.getLogger(__name__)\n\n\nclass CompatThreadPoolExecutor(ThreadPoolExecutor):\n    if version_info < (3, 9):\n        def shutdown(self, wait=True, cancel_futures=False):  # pragma: no cover\n            with self._shutdown_lock:\n                self._shutdown = True\n                if cancel_futures:\n                    # Drain all work items from the queue, and then cancel their\n                    # associated futures.\n                    while True:\n                        try:\n                            work_item = self._work_queue.get_nowait()\n                        except queue.Empty:\n                            break\n                        if work_item is not None:\n                            work_item.future.cancel()\n\n                # Send a wake-up to prevent threads calling\n                # _work_queue.get(block=True) from permanently blocking.\n                self._work_queue.put(None)\n            if wait:\n                for t in self._threads:\n                    t.join()\n\n\nclass SegmentedStreamWorker(Thread):\n    \"\"\"The general worker thread.\n\n    This thread is responsible for queueing up segments in the\n    writer thread.\n    \"\"\"\n\n    def __init__(self, reader, **kwargs):\n        self.closed = False\n        self.reader = reader\n        self.writer = reader.writer\n        self.stream = reader.stream\n        self.session = reader.session\n\n        self._wait = Event()\n\n        super().__init__(daemon=True, name=f\"Thread-{self.__class__.__name__}\")\n\n    def close(self):\n        \"\"\"Shuts down the thread.\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        log.debug(\"Closing worker thread\")\n\n        self.closed = True\n        self._wait.set()\n\n    def wait(self, time):\n        \"\"\"Pauses the thread for a specified time.\n\n        Returns False if interrupted by another thread and True if the\n        time runs out normally.\n        \"\"\"\n        return not self._wait.wait(time)\n\n    def iter_segments(self):\n        \"\"\"The iterator that generates segments for the worker thread.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        return\n        yield\n\n    def run(self):\n        for segment in self.iter_segments():\n            if self.closed:  # pragma: no cover\n                break\n            self.writer.put(segment)\n\n        # End of stream, tells the writer to exit\n        self.writer.put(None)\n        self.close()\n\n\nclass SegmentedStreamWriter(Thread):\n    \"\"\"The writer thread.\n\n    This thread is responsible for fetching segments, processing them\n    and finally writing the data to the buffer.\n    \"\"\"\n\n    def __init__(self, reader, size=20, retries=None, threads=None, timeout=None):\n        self.closed = False\n        self.reader = reader\n        self.stream = reader.stream\n        self.session = reader.session\n\n        if not retries:\n            retries = self.session.options.get(\"stream-segment-attempts\")\n\n        if not threads:\n            threads = self.session.options.get(\"stream-segment-threads\")\n\n        if not timeout:\n            timeout = self.session.options.get(\"stream-segment-timeout\")\n\n        self.retries = retries\n        self.timeout = timeout\n        self.threads = threads\n        self.executor = CompatThreadPoolExecutor(max_workers=self.threads)\n        self.futures = queue.Queue(size)\n\n        super().__init__(daemon=True, name=f\"Thread-{self.__class__.__name__}\")\n\n    def close(self):\n        \"\"\"Shuts down the thread, its executor and closes the reader (worker thread and buffer).\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        log.debug(\"Closing writer thread\")\n\n        self.closed = True\n        self.reader.close()\n        self.executor.shutdown(wait=True, cancel_futures=True)\n\n    def put(self, segment):\n        \"\"\"Adds a segment to the download pool and write queue.\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        if segment is None:\n            future = None\n        else:\n            future = self.executor.submit(self.fetch, segment, retries=self.retries)\n\n        self.queue(segment, future)\n\n    def queue(self, segment: Any, future: Optional[Future], *data):\n        \"\"\"Puts values into a queue but aborts if this thread is closed.\"\"\"\n        while not self.closed:  # pragma: no branch\n            try:\n                self._futures_put((segment, future, *data))\n                return\n            except queue.Full:  # pragma: no cover\n                continue\n\n    def _futures_put(self, item):\n        self.futures.put(item, block=True, timeout=1)\n\n    def _futures_get(self):\n        return self.futures.get(block=True, timeout=0.5)\n\n    @staticmethod\n    def _future_result(future: Future):\n        return future.result(timeout=0.5)\n\n    def fetch(self, segment):\n        \"\"\"Fetches a segment.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        pass\n\n    def write(self, segment, result, *data):\n        \"\"\"Writes a segment to the buffer.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        pass\n\n    def run(self):\n        while not self.closed:\n            try:\n                segment, future, *data = self._futures_get()\n            except queue.Empty:  # pragma: no cover\n                continue\n\n            # End of stream\n            if future is None:\n                break\n\n            while not self.closed:  # pragma: no branch\n                try:\n                    result = self._future_result(future)\n                except futures.TimeoutError:  # pragma: no cover\n                    continue\n                except futures.CancelledError:  # pragma: no cover\n                    break\n\n                if result is not None:  # pragma: no branch\n                    self.write(segment, result, *data)\n\n                break\n\n        self.close()\n\n\nclass SegmentedStreamReader(StreamIO):\n    __worker__ = SegmentedStreamWorker\n    __writer__ = SegmentedStreamWriter\n\n    def __init__(self, stream, timeout=None):\n        super().__init__()\n        self.session = stream.session\n        self.stream = stream\n        self.timeout = timeout or self.session.options.get(\"stream-timeout\")\n\n        buffer_size = self.session.get_option(\"ringbuffer-size\")\n        self.buffer = RingBuffer(buffer_size)\n        self.writer = self.__writer__(self)\n        self.worker = self.__worker__(self)\n\n    def open(self):\n        self.writer.start()\n        self.worker.start()\n\n    def close(self):\n        self.worker.close()\n        self.writer.close()\n        self.buffer.close()\n\n    def read(self, size):\n        return self.buffer.read(\n            size,\n            block=self.writer.is_alive(),\n            timeout=self.timeout\n        )\n", "path": "src/streamlink/stream/segmented.py"}], "after_files": [{"content": "import logging\nimport queue\nfrom concurrent import futures\nfrom concurrent.futures import Future, ThreadPoolExecutor\nfrom sys import version_info\nfrom threading import Event, Thread, current_thread\nfrom typing import Any, Optional\n\nfrom streamlink.buffers import RingBuffer\nfrom streamlink.stream.stream import StreamIO\n\nlog = logging.getLogger(__name__)\n\n\nclass CompatThreadPoolExecutor(ThreadPoolExecutor):\n    if version_info < (3, 9):\n        def shutdown(self, wait=True, cancel_futures=False):  # pragma: no cover\n            with self._shutdown_lock:\n                self._shutdown = True\n                if cancel_futures:\n                    # Drain all work items from the queue, and then cancel their\n                    # associated futures.\n                    while True:\n                        try:\n                            work_item = self._work_queue.get_nowait()\n                        except queue.Empty:\n                            break\n                        if work_item is not None:\n                            work_item.future.cancel()\n\n                # Send a wake-up to prevent threads calling\n                # _work_queue.get(block=True) from permanently blocking.\n                self._work_queue.put(None)\n            if wait:\n                for t in self._threads:\n                    t.join()\n\n\nclass SegmentedStreamWorker(Thread):\n    \"\"\"The general worker thread.\n\n    This thread is responsible for queueing up segments in the\n    writer thread.\n    \"\"\"\n\n    def __init__(self, reader, **kwargs):\n        self.closed = False\n        self.reader = reader\n        self.writer = reader.writer\n        self.stream = reader.stream\n        self.session = reader.session\n\n        self._wait = Event()\n\n        super().__init__(daemon=True, name=f\"Thread-{self.__class__.__name__}\")\n\n    def close(self):\n        \"\"\"Shuts down the thread.\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        log.debug(\"Closing worker thread\")\n\n        self.closed = True\n        self._wait.set()\n\n    def wait(self, time):\n        \"\"\"Pauses the thread for a specified time.\n\n        Returns False if interrupted by another thread and True if the\n        time runs out normally.\n        \"\"\"\n        return not self._wait.wait(time)\n\n    def iter_segments(self):\n        \"\"\"The iterator that generates segments for the worker thread.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        return\n        yield\n\n    def run(self):\n        for segment in self.iter_segments():\n            if self.closed:  # pragma: no cover\n                break\n            self.writer.put(segment)\n\n        # End of stream, tells the writer to exit\n        self.writer.put(None)\n        self.close()\n\n\nclass SegmentedStreamWriter(Thread):\n    \"\"\"The writer thread.\n\n    This thread is responsible for fetching segments, processing them\n    and finally writing the data to the buffer.\n    \"\"\"\n\n    def __init__(self, reader, size=20, retries=None, threads=None, timeout=None):\n        self.closed = False\n        self.reader = reader\n        self.stream = reader.stream\n        self.session = reader.session\n\n        if not retries:\n            retries = self.session.options.get(\"stream-segment-attempts\")\n\n        if not threads:\n            threads = self.session.options.get(\"stream-segment-threads\")\n\n        if not timeout:\n            timeout = self.session.options.get(\"stream-segment-timeout\")\n\n        self.retries = retries\n        self.timeout = timeout\n        self.threads = threads\n        self.executor = CompatThreadPoolExecutor(max_workers=self.threads)\n        self.futures = queue.Queue(size)\n\n        super().__init__(daemon=True, name=f\"Thread-{self.__class__.__name__}\")\n\n    def close(self):\n        \"\"\"Shuts down the thread, its executor and closes the reader (worker thread and buffer).\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        log.debug(\"Closing writer thread\")\n\n        self.closed = True\n        self.reader.close()\n        self.executor.shutdown(wait=True, cancel_futures=True)\n\n    def put(self, segment):\n        \"\"\"Adds a segment to the download pool and write queue.\"\"\"\n        if self.closed:  # pragma: no cover\n            return\n\n        if segment is None:\n            future = None\n        else:\n            future = self.executor.submit(self.fetch, segment, retries=self.retries)\n\n        self.queue(segment, future)\n\n    def queue(self, segment: Any, future: Optional[Future], *data):\n        \"\"\"Puts values into a queue but aborts if this thread is closed.\"\"\"\n        while not self.closed:  # pragma: no branch\n            try:\n                self._futures_put((segment, future, *data))\n                return\n            except queue.Full:  # pragma: no cover\n                continue\n\n    def _futures_put(self, item):\n        self.futures.put(item, block=True, timeout=1)\n\n    def _futures_get(self):\n        return self.futures.get(block=True, timeout=0.5)\n\n    @staticmethod\n    def _future_result(future: Future):\n        return future.result(timeout=0.5)\n\n    def fetch(self, segment):\n        \"\"\"Fetches a segment.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        pass\n\n    def write(self, segment, result, *data):\n        \"\"\"Writes a segment to the buffer.\n\n        Should be overridden by the inheriting class.\n        \"\"\"\n        pass\n\n    def run(self):\n        while not self.closed:\n            try:\n                segment, future, *data = self._futures_get()\n            except queue.Empty:  # pragma: no cover\n                continue\n\n            # End of stream\n            if future is None:\n                break\n\n            while not self.closed:  # pragma: no branch\n                try:\n                    result = self._future_result(future)\n                except futures.TimeoutError:  # pragma: no cover\n                    continue\n                except futures.CancelledError:  # pragma: no cover\n                    break\n\n                if result is not None:  # pragma: no branch\n                    self.write(segment, result, *data)\n\n                break\n\n        self.close()\n\n\nclass SegmentedStreamReader(StreamIO):\n    __worker__ = SegmentedStreamWorker\n    __writer__ = SegmentedStreamWriter\n\n    def __init__(self, stream, timeout=None):\n        super().__init__()\n        self.session = stream.session\n        self.stream = stream\n        self.timeout = timeout or self.session.options.get(\"stream-timeout\")\n\n        buffer_size = self.session.get_option(\"ringbuffer-size\")\n        self.buffer = RingBuffer(buffer_size)\n        self.writer = self.__writer__(self)\n        self.worker = self.__worker__(self)\n\n    def open(self):\n        self.writer.start()\n        self.worker.start()\n\n    def close(self):\n        self.worker.close()\n        self.writer.close()\n        self.buffer.close()\n\n        current = current_thread()\n        if current is not self.worker:  # pragma: no branch\n            self.worker.join(timeout=self.timeout)\n        if current is not self.writer:  # pragma: no branch\n            self.writer.join(timeout=self.timeout)\n\n    def read(self, size):\n        return self.buffer.read(\n            size,\n            block=self.writer.is_alive(),\n            timeout=self.timeout\n        )\n", "path": "src/streamlink/stream/segmented.py"}]}