cleanup.py

#!/usr/bin/python3
#
# Copyright (C) Citrix Systems Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation; version 2.1 only.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
#
# Script to coalesce and garbage collect COW-based SR's in the background
#

from sm_typing import Any, Dict, Optional, List, override

import os
import os.path
import sys
import time
import signal
import subprocess
import getopt
import datetime
import traceback
import base64
import zlib
import errno
import stat

import XenAPI # pylint: disable=import-error
import util
import lvutil
import lvmcache
import journaler
import fjournaler
import lock
import blktap2
import xs_errors
from refcounter import RefCounter
from ipc import IPCFlag
from lvmanager import LVActivator
from srmetadata import LVMMetadataHandler, VDI_TYPE_TAG
from functools import reduce
from time import monotonic as _time

from constants import NS_PREFIX_LVM, VG_LOCATION, VG_PREFIX
from cowutil import CowImageInfo, CowUtil, getCowUtil
from lvmcowutil import LV_PREFIX, LvmCowUtil
from vditype import VdiType, VdiTypeExtension, VDI_COW_TYPES, VDI_TYPE_TO_EXTENSION

try:
    from linstorcowutil import LinstorCowUtil, MultiLinstorCowUtil
    from linstorjournaler import LinstorJournaler
    from linstorvolumemanager import get_controller_uri
    from linstorvolumemanager import LinstorVolumeManager, LinstorVolumeManagerError, LinstorVolumeOpeners
    from linstorvolumemanager import PERSISTENT_PREFIX as LINSTOR_PERSISTENT_PREFIX

    LINSTOR_AVAILABLE = True
except ImportError:
    LINSTOR_AVAILABLE = False

# Disable automatic leaf-coalescing. Online leaf-coalesce is currently not
# possible due to lvhd_stop_using_() not working correctly. However, we leave
# this option available through the explicit LEAFCLSC_FORCE flag in the VDI
# record for use by the offline tool (which makes the operation safe by pausing
# the VM first)
AUTO_ONLINE_LEAF_COALESCE_ENABLED = True

FLAG_TYPE_ABORT = "abort"     # flag to request aborting of GC/coalesce

# process "lock", used simply as an indicator that a process already exists
# that is doing GC/coalesce on this SR (such a process holds the lock, and we
# check for the fact by trying the lock).
lockGCRunning = None

# process "lock" to indicate that the GC process has been activated but may not
# yet be running, stops a second process from being started.
LOCK_TYPE_GC_ACTIVE = "gc_active"
lockGCActive = None

# Default coalesce error rate limit, in messages per minute. A zero value
# disables throttling, and a negative value disables error reporting.
DEFAULT_COALESCE_ERR_RATE = 1.0 / 60

COALESCE_LAST_ERR_TAG = 'last-coalesce-error'
COALESCE_ERR_RATE_TAG = 'coalesce-error-rate'
VAR_RUN = "/var/run/"
SPEED_LOG_ROOT = VAR_RUN + "{uuid}.speed_log"

N_RUNNING_AVERAGE = 10

NON_PERSISTENT_DIR = '/run/nonpersistent/sm'

# Signal Handler
SIGTERM = False


class AbortException(util.SMException):
    pass

class CancelException(util.SMException):
    pass

def receiveSignal(signalNumber, frame):
    global SIGTERM

    util.SMlog("GC: recieved SIGTERM")
    SIGTERM = True
    return


################################################################################
#
#  Util
#
class Util:
    RET_RC = 1
    RET_STDOUT = 2
    RET_STDERR = 4

    UUID_LEN = 36

    PREFIX = {"G": 1024 * 1024 * 1024, "M": 1024 * 1024, "K": 1024}

    @staticmethod
    def log(text) -> None:
        util.SMlog(text, ident="SMGC")

    @staticmethod
    def logException(tag):
        info = sys.exc_info()
        if info[0] == SystemExit:
            # this should not be happening when catching "Exception", but it is
            sys.exit(0)
        tb = reduce(lambda a, b: "%s%s" % (a, b), traceback.format_tb(info[2]))
        Util.log("*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*")
        Util.log("         ***********************")
        Util.log("         *  E X C E P T I O N  *")
        Util.log("         ***********************")
        Util.log("%s: EXCEPTION %s, %s" % (tag, info[0], info[1]))
        Util.log(tb)
        Util.log("*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*~*")

    @staticmethod
    def doexec(args, expectedRC, inputtext=None, ret=None, log=True):
        "Execute a subprocess, then return its return code, stdout, stderr"
        proc = subprocess.Popen(args,
                                stdin=subprocess.PIPE, \
                                stdout=subprocess.PIPE, \
                                stderr=subprocess.PIPE, \
                                shell=True, \
                                close_fds=True)
        (stdout, stderr) = proc.communicate(inputtext)
        stdout = str(stdout)
        stderr = str(stderr)
        rc = proc.returncode
        if log:
            Util.log("`%s`: %s" % (args, rc))
        if type(expectedRC) != type([]):
            expectedRC = [expectedRC]
        if not rc in expectedRC:
            reason = stderr.strip()
            if stdout.strip():
                reason = "%s (stdout: %s)" % (reason, stdout.strip())
            Util.log("Failed: %s" % reason)
            raise util.CommandException(rc, args, reason)

        if ret == Util.RET_RC:
            return rc
        if ret == Util.RET_STDERR:
            return stderr
        return stdout

    @staticmethod
    def runAbortable(func, ret, ns, abortTest, pollInterval, timeOut, prefSig=signal.SIGKILL):
        """execute func in a separate thread and kill it if abortTest signals
        so"""
        abortSignaled = abortTest()  # check now before we clear resultFlag
        resultFlag = IPCFlag(ns)
        resultFlag.clearAll()
        pid = os.fork()
        if pid:
            startTime = _time()
            try:
                while True:
                    if resultFlag.test("success"):
                        Util.log("  Child process completed successfully")
                        resultFlag.clear("success")
                        return
                    if resultFlag.test("failure"):
                        resultFlag.clear("failure")
                        raise util.SMException("Child process exited with error")
                    if abortTest() or abortSignaled or SIGTERM:
                        os.killpg(pid, prefSig)
                        raise AbortException("Aborting due to signal")
                    if timeOut and _time() - startTime > timeOut:
                        os.killpg(pid, prefSig)
                        resultFlag.clearAll()
                        raise util.SMException("Timed out")
                    time.sleep(pollInterval)
            finally:
                wait_pid = 0
                rc = -1
                count = 0
                while wait_pid == 0 and count < 10:
                    wait_pid, rc = os.waitpid(pid, os.WNOHANG)
                    if wait_pid == 0:
                        time.sleep(2)
                        count += 1

                if wait_pid == 0:
                    Util.log("runAbortable: wait for process completion timed out")
        else:
            os.setpgrp()
            try:
                if func() == ret:
                    resultFlag.set("success")
                else:
                    resultFlag.set("failure")
            except Exception as e:
                Util.log("Child process failed with : (%s)" % e)
                resultFlag.set("failure")
                Util.logException("This exception has occured")
            os._exit(0)

    @staticmethod
    def num2str(number):
        for prefix in ("G", "M", "K"):
            if number >= Util.PREFIX[prefix]:
                return "%.3f%s" % (float(number) / Util.PREFIX[prefix], prefix)
        return "%s" % number

    @staticmethod
    def numBits(val):
        count = 0
        while val:
            count += val & 1
            val = val >> 1
        return count

    @staticmethod
    def countBits(bitmap1, bitmap2):
        """return bit count in the bitmap produced by ORing the two bitmaps"""
        len1 = len(bitmap1)
        len2 = len(bitmap2)
        lenLong = len1
        lenShort = len2
        bitmapLong = bitmap1
        if len2 > len1:
            lenLong = len2
            lenShort = len1
            bitmapLong = bitmap2

        count = 0
        for i in range(lenShort):
            val = bitmap1[i] | bitmap2[i]
            count += Util.numBits(val)

        for i in range(i + 1, lenLong):
            val = bitmapLong[i]
            count += Util.numBits(val)
        return count

    @staticmethod
    def getThisScript():
        thisScript = util.get_real_path(__file__)
        if thisScript.endswith(".pyc"):
            thisScript = thisScript[:-1]
        return thisScript


################################################################################
#
#  XAPI
#
class XAPI:
    USER = "root"
    PLUGIN_ON_SLAVE = "on-slave"

    CONFIG_SM = 0
    CONFIG_OTHER = 1
    CONFIG_ON_BOOT = 2
    CONFIG_ALLOW_CACHING = 3

    CONFIG_NAME = {
            CONFIG_SM: "sm-config",
            CONFIG_OTHER: "other-config",
            CONFIG_ON_BOOT: "on-boot",
            CONFIG_ALLOW_CACHING: "allow_caching"
    }

    class LookupError(util.SMException):
        pass

    def __init__(self, session, srUuid):
        self.session = session
        self._api_session = None
        if self.session is None:
            self._api_session = util.ApiSession("SM-GC")
            self.session = self._api_session.session
        self._srRef = self.session.xenapi.SR.get_by_uuid(srUuid)
        self.srRecord = self.session.xenapi.SR.get_record(self._srRef)
        self.hostUuid = util.get_this_host()
        self._hostRef = self.session.xenapi.host.get_by_uuid(self.hostUuid)
        self.task = None
        self.task_progress = {"coalescable": 0, "done": 0}

    def __del__(self):
        if self._api_session:
            self._api_session.logout()

    @property
    def srRef(self):
        return self._srRef

    def isPluggedHere(self):
        pbds = self.getAttachedPBDs()
        for pbdRec in pbds:
            if pbdRec["host"] == self._hostRef:
                return True
        return False

    def poolOK(self):
        host_recs = self.session.xenapi.host.get_all_records()
        for host_ref, host_rec in host_recs.items():
            if not host_rec["enabled"]:
                Util.log("Host %s not enabled" % host_rec["uuid"])
                return False
        return True

    def isMaster(self):
        if self.srRecord["shared"]:
            pool = list(self.session.xenapi.pool.get_all_records().values())[0]
            return pool["master"] == self._hostRef
        else:
            pbds = self.getAttachedPBDs()
            if len(pbds) < 1:
                raise util.SMException("Local SR not attached")
            elif len(pbds) > 1:
                raise util.SMException("Local SR multiply attached")
            return pbds[0]["host"] == self._hostRef

    def getAttachedPBDs(self):
        """Return PBD records for all PBDs of this SR that are currently
        attached"""
        attachedPBDs = []
        pbds = self.session.xenapi.PBD.get_all_records()
        for pbdRec in pbds.values():
            if pbdRec["SR"] == self._srRef and pbdRec["currently_attached"]:
                attachedPBDs.append(pbdRec)
        return attachedPBDs

    def getOnlineHosts(self):
        return util.get_online_hosts(self.session)

    def ensureInactive(self, hostRef, args):
        text = self.session.xenapi.host.call_plugin( \
                hostRef, self.PLUGIN_ON_SLAVE, "multi", args)
        Util.log("call-plugin returned: '%s'" % text)

    def getRecordHost(self, hostRef):
        return self.session.xenapi.host.get_record(hostRef)

    def _getRefVDI(self, uuid):
        return self.session.xenapi.VDI.get_by_uuid(uuid)

    def getRefVDI(self, vdi):
        return self._getRefVDI(vdi.uuid)

    def getRecordVDI(self, uuid):
        try:
            ref = self._getRefVDI(uuid)
            return self.session.xenapi.VDI.get_record(ref)
        except XenAPI.Failure:
            return None

    def singleSnapshotVDI(self, vdi):
        return self.session.xenapi.VDI.snapshot(vdi.getRef(),
                {"type": "internal"})

    def forgetVDI(self, srUuid, vdiUuid):
        """Forget the VDI, but handle the case where the VDI has already been
        forgotten (i.e. ignore errors)"""
        try:
            vdiRef = self.session.xenapi.VDI.get_by_uuid(vdiUuid)
            self.session.xenapi.VDI.forget(vdiRef)
        except XenAPI.Failure:
            pass

    def getConfigVDI(self, vdi, key):
        kind = vdi.CONFIG_TYPE[key]
        if kind == self.CONFIG_SM:
            cfg = self.session.xenapi.VDI.get_sm_config(vdi.getRef())
        elif kind == self.CONFIG_OTHER:
            cfg = self.session.xenapi.VDI.get_other_config(vdi.getRef())
        elif kind == self.CONFIG_ON_BOOT:
            cfg = self.session.xenapi.VDI.get_on_boot(vdi.getRef())
        elif kind == self.CONFIG_ALLOW_CACHING:
            cfg = self.session.xenapi.VDI.get_allow_caching(vdi.getRef())
        else:
            assert(False)
        Util.log("Got %s for %s: %s" % (self.CONFIG_NAME[kind], vdi, repr(cfg)))
        return cfg

    def removeFromConfigVDI(self, vdi, key):
        kind = vdi.CONFIG_TYPE[key]
        if kind == self.CONFIG_SM:
            self.session.xenapi.VDI.remove_from_sm_config(vdi.getRef(), key)
        elif kind == self.CONFIG_OTHER:
            self.session.xenapi.VDI.remove_from_other_config(vdi.getRef(), key)
        else:
            assert(False)

    def addToConfigVDI(self, vdi, key, val):
        kind = vdi.CONFIG_TYPE[key]
        if kind == self.CONFIG_SM:
            self.session.xenapi.VDI.add_to_sm_config(vdi.getRef(), key, val)
        elif kind == self.CONFIG_OTHER:
            self.session.xenapi.VDI.add_to_other_config(vdi.getRef(), key, val)
        else:
            assert(False)

    def isSnapshot(self, vdi):
        return self.session.xenapi.VDI.get_is_a_snapshot(vdi.getRef())

    def markCacheSRsDirty(self):
        sr_refs = self.session.xenapi.SR.get_all_records_where( \
                'field "local_cache_enabled" = "true"')
        for sr_ref in sr_refs:
            Util.log("Marking SR %s dirty" % sr_ref)
            util.set_dirty(self.session, sr_ref)

    def srUpdate(self):
        Util.log("Starting asynch srUpdate for SR %s" % self.srRecord["uuid"])
        abortFlag = IPCFlag(self.srRecord["uuid"])
        task = self.session.xenapi.Async.SR.update(self._srRef)
        cancelTask = True
        try:
            for i in range(60):
                status = self.session.xenapi.task.get_status(task)
                if not status == "pending":
                    Util.log("SR.update_asynch status changed to [%s]" % status)
                    cancelTask = False
                    return
                if abortFlag.test(FLAG_TYPE_ABORT):
                    Util.log("Abort signalled during srUpdate, cancelling task...")
                    try:
                        self.session.xenapi.task.cancel(task)
                        cancelTask = False
                        Util.log("Task cancelled")
                    except:
                        pass
                    return
                time.sleep(1)
        finally:
            if cancelTask:
                self.session.xenapi.task.cancel(task)
            self.session.xenapi.task.destroy(task)
        Util.log("Asynch srUpdate still running, but timeout exceeded.")

    def update_task(self):
        self.session.xenapi.task.set_other_config(
            self.task,
            {
                "applies_to": self._srRef
            })
        total = self.task_progress['coalescable'] + self.task_progress['done']
        if (total > 0):
            self.session.xenapi.task.set_progress(
                self.task, float(self.task_progress['done']) / total)

    def create_task(self, label, description):
        self.task = self.session.xenapi.task.create(label, description)
        self.update_task()

    def update_task_progress(self, key, value):
        self.task_progress[key] = value
        if self.task:
            self.update_task()

    def set_task_status(self, status):
        if self.task:
            self.session.xenapi.task.set_status(self.task, status)


