| 152 | |
| 153 | |
| 154 | def test_interrupt(): |
| 155 | # Windows implements `queue.get` using polling, |
| 156 | # which means we can set an exception to interrupt the call to `get`. |
| 157 | # Python 3 on other platforms requires sending SIGINT to the main thread. |
| 158 | if sys.platform == "win32": |
| 159 | from _thread import interrupt_main |
| 160 | else: |
| 161 | main_thread = threading.get_ident() |
| 162 | |
| 163 | def interrupt_main() -> None: |
| 164 | signal.pthread_kill(main_thread, signal.SIGINT) |
| 165 | |
| 166 | in_clog_event = threading.Event() |
| 167 | clog_event = threading.Event() |
| 168 | |
| 169 | def clog(in_clog_event: threading.Event, clog_event: threading.Event) -> None: |
| 170 | in_clog_event.set() |
| 171 | clog_event.wait() |
| 172 | |
| 173 | def interrupt(in_clog_event: threading.Event) -> None: |
| 174 | in_clog_event.wait() |
| 175 | interrupt_main() |
| 176 | |
| 177 | dsk = {("x", i): (clog, in_clog_event, clog_event) for i in range(20)} |
| 178 | dsk["x"] = (len, list(dsk.keys())) |
| 179 | |
| 180 | interrupter = threading.Thread(target=interrupt, args=(in_clog_event,)) |
| 181 | interrupter.start() |
| 182 | |
| 183 | # Use explicitly created ThreadPoolExecutor to avoid leaking threads after |
| 184 | # the KeyboardInterrupt |
| 185 | with ThreadPoolExecutor(CPU_COUNT) as pool: |
| 186 | with pytest.raises(KeyboardInterrupt): |
| 187 | get(dsk, "x", pool=pool) |
| 188 | clog_event.set() |
| 189 | interrupter.join() |