Change in vdsm[master]: scheduler: use single instance

Monday, 20 July 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(a)gmail.com&gt;
---
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(a)gmail.com&gt;

    

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011