How to Implement a Bulkhead Pattern with Threading in Python

import threading
import time
import random


class Bulkhead:
    def __init__(self, workers: int):
        self._semaphore = threading.BoundedSemaphore(workers)
        self._lock = threading.Lock()
        self._active = 0

    def run(self, task):
        with self._semaphore:
            with self._lock:
                self._active += 1
            try:
                task()
            finally:
                with self._lock:
                    self._active -= 1

    @property
    def active(self):
        with self._lock:
            return self._active


def task(name):
    print(f"Start {name}, active={bulkhead.active}")
    time.sleep(random.uniform(0.05, 0.2))
    print(f"End {name}, active={bulkhead.active}")


if __name__ == "__main__":
    bulkhead = Bulkhead(workers=2)
    names = [f"task-{i}" for i in range(6)]
    threads = [threading.Thread(target=bulkhead.run, args=(lambda n=n: task(n),)) for n in names]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"Final active: {bulkhead.active}")