Consider the following code:

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())

Run with Python 3.10.6 (Fedora 35), it works just like in the schoolbook.

However, when I run it with Python 3.8.10 (Ubuntu 20.04), I get the following error:

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

It's async with sem line and the Semaphore object that cause the error. Without it, everything works without errors, but not the way I want it to.

I can't provide the loop parameter anywhere, for even where it's allowed, it has been deprecated since Python 3.8 and removed in Python 3.10.

How to make the code work with Python 3.8?

Update. A glimpse at the asyncio code showed that the Python versions differ a lot. However, the Semaphores can't be just broken in 3.8, right?

CodePudding user response：

As discussed in this answer, pre-python 3.10 Semaphore sets its loop on __init__ based on the current running loop, while asyncio.run starts a new loop. And so, when you try and async.run your coros, you are using a different loop than your Semaphore is defined on, for which the correct error message really is got Future <Future pending> attached to a different loop.

Fortunately, making the code work on both python versions is not too hard:

Solution 1

Don't make a new loop, use the existing loop to run your function:

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()

Solution 2

Initialize the semaphore object within the loop created by 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())

Both snippets work on python3.8 and python3.10. Presumably it was because of weirdness like this that they removed the loop parameter from most of asyncio in python 3.10.

Compare the __init__ for semaphore from 3.8 compared to 3.10:

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