storage: convert lvm driver to async version

LVM operations can take significant amount of time. This is especially
visible when stopping a VM (`vm.storage.stop()`) - in that time the
whole qubesd freeze for about 2 seconds.

Fix this by making all the ThinVolume methods a coroutines (where
supported). Each public coroutine is also wrapped with locking on
volume._lock to avoid concurrency-related problems.
This all also require changing internal helper functions to
coroutines. There are two functions that still needs to be called from
non-coroutine call sites:
 - init_cache/reset_cache (initial cache fill, ThinPool.setup())
 - qubes_lvm (ThinVolume.export()

So, those two functions need to live in two variants. Extract its common
code to separate functions to reduce code duplications.

Fixes QubesOS/qubes-issues#4283

qubes/storage/lvm.py

@@ -18,7 +18,7 @@
 #
 
 ''' Driver for storing vm images in a LVM thin pool '''
-
+import functools
 import logging
 import os
 import subprocess
@@ -195,26 +195,14 @@ class ThinPool(qubes.storage.Pool):
             return 0
 
 
-def init_cache(log=logging.getLogger('qubes.storage.lvm')):
-    cmd = ['lvs', '--noheadings', '-o',
-           'vg_name,pool_lv,name,lv_size,data_percent,lv_attr,origin',
-           '--units', 'b', '--separator', ';']
-    if os.getuid() != 0:
-        cmd.insert(0, 'sudo')
-    environ = os.environ.copy()
-    environ['LC_ALL'] = 'C.utf8'
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-        close_fds=True, env=environ)
-    out, err = p.communicate()
-    return_code = p.returncode
-    if return_code == 0 and err:
-        log.warning(err)
-    elif return_code != 0:
-        raise qubes.storage.StoragePoolException(err)
+_init_cache_cmd = ['lvs', '--noheadings', '-o',
+   'vg_name,pool_lv,name,lv_size,data_percent,lv_attr,origin',
+   '--units', 'b', '--separator', ';']
 
+def _parse_lvm_cache(lvm_output):
     result = {}
 
-    for line in out.splitlines():
+    for line in lvm_output.splitlines():
         line = line.decode().strip()
         pool_name, pool_lv, name, size, usage_percent, attr, \
             origin = line.split(';', 6)
@@ -228,6 +216,42 @@ def init_cache(log=logging.getLogger('qubes.storage.lvm')):
 
     return result
 
+def init_cache(log=logging.getLogger('qubes.storage.lvm')):
+    cmd = _init_cache_cmd
+    if os.getuid() != 0:
+        cmd.insert(0, 'sudo')
+    environ = os.environ.copy()
+    environ['LC_ALL'] = 'C.utf8'
+    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        close_fds=True, env=environ)
+    out, err = p.communicate()
+    return_code = p.returncode
+    if return_code == 0 and err:
+        log.warning(err)
+    elif return_code != 0:
+        raise qubes.storage.StoragePoolException(err)
+
+    return _parse_lvm_cache(out)
+
+@asyncio.coroutine
+def init_cache_coro(log=logging.getLogger('qubes.storage.lvm')):
+    cmd = _init_cache_cmd
+    if os.getuid() != 0:
+        cmd = ['sudo'] + cmd
+    environ = os.environ.copy()
+    environ['LC_ALL'] = 'C.utf8'
+    p = yield from asyncio.create_subprocess_exec(*cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        close_fds=True, env=environ)
+    out, err = yield from p.communicate()
+    return_code = p.returncode
+    if return_code == 0 and err:
+        log.warning(err)
+    elif return_code != 0:
+        raise qubes.storage.StoragePoolException(err)
+
+    return _parse_lvm_cache(out)
 
 size_cache = init_cache()
 
@@ -243,6 +267,21 @@ def _revision_sort_key(revision):
         revision = revision.split('-')[0]
     return int(revision)
 
