A semaphore that prevents release() being called too many times. If `.release` would increment the semaphore's value past the initial value, it raises `ValueError`. Semaphores are mostly used to guard resources with limited capacity, so a semaphore released too many times is a sign
| 465 | |
| 466 | |
| 467 | class BoundedSemaphore(Semaphore): |
| 468 | """A semaphore that prevents release() being called too many times. |
| 469 | |
| 470 | If `.release` would increment the semaphore's value past the initial |
| 471 | value, it raises `ValueError`. Semaphores are mostly used to guard |
| 472 | resources with limited capacity, so a semaphore released too many times |
| 473 | is a sign of a bug. |
| 474 | """ |
| 475 | |
| 476 | def __init__(self, value: int = 1) -> None: |
| 477 | super().__init__(value=value) |
| 478 | self._initial_value = value |
| 479 | |
| 480 | def release(self) -> None: |
| 481 | """Increment the counter and wake one waiter.""" |
| 482 | if self._value >= self._initial_value: |
| 483 | raise ValueError("Semaphore released too many times") |
| 484 | super().release() |
| 485 | |
| 486 | |
| 487 | class Lock(object): |