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