+def locked(method):
+    '''Decorator running given Volume's coroutine under a lock.
+    Needs to be added after wrapping with @asyncio.coroutine, for example:
+
+    >>>@locked
+    >>>@asyncio.coroutine
+    >>>def start(self):
+    >>>    pass
+    '''
+    @asyncio.coroutine
+    @functools.wraps(method)
+    def wrapper(self, *args, **kwargs):
+        with (yield from self._lock):  # pylint: disable=protected-access
+            return (yield from method(self, *args, **kwargs))
+    return wrapper
 
 class ThinVolume(qubes.storage.Volume):
     ''' Default LVM thin volume implementation
@@ -260,6 +299,7 @@ class ThinVolume(qubes.storage.Volume):
             self._vid_import = self.vid + '-import'
 
         self._size = size
+        self._lock = asyncio.Lock()
 
     @property
     def path(self):
@@ -307,6 +347,7 @@ class ThinVolume(qubes.storage.Volume):
         raise qubes.storage.StoragePoolException(
             "You shouldn't use lvm size setter")
 
+    @asyncio.coroutine
     def _reset(self):
         ''' Resets a volatile volume '''
         assert not self.snap_on_start and not self.save_on_stop, \
@@ -314,14 +355,15 @@ class ThinVolume(qubes.storage.Volume):
         self.log.debug('Resetting volatile %s', self.vid)
         try:
             cmd = ['remove', self.vid]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
         except qubes.storage.StoragePoolException:
             pass
         # pylint: disable=protected-access
         cmd = ['create', self.pool._pool_id, self.vid.split('/')[1],
                str(self.size)]
-        qubes_lvm(cmd, self.log)
+        yield from qubes_lvm_coro(cmd, self.log)
 
+    @asyncio.coroutine
     def _remove_revisions(self, revisions=None):
         '''Remove old volume revisions.
 
@@ -342,10 +384,11 @@ class ThinVolume(qubes.storage.Volume):
             assert rev_id != self._vid_current
             try:
                 cmd = ['remove', self.vid + '-' + rev_id]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
             except qubes.storage.StoragePoolException:
                 pass
 