################################################################################
#
#  VDI
#
class VDI(object):
    """Object representing a VDI of a COW-based SR"""

    POLL_INTERVAL = 1
    POLL_TIMEOUT = 30
    DEVICE_MAJOR = 202

    # config keys & values
    DB_VDI_PARENT = "vhd-parent"
    DB_VDI_TYPE = "vdi_type"
    DB_VDI_BLOCKS = "vhd-blocks"
    DB_VDI_PAUSED = "paused"
    DB_VDI_RELINKING = "relinking"
    DB_VDI_ACTIVATING = "activating"
    DB_GC = "gc"
    DB_COALESCE = "coalesce"
    DB_LEAFCLSC = "leaf-coalesce"  # config key
    DB_GC_NO_SPACE = "gc_no_space"
    LEAFCLSC_DISABLED = "false"  # set by user; means do not leaf-coalesce
    LEAFCLSC_FORCE = "force"     # set by user; means skip snap-coalesce
    LEAFCLSC_OFFLINE = "offline"  # set here for informational purposes: means
    # no space to snap-coalesce or unable to keep
    # up with VDI. This is not used by the SM, it
    # might be used by external components.
    DB_ONBOOT = "on-boot"
    ONBOOT_RESET = "reset"
    DB_ALLOW_CACHING = "allow_caching"

    CONFIG_TYPE = {
            DB_VDI_PARENT: XAPI.CONFIG_SM,
            DB_VDI_TYPE: XAPI.CONFIG_SM,
            DB_VDI_BLOCKS: XAPI.CONFIG_SM,
            DB_VDI_PAUSED: XAPI.CONFIG_SM,
            DB_VDI_RELINKING: XAPI.CONFIG_SM,
            DB_VDI_ACTIVATING: XAPI.CONFIG_SM,
            DB_GC: XAPI.CONFIG_OTHER,
            DB_COALESCE: XAPI.CONFIG_OTHER,
            DB_LEAFCLSC: XAPI.CONFIG_OTHER,
            DB_ONBOOT: XAPI.CONFIG_ON_BOOT,
            DB_ALLOW_CACHING: XAPI.CONFIG_ALLOW_CACHING,
            DB_GC_NO_SPACE: XAPI.CONFIG_SM
    }

    LIVE_LEAF_COALESCE_MAX_SIZE = 20 * 1024 * 1024  # bytes
    LIVE_LEAF_COALESCE_TIMEOUT = 10  # seconds
    TIMEOUT_SAFETY_MARGIN = 0.5  # extra margin when calculating
    # feasibility of leaf coalesce

    JRN_RELINK = "relink"  # journal entry type for relinking children
    JRN_COALESCE = "coalesce"  # to communicate which VDI is being coalesced
    JRN_LEAF = "leaf"  # used in coalesce-leaf

    STR_TREE_INDENT = 4

    def __init__(self, sr, uuid, vdi_type):
        self.sr = sr
        self.scanError = True
        self.uuid = uuid
        self.vdi_type = vdi_type
        self.fileName = ""
        self.parentUuid = ""
        self.sizeVirt = -1
        self._sizePhys = -1
        self._sizeAllocated = -1
        self._hidden = False
        self.parent = None
        self.children = []
        self._vdiRef = None
        self.cowutil = getCowUtil(vdi_type)
        self._clearRef()

    @staticmethod
    def extractUuid(path):
        raise NotImplementedError("Implement in sub class")

    def load(self, info=None) -> None:
        """Load VDI info"""
        pass

    def getDriverName(self) -> str:
        return self.vdi_type

    def getRef(self):
        if self._vdiRef is None:
            self._vdiRef = self.sr.xapi.getRefVDI(self)
        return self._vdiRef

    def getConfig(self, key, default=None):
        config = self.sr.xapi.getConfigVDI(self, key)
        if key == self.DB_ONBOOT or key == self.DB_ALLOW_CACHING:
            val = config
        else:
            val = config.get(key)
        if val:
            return val
        return default

    def setConfig(self, key, val):
        self.sr.xapi.removeFromConfigVDI(self, key)
        self.sr.xapi.addToConfigVDI(self, key, val)
        Util.log("Set %s = %s for %s" % (key, val, self))

    def delConfig(self, key):
        self.sr.xapi.removeFromConfigVDI(self, key)
        Util.log("Removed %s from %s" % (key, self))

    def ensureUnpaused(self):
        if self.getConfig(self.DB_VDI_PAUSED) == "true":
            Util.log("Unpausing VDI %s" % self)
            self.unpause()

    def pause(self, failfast=False) -> None:
        if not blktap2.VDI.tap_pause(self.sr.xapi.session, self.sr.uuid,
                self.uuid, failfast):
            raise util.SMException("Failed to pause VDI %s" % self)

    def _report_tapdisk_unpause_error(self):
        try:
            xapi = self.sr.xapi.session.xenapi
            sr_ref = xapi.SR.get_by_uuid(self.sr.uuid)
            msg_name = "failed to unpause tapdisk"
            msg_body = "Failed to unpause tapdisk for VDI %s, " \
                    "VMs using this tapdisk have lost access " \
                    "to the corresponding disk(s)" % self.uuid
            xapi.message.create(msg_name, "4", "SR", self.sr.uuid, msg_body)
        except Exception as e:
            util.SMlog("failed to generate message: %s" % e)

    def unpause(self):
        if not blktap2.VDI.tap_unpause(self.sr.xapi.session, self.sr.uuid,
                self.uuid):
            self._report_tapdisk_unpause_error()
            raise util.SMException("Failed to unpause VDI %s" % self)

    def refresh(self, ignoreNonexistent=True):
        """Pause-unpause in one step"""
        self.sr.lock()
        try:
            try:
                if not blktap2.VDI.tap_refresh(self.sr.xapi.session,
                        self.sr.uuid, self.uuid):
                    self._report_tapdisk_unpause_error()
                    raise util.SMException("Failed to refresh %s" % self)
            except XenAPI.Failure as e:
                if util.isInvalidVDI(e) and ignoreNonexistent:
                    Util.log("VDI %s not found, ignoring" % self)
                    return
                raise
        finally:
            self.sr.unlock()

    def isSnapshot(self):
        return self.sr.xapi.isSnapshot(self)

    def isAttachedRW(self):
        return util.is_attached_rw(
                self.sr.xapi.session.xenapi.VDI.get_sm_config(self.getRef()))

    def getVDIBlocks(self):
        val = self.updateBlockInfo()
        bitmap = zlib.decompress(base64.b64decode(val))
        return bitmap

    def isCoalesceable(self):
        """A VDI is coalesceable if it has no siblings and is not a leaf"""
        return (
            not self.scanError and
            self.parent and
            len(self.parent.children) == 1 and
            self.isHidden() and
            len(self.children) > 0 and (
                # Conditions below are Qcow2 specific:
                # A Qcow2 chain can't be coalesce with more than one leaf attached.
                # Put it another way, Qcow2 leaves activated on multiple hosts must
                # prevent the coalesce of the chain.
                self.vdi_type != VdiType.QCOW2 or
                (self.vdi_type == VdiType.QCOW2 and len(self.sr.hasLeavesAttachedOn(self)) <= 1)
            )
        )

    def isLeafCoalesceable(self):
        """A VDI is leaf-coalesceable if it has no siblings and is a leaf"""
        return not self.scanError and \
                self.parent and \
                len(self.parent.children) == 1 and \
                not self.isHidden() and \
                len(self.children) == 0

    def canLiveCoalesce(self, speed):
        """Can we stop-and-leaf-coalesce this VDI? The VDI must be
        isLeafCoalesceable() already"""
        feasibleSize = False
        allowedDownTime = \
                self.TIMEOUT_SAFETY_MARGIN * self.LIVE_LEAF_COALESCE_TIMEOUT
        allocated_size = self.getAllocatedSize()
        if speed:
            feasibleSize = \
                allocated_size // speed < allowedDownTime
        else:
            feasibleSize = \
                allocated_size < self.LIVE_LEAF_COALESCE_MAX_SIZE

        return (feasibleSize or
                self.getConfig(self.DB_LEAFCLSC) == self.LEAFCLSC_FORCE)

    def getAllPrunable(self):
        if len(self.children) == 0:  # base case
            # it is possible to have a hidden leaf that was recently coalesced
            # onto its parent, its children already relinked but not yet
            # reloaded - in which case it may not be garbage collected yet:
            # some tapdisks could still be using the file.
            if self.sr.journaler.get(self.JRN_RELINK, self.uuid):
                return []
            if not self.scanError and self.isHidden():
                return [self]
            return []

        thisPrunable = True
        vdiList = []
        for child in self.children:
            childList = child.getAllPrunable()
            vdiList.extend(childList)
            if child not in childList:
                thisPrunable = False

        # We can destroy the current VDI if all childs are hidden BUT the
        # current VDI must be hidden too to do that!
        # Example in this case (after a failed live leaf coalesce):
        #
        # SMGC: [32436] SR 07ed ('linstor-nvme-sr') (2 VDIs in 1 VHD trees):
        # SMGC: [32436]         b5458d61(1.000G/4.127M)
        # SMGC: [32436]             *OLD_b545(1.000G/4.129M)
        #
        # OLD_b545 is hidden and must be removed, but b5458d61 not.
        # Normally we are not in this function when the delete action is
        # executed but in `_liveLeafCoalesce`.

        if not self.scanError and not self.isHidden() and thisPrunable:
            vdiList.append(self)
        return vdiList

    def getSizePhys(self) -> int:
        return self._sizePhys

    def getAllocatedSize(self) -> int:
        return self._sizeAllocated

    def getTreeRoot(self):
        "Get the root of the tree that self belongs to"
        root = self
        while root.parent:
            root = root.parent
        return root

    def getTreeHeight(self):
        "Get the height of the subtree rooted at self"
        if len(self.children) == 0:
            return 1

        maxChildHeight = 0
        for child in self.children:
            childHeight = child.getTreeHeight()
            if childHeight > maxChildHeight:
                maxChildHeight = childHeight

        return maxChildHeight + 1

    def getAllLeaves(self) -> List["VDI"]:
        "Get all leaf nodes in the subtree rooted at self"
        if len(self.children) == 0:
            return [self]

        leaves = []
        for child in self.children:
            leaves.extend(child.getAllLeaves())
        return leaves

    def updateBlockInfo(self) -> Optional[str]:
        val = base64.b64encode(self._queryCowBlocks()).decode()
        try:
            self.setConfig(VDI.DB_VDI_BLOCKS, val)
        except Exception:
            if self.vdi_type != VdiType.QCOW2:
                raise
            # Sometime with QCOW2, our allocation table is too big to be stored in XAPI, in this case we do not store it
            # and we write `skipped` instead so that hasWork is happy (and the GC doesn't run in loop indefinitely).
            self.setConfig(VDI.DB_VDI_BLOCKS, "skipped")

        return val

    def rename(self, uuid) -> None:
        "Rename the VDI file"
        assert(not self.sr.vdis.get(uuid))
        self._clearRef()
        oldUuid = self.uuid
        self.uuid = uuid
        self.children = []
        # updating the children themselves is the responsibility of the caller
        del self.sr.vdis[oldUuid]
        self.sr.vdis[self.uuid] = self

    def delete(self) -> None:
        "Physically delete the VDI"
        lock.Lock.cleanup(self.uuid, NS_PREFIX_LVM + self.sr.uuid)
        lock.Lock.cleanupAll(self.uuid)
        self._clear()

    def getParent(self) -> str:
        return self.cowutil.getParent(self.path, lambda x: x.strip())

    def repair(self, parent) -> None:
        self.cowutil.repair(parent)

    @override
    def __str__(self) -> str:
        strHidden = ""
        if self.isHidden():
            strHidden = "*"
        strSizeVirt = "?"
        if self.sizeVirt > 0:
            strSizeVirt = Util.num2str(self.sizeVirt)
        strSizePhys = "?"
        if self._sizePhys > 0:
            strSizePhys = "/%s" % Util.num2str(self._sizePhys)
        strSizeAllocated = "?"
        if self._sizeAllocated >= 0:
            strSizeAllocated = "/%s" % Util.num2str(self._sizeAllocated)
        strType = "[{}]".format(self.vdi_type)

        return "%s%s(%s%s%s)%s" % (strHidden, self.uuid[0:8], strSizeVirt,
                strSizePhys, strSizeAllocated, strType)

    def validate(self, fast=False) -> None:
        if self.cowutil.check(self.path, fast=fast) != CowUtil.CheckResult.Success:
            raise util.SMException("COW image %s corrupted" % self)

    def _clear(self):
        self.uuid = ""
        self.path = ""
        self.parentUuid = ""
        self.parent = None
        self._clearRef()

    def _clearRef(self):
        self._vdiRef = None

    @staticmethod
    def _cancel_exception(sig, frame):
        raise CancelException()

    def _call_plugin_coalesce(self, hostRef, leaf):
        signal.signal(signal.SIGTERM, self._cancel_exception)
        args = {"path": self.path, "vdi_type": self.vdi_type, "leaf_path": leaf.path}
        Util.log("Calling remote coalesce plugin with: {}".format(args))
        try:
            ret = self.sr.xapi.session.xenapi.host.call_plugin( \
                    hostRef, XAPI.PLUGIN_ON_SLAVE, "commit_tapdisk", args)
            Util.log("Remote coalesce returned {}".format(ret))
        except CancelException:
            Util.log(f"Cancelling online coalesce following signal {args}")
            self.sr.xapi.session.xenapi.host.call_plugin( \
                hostRef, XAPI.PLUGIN_ON_SLAVE, "commit_cancel", args)
            raise
        except Exception:
            raise

    def _doCoalesceOnHost(self, hostRef, leaf):
        self.parent._increaseSizeVirt(self.sizeVirt)
        self.sr._updateSlavesOnResize(self.parent)

        self._coalesceCowImageOnHost(hostRef, leaf)

        #self._verifyContents(0)
        self.parent.updateBlockInfo()

    def _isOpenOnHosts(self) -> Optional[str]:
        for pbdRecord in self.sr.xapi.getAttachedPBDs():
            hostRef = pbdRecord["host"]
            args = {"path": self.path}
            is_openers = util.strtobool(self.sr.xapi.session.xenapi.host.call_plugin( \
                    hostRef, XAPI.PLUGIN_ON_SLAVE, "is_openers", args))
            if is_openers:
                return hostRef
        return None

    def _doCoalesce(self) -> None:
        """Coalesce self onto parent. Only perform the actual coalescing of
        an image, but not the subsequent relinking. We'll do that as the next step,
        after reloading the entire SR in case things have changed while we
        were coalescing"""
        self.validate()
        self.parent.validate(True)
        self.parent._increaseSizeVirt(self.sizeVirt)
        self.sr._updateSlavesOnResize(self.parent)
        self._coalesceCowImage(0)
        self.parent.validate(True)
        #self._verifyContents(0)
        self.parent.updateBlockInfo()

    def _verifyContents(self, timeOut):
        Util.log("  Coalesce verification on %s" % self)
        abortTest = lambda: IPCFlag(self.sr.uuid).test(FLAG_TYPE_ABORT)
        Util.runAbortable(lambda: self._runTapdiskDiff(), True,
                self.sr.uuid, abortTest, VDI.POLL_INTERVAL, timeOut)
        Util.log("  Coalesce verification succeeded")

    def _runTapdiskDiff(self):
        cmd = "tapdisk-diff -n %s:%s -m %s:%s" % \
                (self.getDriverName(), self.path, \
                self.parent.getDriverName(), self.parent.path)
        Util.doexec(cmd, 0)
        return True

    @staticmethod
    def _reportCoalesceError(vdi, ce):
        """Reports a coalesce error to XenCenter.

        vdi: the VDI object on which the coalesce error occured
        ce: the CommandException that was raised"""

        msg_name = os.strerror(ce.code)
        if ce.code == errno.ENOSPC:
            # TODO We could add more information here, e.g. exactly how much
            # space is required for the particular coalesce, as well as actions
            # to be taken by the user and consequences of not taking these
            # actions.
            msg_body = 'Run out of space while coalescing.'
        elif ce.code == errno.EIO:
            msg_body = 'I/O error while coalescing.'
        else:
            msg_body = ''
        util.SMlog('Coalesce failed on SR %s: %s (%s)'
                % (vdi.sr.uuid, msg_name, msg_body))

        # Create a XenCenter message, but don't spam.
        xapi = vdi.sr.xapi.session.xenapi
        sr_ref = xapi.SR.get_by_uuid(vdi.sr.uuid)
        oth_cfg = xapi.SR.get_other_config(sr_ref)
        if COALESCE_ERR_RATE_TAG in oth_cfg:
            coalesce_err_rate = float(oth_cfg[COALESCE_ERR_RATE_TAG])
        else:
            coalesce_err_rate = DEFAULT_COALESCE_ERR_RATE

        xcmsg = False
        if coalesce_err_rate == 0:
            xcmsg = True
        elif coalesce_err_rate > 0:
            now = datetime.datetime.now()
            sm_cfg = xapi.SR.get_sm_config(sr_ref)
            if COALESCE_LAST_ERR_TAG in sm_cfg:
                # seconds per message (minimum distance in time between two
                # messages in seconds)
                spm = datetime.timedelta(seconds=(1.0 / coalesce_err_rate) * 60)
                last = datetime.datetime.fromtimestamp(
                        float(sm_cfg[COALESCE_LAST_ERR_TAG]))
                if now - last >= spm:
                    xapi.SR.remove_from_sm_config(sr_ref,
                            COALESCE_LAST_ERR_TAG)
                    xcmsg = True
            else:
                xcmsg = True
            if xcmsg:
                xapi.SR.add_to_sm_config(sr_ref, COALESCE_LAST_ERR_TAG,
                        str(now.strftime('%s')))
        if xcmsg:
            xapi.message.create(msg_name, "3", "SR", vdi.sr.uuid, msg_body)

    def coalesce(self) -> int:
        return self.cowutil.coalesce(self.path)

    @staticmethod
    def _doCoalesceCowImage(vdi: "VDI"):
        try:
            startTime = time.time()
            allocated_size = vdi.getAllocatedSize()
            coalesced_size = vdi.coalesce()
            endTime = time.time()
            vdi.sr.recordStorageSpeed(startTime, endTime, coalesced_size)
        except util.CommandException as ce:
            # We use try/except for the following piece of code because it runs
            # in a separate process context and errors will not be caught and
            # reported by anyone.
            try:
                # Report coalesce errors back to user via XC
                VDI._reportCoalesceError(vdi, ce)
            except Exception as e:
                util.SMlog('failed to create XenCenter message: %s' % e)
            raise ce
        except:
            raise

    def _vdi_is_raw(self, vdi_path):
        """
        Given path to vdi determine if it is raw
        """
        uuid = self.extractUuid(vdi_path)
        return self.sr.vdis[uuid].vdi_type == VdiType.RAW

    def _coalesceCowImage(self, timeOut):
        Util.log("  Running COW coalesce on %s" % self)
        def abortTest():
            if self.cowutil.isCoalesceableOnRemote():
                file = self.sr._gc_running_file(self)
                try:
                    with open(file, "r") as f:
                        if not f.read():
                            return True
                except OSError as e:
                    if e.errno == errno.ENOENT:
                        util.SMlog("File {} does not exist".format(file))
                    else:
                        util.SMlog("IOError: {}".format(e))
                    return True
            return IPCFlag(self.sr.uuid).test(FLAG_TYPE_ABORT)

        try:
            util.fistpoint.activate_custom_fn(
                "cleanup_coalesceVHD_inject_failure",
                util.inject_failure)
            Util.runAbortable(lambda: VDI._doCoalesceCowImage(self), None,
                    self.sr.uuid, abortTest, VDI.POLL_INTERVAL, timeOut)
        except:
            # Exception at this phase could indicate a failure in COW coalesce
            # or a kill of COW coalesce by runAbortable due to timeOut
            # Try a repair and reraise the exception
            parent = ""
            try:
                parent = self.getParent()
                if not self._vdi_is_raw(parent):
                    # Repair error is logged and ignored. Error reraised later
                    util.SMlog('Coalesce failed on %s, attempting repair on ' \
                               'parent %s' % (self.uuid, parent))
                    self.repair(parent)
            except Exception as e:
                util.SMlog('(error ignored) Failed to repair parent %s ' \
                           'after failed coalesce on %s, err: %s' %
                           (parent, self.path, e))
            raise

        util.fistpoint.activate("LVHDRT_coalescing_VHD_data", self.sr.uuid)

    def _coalesceCowImageOnHost(self, hostRef, leaf):
        Util.log("  Running COW coalesce on {} via remote host {}".format(self, hostRef))
        def abortTest():
            file = self.sr._gc_running_file(self)
            try:
                with open(file, "r") as f:
                    if not f.read():
                        Util.log("abortTest: Cancelling coalesce")
                        return True
            except OSError as e:
                if e.errno == errno.ENOENT:
                    Util.log("File {} does not exist".format(file))
                else:
                    Util.log("IOError: {}".format(e))
                return True
            return False

        Util.runAbortable(lambda: self._call_plugin_coalesce(hostRef, leaf),
                          None, self.sr.uuid, abortTest, VDI.POLL_INTERVAL, 0, prefSig=signal.SIGTERM)

    def _relinkSkip(self) -> None:
        """Relink children of this VDI to point to the parent of this VDI"""
        abortFlag = IPCFlag(self.sr.uuid)
        for child in self.children:
            if abortFlag.test(FLAG_TYPE_ABORT):
                raise AbortException("Aborting due to signal")
            Util.log("  Relinking %s from %s to %s" % \
                    (child, self, self.parent))
            util.fistpoint.activate("LVHDRT_relinking_grandchildren", self.sr.uuid)
            child._setParent(self.parent)
        self.children = []

    def _reloadChildren(self, vdiSkip):
        """Pause & unpause all VDIs in the subtree to cause blktap to reload
        the COW image metadata for this file in any online VDI"""
        abortFlag = IPCFlag(self.sr.uuid)
        for child in self.children:
            if child == vdiSkip:
                continue
            if abortFlag.test(FLAG_TYPE_ABORT):
                raise AbortException("Aborting due to signal")
            Util.log("  Reloading VDI %s" % child)
            child._reload()

    def _reload(self):
        """Pause & unpause to cause blktap to reload the image metadata"""
        for child in self.children:
            child._reload()

        # only leaves can be attached
        if len(self.children) == 0:
            try:
                self.delConfig(VDI.DB_VDI_RELINKING)
            except XenAPI.Failure as e:
                if not util.isInvalidVDI(e):
                    raise
            self.refresh()

    def _needRelink(self, list_not_to_relink):
        """
        If we coalesce up the chain, we shouldn't need to do the relink at all, we only need to do the relink on the children if their direct parent was the one we were coalescing
        """
        if not list_not_to_relink:
            return True
        if self.uuid in list_not_to_relink:
            return False
        else:
            return True

    def _tagChildrenForRelink(self, list_not_to_relink=None):
        if len(self.children) == 0:
            retries = 0
            try:
                while retries < 15:
                    retries += 1
                    if self.getConfig(VDI.DB_VDI_ACTIVATING) is not None:
                        Util.log("VDI %s is activating, wait to relink" %
                                 self.uuid)
                    else:
                        if self._needRelink(list_not_to_relink):
                            self.setConfig(VDI.DB_VDI_RELINKING, "True")

                            if self.getConfig(VDI.DB_VDI_ACTIVATING):
                                self.delConfig(VDI.DB_VDI_RELINKING)
                                Util.log("VDI %s started activating while tagging" %
                                        self.uuid)
                            else:
                                return
                        else:
                            Util.log(f"Not adding relinking tag to VDI {self.uuid}")
                            return
                    time.sleep(2)

                raise util.SMException("Failed to tag vdi %s for relink" % self)
            except XenAPI.Failure as e:
                if not util.isInvalidVDI(e):
                    raise

        for child in self.children:
            child._tagChildrenForRelink(list_not_to_relink)

    def _loadInfoParent(self):
        ret = self.cowutil.getParent(self.path, LvmCowUtil.extractUuid)
        if ret:
            self.parentUuid = ret

    def _setParent(self, parent) -> None:
        self.cowutil.setParent(self.path, parent.path, False)
        self.parent = parent
        self.parentUuid = parent.uuid
        parent.children.append(self)
        try:
            self.setConfig(self.DB_VDI_PARENT, self.parentUuid)
            Util.log("Updated the vhd-parent field for child %s with %s" % \
                     (self.uuid, self.parentUuid))
        except:
            Util.log("Failed to update %s with vhd-parent field %s" % \
                     (self.uuid, self.parentUuid))

    def _ensureParentActiveForRelink(self) -> None:
        pass

    def _update_vhd_parent(self, real_parent_uuid):
        try:
            self.setConfig(self.DB_VDI_PARENT, real_parent_uuid)
            Util.log("Updated the vhd-parent field for child %s with real parent %s following a online coalesce" % \
                    (self.uuid, real_parent_uuid))
        except:
            Util.log("Failed to update %s with vhd-parent field %s" % \
                    (self.uuid, real_parent_uuid))

    def isHidden(self) -> bool:
        if self._hidden is None:
            self._loadInfoHidden()
        return self._hidden

    def _loadInfoHidden(self) -> None:
        hidden = self.cowutil.getHidden(self.path)
        self._hidden = (hidden != 0)

    def _setHidden(self, hidden=True) -> None:
        self._hidden = None
        self.cowutil.setHidden(self.path, hidden)
        self._hidden = hidden

    def _increaseSizeVirt(self, size, atomic=True) -> None:
        """ensure the virtual size of 'self' is at least 'size'. Note that
        resizing a COW image must always be offline and atomically: the file must
        not be open by anyone and no concurrent operations may take place.
        Thus we use the Agent API call for performing paused atomic
        operations. If the caller is already in the atomic context, it must
        call with atomic = False"""
        if self.sizeVirt >= size:
            return
        Util.log("  Expanding COW image virt size for VDI %s: %s -> %s" % \
                (self, Util.num2str(self.sizeVirt), Util.num2str(size)))

        msize = self.cowutil.getMaxResizeSize(self.path)
        if (size <= msize):
            self.cowutil.setSizeVirtFast(self.path, size)
        else:
            if atomic:
                vdiList = self._getAllSubtree()
                self.sr.lock()
                try:
                    self.sr.pauseVDIs(vdiList)
                    try:
                        self._setSizeVirt(size)
                    finally:
                        self.sr.unpauseVDIs(vdiList)
                finally:
                    self.sr.unlock()
            else:
                self._setSizeVirt(size)

        self.sizeVirt = self.cowutil.getSizeVirt(self.path)

    def _setSizeVirt(self, size) -> None:
        """WARNING: do not call this method directly unless all VDIs in the
        subtree are guaranteed to be unplugged (and remain so for the duration
        of the operation): this operation is only safe for offline COW images"""
        jFile = os.path.join(self.sr.path, self.uuid)
        self.cowutil.setSizeVirt(self.path, size, jFile)

    def _queryCowBlocks(self) -> bytes:
        return self.cowutil.getBlockBitmap(self.path)

    def _getCoalescedSizeData(self):
        """Get the data size of the resulting image if we coalesce self onto
        parent. We calculate the actual size by using the image block allocation
        information (as opposed to just adding up the two image sizes to get an
        upper bound)"""
        # make sure we don't use stale BAT info from vdi_rec since the child
        # was writable all this time
        self.delConfig(VDI.DB_VDI_BLOCKS)
        blocksChild = self.getVDIBlocks()
        blocksParent = self.parent.getVDIBlocks()
        numBlocks = Util.countBits(blocksChild, blocksParent)
        Util.log("Num combined blocks = %d" % numBlocks)
        sizeData = numBlocks * self.cowutil.getBlockSize(self.path)
        assert(sizeData <= self.sizeVirt)
        return sizeData

    def _calcExtraSpaceForCoalescing(self) -> int:
        sizeData = self._getCoalescedSizeData()
        sizeCoalesced = sizeData + self.cowutil.calcOverheadBitmap(sizeData) + \
                self.cowutil.calcOverheadEmpty(self.sizeVirt)
        Util.log("Coalesced size = %s" % Util.num2str(sizeCoalesced))
        return sizeCoalesced - self.parent.getSizePhys()

    def _calcExtraSpaceForLeafCoalescing(self) -> int:
        """How much extra space in the SR will be required to
        [live-]leaf-coalesce this VDI"""
        # the space requirements are the same as for inline coalesce
        return self._calcExtraSpaceForCoalescing()

    def _calcExtraSpaceForSnapshotCoalescing(self) -> int:
        """How much extra space in the SR will be required to
        snapshot-coalesce this VDI"""
        return self._calcExtraSpaceForCoalescing() + \
                self.cowutil.calcOverheadEmpty(self.sizeVirt)  # extra snap leaf

    def _getAllSubtree(self):
        """Get self and all VDIs in the subtree of self as a flat list"""
        vdiList = [self]
        for child in self.children:
            vdiList.extend(child._getAllSubtree())
        return vdiList


