Change in vdsm[master]: migration: Add DynamicThrottling semaphores

mbetak at redhat.com mbetak at redhat.com
Tue Feb 9 15:16:14 UTC 2016


Martin Betak has uploaded a new change for review.

Change subject: migration: Add DynamicThrottling semaphores
......................................................................

migration: Add DynamicThrottling semaphores

Change-Id: I67376d0bd990f89e0a013887cef1a0cb05fb855d
Signed-off-by: Martin Betak <mbetak at redhat.com>
---
M tests/vmUtilsTests.py
M vdsm/clientIF.py
M vdsm/virt/migration.py
M vdsm/virt/utils.py
4 files changed, 104 insertions(+), 11 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/04/53304/1

diff --git a/tests/vmUtilsTests.py b/tests/vmUtilsTests.py
index 19e64e7..abf2fbe 100644
--- a/tests/vmUtilsTests.py
+++ b/tests/vmUtilsTests.py
@@ -194,3 +194,29 @@
 
     def _helper(self):
         self.done.set()
+
+
+class DynamicSemaphoreTests(TestCaseBase):
+
+    def setUp(self):
+        self.sem = utils.DynamicBoundedSemaphore(5)
+
+    def test_basic_operations(self):
+        for i in range(5):
+            self.sem.acquire()
+        for i in range(5):
+            self.sem.release()
+        self.assertEquals(5, self.sem._value)
+
+    def test_set_bound_increase(self):
+        self.sem.set_bound(10)
+        for i in range(10):
+            self.sem.acquire()
+        self.assertEquals(0, self.sem._value)
+
+    def test_set_bound_decrease(self):
+        self.sem.set_bound(0)
+        success = self.sem.acquire(blocking=False)
+        self.assertFalse(success, 'It should be not possible to obtain '
+                                  'semaphore with value 0')
+        self.assertEqual(0, self.sem._value)
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index 4f9d85f..39ba2f7 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -471,7 +471,7 @@
             # API response.
             mog = min(config.getint('vars', 'max_outgoing_migrations'),
                       caps.CpuTopology().cores())
-            migration.SourceThread.setMaxOutgoingMigrations(mog)
+            migration.SourceThread.ongoingMigrations.set_bound(mog)
 
             recovery.all_vms(self)
 
diff --git a/vdsm/virt/migration.py b/vdsm/virt/migration.py
index 8ae03ad..090124c 100644
--- a/vdsm/virt/migration.py
+++ b/vdsm/virt/migration.py
@@ -35,6 +35,7 @@
 from vdsm.define import NORMAL, Mbytes, errCode
 from vdsm.sslcompat import sslutils
 from virt.utils import run_async
+from virt.utils import DynamicBoundedSemaphore
 from yajsonrpc import \
     JsonRpcNoResponseError, \
     JsonRpcBindingsError
@@ -56,16 +57,22 @@
 VIR_MIGRATE_PARAM_GRAPHICS_URI = 'graphics_uri'
 
 
-_incomingMigrations = threading.BoundedSemaphore(
+incomingMigrations = DynamicBoundedSemaphore(
     min(config.getint('vars', 'max_incoming_migrations'),
         caps.CpuTopology().cores()))
 
 
 spawn_vm = functools.partial(
     run_async,
-    resource=_incomingMigrations,
+    resource=incomingMigrations,
     error='migrateLimit'
 )
+
+
+class MigrationConfigurationError(RuntimeError):
+    """
+    Failed to set requested global migration option(s)
+    """
 
 
 class MigrationDestinationSetupError(RuntimeError):
@@ -85,14 +92,7 @@
     """
     A thread that takes care of migration on the source vdsm.
     """
-    _ongoingMigrations = threading.BoundedSemaphore(1)
-
-    @classmethod
-    def setMaxOutgoingMigrations(cls, n):
-        """Set the initial value of the _ongoingMigrations semaphore.
-
-        must not be called after any vm has been run."""
-        cls._ongoingMigrations = threading.BoundedSemaphore(n)
+    ongoingMigrations = DynamicBoundedSemaphore(1)
 
     def __init__(self, vm, dst='', dstparams='',
                  mode=MODE_REMOTE, method=METHOD_ONLINE,
diff --git a/vdsm/virt/utils.py b/vdsm/virt/utils.py
index 51b6d54..5ba32b6 100644
--- a/vdsm/virt/utils.py
+++ b/vdsm/virt/utils.py
@@ -173,3 +173,70 @@
     started.wait()
     if starting_error[0] is not None:
         raise AsyncStartError(starting_error[0])
+
+
+class DynamicBoundedSemaphore(object):
+    """
+    Bounded Semaphore with the additional ability
+    to dynamically adjust its bound.
+    """
+
+    def __init__(self, value):
+        self._cond = threading.Condition(threading.Lock())
+        self._value = value
+        self._bound = value
+
+    def acquire(self, blocking=True):
+        """ Same behavior as threading.BoundedSemaphore.acquire """
+        rc = False
+        with self._cond:
+            # to enable runtime adjustment of semaphore bound
+            # we allow the _value counter to reach negative values
+            while self._value <= 0:
+                if not blocking:
+                    break
+                self._cond.wait()
+            else:
+                self._value -= 1
+                rc = True
+        return rc
+
+    __enter__ = acquire
+
+    def release(self):
+        """ Same behavior as threading.BoundedSemaphore.release """
+        with self._cond:
+            if self._value >= self._bound:
+                raise ValueError("Dynamic Semaphore released too many times")
+            self._value += 1
+            self._cond.notify()
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.release()
+
+    def set_bound(self, value):
+        """ Dynamically updates semaphore bound.
+
+        When the the specified value is larger than the previous bound,
+        it releases the semaphore the required number of times (and possibly
+        wakes that number of waiting threads).
+
+        When the specified value is smaller than the previous bound,
+        it simply decreases the current value, which may in doing so become
+        negative. Semaphore with value <= 0 is considered unavailable and
+        appropriate number of `release()` calls or a new `setBound(n)` is
+        required to make it obtainable again.
+
+        """
+        with self._cond:
+            delta = value - self._bound
+            self._bound = value
+            if delta < 0:
+                self._value += delta
+
+        # if we are increasing the bound we need to do this outside of
+        # context manager otherwise release() would deadlock since it also
+        # tries to obtain the lock
+        if delta > 0:
+            for i in range(delta):
+                self.release()


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

Gerrit-MessageType: newchange
Gerrit-Change-Id: I67376d0bd990f89e0a013887cef1a0cb05fb855d
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Martin Betak <mbetak at redhat.com>


More information about the vdsm-patches mailing list