+    @asyncio.coroutine
     def _commit(self, vid_to_commit=None, keep=False):
         '''
         Commit temporary volume into current one. By default
@@ -368,8 +411,7 @@ class ThinVolume(qubes.storage.Volume):
             assert hasattr(self, '_vid_snap')
             vid_to_commit = self._vid_snap
 
-        # TODO: when converting this function to coroutine, this _must_ be
-        # under a lock
+        assert self._lock.locked()
         if not os.path.exists('/dev/' + vid_to_commit):
             # nothing to commit
             return
@@ -377,21 +419,23 @@ class ThinVolume(qubes.storage.Volume):
         if self._vid_current == self.vid:
             cmd = ['rename', self.vid,
                    '{}-{}-back'.format(self.vid, int(time.time()))]
-            qubes_lvm(cmd, self.log)
-            reset_cache()
+            yield from qubes_lvm_coro(cmd, self.log)
+            yield from reset_cache_coro()
 
         cmd = ['clone' if keep else 'rename',
                vid_to_commit,
                self.vid]
-        qubes_lvm(cmd, self.log)
-        reset_cache()
+        yield from qubes_lvm_coro(cmd, self.log)
+        yield from reset_cache_coro()
         # make sure the one we've committed right now is properly
         # detected as the current one - before removing anything
         assert self._vid_current == self.vid
 
         # and remove old snapshots, if needed
-        self._remove_revisions()
+        yield from self._remove_revisions()
 
+    @locked
+    @asyncio.coroutine
     def create(self):
         assert self.vid
         assert self.size
@@ -405,32 +449,34 @@ class ThinVolume(qubes.storage.Volume):
                     self.vid.split('/', 1)[1],
                     str(self.size)
                 ]
-            qubes_lvm(cmd, self.log)
-            reset_cache()
+            yield from qubes_lvm_coro(cmd, self.log)
+            yield from reset_cache_coro()
         return self
 
+    @locked
+    @asyncio.coroutine
     def remove(self):
         assert self.vid
         try:
             if os.path.exists('/dev/' + self._vid_snap):
                 cmd = ['remove', self._vid_snap]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
         except AttributeError:
             pass
 
         try:
             if os.path.exists('/dev/' + self._vid_import):
                 cmd = ['remove', self._vid_import]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
         except AttributeError:
             pass
 
-        self._remove_revisions(self.revisions.keys())
+        yield from self._remove_revisions(self.revisions.keys())
         if not os.path.exists(self.path):
             return
         cmd = ['remove', self.path]
-        qubes_lvm(cmd, self.log)
-        reset_cache()
+        yield from qubes_lvm_coro(cmd, self.log)
+        yield from reset_cache_coro()
         # pylint: disable=protected-access
         self.pool._volume_objects_cache.pop(self.vid, None)
 
@@ -441,6 +487,7 @@ class ThinVolume(qubes.storage.Volume):
         devpath = self.path
         return devpath
 
+    @locked
     @asyncio.coroutine
     def import_volume(self, src_volume):
         if not src_volume.save_on_stop:
@@ -456,13 +503,13 @@ class ThinVolume(qubes.storage.Volume):
         # pylint: disable=line-too-long
         if isinstance(src_volume.pool, ThinPool) and \
                 src_volume.pool.thin_pool == self.pool.thin_pool:  # NOQA
-            self._commit(src_volume.path[len('/dev/'):], keep=True)
+            yield from self._commit(src_volume.path[len('/dev/'):], keep=True)
         else:
             cmd = ['create',
                    self.pool._pool_id,  # pylint: disable=protected-access
                    self._vid_import.split('/')[1],
                    str(src_volume.size)]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
             src_path = src_volume.export()
             cmd = ['dd', 'if=' + src_path, 'of=/dev/' + self._vid_import,
                 'conv=sparse', 'status=none']
@@ -474,14 +521,16 @@ class ThinVolume(qubes.storage.Volume):
             yield from p.wait()
             if p.returncode != 0:
                 cmd = ['remove', self._vid_import]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
                 raise qubes.storage.StoragePoolException(
                     'Failed to import volume {!r}, dd exit code: {}'.format(
                         src_volume, p.returncode))
-            self._commit(self._vid_import)
+            yield from self._commit(self._vid_import)
 
         return self
 
+    @locked
+    @asyncio.coroutine
     def import_data(self):
         ''' Returns an object that can be `open()`. '''
         if self.is_dirty():
@@ -492,21 +541,23 @@ class ThinVolume(qubes.storage.Volume):
         # pylint: disable=protected-access
         cmd = ['create', self.pool._pool_id, self._vid_import.split('/')[1],
                str(self.size)]
-        qubes_lvm(cmd, self.log)
-        reset_cache()
+        yield from qubes_lvm_coro(cmd, self.log)
+        yield from reset_cache_coro()
         devpath = '/dev/' + self._vid_import
         return devpath
 
+    @locked
+    @asyncio.coroutine
     def import_data_end(self, success):
         '''Either commit imported data, or discard temporary volume'''
         if not os.path.exists('/dev/' + self._vid_import):
             raise qubes.storage.StoragePoolException(
                 'No import operation in progress on {}'.format(self.vid))
         if success:
-            self._commit(self._vid_import)
+            yield from self._commit(self._vid_import)
         else:
             cmd = ['remove', self._vid_import]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
 
     def abort_if_import_in_progress(self):
         try:
@@ -531,6 +582,8 @@ class ThinVolume(qubes.storage.Volume):
         return (size_cache[self._vid_snap]['origin'] !=
                self.source.path.split('/')[-1])
 
+    @locked
+    @asyncio.coroutine
     def revert(self, revision=None):
         if self.is_dirty():
             raise qubes.storage.StoragePoolException(
@@ -547,12 +600,14 @@ class ThinVolume(qubes.storage.Volume):
 
         if self.vid in size_cache:
             cmd = ['remove', self.vid]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
         cmd = ['clone', self.vid + '-' + revision, self.vid]
-        qubes_lvm(cmd, self.log)
-        reset_cache()
+        yield from qubes_lvm_coro(cmd, self.log)
+        yield from reset_cache_coro()
         return self
 
+    @locked
+    @asyncio.coroutine
     def resize(self, size):
         ''' Expands volume, throws
             :py:class:`qubst.storage.qubes.storage.StoragePoolException` if
@@ -574,20 +629,21 @@ class ThinVolume(qubes.storage.Volume):
 
         if self.is_dirty():
             cmd = ['extend', self._vid_snap, str(size)]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
         elif hasattr(self, '_vid_import') and \
                 os.path.exists('/dev/' + self._vid_import):
             cmd = ['extend', self._vid_import, str(size)]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
         elif self.save_on_stop or not self.snap_on_start:
             cmd = ['extend', self._vid_current, str(size)]
-            qubes_lvm(cmd, self.log)
-        reset_cache()
+            yield from qubes_lvm_coro(cmd, self.log)
+        yield from reset_cache_coro()
 
+    @asyncio.coroutine
     def _snapshot(self):
         try:
             cmd = ['remove', self._vid_snap]
-            qubes_lvm(cmd, self.log)
+            yield from qubes_lvm_coro(cmd, self.log)
         except:  # pylint: disable=bare-except
             pass
 
@@ -595,32 +651,36 @@ class ThinVolume(qubes.storage.Volume):
             cmd = ['clone', self._vid_current, self._vid_snap]
         else:
             cmd = ['clone', self.source.path, self._vid_snap]
-        qubes_lvm(cmd, self.log)
+        yield from qubes_lvm_coro(cmd, self.log)
 
