File size: 1,701 Bytes
201dc77 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """U10-7: a pool that breaks mid-batch must never score the tail INCORRECT."""
from concurrent.futures.process import BrokenProcessPool
import pytest
from align.eval.verifier import Outcome, OutcomeVerifier, VerifierConfig
def _stub_scorer(gold, preds, timeout):
return gold == preds[0]
class _Fut:
def __init__(self, val):
self.val = val
def result(self, timeout=None):
if isinstance(self.val, BaseException):
raise self.val
return self.val
class _DiesAfter:
"""Worker 'OOM-killed' after `ok_first` submissions: the rest break."""
_processes = {}
def __init__(self, ok_first):
self.n = 0
self.ok_first = ok_first
def submit(self, fn, gold, preds, timeout):
self.n += 1
if self.n <= self.ok_first:
return _Fut(fn(gold, preds, timeout))
return _Fut(BrokenProcessPool("worker died"))
def shutdown(self, **kw):
pass
def _verifier(factory):
return OutcomeVerifier(
VerifierConfig(workers=2), _score_fn=_stub_scorer, _executor_factory=factory
)
def test_persistently_broken_tail_raises_not_incorrect():
v = _verifier(lambda n: _DiesAfter(ok_first=3))
golds = [str(i) for i in range(10)]
with pytest.raises(RuntimeError, match="BrokenProcessPool"):
v.verify_batch(golds, golds)
def test_broken_tail_is_retried_on_a_fresh_pool():
pools = []
def factory(n):
pools.append(_DiesAfter(ok_first=3 if not pools else 99))
return pools[-1]
v = _verifier(factory)
golds = [str(i) for i in range(10)]
out = v.verify_batch(golds, golds)
assert out == [Outcome.CORRECT] * 10 and len(pools) == 2
|