Add multiprocessing.Lock.locked()

This commit is contained in:
Rémi Lapeyre 2020-06-05 15:48:16 +02:00
parent e005ead49b
commit c0a2e2161d
4 changed files with 13 additions and 0 deletions

View File

@ -1301,6 +1301,12 @@ object -- see :ref:`multiprocessing-managers`.
Behavior is the same as in :meth:`threading.Lock.release` except that
when invoked on an unlocked lock, a :exc:`ValueError` is raised.
.. method:: locked()
Return true if the lock is acquired.
.. versionadded:: 3.10
.. class:: RLock()

View File

@ -161,6 +161,9 @@ class Lock(SemLock):
def __init__(self, *, ctx):
SemLock.__init__(self, SEMAPHORE, 1, 1, ctx=ctx)
def locked(self):
return self._semlock._is_zero()
def __repr__(self):
try:
if self._semlock._is_mine():

View File

@ -1230,8 +1230,10 @@ class _TestLock(BaseTestCase):
def test_lock(self):
lock = self.Lock()
self.assertEqual(lock.acquire(), True)
self.assertTrue(lock.locked())
self.assertEqual(lock.acquire(False), False)
self.assertEqual(lock.release(), None)
self.assertFalse(lock.locked())
self.assertRaises((ValueError, threading.ThreadError), lock.release)
def test_rlock(self):

View File

@ -0,0 +1,2 @@
:class:`multiprocessing.Lock` now has a `locked()` method that returns
whether the lock is currently held. Patch contributed by Rémi Lapeyre.