[ovirt-devel] [VDSM] [PATCH] Add the missing locked() interface

[ybronhei]

On 05/07/2014 07:48 PM, Nir Soffer wrote:
>
> threading.Lock has a little known locked() method, documented in
> https://docs.python.org/2.6/library/thread.html#thread.lock.locked
>
> This method is not very useful, but since pthreading.Lock must be
> drop-in replacment for threading.Lock, we must implement it.
>
> This patch adds the missing method, implementing it in the same way
> Python implemnts it.  Since RLock does not have this method, RLock does
> not extend Lock now.
>
> Signed-off-by: Nir Soffer <nsoffer at redhat.com>
> ---
>   pthreading.py | 18 ++++++++++++++----
>   tests.py      |  7 +++++++
>   2 files changed, 21 insertions(+), 4 deletions(-)
>
> diff --git a/pthreading.py b/pthreading.py
> index 2e9e2d6..f1ea056 100644
> --- a/pthreading.py
> +++ b/pthreading.py
> @@ -51,9 +51,9 @@ import os
>   import pthread
>
>
> -class Lock(pthread.Mutex):
> +class _Lock(pthread.Mutex):
>       """
> -    Lock class mimics Python native threading.Lock() API on top of
> +    _Lock class mimics Python native threading.Lock() API on top of
>       the POSIX thread mutex synchronization primitive.
>       """
>       def __enter__(self):
> @@ -78,9 +78,19 @@ class Lock(pthread.Mutex):
>           self.unlock()
>
>
> -class RLock(Lock):
> +class Lock(_Lock):
> +    def locked(self):
> +        # Yes, this is horrible hack, and the same one used by Python
> +        # threadmodule.c. But this is part of Python lock interface.
> +        if self.acquire(blocking=False):
> +            self.release()
> +            return False
> +        return True
> +
> +
> +class RLock(_Lock):
>       def __init__(self):
> -        pthread.Mutex.__init__(self, recursive=True)
> +        _Lock.__init__(self, recursive=True)
>
>
>   class Condition(object):
> diff --git a/tests.py b/tests.py
> index d651288..f4c9746 100644
> --- a/tests.py
> +++ b/tests.py
> @@ -60,6 +60,13 @@ class LockTests(TestCaseBase):
>           self.assertTrue(lock.acquire())
>           self.assertTrue(lock.acquire(False))
>
> +    def testLocked(self):
> +        lock = pthreading.Lock()
> +        self.assertFalse(lock.locked())
> +        with lock:
> +            self.assertTrue(lock.locked())
> +        self.assertFalse(lock.locked())
> +
>
>   class Flag(object):
>       def __init__(self):
>
+1
thanks

-- 
Yaniv Bronhaim.