Skip to content

Commit 5a076d8

Browse files
committed
test: add parallel execution safety tests
Add tests to verify parallel sample execution doesn't cause temp file conflicts or race conditions: - test_parallel_custom_fn_isolation: verifies temp files aren't clobbered - test_parallel_preserves_order: ensures results maintain sample order - test_parallel_error_handling: validates graceful error handling
1 parent fa63a3a commit 5a076d8

1 file changed

Lines changed: 125 additions & 0 deletions

File tree

‎tests/benchmarks/test_benchmarks.py‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -242,6 +242,131 @@ def test_evaluate(self):
242242
assert 0 < result["f1"] < 1.0
243243

244244

245+
class TestParallelExecution:
246+
"""Tests for parallel execution safety."""
247+
248+
def test_parallel_custom_fn_isolation(self):
249+
"""Test that parallel execution with custom functions don't interfere."""
250+
import tempfile
251+
import threading
252+
import time
253+
from pathlib import Path
254+
255+
from benchmarks.runner import BenchmarkRunner
256+
from benchmarks.tasks.niah import NIAHBenchmark
257+
258+
# Track which thread processed which sample
259+
thread_ids = {}
260+
lock = threading.Lock()
261+
262+
def custom_inference(sample):
263+
"""Custom inference that tests isolation by writing temp files."""
264+
# Create a temp file unique to this invocation
265+
with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f:
266+
temp_path = f.name
267+
f.write(sample.id)
268+
269+
# Small delay to increase chance of race conditions
270+
time.sleep(0.01)
271+
272+
# Read back and verify it's still our data
273+
with open(temp_path) as f:
274+
read_id = f.read()
275+
276+
# Clean up
277+
Path(temp_path).unlink()
278+
279+
# Record thread info
280+
with lock:
281+
thread_ids[sample.id] = threading.current_thread().ident
282+
283+
# Return expected answer if our temp file wasn't clobbered
284+
if read_id == sample.id:
285+
return sample.expected_answer
286+
else:
287+
return f"CLOBBERED: expected {sample.id}, got {read_id}"
288+
289+
benchmark = NIAHBenchmark(context_length=1000)
290+
runner = BenchmarkRunner(max_workers=4)
291+
292+
result = runner.run(
293+
benchmark,
294+
method="custom",
295+
custom_fn=custom_inference,
296+
num_samples=8,
297+
seed=42,
298+
max_workers=4,
299+
)
300+
301+
# All samples should be correct (temp files weren't clobbered)
302+
assert result.accuracy == 1.0, f"Some samples were clobbered: {[sr for sr in result.sample_results if not sr.is_correct]}"
303+
304+
# Verify multiple threads were used
305+
unique_threads = set(thread_ids.values())
306+
assert len(unique_threads) > 1, "Expected multiple threads to be used"
307+
308+
def test_parallel_preserves_order(self):
309+
"""Test that parallel results maintain original sample order."""
310+
from benchmarks.runner import BenchmarkRunner
311+
from benchmarks.tasks.niah import NIAHBenchmark
312+
313+
def custom_inference(sample):
314+
return sample.expected_answer
315+
316+
benchmark = NIAHBenchmark(context_length=1000)
317+
runner = BenchmarkRunner(max_workers=4)
318+
319+
result = runner.run(
320+
benchmark,
321+
method="custom",
322+
custom_fn=custom_inference,
323+
num_samples=10,
324+
seed=42,
325+
max_workers=4,
326+
)
327+
328+
# Verify results are in expected order
329+
expected_ids = [f"niah-{i:04d}" for i in range(10)]
330+
actual_ids = [sr.sample_id for sr in result.sample_results]
331+
assert actual_ids == expected_ids, f"Order mismatch: {actual_ids}"
332+
333+
def test_parallel_error_handling(self):
334+
"""Test that errors in parallel execution are handled gracefully."""
335+
from benchmarks.runner import BenchmarkRunner
336+
from benchmarks.tasks.niah import NIAHBenchmark
337+
338+
call_count = {"count": 0}
339+
340+
def flaky_inference(sample):
341+
call_count["count"] += 1
342+
if call_count["count"] % 3 == 0:
343+
raise ValueError(f"Simulated error for sample {sample.id}")
344+
return sample.expected_answer
345+
346+
benchmark = NIAHBenchmark(context_length=1000)
347+
runner = BenchmarkRunner(max_workers=2)
348+
349+
result = runner.run(
350+
benchmark,
351+
method="custom",
352+
custom_fn=flaky_inference,
353+
num_samples=6,
354+
seed=42,
355+
max_workers=2,
356+
)
357+
358+
# Should complete without crashing
359+
assert len(result.sample_results) == 6
360+
361+
# Some should have errors
362+
errors = [sr for sr in result.sample_results if sr.error]
363+
assert len(errors) > 0, "Expected some errors"
364+
365+
# Errors should have zero metrics
366+
for sr in errors:
367+
assert sr.is_correct is False
368+
369+
245370
class TestBenchmarkIntegration:
246371
"""Integration tests for benchmark framework."""
247372

0 commit comments

Comments
 (0)