File size: 2,406 Bytes
abd39d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | """Audio consumer queue, bounded backlog, multi-worker support, and deterministic shutdown.
Only the translation producer mutates the book scope. The job stays active until
close() joins the consumer workers, so reset/delete/rename cannot race audio output.
"""
import queue
import threading
class AudioQueue:
def __init__(self, consume, checkpoint, cancel, maxsize=2, workers=1):
self.consume = consume
self.checkpoint = checkpoint
self.cancel = cancel
self.queue = queue.Queue(maxsize=maxsize)
self.error = None
self.closed = False
self.workers = []
for i in range(max(1, int(workers))):
t = threading.Thread(target=self._run, name=f'chapter-audio-{i}', daemon=True)
t.start()
self.workers.append(t)
@property
def worker(self):
"""Backward compatibility for checks like `q.worker.is_alive()`."""
return self.workers[0] if self.workers else None
def submit(self, item):
if self.closed:
raise RuntimeError('Audio queue is closed')
while any(w.is_alive() for w in self.workers):
if self.error:
raise RuntimeError('Audio worker failed') from self.error
if not self.checkpoint():
self.cancel(item)
return False
try:
self.queue.put(item, timeout=0.2)
return True
except queue.Full:
continue
raise RuntimeError('Audio worker stopped unexpectedly') from self.error
def _run(self):
while True:
item = self.queue.get()
try:
if item is None:
return
if self.error is not None or not self.checkpoint():
self.cancel(item)
else:
self.consume(item)
except Exception as exc:
# Do not abandon a full queue and deadlock its producer.
self.error = exc
finally:
self.queue.task_done()
def close(self):
if not self.closed:
self.closed = True
for _ in self.workers:
self.queue.put(None)
for w in self.workers:
w.join()
if self.error:
raise RuntimeError('Audio worker failed') from self.error
|