class FileVDI(VDI):
    """Object representing a VDI in a file-based SR (EXT or NFS)"""

    @override
    @staticmethod
    def extractUuid(path):
        fileName = os.path.basename(path)
        return os.path.splitext(fileName)[0]

    def __init__(self, sr, uuid, vdi_type):
        VDI.__init__(self, sr, uuid, vdi_type)
        self.fileName = "%s%s" % (self.uuid, VDI_TYPE_TO_EXTENSION[self.vdi_type])

    @override
    def load(self, info=None) -> None:
        if not info:
            if not util.pathexists(self.path):
                raise util.SMException("%s not found" % self.path)
            try:
                info = self.cowutil.getInfo(self.path, self.extractUuid)
            except util.SMException:
                Util.log(" [VDI %s: failed to read COW image metadata]" % self.uuid)
                return
        self.parent = None
        self.children = []
        self.parentUuid = info.parentUuid
        self.sizeVirt = info.sizeVirt
        self._sizePhys = info.sizePhys
        self._sizeAllocated = info.sizeAllocated
        self._hidden = info.hidden
        self.scanError = False
        self.path = os.path.join(self.sr.path, "%s%s" % \
                (self.uuid, VDI_TYPE_TO_EXTENSION[self.vdi_type]))

    @override
    def rename(self, uuid) -> None:
        oldPath = self.path
        VDI.rename(self, uuid)
        self.fileName = "%s%s" % (self.uuid, VDI_TYPE_TO_EXTENSION[self.vdi_type])
        self.path = os.path.join(self.sr.path, self.fileName)
        assert(not util.pathexists(self.path))
        Util.log("Renaming %s -> %s" % (oldPath, self.path))
        os.rename(oldPath, self.path)

    @override
    def delete(self) -> None:
        if len(self.children) > 0:
            raise util.SMException("VDI %s has children, can't delete" % \
                    self.uuid)
        try:
            self.sr.lock()
            try:
                os.unlink(self.path)
                self.sr.forgetVDI(self.uuid)
            finally:
                self.sr.unlock()
        except OSError:
            raise util.SMException("os.unlink(%s) failed" % self.path)
        VDI.delete(self)

    @override
    def getAllocatedSize(self) -> int:
        if self._sizeAllocated == -1:
            self._sizeAllocated = self.cowutil.getAllocatedSize(self.path)
        return self._sizeAllocated


class LVMVDI(VDI):
    """Object representing a VDI in an LVM SR"""

    JRN_ZERO = "zero"  # journal entry type for zeroing out end of parent

    @override
    def load(self, info=None) -> None:
        # `info` is always set. `None` default value is only here to match parent method.
        assert info, "No info given to LVMVDI.load"
        self.parent = None
        self.children = []
        self._sizePhys = -1
        self._sizeAllocated = -1
        self.scanError = info.scanError
        self.sizeLV = info.sizeLV
        self.sizeVirt = info.sizeVirt
        self.fileName = info.lvName
        self.lvActive = info.lvActive
        self.lvOpen = info.lvOpen
        self.lvReadonly = info.lvReadonly
        self._hidden = info.hidden
        self.parentUuid = info.parentUuid
        self.path = os.path.join(self.sr.path, self.fileName)
        self.lvmcowutil = LvmCowUtil(self.cowutil)

    @override
    @staticmethod
    def extractUuid(path):
        return LvmCowUtil.extractUuid(path)

    def inflate(self, size):
        """inflate the LV containing the COW image to 'size'"""
        if not VdiType.isCowImage(self.vdi_type):
            return
        self._activate()
        self.sr.lock()
        try:
            self.lvmcowutil.inflate(self.sr.journaler, self.sr.uuid, self.uuid, self.vdi_type, size)
            util.fistpoint.activate("LVHDRT_inflating_the_parent", self.sr.uuid)
        finally:
            self.sr.unlock()
        self.sizeLV = self.sr.lvmCache.getSize(self.fileName)
        self._sizePhys = -1
        self._sizeAllocated = -1

    def deflate(self):
        """deflate the LV containing the image to minimum"""
        if not VdiType.isCowImage(self.vdi_type):
            return
        self._activate()
        self.sr.lock()
        try:
            self.lvmcowutil.deflate(self.sr.lvmCache, self.fileName, self.getSizePhys())
        finally:
            self.sr.unlock()
        self.sizeLV = self.sr.lvmCache.getSize(self.fileName)
        self._sizePhys = -1
        self._sizeAllocated = -1

    def inflateFully(self):
        self.inflate(self.lvmcowutil.calcVolumeSize(self.sizeVirt))

    def inflateParentForCoalesce(self):
        """Inflate the parent only as much as needed for the purposes of
        coalescing"""
        if not VdiType.isCowImage(self.parent.vdi_type):
            return
        inc = self._calcExtraSpaceForCoalescing()
        if inc > 0:
            util.fistpoint.activate("LVHDRT_coalescing_before_inflate_grandparent", self.sr.uuid)
            self.parent.inflate(self.parent.sizeLV + inc)

    @override
    def updateBlockInfo(self) -> Optional[str]:
        if VdiType.isCowImage(self.vdi_type):
            return VDI.updateBlockInfo(self)
        return None

    @override
    def rename(self, uuid) -> None:
        oldUuid = self.uuid
        oldLVName = self.fileName
        VDI.rename(self, uuid)
        self.fileName = LV_PREFIX[self.vdi_type] + self.uuid
        self.path = os.path.join(self.sr.path, self.fileName)
        assert(not self.sr.lvmCache.checkLV(self.fileName))

        self.sr.lvmCache.rename(oldLVName, self.fileName)
        if self.sr.lvActivator.get(oldUuid, False):
            self.sr.lvActivator.replace(oldUuid, self.uuid, self.fileName, False)

        ns = NS_PREFIX_LVM + self.sr.uuid
        (cnt, bcnt) = RefCounter.check(oldUuid, ns)
        RefCounter.set(self.uuid, cnt, bcnt, ns)
        RefCounter.reset(oldUuid, ns)

    @override
    def delete(self) -> None:
        if len(self.children) > 0:
            raise util.SMException("VDI %s has children, can't delete" % \
                    self.uuid)
        self.sr.lock()
        try:
            self.sr.lvmCache.remove(self.fileName)
            self.sr.forgetVDI(self.uuid)
        finally:
            self.sr.unlock()
        RefCounter.reset(self.uuid, NS_PREFIX_LVM + self.sr.uuid)
        VDI.delete(self)

    @override
    def getSizePhys(self) -> int:
        if self._sizePhys == -1:
            self._loadInfoSizePhys()
        return self._sizePhys

    def _loadInfoSizePhys(self):
        """Get the physical utilization of the COW image file. We do it individually
        (and not using the COW batch scanner) as an optimization: this info is
        relatively expensive and we need it only for VDI's involved in
        coalescing."""
        if not VdiType.isCowImage(self.vdi_type):
            return
        self._activate()
        self._sizePhys = self.cowutil.getSizePhys(self.path)
        if self._sizePhys <= 0:
            raise util.SMException("phys size of %s = %d" % \
                    (self, self._sizePhys))

    @override
    def getAllocatedSize(self) -> int:
        if self._sizeAllocated == -1:
            self._loadInfoSizeAllocated()
        return self._sizeAllocated

    def _loadInfoSizeAllocated(self):
        """
        Get the allocated size of the COW volume.
        """
        if not VdiType.isCowImage(self.vdi_type):
            return
        self._activate()
        self._sizeAllocated = self.cowutil.getAllocatedSize(self.path)

    @override
    def _loadInfoHidden(self) -> None:
        if not VdiType.isCowImage(self.vdi_type):
            self._hidden = self.sr.lvmCache.getHidden(self.fileName)
        else:
            VDI._loadInfoHidden(self)

    @override
    def _setHidden(self, hidden=True) -> None:
        if not VdiType.isCowImage(self.vdi_type):
            self._hidden = None
            self.sr.lvmCache.setHidden(self.fileName, hidden)
            self._hidden = hidden
        else:
            VDI._setHidden(self, hidden)

    @override
    def __str__(self) -> str:
        strType = self.vdi_type
        if self.vdi_type == VdiType.RAW:
            strType = "RAW"
        strHidden = ""
        if self.isHidden():
            strHidden = "*"
        strSizePhys = ""
        if self._sizePhys > 0:
            strSizePhys = Util.num2str(self._sizePhys)
        strSizeAllocated = ""
        if self._sizeAllocated >= 0:
            strSizeAllocated = Util.num2str(self._sizeAllocated)
        strActive = "n"
        if self.lvActive:
            strActive = "a"
        if self.lvOpen:
            strActive += "o"
        return "%s%s[%s](%s/%s/%s/%s|%s)" % (strHidden, self.uuid[0:8], strType,
                Util.num2str(self.sizeVirt), strSizePhys, strSizeAllocated,
                Util.num2str(self.sizeLV), strActive)

    @override
    def validate(self, fast=False) -> None:
        if VdiType.isCowImage(self.vdi_type):
            VDI.validate(self, fast)

    def _setChainRw(self) -> List[str]:
        """
        Set the readonly LV and children writable.
        It's needed because the coalesce can be done by tapdisk directly
        and it will need to write parent information for children.
        The VDI we want to coalesce into it's parent need to be writable for libqcow coalesce part.
        Return a list of the LV that were previously readonly to be made RO again after the coalesce.
        """
        was_ro = []
        if self.lvReadonly:
            self.sr.lvmCache.setReadonly(self.fileName, False)
            was_ro.append(self.fileName)

        for child in self.children:
            if child.lvReadonly:
                self.sr.lvmCache.setReadonly(child.fileName, False)
                was_ro.append(child.fileName)

        return was_ro

    def _setChainRo(self, was_ro: List[str]) -> None:
        """Set the list of LV in parameters to readonly"""
        for lvName in was_ro:
            self.sr.lvmCache.setReadonly(lvName, True)

    @override
    def _doCoalesce(self) -> None:
        """LVMVDI parents must first be activated, inflated, and made writable"""
        was_ro = []
        try:
            self._activateChain()
            self.sr.lvmCache.setReadonly(self.parent.fileName, False)
            self.parent.validate()
            self.inflateParentForCoalesce()
            was_ro = self._setChainRw()
            VDI._doCoalesce(self)
        finally:
            self.parent._loadInfoSizePhys()
            self.parent.deflate()
            self.sr.lvmCache.setReadonly(self.parent.fileName, True)
            self._setChainRo(was_ro)

    @override
    def _setParent(self, parent) -> None:
        self._activate()
        if self.lvReadonly:
            self.sr.lvmCache.setReadonly(self.fileName, False)

        try:
            self.cowutil.setParent(self.path, parent.path, parent.vdi_type == VdiType.RAW)
        finally:
            if self.lvReadonly:
                self.sr.lvmCache.setReadonly(self.fileName, True)
        self._deactivate()
        self.parent = parent
        self.parentUuid = parent.uuid
        parent.children.append(self)
        try:
            self.setConfig(self.DB_VDI_PARENT, self.parentUuid)
            Util.log("Updated the VDI-parent field for child %s with %s" % \
                     (self.uuid, self.parentUuid))
        except:
            Util.log("Failed to update the VDI-parent with %s for child %s" % \
                     (self.parentUuid, self.uuid))

    def _activate(self):
        self.sr.lvActivator.activate(self.uuid, self.fileName, False)

    def _activateChain(self):
        vdi = self
        while vdi:
            vdi._activate()
            vdi = vdi.parent

    def _deactivate(self):
        self.sr.lvActivator.deactivate(self.uuid, False)

    @override
    def _ensureParentActiveForRelink(self) -> None:
        self.parent._activate()

    @override
    def _increaseSizeVirt(self, size, atomic=True) -> None:
        "ensure the virtual size of 'self' is at least 'size'"
        self._activate()
        if VdiType.isCowImage(self.vdi_type):
            VDI._increaseSizeVirt(self, size, atomic)
            return

        # raw VDI case
        offset = self.sizeLV
        if self.sizeVirt < size:
            oldSize = self.sizeLV
            self.sizeLV = util.roundup(lvutil.LVM_SIZE_INCREMENT, size)
            Util.log("  Growing %s: %d->%d" % (self.path, oldSize, self.sizeLV))
            self.sr.lvmCache.setSize(self.fileName, self.sizeLV)
            offset = oldSize
        unfinishedZero = False
        jval = self.sr.journaler.get(self.JRN_ZERO, self.uuid)
        if jval:
            unfinishedZero = True
            offset = int(jval)
        length = self.sizeLV - offset
        if not length:
            return

        if unfinishedZero:
            Util.log("  ==> Redoing unfinished zeroing out")
        else:
            self.sr.journaler.create(self.JRN_ZERO, self.uuid, \
                    str(offset))
        Util.log("  Zeroing %s: from %d, %dB" % (self.path, offset, length))
        abortTest = lambda: IPCFlag(self.sr.uuid).test(FLAG_TYPE_ABORT)
        func = lambda: util.zeroOut(self.path, offset, length)
        Util.runAbortable(func, True, self.sr.uuid, abortTest,
                VDI.POLL_INTERVAL, 0)
        self.sr.journaler.remove(self.JRN_ZERO, self.uuid)

    @override
    def _setSizeVirt(self, size) -> None:
        """WARNING: do not call this method directly unless all VDIs in the
        subtree are guaranteed to be unplugged (and remain so for the duration
        of the operation): this operation is only safe for offline COW images."""
        self._activate()
        jFile = self.lvmcowutil.createResizeJournal(self.sr.lvmCache, self.uuid)
        try:
            self.lvmcowutil.setSizeVirt(self.sr.journaler, self.sr.uuid, self.uuid, self.vdi_type, size, jFile)
        finally:
            self.lvmcowutil.destroyResizeJournal(self.sr.lvmCache, self.uuid)

    @override
    def _queryCowBlocks(self) -> bytes:
        self._activate()
        return VDI._queryCowBlocks(self)

    @override
    def getParent(self) -> str:
        self._activate()
        parent = VDI.getParent(self)
        self._deactivate()
        return parent

    @override
    def _calcExtraSpaceForCoalescing(self) -> int:
        if not VdiType.isCowImage(self.parent.vdi_type):
            return 0  # raw parents are never deflated in the first place
        sizeCoalesced = self.lvmcowutil.calcVolumeSize(self._getCoalescedSizeData())
        Util.log("Coalesced size = %s" % Util.num2str(sizeCoalesced))
        return sizeCoalesced - self.parent.sizeLV

    @override
    def _calcExtraSpaceForLeafCoalescing(self) -> int:
        """How much extra space in the SR will be required to
        [live-]leaf-coalesce this VDI"""
        # we can deflate the leaf to minimize the space requirements
        deflateDiff = self.sizeLV - lvutil.calcSizeLV(self.getSizePhys())
        return self._calcExtraSpaceForCoalescing() - deflateDiff

    @override
    def _calcExtraSpaceForSnapshotCoalescing(self) -> int:
        return self._calcExtraSpaceForCoalescing() + \
                lvutil.calcSizeLV(self.getSizePhys())


