> Would you prepare a patch for number 3?

I will give it a try and see what I come up with.

> I am afraid we can add another hard-to-debug multi-threaded problem by complicating the data structure.

Yeah this was my concern too, the adding and removing from the WeakDict[AbstractEventLoop, WeakSet[Task]] for `_all_tasks` could still cause issues. Specifically the whole WeakSet class is not threadsafe so I would assume WeakDict is the same, there may not be a nice way of ensuring a combination of GC + the IterationGuard doesn't come and mess up the dict even if I wrap it in a threading lock.

Another option would be to have the WeakSet[Task] attached to the loop itself then because using the same loop in multiple threads not at all thread safe already that would contain the problem. You mentioned "third-party loops" which may make this option impossible.

> I'm just curious why do you call `all_tasks()` at all?
> In my mind, the only non-debug usage is `asyncio.run()`

In reality we aren't using `all_tasks()` directly. We are calling `asyncio.run()` from multiple threads which triggers the issue. The repro I provided was just a more reliable way of triggering the issue. I will paste a slightly more real-world example of how this happened below. This version is a little more messy and harder to see exactly what the problem is which is why I started with the other one.

```
import asyncio
from threading import Thread


async def do_nothing(n=0):
    await asyncio.sleep(n)


async def loop_tasks():
    loop = asyncio.get_event_loop()
    while True:
        loop.create_task(do_nothing())
        await asyncio.sleep(0.01)


async def make_tasks(n):
    loop = asyncio.get_event_loop()
    for i in range(n):
        loop.create_task(do_nothing(1))
    await asyncio.sleep(1)


def make_lots_of_tasks():
    while True:
        asyncio.run(make_tasks(10000))


for i in range(10):
    t = Thread(target=make_lots_of_tasks)
    t.start()

asyncio.run(loop_tasks())
```