# -*- coding: utf-8 -*- """Run a batch of job files on a background thread without ever throwing. WHY THIS MODULE IS THE ONLY PLACE THIS CODE MAY LIVE. A pyRevit button command can return while a thread it started is still running, and pyRevit then tears down the command script's module scope. A name looked up from that scope -- a module-level import such as `io`, `os` or `liftexport` -- is gone. The lookup misses, and the miss happens on a NON-MAIN thread, and Revit has no way to catch an exception on a non-main thread: an unhandled one ends the whole process (measured 2026-08-27: UnboundNameException in a Panel Lift Check batch thread, ~7 s after the command returned; the event log named IronPython lookup-of-global as the killer). A thread cannot be made safe by wrapping each call -- the miss happens at name lookup, before any body runs. It is only safe with NO global name lookup at all. Everything this worker needs arrives as a default parameter, bound at definition time (when the scope it was defined in is still alive), and the whole body is wrapped so nothing can escape. A lost log line is acceptable; a dead Revit is not. ONE PRINCIPLE, stated because it was learned the hard way: a function whose contract is "never raises" must keep EVERY statement inside the protection -- including the first one. `list(compute)` can raise before any try; so can unpacking a row; so can `os.path.splitext`. The outer try/except below exists for exactly those, and the tests feed the worker malformed inputs to prove it stays silent. The free functions below are what the tests exercise: build the same callable, inject a job that raises and a status path that cannot be written, and assert it still does not raise. IronPython 2.7 is assumed: no f-strings, no pathlib. MANY AT ONCE, AND WHY THAT IS SAFE HERE. Each job is already an isolated CPython subprocess that reads one job file and writes its own report, log and summary beside it. Nothing is shared between two of them -- not a file, not a number, not a random seed -- so running eight at once is a scheduling change and not a numerical one: the reports come out identical, only sooner. Measured 2026-08-30 on the 184-piece reference sweep: 42 minutes serial, 8.6 minutes on eight workers. What running many at once DOES touch is coordination, and coordination is exactly the thing this module is paranoid about. So the extra machinery obeys the same rule as everything else here: the queue, the lock and the thread factory are all created and bound in `make_worker`, on the UI thread, while the defining scope is alive; the functions the threads run are defined there too, with every name they need bound as a default parameter. A worker thread looks nothing up. `pump` carries its own outer try, because an exception escaping THERE is the same dead Revit as one escaping `worker` -- there are now N threads that can kill the process instead of one. """ # Most jobs a batch will compute at the same time, whatever the machine has. # Eight subprocesses is already more than the sweep needs to fit inside a # coffee break, and the number exists to stop a 64-core workstation from # starting 62 of them. MAX_WORKERS = 8 # Cores left for everything else. Revit is open, the user is still working in # it, and a batch that makes the model unusable is a batch nobody starts. SPARE_CORES = 2 def worker_count(os=None, cap=None, spare=None): """How many jobs to compute at once: min(cores - 2, 8), and never 0. Called on the UI thread, before any thread exists, and the answer is then bound as a default parameter -- so a background thread never asks this question and never touches `os.environ`. NUMBER_OF_PROCESSORS is read before `os.cpu_count`, because it is the one that exists on IronPython 2.7 (Windows sets it for every process) and `cpu_count` does not arrive until CPython 3.4. A machine that answers neither gets 1, which is exactly today's behaviour. """ import os as _os os = os or _os cap = MAX_WORKERS if cap is None else cap spare = SPARE_CORES if spare is None else spare cores = 0 try: cores = int(os.environ.get('NUMBER_OF_PROCESSORS') or 0) except Exception: # pylint: disable=broad-except cores = 0 if not cores: try: cores = int(os.cpu_count() or 0) # CPython 3.4+ only except Exception: # pylint: disable=broad-except cores = 0 if not cores: return 1 count = cores - spare if count < 1: count = 1 if count > cap: count = cap return count def make_worker(compute, status_path, run_job, io=None, os=None, workers=None, threading=None): """A zero-argument callable that runs a batch and never raises. compute -- list of [name, job_path] rows, as the batch engine in Panel Lift Check builds them. The list is copied, so the caller may reuse its own after handing it over. status_path -- where progress lines are appended. run_job -- callable(name, job_path, out_path) that computes one report. It may raise; the worker records an ERROR line and continues with the next job. workers -- how many jobs to compute at once. None asks the machine (see `worker_count`). 1 is the old behaviour exactly: one job after another, on the calling thread, no extra thread created at all. io, os, threading -- injected for tests (a fake io whose open raises, say). Defaults to the real modules, bound NOW -- at def time, while this module's scope is alive, which is the whole point: a background thread must never look anything up from a scope that may have been torn down. """ import io as _io import os as _os import threading as _threading io = io or _io os = os or _os threading = threading or _threading if workers is None: workers = worker_count(os) # One lock for the queue AND the status file. They are never held at the # same time -- take() releases before run_job starts, note() takes it # after -- so one lock cannot deadlock against itself, and two locks # would be two things to reason about for no gain. try: lock = threading.Lock() except Exception: # pylint: disable=broad-except lock = None workers = 1 # no lock, no sharing: run it the old way def _grab(lock=lock): try: if lock is not None: lock.acquire() except Exception: # pylint: disable=broad-except pass def _drop(lock=lock): try: if lock is not None: lock.release() except Exception: # pylint: disable=broad-except pass def note(line, io=io, status_path=status_path, mode='a', grab=_grab, drop=_drop): """One line into the status file. Never raises, ever. acquire, write and release are three separate guards on purpose: a release() on a lock that was never acquired raises, so it cannot sit in a `finally` that runs after a failed acquire. """ grab() try: with io.open(status_path, mode, encoding='utf-8') as fh: fh.write(line + '\n') except Exception: # pylint: disable=broad-except pass # cannot write the log -> stay silent drop() def take(queue, grab=_grab, drop=_drop): """The next [name, job_path], or None when the batch is empty.""" grab() row = None try: if queue: row = queue.pop(0) except Exception: # pylint: disable=broad-except row = None drop() return row def pump(queue, take=take, note=note, run_job=run_job, os=os): """Drain the shared queue. Runs ON a worker thread, so it never raises: an exception here ends the process the same way one in `worker` does, and there are now several of these.""" try: while True: row = take(queue) if row is None: return try: name, job_path = row out_path = os.path.splitext(job_path)[0] + '.html' run_job(name, job_path, out_path) line = 'DONE {0}'.format(name) except Exception as exc: # pylint: disable=broad-except # A row that would not even unpack still gets a line. # Dropping it -- which the serial version did, by letting # the outer guard swallow the whole rest of the batch -- # is a piece that silently never ran. try: line = 'ERROR {0}: {1}'.format(row[0], exc) except Exception: # pylint: disable=broad-except line = 'ERROR (unreadable row): {0}'.format(exc) note(line) except Exception: # pylint: disable=broad-except pass def worker(compute=compute, note=note, pump=pump, threading=threading, workers=workers): try: queue = list(compute) # The count is on the header line because a batch that is running # eight at a time and one that is running one look identical from # outside, and they are 42 minutes apart. note('BATCH RUNNING ({0} panels), {1} at a time' .format(len(queue), workers), mode='w') helpers = [] # A counted `while` rather than `range(...)`: one fewer name this # thread has to resolve through a scope that may be gone. The # calling thread is worker number one, so it starts workers - 1. started = 1 while started < workers: started += 1 try: thread = threading.Thread(target=pump, args=(queue,)) try: thread.daemon = False except Exception: # pylint: disable=broad-except pass thread.start() helpers.append(thread) except Exception: # pylint: disable=broad-except # Could not start another thread. The ones already going # plus this one still finish the batch, just slower. break pump(queue) # this thread pulls its share too for thread in helpers: try: thread.join() except Exception: # pylint: disable=broad-except pass note('BATCH FINISHED') except Exception: # pylint: disable=broad-except # "Never raises" includes the first statement. list(compute) can # raise before any of the inner guards run; this one catches it. pass return worker