classification
Title: asyncio.create_task weakrefset race condition
Type: Stage: resolved
Components: asyncio Versions: Python 3.11, Python 3.10, Python 3.9
process
Status: closed Resolution: fixed
Dependencies: Superseder:
Assigned To: Nosy List: asvetlov, bjs, graingert, lukasz.langa, miss-islington, pitrou, yselivanov
Priority: normal Keywords: patch

Created on 2021-08-20 14:07 by graingert, last changed 2021-08-28 18:55 by lukasz.langa. This issue is now closed.

Pull Requests
URL Status Linked Edit
PR 27909 closed graingert, 2021-08-23 18:13
PR 27921 merged graingert, 2021-08-23 19:54
PR 28013 merged miss-islington, 2021-08-28 17:07
PR 28014 merged miss-islington, 2021-08-28 17:07
Messages (9)
msg399969 - (view) Author: Thomas Grainger (graingert) * Date: 2021-08-20 14:07
with the following demo script I can get a IndexError: pop from empty list

```
import itertools
import asyncio
import concurrent.futures
import sys
import threading

threads = 200

def test_all_tasks_threading() -> None:
    async def foo() -> None:
        await asyncio.sleep(0)

    async def create_tasks() -> None:
        for i in range(1000):
            asyncio.create_task(foo())

        await asyncio.sleep(0)

    results = []
    with concurrent.futures.ThreadPoolExecutor(threads) as tpe:
        for f in concurrent.futures.as_completed(
            tpe.submit(asyncio.run, create_tasks()) for i in range(threads)
        ):
            results.append(f.result())
    assert results == [None] * threads


def main():
    for i in itertools.count():
        test_all_tasks_threading()
        print(f"worked {i}")
    return 0


if __name__ == "__main__":
    sys.exit(main())
```



```
worked 0
worked 1
worked 2
worked 3
worked 4
worked 5
worked 6
worked 7
worked 8
worked 9
worked 10
worked 11
worked 12
worked 13
worked 14
worked 15
worked 16
worked 17
worked 18
Traceback (most recent call last):
  File "/home/graingert/projects/asyncio-demo/demo.py", line 36, in <module>
    sys.exit(main())
  File "/home/graingert/projects/asyncio-demo/demo.py", line 30, in main
    test_all_tasks_threading()
  File "/home/graingert/projects/asyncio-demo/demo.py", line 24, in test_all_tasks_threading
    results.append(f.result())
  File "/usr/lib/python3.9/concurrent/futures/_base.py", line 438, in result
    return self.__get_result()
  File "/usr/lib/python3.9/concurrent/futures/_base.py", line 390, in __get_result
    raise self._exception
  File "/usr/lib/python3.9/concurrent/futures/thread.py", line 52, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/usr/lib/python3.9/asyncio/runners.py", line 48, in run
    loop.run_until_complete(loop.shutdown_asyncgens())
  File "/usr/lib/python3.9/asyncio/base_events.py", line 621, in run_until_complete
    future = tasks.ensure_future(future, loop=self)
  File "/usr/lib/python3.9/asyncio/tasks.py", line 667, in ensure_future
    task = loop.create_task(coro_or_future)
  File "/usr/lib/python3.9/asyncio/base_events.py", line 433, in create_task
    task = tasks.Task(coro, loop=self, name=name)
  File "/usr/lib/python3.9/_weakrefset.py", line 84, in add
    self._commit_removals()
  File "/usr/lib/python3.9/_weakrefset.py", line 57, in _commit_removals
    discard(l.pop())
IndexError: pop from empty list
sys:1: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_asyncgens' was never awaited
Task was destroyed but it is pending!
task: <Task pending name='Task-74178' coro=<BaseEventLoop.shutdown_asyncgens() running at /usr/lib/python3.9/asyncio/base_events.py:530>>
```

here's a live demo on github actions: https://github.com/graingert/asyncio-backport/runs/3380502247#step:5:90
msg399971 - (view) Author: Thomas Grainger (graingert) * Date: 2021-08-20 14:11
still happens on 3.10 even though there's an `with _IterationGuard`

```
worked 0
Traceback (most recent call last):
  File "/home/graingert/projects/asyncio-demo/demo.py", line 36, in <module>
    sys.exit(main())
  File "/home/graingert/projects/asyncio-demo/demo.py", line 30, in main
    test_all_tasks_threading()
  File "/home/graingert/projects/asyncio-demo/demo.py", line 24, in test_all_tasks_threading
    results.append(f.result())
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 438, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 390, in __get_result
    raise self._exception
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 52, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/usr/lib/python3.10/asyncio/runners.py", line 47, in run
    _cancel_all_tasks(loop)
  File "/usr/lib/python3.10/asyncio/runners.py", line 56, in _cancel_all_tasks
    to_cancel = tasks.all_tasks(loop)
  File "/usr/lib/python3.10/asyncio/tasks.py", line 53, in all_tasks
    tasks = list(_all_tasks)
  File "/usr/lib/python3.10/_weakrefset.py", line 60, in __iter__
    with _IterationGuard(self):
  File "/usr/lib/python3.10/_weakrefset.py", line 33, in __exit__
    w._commit_removals()
  File "/usr/lib/python3.10/_weakrefset.py", line 57, in _commit_removals
    discard(l.pop())
IndexError: pop from empty list
```
msg399972 - (view) Author: Thomas Grainger (graingert) * Date: 2021-08-20 14:12
interestingly 3.10 didn't show:

