POSIX, signals-based interruptible runner.
(func, args=None, kwargs=None, timeout=DEFAULT_TIMEOUT)
| 78 | from signal import signal as create_signal |
| 79 | |
| 80 | def interruptible(func, args=None, kwargs=None, timeout=DEFAULT_TIMEOUT): |
| 81 | """ |
| 82 | POSIX, signals-based interruptible runner. |
| 83 | """ |
| 84 | |
| 85 | def handler(signum, frame): |
| 86 | raise TimeoutError |
| 87 | |
| 88 | try: |
| 89 | create_signal(SIGALRM, handler) |
| 90 | setitimer(ITIMER_REAL, timeout) |
| 91 | return NO_ERROR, func(*(args or ()), **(kwargs or {})) |
| 92 | |
| 93 | except TimeoutError: |
| 94 | return TIMEOUT_MSG % locals(), NO_VALUE |
| 95 | |
| 96 | except Exception: |
| 97 | return ERROR_MSG + traceback_format_exc(), NO_VALUE |
| 98 | |
| 99 | finally: |
| 100 | setitimer(ITIMER_REAL, 0) |
| 101 | |
| 102 | elif on_windows: |
| 103 | """ |
nothing calls this directly
no test coverage detected