From 24ca3e08437523be3e2ca174dc82ee3f95900262 Mon Sep 17 00:00:00 2001
From: FU-max-boop <214359569+FU-max-boop@users.noreply.github.com>
Date: Tue, 11 Aug 2026 12:03:22 +0800
Subject: [PATCH 1/5] fix: terminate interrupted parallel workers

---
 fastembed/parallel_processor.py  | 12 +++++-
 tests/test_parallel_processor.py | 73 ++++++++++++++++++++++++++++++++
 2 files changed, 84 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_parallel_processor.py

diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py
index bfaaf0c54..90e3de811 100644
--- a/fastembed/parallel_processor.py
+++ b/fastembed/parallel_processor.py
@@ -153,6 +153,7 @@ def ordered_map(self, stream: Iterable[Any], *args: Any, **kwargs: Any) -> Itera
     def semi_ordered_map(
         self, stream: Iterable[Any], *args: Any, **kwargs: Any
     ) -> Iterable[tuple[int, Any]]:
+        completed = False
         try:
             self.start(**kwargs)
 
@@ -196,10 +197,19 @@ def semi_ordered_map(
                     raise RuntimeError("Thread unexpectedly terminated")
                 yield out_item
                 read += 1
+            completed = True
         finally:
             assert self.input_queue is not None, "Input queue is None"
             assert self.output_queue is not None, "Output queue is None"
-            self.join()
+            if completed:
+                self.join()
+            else:
+                # The generator may be closed before the input stream is exhausted (for
+                # example, when a caller only consumes a prefix of the embeddings). In that
+                # case workers have not necessarily received their stop signals and a normal
+                # join would wait forever for processes blocked on the input queue.
+                self.emergency_shutdown = True
+                self.join_or_terminate()
             self.input_queue.close()
             self.output_queue.close()
             if self.emergency_shutdown:
diff --git a/tests/test_parallel_processor.py b/tests/test_parallel_processor.py
new file mode 100644
index 000000000..695123d82
--- /dev/null
+++ b/tests/test_parallel_processor.py
@@ -0,0 +1,73 @@
+from queue import Empty
+from unittest.mock import MagicMock
+
+import pytest
+
+from fastembed.parallel_processor import ParallelWorkerPool, Worker
+
+
+class EchoWorker(Worker):
+    @classmethod
+    def start(cls, **kwargs):
+        return cls()
+
+    def process(self, items):
+        yield from items
+
+
+def make_stubbed_pool(monkeypatch):
+    pool = ParallelWorkerPool(1, EchoWorker)
+    input_queue = MagicMock()
+    output_queue = MagicMock()
+    output_queue.get_nowait.side_effect = Empty
+    output_queue.get.return_value = (0, "result")
+    process = MagicMock()
+    process.is_alive.return_value = True
+
+    def start(**kwargs):
+        pool.input_queue = input_queue
+        pool.output_queue = output_queue
+        pool.processes = [process]
+
+    monkeypatch.setattr(pool, "start", start)
+    return pool, input_queue, output_queue, process
+
+
+def test_closing_partial_parallel_iterator_terminates_workers(monkeypatch):
+    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
+    results = pool.ordered_map(["input"])
+
+    assert next(results) == "result"
+    results.close()
+
+    process.join.assert_called_once_with(timeout=1)
+    process.terminate.assert_called_once_with()
+    input_queue.cancel_join_thread.assert_called_once_with()
+    output_queue.cancel_join_thread.assert_called_once_with()
+
+
+def test_parallel_iterator_terminates_workers_when_input_raises(monkeypatch):
+    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
+
+    def failing_stream():
+        yield "input"
+        raise RuntimeError("stream failed")
+
+    with pytest.raises(RuntimeError, match="stream failed"):
+        list(pool.semi_ordered_map(failing_stream()))
+
+    process.join.assert_called_once_with(timeout=1)
+    process.terminate.assert_called_once_with()
+    input_queue.cancel_join_thread.assert_called_once_with()
+    output_queue.cancel_join_thread.assert_called_once_with()
+
+
+def test_exhausted_parallel_iterator_joins_workers_gracefully(monkeypatch):
+    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
+
+    assert list(pool.semi_ordered_map(["input"])) == [(0, "result")]
+
+    process.join.assert_called_once_with()
+    process.terminate.assert_not_called()
+    input_queue.join_thread.assert_called_once_with()
+    output_queue.join_thread.assert_called_once_with()

From 02394b39b6d464fb2177bbcbd9ac919d3336a69a Mon Sep 17 00:00:00 2001
From: FU-max-boop <214359569+FU-max-boop@users.noreply.github.com>
Date: Tue, 11 Aug 2026 15:59:10 +0800
Subject: [PATCH 2/5] fix: reap terminated parallel workers

---
 fastembed/parallel_processor.py  |  2 ++
 tests/test_parallel_processor.py | 51 ++++++++++++++++++++++++++++++--
 2 files changed, 50 insertions(+), 3 deletions(-)

diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py
index 90e3de811..ff84b1f47 100644
--- a/fastembed/parallel_processor.py
+++ b/fastembed/parallel_processor.py
@@ -111,6 +111,7 @@ def __init__(
         self.num_active_workers: BaseValue | None = None
 
     def start(self, **kwargs: Any) -> None:
+        self.emergency_shutdown = False
         self.input_queue = self.ctx.Queue(self.queue_size)
         self.output_queue = self.ctx.Queue(self.queue_size)
 
@@ -241,6 +242,7 @@ def join_or_terminate(self, timeout: int = 1) -> None:
             process.join(timeout=timeout)
             if process.is_alive():
                 process.terminate()
+                process.join(timeout=timeout)
         self.processes.clear()
 
     def join(self) -> None:
diff --git a/tests/test_parallel_processor.py b/tests/test_parallel_processor.py
index 695123d82..5f781549a 100644
--- a/tests/test_parallel_processor.py
+++ b/tests/test_parallel_processor.py
@@ -1,5 +1,6 @@
+from multiprocessing import get_context
 from queue import Empty
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, call
 
 import pytest
 
@@ -40,7 +41,7 @@ def test_closing_partial_parallel_iterator_terminates_workers(monkeypatch):
     assert next(results) == "result"
     results.close()
 
-    process.join.assert_called_once_with(timeout=1)
+    assert process.join.call_args_list == [call(timeout=1), call(timeout=1)]
     process.terminate.assert_called_once_with()
     input_queue.cancel_join_thread.assert_called_once_with()
     output_queue.cancel_join_thread.assert_called_once_with()
@@ -56,7 +57,7 @@ def failing_stream():
     with pytest.raises(RuntimeError, match="stream failed"):
         list(pool.semi_ordered_map(failing_stream()))
 
-    process.join.assert_called_once_with(timeout=1)
+    assert process.join.call_args_list == [call(timeout=1), call(timeout=1)]
     process.terminate.assert_called_once_with()
     input_queue.cancel_join_thread.assert_called_once_with()
     output_queue.cancel_join_thread.assert_called_once_with()
@@ -71,3 +72,47 @@ def test_exhausted_parallel_iterator_joins_workers_gracefully(monkeypatch):
     process.terminate.assert_not_called()
     input_queue.join_thread.assert_called_once_with()
     output_queue.join_thread.assert_called_once_with()
+
+
+def test_reused_pool_returns_to_graceful_queue_cleanup():
+    pool = ParallelWorkerPool(1, EchoWorker)
+    ctx = MagicMock()
+    pool.ctx = ctx
+
+    first_input_queue = MagicMock()
+    first_output_queue = MagicMock()
+    first_output_queue.get_nowait.return_value = (0, "first result")
+    second_input_queue = MagicMock()
+    second_output_queue = MagicMock()
+    second_output_queue.get_nowait.return_value = (0, "second result")
+    first_process = MagicMock()
+    first_process.is_alive.return_value = True
+    second_process = MagicMock()
+
+    ctx.Queue.side_effect = [
+        first_input_queue,
+        first_output_queue,
+        second_input_queue,
+        second_output_queue,
+    ]
+    ctx.Value.side_effect = [
+        get_context().Value("i", 1),
+        get_context().Value("i", 1),
+    ]
+    ctx.Process.side_effect = [first_process, second_process]
+
+    first_results = pool.ordered_map(["first input"])
+    assert next(first_results) == "first result"
+    first_results.close()
+
+    assert pool.emergency_shutdown is True
+    first_input_queue.cancel_join_thread.assert_called_once_with()
+    first_output_queue.cancel_join_thread.assert_called_once_with()
+
+    assert list(pool.semi_ordered_map(["second input"])) == [(0, "second result")]
+
+    assert pool.emergency_shutdown is False
+    second_input_queue.join_thread.assert_called_once_with()
+    second_output_queue.join_thread.assert_called_once_with()
+    second_input_queue.cancel_join_thread.assert_not_called()
+    second_output_queue.cancel_join_thread.assert_not_called()

From f1f77fee10d1722056cd0401f5fb1042e668005d Mon Sep 17 00:00:00 2001
From: FU-max-boop <214359569+FU-max-boop@users.noreply.github.com>
Date: Tue, 11 Aug 2026 18:06:31 +0800
Subject: [PATCH 3/5] test: cover worker cleanup after pool reuse

---
 tests/test_parallel_processor.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/tests/test_parallel_processor.py b/tests/test_parallel_processor.py
index 5f781549a..7ee2271a5 100644
--- a/tests/test_parallel_processor.py
+++ b/tests/test_parallel_processor.py
@@ -106,6 +106,8 @@ def test_reused_pool_returns_to_graceful_queue_cleanup():
     first_results.close()
 
     assert pool.emergency_shutdown is True
+    assert first_process.join.call_args_list == [call(timeout=1), call(timeout=1)]
+    first_process.terminate.assert_called_once_with()
     first_input_queue.cancel_join_thread.assert_called_once_with()
     first_output_queue.cancel_join_thread.assert_called_once_with()
 

From 9a35d85744e71745358dbd102780bf847c266349 Mon Sep 17 00:00:00 2001
From: George Panchuk <george.panchuk@qdrant.tech>
Date: Sat, 26 Sep 2026 18:07:30 +0700
Subject: [PATCH 4/5] fix: bound parallel worker shutdown and free unsent input
 batches

---
 fastembed/parallel_processor.py  |  18 ++++-
 tests/test_parallel_processor.py | 130 ++++++-------------------------
 2 files changed, 40 insertions(+), 108 deletions(-)

diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py
index ff84b1f47..9791fcb3e 100644
--- a/fastembed/parallel_processor.py
+++ b/fastembed/parallel_processor.py
@@ -1,5 +1,6 @@
 import logging
 import os
+import time
 from collections import defaultdict
 from copy import deepcopy
 from enum import Enum
@@ -111,8 +112,10 @@ def __init__(
         self.num_active_workers: BaseValue | None = None
 
     def start(self, **kwargs: Any) -> None:
-        self.emergency_shutdown = False
         self.input_queue = self.ctx.Queue(self.queue_size)
+        # An emergency shutdown unblocks the feeder thread with EPIPE (see semi_ordered_map), let it
+        # exit quietly instead of printing a traceback. ProcessPoolExecutor does the same.
+        self.input_queue._ignore_epipe = True  # type: ignore[attr-defined]
         self.output_queue = self.ctx.Queue(self.queue_size)
 
         ctx_value = self.ctx.Value("i", self.num_workers)
@@ -211,6 +214,10 @@ 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]
             self.input_queue.close()
             self.output_queue.close()
             if self.emergency_shutdown:
@@ -238,11 +245,16 @@ def join_or_terminate(self, timeout: int = 1) -> None:
         @param timeout:
         @return:
         """
+        # One deadline for the whole pool: workers that can't finish (e.g. after an early close)
+        # would otherwise each use up the full timeout, one after another.
+        deadline = time.monotonic() + timeout
+        for process in self.processes:
+            process.join(timeout=max(0.0, deadline - time.monotonic()))
         for process in self.processes:
-            process.join(timeout=timeout)
             if process.is_alive():
                 process.terminate()
-                process.join(timeout=timeout)
+        for process in self.processes:
+            process.join(timeout=timeout)
         self.processes.clear()
 
     def join(self) -> None:
diff --git a/tests/test_parallel_processor.py b/tests/test_parallel_processor.py
index 7ee2271a5..c9d4fb798 100644
--- a/tests/test_parallel_processor.py
+++ b/tests/test_parallel_processor.py
@@ -1,6 +1,6 @@
-from multiprocessing import get_context
-from queue import Empty
-from unittest.mock import MagicMock, call
+import threading
+from itertools import count
+from multiprocessing import get_all_start_methods
 
 import pytest
 
@@ -16,105 +16,25 @@ def process(self, items):
         yield from items
 
 
-def make_stubbed_pool(monkeypatch):
-    pool = ParallelWorkerPool(1, EchoWorker)
-    input_queue = MagicMock()
-    output_queue = MagicMock()
-    output_queue.get_nowait.side_effect = Empty
-    output_queue.get.return_value = (0, "result")
-    process = MagicMock()
-    process.is_alive.return_value = True
-
-    def start(**kwargs):
-        pool.input_queue = input_queue
-        pool.output_queue = output_queue
-        pool.processes = [process]
-
-    monkeypatch.setattr(pool, "start", start)
-    return pool, input_queue, output_queue, process
-
-
-def test_closing_partial_parallel_iterator_terminates_workers(monkeypatch):
-    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
-    results = pool.ordered_map(["input"])
-
-    assert next(results) == "result"
-    results.close()
-
-    assert process.join.call_args_list == [call(timeout=1), call(timeout=1)]
-    process.terminate.assert_called_once_with()
-    input_queue.cancel_join_thread.assert_called_once_with()
-    output_queue.cancel_join_thread.assert_called_once_with()
-
-
-def test_parallel_iterator_terminates_workers_when_input_raises(monkeypatch):
-    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
-
-    def failing_stream():
-        yield "input"
-        raise RuntimeError("stream failed")
-
-    with pytest.raises(RuntimeError, match="stream failed"):
-        list(pool.semi_ordered_map(failing_stream()))
-
-    assert process.join.call_args_list == [call(timeout=1), call(timeout=1)]
-    process.terminate.assert_called_once_with()
-    input_queue.cancel_join_thread.assert_called_once_with()
-    output_queue.cancel_join_thread.assert_called_once_with()
-
-
-def test_exhausted_parallel_iterator_joins_workers_gracefully(monkeypatch):
-    pool, input_queue, output_queue, process = make_stubbed_pool(monkeypatch)
-
-    assert list(pool.semi_ordered_map(["input"])) == [(0, "result")]
-
-    process.join.assert_called_once_with()
-    process.terminate.assert_not_called()
-    input_queue.join_thread.assert_called_once_with()
-    output_queue.join_thread.assert_called_once_with()
-
-
-def test_reused_pool_returns_to_graceful_queue_cleanup():
-    pool = ParallelWorkerPool(1, EchoWorker)
-    ctx = MagicMock()
-    pool.ctx = ctx
-
-    first_input_queue = MagicMock()
-    first_output_queue = MagicMock()
-    first_output_queue.get_nowait.return_value = (0, "first result")
-    second_input_queue = MagicMock()
-    second_output_queue = MagicMock()
-    second_output_queue.get_nowait.return_value = (0, "second result")
-    first_process = MagicMock()
-    first_process.is_alive.return_value = True
-    second_process = MagicMock()
-
-    ctx.Queue.side_effect = [
-        first_input_queue,
-        first_output_queue,
-        second_input_queue,
-        second_output_queue,
-    ]
-    ctx.Value.side_effect = [
-        get_context().Value("i", 1),
-        get_context().Value("i", 1),
-    ]
-    ctx.Process.side_effect = [first_process, second_process]
-
-    first_results = pool.ordered_map(["first input"])
-    assert next(first_results) == "first result"
-    first_results.close()
-
-    assert pool.emergency_shutdown is True
-    assert first_process.join.call_args_list == [call(timeout=1), call(timeout=1)]
-    first_process.terminate.assert_called_once_with()
-    first_input_queue.cancel_join_thread.assert_called_once_with()
-    first_output_queue.cancel_join_thread.assert_called_once_with()
-
-    assert list(pool.semi_ordered_map(["second input"])) == [(0, "second result")]
-
-    assert pool.emergency_shutdown is False
-    second_input_queue.join_thread.assert_called_once_with()
-    second_output_queue.join_thread.assert_called_once_with()
-    second_input_queue.cancel_join_thread.assert_not_called()
-    second_output_queue.cancel_join_thread.assert_not_called()
+# semi_ordered_map is closed by garbage collection, so an error in its cleanup is only reported as unraisable
+@pytest.mark.filterwarnings("error::pytest.PytestUnraisableExceptionWarning")
+def test_closing_partially_consumed_iterator_stops_workers():
+    start_method = "forkserver" if "forkserver" in get_all_start_methods() else "spawn"
+    pool = ParallelWorkerPool(2, EchoWorker, start_method=start_method)
+    # the stream never ends, so the workers never get their stop signals
+    results = pool.ordered_map(count())
+    assert next(results) == 0
+    workers = list(pool.processes)
+
+    # close() used to wait in join() forever, run it in a thread so a regression fails instead of
+    # hanging the test session
+    closer = threading.Thread(target=results.close, daemon=True)
+    closer.start()
+    closer.join(timeout=30)
+    try:
+        assert not closer.is_alive(), "closing the iterator hung"
+        assert not any(worker.is_alive() for worker in workers)
+    finally:
+        for worker in workers:
+            if worker.is_alive():
+                worker.kill()

From 5478edfcd950b6f921b5b3b5ff90af64f1895d11 Mon Sep 17 00:00:00 2001
From: George Panchuk <george.panchuk@qdrant.tech>
Date: Sat, 26 Sep 2026 19:58:10 +0700
Subject: [PATCH 5/5] fix: reset emergency shutdown on start

---
 fastembed/parallel_processor.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/fastembed/parallel_processor.py b/fastembed/parallel_processor.py
index 9791fcb3e..ffb3d6ec3 100644
--- a/fastembed/parallel_processor.py
+++ b/fastembed/parallel_processor.py
@@ -112,6 +112,7 @@ def __init__(
         self.num_active_workers: BaseValue | None = None
 
     def start(self, **kwargs: Any) -> None:
+        self.emergency_shutdown = False
         self.input_queue = self.ctx.Queue(self.queue_size)
         # An emergency shutdown unblocks the feeder thread with EPIPE (see semi_ordered_map), let it
         # exit quietly instead of printing a traceback. ProcessPoolExecutor does the same.