class LinstorVDI(VDI):
    """Object representing a VDI in a LINSTOR SR"""

    VOLUME_LOCK_TIMEOUT = 30

    @override
    def load(self, info=None) -> None:
        self.parentUuid = info.parentUuid
        self.scanError = True
        self.parent = None
        self.children = []

        self.fileName = self.sr._linstor.get_volume_name(self.uuid)
        self.path = self.sr._linstor.build_device_path(self.fileName)
        self.linstorcowutil = LinstorCowUtil(self.sr.xapi.session, self.sr._linstor, info.vdiType)

        if not info:
            try:
                info = self.linstorcowutil.get_info(self.uuid)
            except util.SMException:
                Util.log(
                    ' [VDI {}: failed to read COW image metadata]'.format(self.uuid)
                )
                return

        self.parentUuid = info.parentUuid
        self.sizeVirt = info.sizeVirt
        self._sizePhys = -1
        self._sizeAllocated = -1
        self.drbd_size = -1
        self._hidden = info.hidden
        self.scanError = False

    @override
    def getSizePhys(self, fetch=False) -> int:
        if self._sizePhys < 0 or fetch:
            self._sizePhys = self.linstorcowutil.get_size_phys(self.uuid)
        return self._sizePhys

    def getDrbdSize(self, fetch=False):
        if self.drbd_size < 0 or fetch:
            self.drbd_size = self.linstorcowutil.get_drbd_size(self.uuid)
        return self.drbd_size

    @override
    def getAllocatedSize(self) -> int:
        if self._sizeAllocated == -1:
            if VdiType.isCowImage(self.vdi_type):
                self._sizeAllocated = self.linstorcowutil.get_allocated_size(self.uuid)
        return self._sizeAllocated

    def inflate(self, size):
        if not VdiType.isCowImage(self.vdi_type):
            return
        self.sr.lock()
        try:
            # Ensure we use the real DRBD size and not the cached one.
            # Why? Because this attribute can be changed if volume is resized by user.
            self.drbd_size = self.getDrbdSize(fetch=True)
            self.linstorcowutil.inflate(self.sr.journaler, self.uuid, self.path, size, self.drbd_size)
        finally:
            self.sr.unlock()
        self.drbd_size = -1
        self._sizePhys = -1
        self._sizeAllocated = -1

    def deflate(self):
        if not VdiType.isCowImage(self.vdi_type):
            return
        self.sr.lock()
        try:
            # Ensure we use the real sizes and not the cached info.
            self.drbd_size = self.getDrbdSize(fetch=True)
            self._sizePhys = self.getSizePhys(fetch=True)
            self.linstorcowutil.force_deflate(self.path, self._sizePhys, self.drbd_size, zeroize=False)
        finally:
            self.sr.unlock()
        self.drbd_size = -1
        self._sizePhys = -1
        self._sizeAllocated = -1

    def inflateFully(self):
        if VdiType.isCowImage(self.vdi_type):
            self.inflate(self.linstorcowutil.compute_volume_size(self.sizeVirt))

    @override
    def rename(self, uuid) -> None:
        Util.log('Renaming {} -> {} (path={})'.format(
            self.uuid, uuid, self.path
        ))
        self.sr._linstor.update_volume_uuid(self.uuid, uuid)
        VDI.rename(self, uuid)

    @override
    def delete(self) -> None:
        if len(self.children) > 0:
            raise util.SMException(
                'VDI {} has children, can\'t delete'.format(self.uuid)
            )
        self.sr.lock()
        try:
            self.sr._linstor.destroy_volume(self.uuid)
            self.sr.forgetVDI(self.uuid)
        finally:
            self.sr.unlock()
        VDI.delete(self)

    @override
    def validate(self, fast=False) -> None:
        if VdiType.isCowImage(self.vdi_type) and self.linstorcowutil.check(self.uuid, fast=fast) != CowUtil.CheckResult.Success:
            raise util.SMException('COW image {} corrupted'.format(self))

    @override
    def pause(self, failfast=False) -> None:
        self.sr._linstor.ensure_volume_is_not_locked(
            self.uuid, timeout=self.VOLUME_LOCK_TIMEOUT
        )
        return super(LinstorVDI, self).pause(failfast)

    @override
    def coalesce(self) -> int:
        # Note: We raise `SMException` here to skip the current coalesce in case of failure.
        # Using another exception we can't execute the next coalesce calls.
        return self.linstorcowutil.force_coalesce(self.path)

    @override
    def getParent(self) -> str:
        return self.linstorcowutil.get_parent(
            self.sr._linstor.get_volume_uuid_from_device_path(self.path)
        )

    @override
    def repair(self, parent_uuid) -> None:
        self.linstorcowutil.force_repair(
            self.sr._linstor.get_device_path(parent_uuid)
        )

    @override
    def _relinkSkip(self) -> None:
        abortFlag = IPCFlag(self.sr.uuid)
        for child in self.children:
            if abortFlag.test(FLAG_TYPE_ABORT):
                raise AbortException('Aborting due to signal')
            Util.log(
                '  Relinking {} from {} to {}'.format(
                    child, self, self.parent
                )
            )

            session = child.sr.xapi.session
            sr_uuid = child.sr.uuid
            vdi_uuid = child.uuid
            try:
                self.sr._linstor.ensure_volume_is_not_locked(
                    vdi_uuid, timeout=self.VOLUME_LOCK_TIMEOUT
                )
                blktap2.VDI.tap_pause(session, sr_uuid, vdi_uuid)
                child._setParent(self.parent)
            finally:
                blktap2.VDI.tap_unpause(session, sr_uuid, vdi_uuid)
        self.children = []

    @override
    def _setParent(self, parent) -> None:
        self.sr._linstor.get_device_path(self.uuid)
        self.linstorcowutil.force_parent(self.path, parent.path)
        self.parent = parent
        self.parentUuid = parent.uuid
        parent.children.append(self)
        try:
            self.setConfig(self.DB_VDI_PARENT, self.parentUuid)
            Util.log("Updated the vhd-parent field for child %s with %s" % \
                     (self.uuid, self.parentUuid))
        except:
            Util.log("Failed to update %s with vhd-parent field %s" % \
                     (self.uuid, self.parentUuid))

    @override
    def _doCoalesce(self) -> None:
        try:
            self._activateChain()
            self.parent.validate()
            self._inflateParentForCoalesce()
            VDI._doCoalesce(self)
        finally:
            self.parent.deflate()

    def _activateChain(self):
        vdi = self
        while vdi:
            try:
                p = self.sr._linstor.get_device_path(vdi.uuid)
            except Exception as e:
                # Use SMException to skip coalesce.
                # Otherwise the GC is stopped...
                raise util.SMException(str(e))
            vdi = vdi.parent

    @override
    def _setHidden(self, hidden=True) -> None:
        HIDDEN_TAG = 'hidden'

        if not VdiType.isCowImage(self.vdi_type):
            self._hidden = None
            self.sr._linstor.update_volume_metadata(self.uuid, {
                HIDDEN_TAG: hidden
            })
            self._hidden = hidden
        else:
            VDI._setHidden(self, hidden)

    @override
    def _increaseSizeVirt(self, size, atomic=True):
        if self.vdi_type == VdiType.RAW:
            offset = self.drbd_size
            if self.sizeVirt < size:
                oldSize = self.drbd_size
                self.drbd_size = LinstorVolumeManager.round_up_volume_size(size)
                Util.log("  Growing %s: %d->%d" % (self.path, oldSize, self.drbd_size))
                self.sr._linstor.resize_volume(self.uuid, self.drbd_size)
                offset = oldSize
            unfinishedZero = False
            jval = self.sr.journaler.get(LinstorJournaler.ZERO, self.uuid)
            if jval:
                unfinishedZero = True
                offset = int(jval)
            length = self.drbd_size - offset
            if not length:
                return

            if unfinishedZero:
                Util.log("  ==> Redoing unfinished zeroing out")
            else:
                self.sr.journaler.create(LinstorJournaler.ZERO, self.uuid, str(offset))
            Util.log("  Zeroing %s: from %d, %dB" % (self.path, offset, length))
            abortTest = lambda: IPCFlag(self.sr.uuid).test(FLAG_TYPE_ABORT)
            func = lambda: util.zeroOut(self.path, offset, length)
            Util.runAbortable(func, True, self.sr.uuid, abortTest, VDI.POLL_INTERVAL, 0)
            self.sr.journaler.remove(LinstorJournaler.ZERO, self.uuid)
            return

        if self.sizeVirt >= size:
            return
        Util.log("  Expanding COW image virt size for VDI %s: %s -> %s" % \
                (self, Util.num2str(self.sizeVirt), Util.num2str(size)))

        msize = self.linstorcowutil.get_max_resize_size(self.uuid) * 1024 * 1024
        if (size <= msize):
            self.linstorcowutil.set_size_virt_fast(self.path, size)
        else:
            if atomic:
                vdiList = self._getAllSubtree()
                self.sr.lock()
                try:
                    self.sr.pauseVDIs(vdiList)
                    try:
                        self._setSizeVirt(size)
                    finally:
                        self.sr.unpauseVDIs(vdiList)
                finally:
                    self.sr.unlock()
            else:
                self._setSizeVirt(size)

        self.sizeVirt = self.linstorcowutil.get_size_virt(self.uuid)

    @override
    def _setSizeVirt(self, size) -> None:
        jfile = self.uuid + '-jvhd'
        self.sr._linstor.create_volume(
            jfile, self.cowutil.getResizeJournalSize(), persistent=False, volume_name=jfile
        )
        try:
            self.inflate(self.linstorcowutil.compute_volume_size(size))
            self.linstorcowutil.set_size_virt(self.path, size, jfile)
        finally:
            try:
                self.sr._linstor.destroy_volume(jfile)
            except Exception:
                # We can ignore it, in any case this volume is not persistent.
                pass

    @override
    def _queryCowBlocks(self) -> bytes:
        return self.linstorcowutil.get_block_bitmap(self.uuid)

    def _inflateParentForCoalesce(self):
        if not VdiType.isCowImage(self.parent.vdi_type):
            return
        inc = self._calcExtraSpaceForCoalescing()
        if inc > 0:
            self.parent.inflate(self.parent.getDrbdSize() + inc)

    @override
    def _calcExtraSpaceForCoalescing(self) -> int:
        if not VdiType.isCowImage(self.parent.vdi_type):
            return 0
        size_coalesced = self.linstorcowutil.compute_volume_size(self._getCoalescedSizeData())
        Util.log("Coalesced size = %s" % Util.num2str(size_coalesced))
        return size_coalesced - self.parent.getDrbdSize()

    @override
    def _calcExtraSpaceForLeafCoalescing(self) -> int:
        assert self.getDrbdSize() > 0
        assert self.getSizePhys() > 0
        deflate_diff = self.getDrbdSize() - LinstorVolumeManager.round_up_volume_size(self.getSizePhys())
        assert deflate_diff >= 0
        return self._calcExtraSpaceForCoalescing() - deflate_diff

    @override
    def _calcExtraSpaceForSnapshotCoalescing(self) -> int:
        assert self.getSizePhys() > 0
        return self._calcExtraSpaceForCoalescing() + \
            LinstorVolumeManager.round_up_volume_size(self.getSizePhys())

