|
1 | 1 | import asyncio |
| 2 | +import contextvars |
| 3 | +import _thread |
2 | 4 | import threading |
3 | 5 | import unittest |
4 | 6 | from threading import Thread |
@@ -232,6 +234,68 @@ def factory(self, loop, coro, **kwargs): |
232 | 234 | @unittest.skipUnless(hasattr(asyncio.tasks, "_c_all_tasks"), "requires _asyncio") |
233 | 235 | class TestCFreeThreading(TestFreeThreading, TestCase): |
234 | 236 |
|
| 237 | + def test_task_created_during_thread_state_clear_is_tracked(self): |
| 238 | + """Tasks created by a thread-state finalizer must not be orphaned.""" |
| 239 | + loop = asyncio.new_event_loop() |
| 240 | + loop_ready = threading.Event() |
| 241 | + task_started = threading.Event() |
| 242 | + task_finished = threading.Event() |
| 243 | + thread_done = threading.Event() |
| 244 | + blocker = [] |
| 245 | + |
| 246 | + async def blocked(): |
| 247 | + task_started.set() |
| 248 | + await blocker[0] |
| 249 | + task_finished.set() |
| 250 | + |
| 251 | + def initialize_loop(): |
| 252 | + blocker.append(loop.create_future()) |
| 253 | + loop_ready.set() |
| 254 | + |
| 255 | + def run_loop(): |
| 256 | + asyncio.set_event_loop(loop) |
| 257 | + loop.call_soon(initialize_loop) |
| 258 | + try: |
| 259 | + loop.run_forever() |
| 260 | + finally: |
| 261 | + loop.close() |
| 262 | + |
| 263 | + loop_thread = Thread(target=run_loop) |
| 264 | + loop_thread.start() |
| 265 | + try: |
| 266 | + self.assertTrue(loop_ready.wait(10)) |
| 267 | + |
| 268 | + created = [] |
| 269 | + |
| 270 | + class CreatesTaskOnClear: |
| 271 | + def __del__(self): |
| 272 | + task = loop.create_task(blocked()) |
| 273 | + created.append(task) |
| 274 | + # loop.create_task() does not wake a loop running in |
| 275 | + # another thread. |
| 276 | + loop.call_soon_threadsafe(lambda: None) |
| 277 | + |
| 278 | + var = contextvars.ContextVar("var") |
| 279 | + |
| 280 | + def worker(): |
| 281 | + var.set(CreatesTaskOnClear()) |
| 282 | + thread_done.set() |
| 283 | + |
| 284 | + _thread.start_new_thread(worker, ()) |
| 285 | + self.assertTrue(thread_done.wait(10)) |
| 286 | + |
| 287 | + self.assertTrue(task_started.wait(10)) |
| 288 | + self.assertEqual(len(created), 1) |
| 289 | + self.assertIn(created[0], asyncio.all_tasks(loop)) |
| 290 | + finally: |
| 291 | + if blocker: |
| 292 | + loop.call_soon_threadsafe(blocker[0].set_result, None) |
| 293 | + if task_started.is_set(): |
| 294 | + self.assertTrue(task_finished.wait(10)) |
| 295 | + loop.call_soon_threadsafe(loop.stop) |
| 296 | + loop_thread.join(10) |
| 297 | + self.assertFalse(loop_thread.is_alive()) |
| 298 | + |
235 | 299 | def setUp(self): |
236 | 300 | self._old_current_task = asyncio.current_task |
237 | 301 | asyncio.current_task = asyncio.tasks.current_task = asyncio.tasks._c_current_task |
|
0 commit comments