+    @locked
+    @asyncio.coroutine
     def start(self):
         self.abort_if_import_in_progress()
         try:
             if self.snap_on_start or self.save_on_stop:
                 if not self.save_on_stop or not self.is_dirty():
-                    self._snapshot()
+                    yield from self._snapshot()
             else:
-                self._reset()
+                yield from self._reset()
         finally:
-            reset_cache()
+            yield from reset_cache_coro()
         return self
 
+    @locked
+    @asyncio.coroutine
     def stop(self):
         try:
             if self.save_on_stop:
-                self._commit()
+                yield from self._commit()
             if self.snap_on_start and not self.save_on_stop:
                 cmd = ['remove', self._vid_snap]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
             elif not self.snap_on_start and not self.save_on_stop:
                 cmd = ['remove', self.vid]
-                qubes_lvm(cmd, self.log)
+                yield from qubes_lvm_coro(cmd, self.log)
         finally:
-            reset_cache()
+            yield from reset_cache_coro()
         return self
 
     def verify(self):
@@ -671,9 +731,14 @@ def pool_exists(pool_id):
     except KeyError:
         return False
 
+def _get_lvm_cmdline(cmd):
+    ''' Build command line for :program:`lvm` call.
+    The purpose of this function is to keep all the detailed lvm options in
+    one place.
 
-def qubes_lvm(cmd, log=logging.getLogger('qubes.storage.lvm')):
-    ''' Call :program:`lvm` to execute an LVM operation '''
+    :param cmd: array of str, where cmd[0] is action and the rest are arguments
+    :return array of str appropriate for subprocess.Popen
+    '''
     action = cmd[0]
     if action == 'remove':
         lvm_cmd = ['lvremove', '-f', cmd[1]]
@@ -698,28 +763,57 @@ def qubes_lvm(cmd, log=logging.getLogger('qubes.storage.lvm')):
         cmd = ['sudo', 'lvm'] + lvm_cmd
     else:
         cmd = ['lvm'] + lvm_cmd
-    environ = os.environ.copy()
-    environ['LC_ALL'] = 'C.utf8'
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-        close_fds=True, env=environ)
-    out, err = p.communicate()
-    err = err.decode()
+
+    return cmd
+
+def _process_lvm_output(returncode, stdout, stderr, log):
+    '''Process output of LVM, determine if the call was successful and
+    possibly log warnings.'''
     # Filter out warning about intended over-provisioning.
     # Upstream discussion about missing option to silence it:
     # https://bugzilla.redhat.com/1347008
-    err = '\n'.join(line for line in err.splitlines()
+    err = '\n'.join(line for line in stderr.decode().splitlines()
         if 'exceeds the size of thin pool' not in line)
-    return_code = p.returncode
-    if out:
-        log.debug(out)
-    if return_code == 0 and err:
+    if stdout:
+        log.debug(stdout)
+    if returncode == 0 and err:
         log.warning(err)
-    elif return_code != 0:
+    elif returncode != 0:
         assert err, "Command exited unsuccessful, but printed nothing to stderr"
         err = err.replace('%', '%%')
         raise qubes.storage.StoragePoolException(err)
     return True
 
+def qubes_lvm(cmd, log=logging.getLogger('qubes.storage.lvm')):
+    ''' Call :program:`lvm` to execute an LVM operation '''
+    # the only caller for this non-coroutine version is ThinVolume.export()
+    cmd = _get_lvm_cmdline(cmd)
+    environ = os.environ.copy()
+    environ['LC_ALL'] = 'C.utf8'
+    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        close_fds=True, env=environ)
+    out, err = p.communicate()
+    return _process_lvm_output(p.returncode, out, err, log)
+
+@asyncio.coroutine
+def qubes_lvm_coro(cmd, log=logging.getLogger('qubes.storage.lvm')):
+    ''' Call :program:`lvm` to execute an LVM operation
+
+    Coroutine version of :py:func:`qubes_lvm`'''
+    cmd = _get_lvm_cmdline(cmd)
+    environ = os.environ.copy()
+    environ['LC_ALL'] = 'C.utf8'
+    p = yield from asyncio.create_subprocess_exec(*cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        close_fds=True, env=environ)
+    out, err = yield from p.communicate()
+    return _process_lvm_output(p.returncode, out, err, log)
+
 
 def reset_cache():
     qubes.storage.lvm.size_cache = init_cache()
+
+@asyncio.coroutine
+def reset_cache_coro():
+    qubes.storage.lvm.size_cache = yield from init_cache_coro()