################################################################################
#
# SR
#
class SR(object):
    class LogFilter:
        def __init__(self, sr):
            self.sr = sr
            self.stateLogged = False
            self.prevState = {}
            self.currState = {}

        def logState(self):
            changes = ""
            self.currState.clear()
            for vdi in self.sr.vdiTrees:
                self.currState[vdi.uuid] = self._getTreeStr(vdi)
                if not self.prevState.get(vdi.uuid) or \
                        self.prevState[vdi.uuid] != self.currState[vdi.uuid]:
                    changes += self.currState[vdi.uuid]

            for uuid in self.prevState:
                if not self.currState.get(uuid):
                    changes += "Tree %s gone\n" % uuid

            result = "SR %s (%d VDIs in %d COW trees): " % \
                    (self.sr, len(self.sr.vdis), len(self.sr.vdiTrees))

            if len(changes) > 0:
                if self.stateLogged:
                    result += "showing only COW trees that changed:"
                result += "\n%s" % changes
            else:
                result += "no changes"

            for line in result.split("\n"):
                Util.log("%s" % line)
            self.prevState.clear()
            for key, val in self.currState.items():
                self.prevState[key] = val
            self.stateLogged = True

        def logNewVDI(self, uuid):
            if self.stateLogged:
                Util.log("Found new VDI when scanning: %s" % uuid)

        def _getTreeStr(self, vdi, indent=8):
            treeStr = "%s%s\n" % (" " * indent, vdi)
            for child in vdi.children:
                treeStr += self._getTreeStr(child, indent + VDI.STR_TREE_INDENT)
            return treeStr

    TYPE_FILE = "file"
    TYPE_LVHD = "lvhd"
    TYPE_LINSTOR = "linstor"
    TYPES = [TYPE_LVHD, TYPE_FILE, TYPE_LINSTOR]

    LOCK_RETRY_INTERVAL = 3
    LOCK_RETRY_ATTEMPTS = 20
    LOCK_RETRY_ATTEMPTS_LOCK = 100

    SCAN_RETRY_ATTEMPTS = 3

    JRN_CLONE = "clone"  # journal entry type for the clone operation (from SM)
    TMP_RENAME_PREFIX = "OLD_"

    KEY_OFFLINE_COALESCE_NEEDED = "leaf_coalesce_need_offline"
    KEY_OFFLINE_COALESCE_OVERRIDE = "leaf_coalesce_offline_override"

    @staticmethod
    def getInstance(uuid, xapiSession, createLock=True, force=False):
        xapi = XAPI(xapiSession, uuid)
        type = normalizeType(xapi.srRecord["type"])
        if type == SR.TYPE_FILE:
            return FileSR(uuid, xapi, createLock, force)
        elif type == SR.TYPE_LVHD:
            return LVMSR(uuid, xapi, createLock, force)
        elif type == SR.TYPE_LINSTOR:
            return LinstorSR(uuid, xapi, createLock, force)
        raise util.SMException("SR type %s not recognized" % type)

    def __init__(self, uuid, xapi, createLock, force):
        self.logFilter = self.LogFilter(self)
        self.uuid = uuid
        self.path = ""
        self.name = ""
        self.vdis = {}
        self.vdiTrees = []
        self.journaler = None
        self.xapi = xapi
        self._locked = 0
        self._srLock = None
        if createLock:
            self._srLock = lock.Lock(lock.LOCK_TYPE_SR, self.uuid)
        else:
            Util.log("Requested no SR locking")
        self.name = self.xapi.srRecord["name_label"]
        self._failedCoalesceTargets = []

        if not self.xapi.isPluggedHere():
            if force:
                Util.log("SR %s not attached on this host, ignoring" % uuid)
            else:
                if not self.wait_for_plug():
                    raise util.SMException("SR %s not attached on this host" % uuid)

        if force:
            Util.log("Not checking if we are Master (SR %s)" % uuid)
        elif not self.xapi.isMaster():
            raise util.SMException("This host is NOT master, will not run")

        self.no_space_candidates = {}

    def msg_cleared(self, xapi_session, msg_ref):
        try:
            msg = xapi_session.xenapi.message.get_record(msg_ref)
        except XenAPI.Failure:
            return True

        return msg is None

    def check_no_space_candidates(self):
        with util.ApiSession("SM-GC-check_no_space") as xapi_session:
            msg_id = self.xapi.srRecord["sm_config"].get(VDI.DB_GC_NO_SPACE)
            if self.no_space_candidates:
                if msg_id is None or self.msg_cleared(xapi_session, msg_id):
                    util.SMlog("Could not coalesce due to a lack of space "
                               f"in SR {self.uuid}")
                    msg_body = ("Unable to perform data coalesce due to a lack "
                                f"of space in SR {self.uuid}")
                    msg_id = xapi_session.xenapi.message.create(
                        'SM_GC_NO_SPACE',
                        3,
                        "SR",
                        self.uuid,
                        msg_body)
                    xapi_session.xenapi.SR.remove_from_sm_config(
                        self.xapi.srRef, VDI.DB_GC_NO_SPACE)
                    xapi_session.xenapi.SR.add_to_sm_config(
                        self.xapi.srRef, VDI.DB_GC_NO_SPACE, msg_id)

                for candidate in self.no_space_candidates.values():
                    candidate.setConfig(VDI.DB_GC_NO_SPACE, msg_id)
            elif msg_id is not None:
                # Everything was coalescable, remove the message
                xapi_session.xenapi.SR.remove_from_sm_config(self.xapi.srRef, VDI.DB_GC_NO_SPACE)
                xapi_session.xenapi.message.destroy(msg_id)

    def clear_no_space_msg(self, vdi):
        msg_id = None
        try:
            msg_id = vdi.getConfig(VDI.DB_GC_NO_SPACE)
        except XenAPI.Failure:
            pass

        self.no_space_candidates.pop(vdi.uuid, None)
        if msg_id is not None:
            vdi.delConfig(VDI.DB_GC_NO_SPACE)


    def wait_for_plug(self):
        for _ in range(1, 10):
            time.sleep(2)
            if self.xapi.isPluggedHere():
                return True
        return False

    def gcEnabled(self, refresh=True):
        if refresh:
            self.xapi.srRecord = \
                    self.xapi.session.xenapi.SR.get_record(self.xapi._srRef)
        if self.xapi.srRecord["other_config"].get(VDI.DB_GC) == "false":
            Util.log("GC is disabled for this SR, abort")
            return False
        return True

    def scan(self, force=False) -> None:
        """Scan the SR and load VDI info for each VDI. If called repeatedly,
        update VDI objects if they already exist"""
        pass

    def scanLocked(self, force=False):
        self.lock()
        try:
            self.scan(force)
        finally:
            self.unlock()

    def getVDI(self, uuid):
        return self.vdis.get(uuid)

    def hasWork(self):
        if len(self.findGarbage()) > 0:
            return True
        if self.findCoalesceable():
            return True
        if self.findLeafCoalesceable():
            return True
        if self.needUpdateBlockInfo():
            return True
        return False

    def findCoalesceable(self):
        """Find a coalesceable VDI. Return a vdi that should be coalesced
        (choosing one among all coalesceable candidates according to some
        criteria) or None if there is no VDI that could be coalesced"""

        candidates = []

        srSwitch = self.xapi.srRecord["other_config"].get(VDI.DB_COALESCE)
        if srSwitch == "false":
            Util.log("Coalesce disabled for this SR")
            return candidates

        # finish any VDI for which a relink journal entry exists first
        journals = self.journaler.getAll(VDI.JRN_RELINK)
        for uuid in journals:
            vdi = self.getVDI(uuid)
            if vdi and vdi not in self._failedCoalesceTargets:
                return vdi

        for vdi in self.vdis.values():
            if vdi.isCoalesceable() and vdi not in self._failedCoalesceTargets:
                candidates.append(vdi)
                Util.log("%s is coalescable" % vdi.uuid)

        self.xapi.update_task_progress("coalescable", len(candidates))

        # pick one in the tallest tree
        treeHeight = dict()
        for c in candidates:
            height = c.getTreeRoot().getTreeHeight()
            if treeHeight.get(height):
                treeHeight[height].append(c)
            else:
                treeHeight[height] = [c]

        freeSpace = self.getFreeSpace()
        heights = list(treeHeight.keys())
        heights.sort(reverse=True)
        for h in heights:
            for c in treeHeight[h]:
                spaceNeeded = c._calcExtraSpaceForCoalescing()
                if spaceNeeded <= freeSpace:
                    Util.log("Coalesce candidate: %s (tree height %d)" % (c, h))
                    self.clear_no_space_msg(c)
                    return c
                else:
                    self.no_space_candidates[c.uuid] = c
                    Util.log("No space to coalesce %s (free space: %d)" % \
                            (c, freeSpace))
        return None

    def getSwitch(self, key):
        return self.xapi.srRecord["other_config"].get(key)

    def forbiddenBySwitch(self, switch, condition, fail_msg):
        srSwitch = self.getSwitch(switch)
        ret = False
        if srSwitch:
            ret = srSwitch == condition

        if ret:
            Util.log(fail_msg)

        return ret

    def leafCoalesceForbidden(self):
        return (self.forbiddenBySwitch(VDI.DB_COALESCE,
                                       "false",
                                       "Coalesce disabled for this SR") or
                self.forbiddenBySwitch(VDI.DB_LEAFCLSC,
                                       VDI.LEAFCLSC_DISABLED,
                                       "Leaf-coalesce disabled for this SR"))

    def findLeafCoalesceable(self):
        """Find leaf-coalesceable VDIs in each COW tree"""

        candidates = []
        if self.leafCoalesceForbidden():
            return candidates

        self.gatherLeafCoalesceable(candidates)

        self.xapi.update_task_progress("coalescable", len(candidates))

        freeSpace = self.getFreeSpace()
        for candidate in candidates:
            # check the space constraints to see if leaf-coalesce is actually
            # feasible for this candidate
            spaceNeeded = candidate._calcExtraSpaceForSnapshotCoalescing()
            spaceNeededLive = spaceNeeded
            if spaceNeeded > freeSpace:
                spaceNeededLive = candidate._calcExtraSpaceForLeafCoalescing()
                if candidate.canLiveCoalesce(self.getStorageSpeed()):
                    spaceNeeded = spaceNeededLive

            if spaceNeeded <= freeSpace:
                Util.log("Leaf-coalesce candidate: %s" % candidate)
                self.clear_no_space_msg(candidate)
                return candidate
            else:
                Util.log("No space to leaf-coalesce %s (free space: %d)" % \
                        (candidate, freeSpace))
                if spaceNeededLive <= freeSpace:
                    Util.log("...but enough space if skip snap-coalesce")
                    candidate.setConfig(VDI.DB_LEAFCLSC,
                                        VDI.LEAFCLSC_OFFLINE)
                self.no_space_candidates[candidate.uuid] = candidate

        return None

    def gatherLeafCoalesceable(self, candidates):
        for vdi in self.vdis.values():
            if not vdi.isLeafCoalesceable():
                continue
            if vdi in self._failedCoalesceTargets:
                continue
            if vdi.getConfig(vdi.DB_ONBOOT) == vdi.ONBOOT_RESET:
                Util.log("Skipping reset-on-boot %s" % vdi)
                continue
            if vdi.getConfig(vdi.DB_ALLOW_CACHING):
                Util.log("Skipping allow_caching=true %s" % vdi)
                continue
            if vdi.getConfig(vdi.DB_LEAFCLSC) == vdi.LEAFCLSC_DISABLED:
                Util.log("Leaf-coalesce disabled for %s" % vdi)
                continue
            if not (AUTO_ONLINE_LEAF_COALESCE_ENABLED or
                    vdi.getConfig(vdi.DB_LEAFCLSC) == vdi.LEAFCLSC_FORCE):
                continue
            candidates.append(vdi)

    def coalesce(self, vdi, dryRun=False):
        """Coalesce vdi onto parent"""
        Util.log("Coalescing %s -> %s" % (vdi, vdi.parent))
        if dryRun:
            return

        try:
            self._coalesce(vdi)
        except util.SMException as e:
            if isinstance(e, AbortException):
                self.cleanup()
                raise
            else:
                self._failedCoalesceTargets.append(vdi)
                Util.logException("coalesce")
                Util.log("Coalesce failed, skipping")
        self.cleanup()

    def coalesceLeaf(self, vdi, dryRun=False):
        """Leaf-coalesce vdi onto parent"""
        Util.log("Leaf-coalescing %s -> %s" % (vdi, vdi.parent))
        if dryRun:
            return

        try:
            uuid = vdi.uuid
            try:
                # "vdi" object will no longer be valid after this call
                if vdi.cowutil.isCoalesceableOnRemote():
                    Util.log("We will live coalesce leaf: {uuid}".format(uuid=vdi.uuid))
                    self._liveLeafCoalesce(vdi, coalesce_on_remote=True)
                else:
                    Util.log("We can't live coalesce leaf: {uuid}".format(uuid=vdi.uuid))
                    self._coalesceLeaf(vdi)
            finally:
                vdi = self.getVDI(uuid)
                if vdi:
                    vdi.delConfig(vdi.DB_LEAFCLSC)
        except AbortException:
            self.cleanup()
            raise
        except (util.SMException, XenAPI.Failure) as e:
            self._failedCoalesceTargets.append(vdi)
            Util.logException("leaf-coalesce")
            Util.log("Leaf-coalesce failed on %s, skipping" % vdi)
        self.cleanup()

    def garbageCollect(self, dryRun=False):
        vdiList = self.findGarbage()
        Util.log("Found %d VDIs for deletion:" % len(vdiList))
        for vdi in vdiList:
            Util.log("  %s" % vdi)
        if not dryRun:
            self.deleteVDIs(vdiList)
        self.cleanupJournals(dryRun)

    def findGarbage(self):
        vdiList = []
        for vdi in self.vdiTrees:
            vdiList.extend(vdi.getAllPrunable())
        return vdiList

    def deleteVDIs(self, vdiList) -> None:
        for vdi in vdiList:
            if IPCFlag(self.uuid).test(FLAG_TYPE_ABORT):
                raise AbortException("Aborting due to signal")
            Util.log("Deleting unlinked VDI %s" % vdi)
            self.deleteVDI(vdi)

    def deleteVDI(self, vdi) -> None:
        assert(len(vdi.children) == 0)
        del self.vdis[vdi.uuid]
        if vdi.parent:
            vdi.parent.children.remove(vdi)
        if vdi in self.vdiTrees:
            self.vdiTrees.remove(vdi)
        vdi.delete()

    def forgetVDI(self, vdiUuid) -> None:
        self.xapi.forgetVDI(self.uuid, vdiUuid)

    def pauseVDIs(self, vdiList) -> None:
        paused = []
        failed = False
        for vdi in vdiList:
            try:
                vdi.pause()
                paused.append(vdi)
            except:
                Util.logException("pauseVDIs")
                failed = True
                break

        if failed:
            self.unpauseVDIs(paused)
            raise util.SMException("Failed to pause VDIs")

    def unpauseVDIs(self, vdiList):
        failed = False
        for vdi in vdiList:
            try:
                vdi.unpause()
            except:
                Util.log("ERROR: Failed to unpause VDI %s" % vdi)
                failed = True
        if failed:
            raise util.SMException("Failed to unpause VDIs")

    def getFreeSpace(self) -> int:
        return 0

    def cleanup(self):
        Util.log("In cleanup")
        return

    @override
    def __str__(self) -> str:
        if self.name:
            ret = "%s ('%s')" % (self.uuid[0:4], self.name)
        else:
            ret = "%s" % self.uuid
        return ret

    def lock(self):
        """Acquire the SR lock. Nested acquire()'s are ok. Check for Abort
        signal to avoid deadlocking (trying to acquire the SR lock while the
        lock is held by a process that is trying to abort us)"""
        if not self._srLock:
            return

        if self._locked == 0:
            abortFlag = IPCFlag(self.uuid)
            for i in range(SR.LOCK_RETRY_ATTEMPTS_LOCK):
                if self._srLock.acquireNoblock():
                    self._locked += 1
                    return
                if abortFlag.test(FLAG_TYPE_ABORT):
                    raise AbortException("Abort requested")
                time.sleep(SR.LOCK_RETRY_INTERVAL)
            raise util.SMException("Unable to acquire the SR lock")

        self._locked += 1

    def unlock(self):
        if not self._srLock:
            return
        assert(self._locked > 0)
        self._locked -= 1
        if self._locked == 0:
            self._srLock.release()

    def needUpdateBlockInfo(self) -> bool:
        for vdi in self.vdis.values():
            if vdi.scanError or len(vdi.children) == 0:
                continue
            if not vdi.getConfig(vdi.DB_VDI_BLOCKS):
                return True
        return False

    def updateBlockInfo(self) -> None:
        for vdi in self.vdis.values():
            if vdi.scanError or len(vdi.children) == 0:
                continue
            if not vdi.getConfig(vdi.DB_VDI_BLOCKS):
                vdi.updateBlockInfo()

    def cleanupCoalesceJournals(self):
        """Remove stale coalesce VDI indicators"""
        entries = self.journaler.getAll(VDI.JRN_COALESCE)
        for uuid, jval in entries.items():
            self.journaler.remove(VDI.JRN_COALESCE, uuid)

    def cleanupJournals(self, dryRun=False):
        """delete journal entries for non-existing VDIs"""
        for t in [LVMVDI.JRN_ZERO, VDI.JRN_RELINK, SR.JRN_CLONE]:
            entries = self.journaler.getAll(t)
            for uuid, jval in entries.items():
                if self.getVDI(uuid):
                    continue
                if t == SR.JRN_CLONE:
                    baseUuid, clonUuid = jval.split("_")
                    if self.getVDI(baseUuid):
                        continue
                Util.log("  Deleting stale '%s' journal entry for %s "
                        "(%s)" % (t, uuid, jval))
                if not dryRun:
                    self.journaler.remove(t, uuid)

    def cleanupCache(self, maxAge=-1) -> int:
        return 0

    def hasLeavesAttachedOn(self, vdi: VDI):
        leaves = vdi.getAllLeaves()
        leaves_vdi = [leaf.uuid for leaf in leaves]
        return util.get_hosts_attached_on_with_vdi_uuid(self.xapi.session, leaves_vdi)

    def _gc_running_file(self, vdi: VDI):
        run_file = "gc_running_{}".format(vdi.uuid)
        return os.path.join(NON_PERSISTENT_DIR, str(self.uuid), run_file)

    def _create_running_file(self, vdi: VDI):
        with open(self._gc_running_file(vdi), "w") as f:
            f.write("1")

    def _delete_running_file(self, vdi: VDI):
        os.unlink(self._gc_running_file(vdi))

    def _coalesce(self, vdi: VDI):
        list_not_to_relink = None
        if self.journaler.get(vdi.JRN_RELINK, vdi.uuid):
            # this means we had done the actual coalescing already and just
            # need to finish relinking and/or refreshing the children
            Util.log("==> Coalesce apparently already done: skipping")

            # The parent volume must be active for the parent change to occur.
            # The parent volume may become inactive if the host is rebooted.
            vdi._ensureParentActiveForRelink()
        else:
            # JRN_COALESCE is used to check which VDI is being coalesced in
            # order to decide whether to abort the coalesce. We remove the
            # journal as soon as the COW coalesce step is done, because we
            # don't expect the rest of the process to take long

            if os.path.exists(self._gc_running_file(vdi)):
                util.SMlog("gc_running already exist for {}. Ignoring...".format(self.uuid))

            self._create_running_file(vdi)

            self.journaler.create(vdi.JRN_COALESCE, vdi.uuid, "1")
            host_refs = self.hasLeavesAttachedOn(vdi)
            # This check of multiple host_refs was done earlier in `isCoalesceable` but
            # we recheck here to be sure another leaf wasn't activated in the meantime.
            if vdi.cowutil.isCoalesceableOnRemote() and len(host_refs) > 1:
                Util.log("Not coalesceable, chain activated more than once")
                Util.log(f"VDI '{vdi.uuid}' has leaves attached on {host_refs}")
                raise Exception("Not coalesceable, chain activated more than once")

            try:
                if host_refs and vdi.cowutil.isCoalesceableOnRemote():
                    #Leaf opened on another host, we need to call online coalesce
                    Util.log("Remote coalesce for {}".format(vdi.path))

                    leaf_for_coalesce_uuid, host_ref = next(iter(host_refs.items())) # First host_ref since we should only have one
                    leaf_for_coalesce = [leaf for leaf in vdi.getAllLeaves() if leaf.uuid == leaf_for_coalesce_uuid][0]

                    vdi._doCoalesceOnHost(host_ref, leaf_for_coalesce)
                    # If we use a host OpaqueRef to do a online coalesce, this vdi will not need to be relinked since it was done by tapdisk
                    # If we coalesce up the chain, we shouldn't need to do the relink at all, we only need to do the relink on the children if their direct parent was the one we were coalescing
                    for child in vdi.children:
                        real_parent_uuid = child.extractUuid(child.getParent())
                        if real_parent_uuid == vdi.parent.uuid:
                            child._update_vhd_parent(real_parent_uuid) # We update the sm-config:vhd-parent value for this VDI since it has already been relinked
                            list_not_to_relink = [leaf.uuid for leaf in child.getAllLeaves()]
                else:
                    Util.log("Offline coalesce for {}".format(vdi.path))
                    vdi._doCoalesce()
            except Exception as e:
                Util.log("EXCEPTION while coalescing: {}".format(e))
                self._delete_running_file(vdi)
                raise

            self.journaler.remove(vdi.JRN_COALESCE, vdi.uuid)
            self._delete_running_file(vdi)

            util.fistpoint.activate("LVHDRT_before_create_relink_journal", self.uuid)

            # we now need to relink the children: lock the SR to prevent ops
            # like SM.clone from manipulating the VDIs we'll be relinking and
            # rescan the SR first in case the children changed since the last
            # scan
            self.journaler.create(vdi.JRN_RELINK, vdi.uuid, "1")

        self.lock()
        try:
            vdi.parent._tagChildrenForRelink(list_not_to_relink)
            self.scan()
            vdi._relinkSkip()
        finally:
            self.unlock()
            # Reload the children to leave things consistent
            vdi.parent._reloadChildren(vdi)
        self.journaler.remove(vdi.JRN_RELINK, vdi.uuid)

        self.deleteVDI(vdi)

    class CoalesceTracker:
        GRACE_ITERATIONS = 2
        MAX_ITERATIONS_NO_PROGRESS = 3
        MAX_ITERATIONS = 20
        MAX_INCREASE_FROM_MINIMUM = 1.2
        HISTORY_STRING = "Iteration: {its} -- Initial size {initSize}" \
                         " --> Final size {finSize}"

        def __init__(self, sr):
            self.itsNoProgress = 0
            self.its = 0
            self.minSize = float("inf")
            self._history = []
            self.reason = ""
            self.startSize = None
            self.finishSize = None
            self.sr = sr
            self.grace_remaining = self.GRACE_ITERATIONS

        @property
        def history(self):
            return [x['msg'] for x in self._history]

        def moving_average(self):
            """
            Calculate a three point moving average
            """
            assert len(self._history) >= 3

            mv_average = sum([x['finalsize'] for x in self._history]) / len(self._history)
            util.SMlog(f'Calculated moving average as {mv_average}')
            return mv_average

        def abortCoalesce(self, prevSize, curSize):
            self.its += 1
            self._history.append(
                {
                    'finalsize': curSize,
                    'msg': self.HISTORY_STRING.format(its=self.its,
                                                      initSize=prevSize,
                                                      finSize=curSize)
                }
            )

            self.finishSize = curSize

            if self.startSize is None:
                self.startSize = prevSize

            if curSize < self.minSize:
                self.minSize = curSize

            if prevSize < self.minSize:
                self.minSize = prevSize

            if self.its < 4:
                # Perform at least three iterations
                return False

            if prevSize >= curSize or curSize < self.moving_average():
                # We made progress
                return False
            else:
                self.itsNoProgress += 1
                Util.log("No progress, attempt:"
                         " {attempt}".format(attempt=self.itsNoProgress))
                util.fistpoint.activate("cleanup_tracker_no_progress", self.sr.uuid)

            if self.its > self.MAX_ITERATIONS:
                max = self.MAX_ITERATIONS
                self.reason = \
                    "Max iterations ({max}) exceeded".format(max=max)
                return True

            if self.itsNoProgress > self.MAX_ITERATIONS_NO_PROGRESS:
                max = self.MAX_ITERATIONS_NO_PROGRESS
                self.reason = \
                    "No progress made for {max} iterations".format(max=max)
                return True

            maxSizeFromMin = self.MAX_INCREASE_FROM_MINIMUM * self.minSize
            if curSize > maxSizeFromMin:
                self.grace_remaining -= 1
                if self.grace_remaining == 0:
                    self.reason = "Unexpected bump in size," \
                        " compared to minimum achieved"

                    return True

            return False

        def printSizes(self):
            Util.log("Starting size was         {size}"
                     .format(size=self.startSize))
            Util.log("Final size was            {size}"
                     .format(size=self.finishSize))
            Util.log("Minimum size achieved was {size}"
                     .format(size=self.minSize))

        def printReasoning(self):
            Util.log("Aborted coalesce")
            for hist in self.history:
                Util.log(hist)
            Util.log(self.reason)
            self.printSizes()

        def printSummary(self):
            if self.its == 0:
                return

            if self.reason:
                Util.log("Aborted coalesce")
                Util.log(self.reason)
            else:
                Util.log("Coalesce summary")

            Util.log(f"Performed {self.its} iterations")
            self.printSizes()


    def _coalesceLeaf(self, vdi):
        """Leaf-coalesce VDI vdi. Return true if we succeed, false if we cannot
        complete due to external changes, namely vdi_delete and vdi_snapshot
        that alter leaf-coalescibility of vdi"""
        tracker = self.CoalesceTracker(self)
        while not vdi.canLiveCoalesce(self.getStorageSpeed()):
            prevSizePhys = vdi.getSizePhys()
            if not self._snapshotCoalesce(vdi):
                return False
            if tracker.abortCoalesce(prevSizePhys, vdi.getSizePhys()):
                tracker.printReasoning()
                raise util.SMException("VDI {uuid} could not be coalesced"
                                       .format(uuid=vdi.uuid))
        tracker.printSummary()
        return self._liveLeafCoalesce(vdi)

    def calcStorageSpeed(self, startTime, endTime, coalescedSize):
        speed = None
        total_time = endTime - startTime
        if total_time > 0:
            speed = float(coalescedSize) / float(total_time)
        return speed

    def writeSpeedToFile(self, speed):
        content = []
        speedFile = None
        path = SPEED_LOG_ROOT.format(uuid=self.uuid)
        self.lock()
        try:
            Util.log("Writing to file: {myfile}".format(myfile=path))
            lines = ""
            if not os.path.isfile(path):
                lines = str(speed) + "\n"
            else:
                speedFile = open(path, "r+")
                content = speedFile.readlines()
                content.append(str(speed) + "\n")
                if len(content) > N_RUNNING_AVERAGE:
                    del content[0]
                lines = "".join(content)

            util.atomicFileWrite(path, VAR_RUN, lines)
        finally:
            if speedFile is not None:
                speedFile.close()
            Util.log("Closing file: {myfile}".format(myfile=path))
            self.unlock()

    def recordStorageSpeed(self, startTime, endTime, coalescedSize):
        speed = self.calcStorageSpeed(startTime, endTime, coalescedSize)
        if speed is None:
            return

        self.writeSpeedToFile(speed)

    def getStorageSpeed(self):
        speedFile = None
        path = SPEED_LOG_ROOT.format(uuid=self.uuid)
        self.lock()
        try:
            speed = None
            if os.path.isfile(path):
                speedFile = open(path)
                content = speedFile.readlines()
                try:
                    content = [float(i) for i in content]
                except ValueError:
                    Util.log("Something bad in the speed log:{log}".
                             format(log=speedFile.readlines()))
                    return speed

                if len(content):
                    speed = sum(content) / float(len(content))
                    if speed <= 0:
                        # Defensive, should be impossible.
                        Util.log("Bad speed: {speed} calculated for SR: {uuid}".
                             format(speed=speed, uuid=self.uuid))
                        speed = None
                else:
                    Util.log("Speed file empty for SR: {uuid}".
                             format(uuid=self.uuid))
            else:
                Util.log("Speed log missing for SR: {uuid}".
                         format(uuid=self.uuid))
            return speed
        finally:
            if not (speedFile is None):
                speedFile.close()
            self.unlock()

    def _snapshotCoalesce(self, vdi):
        # Note that because we are not holding any locks here, concurrent SM
        # operations may change this tree under our feet. In particular, vdi
        # can be deleted, or it can be snapshotted.
        assert(AUTO_ONLINE_LEAF_COALESCE_ENABLED)
        Util.log("Single-snapshotting %s" % vdi)
        util.fistpoint.activate("LVHDRT_coaleaf_delay_1", self.uuid)
        try:
            ret = self.xapi.singleSnapshotVDI(vdi)
            Util.log("Single-snapshot returned: %s" % ret)
        except XenAPI.Failure as e:
            if util.isInvalidVDI(e):
                Util.log("The VDI appears to have been concurrently deleted")
                return False
            raise
        self.scanLocked()
        tempSnap = vdi.parent
        if not tempSnap.isCoalesceable():
            Util.log("The VDI appears to have been concurrently snapshotted")
            return False
        Util.log("Coalescing parent %s" % tempSnap)
        util.fistpoint.activate("LVHDRT_coaleaf_delay_2", self.uuid)
        sizePhys = vdi.getSizePhys()
        self._coalesce(tempSnap)
        if not vdi.isLeafCoalesceable():
            Util.log("The VDI tree appears to have been altered since")
            return False
        return True

    def _liveLeafCoalesce(self, vdi: VDI, coalesce_on_remote: bool = False) -> bool:
        util.fistpoint.activate("LVHDRT_coaleaf_delay_3", self.uuid)
        need_refresh = False
        self.lock()
        try:
            self.scan()
            if not self.getVDI(vdi.uuid):
                Util.log("The VDI appears to have been deleted meanwhile")
                return False
            if not vdi.isLeafCoalesceable():
                Util.log("The VDI is no longer leaf-coalesceable")
                return False

            uuid = vdi.uuid
            if not coalesce_on_remote:
                vdi.pause(failfast=True)
            try:
                try:
                    self._create_running_file(vdi)
                    # "vdi" object will no longer be valid after this call
                    self._doCoalesceLeaf(vdi, coalesce_on_remote)
                    need_refresh = True
                except:
                    Util.logException("_doCoalesceLeaf")
                    self._handleInterruptedCoalesceLeaf()
                    raise
            finally:
                vdi = self.getVDI(uuid)
                if vdi:
                    vdi.ensureUnpaused()
                    if need_refresh:
                        vdi.refresh()
                self._delete_running_file(vdi)
                vdiOld = self.getVDI(self.TMP_RENAME_PREFIX + uuid)
                if vdiOld:
                    util.fistpoint.activate("LVHDRT_coaleaf_before_delete", self.uuid)
                    self.deleteVDI(vdiOld)
                    util.fistpoint.activate("LVHDRT_coaleaf_after_delete", self.uuid)
        finally:
            self.cleanup()
            self.unlock()
            self.logFilter.logState()
        return True

    def _doCoalesceLeaf(self, vdi: VDI, coalesce_on_remote: bool):
        """Actual coalescing of a leaf VDI onto parent. Must be called in an
        offline/atomic context"""
        self.journaler.create(VDI.JRN_LEAF, vdi.uuid, vdi.parent.uuid)
        self._prepareCoalesceLeaf(vdi)
        vdi.parent._setHidden(False)
        vdi.parent._increaseSizeVirt(vdi.sizeVirt, False)
        host_refs = self.hasLeavesAttachedOn(vdi) if coalesce_on_remote else None
        if host_refs:
            util.fistpoint.activate("LVHDRT_coaleaf_before_coalesce", self.uuid)
            _, host_ref = next(iter(host_refs.items()))
            vdi._coalesceCowImageOnHost(host_ref, vdi) # vdi is the leaf for the online coalesce
            util.fistpoint.activate("LVHDRT_coaleaf_after_coalesce", self.uuid)
        else:
            vdi.validate(True)
            vdi.parent.validate(True)
            util.fistpoint.activate("LVHDRT_coaleaf_before_coalesce", self.uuid)
            timeout = vdi.LIVE_LEAF_COALESCE_TIMEOUT
            if vdi.getConfig(vdi.DB_LEAFCLSC) == vdi.LEAFCLSC_FORCE:
                Util.log("Leaf-coalesce forced, will not use timeout")
                timeout = 0
            vdi._coalesceCowImage(timeout)
            util.fistpoint.activate("LVHDRT_coaleaf_after_coalesce", self.uuid)
            vdi.parent.validate(True)
            #vdi._verifyContents(timeout / 2)

        # rename
        vdiUuid = vdi.uuid
        oldName = vdi.fileName
        origParentUuid = vdi.parent.uuid
        vdi.rename(self.TMP_RENAME_PREFIX + vdiUuid)
        util.fistpoint.activate("LVHDRT_coaleaf_one_renamed", self.uuid)
        vdi.parent.rename(vdiUuid)
        util.fistpoint.activate("LVHDRT_coaleaf_both_renamed", self.uuid)
        self._updateSlavesOnRename(vdi.parent, oldName, origParentUuid)

        # Note that "vdi.parent" is now the single remaining leaf and "vdi" is
        # garbage

        # update the VDI record
        if vdi.parent.vdi_type == VdiType.RAW:
            vdi.parent.setConfig(VDI.DB_VDI_TYPE, VdiType.RAW)
        vdi.parent.delConfig(VDI.DB_VDI_BLOCKS)
        util.fistpoint.activate("LVHDRT_coaleaf_after_vdirec", self.uuid)

        self._updateNode(vdi)

        # delete the obsolete leaf & inflate the parent (in that order, to
        # minimize free space requirements)
        parent = vdi.parent
        vdi._setHidden(True)
        vdi.parent.children = []
        vdi.parent = None

        if parent.parent is None:
            parent.delConfig(VDI.DB_VDI_PARENT)

        extraSpace = self._calcExtraSpaceNeeded(vdi, parent)
        freeSpace = self.getFreeSpace()
        if freeSpace < extraSpace:
            # don't delete unless we need the space: deletion is time-consuming
            # because it requires contacting the slaves, and we're paused here
            util.fistpoint.activate("LVHDRT_coaleaf_before_delete", self.uuid)
            self.deleteVDI(vdi)
            util.fistpoint.activate("LVHDRT_coaleaf_after_delete", self.uuid)

        util.fistpoint.activate("LVHDRT_coaleaf_before_remove_j", self.uuid)
        self.journaler.remove(VDI.JRN_LEAF, vdiUuid)

        self.forgetVDI(origParentUuid)
        self._finishCoalesceLeaf(parent)
        self._updateSlavesOnResize(parent)

    def _calcExtraSpaceNeeded(self, child, parent) -> int:
        assert(VdiType.isCowImage(parent.vdi_type))
        extra = child.getSizePhys() - parent.getSizePhys()
        if extra < 0:
            extra = 0
        return extra

    def _prepareCoalesceLeaf(self, vdi) -> None:
        pass

    def _updateNode(self, vdi) -> None:
        pass

    def _finishCoalesceLeaf(self, parent) -> None:
        pass

    def _updateSlavesOnUndoLeafCoalesce(self, parent, child) -> None:
        pass

    def _updateSlavesOnRename(self, vdi, oldName, origParentUuid) -> None:
        pass

    def _updateSlavesOnResize(self, vdi) -> None:
        pass

    def _removeStaleVDIs(self, uuidsPresent) -> None:
        for uuid in list(self.vdis.keys()):
            if not uuid in uuidsPresent:
                Util.log("VDI %s disappeared since last scan" % \
                        self.vdis[uuid])
                del self.vdis[uuid]

    def _handleInterruptedCoalesceLeaf(self) -> None:
        """An interrupted leaf-coalesce operation may leave the COW tree in an
        inconsistent state. If the old-leaf VDI is still present, we revert the
        operation (in case the original error is persistent); otherwise we must
        finish the operation"""
        pass

    def _buildTree(self, force):
        self.vdiTrees = []
        for vdi in self.vdis.values():
            if vdi.parentUuid:
                parent = self.getVDI(vdi.parentUuid)
                if not parent:
                    if vdi.uuid.startswith(self.TMP_RENAME_PREFIX):
                        self.vdiTrees.append(vdi)
                        continue
                    if force:
                        Util.log("ERROR: Parent VDI %s not found! (for %s)" % \
                                (vdi.parentUuid, vdi.uuid))
                        self.vdiTrees.append(vdi)
                        continue
                    else:
                        raise util.SMException("Parent VDI %s of %s not " \
                                "found" % (vdi.parentUuid, vdi.uuid))
                vdi.parent = parent
                parent.children.append(vdi)
            else:
                self.vdiTrees.append(vdi)


