From afa70013db73c3068f910fc8cef5c01206483012 Mon Sep 17 00:00:00 2001 From: George Panchuk Date: Sat, 26 Sep 2026 21:02:28 +0700 Subject: [PATCH 1/2] fix: improve parallel processes cleanup --- fastembed/parallel_processor.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py index ffb3d6ec..a8c00c28 100644 --- a/fastembed/parallel_processor.py +++ b/fastembed/parallel_processor.py @@ -215,10 +215,14 @@ def semi_ordered_map( # join would wait forever for processes blocked on the input queue. self.emergency_shutdown = True self.join_or_terminate() - # Nothing reads the input pipe anymore, so the feeder thread can be stuck writing to it, - # holding every batch it hasn't sent. Closing our read end fails that write with EPIPE and - # lets the thread exit, same as Queue._terminate_broken() in python 3.12+. - self.input_queue._reader.close() # type: ignore[attr-defined] + # Nothing reads the input pipe anymore, so the feeder thread can be stuck + # writing to it, holding every item it hasn't sent. On POSIX, closing our + # read end fails that write with EPIPE; Windows also needs to cancel the send. + terminate_broken = getattr(self.input_queue, "_terminate_broken", None) + if terminate_broken is not None: + terminate_broken() + else: + self.input_queue._reader.close() self.input_queue.close() self.output_queue.close() if self.emergency_shutdown: From b78456011734a1ce1a0df4a9757e8b48edd12b32 Mon Sep 17 00:00:00 2001 From: George Panchuk Date: Sat, 26 Sep 2026 21:05:18 +0700 Subject: [PATCH 2/2] fix: fix mypy --- fastembed/parallel_processor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py index a8c00c28..36688d87 100644 --- a/fastembed/parallel_processor.py +++ b/fastembed/parallel_processor.py @@ -222,7 +222,7 @@ def semi_ordered_map( if terminate_broken is not None: terminate_broken() else: - self.input_queue._reader.close() + self.input_queue._reader.close() # type: ignore[attr-defined] self.input_queue.close() self.output_queue.close() if self.emergency_shutdown: