Hello Dan Kenigsberg,
I'd like you to do a code review. Please visit
http://gerrit.ovirt.org/11465
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(a)redhat.com>
Reviewed-on:
http://gerrit.ovirt.org/10282
Reviewed-by: Dan Kenigsberg <danken(a)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(a)redhat.com>
Gerrit-Reviewer: Dan Kenigsberg <danken(a)redhat.com>