class FileSR(SR):
    TYPE = SR.TYPE_FILE
    CACHE_FILE_EXT = ".vhdcache"
    # cache cleanup actions
    CACHE_ACTION_KEEP = 0
    CACHE_ACTION_REMOVE = 1
    CACHE_ACTION_REMOVE_IF_INACTIVE = 2

    def __init__(self, uuid, xapi, createLock, force):
        SR.__init__(self, uuid, xapi, createLock, force)
        self.path = "/var/run/sr-mount/%s" % self.uuid
        self.journaler = fjournaler.Journaler(self.path)

    @override
    def scan(self, force=False) -> None:
        if not util.pathexists(self.path):
            raise util.SMException("directory %s not found!" % self.uuid)

        uuidsPresent: List[str] = []

        for vdi_type in VDI_COW_TYPES:
            scan_result = self._scan(vdi_type, force)
            for uuid, image_info in scan_result.items():
                vdi = self.getVDI(uuid)
                if not vdi:
                    self.logFilter.logNewVDI(uuid)
                    vdi = FileVDI(self, uuid, vdi_type)
                    self.vdis[uuid] = vdi
                vdi.load(image_info)
            uuidsPresent.extend(scan_result.keys())

        rawList = [x for x in os.listdir(self.path) if x.endswith(VdiTypeExtension.RAW)]
        for rawName in rawList:
            uuid = FileVDI.extractUuid(rawName)
            uuidsPresent.append(uuid)
            vdi = self.getVDI(uuid)
            if not vdi:
                self.logFilter.logNewVDI(uuid)
                vdi = FileVDI(self, uuid, VdiType.RAW)
                self.vdis[uuid] = vdi
        self._removeStaleVDIs(uuidsPresent)
        self._buildTree(force)
        self.logFilter.logState()
        self._handleInterruptedCoalesceLeaf()

    @override
    def getFreeSpace(self) -> int:
        return util.get_fs_size(self.path) - util.get_fs_utilisation(self.path)

    @override
    def deleteVDIs(self, vdiList) -> None:
        rootDeleted = False
        for vdi in vdiList:
            if not vdi.parent:
                rootDeleted = True
                break
        SR.deleteVDIs(self, vdiList)
        if self.xapi.srRecord["type"] == "nfs" and rootDeleted:
            self.xapi.markCacheSRsDirty()

    @override
    def cleanupCache(self, maxAge=-1) -> int:
        """Clean up IntelliCache cache files. Caches for leaf nodes are
        removed when the leaf node no longer exists or its allow-caching
        attribute is not set. Caches for parent nodes are removed when the
        parent node no longer exists or it hasn't been used in more than
        <maxAge> hours.
        Return number of caches removed.
        """
        numRemoved = 0
        cacheFiles = [x for x in os.listdir(self.path) if self._isCacheFileName(x)]
        Util.log("Found %d cache files" % len(cacheFiles))
        cutoff = datetime.datetime.now() - datetime.timedelta(hours=maxAge)
        for cacheFile in cacheFiles:
            uuid = cacheFile[:-len(self.CACHE_FILE_EXT)]
            action = self.CACHE_ACTION_KEEP
            rec = self.xapi.getRecordVDI(uuid)
            if not rec:
                Util.log("Cache %s: VDI doesn't exist" % uuid)
                action = self.CACHE_ACTION_REMOVE
            elif rec["managed"] and not rec["allow_caching"]:
                Util.log("Cache %s: caching disabled" % uuid)
                action = self.CACHE_ACTION_REMOVE
            elif not rec["managed"] and maxAge >= 0:
                lastAccess = datetime.datetime.fromtimestamp( \
                        os.path.getatime(os.path.join(self.path, cacheFile)))
                if lastAccess < cutoff:
                    Util.log("Cache %s: older than %d hrs" % (uuid, maxAge))
                    action = self.CACHE_ACTION_REMOVE_IF_INACTIVE

            if action == self.CACHE_ACTION_KEEP:
                Util.log("Keeping cache %s" % uuid)
                continue

            lockId = uuid
            parentUuid = None
            if rec and rec["managed"]:
                parentUuid = rec["sm_config"].get("vhd-parent")
            if parentUuid:
                lockId = parentUuid

            cacheLock = lock.Lock(blktap2.VDI.LOCK_CACHE_SETUP, lockId)
            cacheLock.acquire()
            try:
                if self._cleanupCache(uuid, action):
                    numRemoved += 1
            finally:
                cacheLock.release()
        return numRemoved

    def _cleanupCache(self, uuid, action):
        assert(action != self.CACHE_ACTION_KEEP)
        rec = self.xapi.getRecordVDI(uuid)
        if rec and rec["allow_caching"]:
            Util.log("Cache %s appears to have become valid" % uuid)
            return False

        fullPath = os.path.join(self.path, uuid + self.CACHE_FILE_EXT)
        tapdisk = blktap2.Tapdisk.find_by_path(fullPath)
        if tapdisk:
            if action == self.CACHE_ACTION_REMOVE_IF_INACTIVE:
                Util.log("Cache %s still in use" % uuid)
                return False
            Util.log("Shutting down tapdisk for %s" % fullPath)
            tapdisk.shutdown()

        Util.log("Deleting file %s" % fullPath)
        os.unlink(fullPath)
        return True

    def _isCacheFileName(self, name):
        return (len(name) == Util.UUID_LEN + len(self.CACHE_FILE_EXT)) and \
                name.endswith(self.CACHE_FILE_EXT)

    def _scan(self, vdi_type, force):
        for i in range(SR.SCAN_RETRY_ATTEMPTS):
            error = False
            pattern = os.path.join(self.path, "*%s" % VDI_TYPE_TO_EXTENSION[vdi_type])
            scan_result = getCowUtil(vdi_type).getAllInfoFromVG(pattern, FileVDI.extractUuid)
            for uuid, vdiInfo in scan_result.items():
                if vdiInfo.error:
                    error = True
                    break
            if not error:
                return scan_result
            Util.log("Scan error on attempt %d" % i)
        if force:
            return scan_result
        raise util.SMException("Scan error")

    @override
    def deleteVDI(self, vdi) -> None:
        self._checkSlaves(vdi)
        SR.deleteVDI(self, vdi)

    def _checkSlaves(self, vdi):
        onlineHosts = self.xapi.getOnlineHosts()
        abortFlag = IPCFlag(self.uuid)
        for pbdRecord in self.xapi.getAttachedPBDs():
            hostRef = pbdRecord["host"]
            if hostRef == self.xapi._hostRef:
                continue
            if abortFlag.test(FLAG_TYPE_ABORT):
                raise AbortException("Aborting due to signal")
            try:
                self._checkSlave(hostRef, vdi)
            except XenAPI.Failure:
                if hostRef in onlineHosts:
                    raise

    def _checkSlave(self, hostRef, vdi):
        call = (hostRef, "nfs-on-slave", "check", {'path': vdi.path})
        Util.log("Checking with slave: %s" % repr(call))
        _host = self.xapi.session.xenapi.host
        text = _host.call_plugin( * call)

    @override
    def _handleInterruptedCoalesceLeaf(self) -> None:
        entries = self.journaler.getAll(VDI.JRN_LEAF)
        for uuid, parentUuid in entries.items():
            fileList = os.listdir(self.path)
            childName = uuid + VdiTypeExtension.VHD
            tmpChildName = self.TMP_RENAME_PREFIX + uuid + VdiTypeExtension.VHD
            parentName1 = parentUuid + VdiTypeExtension.VHD
            parentName2 = parentUuid + VdiTypeExtension.RAW
            parentPresent = (parentName1 in fileList or parentName2 in fileList)
            if parentPresent or tmpChildName in fileList:
                self._undoInterruptedCoalesceLeaf(uuid, parentUuid)
            else:
                self._finishInterruptedCoalesceLeaf(uuid, parentUuid)
            self.journaler.remove(VDI.JRN_LEAF, uuid)
            vdi = self.getVDI(uuid)
            if vdi:
                vdi.ensureUnpaused()

    def _undoInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log("*** UNDO LEAF-COALESCE")
        parent = self.getVDI(parentUuid)
        if not parent:
            parent = self.getVDI(childUuid)
            if not parent:
                raise util.SMException("Neither %s nor %s found" % \
                        (parentUuid, childUuid))
            Util.log("Renaming parent back: %s -> %s" % (childUuid, parentUuid))
            parent.rename(parentUuid)
        util.fistpoint.activate("LVHDRT_coaleaf_undo_after_rename", self.uuid)

        child = self.getVDI(childUuid)
        if not child:
            child = self.getVDI(self.TMP_RENAME_PREFIX + childUuid)
            if not child:
                raise util.SMException("Neither %s nor %s found" % \
                        (childUuid, self.TMP_RENAME_PREFIX + childUuid))
            Util.log("Renaming child back to %s" % childUuid)
            child.rename(childUuid)
            Util.log("Updating the VDI record")
            child.setConfig(VDI.DB_VDI_PARENT, parentUuid)
            child.setConfig(VDI.DB_VDI_TYPE, child.vdi_type)
            util.fistpoint.activate("LVHDRT_coaleaf_undo_after_rename2", self.uuid)

        if child.isHidden():
            child._setHidden(False)
        if not parent.isHidden():
            parent._setHidden(True)
        self._updateSlavesOnUndoLeafCoalesce(parent, child)
        util.fistpoint.activate("LVHDRT_coaleaf_undo_end", self.uuid)
        Util.log("*** leaf-coalesce undo successful")
        if util.fistpoint.is_active("LVHDRT_coaleaf_stop_after_recovery"):
            child.setConfig(VDI.DB_LEAFCLSC, VDI.LEAFCLSC_DISABLED)

    def _finishInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log("*** FINISH LEAF-COALESCE")
        vdi = self.getVDI(childUuid)
        if not vdi:
            Util.log(f"_finishInterruptedCoalesceLeaf, vdi {childUuid} not found, aborting")
            raise util.SMException("VDI %s not found" % childUuid)
        try:
            self.forgetVDI(parentUuid)
        except XenAPI.Failure:
            Util.logException('_finishInterruptedCoalesceLeaf')
            pass
        self._updateSlavesOnResize(vdi)
        util.fistpoint.activate("LVHDRT_coaleaf_finish_end", self.uuid)
        Util.log("*** finished leaf-coalesce successfully")