qubes/tests/storage_lvm.py

@@ -136,10 +136,10 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(volume.name, 'root')
         self.assertEqual(volume.pool, self.pool.name)
         self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         path = "/dev/%s" % volume.vid
         self.assertTrue(os.path.exists(path), path)
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_003_read_write_volume(self):
         ''' Test read-write volume creation '''
@@ -156,10 +156,10 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(volume.name, 'root')
         self.assertEqual(volume.pool, self.pool.name)
         self.assertEqual(volume.size, qubes.config.defaults['root_img_size'])
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         path = "/dev/%s" % volume.vid
         self.assertTrue(os.path.exists(path), path)
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_004_size(self):
         with self.assertNotRaises(NotImplementedError):
@@ -207,11 +207,11 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
-        self.addCleanup(volume.remove)
+        self.loop.run_until_complete(volume.create())
+        self.addCleanup(self.loop.run_until_complete, volume.remove())
         path = "/dev/%s" % volume.vid
         new_size = 64 * 1024 ** 2
-        volume.resize(new_size)
+        self.loop.run_until_complete(volume.resize(new_size))
         self.assertEqual(self._get_size(path), new_size)
         self.assertEqual(volume.size, new_size)
 
@@ -226,17 +226,17 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
-        self.addCleanup(volume.remove)
-        volume.start()
+        self.loop.run_until_complete(volume.create())
+        self.addCleanup(self.loop.run_until_complete, volume.remove())
+        self.loop.run_until_complete(volume.start())
         path = "/dev/%s" % volume.vid
         path2 = "/dev/%s" % volume._vid_snap
         new_size = 64 * 1024 ** 2
-        volume.resize(new_size)
+        self.loop.run_until_complete(volume.resize(new_size))
         self.assertEqual(self._get_size(path), old_size)
         self.assertEqual(self._get_size(path2), new_size)
         self.assertEqual(volume.size, new_size)
-        volume.stop()
+        self.loop.run_until_complete(volume.stop())
         self.assertEqual(self._get_size(path), new_size)
         self.assertEqual(volume.size, new_size)
 
@@ -271,18 +271,18 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         path_snap = '/dev/' + volume._vid_snap
         self.assertFalse(os.path.exists(path_snap), path_snap)
         origin_uuid = self._get_lv_uuid(volume.path)
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         snap_uuid = self._get_lv_uuid(path_snap)
         self.assertNotEqual(origin_uuid, snap_uuid)
         path = volume.path
         self.assertTrue(path.startswith('/dev/' + volume.vid),
                         '{} does not start with /dev/{}'.format(path, volume.vid))
         self.assertTrue(os.path.exists(path), path)
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_009_interrupted_commit(self):
         ''' Test volume changes commit'''
@@ -317,7 +317,7 @@ class TC_00_ThinPool(ThinPoolBase):
             revisions[1].lstrip('-'): '2018-03-14T22:18:25',
         }
         self.assertEqual(volume.revisions, expected_revisions)
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         self.assertEqual(volume.revisions, expected_revisions)
         snap_uuid = self._get_lv_uuid(path_snap)
         self.assertEqual(orig_uuids['-snap'], snap_uuid)
