GeezerVison

import asyncio
import threading


def func():
    loop = asyncio.EventLoop()

    async def coro():
        await asyncio.sleep(0.01)

    lock = threading.Lock()
    count = 0
    async def main():
        nonlocal count
        while True:
            with lock:
                asyncio.create_task(coro())
                count = len(asyncio.all_tasks(loop))

    th = threading.Thread(target=loop.run_until_complete, args=(main(),))
    th.start()

    def check():
        with lock:
            assert count == len(asyncio.all_tasks(loop))

    threads = [threading.Thread(target=check) for _ in range(10)]
    for t in threads:
        t.start()

    for t in threads:
        t.join()


func()

❯ ./python -m test test_asyncio.test_free_threading -F
Using random seed: 2944425156
0:00:00 load avg: 0.61 Run tests sequentially in a single process
0:00:00 load avg: 0.61 [  1] test_asyncio.test_free_threading
Warning -- Uncaught thread exception: AssertionError
Exception in thread Thread-5 (check):
Warning -- Uncaught thread exception: AssertionError
Exception in thread Thread-2 (check):
Traceback (most recent call last):
Traceback (most recent call last):
  File "/home/realkumaraditya/cpython/Lib/threading.py", line 1054, in _bootstrap_inner
    self.run()
    ~~~~~~~~^^
  File "/home/realkumaraditya/cpython/Lib/threading.py", line 996, in run
    self._target(*self._args, **self._kwargs)
    ~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/realkumaraditya/cpython/Lib/test/test_asyncio/test_free_threading.py", line 86, in check
    self.assertSetEqual(tasks & self.all_tasks(loop), tasks)
    ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/realkumaraditya/cpython/Lib/unittest/case.py", line 1176, in assertSetEqual
    self.fail(self._formatMessage(msg, standardMsg))
    ~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/realkumaraditya/cpython/Lib/unittest/case.py", line 741, in fail
    raise self.failureException(msg)
AssertionError: Items in the second set but not the first:
<Task pending name='Task-1' coro=<TestFreeThreading.test_all_tasks_different_thread.<locals>.main() running at /home/realkumaraditya/cpython/Lib/test/test_asyncio/test_free_threading.py:77> cb=[_run_until_complete_cb() at /home/realkumaraditya/cpython/Lib/asyncio/base_events.py:181]>