Change in vdsm[master]: rwlock: Support timed acquire

Thursday, 25 June 2015

Nir Soffer has uploaded a new change for review.

Change subject: rwlock: Support timed acquire
......................................................................

rwlock: Support timed acquire

Add acquire timeout - with timeout=0, can be used to implement
non-blocking acquire. With bigger timeout, make it easy to fail a
request after a timeout, instead of leaving a blocked thread forever.

Change-Id: Idd9b11452d74b566b8989f41244f2d0534327214
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com&gt;
---
M lib/vdsm/rwlock.py
M tests/rwlock_test.py
2 files changed, 65 insertions(+), 21 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/09/42909/1

diff --git a/lib/vdsm/rwlock.py b/lib/vdsm/rwlock.py
index 900d91d..72f962c 100644
--- a/lib/vdsm/rwlock.py
+++ b/lib/vdsm/rwlock.py
@@ -22,6 +22,10 @@
 import threading
 
 
+class Timeout(Exception):
+    pass
+
+
 class RWLock(object):
 
     def __init__(self):
@@ -30,16 +34,16 @@
         self._readers = set()
         self._writer = None
 
-    def acquireWrite(self):
+    def acquireWrite(self, timeout=None):
         with self._lock:
             if self._writer or self._readers or self._waiters:
-                self._wait(True)
+                self._wait(True, timeout)
             self._writer = threading.current_thread()
 
-    def acquireRead(self):
+    def acquireRead(self, timeout=None):
         with self._lock:
             if self._writer or self._waiters:
-                self._wait(False)
+                self._wait(False, timeout)
             self._readers.add(threading.current_thread())
 
     def release(self):
@@ -60,13 +64,13 @@
             if self._waiters:
                 self._wakeup_waiter()
 
-    def _wait(self, wants_write):
+    def _wait(self, wants_write, timeout):
         waiter = Waiter(wants_write)
         self._waiters.append(waiter)
         try:
             self._lock.release()
             try:
-                waiter.wait()
+                waiter.wait(timeout)
             finally:
                 self._lock.acquire()
         finally:
@@ -84,8 +88,10 @@
         self.wants_write = wants_write
         self._event = threading.Event()
 
-    def wait(self):
-        self._event.wait()
+    def wait(self, timeout):
+        if not self._event.wait(timeout):
+            raise Timeout("Timeout acquiring lock for %s" %
+                          "writing" if self.wants_write else
"reading")
 
     def wakeup(self):
         self._event.set()
diff --git a/tests/rwlock_test.py b/tests/rwlock_test.py
index e67e27a..5ddf066 100644
--- a/tests/rwlock_test.py
+++ b/tests/rwlock_test.py
@@ -25,10 +25,40 @@
 from testlib import VdsmTestCase
 from testValidation import slowtest, stresstest
 
-from vdsm.rwlock import RWLock
+from vdsm.rwlock import RWLock, Timeout
 
 
 class RWLockTests(VdsmTestCase):
+
+    def test_writer_blocks_other_writer(self):
+        lock = RWLock()
+        lock.acquireWrite()
+        try:
+            log = []
+            start_thread(writer, lock, log, timeout=0).join()
+            self.assertEqual(log, ["writer timeout"])
+        finally:
+            lock.release()
+
+    def test_writer_blocks_other_reader(self):
+        lock = RWLock()
+        lock.acquireWrite()
+        try:
+            log = []
+            start_thread(reader, lock, log, timeout=0).join()
+            self.assertEqual(log, ["reader timeout"])
+        finally:
+            lock.release()
+
+    def test_reader_blocks_other_writer(self):
+        lock = RWLock()
+        lock.acquireRead()
+        try:
+            log = []
+            start_thread(writer, lock, log, timeout=0).join()
+            self.assertEqual(log, ["writer timeout"])
+        finally:
+            lock.release()
 
     @slowtest
     def test_concurrent_readers(self):
@@ -194,22 +224,30 @@
         self.assertEqual(log, ["reader acquired"])
 
 
-def writer(lock, log, hold=0):
-    lock.acquireWrite()
+def writer(lock, log, timeout=1, hold=0):
     try:
-        log.append("writer acquired")
-        time.sleep(hold)
-    finally:
-        lock.release()
+        lock.acquireWrite(timeout)
+    except Timeout:
+        log.append("writer timeout")
+    else:
+        try:
+            log.append("writer acquired")
+            time.sleep(hold)
+        finally:
+            lock.release()
 
 
-def reader(lock, log, hold=0):
-    lock.acquireRead()
+def reader(lock, log, timeout=1, hold=0):
     try:
-        log.append("reader acquired")
-        time.sleep(hold)
-    finally:
-        lock.release()
+        lock.acquireRead(timeout)
+    except Timeout:
+        log.append("reader timeout")
+    else:
+        try:
+            log.append("reader acquired")
+            time.sleep(hold)
+        finally:
+            lock.release()
 
 
 def start_thread(func, *args, **kwargs):


-- 
To view, visit https://gerrit.ovirt.org/42909
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Idd9b11452d74b566b8989f41244f2d0534327214
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com&gt;

    

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011