thinp: Extend drive immediately on events

Expose periodic.dispatch() function allowing immediate dispatching of
calls on the periodic executor. This is useful when you want to handle
libvirt events on the periodic executor.

The first user of this facility is the thinp volume monitor. Now when we
receive a block threshold or enospc events we use the periodic dispatch
to extend the relevant drive immediately. This eliminates the 0-2
seconds wait after receiving an event.

Here are test results from 4 runs, each writing 50 GiB to think disk at
~1300 MiB/s. Each run extends the disk 20 times. The VM was not paused
during the test.

| time        |  min  |  avg  |  max  |
|-------------|-------|-------|-------|
| total       |  0.77 |  1.15 |  1.39 |
| extend      |  0.55 |  0.92 |  1.14 |
| refresh     |  0.16 |  0.22 |  0.31 |
| wait        |  0.01 |  0.01 |  0.03 |

Fixes: #85
Signed-off-by: Nir Soffer <nsoffer@redhat.com>

lib/vdsm/virt/periodic.py

@@ -117,6 +117,20 @@ def stop():
     _executor.stop(wait=False)
 
 
+def dispatch(callable, timeout=None, discard=True):
+    """
+    Dispatch callable on the periodic executor.
+
+    Should be used when a periodic operation should run as soon as possible
+    instead of waiting for the next cycle.
+
+    Raises:
+    - vdsm.exception.ResourceExhausted if the executor queue is full
+    - vdsm.executor.NotRunning if the executor is not running
+    """
+    _executor.dispatch(callable, timeout=timeout, discard=discard)
+
+
 class Operation(object):
     """
     Operation runs a callable with a given period until

lib/vdsm/virt/vm.py

@@ -93,6 +93,7 @@
 from vdsm.virt import libvirtxml
 from vdsm.virt import metadata
 from vdsm.virt import migration
+from vdsm.virt import periodic
 from vdsm.virt import sampling
 from vdsm.virt import saslpasswd2
 from vdsm.virt import thinp
@@ -437,7 +438,7 @@ def __init__(self, cif, params, recover=False):
         self._hotunplugged_devices = {}  # { alias: device_object }
 
         self.volume_monitor = thinp.VolumeMonitor(
-            self, self.log, enabled=False)
+            self, self.log, dispatch=periodic.dispatch, enabled=False)
         self._connection = libvirtconnection.get(cif)
         if (recover and
             # status retrieved from the recovery file (legacy style)