02.10.01 — WebDriversPool[Driver]#

Source file: src/ocarina/infra/drivers_pool.py

Thread-safe driver pool, max concurrency enforced by semaphore, monitored async warmup, clean shutdown. No reuse — every acquire() disposes its driver on the way out (clean state).

Constructor#

@final
class WebDriversPool[Driver]:
    def __init__(
        self,
        create_driver: Thunk[BuiltWebDriver[Driver]],
        max_size: int,
        warmup_timeout: float | None = None,
    ) -> None:
        self._create_driver = create_driver
        self._pool: Queue[BuiltWebDriver[Driver]] = Queue(max_size)
        self._semaphore = Semaphore(max_size)
        self._warmup_timeout = (
            warmup_timeout if warmup_timeout is not None and warmup_timeout > 0.1
            else 60.0 * 5
        )

acquire()#

@contextmanager
def acquire(self) -> Iterator[Driver]:
    try:
        driver, dispose = self._pool.get_nowait()
    except Empty:
        self._semaphore.acquire()
        try:
            driver, dispose = self._create_driver()
        except Exception:
            self._semaphore.release()
            raise

    try:
        yield driver
    finally:
        with suppress(Exception):
            dispose()
        self._semaphore.release()

              acquire()
                  │
                  ▼
   ┌──────────────────────────────────┐
   │ pool.get_nowait()                │── OK ─► driver, dispose         ← warmup case
   └──────────────┬───────────────────┘
                  │ Empty
                  ▼
   ┌──────────────────────────────────┐
   │ sem.acquire()                    │  ← blocks if N drivers alive
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ create_driver()                  │── raises ─► sem.release(); raise
   └──────────────┬───────────────────┘
                  ▼
   ┌──────────────────────────────────┐
   │ yield driver                     │  ← caller uses it
   └──────────────┬───────────────────┘
                  ▼
                finally :
                  with suppress : dispose()       ← driver disposed
                  sem.release()                   ← release a slot

1. No reuse#

Queue.get_nowait() consumes the entry. Once pulled, the driver never goes back. At the end (finally), it’s disposed.

No driver is ever used by two tests.

2. No semaphore leak#

• If pool.get_nowait() returns a driver: we use it, dispose, release. ✅
• If pool.get_nowait() raises Empty: we acquire the sem, create, use, dispose, release. ✅
• If create_driver() raises: we release the sem before propagating. ✅
• If dispose() raises: we suppress + release. ✅

Every path releases the semaphore. No leak.

3. Concurrency guaranteed ≤ max_size#

The semaphore is acquired before creation. As long as N drivers are alive (created or queued), the N+1-th caller waits.

warmup()#

def warmup(self) -> None:
    progress = {"count": 0}
    lock = threading.Lock()
    stop_event = threading.Event()

    def worker() -> None:
        while not self._pool.full() and not stop_event.is_set():
            if not self._semaphore.acquire(blocking=False):
                break
            try:
                driver, dispose = self._create_driver()
                self._pool.put((driver, dispose))
                with lock:
                    progress["count"] += 1
            except Exception:
                self._semaphore.release()
                break

    t = threading.Thread(target=worker, daemon=True)
    t.start()

    start_global = time.monotonic()
    last_progress = 0

    while t.is_alive():
        time.sleep(0.5)
        with lock:
            current = progress["count"]

        if current != last_progress:
            last_progress = current
            start_global = time.monotonic()

        if time.monotonic() - start_global > self._warmup_timeout:
            stop_event.set()
            msg = (
                "Warmup stalled (no progress detected)."
                " Some browser processes may still be running."
                " Please check your system (Activity Monitor / Task Manager / Dock)"
                " and close any remaining browser instances."
            )
            self.shutdown()
            raise WarmupTimeoutError(msg)

    t.join()

A progress-based watchdog, not elapsed real time. 30s to create a driver is fine if progress advances. Creation blocking indefinitely → progress stalls → watchdog raises.

shutdown()#

def shutdown(self) -> None:
    while not self._pool.empty():
        _, dispose = self._pool.get_nowait()
        with suppress(Exception):
            dispose()
        self._semaphore.release()

• Disposes every driver in the queue (not the acquired ones — those get handled by their acquire() context).
• Releases the matching semaphores.
• suppress(Exception): a dispose() that raises gets skipped, we move on.

Called:

1. By TestSuite.run on WarmupTimeoutError.
2. Via atexit.register(pool.shutdown) at process end (see infra/selenium/create_drivers_pool.py).

WarmupTimeoutError#

class WarmupTimeoutError(Exception):
    """Raised when warmup is blocked (mostly some macOS edge cases)."""

“macOS edge cases”: with headless off and you close a browser window driven by Ocarina, macOS doesn’t kill the process behind the window.

That very specific case exposed the possibility of an infinite hang, and triggered the naive watchdog.

Dedicated tests#

tests/scenarios/test_drivers_pool.py covers:

• standard acquisition / release,
• warmup,
• WarmupTimeoutError detection when the factor stalls,
• behavior when create_driver raises (sem released),
• shutdown.