Bug report

Bug description:

Maybe I didn't quite understand what this feature did, but I think there's a bug when using thelocked() method with amultiprocessing.[R]Lock.

Here is an example:

import multiprocessing as mpdef acq(lock, event):    lock.acquire()    print(f'Acq: {lock = }')    print(f'Acq: {lock.locked() = }')    event.set()def main():    lock = mp.Lock()    event = mp.Event()    p = mp.Process(target=acq, args=(lock, event))    p.start()    event.wait()    print(f'Main: {lock = }')    print(f'Main: {lock.locked() = }')if __name__ == "__main__":    mp.freeze_support()    main()

output is:

Acq: lock = <Lock(owner=Process-1)>Acq: lock.locked() = TrueMain: lock = <Lock(owner=SomeOtherProcess)>Main: lock.locked() = False

In thelockedmethod, the call toself._semlock._count() != 0 is not appropriate. The internalcount attribute is really used withmultiprocessing.RLock to count number of reentrant calls toacquire for the current thread.
Withmultiprocessing.Lock, thiscount is set to 1 when the lock is acquired (only once).

Whatever, only other threads can obtain this value, but not other processes sharing the[R]Lock.

IMO the test should be replace withself._semlock._is_zero() and the example above should also be add as unit test.

Linked issue/PR

• AcquirerProxy object has no attributelocked #115942
• gh-115942: Addlocked to several multiprocessing locks #115944

CPython versions tested on:

CPython main branch

Operating systems tested on:

macOS

Linked PRs

• gh-132561: Fix the publicmultiprocessing.SemLock.locked method #132586