Hello Dan Kenigsberg,
I'd like you to do a code review. Please visit
to review the following change.
Change subject: clusterlock: add the local locking implementation ......................................................................
clusterlock: add the local locking implementation
In order to have a faster and more lightweight locking mechanism on local storage domains a new cluster lock (based on flock) has been introduced.
Change-Id: I106618a9a61cc96727edaf2e3ab043b2ec28ebe1 Signed-off-by: Federico Simoncelli fsimonce@redhat.com Reviewed-on: http://gerrit.ovirt.org/10282 Reviewed-by: Dan Kenigsberg danken@redhat.com --- M vdsm/storage/clusterlock.py M vdsm/storage/localFsSD.py M vdsm/storage/misc.py 3 files changed, 141 insertions(+), 7 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/65/11465/1
diff --git a/vdsm/storage/clusterlock.py b/vdsm/storage/clusterlock.py index 4525b2f..cabf174 100644 --- a/vdsm/storage/clusterlock.py +++ b/vdsm/storage/clusterlock.py @@ -19,6 +19,7 @@ #
import os +import fcntl import threading import logging import subprocess @@ -127,6 +128,22 @@ self.log.debug("Cluster lock released successfully")
+initSANLockLog = logging.getLogger("initSANLock") + + +def initSANLock(sdUUID, idsPath, leasesPath): + initSANLockLog.debug("Initializing SANLock for domain %s", sdUUID) + + try: + sanlock.init_lockspace(sdUUID, idsPath) + sanlock.init_resource(sdUUID, SDM_LEASE_NAME, + [(leasesPath, SDM_LEASE_OFFSET)]) + except sanlock.SanlockException: + initSANLockLog.error("Cannot initialize SANLock for domain %s", + sdUUID, exc_info=True) + raise se.ClusterLockInitError() + + class SANLock(object): log = logging.getLogger("SANLock")
@@ -141,13 +158,7 @@ self._sanlockfd = None
def initLock(self): - try: - sanlock.init_lockspace(self._sdUUID, self._idsPath) - sanlock.init_resource(self._sdUUID, SDM_LEASE_NAME, - [(self._leasesPath, SDM_LEASE_OFFSET)]) - except sanlock.SanlockException: - self.log.warn("Cannot initialize clusterlock", exc_info=True) - raise se.ClusterLockInitError() + initSANLock(self._sdUUID, self._idsPath, self._leasesPath)
def setParams(self, *args): pass @@ -249,3 +260,100 @@ self._sanlockfd = None self.log.debug("Cluster lock for domain %s successfully released", self._sdUUID) + + +class LocalLock(object): + log = logging.getLogger("LocalLock") + + _globalLockMap = {} + _globalLockMapSync = threading.Lock() + + def __init__(self, sdUUID, idsPath, leasesPath, *args): + self._sdUUID = sdUUID + self._idsPath = idsPath + self._leasesPath = leasesPath + + def initLock(self): + # The LocalLock initialization is based on SANLock to maintain on-disk + # domain format consistent across all the V3 types. + # The advantage is that the domain can be exposed as an NFS/GlusterFS + # domain later on without any modification. + # XXX: Keep in mind that LocalLock and SANLock cannot detect each other + # and therefore concurrently using the same domain as local domain and + # NFS domain (or any other shared file-based domain) will certainly + # lead to disastrous consequences. + initSANLock(self._sdUUID, self._idsPath, self._leasesPath) + + def setParams(self, *args): + pass + + def getReservedId(self): + return MAX_HOST_ID + + def acquireHostId(self, hostId, async): + self.log.debug("Host id for domain %s successfully acquired (id: %s)", + self._sdUUID, hostId) + + def releaseHostId(self, hostId, async, unused): + self.log.debug("Host id for domain %s released successfully (id: %s)", + self._sdUUID, hostId) + + def hasHostId(self, hostId): + return True + + def acquire(self, hostId): + with self._globalLockMapSync: + self.log.info("Acquiring local lock for domain %s (id: %s)", + self._sdUUID, hostId) + + lockFile = self._globalLockMap.get(self._sdUUID, None) + + if lockFile: + try: + misc.NoIntrCall(fcntl.fcntl, lockFile, fcntl.F_GETFD) + except IOError as e: + # We found a stale file descriptor, removing. + del self._globalLockMap[self._sdUUID] + + # Raise any other unkown error. + if e.errno != os.errno.EBADF: + raise + else: + self.log.debug("Local lock already acquired for domain " + "%s (id: %s)", self._sdUUID, hostId) + return # success, the lock was already acquired + + lockFile = misc.NoIntrCall(os.open, self._idsPath, os.O_RDONLY) + + try: + misc.NoIntrCall(fcntl.flock, lockFile, + fcntl.LOCK_EX | fcntl.LOCK_NB) + except IOError as e: + misc.NoIntrCall(os.close, lockFile) + if e.errno in (os.errno.EACCES, os.errno.EAGAIN): + raise se.AcquireLockFailure( + self._sdUUID, e.errno, "Cannot acquire local lock", + str(e)) + raise + else: + self._globalLockMap[self._sdUUID] = lockFile + + self.log.debug("Local lock for domain %s successfully acquired " + "(id: %s)", self._sdUUID, hostId) + + def release(self): + with self._globalLockMapSync: + self.log.info("Releasing local lock for domain %s", self._sdUUID) + + lockFile = self._globalLockMap.get(self._sdUUID, None) + + if not lockFile: + self.log.debug("Local lock already released for domain %s", + self._sdUUID) + return + + misc.NoIntrCall(os.close, lockFile) + del self._globalLockMap[self._sdUUID] + + self.log.debug("Local lock for domain %s successfully released", + self._sdUUID) diff --git a/vdsm/storage/localFsSD.py b/vdsm/storage/localFsSD.py index 198c073..7d59894 100644 --- a/vdsm/storage/localFsSD.py +++ b/vdsm/storage/localFsSD.py @@ -26,9 +26,16 @@ import fileUtils import storage_exception as se import misc +import clusterlock
class LocalFsStorageDomain(fileSD.FileStorageDomain): + # version: (clusterLockClass, hasVolumeLeases) + _clusterLockTable = { + 0: (clusterlock.SafeLease, False), + 2: (clusterlock.SafeLease, False), + 3: (clusterlock.LocalLock, True), + }
@classmethod def _preCreateValidation(cls, sdUUID, domPath, typeSpecificArg, version): diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py index 17d38ee..b26a317 100644 --- a/vdsm/storage/misc.py +++ b/vdsm/storage/misc.py @@ -1344,6 +1344,25 @@ yield respQueue.get()
+def NoIntrCall(fun, *args, **kwargs): + """ + This wrapper is used to handle the interrupt exceptions that might + occur during a system call. + """ + while True: + try: + return fun(*args, **kwargs) + except (IOError, select.error) as e: + if e.args[0] == os.errno.EINTR: + continue + raise + break + + +# NOTE: it would be best to try and unify NoIntrCall and NoIntrPoll. +# We could do so defining a new object that can be used as a placeholer +# for the changing timeout value in the *args/**kwargs. This would +# lead us to rebuilding the function arguments at each loop. def NoIntrPoll(pollfun, timeout=-1): """ This wrapper is used to handle the interrupt exceptions that might occur
-- To view, visit http://gerrit.ovirt.org/11465 To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange Gerrit-Change-Id: I106618a9a61cc96727edaf2e3ab043b2ec28ebe1 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: ovirt-3.2 Gerrit-Owner: Federico Simoncelli fsimonce@redhat.com Gerrit-Reviewer: Dan Kenigsberg danken@redhat.com