Change in vdsm[master]: vdsm: add eventfd and EventFile synchronization
by Federico Simoncelli
Federico Simoncelli has uploaded a new change for review.
Change subject: vdsm: add eventfd and EventFile synchronization
......................................................................
vdsm: add eventfd and EventFile synchronization
Change-Id: I0d237f13c42b1f4505c90d30c6d3c3ecbd1e9fa7
Signed-off-by: Federico Simoncelli <fsimonce(a)redhat.com>
---
M lib/vdsm/Makefile.am
A lib/vdsm/eventfd.py
M tests/Makefile.am
A tests/eventfdTests.py
4 files changed, 251 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/87/33687/1
diff --git a/lib/vdsm/Makefile.am b/lib/vdsm/Makefile.am
index 4bebf28..e712cad 100644
--- a/lib/vdsm/Makefile.am
+++ b/lib/vdsm/Makefile.am
@@ -25,6 +25,7 @@
__init__.py \
compat.py \
define.py \
+ eventfd.py \
exception.py \
ipwrapper.py \
libvirtconnection.py \
diff --git a/lib/vdsm/eventfd.py b/lib/vdsm/eventfd.py
new file mode 100644
index 0000000..b2a7084
--- /dev/null
+++ b/lib/vdsm/eventfd.py
@@ -0,0 +1,140 @@
+#
+# Copyright 2014 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# 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 General Public License for more details.
+#
+# You should have received a copy of the GNU 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
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+"""\
+This module provides the support for eventfd(2).
+
+More information about eventfd and usage examples can be found in the
+eventfd(2) man page.
+
+The EventFile class provides a single synchronization object exposing
+the python Event interface and associated eventfds.
+
+The eventfd() context manager returns a file descriptor that can be
+used to provide the event notice to select, poll and epoll, e.g.
+
+ import os
+ import sys
+ import select
+ import threading
+ import time
+ from vdsm.eventfd import EventFile, DATASIZE
+
+ e = EventFile()
+ p = select.epoll()
+
+ threading.Timer(5, e.set).start()
+
+ with e.eventfd() as efd:
+ p.register(efd, select.EPOLLIN)
+ p.register(sys.stdin.fileno(), select.EPOLLIN)
+
+ print "Echoing lines until event is received"
+ event_received = False
+
+ while not event_received:
+ for fileno, event in p.poll():
+ if not event & select.EPOLLIN:
+ continue
+
+ if fileno == efd:
+ os.read(efd, DATASIZE)
+ event_received = True
+ elif fileno == sys.stdin.fileno():
+ print os.read(sys.stdin.fileno(), 1024),
+
+ print "Event received!"
+
+
+The Event set() semantic is preserved in the eventfd context manager:
+if the event is set then the eventfd already contains the notification.
+This is both to maintain the semantic and to avoid possible races as:
+
+ if not e.is_set():
+ with e.eventfd() as efd:
+ ...
+"""
+
+import os
+import ctypes
+import threading
+
+from contextlib import contextmanager
+
+_libc = ctypes.CDLL('libc.so.6', use_errno=True)
+
+EFD_NONBLOCK = os.O_NONBLOCK
+EFD_CLOEXEC = 02000000 # os.O_CLOEXEC in python 3.3
+EFD_SEMAPHORE = 00000001
+
+DATASIZE = ctypes.sizeof(ctypes.c_ulonglong)
+
+
+def eventfd(initval, flags):
+ return _libc.eventfd(initval, flags)
+
+
+class EventFile(object):
+ def __init__(self, event=None):
+ self.__lock = threading.Lock()
+ self.__fds = set()
+ self.__event = event or threading.Event()
+
+ @staticmethod
+ def __fire_event(fd):
+ os.write(fd, ctypes.c_ulonglong(1))
+
+ def open_eventfd(self):
+ with self.__lock:
+ fd = eventfd(0, 0)
+
+ self.__fds.add(fd)
+
+ if self.__event.is_set():
+ self.__fire_event(fd)
+
+ return fd
+
+ @contextmanager
+ def eventfd(self):
+ fd = self.open_eventfd()
+
+ yield fd
+
+ with self.__lock:
+ self.__fds.remove(fd)
+ os.close(fd)
+
+ def isSet(self):
+ return self.__event.isSet()
+
+ is_set = isSet
+
+ def set(self):
+ with self.__lock:
+ self.__event.set()
+ for fd in self.__fds:
+ self.__fire_event(fd)
+
+ def clear(self):
+ self.__event.clear()
+
+ def wait(self, timeout=None, balancing=True):
+ self.__event.wait(timeout)
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 449d7b1..120712e 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -31,6 +31,7 @@
clientifTests.py \
configNetworkTests.py \
domainMonitorTests.py \
+ eventfdTests.py \
fileVolumeTests.py \
fileUtilTests.py \
fuserTests.py \
diff --git a/tests/eventfdTests.py b/tests/eventfdTests.py
new file mode 100644
index 0000000..be15248
--- /dev/null
+++ b/tests/eventfdTests.py
@@ -0,0 +1,109 @@
+#
+# Copyright 2014 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# 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 General Public License for more details.
+#
+# You should have received a copy of the GNU 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
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+import os
+import select
+from vdsm.eventfd import EventFile, DATASIZE
+from nose.tools import timed, raises, TimeExpired
+
+TEST_TIMEOUT = 1
+WAIT_TIMEOUT = 2
+
+
+def test_set():
+ e = EventFile()
+ e.set()
+ assert e.is_set()
+ assert e.isSet()
+
+
+def text_clear():
+ e = EventFile()
+ e.set()
+ assert e.is_set()
+ e.clear()
+ assert not e.is_set()
+
+
+@timed(TEST_TIMEOUT)
+def test_wait_set():
+ e = EventFile()
+ e.set()
+ e.wait(WAIT_TIMEOUT)
+
+
+@raises(TimeExpired)
+@timed(TEST_TIMEOUT)
+def test_wait_noset():
+ e = EventFile()
+ e.wait(WAIT_TIMEOUT)
+
+
+@timed(TEST_TIMEOUT)
+def test_eventfd_earlyset():
+ e = EventFile()
+ e.set()
+ with e.eventfd() as fd:
+ assert len(__select_and_read(fd)) == DATASIZE
+
+
+@timed(TEST_TIMEOUT)
+def test_eventfd_lateset():
+ e = EventFile()
+ with e.eventfd() as fd:
+ e.set()
+ assert len(__select_and_read(fd)) == DATASIZE
+
+
+@raises(TimeExpired)
+@timed(TEST_TIMEOUT)
+def test_eventfd_noset():
+ e = EventFile()
+ with e.eventfd() as fd:
+ assert len(__select_and_read(fd)) != DATASIZE
+
+
+@timed(TEST_TIMEOUT)
+def test_eventfd_multiple():
+ e = EventFile()
+ e.set()
+ with e.eventfd() as fd1:
+ assert len(__select_and_read(fd1)) == DATASIZE
+ with e.eventfd() as fd2:
+ assert len(__select_and_read(fd2)) == DATASIZE
+ with e.eventfd() as fd3:
+ assert len(__select_and_read(fd3)) == DATASIZE
+
+
+@raises(TimeExpired)
+@timed(TEST_TIMEOUT)
+def test_eventfd_clear():
+ e = EventFile()
+ e.set()
+ e.clear()
+ with e.eventfd() as fd:
+ assert len(__select_and_read(fd)) != DATASIZE
+
+
+def __select_and_read(fd):
+ rd, wr, ex = select.select((fd,), (), (), WAIT_TIMEOUT)
+ if fd in rd:
+ return os.read(fd, DATASIZE)
+ return ''
--
To view, visit http://gerrit.ovirt.org/33687
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I0d237f13c42b1f4505c90d30c6d3c3ecbd1e9fa7
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Federico Simoncelli <fsimonce(a)redhat.com>
7 years, 7 months
Change in vdsm[master]: utils: Add changehash function for change detection
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: utils: Add changehash function for change detection
......................................................................
utils: Add changehash function for change detection
We use Python built-in hash to detect changes in vm state without sending
the state in each response. This function is not suitable for this
usage. Now we use generic utils.changehash(), implemented using md5
hexdigest.
Change-Id: I2242a594383e2d2fe64e3a581f18b8ac662648b0
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M lib/vdsm/utils.py
M vdsm/virt/vm.py
2 files changed, 13 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/45/33045/1
diff --git a/lib/vdsm/utils.py b/lib/vdsm/utils.py
index 23c63e8..1b4a9d5 100644
--- a/lib/vdsm/utils.py
+++ b/lib/vdsm/utils.py
@@ -37,6 +37,7 @@
import glob
import io
import itertools
+import hashlib
import logging
import re
import sys
@@ -1133,3 +1134,13 @@
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
flags |= os.O_NONBLOCK
fcntl.fcntl(fd, fcntl.F_SETFL, flags)
+
+
+def changehash(s):
+ """
+ Returns a hash of string s, suitable for change detection.
+
+ Tipically changehash(s) is sent to client frequently. When a client detect
+ that changehash(s) changed, it ask for s itself, which may be much bigger.
+ """
+ return hashlib.md5(s).hexdigest()
diff --git a/vdsm/virt/vm.py b/vdsm/virt/vm.py
index 941f283..b1567f9 100644
--- a/vdsm/virt/vm.py
+++ b/vdsm/virt/vm.py
@@ -1500,7 +1500,7 @@
self.guestAgent = guestagent.GuestAgent(
self._guestSocketFile, self.cif.channelListener, self.log)
self._lastXMLDesc = '<domain><uuid>%s</uuid></domain>' % self.id
- self._devXmlHash = '0'
+ self._devXmlHash = utils.changehash('')
self._released = False
self._releaseLock = threading.Lock()
self.saveState()
@@ -4495,7 +4495,7 @@
self._lastXMLDesc = self._dom.XMLDesc(0)
devxml = _domParseStr(self._lastXMLDesc).childNodes[0]. \
getElementsByTagName('devices')[0]
- self._devXmlHash = str(hash(devxml.toxml()))
+ self._devXmlHash = utils.changehash(devxml.toxml())
return self._lastXMLDesc
--
To view, visit http://gerrit.ovirt.org/33045
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I2242a594383e2d2fe64e3a581f18b8ac662648b0
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 8 months
Change in vdsm[master]: clusterLock: Acquire, release, and inquire volume leases
by alitke＠redhat.com
Adam Litke has uploaded a new change for review.
Change subject: clusterLock: Acquire, release, and inquire volume leases
......................................................................
clusterLock: Acquire, release, and inquire volume leases
Although we have been creating volume leases on storage for awhile now,
there have been no APIs to manage them. Add this support to
the clusterLock implementations.
Change-Id: Icca901ccd27358767c023cd55b7a3823531d2a5a
Signed-off-by: Adam Litke <alitke(a)redhat.com>
---
M vdsm/storage/clusterlock.py
M vdsm/storage/sd.py
2 files changed, 80 insertions(+), 23 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/22/38622/1
diff --git a/vdsm/storage/clusterlock.py b/vdsm/storage/clusterlock.py
index 51d2906..4077068 100644
--- a/vdsm/storage/clusterlock.py
+++ b/vdsm/storage/clusterlock.py
@@ -73,6 +73,10 @@
"""Raised when the clusterlock class is not supporting inquire"""
+class ResourceLeasesNotSupportedError(Exception):
+ """Raised when the clusterlock class does not support resource leases"""
+
+
class SafeLease(object):
log = logging.getLogger("Storage.SafeLease")
@@ -145,7 +149,13 @@
raise se.AcquireLockFailure(self._sdUUID, rc, out, err)
self.log.debug("Clustered lock acquired successfully")
+ def acquireResource(self, resource, lockDisk, shared=False):
+ raise ResourceLeasesNotSupportedError()
+
def inquire(self):
+ raise InquireNotSupportedError()
+
+ def inquireResource(self, resource, lockDisk):
raise InquireNotSupportedError()
def getLockUtilFullPath(self):
@@ -165,6 +175,8 @@
self.log.debug("Cluster lock released successfully")
+ def releaseResource(self, resource, lockDisk):
+ raise ResourceLeasesNotSupportedError()
initSANLockLog = logging.getLogger("Storage.initSANLock")
@@ -286,13 +298,9 @@
# HOST_FAIL.
return HOST_STATUS_FREE
- # The hostId parameter is maintained here only for compatibility with
- # ClusterLock. We could consider to remove it in the future but keeping it
- # for logging purpose is desirable.
- def acquire(self, hostId):
+ def _acquire(self, resource, lockDisk, shared=False):
with nested(self._lock, SANLock._sanlock_lock):
- self.log.info("Acquiring cluster lock for domain %s (id: %s)",
- self._sdUUID, hostId)
+ self.log.info("Acquiring resource %s, shared=%s", resource, shared)
while True:
if SANLock._sanlock_fd is None:
@@ -304,27 +312,42 @@
"Cannot register to sanlock", str(e))
try:
- sanlock.acquire(self._sdUUID, SDM_LEASE_NAME,
- self.getLockDisk(),
- slkfd=SANLock._sanlock_fd)
+ sanlock.acquire(self._sdUUID, resource, lockDisk,
+ slkfd=SANLock._sanlock_fd, shared=shared)
except sanlock.SanlockException as e:
if e.errno != os.errno.EPIPE:
raise se.AcquireLockFailure(
self._sdUUID, e.errno,
- "Cannot acquire cluster lock", str(e))
+ "Cannot acquire sanlock resource", str(e))
SANLock._sanlock_fd = None
continue
break
- self.log.debug("Cluster lock for domain %s successfully acquired "
- "(id: %s)", self._sdUUID, hostId)
+ self.log.debug("Resource %s successfully acquired", resource)
+
+ # The hostId parameter is maintained here only for compatibility with
+ # ClusterLock. We could consider to remove it in the future but keeping it
+ # for logging purpose is desirable.
+ def acquire(self, hostId):
+ self.log.info("Acquiring cluster lock for domain %s (id: %s)",
+ self._sdUUID, hostId)
+ self._acquire(SDM_LEASE_NAME, self.getLockDisk())
+ self.log.debug("Cluster lock for domain %s successfully acquired "
+ "(id: %s)", self._sdUUID, hostId)
+
+ def acquireResource(self, resource, lockDisk, shared=False):
+ self._acquire(resource, lockDisk, shared)
+ res, owners = self._inquire(resource, lockDisk)
+ self.log.debug("ALITKE: acquire: res:%s owners:%s", res, owners)
+
+ def _inquire(self, resource, lockDisk):
+ res = sanlock.read_resource(*lockDisk[0])
+ owners = sanlock.read_resource_owners(self._sdUUID, resource, lockDisk)
+ return res, owners
def inquire(self):
- resource = sanlock.read_resource(self._leasesPath, SDM_LEASE_OFFSET)
- owners = sanlock.read_resource_owners(self._sdUUID, SDM_LEASE_NAME,
- self.getLockDisk())
-
+ resource, owners = self._inquire(SDM_LEASE_NAME, self.getLockDisk())
if len(owners) == 1:
return resource.get("version"), owners[0].get("host_id")
elif len(owners) > 1:
@@ -334,19 +357,32 @@
return None, None
- def release(self):
+ def inquireResource(self, resource, lockDisk):
+ resource, owners = self._inquire(SDM_LEASE_NAME, self.getLockDisk())
+ return (resource.get("version"),
+ [owner.get("host_id") for owner in owners])
+
+ def _release(self, resource, lockDisk):
with self._lock:
- self.log.info("Releasing cluster lock for domain %s", self._sdUUID)
+ self.log.info("Releasing resource %s", resource)
try:
- sanlock.release(self._sdUUID, SDM_LEASE_NAME,
- self.getLockDisk(), slkfd=SANLock._sanlock_fd)
+ sanlock.release(self._sdUUID, resource, lockDisk,
+ slkfd=SANLock._sanlock_fd)
except sanlock.SanlockException as e:
- raise se.ReleaseLockFailure(self._sdUUID, e)
+ raise se.ReleaseLockFailure(resource, e)
self._sanlockfd = None
- self.log.debug("Cluster lock for domain %s successfully released",
- self._sdUUID)
+ self.log.debug("Resource %s successfully released", resource)
+
+ def release(self):
+ self.log.info("Releasing cluster lock for domain %s", self._sdUUID)
+ self._release(SDM_LEASE_NAME, self.getLockDisk())
+ self.log.debug("Cluster lock for domain %s successfully released",
+ self._sdUUID)
+
+ def releaseResource(self, resource, lockDisk):
+ self._release(resource, lockDisk)
class LocalLock(object):
@@ -462,10 +498,16 @@
self.log.debug("Local lock for domain %s successfully acquired "
"(id: %s)", self._sdUUID, hostId)
+ def acquireResource(self, resource, lockDisk, shared=False):
+ raise ResourceLeasesNotSupportedError()
+
def inquire(self):
with self._globalLockMapSync:
hostId, lockFile = self._getLease()
return self.LVER, hostId if lockFile else None, None
+
+ def inquireResource(self, resource, lockDisk):
+ raise InquireNotSupportedError()
def release(self):
with self._globalLockMapSync:
@@ -483,3 +525,6 @@
self.log.debug("Local lock for domain %s successfully released",
self._sdUUID)
+
+ def releaseResource(self, resource, lockDisk):
+ raise ResourceLeasesNotSupportedError()
\ No newline at end of file
diff --git a/vdsm/storage/sd.py b/vdsm/storage/sd.py
index a771abf..9e6f281 100644
--- a/vdsm/storage/sd.py
+++ b/vdsm/storage/sd.py
@@ -514,6 +514,18 @@
def inquireClusterLock(self):
return self._clusterLock.inquire()
+ def acquireVolumeLease(self, imgUUID, volUUID, shared=False):
+ lockDisk = self.getVolumeLease(imgUUID, volUUID)
+ self._clusterLock.acquireResource(volUUID, [lockDisk], shared)
+
+ def releaseVolumeLease(self, imgUUID, volUUID):
+ lockDisk = self.getVolumeLease(imgUUID, volUUID)
+ self._clusterLock.releaseResource(volUUID, [lockDisk])
+
+ def inquireVolumeLease(self, imgUUID, volUUID):
+ lockDisk = self.getVolumeLease(imgUUID, volUUID)
+ return self._clusterLock.inquireResource(volUUID, [lockDisk])
+
def attach(self, spUUID):
self.invalidateMetadata()
pools = self.getPools()
--
To view, visit https://gerrit.ovirt.org/38622
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Icca901ccd27358767c023cd55b7a3823531d2a5a
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Adam Litke <alitke(a)redhat.com>
7 years, 8 months
Change in vdsm[master]: vm: Cleaner vm stats thread initialization
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: vm: Cleaner vm stats thread initialization
......................................................................
vm: Cleaner vm stats thread initialization
The vm stats thread was initialized too late, leading to unneeded checks
and unclear try except blocks when trying to stop the thread.
This patch changes AdvancedStatsThread so it keeps a thread instance
instead of inheriting from Thread, and initialize it in Vm.__init__, so
it is always available when we access it.
Change-Id: I2917de42b76ee3dc8b27bdc23b33f3c984a7cc68
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M vdsm/virt/sampling.py
M vdsm/virt/vm.py
2 files changed, 11 insertions(+), 18 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/99/39299/1
diff --git a/vdsm/virt/sampling.py b/vdsm/virt/sampling.py
index 3206b97..848bb36 100644
--- a/vdsm/virt/sampling.py
+++ b/vdsm/virt/sampling.py
@@ -401,7 +401,7 @@
return self._samples.last()
-class AdvancedStatsThread(threading.Thread):
+class AdvancedStatsThread(object):
"""
A thread that runs the registered AdvancedStatsFunction objects
for statistic and monitoring purpose.
@@ -412,9 +412,8 @@
"""
Initialize an AdvancedStatsThread object
"""
- threading.Thread.__init__(self)
self.daemon = daemon
-
+ self._thread = None
self._log = log
self._stopEvent = threading.Event()
self._contEvent = threading.Event()
@@ -426,7 +425,7 @@
"""
Register the functions listed as arguments
"""
- if self.isAlive():
+ if self._thread is not None:
raise RuntimeError("AdvancedStatsThread is started")
for statsFunction in args:
@@ -437,7 +436,9 @@
Start the execution of the thread and exit
"""
self._log.debug("Start statistics collection")
- threading.Thread.start(self)
+ self._thread = threading.Thread(target=self._run)
+ self._thread.daemon = self.daemon
+ self._thread.start()
def stop(self):
"""
@@ -464,7 +465,7 @@
def getLastSampleTime(self):
return self._statsTime
- def run(self):
+ def _run(self):
self._log.debug("Stats thread started")
self._contEvent.set()
diff --git a/vdsm/virt/vm.py b/vdsm/virt/vm.py
index 28054bf..1bc04e7 100644
--- a/vdsm/virt/vm.py
+++ b/vdsm/virt/vm.py
@@ -779,7 +779,7 @@
self._initTimeRTC = long(self.conf.get('timeOffset', 0))
self._guestEvent = vmstatus.POWERING_UP
self._guestEventTime = 0
- self._vmStats = None
+ self._vmStats = VmStatsThread(self)
self._guestCpuRunning = False
self._guestCpuLock = threading.Lock()
self._startTime = time.time() - \
@@ -1269,7 +1269,7 @@
return
toSave = self.status()
toSave['startTime'] = self._startTime
- if self.lastStatus != vmstatus.DOWN and self._vmStats:
+ if self.lastStatus != vmstatus.DOWN:
guestInfo = self.guestAgent.getGuestInfo()
toSave['username'] = guestInfo['username']
toSave['guestIPs'] = guestInfo['guestIPs']
@@ -1758,7 +1758,7 @@
decStats = {}
try:
- if self._vmStats and self._vmStats.getLastSampleTime() is not None:
+ if self._vmStats.getLastSampleTime() is not None:
decStats = self._vmStats.get()
self._setUnresponsiveIfTimeout(
stats,
@@ -2001,19 +2001,11 @@
return domxml.toxml()
def startVmStats(self):
- self._vmStats = VmStatsThread(self)
self._vmStats.start()
self._guestEventTime = self._startTime
def stopVmStats(self):
- # this is less clean that it could be, but we can get here from
- # many flows and with various locks held
- # (_releaseLock, _shutdownLock)
- # _vmStats may be None already, and we're good with that.
- try:
- self._vmStats.stop()
- except AttributeError:
- pass
+ self._vmStats.stop()
@staticmethod
def _guestSockCleanup(sock):
--
To view, visit https://gerrit.ovirt.org/39299
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I2917de42b76ee3dc8b27bdc23b33f3c984a7cc68
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 8 months
Change in vdsm[master]: storage: export volume lease state in getVolumeInfo
by alitke＠redhat.com
Adam Litke has uploaded a new change for review.
Change subject: storage: export volume lease state in getVolumeInfo
......................................................................
storage: export volume lease state in getVolumeInfo
In order to support an entity-based polling methodology (ie. in the new
SDM verbs) we must know whether a volume is currently locked by a long
running operation. Extend the getVolumeInfo API to report whether the
lease is free or held.
Change-Id: I55f062a4be15593fdc98518fd0a113976cbe0ae7
Signed-off-by: Adam Litke <alitke(a)redhat.com>
---
M vdsm/rpc/vdsmapi-schema.json
M vdsm/storage/volume.py
2 files changed, 31 insertions(+), 1 deletion(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/23/38623/1
diff --git a/vdsm/rpc/vdsmapi-schema.json b/vdsm/rpc/vdsmapi-schema.json
index 988ce64..cba1008 100644
--- a/vdsm/rpc/vdsmapi-schema.json
+++ b/vdsm/rpc/vdsmapi-schema.json
@@ -7897,6 +7897,19 @@
{'enum': 'VolumeStatus', 'data': ['OK', 'INVALID', 'ILLEGAL']}
##
+# @VolumeLeaseState:
+#
+# An enumeration of Volume lease states.
+#
+# @FREE: The lease is free
+#
+# @HELD: The lease is held
+#
+# Since: 4.10.0
+##
+{'enum': 'VolumeLeaseState', 'data': ['FREE', 'HELD']}
+
+##
# @VolumeInfo:
#
# Information about a Volume.
@@ -7937,6 +7950,8 @@
#
# @children: A list of decendent Volumes that depend on this Volume
#
+# @leaseState: #optional The state of the volume lease
+#
# Since: 4.10.0
##
{'type': 'VolumeInfo',
@@ -7946,7 +7961,8 @@
'description': 'str', 'pool': 'UUID', 'domain': 'UUID',
'image': 'UUID', 'ctime': 'int', '*mtime': 'uint',
'legality': 'VolumeLegality', 'apparentsize': 'uint',
- 'truesize': 'uint', 'status': 'VolumeStatus', 'children': ['UUID']}}
+ 'truesize': 'uint', 'status': 'VolumeStatus', 'children': ['UUID'],
+ '*leaseState': 'VolumeLeaseState'}}
##
# @Volume.getInfo:
diff --git a/vdsm/storage/volume.py b/vdsm/storage/volume.py
index 980b6e3..2e16efc 100644
--- a/vdsm/storage/volume.py
+++ b/vdsm/storage/volume.py
@@ -33,6 +33,7 @@
import fileUtils
import task
from threadLocal import vars
+from clusterlock import InquireNotSupportedError
import resourceFactories
import resourceManager as rm
rmanager = rm.ResourceManager.getInstance()
@@ -85,6 +86,10 @@
ILLEGAL_VOL = "ILLEGAL"
LEGAL_VOL = "LEGAL"
FAKE_VOL = "FAKE"
+
+# Volume lease states
+LEASE_FREE = "FREE"
+LEASE_HELD = "HELD"
log = logging.getLogger('Storage.Volume')
@@ -839,6 +844,11 @@
cls.createMetadata(metaId, meta)
return meta
+ def getLeaseState(self):
+ ver, owners = sdCache.produce(self.sdUUID).inquireVolumeLease(
+ self.imgUUID, self.volUUID)
+ return LEASE_HELD if owners else LEASE_FREE
+
def getInfo(self):
"""
Get volume info
@@ -857,6 +867,10 @@
info['apparentsize'] = str(vsize)
info['truesize'] = str(avsize)
info['status'] = "OK"
+ try:
+ info['leaseState'] = self.getLeaseState()
+ except InquireNotSupportedError:
+ pass
except se.StorageException as e:
self.log.debug("exception: %s:%s" % (str(e.message), str(e.value)))
info['apparentsize'] = "0"
--
To view, visit https://gerrit.ovirt.org/38623
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I55f062a4be15593fdc98518fd0a113976cbe0ae7
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Adam Litke <alitke(a)redhat.com>
7 years, 8 months
Change in vdsm[master]: log: Use INFO log level as default
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: log: Use INFO log level as default
......................................................................
log: Use INFO log level as default
The current logs are much too verbose which cause trouble for users, and
make us look unprofessional. Mature project should not use debug log by
default.
To debug issues that are not clear enough using INFO logs, the relevant
logger level can be modified on a user machine as needed.
Change-Id: I767dcd9bad7b9fbeebb438e9ef13cb0ec3f042ee
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M vdsm/logger.conf.in
1 file changed, 4 insertions(+), 4 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/04/32504/1
diff --git a/vdsm/logger.conf.in b/vdsm/logger.conf.in
index 64b154f..8e963dd 100644
--- a/vdsm/logger.conf.in
+++ b/vdsm/logger.conf.in
@@ -8,18 +8,18 @@
keys=long,simple,none,sysform
[logger_root]
-level=DEBUG
+level=INFO
handlers=syslog,logfile
propagate=0
[logger_vds]
-level=DEBUG
+level=INFO
handlers=syslog,logfile
qualname=vds
propagate=0
[logger_Storage]
-level=DEBUG
+level=INFO
handlers=logfile
qualname=Storage
propagate=0
@@ -31,7 +31,7 @@
propagate=1
[logger_connectivity]
-level=DEBUG
+level=INFO
handlers=connlogfile
qualname=connectivity
propagate=0
--
To view, visit http://gerrit.ovirt.org/32504
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I767dcd9bad7b9fbeebb438e9ef13cb0ec3f042ee
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 9 months
Change in vdsm[master]: fc-scan: Use utilities from vdsm library.
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: fc-scan: Use utilities from vdsm library.
......................................................................
fc-scan: Use utilities from vdsm library.
Replace low level threading code with simpler concurrent.tmap() call and
duplicate monotonic_time() with utils.monotonic_time().
Change-Id: Ic48748d6a43d41e034e16cb4f636ebe627881590
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M vdsm/storage/fc-scan
1 file changed, 24 insertions(+), 46 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/66/38466/1
diff --git a/vdsm/storage/fc-scan b/vdsm/storage/fc-scan
index 344345d..c746ea4 100755
--- a/vdsm/storage/fc-scan
+++ b/vdsm/storage/fc-scan
@@ -38,43 +38,11 @@
import logging
import os
import sys
-import threading
+
+from vdsm import concurrent
+from vdsm import utils
log = logging.getLogger("fc-scan")
-
-
-class Scan(object):
-
- def __init__(self, host):
- self.host = host
- self.succeeded = False
- self.thread = None
-
- def start(self):
- self.thread = threading.Thread(target=self.run)
- self.thread.daemon = True
- self.thread.start()
-
- def wait(self):
- self.thread.join()
-
- def run(self):
- try:
- path = "/sys/class/scsi_host/%s/scan" % self.host
- log.debug("Scanning %s", path)
- start = monotonic_time()
- fd = os.open(path, os.O_WRONLY)
- try:
- os.write(fd, "- - -")
- finally:
- os.close(fd)
- self.succeeded = True
- elapsed = monotonic_time() - start
- log.debug("Scanned %s in %.2f seconds", path, elapsed)
- except OSError as e:
- log.error("Scanning %s failed: %s", path, e)
- except Exception:
- log.exception("Scanning %s failed", path)
def main(args):
@@ -93,22 +61,32 @@
log.debug("No fc_host found")
return 0
- scans = []
-
- for host in hosts:
- s = Scan(host)
- s.start()
- scans.append(s)
-
- for s in scans:
- s.wait()
+ scans = concurrent.tmap(scan_host, hosts)
if not all(s.succeeded for s in scans):
return 1
+ return 0
-def monotonic_time():
- return os.times()[4]
+
+def scan_host(name):
+ try:
+ path = "/sys/class/scsi_host/%s/scan" % name
+ log.debug("Scanning %s", path)
+ start = utils.monotonic_time()
+ fd = os.open(path, os.O_WRONLY)
+ try:
+ os.write(fd, "- - -")
+ finally:
+ os.close(fd)
+ elapsed = utils.monotonic_time() - start
+ log.debug("Scanned %s in %.2f seconds", path, elapsed)
+ except OSError as e:
+ log.error("Scanning %s failed: %s", path, e)
+ raise
+ except Exception:
+ log.exception("Scanning %s failed", path)
+ raise
if __name__ == '__main__':
--
To view, visit https://gerrit.ovirt.org/38466
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic48748d6a43d41e034e16cb4f636ebe627881590
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 9 months
Change in vdsm[master]: lib: Revert and refine error handling in tmap()
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: lib: Revert and refine error handling in tmap()
......................................................................
lib: Revert and refine error handling in tmap()
In commit 2b7155b696 (lib: Simplify and generalize concurrent.tmap()),
we simplified error handling by returning a named tuple with function
results. This turned out less useful then the original error handling.
This patch returns the previous error handling:
- Functions passed to tmap() should not raise - if they raise, this is
considered a bug in the function.
- The last error is raised by tmap() instead of returning the result.
This make it easier to fail loudly for unexpected errors.
- The original exception is re-raised now with the original traceback.
- Error handling is documented properly now
Previously you had to make sure function raises to signal failures:
def func():
try:
code that should not fail...
code that may fail...
code that should not fail...
except ExpectedError:
log.error(...)
raise
except Exception:
log.exception(...)
raise
results = concurrent.tmap(func, values)
if not all(r.succeeded for r in results):
...
Returning the result as is lets us have nicer code:
def func():
code that should not fail...
try:
code that may fail...
except ExpectedError:
log.error(...)
return False
code that should not fail...
return True
succeeded = concurrent.tmap(func, values)
if not all(succeeded):
...
We can ignore unexpected errors, since tmap() will log them and fail
loudly. We can also minimize try except block for expected errors.
Change-Id: I0154b28ff7822c63e77181bbbf444c712bd0c31e
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M lib/vdsm/concurrent.py
M tests/concurrentTests.py
2 files changed, 45 insertions(+), 19 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/11/39211/1
diff --git a/lib/vdsm/concurrent.py b/lib/vdsm/concurrent.py
index 64e072d..5498052 100644
--- a/lib/vdsm/concurrent.py
+++ b/lib/vdsm/concurrent.py
@@ -18,22 +18,42 @@
# Refer to the README and COPYING files for full details of the license
#
+import logging
import threading
-from collections import namedtuple
-
-
-Result = namedtuple("Result", ["succeeded", "value"])
+import sys
def tmap(func, iterable):
+ """
+ Run func with arguments from iterable in multiple threads, a returning the
+ output in order of arguments.
+
+ func should not raise exceptions - we consider this a bug in func, and will
+ fail the call and re-raise the exception in the caller thread.
+
+ Expected exceptions should be handled in func. If the caller likes to
+ handle the error later, func should return it:
+
+ def func(value):
+ try:
+ return something(value)
+ except ExpectedError as e:
+ return e
+
+ Unexpected exceptions should not be handled, as they are logged in the
+ worker threads and re-raised in the caller thread. If multiple excpetions
+ raised, only the last one will be re-raised in the caller thread.
+ """
args = list(iterable)
results = [None] * len(args)
+ error = [None]
def worker(i, f, arg):
try:
- results[i] = Result(True, f(arg))
- except Exception as e:
- results[i] = Result(False, e)
+ results[i] = f(arg)
+ except Exception:
+ error[0] = sys.exc_info()
+ logging.exception("Unhandled exception in tmap worker thread")
threads = []
for i, arg in enumerate(args):
@@ -45,4 +65,8 @@
for t in threads:
t.join()
+ if error[0] is not None:
+ t, v, tb = error[0]
+ raise t, v, tb
+
return results
diff --git a/tests/concurrentTests.py b/tests/concurrentTests.py
index 307e397..5c0646b 100644
--- a/tests/concurrentTests.py
+++ b/tests/concurrentTests.py
@@ -26,13 +26,16 @@
from vdsm import concurrent
+class Error(Exception):
+ pass
+
+
class TMapTests(VdsmTestCase):
def test_results(self):
values = tuple(range(10))
results = concurrent.tmap(lambda x: x, values)
- expected = [concurrent.Result(True, x) for x in values]
- self.assertEqual(results, expected)
+ self.assertEqual(results, list(values))
def test_results_order(self):
def func(x):
@@ -40,8 +43,7 @@
return x
values = tuple(random.random() * 0.1 for x in range(10))
results = concurrent.tmap(func, values)
- expected = [concurrent.Result(True, x) for x in values]
- self.assertEqual(results, expected)
+ self.assertEqual(results, list(values))
def test_concurrency(self):
start = time.time()
@@ -49,12 +51,12 @@
elapsed = time.time() - start
self.assertTrue(0.1 < elapsed < 0.2)
- def test_error(self):
- error = RuntimeError("No result for you!")
-
+ def test_raise_last_error(self):
def func(x):
- raise error
-
- results = concurrent.tmap(func, range(10))
- expected = [concurrent.Result(False, error)] * 10
- self.assertEqual(results, expected)
+ raise Error(x)
+ try:
+ concurrent.tmap(func, (1, 2, 3))
+ except Error as e:
+ self.assertEqual(e.args, (3,))
+ else:
+ self.fail("Exception was not raised")
--
To view, visit https://gerrit.ovirt.org/39211
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I0154b28ff7822c63e77181bbbf444c712bd0c31e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 9 months
Change in vdsm[master]: misc: Safer and simpler itmap
by Nir Soffer
Nir Soffer has uploaded a new change for review.
Change subject: misc: Safer and simpler itmap
......................................................................
misc: Safer and simpler itmap
The previous code had few issues:
- It used unlimited number of threads by default. This may lead to
creation of 100's of threads if you do not specify a value.
- It used non-daemon threads, which could lead to unwanted delay during
vdsm shutdown.
- It tried to yield results before all arguments were handled. This
could lead to unwanted delay in argument processing, if the caller
would block processing the results.
- It started one thread per value, even if maxthreads was smaller than
number of values.
- It was too complicated.
Changes:
- The caller must specify the maximum number of threads.
- Use daemon threads
- Queue all values before yielding results
- Start up to maxthreads worker threads, each processing multiple values
- Simplify the code
- Add test for error handling
Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M tests/miscTests.py
M vdsm/storage/misc.py
2 files changed, 42 insertions(+), 42 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/19/39119/1
diff --git a/tests/miscTests.py b/tests/miscTests.py
index 31f64fa..4b3e3c3 100644
--- a/tests/miscTests.py
+++ b/tests/miscTests.py
@@ -196,7 +196,7 @@
# outOfProcess operation + 1. it let us know that oop and itmap operate
# properly with their limitations
data = frozenset(range(oop.HELPERS_PER_DOMAIN + 1))
- ret = frozenset(misc.itmap(dummy, data, misc.UNLIMITED_THREADS))
+ ret = frozenset(misc.itmap(dummy, data, len(data)))
self.assertEquals(ret, data)
def testMoreThreadsThanArgs(self):
@@ -207,6 +207,13 @@
data = 1
self.assertRaises(ValueError, misc.itmap(int, data, 0).next)
+ def testErrors(self):
+ err = Exception()
+ def dummy(arg):
+ raise err
+ data = [1, 2, 3]
+ self.assertEqual(list(misc.itmap(dummy, data, 4)), [err] * len(data))
+
class RotateFiles(TestCaseBase):
diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py
index eb484c7..463fd04 100644
--- a/vdsm/storage/misc.py
+++ b/vdsm/storage/misc.py
@@ -58,7 +58,6 @@
STR_UUID_SIZE = 36
UUID_HYPHENS = [8, 13, 18, 23]
MEGA = 1 << 20
-UNLIMITED_THREADS = -1
log = logging.getLogger('Storage.Misc')
@@ -882,53 +881,47 @@
raise exception
-def itmap(func, iterable, maxthreads=UNLIMITED_THREADS):
+def itmap(func, iterable, maxthreads):
"""
- Make an iterator that computes the function using
- arguments from the iterable. It works similar to tmap
- by running each operation in a different thread, this
- causes the results not to return in any particular
- order so it's good if you don't care about the order
- of the results.
- maxthreads stands for maximum threads that we can initiate simultaneosly.
- If we reached to max threads the function waits for thread to
- finish before initiate the next one.
+ Return an iterator calling func with arguments from iterable in multiple threads.
+
+ Unlike tmap, the results are not returned in the original order of the
+ arguments, and number of threads is limited to maxthreads.
"""
- if maxthreads < 1 and maxthreads != UNLIMITED_THREADS:
- raise ValueError("Wrong input to function itmap: %s", maxthreads)
+ if maxthreads < 1:
+ raise ValueError("Invalid maxthreads value: %s" % maxthreads)
- respQueue = Queue.Queue()
+ DONE = object()
+ values = Queue.Queue()
+ results = Queue.Queue()
- def wrapper(value):
- try:
- respQueue.put(func(value))
- except Exception as e:
- respQueue.put(e)
+ def worker():
+ while True:
+ value = values.get()
+ if value is DONE:
+ return
+ try:
+ results.put(func(value))
+ except Exception as e:
+ results.put(e)
- threadsCount = 0
- for arg in iterable:
- if maxthreads != UNLIMITED_THREADS:
- if maxthreads == 0:
- # This not supposed to happened. If it does, it's a bug.
- # maxthreads should get to 0 only after threadsCount is
- # greater than 1
- if threadsCount < 1:
- raise RuntimeError("No thread initiated")
- else:
- yield respQueue.get()
- # if yield returns one thread stopped, so we can run
- # another thread in queue
- maxthreads += 1
- threadsCount -= 1
+ count = 0
+ threads = 0
- t = threading.Thread(target=wrapper, args=(arg,))
- t.start()
- threadsCount += 1
- maxthreads -= 1
+ for value in iterable:
+ values.put(value)
+ count += 1
+ if threads < maxthreads:
+ t = threading.Thread(target=worker)
+ t.daemon = True
+ t.start()
+ threads += 1
- # waiting for rest threads to end
- for i in xrange(threadsCount):
- yield respQueue.get()
+ for _ in range(threads):
+ values.put(DONE)
+
+ for _ in xrange(count):
+ yield results.get()
def isAscii(s):
--
To view, visit https://gerrit.ovirt.org/39119
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
7 years, 9 months
Change in vdsm[master]: storage: Make Image.__chainSizeCalc public
by alitke＠redhat.com
Adam Litke has uploaded a new change for review.
Change subject: storage: Make Image.__chainSizeCalc public
......................................................................
storage: Make Image.__chainSizeCalc public
The new SDM copyVolumeData wants to make use of the same logic being
used by the classic copy flows to extend the size of the target volume
to the appropriate size. Make Image.__chainSizeCalc public so it can be
accessed from the SDM code.
Change-Id: Id079eb5067c16f934370e42b5f4e09bbcef1512b
Signed-off-by: Adam Litke <alitke(a)redhat.com>
---
M vdsm/storage/image.py
1 file changed, 2 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/95/38995/1
diff --git a/vdsm/storage/image.py b/vdsm/storage/image.py
index 791b48c..8d5e8c2 100644
--- a/vdsm/storage/image.py
+++ b/vdsm/storage/image.py
@@ -140,7 +140,7 @@
randomStr = misc.randomStr(RENAME_RANDOM_STRING_LEN)
return "%s%s_%s" % (sd.REMOVED_IMAGE_PREFIX, randomStr, uuid)
- def __chainSizeCalc(self, sdUUID, imgUUID, volUUID, size):
+ def chainSizeCalc(self, sdUUID, imgUUID, volUUID, size):
"""
Compute an estimate of the whole chain size
using the sum of the actual size of the chain's volumes
@@ -763,7 +763,7 @@
if volParams['volFormat'] != volume.COW_FORMAT or \
volParams['prealloc'] != volume.SPARSE_VOL:
raise se.IncorrectFormat(self)
- volParams['apparentsize'] = self.__chainSizeCalc(
+ volParams['apparentsize'] = self.chainSizeCalc(
sdUUID, srcImgUUID, srcVolUUID, volParams['size'])
# Find out dest volume parameters
--
To view, visit https://gerrit.ovirt.org/38995
To unsubscribe, visit https://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Id079eb5067c16f934370e42b5f4e09bbcef1512b
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Adam Litke <alitke(a)redhat.com>
7 years, 9 months