Class asyncio.Semaphore has dequeue in implementation for waiters of semaphore and seems like the intention of this dequeue is to maintain FIFO queue of waiters. But it doesn't work, coroutine that releases semaphore can acquire semaphore again. Below is reproducible example:

import asyncio


async def process(idx: int, semaphore: asyncio.Semaphore) -> None:
    while True:
        async with semaphore:
            print(f'ACQUIRE {idx}')
            await asyncio.sleep(1)


async def main() -> None:
    semaphore = asyncio.Semaphore(5)
    await asyncio.gather(*[process(idx, semaphore) for idx in range(20)])

asyncio.run(main())

In console:

/home/y.hyzyla/Work/asynciosemaphorebug/venv/bin/python /home/y.hyzyla/Work/asynciosemaphorebug/main.py
ACQUIRE 0
ACQUIRE 1
ACQUIRE 2
ACQUIRE 3
ACQUIRE 4
ACQUIRE 0
ACQUIRE 1
ACQUIRE 2
ACQUIRE 3
ACQUIRE 4
ACQUIRE 0
ACQUIRE 1
ACQUIRE 2
ACQUIRE 3
ACQUIRE 4
....

Ugly fix, is to add asyncio.sleep right before semaphore.

    while True:
        await asyncio.sleep(0)
        async with semaphore:
            ...

Also, I found a comment on Stack Overflow about race condition in Semaphore implementation, but I don't know about the quality of that comment:

https://stackoverflow.com/questions/55951233/does-pythons-asyncio-lock-acquire-maintain-order#comment112978366_55951304

Good point.
Currently, asyncio lock objects don't provide a strong FIFO guarantee.

In a tight loop, a task can re-acquire the lock just after releasing even if there are pending waiters that were scheduled earlier. It's true also for Lock, Conditional, Event, etc.

The solution requires async release method. Since the change is not backward compatible, a new method should be added, e.g. await sem.release_and_wait() for endorsing the context switch if there are pending waiters.

async context manager can be modified for using the new method without backward compatibility problems easily.

A hero who can help is welcome!

Or, maybe, there is a way to do everything without changing public API.

The idea is: _wake_up_next can create a future which is set by *waked up task* on its acquiring. acquire method should wait for this future first before entering in `while self._value < 0:` loop.

If the future is cancelled, _wake_up_next should be called again and waiting for the next future acquiring to be performed.

If there is no *acquire waiting* future exists -- do everything as usual.

All other lock objects should be modified also.

Thanks for response!

A hero who can help is welcome!

I will try to make PR :raising_hand

Andrew, the same problem exists in asyncio.Queue which is also critical. Here's how I fixed it in edgedb code base: https://github.com/edgedb/edgedb/blob/08e41341024828df22a01cd690b11fcff00bca5e/edb/server/compiler_pool/queue.py#L51-L74

Thanks, Yuri.
I'll take a look.

New changeset 32e7715 by Andrew Svetlov in branch 'main':
bpo-45997: Fix asyncio.Semaphore re-acquiring order (GH-31910)
32e7715

New changeset 9d59381 by Miss Islington (bot) in branch '3.10':
[3.10] bpo-45997: Fix asyncio.Semaphore re-acquiring order (GH-31910) (bpo-32047)
9d59381

New changeset f47984b by Andrew Svetlov in branch '3.9':
[3.9] bpo-45997: Fix asyncio.Semaphore re-acquiring order (GH-31910) (GH-32049)
f47984b

Maybe not the end of story.

This shows how to create a broken semaphore (on commit 1971014):

from asyncio import CancelledError
from asyncio import Semaphore
from asyncio import create_task
from asyncio import gather
from asyncio import run
from asyncio import sleep

async def process(idx, sem, tasks):
    await sleep(idx)
    async with sem:
        await sleep(2)
    tasks[1].cancel()

async def main():
    print('let me run for 3 seconds')
    sem = Semaphore(1)
    tasks = []
    for idx in range(2):
        tasks.append(create_task(process(idx, sem, tasks)))
    await sleep(3)
    await gather(*tasks, return_exceptions=True)

    print(f'sem._value == {sem._value}')
    print('sem is still...', end='', flush=True)
    async with sem:
        print('working!')

run(main())

Expected output:

let me run for 3 seconds
sem._value == 1
sem is still...working!

Actual output:

let me run for 3 seconds
sem._value == 1
sem is still...

PR 93222 has been made for this. Might be helpful.

I was going through the changes trying to understand how all this works in various cases and I noticed that the locked method was not adjusted.

It currently reads:

def locked(self):
    """Returns True if semaphore can not be acquired immediately."""
    return self._value == 0

but after the latest changes that added the _wakeup_scheduled flag should't it actually read:

def locked(self):
    """Returns True if semaphore can not be acquired immediately."""
    return self._wakeup_scheduled or self._value == 0

?

Since the comment indicates that this method returns True if the semaphore cannot be acquired immediately and considering that the condition in the acquire method that prevents an immediate acquiring of the semaphore is:

    while self._wakeup_scheduled or self._value <= 0: ...

it leads me to think that locked method needs to be adjusted to have the same condition.

Someone with a better understanding of how this is supposed to work, please weigh in.