feat(linstor): parallelize SR.scan/cleanup (vhd-util calls)

- Simplify scan logic removing XAPI calls.
- Create one XAPI session / LINSTOR connection in each worker thread.

Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>

Author: Wescoeur

drivers/LinstorSR.py

@@ -20,7 +20,7 @@
 
 try:
     from linstorjournaler import LinstorJournaler
-    from linstorvhdutil import LinstorVhdUtil
+    from linstorvhdutil import LinstorVhdUtil, MultiLinstorVhdUtil
     from linstorvolumemanager import get_controller_uri
     from linstorvolumemanager import get_controller_node_name
     from linstorvolumemanager import LinstorVolumeManager
@@ -372,6 +372,7 @@ def load(self, sr_uuid) -> None:
         self._vdis_loaded = False
         self._all_volume_info_cache = None
         self._all_volume_metadata_cache = None
+        self._multi_vhdutil = None
 
     # To remove in python 3.10.
     # Use directly @staticmethod instead.
@@ -1092,8 +1093,10 @@ def _load_vdis_ex(self):
             else:
                 introduce = True
 
-        # 4. Now check all volume info.
+        # 4. Now process all volume info.
         vdi_to_snaps = {}
+        vdi_uuids = []
+
         for vdi_uuid, volume_info in all_volume_info.items():
             if vdi_uuid.startswith(cleanup.SR.TMP_RENAME_PREFIX):
                 continue
@@ -1202,14 +1205,30 @@ def _load_vdis_ex(self):
                         vdi_to_snaps[snap_uuid] = [vdi_uuid]
 
             # 4.b. Add the VDI in the list.
+            vdi_uuids.append(vdi_uuid)
+
+        # 5. Create VDIs.
+        self._multi_vhdutil = MultiLinstorVhdUtil(self._linstor.uri, self._group_name)
+
+        def load_vdi(vdi_uuid, vhdutil_instance):
             vdi = self.vdi(vdi_uuid)
-            self.vdis[vdi_uuid] = vdi
 
             if USE_KEY_HASH and vdi.vdi_type == vhdutil.VDI_TYPE_VHD:
-                vdi.sm_config_override['key_hash'] = self._vhdutil.get_key_hash(vdi_uuid)
+                vdi.sm_config_override['key_hash'] = vhdutil_instance.get_key_hash(vdi_uuid)
 
-            # 4.c. Update CBT status of disks either just added
-            # or already in XAPI.
+            return vdi
+
+        try:
+            self.vdis = {vdi.uuid: vdi for vdi in self._multi_vhdutil.run(load_vdi, vdi_uuids)}
+        finally:
+            multi_vhdutil = self._multi_vhdutil
+            self._multi_vhdutil = None
+            del multi_vhdutil
+
+        # 6. Update CBT status of disks either just added
+        # or already in XAPI.
+        for vdi in self.vdis.values():
+            volume_metadata = volumes_metadata.get(vdi.uuid)
             cbt_uuid = volume_metadata.get(CBTLOG_TAG)
             if cbt_uuid in cbt_vdis:
                 vdi_ref = xenapi.VDI.get_by_uuid(vdi_uuid)
@@ -1220,7 +1239,7 @@ def _load_vdis_ex(self):
                 self.vdis[vdi_uuid].cbt_enabled = True
                 cbt_vdis.remove(cbt_uuid)
 
-        # 5. Now set the snapshot statuses correctly in XAPI.
+        # 7. Now set the snapshot statuses correctly in XAPI.
         for src_uuid in vdi_to_snaps:
             try:
                 src_ref = xenapi.VDI.get_by_uuid(src_uuid)
@@ -1239,7 +1258,7 @@ def _load_vdis_ex(self):
         # TODO: Check correctly how to use CBT.
         # Update cbt_enabled on the right VDI, check LVM/FileSR code.
 
-        # 6. If we have items remaining in this list,
+        # 8. If we have items remaining in this list,
         # they are cbt_metadata VDI that XAPI doesn't know about.
         # Add them to self.vdis and they'll get added to the DB.
         for cbt_uuid in cbt_vdis:
@@ -1248,10 +1267,10 @@ def _load_vdis_ex(self):
             new_vdi.cbt_enabled = True
             self.vdis[cbt_uuid] = new_vdi
 
-        # 7. Update virtual allocation, build geneology and remove useless VDIs
+        # 9. Update virtual allocation, build geneology and remove useless VDIs
         self.virtual_allocation = 0
 
-        # 8. Build geneology.
+        # 10. Build geneology.
         geneology = {}
 
         for vdi_uuid, vdi in self.vdis.items():
@@ -1265,7 +1284,7 @@ def _load_vdis_ex(self):
             if not vdi.hidden:
                 self.virtual_allocation += vdi.size
 
-        # 9. Remove all hidden leaf nodes to avoid introducing records that
+        # 11. Remove all hidden leaf nodes to avoid introducing records that
         # will be GC'ed.
         for vdi_uuid in list(self.vdis.keys()):
             if vdi_uuid not in geneology and self.vdis[vdi_uuid].hidden:
@@ -2096,13 +2115,15 @@ def _load_this(self):
         volume_metadata = None
         if self.sr._all_volume_metadata_cache:
             volume_metadata = self.sr._all_volume_metadata_cache.get(self.uuid)
-        if volume_metadata is None:
+            assert volume_metadata
+        else:
             volume_metadata = self._linstor.get_volume_metadata(self.uuid)
 
         volume_info = None
         if self.sr._all_volume_info_cache:
             volume_info = self.sr._all_volume_info_cache.get(self.uuid)
-        if volume_info is None:
+            assert volume_info
+        else:
             volume_info = self._linstor.get_volume_info(self.uuid)
 
         # Contains the max physical size used on a disk.
@@ -2119,7 +2140,13 @@ def _load_this(self):
             self.size = volume_info.virtual_size
             self.parent = ''
         else:
-            vhd_info = self.sr._vhdutil.get_vhd_info(self.uuid)
+            if self.sr._multi_vhdutil:
+                vhdutil_instance = self.sr._multi_vhdutil.local_vhdutil
+            else:
+                vhdutil_instance = self.sr._vhdutil
+
+            vhd_info = vhdutil_instance.get_vhd_info(self.uuid)
+
             self.hidden = vhd_info.hidden
             self.size = vhd_info.sizeVirt
             self.parent = vhd_info.parentUuid
