Change in vdsm[master]: migration: Add migrateChangeGlobalParams verb

mbetak at redhat.com mbetak at redhat.com
Fri Oct 16 16:29:22 UTC 2015


Martin Betak has uploaded a new change for review.

Change subject: migration: Add migrateChangeGlobalParams verb
......................................................................

migration: Add migrateChangeGlobalParams verb

Added new verb for setting values of incoming and outgoing migration semaphores.

Added a custom semaphore class that allows for dynamic changes to its value.

Wiki: http://www.ovirt.org/Features/Migration_Enhancements
Change-Id: I4f6d1bcdc29f144d9fcf28a085b7014127cc4f41
Signed-off-by: Martin Betak <mbetak at redhat.com>
---
M lib/vdsm/define.py
M vdsm/API.py
M vdsm/clientIF.py
M vdsm/virt/migration.py
M vdsm/virt/utils.py
5 files changed, 81 insertions(+), 10 deletions(-)


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

diff --git a/lib/vdsm/define.py b/lib/vdsm/define.py
index aee6f09..5d47c6c 100644
--- a/lib/vdsm/define.py
+++ b/lib/vdsm/define.py
@@ -196,6 +196,9 @@
     'migrationLimit': {'status': {
         'code': 80,
         'message': 'Incoming migration limit exceeded'}},
+    'migrateChangeParamsErr': {'status': {
+        'code': 81,
+        'message': 'Error setting migration parameters'}},
     'recovery': {'status': {
         'code': 99,
         'message': 'Recovering from crash or Initializing'}},
diff --git a/vdsm/API.py b/vdsm/API.py
index 8fc9413..b49708e 100644
--- a/vdsm/API.py
+++ b/vdsm/API.py
@@ -594,6 +594,32 @@
         return {'status': doneCode, 'migrationPort': 0,
                 'params': result['vmList']}
 
+    def migrateChangeGlobalParams(self, params):
+        """
+        Change parameters that apply to all migrations.
+
+        :param params: a dictionary containing:
+            *max_outgoing_migrations* - maximum concurrent outgoing migrations
+        """
+
+        try:
+            max_outgoing = params.get('max_outgoing_migrations')
+            if max_outgoing is not None:
+                self.log('Updating max_outgoing_migrations to %s',
+                         max_outgoing)
+                migration.SourceThread.ongoingMigrations.setValue(max_outgoing)
+
+            max_incoming = params.get('max_incoming_migrations')
+            if max_incoming is not None:
+                self.log('Updating max_incoming_migrations to %s',
+                         max_incoming)
+                migration.incomingMigrations.setValue(max_incoming)
+
+            return response.success()
+        except migration.MigrationConfigurationError:
+            return response.error('migrateChangeParamsErr')
+
+
     def monitorCommand(self, command):
         """
         Send a monitor command to the specified VM and wait for the answer.
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index aca99b2..4f2c164 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -464,7 +464,7 @@
             # API response.
             mog = min(config.getint('vars', 'max_outgoing_migrations'),
                       caps.CpuTopology().cores())
-            migration.SourceThread.setMaxOutgoingMigrations(mog)
+            migration.SourceThread.ongoingMigrations.setValue(mog)
 
             # Recover stage 1: domains from libvirt
             doms = getVDSMDomains()
diff --git a/vdsm/virt/migration.py b/vdsm/virt/migration.py
index c8d87e5..b583107 100644
--- a/vdsm/virt/migration.py
+++ b/vdsm/virt/migration.py
@@ -33,6 +33,7 @@
 from vdsm.compat import pickle
 from vdsm.config import config
 from vdsm.define import NORMAL, Mbytes
+from virt.utils import DynamicSemaphore
 from yajsonrpc import \
     JsonRpcNoResponseError, \
     JsonRpcBindingsError
@@ -53,9 +54,15 @@
 VIR_MIGRATE_PARAM_BANDWIDTH = 'bandwidth'
 VIR_MIGRATE_PARAM_GRAPHICS_URI = 'graphics_uri'
 
-incomingMigrations = threading.BoundedSemaphore(
+incomingMigrations = DynamicSemaphore(
     min(config.getint('vars', 'max_incoming_migrations'),
         caps.CpuTopology().cores()))
+
+
+class MigrationConfigurationError(RuntimeError):
+    """
+    Failed to set requested global migration option(s)
+    """
 
 
 class MigrationDestinationSetupError(RuntimeError):
@@ -68,14 +75,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 = DynamicSemaphore(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 c4f47b5..29868c1 100644
--- a/vdsm/virt/utils.py
+++ b/vdsm/virt/utils.py
@@ -159,3 +159,45 @@
         return wrapper
 
     return decorator
+
+
+class DynamicSemaphore(threading.Semaphore):
+    """
+    Extends standard semaphore with the ability to alter the current "bound"
+    by external actors.
+
+    Translates setValue(n) calls to appropriate sequence of acquire() or
+    release() calls so the new bound would be achieved.
+    """
+
+    def __init__(self, value, *args, **kwargs):
+        self.lock = threading.Lock()
+        self.value = value
+        super(DynamicSemaphore, self).__init__(value, *args, **kwargs)
+
+    def setValue(self, newValue):
+        with self.lock:
+            if self.value == newValue:
+                return
+
+            if newValue <= 0:
+                raise ValueError("semaphore value must be >= 0")
+
+            delta = newValue - self.value
+            if delta > 0:
+                self._increase(delta)
+            else:
+                self._decreaseAsync(abs(delta))
+
+            self.value = newValue
+
+    def _increase(self, by):
+        for i in range(by):
+            self.release()
+
+    def _decrease(self, by):
+        for i in range(by):
+            self.acquire()
+
+    def _decreaseAsync(self, by):
+        threading.Thread(target=self._decrease, args=(by,)).start()


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

Gerrit-MessageType: newchange
Gerrit-Change-Id: I4f6d1bcdc29f144d9fcf28a085b7014127cc4f41
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