/
/
/
1"""Simple async-friendly named pipe reader/writer using threads."""
2
3from __future__ import annotations
4
5import asyncio
6import errno as errno_module
7import logging
8import os
9import select
10import time
11from collections.abc import AsyncGenerator
12from contextlib import suppress
13from functools import partial
14from pathlib import Path
15
16_LOGGER = logging.getLogger("named_pipe")
17
18# How long a write waits on a full pipe buffer that the reader never drains.
19WRITE_STALL_TIMEOUT = 2.0
20# Upper bound for a single write, so a barely-moving reader cannot park it for minutes.
21WRITE_TOTAL_TIMEOUT = 30.0
22# Length of one wait slice, which caps how long a reader that left goes unnoticed.
23WRITE_POLL_INTERVAL_MS = 250
24
25
26class AsyncNamedPipeWriter:
27 """Async writer for named pipes."""
28
29 def __init__(self, pipe_path: str, owner_id: str | None = None) -> None:
30 """
31 Initialize named pipe writer.
32
33 :param pipe_path: Filesystem path of the named pipe.
34 :param owner_id: Optional identifier (e.g. player_id) included in log
35 messages so silent failures can be correlated to a specific device.
36 """
37 self._pipe_path = pipe_path
38 self._owner_id = owner_id
39 self._write_fd: int | None = None
40 self._write_lock = asyncio.Lock()
41
42 @property
43 def path(self) -> str:
44 """Return the named pipe path."""
45 return self._pipe_path
46
47 async def create(self) -> None:
48 """Create the named pipe."""
49
50 def _create() -> None:
51 pipe_path = Path(self._pipe_path)
52 if pipe_path.exists():
53 pipe_path.unlink()
54 os.mkfifo(self._pipe_path)
55
56 await asyncio.to_thread(_create)
57
58 async def wait_for_reader(self, timeout: float) -> bool:
59 """
60 Wait until the pipe has a reader attached, so writes are no longer dropped.
61
62 A pipe without a reader accepts nothing, so a writer that is spawning its
63 reader alongside itself waits here before its first write.
64
65 :param timeout: Maximum time to wait for the reader in seconds.
66 :return: True once the pipe can be written to, False if no reader
67 attached before the timeout.
68 """
69 try:
70 async with asyncio.timeout(timeout):
71 while True:
72 # a concurrent write opens the same descriptor from its worker thread
73 async with self._write_lock:
74 if self._ensure_write_fd():
75 return True
76 await asyncio.sleep(0.05)
77 except TimeoutError:
78 return False
79
80 async def write(self, data: bytes) -> bool:
81 """
82 Write data to the named pipe.
83
84 :param data: Data to write.
85 :return: True for a complete write, False when no reader is available,
86 the reader closes, or the write cannot make progress.
87 :raises OSError: If writing fails for another reason.
88 """
89
90 def _write() -> bool:
91 if not self._ensure_write_fd():
92 _LOGGER.debug(
93 "Named pipe write failed: no writable fd for pipe %s (owner=%s, %d bytes dropped)",
94 self._pipe_path,
95 self._log_owner,
96 len(data),
97 )
98 return False
99 # hold on to the descriptor a concurrent remove() may swap out from under us
100 write_fd = self._write_fd
101 if write_fd is None:
102 return False
103 data_view = memoryview(data)
104 total_bytes_written = 0
105 give_up_at = time.monotonic() + WRITE_TOTAL_TIMEOUT
106 stall_ends_at: float | None = None
107 try:
108 while total_bytes_written < len(data_view):
109 # a cancelled write releases the lock but keeps this thread going,
110 # so stop once remove() has taken the descriptor
111 if self._write_fd != write_fd:
112 return False
113 try:
114 bytes_written = os.write(write_fd, data_view[total_bytes_written:])
115 except BlockingIOError:
116 # A full buffer means the reader is behind, not gone, so wait for it
117 # to drain and try again. Only the write itself tells the two apart:
118 # a full pipe whose reader left is not reported as broken everywhere,
119 # so the wait is sliced rather than trusted to end on its own.
120 now = time.monotonic()
121 if stall_ends_at is None:
122 stall_ends_at = now + WRITE_STALL_TIMEOUT
123 if now < stall_ends_at and now < give_up_at:
124 self._wait_writable(write_fd)
125 continue
126 _LOGGER.debug(
127 "Named pipe write stalled on %s "
128 "(owner=%s, %d of %d bytes written): reader is not draining",
129 self._pipe_path,
130 self._log_owner,
131 total_bytes_written,
132 len(data),
133 )
134 return False
135 stall_ends_at = None
136 if bytes_written == 0:
137 _LOGGER.debug(
138 "Named pipe write made no progress on %s "
139 "(owner=%s, %d of %d bytes written)",
140 self._pipe_path,
141 self._log_owner,
142 total_bytes_written,
143 len(data),
144 )
145 return False
146 total_bytes_written += bytes_written
147 return True
148 except OSError as e:
149 if e.errno == errno_module.EPIPE:
150 # Reader closed, reset fd for next attempt. A concurrent remove()
151 # may already have replaced it, and then owns it instead.
152 if self._write_fd == write_fd:
153 with suppress(Exception):
154 os.close(write_fd)
155 self._write_fd = None
156 _LOGGER.debug(
157 "Named pipe write failed (EPIPE) on %s "
158 "(owner=%s, %d of %d bytes written): reader closed",
159 self._pipe_path,
160 self._log_owner,
161 total_bytes_written,
162 len(data),
163 )
164 return False
165 raise
166
167 async with self._write_lock:
168 return await asyncio.to_thread(_write)
169
170 async def remove(self) -> None:
171 """Close write fd and remove the pipe."""
172 # the lock keeps a write in flight on its worker thread from reopening
173 # the descriptor between the close and the unlink
174 async with self._write_lock:
175 if self._write_fd is not None:
176 with suppress(Exception):
177 os.close(self._write_fd)
178 self._write_fd = None
179 pipe_path = Path(self._pipe_path)
180 if pipe_path.exists():
181 with suppress(Exception):
182 pipe_path.unlink()
183
184 def __str__(self) -> str:
185 """Return string representation."""
186 return self._pipe_path
187
188 @property
189 def _log_owner(self) -> str:
190 """Return a short descriptor for logging (owner_id or pipe path)."""
191 return self._owner_id or self._pipe_path
192
193 def _wait_writable(self, write_fd: int) -> None:
194 """
195 Wait a short while for the pipe to accept data again.
196
197 :param write_fd: Descriptor of the pipe's write end.
198 """
199 # poll() rather than select(), which rejects a descriptor of 1024 or above
200 poller = select.poll()
201 poller.register(write_fd, select.POLLOUT)
202 poller.poll(WRITE_POLL_INTERVAL_MS)
203
204 def _ensure_write_fd(self) -> bool:
205 """Open the write end while a reader is attached. Returns True if successful."""
206 if self._write_fd is not None:
207 return True
208 if not Path(self._pipe_path).exists():
209 return False
210 try:
211 self._write_fd = os.open(self._pipe_path, os.O_WRONLY | os.O_NONBLOCK)
212 except OSError as e:
213 if e.errno in (errno_module.ENXIO, errno_module.ENOENT):
214 return False
215 raise
216 return True
217
218
219async def read_named_pipe(
220 pipe_path: str,
221 chunk_size: int = 4096,
222) -> AsyncGenerator[bytes]:
223 """
224 Read raw bytes from a named pipe (FIFO) as an async generator.
225
226 Suspends while the upstream writer is idle and transparently reopens the
227 pipe on writer disconnect so an external-process restart doesn't tear down
228 the consumer.
229
230 :param pipe_path: Filesystem path of the named pipe.
231 :param chunk_size: Maximum bytes returned per yield.
232 """
233 loop = asyncio.get_running_loop()
234 while True:
235 fd = os.open(pipe_path, os.O_RDONLY | os.O_NONBLOCK)
236 try:
237 pipe_file = os.fdopen(fd, "rb", buffering=0)
238 except OSError:
239 os.close(fd)
240 raise
241 # Small StreamReader limit so back-pressure kicks in quickly when the
242 # producer writes faster than realtime (e.g. librespot's pipe backend
243 # which is not natively rate-limited). asyncio's default is 64 KiB.
244 # 32 KiB caps the in-flight backlog at ~180 ms at 44.1 kHz s16 stereo
245 # without being so tight it risks dropping packets from realtime-paced
246 # producers (shairport-sync etc.) under brief consumer-side jitter.
247 reader = asyncio.StreamReader(limit=32768)
248 try:
249 transport, _ = await loop.connect_read_pipe(
250 partial(asyncio.StreamReaderProtocol, reader),
251 pipe_file,
252 )
253 except BaseException:
254 pipe_file.close()
255 raise
256 try:
257 while True:
258 data = await reader.read(chunk_size)
259 if not data:
260 break
261 yield data
262 finally:
263 transport.close()
264 # avoid a tight reopen loop when no writer is present
265 await asyncio.sleep(0.1)
266