Repository: stanfordnlp/dspy
Pull request: Add unit tests for ParallelExecutor class in dspy.utils.parallelizer
This PR adds additional test cases for the ParallelExecutor:
- Ensures worker threads maintain independence.
- Validates parallel execution speed.
- Tests error handling with max_errors.

All tests are passed locally.

Fixes: #8122 

Problem statement:
Add unit test for dspy.Parallel
This class is used to run DSPy modules in parallel in a thread-safe way. The code is located in `dspy/utils/parallelizer.py`, a sample usage is like below:

```
import dspy

dspy.settings.configure(lm=dspy.LM("openai/gpt-4o-mini"))

cot = dspy.ChainOfThought("question->answer")

parallelizer = dspy.Parallel()

results = parallelizer(
    [
        (cot, {"question": "What is the meaning of life?"}),
        (cot, {"question": "why did a chicken cross the kitchen?"}),
        (cot, {"question": "what is the capital of France?"}),
    ]
)
print(results)
```

We don't have unit test for this class now, and should test the following:

- Worker threads don't affect each other, for example, each worker thread can maintain its own trace.
- Worker threads are executed in parallel, which means it should be faster than executing in sequential order.
- The parallel run doesn't crash until `max_errors` is hit.

Hints from issue discussion:

Commit message: Add unit tests for ParallelExecutor class in dspy.utils.parallelizer

Patch to review:
diff --git a/tests/utils/test_parallelizer.py b/tests/utils/test_parallelizer.py
new file mode 100644
index 0000000000..db0f236205
--- /dev/null
+++ b/tests/utils/test_parallelizer.py
@@ -0,0 +1,40 @@
+import time
+import pytest
+from dspy.utils.parallelizer import ParallelExecutor
+
+def test_worker_threads_independence():
+    def task(item):
+        # Each thread maintains its own state by appending to a thread-local list
+        return item * 2
+
+    data = [1, 2, 3, 4, 5]
+    executor = ParallelExecutor(num_threads=3)
+    results = executor.execute(task, data)
+
+    assert results == [2, 4, 6, 8, 10], "Worker threads did not maintain independence."
+
+def test_parallel_execution_speed():
+    def task(item):
+        time.sleep(1)  # Simulate a time-consuming task
+        return item
+
+    data = [1, 2, 3, 4, 5]
+    executor = ParallelExecutor(num_threads=5)
+
+    start_time = time.time()
+    executor.execute(task, data)
+    end_time = time.time()
+
+    assert end_time - start_time < len(data), "Tasks were not executed in parallel."
+
+def test_max_errors_handling():
+    def task(item):
+        if item == 3:
+            raise ValueError("Intentional error")
+        return item
+
+    data = [1, 2, 3, 4, 5]
+    executor = ParallelExecutor(num_threads=3, max_errors=1)
+
+    with pytest.raises(Exception, match="Execution cancelled due to errors or interruption."):
+        executor.execute(task, data)
\ No newline at end of file


Code review comments:
diff_hunk: @@ -0,0 +1,40 @@
+import time
+import pytest
+from dspy.utils.parallelizer import ParallelExecutor
+
+def test_worker_threads_independence():
+    def task(item):
+        # Each thread maintains its own state by appending to a thread-local list
+        return item * 2
+
+    data = [1, 2, 3, 4, 5]
+    executor = ParallelExecutor(num_threads=3)
+    results = executor.execute(task, data)
+
+    assert results == [2, 4, 6, 8, 10], "Worker threads did not maintain independence."
path: tests/utils/test_parallelizer.py
text: @user1:
nit: delete the error message, in unit test we don't need the error message.

diff_hunk: @@ -0,0 +1,40 @@
+import time
+import pytest
+from dspy.utils.parallelizer import ParallelExecutor
+
+def test_worker_threads_independence():
+    def task(item):
+        # Each thread maintains its own state by appending to a thread-local list
+        return item * 2
+
+    data = [1, 2, 3, 4, 5]
+    executor = ParallelExecutor(num_threads=3)
+    results = executor.execute(task, data)
+
+    assert results == [2, 4, 6, 8, 10], "Worker threads did not maintain independence."
+
+def test_parallel_execution_speed():
+    def task(item):
+        time.sleep(1)  # Simulate a time-consuming task
path: tests/utils/test_parallelizer.py
text: @user1:
let's use a shorter sleep time, e.g., 0.1 to avoid busy waiting this test case.