class LVMSR(SR):
    TYPE = SR.TYPE_LVHD
    SUBTYPES = ["lvhdoiscsi", "lvhdohba"]

    def __init__(self, uuid, xapi, createLock, force):
        SR.__init__(self, uuid, xapi, createLock, force)
        self.vgName = "%s%s" % (VG_PREFIX, self.uuid)
        self.path = os.path.join(VG_LOCATION, self.vgName)

        sr_ref = self.xapi.session.xenapi.SR.get_by_uuid(self.uuid)
        other_conf = self.xapi.session.xenapi.SR.get_other_config(sr_ref)
        lvm_conf = other_conf.get('lvm-conf') if other_conf else None
        self.lvmCache = lvmcache.LVMCache(self.vgName, lvm_conf)

        self.lvActivator = LVActivator(self.uuid, self.lvmCache)
        self.journaler = journaler.Journaler(self.lvmCache)

    @override
    def deleteVDI(self, vdi) -> None:
        if self.lvActivator.get(vdi.uuid, False):
            self.lvActivator.deactivate(vdi.uuid, False)
        self._checkSlaves(vdi)
        SR.deleteVDI(self, vdi)

    @override
    def forgetVDI(self, vdiUuid) -> None:
        SR.forgetVDI(self, vdiUuid)
        mdpath = os.path.join(self.path, lvutil.MDVOLUME_NAME)
        LVMMetadataHandler(mdpath).deleteVdiFromMetadata(vdiUuid)

    @override
    def getFreeSpace(self) -> int:
        stats = lvutil._getVGstats(self.vgName)
        return stats['physical_size'] - stats['physical_utilisation']

    @override
    def cleanup(self):
        if not self.lvActivator.deactivateAll():
            Util.log("ERROR deactivating LVs while cleaning up")

    @override
    def needUpdateBlockInfo(self) -> bool:
        for vdi in self.vdis.values():
            if vdi.scanError or not VdiType.isCowImage(vdi.vdi_type) or len(vdi.children) == 0:
                continue
            if not vdi.getConfig(vdi.DB_VDI_BLOCKS):
                return True
        return False

    @override
    def updateBlockInfo(self) -> None:
        numUpdated = 0
        for vdi in self.vdis.values():
            if vdi.scanError or not VdiType.isCowImage(vdi.vdi_type) or len(vdi.children) == 0:
                continue
            if not vdi.getConfig(vdi.DB_VDI_BLOCKS):
                vdi.updateBlockInfo()
                numUpdated += 1
        if numUpdated:
            # deactivate the LVs back sooner rather than later. If we don't
            # now, by the time this thread gets to deactivations, another one
            # might have leaf-coalesced a node and deleted it, making the child
            # inherit the refcount value and preventing the correct decrement
            self.cleanup()

    @override
    def scan(self, force=False) -> None:
        vdis = self._scan(force)
        for uuid, vdiInfo in vdis.items():
            vdi = self.getVDI(uuid)
            if not vdi:
                self.logFilter.logNewVDI(uuid)
                vdi = LVMVDI(self, uuid, vdiInfo.vdiType)
                self.vdis[uuid] = vdi
            vdi.load(vdiInfo)
        self._removeStaleVDIs(vdis.keys())
        self._buildTree(force)
        self.logFilter.logState()
        self._handleInterruptedCoalesceLeaf()

    def _scan(self, force):
        for i in range(SR.SCAN_RETRY_ATTEMPTS):
            error = False
            self.lvmCache.refresh()
            vdis = LvmCowUtil.getVDIInfo(self.lvmCache)
            for uuid, vdiInfo in vdis.items():
                if vdiInfo.scanError:
                    error = True
                    break
            if not error:
                return vdis
            Util.log("Scan error, retrying (%d)" % i)
        if force:
            return vdis
        raise util.SMException("Scan error")

    @override
    def _removeStaleVDIs(self, uuidsPresent) -> None:
        for uuid in list(self.vdis.keys()):
            if not uuid in uuidsPresent:
                Util.log("VDI %s disappeared since last scan" % \
                        self.vdis[uuid])
                del self.vdis[uuid]
                if self.lvActivator.get(uuid, False):
                    self.lvActivator.remove(uuid, False)

    @override
    def _liveLeafCoalesce(self, vdi: VDI, coalesce_on_remote: bool = False) -> bool:
        """If the parent is raw and the child was resized (virt. size), then
        we'll need to resize the parent, which can take a while due to zeroing
        out of the extended portion of the LV. Do it before pausing the child
        to avoid a protracted downtime"""
        if not VdiType.isCowImage(vdi.parent.vdi_type) and vdi.sizeVirt > vdi.parent.sizeVirt:
            self.lvmCache.setReadonly(vdi.parent.fileName, False)
            vdi.parent._increaseSizeVirt(vdi.sizeVirt)

        return SR._liveLeafCoalesce(self, vdi, coalesce_on_remote)

    @override
    def _prepareCoalesceLeaf(self, vdi) -> None:
        vdi._activateChain()
        self.lvmCache.setReadonly(vdi.parent.fileName, False)
        vdi.deflate()
        vdi.inflateParentForCoalesce()

    @override
    def _updateNode(self, vdi) -> None:
        # fix the refcounts: the remaining node should inherit the binary
        # refcount from the leaf (because if it was online, it should remain
        # refcounted as such), but the normal refcount from the parent (because
        # this node is really the parent node) - minus 1 if it is online (since
        # non-leaf nodes increment their normal counts when they are online and
        # we are now a leaf, storing that 1 in the binary refcount).
        ns = NS_PREFIX_LVM + self.uuid
        cCnt, cBcnt = RefCounter.check(vdi.uuid, ns)
        pCnt, pBcnt = RefCounter.check(vdi.parent.uuid, ns)
        pCnt = pCnt - cBcnt
        assert(pCnt >= 0)
        RefCounter.set(vdi.parent.uuid, pCnt, cBcnt, ns)

    @override
    def _finishCoalesceLeaf(self, parent) -> None:
        if not parent.isSnapshot() or parent.isAttachedRW():
            parent.inflateFully()
        else:
            parent.deflate()

    @override
    def _calcExtraSpaceNeeded(self, child, parent) -> int:
        return parent.lvmcowutil.calcVolumeSize(parent.sizeVirt) - parent.sizeLV

    @override
    def _handleInterruptedCoalesceLeaf(self) -> None:
        entries = self.journaler.getAll(VDI.JRN_LEAF)
        for uuid, parentUuid in entries.items():
            undo = False
            for prefix in LV_PREFIX.values():
                parentLV = prefix + parentUuid
                undo = self.lvmCache.checkLV(parentLV)
                if undo:
                    break

            if not undo:
                for prefix in LV_PREFIX.values():
                    tmpChildLV = prefix + uuid
                    undo = self.lvmCache.checkLV(tmpChildLV)
                    if undo:
                        break

            if undo:
                self._undoInterruptedCoalesceLeaf(uuid, parentUuid)
            else:
                self._finishInterruptedCoalesceLeaf(uuid, parentUuid)
            self.journaler.remove(VDI.JRN_LEAF, uuid)
            vdi = self.getVDI(uuid)
            if vdi:
                vdi.ensureUnpaused()

    def _undoInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log("*** UNDO LEAF-COALESCE")
        parent = self.getVDI(parentUuid)
        if not parent:
            parent = self.getVDI(childUuid)
            if not parent:
                raise util.SMException("Neither %s nor %s found" % \
                        (parentUuid, childUuid))
            Util.log("Renaming parent back: %s -> %s" % (childUuid, parentUuid))
            parent.rename(parentUuid)
        util.fistpoint.activate("LVHDRT_coaleaf_undo_after_rename", self.uuid)

        child = self.getVDI(childUuid)
        if not child:
            child = self.getVDI(self.TMP_RENAME_PREFIX + childUuid)
            if not child:
                raise util.SMException("Neither %s nor %s found" % \
                        (childUuid, self.TMP_RENAME_PREFIX + childUuid))
            Util.log("Renaming child back to %s" % childUuid)
            child.rename(childUuid)
            Util.log("Updating the VDI record")
            child.setConfig(VDI.DB_VDI_PARENT, parentUuid)
            child.setConfig(VDI.DB_VDI_TYPE, child.vdi_type)
            util.fistpoint.activate("LVHDRT_coaleaf_undo_after_rename2", self.uuid)

            # refcount (best effort - assume that it had succeeded if the
            # second rename succeeded; if not, this adjustment will be wrong,
            # leading to a non-deactivation of the LV)
            ns = NS_PREFIX_LVM + self.uuid
            cCnt, cBcnt = RefCounter.check(child.uuid, ns)
            pCnt, pBcnt = RefCounter.check(parent.uuid, ns)
            pCnt = pCnt + cBcnt
            RefCounter.set(parent.uuid, pCnt, 0, ns)
            util.fistpoint.activate("LVHDRT_coaleaf_undo_after_refcount", self.uuid)

        parent.deflate()
        child.inflateFully()
        util.fistpoint.activate("LVHDRT_coaleaf_undo_after_deflate", self.uuid)
        if child.isHidden():
            child._setHidden(False)
        if not parent.isHidden():
            parent._setHidden(True)
        if not parent.lvReadonly:
            self.lvmCache.setReadonly(parent.fileName, True)
        self._updateSlavesOnUndoLeafCoalesce(parent, child)
        util.fistpoint.activate("LVHDRT_coaleaf_undo_end", self.uuid)
        Util.log("*** leaf-coalesce undo successful")
        if util.fistpoint.is_active("LVHDRT_coaleaf_stop_after_recovery"):
            child.setConfig(VDI.DB_LEAFCLSC, VDI.LEAFCLSC_DISABLED)

    def _finishInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log("*** FINISH LEAF-COALESCE")
        vdi = self.getVDI(childUuid)
        if not vdi:
            raise util.SMException("VDI %s not found" % childUuid)
        vdi.inflateFully()
        util.fistpoint.activate("LVHDRT_coaleaf_finish_after_inflate", self.uuid)
        try:
            self.forgetVDI(parentUuid)
        except XenAPI.Failure:
            pass
        self._updateSlavesOnResize(vdi)
        util.fistpoint.activate("LVHDRT_coaleaf_finish_end", self.uuid)
        Util.log("*** finished leaf-coalesce successfully")

    def _checkSlaves(self, vdi):
        """Confirm with all slaves in the pool that 'vdi' is not in use. We
        try to check all slaves, including those that the Agent believes are
        offline, but ignore failures for offline hosts. This is to avoid cases
        where the Agent thinks a host is offline but the host is up."""
        args = {"vgName": self.vgName,
                "action1": "deactivateNoRefcount",
                "lvName1": vdi.fileName,
                "action2": "cleanupLockAndRefcount",
                "uuid2": vdi.uuid,
                "ns2": NS_PREFIX_LVM + self.uuid}
        onlineHosts = self.xapi.getOnlineHosts()
        abortFlag = IPCFlag(self.uuid)
        for pbdRecord in self.xapi.getAttachedPBDs():
            hostRef = pbdRecord["host"]
            if hostRef == self.xapi._hostRef:
                continue
            if abortFlag.test(FLAG_TYPE_ABORT):
                raise AbortException("Aborting due to signal")
            Util.log("Checking with slave %s (path %s)" % (
                self.xapi.getRecordHost(hostRef)['hostname'], vdi.path))
            try:
                self.xapi.ensureInactive(hostRef, args)
            except XenAPI.Failure:
                if hostRef in onlineHosts:
                    raise

    @override
    def _updateSlavesOnUndoLeafCoalesce(self, parent, child) -> None:
        slaves = util.get_slaves_attached_on(self.xapi.session, [child.uuid])
        if not slaves:
            Util.log("Update-on-leaf-undo: VDI %s not attached on any slave" % \
                    child)
            return

        tmpName = child.vdi_type + self.TMP_RENAME_PREFIX + child.uuid
        args = {"vgName": self.vgName,
                "action1": "deactivateNoRefcount",
                "lvName1": tmpName,
                "action2": "deactivateNoRefcount",
                "lvName2": child.fileName,
                "action3": "refresh",
                "lvName3": child.fileName,
                "action4": "refresh",
                "lvName4": parent.fileName}
        for slave in slaves:
            Util.log("Updating %s, %s, %s on slave %s" % \
                    (tmpName, child.fileName, parent.fileName,
                     self.xapi.getRecordHost(slave)['hostname']))
            text = self.xapi.session.xenapi.host.call_plugin( \
                    slave, self.xapi.PLUGIN_ON_SLAVE, "multi", args)
            Util.log("call-plugin returned: '%s'" % text)

    @override
    def _updateSlavesOnRename(self, vdi, oldNameLV, origParentUuid) -> None:
        slaves = util.get_slaves_attached_on(self.xapi.session, [vdi.uuid])
        if not slaves:
            Util.log("Update-on-rename: VDI %s not attached on any slave" % vdi)
            return

        args = {"vgName": self.vgName,
                "action1": "deactivateNoRefcount",
                "lvName1": oldNameLV,
                "action2": "refresh",
                "lvName2": vdi.fileName,
                "action3": "cleanupLockAndRefcount",
                "uuid3": origParentUuid,
                "ns3": NS_PREFIX_LVM + self.uuid}
        for slave in slaves:
            Util.log("Updating %s to %s on slave %s" % \
                    (oldNameLV, vdi.fileName,
                     self.xapi.getRecordHost(slave)['hostname']))
            text = self.xapi.session.xenapi.host.call_plugin( \
                    slave, self.xapi.PLUGIN_ON_SLAVE, "multi", args)
            Util.log("call-plugin returned: '%s'" % text)

    @override
    def _updateSlavesOnResize(self, vdi) -> None:
        uuids = [x.uuid for x in vdi.getAllLeaves()]
        slaves = util.get_slaves_attached_on(self.xapi.session, uuids)
        if not slaves:
            util.SMlog("Update-on-resize: %s not attached on any slave" % vdi)
            return
        LvmCowUtil.refreshVolumeOnSlaves(self.xapi.session, self.uuid, self.vgName,
                vdi.fileName, vdi.uuid, slaves)


class LinstorSR(SR):
    TYPE = SR.TYPE_LINSTOR

    def __init__(self, uuid, xapi, createLock, force):
        if not LINSTOR_AVAILABLE:
            raise util.SMException(
                'Can\'t load cleanup LinstorSR: LINSTOR libraries are missing'
            )

        SR.__init__(self, uuid, xapi, createLock, force)
        self.path = LinstorVolumeManager.DEV_ROOT_PATH

        class LinstorProxy:
            def __init__(self, sr: LinstorSR) -> None:
                self.sr = sr

            def __getattr__(self, attr: str) -> Any:
                assert self.sr, "Cannot use `LinstorProxy` without valid `LinstorVolumeManager` instance"
                return getattr(self.sr._linstor, attr)

        self._linstor = None
        self._linstor_proxy = LinstorProxy(self)
        self._reloadLinstor(journaler_only=True)

    @override
    def deleteVDI(self, vdi) -> None:
        self._checkSlaves(vdi)
        SR.deleteVDI(self, vdi)

    @override
    def getFreeSpace(self) -> int:
        return self._linstor.max_volume_size_allowed

    @override
    def scan(self, force=False) -> None:
        all_vdi_info = self._scan(force)
        for uuid, vdiInfo in all_vdi_info.items():
            # When vdiInfo is None, the VDI is RAW.
            vdi = self.getVDI(uuid)
            if not vdi:
                self.logFilter.logNewVDI(uuid)
                vdi = LinstorVDI(self, uuid, vdiInfo.vdiType if vdiInfo else VdiType.RAW)
                self.vdis[uuid] = vdi
            if vdiInfo:
                vdi.load(vdiInfo)
        self._removeStaleVDIs(all_vdi_info.keys())
        self._buildTree(force)
        self.logFilter.logState()
        self._handleInterruptedCoalesceLeaf()

    @override
    def pauseVDIs(self, vdiList) -> None:
        self._linstor.ensure_volume_list_is_not_locked(
            vdiList, timeout=LinstorVDI.VOLUME_LOCK_TIMEOUT
        )
        return super(LinstorSR, self).pauseVDIs(vdiList)

    def _reloadLinstor(self, journaler_only=False):
        session = self.xapi.session
        host_ref = util.get_this_host_ref(session)
        sr_ref = session.xenapi.SR.get_by_uuid(self.uuid)

        pbd = util.find_my_pbd(session, host_ref, sr_ref)
        if pbd is None:
            raise util.SMException('Failed to find PBD')

        dconf = session.xenapi.PBD.get_device_config(pbd)
        group_name = dconf['group-name']

        controller_uri = get_controller_uri()

        if not journaler_only:
            self._linstor = LinstorVolumeManager(
                controller_uri,
                group_name,
                repair=True,
                logger=util.SMlog
            )

        self.journaler = LinstorJournaler(
            group_name,
            uri=None if self._linstor else controller_uri,
            native_client=self._linstor.native_client if self._linstor else None,
            logger=util.SMlog
        )

    def _scan(self, force):
        for i in range(SR.SCAN_RETRY_ATTEMPTS):
            self._reloadLinstor()
            error = False
            try:
                all_vdi_info = self._load_vdi_info()
                for uuid, vdiInfo in all_vdi_info.items():
                    if vdiInfo and vdiInfo.error:
                        error = True
                        break
                if not error:
                    return all_vdi_info
                Util.log('Scan error, retrying ({})'.format(i))
            except Exception as e:
                Util.log('Scan exception, retrying ({}): {}'.format(i, e))
                Util.log(traceback.format_exc())

        if force:
            return all_vdi_info
        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_vdis = []

        def handle_fail(vdi_uuid, e):
            Util.log(f" [VDI {vdi_uuid}: failed to load VDI info]: {e}")
            info = CowImageInfo(vdi_uuid)
            info.error = 1
            return info

        for vdi_uuid, volume_info in all_volume_info.items():
            vdi_type = VdiType.RAW
            try:
                volume_metadata = volumes_metadata[vdi_uuid]
                if not volume_info.name and not list(volume_metadata.items()):
                    continue  # Ignore it, probably deleted.

                if vdi_uuid.startswith('DELETED_'):
                    # Assume it's really a RAW volume of a failed snap without COW header/footer.
                    # We must remove this VDI now without adding it in the VDI list.
                    # Otherwise `Relinking` calls and other actions can be launched on it.
                    # We don't want that...
                    Util.log('Deleting bad VDI {}'.format(vdi_uuid))

                    self.lock()
                    try:
                        self._linstor.destroy_volume(vdi_uuid)
                        try:
                            self.forgetVDI(vdi_uuid)
                        except:
                            pass
                    except Exception as e:
                        Util.log('Cannot delete bad VDI: {}'.format(e))
                    finally:
                        self.unlock()
                    continue

                vdi_type = volume_metadata.get(VDI_TYPE_TAG)
                if VdiType.isCowImage(vdi_type):
                    pending_vdis.append((vdi_uuid, vdi_type))
                else:
                    all_vdi_info[vdi_uuid] = None
            except Exception as e:
                all_vdi_info[vdi_uuid] = handle_fail(vdi_uuid, e)

        multi_cowutil = MultiLinstorCowUtil(self._linstor.uri, self._linstor.group_name)

        def load_info(vdi, multi_cowutil):
            vdi_uuid, vdi_type = vdi
            try:
                vdiInfo = multi_cowutil.get_local_cowutil(vdi_type).get_info(vdi_uuid)
            except Exception as e:
                vdiInfo = handle_fail(vdi_uuid, e)
            vdiInfo.vdiType = vdi_type
            return vdiInfo

        try:
            for vdiInfo in multi_cowutil.run(load_info, pending_vdis):
                all_vdi_info[vdiInfo.uuid] = vdiInfo
        finally:
            del multi_cowutil

        return all_vdi_info

    @override
    def _prepareCoalesceLeaf(self, vdi) -> None:
        vdi._activateChain()
        vdi.deflate()
        vdi._inflateParentForCoalesce()

    @override
    def _finishCoalesceLeaf(self, parent) -> None:
        if not parent.isSnapshot() or parent.isAttachedRW():
            parent.inflateFully()
        else:
            parent.deflate()

    @override
    def _calcExtraSpaceNeeded(self, child, parent) -> int:
        return LinstorCowUtil(
            self.xapi.session, self._linstor, parent.vdi_type
        ).compute_volume_size(parent.sizeVirt) - parent.getDrbdSize()

    def _hasValidDevicePath(self, uuid):
        try:
            self._linstor.get_device_path(uuid)
        except Exception:
            # TODO: Maybe log exception.
            return False
        return True

    @override
    def _liveLeafCoalesce(self, vdi: VDI, coalesce_on_remote: bool = False) -> bool:
        self.lock()
        try:
            self._linstor.ensure_volume_is_not_locked(
                vdi.uuid, timeout=LinstorVDI.VOLUME_LOCK_TIMEOUT
            )
            return super(LinstorSR, self)._liveLeafCoalesce(vdi)
        finally:
            self.unlock()

    @override
    def _handleInterruptedCoalesceLeaf(self) -> None:
        entries = self.journaler.get_all(VDI.JRN_LEAF)
        for uuid, parentUuid in entries.items():
            if self._hasValidDevicePath(parentUuid) or \
                    self._hasValidDevicePath(self.TMP_RENAME_PREFIX + uuid):
                self._undoInterruptedCoalesceLeaf(uuid, parentUuid)
            else:
                self._finishInterruptedCoalesceLeaf(uuid, parentUuid)
            self.journaler.remove(VDI.JRN_LEAF, uuid)
            vdi = self.getVDI(uuid)
            if vdi:
                vdi.ensureUnpaused()

    def _undoInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log('*** UNDO LEAF-COALESCE')
        parent = self.getVDI(parentUuid)
        if not parent:
            parent = self.getVDI(childUuid)
            if not parent:
                raise util.SMException(
                    'Neither {} nor {} found'.format(parentUuid, childUuid)
                )
            Util.log(
                'Renaming parent back: {} -> {}'.format(childUuid, parentUuid)
            )
            parent.rename(parentUuid)

        child = self.getVDI(childUuid)
        if not child:
            child = self.getVDI(self.TMP_RENAME_PREFIX + childUuid)
            if not child:
                raise util.SMException(
                    'Neither {} nor {} found'.format(
                        childUuid, self.TMP_RENAME_PREFIX + childUuid
                    )
                )
            Util.log('Renaming child back to {}'.format(childUuid))
            child.rename(childUuid)
            Util.log('Updating the VDI record')
            child.setConfig(VDI.DB_VDI_PARENT, parentUuid)
            child.setConfig(VDI.DB_VDI_TYPE, child.vdi_type)

        # TODO: Maybe deflate here.

        if child.isHidden():
            child._setHidden(False)
        if not parent.isHidden():
            parent._setHidden(True)
        self._updateSlavesOnUndoLeafCoalesce(parent, child)
        Util.log('*** leaf-coalesce undo successful')

    def _finishInterruptedCoalesceLeaf(self, childUuid, parentUuid):
        Util.log('*** FINISH LEAF-COALESCE')
        vdi = self.getVDI(childUuid)
        if not vdi:
            raise util.SMException('VDI {} not found'.format(childUuid))
        # TODO: Maybe inflate.
        try:
            self.forgetVDI(parentUuid)
        except XenAPI.Failure:
            pass
        self._updateSlavesOnResize(vdi)
        Util.log('*** finished leaf-coalesce successfully')

    def _checkSlaves(self, vdi):
        try:
            openers = self._linstor.get_volume_openers(vdi.uuid)
            for host_openers in openers.values():
                for opener in host_openers.values():
                    if opener['process-name'] != 'tapdisk':
                        raise util.SMException(
                            'VDI {} is in use: {}'.format(vdi.uuid, openers)
                        )
        except LinstorVolumeManagerError as e:
            if e.code != LinstorVolumeManagerError.ERR_VOLUME_NOT_EXISTS:
                raise

    @classmethod
    def abort_gc_from_openers_vdi(cls, vdi_uuid: str, openers: "LinstorVolumeOpeners") -> bool:
        return cls._abort_gc_from_openers(vdi_uuid, True, openers)

    @classmethod
    def abort_gc_from_openers_sr(cls, sr_uuid: str, openers: "LinstorVolumeOpeners") -> bool:
        return cls._abort_gc_from_openers(sr_uuid, False, openers)

    @staticmethod
    def _abort_gc_from_openers(uuid: str, is_vdi_uuid: bool, openers: "LinstorVolumeOpeners") -> bool:
        from linstorcowutil import MANAGER_PLUGIN

        node_name = None

        for host_openers in openers.values():
            for hostname, opener in host_openers.items():
                # Not the most accurate check but it works...
                # `vhd-util` is probably prefixed with a "+" which is ignored here.
                if not opener["process-name"].endswith("vhd-util") or "coalesce" not in opener["cmdline"]:
                    continue

                if not node_name:
                    import socket
                    node_name = socket.gethostname()

                if node_name == hostname:
                    continue

                with util.timeout(5), util.ApiSession("SM-GC-coalescing") as session:
                    sr_uuid = util.get_sr_uuid_from_vdi_uuid(session, uuid) if is_vdi_uuid else uuid
                    util.SMlog(f"LINSTOR volume is coalescing on `{sr_uuid}`. We're going to interrupt the GC...")
                    return util.strtobool(session.xenapi.host.call_plugin(
                        util.get_master_ref(session), MANAGER_PLUGIN, "abortGc", {"srUuid": sr_uuid}
                    ))
        return False



