storage: allow import_data and import_data_end be coroutines

On some storage pools this operation can also be time consuming - for
example require creating temporary volume, and volume.create() already
can be a coroutine.
This is also requirement for making common code used by start()/create()
etc be a coroutine, otherwise neither of them can be and will block
other operations.

Related to QubesOS/qubes-issues#4283

qubes/api/admin.py

@@ -479,7 +479,7 @@ def vm_volume_import(self):
         if not self.dest.is_halted():
             raise qubes.exc.QubesVMNotHaltedError(self.dest)
 
-        path = self.dest.storage.import_data(self.arg)
+        path = yield from self.dest.storage.import_data(self.arg)
         assert ' ' not in path
         size = self.dest.volumes[self.arg].size
 

qubes/api/internal.py

@@ -68,7 +68,8 @@ def vm_volume_import_end(self, untrusted_payload):
         success = untrusted_payload == b'ok'
 
         try:
-            self.dest.storage.import_data_end(self.arg, success=success)
+            yield from self.dest.storage.import_data_end(self.arg,
+                success=success)
         except:
             self.dest.fire_event('domain-volume-import-end', volume=self.arg,
                 success=False)

qubes/storage/__init__.py

@@ -198,6 +198,8 @@ def import_data(self):
             volume data require something more than just writing to a file (
             for example connecting to some other domain, or converting data
             on the fly), the returned path may be a pipe.
+
+            This can be implemented as a coroutine.
         '''
         raise self._not_implemented("import")
 
@@ -207,6 +209,8 @@ def import_data_end(self, success):
 
         This method is called regardless the operation was successful or not.
 
+        This can be implemented as a coroutine.
+
         :param success: True if data import was successful, otherwise False
         '''
         # by default do nothing
@@ -654,24 +658,34 @@ def export(self, volume):
 
         return self.vm.volumes[volume].export()
 
+    @asyncio.coroutine
     def import_data(self, volume):
         ''' Helper function to import volume data (pool.import_data(volume))'''
         assert isinstance(volume, (Volume, str)), \
             "You need to pass a Volume or pool name as str"
         if isinstance(volume, Volume):
-            return volume.import_data()
+            ret = volume.import_data()
+        else:
+            ret = self.vm.volumes[volume].import_data()
 
-        return self.vm.volumes[volume].import_data()
+        if asyncio.iscoroutine(ret):
+            ret = yield from ret
+        return ret
 
+    @asyncio.coroutine
     def import_data_end(self, volume, success):
         ''' Helper function to finish/cleanup data import
         (pool.import_data_end( volume))'''
         assert isinstance(volume, (Volume, str)), \
             "You need to pass a Volume or pool name as str"
         if isinstance(volume, Volume):
-            return volume.import_data_end(success=success)
+            ret = volume.import_data_end(success=success)
+        else:
+            ret = self.vm.volumes[volume].import_data_end(success=success)
 
-        return self.vm.volumes[volume].import_data_end(success=success)
+        if asyncio.iscoroutine(ret):
+            ret = yield from ret
+        return ret
 
 
 class VolumesCollection: