Benchmark async context switching
(num_tests, num_fibers)
| 78 | |
| 79 | |
| 80 | async def test_context_switch_async(num_tests, num_fibers): |
| 81 | """Benchmark async context switching""" |
| 82 | global swap_count |
| 83 | total_ops = 0 |
| 84 | |
| 85 | for _ in range(num_tests): |
| 86 | swap_count = 0 |
| 87 | stop_event = asyncio.Event() |
| 88 | fibers = [asyncio.create_task(worker(stop_event)) for _ in range(num_fibers)] |
| 89 | |
| 90 | start_time = runtime_time() |
| 91 | while (runtime_time() - start_time) < 1000: |
| 92 | await asyncio.sleep(0) # give control to other tasks |
| 93 | |
| 94 | total_ops += swap_count |
| 95 | swap_count = None |
| 96 | stop_event.set() |
| 97 | await asyncio.gather(*fibers) |
| 98 | |
| 99 | return int(total_ops / num_tests) |
| 100 | |
| 101 | |
| 102 | # ---------------------- Math Operations ---------------------- |
no test coverage detected