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