| 503 | |
| 504 | |
| 505 | def test_threading(): |
| 506 | import _thread |
| 507 | |
| 508 | def hook(event, args): |
| 509 | if event.startswith(("_thread.", "cpython.PyThreadState", "test.")): |
| 510 | print(event, args) |
| 511 | |
| 512 | sys.addaudithook(hook) |
| 513 | |
| 514 | lock = _thread.allocate_lock() |
| 515 | lock.acquire() |
| 516 | |
| 517 | class test_func: |
| 518 | def __repr__(self): return "<test_func>" |
| 519 | def __call__(self): |
| 520 | sys.audit("test.test_func") |
| 521 | lock.release() |
| 522 | |
| 523 | i = _thread.start_new_thread(test_func(), ()) |
| 524 | lock.acquire() |
| 525 | |
| 526 | handle = _thread.start_joinable_thread(test_func()) |
| 527 | handle.join() |
| 528 | |
| 529 | |
| 530 | def test_threading_abort(): |