Change in vdsm[master]: scheduler: use single instance
piotr.kliczewski at gmail.com
piotr.kliczewski at gmail.com
Mon Jul 20 11:31:55 UTC 2015
Piotr Kliczewski has uploaded a new change for review.
Change subject: scheduler: use single instance
......................................................................
scheduler: use single instance
We used to have one instance of a scheduler used for periodic
operations. With this patch we prepare code for using single scheduler
in more places like jsonrpc server.
Change-Id: If83eded458f8304d802fcd75839e7a916146918b
Signed-off-by: pkliczewski <piotr.kliczewski at gmail.com>
---
M vdsm/clientIF.py
M vdsm/vdsm
M vdsm/virt/periodic.py
3 files changed, 67 insertions(+), 60 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/25/43825/1
diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index 44a0f96..e12aca7 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -40,6 +40,7 @@
from vdsm import sslutils
from vdsm import libvirtconnection
from vdsm import constants
+from vdsm import schedule
from vdsm import utils
import caps
import blkid
@@ -95,6 +96,8 @@
self.bindings = {}
self._broker_client = None
self._subscriptions = defaultdict(list)
+ self._scheduler = schedule.Scheduler(name="periodic.Scheduler",
+ clock=utils.monotonic_time)
if _glusterEnabled:
self.gluster = gapi.GlusterApi(self, log)
else:
@@ -141,6 +144,10 @@
@property
def ready(self):
return (self.irs is None or self.irs.ready) and not self._recovery
+
+ @property
+ def scheduler(self):
+ return self._scheduler
def notify(self, event_id, **kwargs):
"""
@@ -280,6 +287,7 @@
secret.clear()
self.channelListener.stop()
self._hostStats.stop()
+ self._scheduler.stop()
if self.irs:
return self.irs.prepareForShutdown()
else:
@@ -288,6 +296,8 @@
self._shutdownSemaphore.release()
def start(self):
+ self._scheduler.start()
+
for binding in self.bindings.values():
binding.start()
self.thread = threading.Thread(target=self._reactor.process_requests,
diff --git a/vdsm/vdsm b/vdsm/vdsm
index 67f21a4..ceef742 100755
--- a/vdsm/vdsm
+++ b/vdsm/vdsm
@@ -95,14 +95,15 @@
install_manhole({'irs': irs, 'cif': cif})
cif.start()
- periodic.start(cif)
+ runner = periodic.PeriodicOperationRunner()
+ runner.start(cif)
try:
while running[0]:
sigutils.wait_for_signal()
profile.stop()
finally:
- periodic.stop()
+ runner.stop()
cif.prepareForShutdown()
diff --git a/vdsm/virt/periodic.py b/vdsm/virt/periodic.py
index b023976..ff0a379 100644
--- a/vdsm/virt/periodic.py
+++ b/vdsm/virt/periodic.py
@@ -27,9 +27,7 @@
from vdsm import executor
from vdsm import libvirtconnection
-from vdsm import schedule
from vdsm.config import config
-from vdsm.utils import monotonic_time
from . import sampling
@@ -41,13 +39,6 @@
_TASKS = _WORKERS * _TASK_PER_WORKER
-_scheduler = schedule.Scheduler(name="periodic.Scheduler",
- clock=monotonic_time)
-
-_executor = executor.Executor(name="periodic.Executor",
- workers_count=_WORKERS,
- max_tasks=_TASKS,
- scheduler=_scheduler)
_operations = []
@@ -58,63 +49,69 @@
return interval / 2.
-def start(cif):
- global _operations
+class PeriodicOperationRunner(object):
- _scheduler.start()
- _executor.start()
+ def start(self, cif):
+ global _operations
+ self._scheduler = cif.scheduler
- def per_vm_operation(func, period):
- disp = VmDispatcher(
- cif.getVMs, _executor, func, _timeout_from(period))
- return Operation(disp, period)
+ self._executor = executor.Executor(name="periodic.Executor",
+ workers_count=_WORKERS,
+ max_tasks=_TASKS,
+ scheduler=self._scheduler)
+ self._executor.start()
- _operations = [
- # needs dispatching becuse updating the volume stats needs the
- # access the storage, thus can block.
- per_vm_operation(
- UpdateVolumes,
- config.getint('irs', 'vol_size_sample_interval')),
+ def per_vm_operation(func, period):
+ disp = VmDispatcher(
+ cif.getVMs, self._executor, func, _timeout_from(period))
+ return Operation(disp, period, self._scheduler, self._executor)
- # needs dispatching becuse access FS and libvirt data
- per_vm_operation(
- NumaInfoMonitor,
- config.getint('vars', 'vm_sample_numa_interval')),
+ _operations = [
+ # needs dispatching becuse updating the volume stats needs the
+ # access the storage, thus can block.
+ per_vm_operation(
+ UpdateVolumes,
+ config.getint('irs', 'vol_size_sample_interval')),
- # Job monitoring need QEMU monitor access.
- per_vm_operation(
- BlockjobMonitor,
- config.getint('vars', 'vm_sample_jobs_interval')),
+ # needs dispatching becuse access FS and libvirt data
+ per_vm_operation(
+ NumaInfoMonitor,
+ config.getint('vars', 'vm_sample_numa_interval')),
- # libvirt sampling using bulk stats can block, but unresponsive
- # domains are handled inside VMBulkSampler for performance reasons;
- # thus, does not need dispatching.
- Operation(
- sampling.VMBulkSampler(
- libvirtconnection.get(cif),
- cif.getVMs,
- sampling.stats_cache),
- config.getint('vars', 'vm_sample_interval')),
+ # Job monitoring need QEMU monitor access.
+ per_vm_operation(
+ BlockjobMonitor,
+ config.getint('vars', 'vm_sample_jobs_interval')),
- # we do this only until we get high water mark notifications
- # from qemu. Access storage and/or qemu monitor, so can block,
- # thus we need dispatching.
- per_vm_operation(
- DriveWatermarkMonitor,
- config.getint('vars', 'vm_watermark_interval'))
+ # libvirt sampling using bulk stats can block, but unresponsive
+ # domains are handled inside VMBulkSampler for performance reasons;
+ # thus, does not need dispatching.
+ Operation(
+ sampling.VMBulkSampler(
+ libvirtconnection.get(cif),
+ cif.getVMs,
+ sampling.stats_cache),
+ config.getint('vars', 'vm_sample_interval'),
+ self._scheduler,
+ self._executor),
- ]
+ # we do this only until we get high water mark notifications
+ # from qemu. Access storage and/or qemu monitor, so can block,
+ # thus we need dispatching.
+ per_vm_operation(
+ DriveWatermarkMonitor,
+ config.getint('vars', 'vm_watermark_interval'))
- for op in _operations:
- op.start()
+ ]
+ for op in _operations:
+ op.start()
-def stop():
- for op in _operations:
- op.stop()
+ def stop(self):
+ for op in _operations:
+ op.stop()
- _executor.stop(wait=False)
- _scheduler.stop(wait=False)
+ self._executor.stop(wait=False)
class Operation(object):
@@ -128,8 +125,7 @@
_log = logging.getLogger("virt.periodic.Operation")
- def __init__(self, func, period, timeout=0,
- scheduler=None, executor=None):
+ def __init__(self, func, period, scheduler, executor, timeout=0):
"""
parameters:
@@ -144,8 +140,8 @@
self._func = func
self._period = period
self._timeout = _timeout_from(period) if timeout == 0 else timeout
- self._scheduler = _scheduler if scheduler is None else scheduler
- self._executor = _executor if executor is None else executor
+ self._scheduler = scheduler
+ self._executor = executor
self._lock = threading.Lock()
self._running = False
self._call = None
--
To view, visit https://gerrit.ovirt.org/43825
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: If83eded458f8304d802fcd75839e7a916146918b
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski at gmail.com>
More information about the vdsm-patches
mailing list