@@ -2223,32 +2250,26 @@ def _determine_type_and_path(self):
         Determine whether this is a RAW or a VHD VDI.
         """
 
-        # 1. Check vdi_ref and vdi_type in config.
-        try:
-            vdi_ref = self.session.xenapi.VDI.get_by_uuid(self.uuid)
-            if vdi_ref:
-                sm_config = self.session.xenapi.VDI.get_sm_config(vdi_ref)
-                vdi_type = sm_config.get('vdi_type')
-                if vdi_type:
-                    # Update parent fields.
-                    self.vdi_type = vdi_type
-                    self.sm_config_override = sm_config
-                    self._update_device_name(
-                        self._linstor.get_volume_name(self.uuid)
-                    )
-                    return
-        except Exception:
-            pass
+        if self.sr._all_volume_metadata_cache:
+            # We are currently loading all volumes.
+            volume_metadata = self.sr._all_volume_metadata_cache.get(self.uuid)
+            if not volume_metadata:
+                raise xs_errors.XenError(
+                    'VDIUnavailable',
+                    opterr='failed to get metadata'
+                )
+        else:
+            # Simple load.
+            volume_metadata = self._linstor.get_volume_metadata(self.uuid)
 
-        # 2. Otherwise use the LINSTOR volume manager directly.
-        # It's probably a new VDI created via snapshot.
-        volume_metadata = self._linstor.get_volume_metadata(self.uuid)
+        # Set type and path.
         self.vdi_type = volume_metadata.get(VDI_TYPE_TAG)
         if not self.vdi_type:
             raise xs_errors.XenError(
                 'VDIUnavailable',
                 opterr='failed to get vdi_type in metadata'
             )
+
         self._update_device_name(self._linstor.get_volume_name(self.uuid))
 
     def _update_device_name(self, device_name):

drivers/cleanup.py

@@ -45,6 +45,7 @@
 import lock
 import blktap2
 import xs_errors
+
 from refcounter import RefCounter
 from ipc import IPCFlag
 from lvmanager import LVActivator
@@ -54,7 +55,7 @@
 
 try:
     from linstorjournaler import LinstorJournaler
-    from linstorvhdutil import LinstorVhdUtil
+    from linstorvhdutil import LinstorVhdUtil, MultiLinstorVhdUtil
     from linstorvolumemanager import get_controller_uri
     from linstorvolumemanager import LinstorVolumeManager
     from linstorvolumemanager import LinstorVolumeManagerError
@@ -3501,12 +3502,18 @@ def _scan(self, force):
         raise util.SMException('Scan error')
 
     def _load_vdi_info(self):
+        all_volume_info = self._linstor.get_volumes_with_info()
+        volumes_metadata = self._linstor.get_volumes_with_metadata()
+
         all_vdi_info = {}
+        pending_vdi_uuids = []
 
-        # TODO: Ensure metadata contains the right info.
+        def handle_fail(vdi_uuid, e):
+            Util.log(f" [VDI {vdi_uuid}: failed to load VDI info]: {e}")
+            info = vhdutil.VHDInfo(vdi_uuid)
+            info.error = 1
+            return info
 
-        all_volume_info = self._linstor.get_volumes_with_info()
-        volumes_metadata = self._linstor.get_volumes_with_metadata()
         for vdi_uuid, volume_info in all_volume_info.items():
             try:
                 volume_metadata = volumes_metadata[vdi_uuid]
@@ -3534,34 +3541,26 @@ def _load_vdi_info(self):
                     continue
 
                 vdi_type = volume_metadata.get(VDI_TYPE_TAG)
-                volume_name = self._linstor.get_volume_name(vdi_uuid)
-                if volume_name.startswith(LINSTOR_PERSISTENT_PREFIX):
-                    # Always RAW!
-                    info = None
-                elif vdi_type == vhdutil.VDI_TYPE_VHD:
-                    info = self._vhdutil.get_vhd_info(vdi_uuid)
+                if vdi_type == vhdutil.VDI_TYPE_VHD:
+                    pending_vdi_uuids.append(vdi_uuid)
                 else:
-                    # Ensure it's not a VHD...
-                    try:
-                        info = self._vhdutil.get_vhd_info(vdi_uuid)
-                    except:
-                        try:
-                            self._vhdutil.force_repair(
-                                self._linstor.get_device_path(vdi_uuid)
-                            )
-                            info = self._vhdutil.get_vhd_info(vdi_uuid)
-                        except:
-                            info = None
+                    all_vdi_info[vdi_uuid] = None
+            except Exception as e:
+                all_vdi_info[vdi_uuid] = handle_fail(vdi_uuid, e)
 
+        multi_vhdutil = MultiLinstorVhdUtil(self._linstor.uri, self._linstor.group_name)
+
+        def load_info(vdi_uuid, vhdutil_instance):
+            try:
+                return vhdutil_instance.get_vhd_info(vdi_uuid)
             except Exception as e:
-                Util.log(
-                    ' [VDI {}: failed to load VDI info]: {}'
-                    .format(vdi_uuid, e)
-                )
-                info = vhdutil.VHDInfo(vdi_uuid)
-                info.error = 1
+                return handle_fail(vdi_uuid, e)
 
-            all_vdi_info[vdi_uuid] = info
+        try:
+            for vdiInfo in multi_vhdutil.run(load_info, pending_vdi_uuids):
+                all_vdi_info[vdiInfo.uuid] = vdiInfo
+        finally:
+            del multi_vhdutil
 
         return all_vdi_info
 

drivers/linstorvhdutil.py

@@ -18,10 +18,13 @@
 
 from linstorjournaler import LinstorJournaler
 from linstorvolumemanager import LinstorVolumeManager
+
+from concurrent.futures import ThreadPoolExecutor
 import base64
 import errno
 import json
 import socket
+import threading
 import time
 import util
 import vhdutil
@@ -597,3 +600,72 @@ def _zeroize(path, size):
                 'EIO',
                 opterr='Failed to zero out VHD footer {}'.format(path)
             )
+
+class MultiLinstorVhdUtil:
+    class ExecutorData(threading.local):
+        def __init__(self):
+            self._clear()
+
+        def _clear(self):
+            self.session = None
+            self.linstor = None
+            self.vhdutil = None
+
+    class Load:
+        def __init__(self, session):
+            self.session = session
+
+        def cleanup(self):
+            if self.session:
+                self.session.xenapi.session.logout()
+                self.session = None
+
+    def __init__(self, uri, group_name) -> None:
+        self._uri = uri
+        self._group_name = group_name
+        self._loads: self.Load = []
+        self._executor_data = self.ExecutorData()
+
+    def __del__(self):
+        self._cleanup()
+
+    def run(self, cb, vdi_uuids):
+        def wrapper(cb, vdi_uuid):
+            if not self._executor_data.session:
+                self._init_executor_thread()
+            return cb(vdi_uuid, self._executor_data.vhdutil)
+
+        with ThreadPoolExecutor() as executor:
+            return executor.map(lambda vdi_uuid: wrapper(cb, vdi_uuid), vdi_uuids)
+
+    @property
+    def local_vhdutil(self):
+        return self._executor_data.vhdutil
+
+    def _init_executor_thread(self):
+        session = util.get_localAPI_session()
+        load = self.Load(session)
+        try:
+            linstor = LinstorVolumeManager(
+                self._uri,
+                self._group_name,
+                repair=False,
+                logger=util.SMlog
+            )
+            self._executor_data.linstor = linstor
+            self._executor_data.vhdutil = LinstorVhdUtil(session, linstor)
+            self._executor_data.session = session
+        except:
+            self._executor_data.clear()
+            load.cleanup()
+            raise
+
+        self._loads.append(load)
+
+    def _cleanup(self):
+        for load in self._loads:
+            try:
+                load.cleanup()
+            except Exception as e:
+                util.SMlog(f"Failed to clean load executor: {e}")
+        self._loads.clear()

drivers/linstorvolumemanager.py

@@ -90,9 +90,7 @@ def get_all_volume_openers(resource_name, volume):
     volume = str(volume)
     openers = {}
 
-    # Make sure this call never stucks because this function can be called
-    # during HA init and in this case we can wait forever.
-    session = util.timeout_call(10, util.get_localAPI_session)
+    session = util.get_localAPI_session()
 
     hosts = session.xenapi.host.get_all_records()
     for host_ref, host_record in hosts.items():
@@ -257,7 +255,7 @@ class LinstorVolumeManager(object):
     """
 
     __slots__ = (
-        '_linstor', '_logger', '_redundancy',
+        '_linstor', '_uri', '_logger', '_redundancy',
         '_base_group_name', '_group_name', '_ha_group_name',
         '_volumes', '_storage_pools', '_storage_pools_time',
         '_kv_cache', '_resource_cache', '_volume_info_cache',
@@ -359,6 +357,7 @@ def __init__(
         :param int attempt_count: Number of attempts to join the controller.
         """
 
+        self._uri = uri
         self._linstor = self._create_linstor_instance(
             uri, attempt_count=attempt_count
         )
@@ -402,6 +401,10 @@ def __init__(
         self._resources_info_cache = None
         self._build_volumes(repair=repair)
 
+    @property
+    def uri(self) -> str:
+        return self._uri
+
     @property
     def group_name(self):
         """