@@ -326,7 +326,7 @@ class TC_00_ThinPool(ThinPoolBase):
                          '/dev/' + volume.vid + revisions[1])
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = [521065906]
-            volume.stop()
+            self.loop.run_until_complete(volume.stop())
         expected_revisions = {
             revisions[0].lstrip('-'): '2018-03-14T22:18:24',
             revisions[1].lstrip('-'): '2018-03-14T22:18:25',
@@ -337,7 +337,7 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(snap_uuid, self._get_lv_uuid(volume.path))
         self.assertFalse(os.path.exists(path_snap), path_snap)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_010_migration1(self):
         '''Start with old revisions, then start interacting using new code'''
@@ -371,7 +371,7 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(volume.revisions, expected_revisions)
         self.assertEqual(volume.path, '/dev/' + volume.vid)
 
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         snap_uuid = self._get_lv_uuid(path_snap)
         self.assertNotEqual(orig_uuids[''], snap_uuid)
         snap_origin_uuid = self._get_lv_origin_uuid(path_snap)
@@ -382,7 +382,7 @@ class TC_00_ThinPool(ThinPoolBase):
 
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = ('1521065906', '1521065907')
-            volume.stop()
+            self.loop.run_until_complete(volume.stop())
         revisions.extend(['-1521065906-back'])
         expected_revisions = {
             revisions[2].lstrip('-'): '2018-03-14T22:18:25',
@@ -397,7 +397,7 @@ class TC_00_ThinPool(ThinPoolBase):
         prev_path = '/dev/' + volume.vid + revisions[3]
         self.assertEqual(self._get_lv_uuid(prev_path), orig_uuids[''])
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
         for rev in revisions:
             path = '/dev/' + volume.vid + rev
             self.assertFalse(os.path.exists(path), path)
@@ -438,7 +438,7 @@ class TC_00_ThinPool(ThinPoolBase):
 
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = ('1521065906', '1521065907')
-            volume.stop()
+            self.loop.run_until_complete(volume.stop())
         revisions.extend(['-1521065906-back'])
         expected_revisions = {
             revisions[2].lstrip('-'): '2018-03-14T22:18:26',
@@ -452,7 +452,7 @@ class TC_00_ThinPool(ThinPoolBase):
         prev_path = '/dev/' + volume.vid + revisions[2]
         self.assertEqual(self._get_lv_uuid(prev_path), orig_uuids[''])
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
         for rev in revisions:
             path = '/dev/' + volume.vid + rev
             self.assertFalse(os.path.exists(path), path)
@@ -487,14 +487,14 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertTrue(volume.path, '/dev/' + volume.vid)
         self.assertTrue(volume.is_dirty())
 
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         self.assertEqual(volume.revisions, expected_revisions)
         self.assertEqual(volume.path, '/dev/' + volume.vid)
         # -snap LV should be unchanged
         self.assertEqual(self._get_lv_uuid(volume._vid_snap),
                          orig_uuids['-snap'])
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
         for rev in revisions:
             path = '/dev/' + volume.vid + rev
             self.assertFalse(os.path.exists(path), path)
@@ -531,12 +531,12 @@ class TC_00_ThinPool(ThinPoolBase):
 
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = ('1521065906', '1521065907')
-            volume.stop()
+            self.loop.run_until_complete(volume.stop())
         expected_revisions = {}
         self.assertEqual(volume.revisions, expected_revisions)
         self.assertEqual(volume.path, '/dev/' + volume.vid)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
         for rev in revisions:
             path = '/dev/' + volume.vid + rev
             self.assertFalse(os.path.exists(path), path)
@@ -555,13 +555,13 @@ class TC_00_ThinPool(ThinPoolBase):
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
         # mock logging, to not interfere with time.time() mock
         volume.log = unittest.mock.Mock()
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         self.assertFalse(volume.is_dirty())
         path = volume.path
         expected_revisions = {}
         self.assertEqual(volume.revisions, expected_revisions)
 
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         self.assertEqual(volume.revisions, expected_revisions)
         path_snap = '/dev/' + volume._vid_snap
         snap_uuid = self._get_lv_uuid(path_snap)
@@ -570,14 +570,14 @@ class TC_00_ThinPool(ThinPoolBase):
 
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = [521065906]
-            volume.stop()
+            self.loop.run_until_complete(volume.stop())
         self.assertFalse(volume.is_dirty())
         self.assertEqual(volume.revisions, {})
         self.assertEqual(volume.path, '/dev/' + volume.vid)
         self.assertEqual(snap_uuid, self._get_lv_uuid(volume.path))
         self.assertFalse(os.path.exists(path_snap), path_snap)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_020_revert_last(self):
         ''' Test volume revert'''
@@ -591,11 +591,11 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
-        volume.start()
-        volume.stop()
-        volume.start()
-        volume.stop()
+        self.loop.run_until_complete(volume.create())
+        self.loop.run_until_complete(volume.start())
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.start())
+        self.loop.run_until_complete(volume.stop())
         self.assertEqual(len(volume.revisions), 2)
         revisions = volume.revisions
         revision_id = max(revisions.keys())
@@ -604,7 +604,7 @@ class TC_00_ThinPool(ThinPoolBase):
         rev_uuid = self._get_lv_uuid(volume.vid + '-' + revision_id)
         self.assertFalse(volume.is_dirty())
         self.assertNotEqual(current_uuid, rev_uuid)
-        volume.revert()
+        self.loop.run_until_complete(volume.revert())
         path_snap = '/dev/' + volume._vid_snap
         self.assertFalse(os.path.exists(path_snap), path_snap)
         self.assertEqual(current_path, volume.path)
@@ -612,7 +612,7 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(new_uuid, rev_uuid)
         self.assertEqual(volume.revisions, revisions)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_021_revert_earlier(self):
         ''' Test volume revert'''
@@ -626,11 +626,11 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
-        volume.start()
-        volume.stop()
-        volume.start()
-        volume.stop()
+        self.loop.run_until_complete(volume.create())
+        self.loop.run_until_complete(volume.start())
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.start())
+        self.loop.run_until_complete(volume.stop())
         self.assertEqual(len(volume.revisions), 2)
         revisions = volume.revisions
         revision_id = min(revisions.keys())
@@ -639,7 +639,7 @@ class TC_00_ThinPool(ThinPoolBase):
         rev_uuid = self._get_lv_uuid(volume.vid + '-' + revision_id)
         self.assertFalse(volume.is_dirty())
         self.assertNotEqual(current_uuid, rev_uuid)
-        volume.revert(revision_id)
+        self.loop.run_until_complete(volume.revert(revision_id))
         path_snap = '/dev/' + volume._vid_snap
         self.assertFalse(os.path.exists(path_snap), path_snap)
         self.assertEqual(current_path, volume.path)
@@ -647,7 +647,7 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(new_uuid, rev_uuid)
         self.assertEqual(volume.revisions, revisions)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_030_import_data(self):
         ''' Test volume import'''
@@ -661,14 +661,14 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         current_uuid = self._get_lv_uuid(volume.path)
         self.assertFalse(volume.is_dirty())
-        import_path = volume.import_data()
+        import_path = self.loop.run_until_complete(volume.import_data())
         import_uuid = self._get_lv_uuid(import_path)
         self.assertNotEqual(current_uuid, import_uuid)
         # success - commit data
-        volume.import_data_end(True)
+        self.loop.run_until_complete(volume.import_data_end(True))
         new_current_uuid = self._get_lv_uuid(volume.path)
         self.assertEqual(new_current_uuid, import_uuid)
         revisions = volume.revisions
@@ -678,7 +678,7 @@ class TC_00_ThinPool(ThinPoolBase):
                          self._get_lv_uuid(volume.vid + '-' + revision))
         self.assertFalse(os.path.exists(import_path), import_path)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_031_import_data_fail(self):
         ''' Test volume import'''
@@ -692,21 +692,21 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         current_uuid = self._get_lv_uuid(volume.path)
         self.assertFalse(volume.is_dirty())
-        import_path = volume.import_data()
+        import_path = self.loop.run_until_complete(volume.import_data())
         import_uuid = self._get_lv_uuid(import_path)
         self.assertNotEqual(current_uuid, import_uuid)
         # fail - discard data
-        volume.import_data_end(False)
+        self.loop.run_until_complete(volume.import_data_end(False))
         new_current_uuid = self._get_lv_uuid(volume.path)
         self.assertEqual(new_current_uuid, current_uuid)
         revisions = volume.revisions
         self.assertEqual(len(revisions), 0)
         self.assertFalse(os.path.exists(import_path), import_path)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_032_import_volume_same_pool(self):
         '''Import volume from the same pool'''
@@ -721,7 +721,7 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         vm = qubes.tests.storage.TestVM(self)
         source_volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        source_volume.create()
+        self.loop.run_until_complete(source_volume.create())
 
         source_uuid = self._get_lv_uuid(source_volume.path)
 
@@ -738,7 +738,7 @@ class TC_00_ThinPool(ThinPoolBase):
         volume.log = unittest.mock.Mock()
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = [1521065905]
-            volume.create()
+            self.loop.run_until_complete(volume.create())
 
         self.assertEqual(volume.revisions, {})
         uuid_before = self._get_lv_uuid(volume.path)
@@ -760,8 +760,8 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         self.assertEqual(volume.revisions, expected_revisions)
 
-        volume.remove()
-        source_volume.remove()
+        self.loop.run_until_complete(volume.remove())
+        self.loop.run_until_complete(source_volume.remove())
 
     def test_033_import_volume_different_pool(self):
         '''Import volume from a different pool'''
@@ -780,7 +780,7 @@ class TC_00_ThinPool(ThinPoolBase):
         volume.log = unittest.mock.Mock()
         with unittest.mock.patch('time.time') as mock_time:
             mock_time.side_effect = [1521065905]
-            volume.create()
+            self.loop.run_until_complete(volume.create())
 
         self.assertEqual(volume.revisions, {})
         uuid_before = self._get_lv_uuid(volume.path)
@@ -807,7 +807,7 @@ class TC_00_ThinPool(ThinPoolBase):
         }
         self.assertEqual(volume.revisions, expected_revisions)
 
-        volume.remove()
+        self.loop.run_until_complete(volume.remove())
 
     def test_040_volatile(self):
         '''Volatile volume test'''