sys:1: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_asyncgens' was never awaited
msg399973 - (view) Author: Ben (bjs) * Date: 2021-08-20 14:26
I can reproduce on 3.9.6

A little digging and it seems asyncio imports Task from _asyncio
and _asyncio's implementation (in asynciomodule.c) of Task has an __init__ which adds the task to the `all_tasks` weakref.WeakSet
which appears to be implemented in Python (in Lib/_weakrefset.py)

weakref.WeakSet is not thread-safe,  which means concurrent create_task's in different threads (even on separate event loops) is not safe.
msg400175 - (view) Author: Thomas Grainger (graingert) * Date: 2021-08-23 20:12
> weakref.WeakSet is not thread-safe,  which means concurrent create_task's in different threads (even on separate event loops) is not safe.

actually it looks like WeakSet is *supposed* to be thread-safe


```
import patchy

patchy.patch(
    "weakref:WeakSet._commit_removals",
    """\
    @@ -1,5 +1,10 @@
     def _commit_removals(self):
    -    l = self._pending_removals
    +    pop = self._pending_removals.pop
         discard = self.data.discard
    -    while l:
    -        discard(l.pop())
    +    while True:
    +        try:
    +            item = pop()
    +        except IndexError:
    +            return
    +        else:
    +            discard(item)
    """
)

import itertools
import asyncio
import concurrent.futures
import sys
import threading

threads = 200

def test_all_tasks_threading() -> None:
    async def foo() -> None:
        await asyncio.sleep(0)

    async def create_tasks() -> None:
        for i in range(1000):
            asyncio.create_task(foo())

        await asyncio.sleep(0)

    results = []
    with concurrent.futures.ThreadPoolExecutor(threads) as tpe:
        for f in concurrent.futures.as_completed(
            tpe.submit(asyncio.run, create_tasks()) for i in range(threads)
        ):
            results.append(f.result())
    assert results == [None] * threads


def main():
    for i in itertools.count():
        test_all_tasks_threading()
        print(f"worked {i}")
    return 0


if __name__ == "__main__":
    sys.exit(main())
```
msg400483 - (view) Author: Łukasz Langa (lukasz.langa) * (Python committer) Date: 2021-08-28 17:07
New changeset 206b21ed9f64fedff67bfea7cf73e423e3e32393 by Thomas Grainger in branch 'main':
bpo-44962: Fix a race in WeakKeyDict, WeakValueDict and WeakSet when two threads attempt to commit the last pending removal (GH-27921)
https://github.com/python/cpython/commit/206b21ed9f64fedff67bfea7cf73e423e3e32393
msg400488 - (view) Author: miss-islington (miss-islington) Date: 2021-08-28 18:09
New changeset 8aa64cc45bff516a6db1f3a3c037cbcce9417fea by Miss Islington (bot) in branch '3.10':
bpo-44962: Fix a race in WeakKeyDict, WeakValueDict and WeakSet when two threads attempt to commit the last pending removal (GH-27921)
https://github.com/python/cpython/commit/8aa64cc45bff516a6db1f3a3c037cbcce9417fea
msg400494 - (view) Author: Łukasz Langa (lukasz.langa) * (Python committer) Date: 2021-08-28 18:54
New changeset 166ad706066a2aad84d0ae5b1594c88904fbb939 by Miss Islington (bot) in branch '3.9':
bpo-44962: Fix a race in WeakKeyDict, WeakValueDict and WeakSet when two threads attempt to commit the last pending removal (GH-27921) (GH-28014)
https://github.com/python/cpython/commit/166ad706066a2aad84d0ae5b1594c88904fbb939
msg400495 - (view) Author: Łukasz Langa (lukasz.langa) * (Python committer) Date: 2021-08-28 18:55
Thanks, Thomas! ✨ 🍰 ✨
History
Date User Action Args
2021-08-28 18:55:38lukasz.langasetstatus: open -> closed
versions: + Python 3.11, - Python 3.7, Python 3.8
messages: + msg400495

resolution: fixed
stage: patch review -> resolved
2021-08-28 18:54:52lukasz.langasetmessages: + msg400494
2021-08-28 18:09:29miss-islingtonsetmessages: + msg400488
2021-08-28 17:07:59miss-islingtonsetpull_requests: + pull_request26457
2021-08-28 17:07:55lukasz.langasetnosy: + lukasz.langa
messages: + msg400483
2021-08-28 17:07:55miss-islingtonsetnosy: + miss-islington
pull_requests: + pull_request26456
2021-08-24 10:12:02graingertsetnosy: + pitrou
2021-08-23 20:12:53graingertsetmessages: + msg400175
2021-08-23 19:54:41graingertsetpull_requests: + pull_request26373
2021-08-23 18:13:17graingertsetkeywords: + patch
stage: patch review
pull_requests: + pull_request26365
2021-08-20 14:26:47bjssetnosy: + bjs
messages: + msg399973
2021-08-20 14:12:57graingertsetmessages: + msg399972
2021-08-20 14:11:27graingertsetmessages: + msg399971
2021-08-20 14:09:44graingertsetversions: + Python 3.10
2021-08-20 14:09:21graingertsetversions: + Python 3.7
2021-08-20 14:07:52graingertcreate