asyncio：适用于 Python 3.10 但不适用于 Python 3.8

Question

考虑以下代码：

import asyncio

sem: asyncio.Semaphore = asyncio.Semaphore(2)


async def async_run() -> None:
    async def async_task() -> None:
        async with sem:
            await asyncio.sleep(1)
            print('spam')

    await asyncio.gather(*[async_task() for _ in range(3)])


asyncio.run(async_run())

使用 Python 3.10.6 (Fedora 35) 运行，就像教科书中一样。

但是，当我使用 Python 3.8.10 (Ubuntu 20.04) 运行它时，出现以下错误：

Traceback (most recent call last):
  File "main.py", line 21, in <module>
    asyncio.run(async_run())
  File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "main.py", line 18, in async_run
    print(future_entry_index, await future_entry)
  File "/usr/lib/python3.8/asyncio/tasks.py", line 619, in _wait_for_one
    return f.result()  # May raise f.exception().
  File "main.py", line 11, in async_task
    async with sem:
  File "/usr/lib/python3.8/asyncio/locks.py", line 97, in __aenter__
    await self.acquire()
  File "/usr/lib/python3.8/asyncio/locks.py", line 496, in acquire
    await fut
RuntimeError: Task <Task pending name='Task-4' coro=<async_run.<locals>.async_task() running at main.py:11> cb=[as_completed.<locals>._on_completion() at /usr/lib/python3.8/asyncio/tasks.py:606]> got Future <Future pending> attached to a different loop

导致错误的是 async with sem 行和 Semaphore 对象。没有它，一切都没有错误，但不是我想要的方式。

我无法在任何地方提供 loop 参数，因为即使在允许的地方，它自 Python 3.8 以来已被弃用并在 Python 3.10 中被删除。

如何使代码与 Python 3.8 一起工作？

更新。瞥一眼 asyncio 代码就会发现 Python 版本差异很大。然而，信号量不能在 3.8 中被破坏，对吧？

Answer 1

正如this answer 中所讨论的，pre-python 3.10 Semaphore 根据当前运行的循环在__init__ 上设置它的循环，而asyncio.run 开始一个新的循环。因此，当您尝试 async.run 您的 coros 时，您使用的循环与定义的 Semaphore 不同，正确的错误消息实际上是 got Future <Future pending> attached to a different loop。

幸运的是，让代码在两个 python 版本上工作并不难：

解决方案 1

不要创建新循环，使用现有循环来运行您的函数：

import asyncio

sem: asyncio.Semaphore = asyncio.Semaphore(value=2)


async def async_task() -> None:
    async with sem:
        await asyncio.sleep(1)
        print(f"spam {sem._value}")

async def async_run() -> None:
    await asyncio.gather(*[async_task() for _ in range(3)])

loop = asyncio.get_event_loop()
loop.run_until_complete(async_run())
loop.close()

方案二

在 asyncio.run 创建的循环中初始化信号量对象：

import asyncio

async def async_task2(sem) -> None:
    async with sem:
        await asyncio.sleep(1)
        print(f"spam {sem._value}")

async def async_run2() -> None:
    sem = asyncio.Semaphore(2)
    await asyncio.gather(*[async_task2(sem) for _ in range(3)])


asyncio.run(async_run2())

sn-ps 都适用于 python3.8 和 python3.10。想必他们removed the loop parameter from most of asyncio in python 3.10就是因为这样的怪事。

比较 3.8 和 3.10 中信号量的 __init__：

Python 3.8

class Semaphore(_ContextManagerMixin):
    """A Semaphore implementation.
    A semaphore manages an internal counter which is decremented by each
    acquire() call and incremented by each release() call. The counter
    can never go below zero; when acquire() finds that it is zero, it blocks,
    waiting until some other thread calls release().
    Semaphores also support the context management protocol.
    The optional argument gives the initial value for the internal
    counter; it defaults to 1. If the value given is less than 0,
    ValueError is raised.
    """

    def __init__(self, value=1, *, loop=None):
        if value < 0:
            raise ValueError("Semaphore initial value must be >= 0")
        self._value = value
        self._waiters = collections.deque()
        if loop is None:
            self._loop = events.get_event_loop()
        else:
            self._loop = loop
            warnings.warn("The loop argument is deprecated since Python 3.8, "
                          "and scheduled for removal in Python 3.10.",
                          DeprecationWarning, stacklevel=2)

Python 3.10:

class Semaphore(_ContextManagerMixin, mixins._LoopBoundMixin):
    """A Semaphore implementation.
    A semaphore manages an internal counter which is decremented by each
    acquire() call and incremented by each release() call. The counter
    can never go below zero; when acquire() finds that it is zero, it blocks,
    waiting until some other thread calls release().
    Semaphores also support the context management protocol.
    The optional argument gives the initial value for the internal
    counter; it defaults to 1. If the value given is less than 0,
    ValueError is raised.
    """

    def __init__(self, value=1, *, loop=mixins._marker):
        super().__init__(loop=loop)
        if value < 0:
            raise ValueError("Semaphore initial value must be >= 0")
        self._value = value
        self._waiters = collections.deque()
        self._wakeup_scheduled = False