@@ -821,21 +821,21 @@ class TC_00_ThinPool(ThinPoolBase):
         volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
         # volatile volume don't need any file, verify should succeed
         self.assertTrue(volume.verify())
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         self.assertTrue(volume.verify())
         self.assertFalse(volume.save_on_stop)
         self.assertFalse(volume.snap_on_start)
         path = volume.path
         self.assertEqual(path, '/dev/' + volume.vid)
         self.assertFalse(os.path.exists(path))
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         self.assertTrue(os.path.exists(path))
         vol_uuid = self._get_lv_uuid(path)
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         self.assertTrue(os.path.exists(path))
         vol_uuid2 = self._get_lv_uuid(path)
         self.assertNotEqual(vol_uuid, vol_uuid2)
-        volume.stop()
+        self.loop.run_until_complete(volume.stop())
         self.assertFalse(os.path.exists(path))
 
     def test_050_snapshot_volume(self):
@@ -850,7 +850,7 @@ class TC_00_ThinPool(ThinPoolBase):
         vm = qubes.tests.storage.TestVM(self)
         volume_origin = self.app.get_pool(self.pool.name).init_volume(
             vm, config_origin)
-        volume_origin.create()
+        self.loop.run_until_complete(volume_origin.create())
         config_snapshot = {
             'name': 'root2',
             'pool': self.pool.name,
@@ -868,11 +868,11 @@ class TC_00_ThinPool(ThinPoolBase):
         # only origin volume really needs to exist, verify should succeed
         # even before create
         self.assertTrue(volume.verify())
-        volume.create()
+        self.loop.run_until_complete(volume.create())
         path = volume.path
         self.assertEqual(path, '/dev/' + volume.vid)
         self.assertFalse(os.path.exists(path), path)
-        volume.start()
+        self.loop.run_until_complete(volume.start())
         # snapshot volume isn't considered dirty at any time
         self.assertFalse(volume.is_dirty())
         # not outdated yet
@@ -882,13 +882,13 @@ class TC_00_ThinPool(ThinPoolBase):
         self.assertEqual(origin_uuid, snap_origin_uuid)
 
         # now make it outdated
-        volume_origin.start()
-        volume_origin.stop()
+        self.loop.run_until_complete(volume_origin.start())
+        self.loop.run_until_complete(volume_origin.stop())
         self.assertTrue(volume.is_outdated())
         origin_uuid = self._get_lv_uuid(volume_origin.path)
         self.assertNotEqual(origin_uuid, snap_origin_uuid)
 
-        volume.stop()
+        self.loop.run_until_complete(volume.stop())
         # stopped volume is never outdated
         self.assertFalse(volume.is_outdated())
         path = volume.path
@@ -896,8 +896,8 @@ class TC_00_ThinPool(ThinPoolBase):
         path = '/dev/' + volume._vid_snap
         self.assertFalse(os.path.exists(path), path)
 
-        volume.remove()
-        volume_origin.remove()
+        self.loop.run_until_complete(volume.remove())
+        self.loop.run_until_complete(volume_origin.remove())
 
     def test_100_pool_list_volumes(self):
         config = {
@@ -911,24 +911,24 @@ class TC_00_ThinPool(ThinPoolBase):
         config2 = config.copy()
         vm = qubes.tests.storage.TestVM(self)
         volume1 = self.app.get_pool(self.pool.name).init_volume(vm, config)
-        volume1.create()
+        self.loop.run_until_complete(volume1.create())
         config2['name'] = 'private'
         volume2 = self.app.get_pool(self.pool.name).init_volume(vm, config2)
-        volume2.create()
+        self.loop.run_until_complete(volume2.create())
 
         # create some revisions
-        volume1.start()
-        volume1.stop()
+        self.loop.run_until_complete(volume1.start())
+        self.loop.run_until_complete(volume1.stop())
 
         # and have one in dirty state
-        volume2.start()
+        self.loop.run_until_complete(volume2.start())
 
         self.assertIn(volume1, list(self.pool.volumes))
         self.assertIn(volume2, list(self.pool.volumes))
-        volume1.remove()
+        self.loop.run_until_complete(volume1.remove())
         self.assertNotIn(volume1, list(self.pool.volumes))
         self.assertIn(volume2, list(self.pool.volumes))
-        volume2.remove()
+        self.loop.run_until_complete(volume2.remove())
         self.assertNotIn(volume1, list(self.pool.volumes))
         self.assertNotIn(volume1, list(self.pool.volumes))