will
25592fa6c2
ci / test (push) Successful in 6m14s
History-free snapshot of the private development repository at 77a82eb. Code, docs and the public catalogue only: no plant or job knowledge layers, no client, job or producer names (tools/scrub.py clean against the widened term list; a second word-level grep clean). The intro deck is left out because its slide images cannot be checked for job drawings.
381 lines
13 KiB
Python
381 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""The batch worker must never throw, and must never look a module global up.
|
|
|
|
A pyRevit command can return while the thread it started is still running,
|
|
and pyRevit then tears the command module's scope down. A thread that looks
|
|
a module global up from that scope raises UnboundNameException -- on a
|
|
non-main thread, which Revit cannot catch, so the whole process dies
|
|
(measured 2026-08-27, Panel Lift Check batch thread, event log
|
|
IronPython.PythonOps.GetVariable as the killer).
|
|
|
|
These tests build the same callable vibe.batchrun.make_worker returns,
|
|
inject the failure modes that used to kill Revit -- a job that raises, a
|
|
status path that cannot be written -- and assert the worker still does not
|
|
raise, and that the error reached the status file (or was silently
|
|
dropped when even that was impossible). Plus the static half: the
|
|
worker's code object contains no name that would resolve to a module
|
|
global of the command script; every free name is a parameter bound at
|
|
definition time.
|
|
"""
|
|
# pylint: disable=import-error,invalid-name,broad-except
|
|
import threading as _threading
|
|
import time
|
|
|
|
from vibe import batchrun
|
|
|
|
|
|
class BoomIO(object):
|
|
"""An io whose open always fails -- the status file cannot be written."""
|
|
|
|
def open(self, path, mode, encoding=None):
|
|
raise IOError('cannot write status: {0}'.format(path))
|
|
|
|
|
|
class RecorderIO(object):
|
|
"""An io that records every write, so the test can read the log back."""
|
|
|
|
def __init__(self):
|
|
self.writes = []
|
|
|
|
def open(self, path, mode, encoding=None):
|
|
return _FakeFile(self)
|
|
|
|
|
|
class _FakeFile(object):
|
|
def __init__(self, io):
|
|
self._io = io
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def write(self, text):
|
|
self._io.writes.append(text)
|
|
|
|
|
|
def test_worker_survives_a_job_that_raises_and_records_it():
|
|
"""An exploding liftexport.run must not propagate; the ERROR line does."""
|
|
recorder = RecorderIO()
|
|
|
|
def run_job(name, job_path, out_path):
|
|
if name == 'A':
|
|
raise RuntimeError('the calculation exploded')
|
|
return None
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json'], ['B', 'b.job.json']],
|
|
'batch-status.txt', run_job, io=recorder)
|
|
|
|
worker() # must not raise
|
|
|
|
text = '\n'.join(recorder.writes)
|
|
assert 'BATCH RUNNING (2 panels)' in text
|
|
assert 'ERROR A: the calculation exploded' in text
|
|
assert 'DONE B' in text
|
|
assert 'BATCH FINISHED' in text
|
|
|
|
|
|
def test_worker_survives_an_unwritable_status_path():
|
|
"""Even when every status write fails, nothing may escape the thread."""
|
|
recorder = RecorderIO()
|
|
|
|
def run_job(name, job_path, out_path):
|
|
return None
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json']], 'batch-status.txt', run_job, io=BoomIO())
|
|
|
|
worker() # must not raise
|
|
assert recorder.writes == [] # and it truly dropped the log, not faked it
|
|
|
|
|
|
def test_worker_continues_after_a_failed_job():
|
|
"""One bad piece must not stop the rest of the batch.
|
|
|
|
workers=1 on purpose: this asserts the ORDER, and order is exactly what
|
|
more than one worker is allowed to change.
|
|
"""
|
|
recorder = RecorderIO()
|
|
ran = []
|
|
|
|
def run_job(name, job_path, out_path):
|
|
ran.append(name)
|
|
if name == 'A':
|
|
raise RuntimeError('boom')
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json'], ['B', 'b.job.json'], ['C', 'c.job.json']],
|
|
'batch-status.txt', run_job, io=recorder, workers=1)
|
|
|
|
worker()
|
|
assert ran == ['A', 'B', 'C']
|
|
text = '\n'.join(recorder.writes)
|
|
assert 'DONE C' in text
|
|
|
|
|
|
def test_one_worker_starts_no_thread_at_all():
|
|
"""workers=1 is today's behaviour exactly, including the thread count."""
|
|
recorder = RecorderIO()
|
|
spawned = []
|
|
|
|
class CountingThreading(object):
|
|
Lock = _threading.Lock
|
|
|
|
@staticmethod
|
|
def Thread(target=None, args=()):
|
|
spawned.append(target)
|
|
return _threading.Thread(target=target, args=args)
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json']], 's', lambda *a: None, io=recorder,
|
|
workers=1, threading=CountingThreading)
|
|
worker()
|
|
assert spawned == []
|
|
|
|
|
|
def test_every_piece_runs_exactly_once_across_many_workers():
|
|
"""The shared queue is the whole coordination risk: no piece may be
|
|
computed twice and none may be skipped."""
|
|
recorder = RecorderIO()
|
|
ran = []
|
|
lock = _threading.Lock()
|
|
|
|
def run_job(name, job_path, out_path):
|
|
time.sleep(0.001) # long enough for the threads to overlap
|
|
lock.acquire()
|
|
ran.append(name)
|
|
lock.release()
|
|
|
|
rows = [['P{0}'.format(i), 'p{0}.job.json'.format(i)] for i in range(60)]
|
|
worker = batchrun.make_worker(rows, 's', run_job, io=recorder, workers=8)
|
|
worker()
|
|
|
|
assert sorted(ran) == sorted(r[0] for r in rows)
|
|
text = '\n'.join(recorder.writes)
|
|
for row in rows:
|
|
assert 'DONE {0}'.format(row[0]) in text
|
|
assert text.count('BATCH FINISHED') == 1
|
|
|
|
|
|
def test_the_batch_really_overlaps():
|
|
"""Not just correct, actually concurrent -- otherwise the whole change
|
|
bought nothing. Eight workers on eight jobs that each sleep must finish
|
|
in well under eight sleeps."""
|
|
recorder = RecorderIO()
|
|
|
|
def run_job(name, job_path, out_path):
|
|
time.sleep(0.25)
|
|
|
|
rows = [['P{0}'.format(i), 'p{0}.job.json'.format(i)] for i in range(8)]
|
|
worker = batchrun.make_worker(rows, 's', run_job, io=recorder, workers=8)
|
|
start = time.time()
|
|
worker()
|
|
elapsed = time.time() - start
|
|
assert elapsed < 1.0, elapsed # serial would be 2.0 s
|
|
|
|
|
|
def test_finished_is_written_after_every_worker_has_stopped():
|
|
"""BATCH FINISHED means finished. A reader polling the status file uses
|
|
it to decide the reports are all on disk."""
|
|
recorder = RecorderIO()
|
|
|
|
def run_job(name, job_path, out_path):
|
|
time.sleep(0.02)
|
|
|
|
rows = [['P{0}'.format(i), 'p{0}.job.json'.format(i)] for i in range(20)]
|
|
worker = batchrun.make_worker(rows, 's', run_job, io=recorder, workers=8)
|
|
worker()
|
|
lines = [w.strip() for w in recorder.writes if w.strip()]
|
|
assert lines[-1] == 'BATCH FINISHED'
|
|
assert len([l for l in lines if l.startswith('DONE')]) == 20
|
|
|
|
|
|
def test_the_header_says_how_many_run_at_once():
|
|
"""Eight at a time and one at a time look identical from outside, and
|
|
they are 40 minutes apart on the reference sweep."""
|
|
recorder = RecorderIO()
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json'], ['B', 'b.job.json']], 's',
|
|
lambda *a: None, io=recorder, workers=4)
|
|
worker()
|
|
text = '\n'.join(recorder.writes)
|
|
assert 'BATCH RUNNING (2 panels)' in text # the old line survives
|
|
assert '4 at a time' in text
|
|
|
|
|
|
def test_a_thread_that_will_not_start_still_finishes_the_batch():
|
|
"""No thread, no batch is not acceptable: the calling thread pulls too."""
|
|
recorder = RecorderIO()
|
|
ran = []
|
|
|
|
class NoThreads(object):
|
|
Lock = _threading.Lock
|
|
|
|
@staticmethod
|
|
def Thread(target=None, args=()):
|
|
raise RuntimeError('out of threads')
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json'], ['B', 'b.job.json']], 's',
|
|
lambda name, j, o: ran.append(name), io=recorder,
|
|
workers=8, threading=NoThreads)
|
|
worker() # must not raise
|
|
assert ran == ['A', 'B']
|
|
assert 'BATCH FINISHED' in '\n'.join(recorder.writes)
|
|
|
|
|
|
def test_no_lock_falls_back_to_one_worker():
|
|
"""Without a lock there is no safe shared queue, so it runs serially
|
|
rather than racing one."""
|
|
recorder = RecorderIO()
|
|
ran = []
|
|
|
|
class NoLock(object):
|
|
@staticmethod
|
|
def Lock():
|
|
raise RuntimeError('no locks here')
|
|
|
|
@staticmethod
|
|
def Thread(target=None, args=()):
|
|
raise AssertionError('must not spawn without a lock')
|
|
|
|
worker = batchrun.make_worker(
|
|
[['A', 'a.job.json'], ['B', 'b.job.json']], 's',
|
|
lambda name, j, o: ran.append(name), io=recorder,
|
|
threading=NoLock)
|
|
worker()
|
|
assert ran == ['A', 'B']
|
|
|
|
|
|
def test_a_malformed_row_no_longer_swallows_the_rest_of_the_batch():
|
|
"""The serial version let the outer guard eat the whole remainder when a
|
|
row would not unpack -- every piece after it silently never ran."""
|
|
recorder = RecorderIO()
|
|
ran = []
|
|
|
|
worker = batchrun.make_worker(
|
|
[[], ['B', 'b.job.json']], 's',
|
|
lambda name, j, o: ran.append(name), io=recorder, workers=1)
|
|
worker() # must not raise
|
|
assert ran == ['B']
|
|
assert 'BATCH FINISHED' in '\n'.join(recorder.writes)
|
|
|
|
|
|
# --------------------------------------------------- how many run at once
|
|
|
|
def test_worker_count_leaves_two_cores_for_revit():
|
|
"""Revit is open and the user is still working in it."""
|
|
assert batchrun.worker_count(_FakeOS('16')) == 8 # capped
|
|
assert batchrun.worker_count(_FakeOS('8')) == 6
|
|
assert batchrun.worker_count(_FakeOS('4')) == 2
|
|
|
|
|
|
def test_worker_count_never_returns_zero():
|
|
"""A batch that starts no workers computes nothing at all."""
|
|
assert batchrun.worker_count(_FakeOS('2')) == 1
|
|
assert batchrun.worker_count(_FakeOS('1')) == 1
|
|
assert batchrun.worker_count(_FakeOS('')) == 1
|
|
assert batchrun.worker_count(_FakeOS('not a number')) == 1
|
|
|
|
|
|
def test_worker_count_is_capped():
|
|
"""A 64-core workstation must not start 62 CPython subprocesses."""
|
|
assert batchrun.worker_count(_FakeOS('64')) == batchrun.MAX_WORKERS
|
|
|
|
|
|
class _FakeOS(object):
|
|
"""An os that reports a given NUMBER_OF_PROCESSORS and nothing else.
|
|
|
|
No cpu_count attribute, deliberately: that is IronPython 2.7, where the
|
|
environment variable is the only answer available.
|
|
"""
|
|
|
|
def __init__(self, cores):
|
|
self.environ = {'NUMBER_OF_PROCESSORS': cores}
|
|
|
|
|
|
def test_worker_looks_up_no_module_global_names():
|
|
"""Static half: every free name is a bound parameter, not a global.
|
|
|
|
co_names holds the names a code object resolves by LOAD_GLOBAL /
|
|
LOAD_ATTR. The module globals that killed Revit -- io, os, liftexport,
|
|
the caller's compute/status_path/run_job/python_exe -- must not be
|
|
among them: they are all parameters bound at definition time, when the
|
|
defining scope was still alive. (Method/attribute names like 'open'
|
|
and 'write' belong in co_names and are fine; the point is that the
|
|
MODULE names do not.)
|
|
"""
|
|
def quiet(name, job_path, out_path):
|
|
return None
|
|
|
|
worker = batchrun.make_worker([], 's', quiet)
|
|
|
|
names = worker.__code__.co_names
|
|
for forbidden in ('io', 'os', 'liftexport', 'compute', 'status_path',
|
|
'run_job', 'python_exe',
|
|
# added with the parallel batch: the module and the
|
|
# helpers it needs are parameters too, and `worker_count`
|
|
# is answered on the UI thread before any thread exists.
|
|
'threading', 'worker_count', 'MAX_WORKERS',
|
|
'SPARE_CORES', 'note', 'pump', 'take', 'lock'):
|
|
assert forbidden not in names, forbidden
|
|
|
|
|
|
def test_the_functions_that_run_on_worker_threads_look_up_nothing_either():
|
|
"""There are now N threads that can kill Revit, not one.
|
|
|
|
`pump`, `take` and `note` run on the spawned threads, so they are held to
|
|
the same rule as `worker`: every module name they need is a parameter
|
|
bound in `make_worker`, while its scope was alive.
|
|
"""
|
|
holder = {}
|
|
|
|
class Spy(object):
|
|
Lock = _threading.Lock
|
|
|
|
@staticmethod
|
|
def Thread(target=None, args=()):
|
|
holder['pump'] = target
|
|
return _threading.Thread(target=target, args=args)
|
|
|
|
# A recorder io, like every sibling test: this one actually CALLS the
|
|
# worker, and the real io would write a file named 's' into the CWD.
|
|
worker = batchrun.make_worker([], 's', lambda *a: None, workers=2,
|
|
threading=Spy, io=RecorderIO())
|
|
worker()
|
|
pump = holder.get('pump')
|
|
assert pump is not None
|
|
# pump's own free names, plus those of the take/note it was handed.
|
|
seen = set(pump.__code__.co_names)
|
|
for closed in pump.__defaults__ or ():
|
|
code = getattr(closed, '__code__', None)
|
|
if code is not None:
|
|
seen |= set(code.co_names)
|
|
for forbidden in ('io', 'os', 'liftexport', 'threading', 'batchrun',
|
|
'status_path', 'run_job', 'lock', 'compute'):
|
|
assert forbidden not in seen, forbidden
|
|
|
|
|
|
def test_worker_survives_malformed_inputs():
|
|
"""The outer guard covers statements before any inner try.
|
|
|
|
"A function whose contract is never raises must keep EVERY statement
|
|
inside the protection, including the first one" -- list(compute),
|
|
unpacking a row, and os.path.splitext can all raise before the inner
|
|
guards run.
|
|
"""
|
|
recorder = RecorderIO()
|
|
|
|
def quiet(name, job_path, out_path):
|
|
return None
|
|
|
|
malformed = [
|
|
None, # compute not iterable
|
|
[[], []], # row not unpackable
|
|
[['A', None]], # job_path not a path
|
|
]
|
|
for compute in malformed:
|
|
worker = batchrun.make_worker(compute, 's', quiet, io=recorder)
|
|
worker() # must not raise, for every shape |