Dump thread stacks when blocked on lock acquisition.
(lock_name: str, elapsed: float)
| 80 | |
| 81 | |
| 82 | def _dump_blocking_stacks(lock_name: str, elapsed: float) -> None: |
| 83 | """Dump thread stacks when blocked on lock acquisition.""" |
| 84 | try: |
| 85 | from ci.util.test_env import dump_thread_stacks |
| 86 | |
| 87 | yellow = "\033[33m" |
| 88 | reset = "\033[0m" |
| 89 | print( |
| 90 | f"{yellow}[LOCK] Blocked on '{lock_name}' for {elapsed:.0f}s - dumping stacks:{reset}" |
| 91 | ) |
| 92 | dump_thread_stacks() |
| 93 | except KeyboardInterrupt as ki: |
| 94 | from ci.util.global_interrupt_handler import handle_keyboard_interrupt |
| 95 | |
| 96 | handle_keyboard_interrupt(ki) |
| 97 | raise |
| 98 | except Exception: |
| 99 | pass # Best-effort diagnostic |
| 100 | |
| 101 | |
| 102 | class BuildLock: |
no test coverage detected