################################################################################
#
#  Helpers
#
def daemonize():
    pid = os.fork()
    if pid:
        os.waitpid(pid, 0)
        Util.log("New PID [%d]" % pid)
        return False
    os.chdir("/")
    os.setsid()
    pid = os.fork()
    if pid:
        Util.log("Will finish as PID [%d]" % pid)
        os._exit(0)
    for fd in [0, 1, 2]:
        try:
            os.close(fd)
        except OSError:
            pass
    # we need to fill those special fd numbers or pread won't work
    sys.stdin = open("/dev/null", 'r')
    sys.stderr = open("/dev/null", 'w')
    sys.stdout = open("/dev/null", 'w')
    # As we're a new process we need to clear the lock objects
    lock.Lock.clearAll()
    return True


def normalizeType(type):
    if type in LVMSR.SUBTYPES:
        type = SR.TYPE_LVHD
    if type in ["lvm", "lvmoiscsi", "lvmohba", "lvmofcoe"]:
        # temporary while LVHD is symlinked as LVM
        type = SR.TYPE_LVHD
    if type in [
        "ext", "nfs", "ocfsoiscsi", "ocfsohba", "smb", "cephfs", "glusterfs",
        "moosefs", "xfs", "zfs", "largeblock"
    ]:
        type = SR.TYPE_FILE
    if type in ["linstor"]:
        type = SR.TYPE_LINSTOR
    if type not in SR.TYPES:
        raise util.SMException("Unsupported SR type: %s" % type)
    return type

GCPAUSE_DEFAULT_SLEEP = 5 * 60


def _gc_init_file(sr_uuid):
    return os.path.join(NON_PERSISTENT_DIR, str(sr_uuid), 'gc_init')


def _create_init_file(sr_uuid):
    util.makedirs(os.path.join(NON_PERSISTENT_DIR, str(sr_uuid)))
    with open(os.path.join(_gc_init_file(sr_uuid)), 'w+') as f:
        f.write('1')


def _gcLoopPause(sr, dryRun=False, immediate=False):
    if immediate:
        return

    # Check to see if the GCPAUSE_FISTPOINT is present. If so the fist
    # point will just return. Otherwise, fall back on an abortable sleep.

    if util.fistpoint.is_active(util.GCPAUSE_FISTPOINT):

        util.fistpoint.activate_custom_fn(util.GCPAUSE_FISTPOINT,
                                          lambda *args: None)
    elif os.path.exists(_gc_init_file(sr.uuid)):
        def abortTest():
            return IPCFlag(sr.uuid).test(FLAG_TYPE_ABORT)

        # If time.sleep hangs we are in deep trouble, however for
        # completeness we set the timeout of the abort thread to
        # 110% of GCPAUSE_DEFAULT_SLEEP.
        Util.log("GC active, about to go quiet")
        Util.runAbortable(lambda: time.sleep(GCPAUSE_DEFAULT_SLEEP),
                          None, sr.uuid, abortTest, VDI.POLL_INTERVAL,
                          GCPAUSE_DEFAULT_SLEEP * 1.1)
        Util.log("GC active, quiet period ended")


def _gcLoop(sr, dryRun=False, immediate=False):
    if not lockGCActive.acquireNoblock():
        Util.log("Another GC instance already active, exiting")
        return

    # Check we're still attached after acquiring locks
    if not sr.xapi.isPluggedHere():
        Util.log("SR no longer attached, exiting")
        return

    # Clean up Intellicache files
    sr.cleanupCache()

    # Track how many we do
    coalesced = 0
    task_status = "success"
    try:
        # Check if any work needs to be done
        if not sr.xapi.isPluggedHere():
            Util.log("SR no longer attached, exiting")
            return
        sr.scanLocked()
        if not sr.hasWork():
            Util.log("No work, exiting")
            return
        sr.xapi.create_task(
            "Garbage Collection",
            "Garbage collection for SR %s" % sr.uuid)
        _gcLoopPause(sr, dryRun, immediate=immediate)
        while True:
            if SIGTERM:
                Util.log("Term requested")
                return

            if not sr.xapi.isPluggedHere():
                Util.log("SR no longer attached, exiting")
                break
            sr.scanLocked()
            if not sr.hasWork():
                Util.log("No work, exiting")
                break

            if not lockGCRunning.acquireNoblock():
                Util.log("Unable to acquire GC running lock.")
                return
            try:
                if not sr.gcEnabled():
                    break

                sr.xapi.update_task_progress("done", coalesced)

                sr.cleanupCoalesceJournals()
                # Create the init file here in case startup is waiting on it
                _create_init_file(sr.uuid)
                sr.scanLocked()
                sr.updateBlockInfo()

                howmany = len(sr.findGarbage())
                if howmany > 0:
                    Util.log("Found %d orphaned vdis" % howmany)
                    sr.lock()
                    try:
                        sr.garbageCollect(dryRun)
                    finally:
                        sr.unlock()
                    sr.xapi.srUpdate()

                candidate = sr.findCoalesceable()
                if candidate:
                    util.fistpoint.activate(
                        "LVHDRT_finding_a_suitable_pair", sr.uuid)
                    sr.coalesce(candidate, dryRun)
                    sr.xapi.srUpdate()
                    coalesced += 1
                    continue

                candidate = sr.findLeafCoalesceable()
                if candidate:
                    sr.coalesceLeaf(candidate, dryRun)
                    sr.xapi.srUpdate()
                    coalesced += 1
                    continue

            finally:
                lockGCRunning.release()
    except:
        task_status = "failure"
        raise
    finally:
        sr.xapi.set_task_status(task_status)
        Util.log("GC process exiting, no work left")
        _create_init_file(sr.uuid)
        lockGCActive.release()


def _gc(session, srUuid, dryRun=False, immediate=False):
    init(srUuid)
    sr = SR.getInstance(srUuid, session)
    if not sr.gcEnabled(False):
        return

    try:
        _gcLoop(sr, dryRun, immediate=immediate)
    finally:
        sr.check_no_space_candidates()
        sr.cleanup()
        sr.logFilter.logState()
        del sr.xapi


def _abort(srUuid, soft=False):
    """Aborts an GC/coalesce.

    srUuid: the UUID of the SR whose GC/coalesce must be aborted
    soft: If set to True and there is a pending abort signal, the function
    doesn't do anything. If set to False, a new abort signal is issued.

    returns: If soft is set to False, we return True holding lockGCActive. If
    soft is set to False and an abort signal is pending, we return False
    without holding lockGCActive. An exception is raised in case of error."""
    Util.log("=== SR %s: abort ===" % (srUuid))
    init(srUuid)
    if not lockGCActive.acquireNoblock():
        gotLock = False
        Util.log("Aborting currently-running instance (SR %s)" % srUuid)
        abortFlag = IPCFlag(srUuid)
        if not abortFlag.set(FLAG_TYPE_ABORT, soft):
            return False
        for i in range(SR.LOCK_RETRY_ATTEMPTS):
            gotLock = lockGCActive.acquireNoblock()
            if gotLock:
                break
            time.sleep(SR.LOCK_RETRY_INTERVAL)
        abortFlag.clear(FLAG_TYPE_ABORT)
        if not gotLock:
            raise util.CommandException(code=errno.ETIMEDOUT,
                    reason="SR %s: error aborting existing process" % srUuid)
    return True


def init(srUuid):
    global lockGCRunning
    if not lockGCRunning:
        lockGCRunning = lock.Lock(lock.LOCK_TYPE_GC_RUNNING, srUuid)
    global lockGCActive
    if not lockGCActive:
        lockGCActive = LockActive(srUuid)


class LockActive:
    """
    Wraps the use of LOCK_TYPE_GC_ACTIVE such that the lock cannot be acquired
    if another process holds the SR lock.
    """
    def __init__(self, srUuid):
        self._lock = lock.Lock(LOCK_TYPE_GC_ACTIVE, srUuid)
        self._srLock = lock.Lock(lock.LOCK_TYPE_SR, srUuid)

    def acquireNoblock(self):
        self._srLock.acquire()

        try:
            return self._lock.acquireNoblock()
        finally:
            self._srLock.release()

    def release(self):
        self._lock.release()


def usage():
    output = """Garbage collect and/or coalesce COW images in a COW-based SR

Parameters:
    -u --uuid UUID   SR UUID
 and one of:
    -g --gc          garbage collect, coalesce, and repeat while there is work
    -G --gc_force    garbage collect once, aborting any current operations
    -c --cache-clean <max_age> clean up IntelliCache cache files older than
                     max_age hours
    -a --abort       abort any currently running operation (GC or coalesce)
    -q --query       query the current state (GC'ing, coalescing or not running)
    -x --disable     disable GC/coalesce (will be in effect until you exit)
    -t --debug       see Debug below

Options:
    -b --background  run in background (return immediately) (valid for -g only)
    -f --force       continue in the presence of COW images with errors (when doing
                     GC, this might cause removal of any such images) (only valid
                     for -G) (DANGEROUS)

Debug:
    The --debug parameter enables manipulation of LVHD VDIs for debugging
    purposes.  ** NEVER USE IT ON A LIVE VM **
    The following parameters are required:
    -t --debug <cmd> <cmd> is one of "activate", "deactivate", "inflate",
                     "deflate".
    -v --vdi_uuid    VDI UUID
    """
    #-d --dry-run     don't actually perform any SR-modifying operations
    print(output)
    Util.log("(Invalid usage)")
    sys.exit(1)


##############################################################################
#
#  API
#
def abort(srUuid, soft=False):
    """Abort GC/coalesce if we are currently GC'ing or coalescing a VDI pair.
    """
    if _abort(srUuid, soft):
        stop_gc_service(srUuid)
        Util.log("abort: releasing the process lock")
        lockGCActive.release()
        return True
    else:
        return False


def run_gc(session, srUuid, dryRun, immediate=False):
    try:
        _gc(session, srUuid, dryRun, immediate=immediate)
        return 0
    except AbortException:
        Util.log("Aborted")
        return 2
    except Exception:
        Util.logException("gc")
        Util.log("* * * * * SR %s: ERROR\n" % srUuid)
        return 1


def gc(session, srUuid, inBackground, dryRun=False):
    """Garbage collect all deleted VDIs in SR "srUuid". Fork & return
    immediately if inBackground=True.

    The following algorithm is used:
    1. If we are already GC'ing in this SR, return
    2. If we are already coalescing a VDI pair:
        a. Scan the SR and determine if the VDI pair is GC'able
        b. If the pair is not GC'able, return
        c. If the pair is GC'able, abort coalesce
    3. Scan the SR
    4. If there is nothing to collect, nor to coalesce, return
    5. If there is something to collect, GC all, then goto 3
    6. If there is something to coalesce, coalesce one pair, then goto 3
    """
    Util.log("=== SR %s: gc ===" % srUuid)

    signal.signal(signal.SIGTERM, receiveSignal)

    if inBackground:
        if daemonize():
            # we are now running in the background. Catch & log any errors
            # because there is no other way to propagate them back at this
            # point

            run_gc(None, srUuid, dryRun)
            os._exit(0)
    else:
        os._exit(run_gc(session, srUuid, dryRun, immediate=True))


def start_gc(session, sr_uuid):
    """
    This function is used to try to start a backgrounded GC session by forking
    the current process. If using the systemd version, call start_gc_service() instead.
    """
    # don't bother if an instance already running (this is just an
    # optimization to reduce the overhead of forking a new process if we
    # don't have to, but the process will check the lock anyways)
    lockRunning = lock.Lock(lock.LOCK_TYPE_GC_RUNNING, sr_uuid)
    if not lockRunning.acquireNoblock():
        if should_preempt(session, sr_uuid):
            util.SMlog("Aborting currently-running coalesce of garbage VDI")
            try:
                if not abort(sr_uuid, soft=True):
                    util.SMlog("The GC has already been scheduled to re-start")
            except util.CommandException as e:
                if e.code != errno.ETIMEDOUT:
                    raise
                util.SMlog('failed to abort the GC')
        else:
            util.SMlog("A GC instance already running, not kicking")
            return
    else:
        lockRunning.release()

    util.SMlog(f"Starting GC file is {__file__}")
    subprocess.run([__file__, '-b', '-u', sr_uuid, '-g'],
                   stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)

def _gc_service_cmd(sr_uuid, action, extra_args=None):
    """
    Build and run the systemctl command for the GC service using util.doexec.
    """
    sr_uuid_esc = sr_uuid.replace("-", "\\x2d")
    cmd=["/usr/bin/systemctl", "--quiet"]
    if extra_args:
        cmd.extend(extra_args)
    cmd += [action, f"SMGC@{sr_uuid_esc}"]
    return util.doexec(cmd)


def start_gc_service(sr_uuid, wait=False):
    """
    This starts the templated systemd service which runs GC on the given SR UUID.
    If the service was already started, this is a no-op.

    Because the service is a one-shot with RemainAfterExit=no, when called with
    wait=True this will run the service synchronously and will not return until the
    run has finished. This is used to force a run of the GC instead of just kicking it
    in the background.
    """
    util.SMlog(f"Kicking SMGC@{sr_uuid}...")
    _gc_service_cmd(sr_uuid, "start", extra_args=None if wait else ["--no-block"])


def stop_gc_service(sr_uuid):
    """
    Stops the templated systemd service which runs GC on the given SR UUID.
    """
    util.SMlog(f"Stopping SMGC@{sr_uuid}...")
    (rc, _stdout, stderr) = _gc_service_cmd(sr_uuid, "stop")
    if rc != 0:
        util.SMlog(f"Failed to stop gc service `SMGC@{sr_uuid}`: `{stderr}`")


def wait_for_completion(sr_uuid):
    while get_state(sr_uuid):
        time.sleep(5)


def gc_force(session, srUuid, force=False, dryRun=False, lockSR=False):
    """Garbage collect all deleted VDIs in SR "srUuid". The caller must ensure
    the SR lock is held.
    The following algorithm is used:
    1. If we are already GC'ing or coalescing a VDI pair, abort GC/coalesce
    2. Scan the SR
    3. GC
    4. return
    """
    Util.log("=== SR %s: gc_force ===" % srUuid)
    init(srUuid)
    sr = SR.getInstance(srUuid, session, lockSR, True)
    if not lockGCActive.acquireNoblock():
        abort(srUuid)
    else:
        Util.log("Nothing was running, clear to proceed")

    if force:
        Util.log("FORCED: will continue even if there are COW image errors")
    sr.scanLocked(force)
    sr.cleanupCoalesceJournals()

    try:
        sr.cleanupCache()
        sr.garbageCollect(dryRun)
    finally:
        sr.cleanup()
        sr.logFilter.logState()
        lockGCActive.release()


def get_state(srUuid):
    """Return whether GC/coalesce is currently running or not. This asks systemd for
    the state of the templated SMGC service and will return True if it is "activating"
    or "running" (for completeness, as in practice it will never achieve the latter state)
    """
    sr_uuid_esc = srUuid.replace("-", "\\x2d")
    cmd=[ "/usr/bin/systemctl", "is-active", f"SMGC@{sr_uuid_esc}"]
    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, close_fds=True)
    state = result.stdout.decode('utf-8').rstrip()
    if state == "activating" or state == "running":
        return True
    return False


def should_preempt(session, srUuid):
    sr = SR.getInstance(srUuid, session)
    entries = sr.journaler.getAll(VDI.JRN_COALESCE)
    if len(entries) == 0:
        return False
    elif len(entries) > 1:
        raise util.SMException("More than one coalesce entry: " + str(entries))
    sr.scanLocked()
    coalescedUuid = entries.popitem()[0]
    garbage = sr.findGarbage()
    for vdi in garbage:
        if vdi.uuid == coalescedUuid:
            return True
    return False


def get_coalesceable_leaves(session, srUuid, vdiUuids):
    coalesceable = []
    sr = SR.getInstance(srUuid, session)
    sr.scanLocked()
    for uuid in vdiUuids:
        vdi = sr.getVDI(uuid)
        if not vdi:
            raise util.SMException("VDI %s not found" % uuid)
        if vdi.isLeafCoalesceable():
            coalesceable.append(uuid)
    return coalesceable


def cache_cleanup(session, srUuid, maxAge):
    sr = SR.getInstance(srUuid, session)
    return sr.cleanupCache(maxAge)


def debug(sr_uuid, cmd, vdi_uuid):
    Util.log("Debug command: %s" % cmd)
    sr = SR.getInstance(sr_uuid, None)
    if not isinstance(sr, LVMSR):
        print("Error: not an LVHD SR")
        return
    sr.scanLocked()
    vdi = sr.getVDI(vdi_uuid)
    if not vdi:
        print("Error: VDI %s not found")
        return
    print("Running %s on SR %s" % (cmd, sr))
    print("VDI before: %s" % vdi)
    if cmd == "activate":
        vdi._activate()
        print("VDI file: %s" % vdi.path)
    if cmd == "deactivate":
        ns = NS_PREFIX_LVM + sr.uuid
        sr.lvmCache.deactivate(ns, vdi.uuid, vdi.fileName, False)
    if cmd == "inflate":
        vdi.inflateFully()
        sr.cleanup()
    if cmd == "deflate":
        vdi.deflate()
        sr.cleanup()
    sr.scanLocked()
    print("VDI after:  %s" % vdi)


def abort_optional_reenable(uuid):
    print("Disabling GC/coalesce for %s" % uuid)
    ret = _abort(uuid)
    input("Press enter to re-enable...")
    print("GC/coalesce re-enabled")
    lockGCRunning.release()
    if ret:
        lockGCActive.release()


##############################################################################
#
#  CLI
#
def main():
    action = ""
    maxAge = 0
    uuid = ""
    background = False
    force = False
    dryRun = False
    debug_cmd = ""
    vdi_uuid = ""
    shortArgs = "gGc:aqxu:bfdt:v:"
    longArgs = ["gc", "gc_force", "clean_cache", "abort", "query", "disable",
            "uuid=", "background", "force", "dry-run", "debug=", "vdi_uuid="]

    try:
        opts, args = getopt.getopt(sys.argv[1:], shortArgs, longArgs)
    except getopt.GetoptError:
        usage()
    for o, a in opts:
        if o in ("-g", "--gc"):
            action = "gc"
        if o in ("-G", "--gc_force"):
            action = "gc_force"
        if o in ("-c", "--clean_cache"):
            action = "clean_cache"
            maxAge = int(a)
        if o in ("-a", "--abort"):
            action = "abort"
        if o in ("-q", "--query"):
            action = "query"
        if o in ("-x", "--disable"):
            action = "disable"
        if o in ("-u", "--uuid"):
            uuid = a
        if o in ("-b", "--background"):
            background = True
        if o in ("-f", "--force"):
            force = True
        if o in ("-d", "--dry-run"):
            Util.log("Dry run mode")
            dryRun = True
        if o in ("-t", "--debug"):
            action = "debug"
            debug_cmd = a
        if o in ("-v", "--vdi_uuid"):
            vdi_uuid = a

    if not action or not uuid:
        usage()
    if action == "debug" and not (debug_cmd and vdi_uuid) or \
            action != "debug" and (debug_cmd or vdi_uuid):
        usage()

    if action != "query" and action != "debug":
        print("All output goes to log")

    if action == "gc":
        gc(None, uuid, background, dryRun)
    elif action == "gc_force":
        gc_force(None, uuid, force, dryRun, True)
    elif action == "clean_cache":
        cache_cleanup(None, uuid, maxAge)
    elif action == "abort":
        abort(uuid)
    elif action == "query":
        print("Currently running: %s" % get_state(uuid))
    elif action == "disable":
        abort_optional_reenable(uuid)
    elif action == "debug":
        debug(uuid, debug_cmd, vdi_uuid)


if __name__ == '__main__':
    main()