fix: terminate interrupted parallel workers

[FU-max-boop]

Summary

• detect when a parallel result iterator exits before it is fully consumed
• use the bounded emergency shutdown path instead of joining workers that may still be blocked on the input queue
• preserve graceful joins for fully consumed iterators
• cover early close, input failure, and normal exhaustion with regression tests

Testing

• pytest -q tests/test_parallel_processor.py tests/test_common.py (4 passed)
• real multiprocessing reproduction: partial ordered_map() close returns and leaves zero tracked processes
• ruff check fastembed/parallel_processor.py tests/test_parallel_processor.py
• ruff format --check fastembed/parallel_processor.py tests/test_parallel_processor.py
• pytest --collect-only -q (92 tests collected)

All Submissions

• Followed the contributing guidelines
• Checked for other open pull requests covering the same change

Closes #669

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 6926973c-4f56-4152-8468-c48ed6abd40e

📥 Commits

Reviewing files that changed from the base of the PR and between 8ed9811 and 859206f.

📒 Files selected for processing (1)

• tests/test_parallel_processor.py

🚧 Files skipped from review as they are similar to previous changes (1)

• tests/test_parallel_processor.py

---

📝 Walkthrough

Walkthrough

semi_ordered_map now tracks normal completion and selects graceful or emergency worker cleanup. Early iterator closure and input-generation errors use termination cleanup, while exhausted iteration uses a regular join. start() resets emergency-shutdown state for pool reuse. Tests cover worker termination, joining, queue-thread cleanup, input failures, normal exhaustion, and reuse.

Estimated code review effort: 2 (Simple) | ~10 minutes

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the fix for interrupted parallel worker cleanup.
Description check	✅ Passed	The description directly explains the cleanup behavior, regression coverage, and validation for the reported issue.
Linked Issues check	✅ Passed	The changes address issue #669 by terminating workers on early iterator exit while preserving graceful joins after normal exhaustion.
Out of Scope Changes check	✅ Passed	The code and tests remain focused on parallel worker cleanup and the requirements described in issue #669.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@fastembed/parallel_processor.py`:
- Around line 211-212: Update the emergency shutdown flow around
join_or_terminate to join every worker after process.terminate(), using the
existing timeout, before clearing self.processes. Extend both emergency-shutdown
tests to verify that each terminated worker receives the second join() call.
- Line 156: Reset ParallelWorkerPool.emergency_shutdown to False at the
beginning of start(), before creating new queues, so a reused pool performs
normal join_thread() cleanup after a prior early close. Add a test covering
sequential reuse of the same pool after early shutdown.

🪄 Autofix

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 3f807871-53de-4475-82bb-84eda11975a9

📥 Commits

Reviewing files that changed from the base of the PR and between f613647 and b15b5a7.

📒 Files selected for processing (2)

• fastembed/parallel_processor.py
• tests/test_parallel_processor.py

[coderabbitai]

🧹 Nitpick comments (1)

tests/test_parallel_processor.py (1)

104-110: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Assert worker termination after pool reuse.

After first_results.close(), this test checks the emergency flag and queue cancellation, but it does not check first_process.terminate() or the bounded join() calls. A regression limited to a reused pool could leave the first worker alive and still pass this test. Add the worker lifecycle assertions used by the fresh-pool test.

Proposed assertions

     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()

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_parallel_processor.py` around lines 104 - 110, Extend the
reused-pool assertions after first_results.close() to verify
first_process.terminate() was called and that its bounded join() calls match the
fresh-pool test. Keep the existing emergency_shutdown and queue cancellation
assertions unchanged, using the same lifecycle assertion pattern and timeout
expectations already established for the fresh-pool case.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@tests/test_parallel_processor.py`:
- Around line 104-110: Extend the reused-pool assertions after
first_results.close() to verify first_process.terminate() was called and that
its bounded join() calls match the fresh-pool test. Keep the existing
emergency_shutdown and queue cancellation assertions unchanged, using the same
lifecycle assertion pattern and timeout expectations already established for the
fresh-pool case.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 5437ebdf-c331-4a07-a51e-3cae8d76b00d

📥 Commits

Reviewing files that changed from the base of the PR and between b15b5a7 and 8ed9811.

📒 Files selected for processing (2)

• fastembed/parallel_processor.py
• tests/test_parallel_processor.py

🚧 Files skipped from review as they are similar to previous changes (1)

• fastembed/parallel_processor.py