Vinzenz Feenstra has uploaded a new change for review.
Change subject: Added support for reporting the timezone and os info
......................................................................
Added support for reporting the timezone and os info
The guest agent reports now the timezone and more detailed os information.
This patch introduces the support for those new messages.
Change-Id: Iadb85a5003fcb4693b5a1174655acf59193340b5
Guest-Agent-API-Version: 2
Bug-Url: https://bugzilla.redhat.com/869296
Signed-off-by: Vinzenz Feenstra <vfeenstr(a)redhat.com>
---
M vdsm/rpc/vdsmapi-schema.json
M vdsm/virt/guestagent.py
2 files changed, 50 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/40/28940/1
diff --git a/vdsm/rpc/vdsmapi-schema.json b/vdsm/rpc/vdsmapi-schema.json
index 2ccf963..869b78c 100644
--- a/vdsm/rpc/vdsmapi-schema.json
+++ b/vdsm/rpc/vdsmapi-schema.json
@@ -3456,6 +3456,12 @@
#
# @clientIp: The IP address of the client connected to the display
#
+# @guestOsInfo: #optional Guest OS information like architecture,
+# type, and version (Since 4.15.0)
+#
+# @guestTimezone: #optional Timezone and time offset of the guest OS
+# (Since 4.15.0)
+#
# Since: 4.14.1
##
{'type': 'VMFullInfo',
@@ -3470,7 +3476,45 @@
'guestFQDN': 'str', 'displayIp': 'str', 'keyboardLayout': 'str',
'displayPort': 'uint', 'guestIPs': 'str', 'smartcardEnable': 'bool',
'nicModel': 'VmInterfaceDeviceModel', 'pitReinjection': 'bool',
- 'status': 'str', 'clientIp': 'str'}}
+ 'status': 'str', 'clientIp': 'str', '*guestOsInfo': 'VMGuestOsInfo',
+ '*guestTimezone': 'VMGuestTimezone'}}
+
+##
+# @VMGuestTimezone:
+#
+# Timezone configuration of the guest operating system
+#
+# @zone: Name of the timezone native to the Guest OS
+#
+# @offset: Time offset of the configured timezone in minutes
+#
+# Since: 4.15.0
+##
+{'type': 'VMGuestTimezone',
+ 'data': {'zone': 'string', 'offset': 'uint'}}
+
+##
+# @VMGuestOsInfo:
+#
+# Information about the guest operating system on a VM
+#
+# @version: Operating system version
+#
+# @distribution: Name of the linux distribution
+#
+# @codename: Codename of the OS
+#
+# @arch: Architecture of the OS (x86, x86_64, ppc64, ...)
+#
+# @type: Type of the OS (linux, windows, ...)
+#
+# @kernel: Kernel version reported by a linux guest
+#
+# Since: 4.15.0
+##
+{'type': 'VMGuestOsInfo',
+ 'data': {'version': 'string', 'distribution': 'string', 'codename': 'string',
+ 'arch': 'string', 'type': 'string', 'kernel': 'string'}}
##
# @Host.getVMFullList:
diff --git a/vdsm/virt/guestagent.py b/vdsm/virt/guestagent.py
index 31c0a26..8ff262f 100644
--- a/vdsm/virt/guestagent.py
+++ b/vdsm/virt/guestagent.py
@@ -30,7 +30,7 @@
from . import vmstatus
-_MAX_SUPPORTED_API_VERSION = 1
+_MAX_SUPPORTED_API_VERSION = 2
_IMPLICIT_API_VERSION_ZERO = 0
_MESSAGE_API_VERSION_LOOKUP = {
@@ -255,6 +255,10 @@
self.guestInfo['guestName'] = args['name']
elif message == 'os-version':
self.guestInfo['guestOs'] = args['version']
+ elif message == 'os-info':
+ self.guestInfo['guestOsInfo'] = args
+ elif message == 'timezone':
+ self.guestInfo['guestTimezone'] = args
elif message == 'network-interfaces':
interfaces = []
old_ips = ''
--
To view, visit http://gerrit.ovirt.org/28940
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Iadb85a5003fcb4693b5a1174655acf59193340b5
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Vinzenz Feenstra <vfeenstr(a)redhat.com>
Vinzenz Feenstra has uploaded a new change for review.
Change subject: vm: Automatically add a serial port for a console device
......................................................................
vm: Automatically add a serial port for a console device
Without a serial device the console support does not work for
linux systems. We switch from virtio to serial and append a serial
port if a console device was defined.
Change-Id: Ifa7b02a7bcaad63017c35c811a194fa42e2b694f
Signed-off-by: Vinzenz Feenstra <vfeenstr(a)redhat.com>
---
M vdsm/vm.py
1 file changed, 14 insertions(+), 1 deletion(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/79/25979/1
diff --git a/vdsm/vm.py b/vdsm/vm.py
index 9cb0e82..7ce7c59 100644
--- a/vdsm/vm.py
+++ b/vdsm/vm.py
@@ -1171,6 +1171,18 @@
commandLine.appendChildWithArgs('qemu:arg', value='keyboard')
self.dom.appendChild(commandLine)
+ def appendSerial(self):
+ """
+ Add a serial port for the console device if it exists
+ <serial type='pty'>
+ <target port='0'>
+ </serial>
+ """
+ if len(self._devices.getElementsByTagName('console')) == 1:
+ s = XMLElement('serial', type='pty')
+ s.appendChildWithArgs('target', port='0')
+ self._devices.appendChild(s)
+
def appendGraphics(self):
"""
Add graphics section to domain xml.
@@ -1888,7 +1900,7 @@
</console>
"""
m = self.createXmlElem('console', 'pty')
- m.appendChildWithArgs('target', type='virtio', port='0')
+ m.appendChildWithArgs('target', type='serial', port='0')
return m
@@ -3025,6 +3037,7 @@
domxml.appendEmulator()
self._appendDevices(domxml)
+ domxml.appendSerial()
for drive in self._devices[DISK_DEVICES][:]:
for leaseElement in drive.getLeasesXML():
--
To view, visit http://gerrit.ovirt.org/25979
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ifa7b02a7bcaad63017c35c811a194fa42e2b694f
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Vinzenz Feenstra <vfeenstr(a)redhat.com>
Antoni Segura Puimedon has uploaded a new change for review.
Change subject: net_tc: add support for mirroring all (not just IPv4) traffic
......................................................................
net_tc: add support for mirroring all (not just IPv4) traffic
Up until now the filter was only acting on packets with IPv4 headers
because of the 'protocol ip' part of the tc filter. This makes the
mirroring protocol agnostic.
Change-Id: I9de381e67fae4cb79fe41dbd9d1b60b72eb84de4
Signed-off-by: Antoni S. Puimedon <asegurap(a)redhat.com>
---
M vdsm/network/tc.py
1 file changed, 2 insertions(+), 2 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/88/27388/1
diff --git a/vdsm/network/tc.py b/vdsm/network/tc.py
index 35c4a92..de1837a 100644
--- a/vdsm/network/tc.py
+++ b/vdsm/network/tc.py
@@ -127,7 +127,7 @@
if filt.prio:
command.extend(['prio', filt.prio])
command.extend(['handle', filt.handle])
- command.extend(['protocol', 'ip', 'u32', 'match', 'u8', '0', '0'])
+ command.extend(['protocol', 'all', 'u32', 'match', 'u8', '0', '0'])
for a in filt.actions:
command.extend(['action', 'mirred', 'egress', 'mirror',
'dev', a.target])
@@ -203,7 +203,7 @@
out = _process_request([EXT_TC, 'filter', 'show', 'dev', dev,
'parent', parent])
- HEADER = 'filter protocol ip pref '
+ HEADER = 'filter protocol all pref '
prio = handle = None
actions = []
prevline = ' '
--
To view, visit http://gerrit.ovirt.org/27388
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I9de381e67fae4cb79fe41dbd9d1b60b72eb84de4
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Antoni Segura Puimedon <asegurap(a)redhat.com>
Nir Soffer has uploaded a new change for review.
Change subject: schedule: Introduce scheduling libary
......................................................................
schedule: Introduce scheduling libary
This moudule provides a Scheduler class scheduling execution of
callables on a background thread.
This should be part of the new scalable vm sampling implemntation, and
can be used also whenever you like to perform a short task on a
background thread, without waiting for the completion of the task.
See the module docstring and tests for usage examples.
Change-Id: Ie3764806d93bd37c3b5924080eb5ae4d29e4f4e0
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com>
---
M debian/vdsm-python.install
M lib/vdsm/Makefile.am
A lib/vdsm/schedule.py
M tests/Makefile.am
A tests/scheduleTests.py
M vdsm.spec.in
6 files changed, 346 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/07/29607/1
diff --git a/debian/vdsm-python.install b/debian/vdsm-python.install
index 2d4bba6..1775241 100644
--- a/debian/vdsm-python.install
+++ b/debian/vdsm-python.install
@@ -16,6 +16,7 @@
./usr/lib/python2.7/dist-packages/vdsm/netlink/link.py
./usr/lib/python2.7/dist-packages/vdsm/profile.py
./usr/lib/python2.7/dist-packages/vdsm/qemuimg.py
+./usr/lib/python2.7/dist-packages/vdsm/schedule.py
./usr/lib/python2.7/dist-packages/vdsm/sslutils.py
./usr/lib/python2.7/dist-packages/vdsm/tool/__init__.py
./usr/lib/python2.7/dist-packages/vdsm/tool/dummybr.py
diff --git a/lib/vdsm/Makefile.am b/lib/vdsm/Makefile.am
index c074bb3..89a5573 100644
--- a/lib/vdsm/Makefile.am
+++ b/lib/vdsm/Makefile.am
@@ -32,6 +32,7 @@
netinfo.py \
profile.py \
qemuimg.py \
+ schedule.py \
SecureXMLRPCServer.py \
sslutils.py \
utils.py \
diff --git a/lib/vdsm/schedule.py b/lib/vdsm/schedule.py
new file mode 100644
index 0000000..bd75924
--- /dev/null
+++ b/lib/vdsm/schedule.py
@@ -0,0 +1,213 @@
+#
+# 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 a Scheduler class scheduling execution of
+a callable on a background thread.
+
+To use a scheduler, create an instance:
+
+ scheduler = schedule.Scheduler()
+
+When you want to schedule some callable:
+
+ def task():
+ print '30 seconds passed'
+
+ scheduler.schedule(30.0, task)
+
+task will be called after 30.0 seconds on the scheduler background thread.
+
+If you need to cancel a scheduled call, keep the ScheduledCall object returned
+from Scheduler.schedule(), and cancel the task:
+
+ scheduled_call = scheduler.schedule(30.0, call)
+ ...
+ scheduled_call.cancel()
+
+Finally, when the scheduler is not needed any more:
+
+ scheduler.cancel()
+
+This will cancel any pending calls and terminate the scheduler thread.
+"""
+
+import heapq
+import logging
+import threading
+import time
+
+from . import utils
+
+
+class Scheduler(object):
+ """
+ Schedule calls for future execution in a background thread.
+
+ This class is thread safe; multiple threads can schedule calls or cancel
+ the scheudler.
+ """
+
+ DEFAULT_DELAY = 30.0 # Used if no timeout are scheduled
+
+ _log = logging.getLogger("vds.Scheduler")
+
+ def __init__(self):
+ self._log.debug("Starting scheduler")
+ self._cond = threading.Condition(threading.Lock())
+ self._running = True
+ self._timeouts = []
+ t = threading.Thread(target=self._run)
+ t.daemon = True
+ t.start()
+
+ def schedule(self, delay, callee):
+ """
+ Schedule callee to be called after delay seconds on the scheduler
+ thread.
+
+ Callee must not block or take excessive time to complete. It it does
+ not finish quickly, it may delay other scheduled calls on the scheduler
+ thread.
+
+ Returns a ScheduledCall that may be canceled if callee was not called
+ yet.
+ """
+ deadline = time.time() + delay
+ timeout = _Timeout(deadline, callee)
+ self._log.debug("Schedulng %s", timeout)
+ with self._cond:
+ if self._running:
+ heapq.heappush(self._timeouts, timeout)
+ self._cond.notify()
+ else:
+ timeout.cancel()
+ return ScheduledCall(timeout)
+
+ def cancel(self):
+ """
+ Cancel all schedueld calls and invalidate the scheduler. Calls
+ scheduled after a scheduler was cancel will never be called.
+ """
+ self._log.debug("Canceling scheduler")
+ with self._cond:
+ self._running = False
+ self._cond.notify()
+
+ @utils.traceback(on=_log.name)
+ def _run(self):
+ try:
+ self._log.debug("started")
+ self._loop()
+ self._log.debug("canceled")
+ finally:
+ self._cleanup()
+
+ def _loop(self):
+ while True:
+ with self._cond:
+ if not self._running:
+ return
+ delay = self._time_until_deadline()
+ if delay > 0.0:
+ self._cond.wait(delay)
+ if not self._running:
+ return
+ expired = self._pop_expired_timeouts()
+ for timeout in expired:
+ timeout.fire()
+
+ def _time_until_deadline(self):
+ if self._timeouts:
+ return self._timeouts[0].deadline - time.time()
+ return self.DEFAULT_DELAY
+
+ def _pop_expired_timeouts(self):
+ now = time.time()
+ expired = []
+ while self._timeouts:
+ timeout = self._timeouts[0]
+ if timeout.deadline > now:
+ break
+ heapq.heappop(self._timeouts)
+ expired.append(timeout)
+ return expired
+
+ def _cleanup(self):
+ # Help the garbage collector by breaking reference cycles
+ with self._cond:
+ for timeout in self._timeouts:
+ timeout.cancel()
+
+
+class ScheduledCall(object):
+ """
+ Returned when a callable is scheduled to be called after delay. The caller
+ may cancel the call if it was not called yet.
+
+ This class is thread safe; any thread can cacnel a call.
+ """
+
+ _log = logging.getLogger("vds.Scheduler")
+
+ def __init__(self, timeout):
+ self._timeout = timeout
+
+ @property
+ def deadline(self):
+ return self._timeout.deadline
+
+ def cancel(self):
+ self._log.debug("Canceling %s", self)
+ self._timeout.cancel()
+
+
+# Sentinel for marking timeouts as invalid. Callable so we can invaliate a
+# timeout in a thread safe manner without locks.
+def _INVALID():
+ pass
+
+
+class _Timeout(object):
+ """
+ Created for each scheduled call.
+ """
+
+ _log = logging.getLogger('vds.Timeout')
+
+ def __init__(self, deadline, callee):
+ self.deadline = deadline
+ self.callee = callee
+
+ def fire(self):
+ if self.callee is _INVALID:
+ return
+ try:
+ self.callee()
+ except Exception:
+ self._log.exception("Unhandled exception in scheduled call")
+ finally:
+ self.callee = _INVALID
+
+ def cancel(self):
+ self.callee = _INVALID
+
+ def __cmp__(self, other):
+ return cmp(self.deadline, other.deadline)
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 6507165..786dea4 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -65,6 +65,7 @@
remoteFileHandlerTests.py \
resourceManagerTests.py \
samplingTests.py \
+ scheduleTests.py \
schemaTests.py \
securableTests.py \
sslTests.py \
diff --git a/tests/scheduleTests.py b/tests/scheduleTests.py
new file mode 100644
index 0000000..0c370d9
--- /dev/null
+++ b/tests/scheduleTests.py
@@ -0,0 +1,129 @@
+#
+# 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 threading
+import time
+
+from vdsm import schedule
+from testrunner import VdsmTestCase
+
+
+class SchedulerTests(VdsmTestCase):
+
+ # Time to wait for completion, so test will not fail on overloaded
+ # machines. If tests fails on CI, increase this value.
+ GRACETIME = 0.1
+
+ MAX_TASKS = 1000
+
+ def setUp(self):
+ self.scheduler = schedule.Scheduler()
+
+ def tearDown(self):
+ self.scheduler.cancel()
+
+ def test_schedule(self):
+ delay = 0.3
+ task1 = Task()
+ task2 = Task()
+ timeout1 = self.scheduler.schedule(delay, task1)
+ self.scheduler.schedule(10, task2)
+ task1.wait(delay + self.GRACETIME)
+ self.assertTrue(timeout1.deadline <= task1.call_time)
+ self.assertTrue(task1.call_time < timeout1.deadline + self.GRACETIME)
+ self.assertEquals(task2.call_time, None)
+
+ def test_schedule_many(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertTrue(timeout.deadline <= task.call_time)
+ self.assertTrue(task.call_time < timeout.deadline + self.GRACETIME)
+
+ def test_continue_after_failures(self):
+ self.scheduler.schedule(0.3, FailingTask())
+ task = Task()
+ self.scheduler.schedule(0.4, task)
+ task.wait(0.4 + self.GRACETIME)
+ self.assertTrue(task.call_time is not None)
+
+ def test_cancel_timeout(self):
+ delay = 0.3
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ timeout.cancel()
+ task.wait(delay + self.GRACETIME)
+ self.assertEquals(task.call_time, None)
+
+ def test_cancel_many(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ for task, timeout in tasks:
+ timeout.cancel()
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertEquals(task.call_time, None)
+
+ def test_cancel(self):
+ delay = 0.3
+ tasks = []
+ for i in range(self.MAX_TASKS):
+ task = Task()
+ timeout = self.scheduler.schedule(delay, task)
+ tasks.append((task, timeout))
+ self.scheduler.cancel()
+ last_task = tasks[-1][0]
+ last_task.wait(delay + self.GRACETIME)
+ for task, timeout in tasks:
+ self.assertEquals(task.call_time, None)
+
+
+class Task(object):
+
+ def __init__(self):
+ self.cond = threading.Condition(threading.Lock())
+ self.call_time = None
+
+ def __call__(self):
+ with self.cond:
+ self.call_time = time.time()
+ self.cond.notify()
+
+ def wait(self, timeout):
+ with self.cond:
+ if self.call_time is None:
+ self.cond.wait(timeout)
+
+
+class FailingTask(object):
+
+ def __call__(self):
+ raise Exception("This task is broken")
diff --git a/vdsm.spec.in b/vdsm.spec.in
index dfca5bd..8ba0477 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -1163,6 +1163,7 @@
%{python_sitearch}/%{vdsm_name}/qemuimg.py*
%{python_sitearch}/%{vdsm_name}/SecureXMLRPCServer.py*
%{python_sitearch}/%{vdsm_name}/netconfpersistence.py*
+%{python_sitearch}/%{vdsm_name}/schedule.py*
%{python_sitearch}/%{vdsm_name}/sslutils.py*
%{python_sitearch}/%{vdsm_name}/utils.py*
%{python_sitearch}/%{vdsm_name}/vdscli.py*
--
To view, visit http://gerrit.ovirt.org/29607
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie3764806d93bd37c3b5924080eb5ae4d29e4f4e0
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com>
Francesco Romani has uploaded a new change for review.
Change subject: threadpool: add a thread pool with watchdog
......................................................................
threadpool: add a thread pool with watchdog
Add a new thread pool implementation with built-in
watchdog.
The watchdog is needed to accomodate the needs of
the virt sampling code.
The virt sampling needs to deal with possibly blocking
I/O like calls which cannot be made not-blocking.
Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2
Signed-off-by: Francesco Romani <fromani(a)redhat.com>
---
M lib/threadpool/Makefile.am
A lib/threadpool/examples/demo.py
A lib/threadpool/schedqueue.py
A lib/threadpool/tests/test_schedqueue.py
A lib/threadpool/tests/test_timetrackingthread.py
A lib/threadpool/tests/test_worker.py
A lib/threadpool/tests/test_workqueue.py
A lib/threadpool/watchedpool.py
A lib/threadpool/watchman.py
A lib/threadpool/worker.py
A lib/threadpool/workqueue.py
11 files changed, 1,003 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/91/29191/1
diff --git a/lib/threadpool/Makefile.am b/lib/threadpool/Makefile.am
index 3a71a5d..8b7098e 100644
--- a/lib/threadpool/Makefile.am
+++ b/lib/threadpool/Makefile.am
@@ -22,4 +22,9 @@
dist_threadpool_PYTHON = \
__init__.py \
+ schedqueue.py \
+ watchedpool.py \
+ watchman.py \
+ worker.py \
+ workqueue.py \
$(NULL)
diff --git a/lib/threadpool/examples/demo.py b/lib/threadpool/examples/demo.py
new file mode 100644
index 0000000..cf80930
--- /dev/null
+++ b/lib/threadpool/examples/demo.py
@@ -0,0 +1,62 @@
+#
+# 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 logging
+import random
+import sys
+import time
+
+from watchedpool import WatchedThreadPool
+
+
+def _test_fun(*args):
+ """
+ just block the caller for a random time span.
+ """
+ tmo = random.randint(1, 15)
+ logging.info('started, waiting: %.3fs', tmo)
+ time.sleep(tmo)
+ logging.info('done')
+
+
+def _main(args):
+ """
+ main test driver
+ """
+ nworkers = int(args[0]) if len(args) >= 1 else 1
+ pool = WatchedThreadPool(nworkers)
+ pool.start()
+ for tag in range(nworkers):
+ pool.submit_periodic(4, _test_fun)
+
+ try:
+ while True:
+ time.sleep(1.0)
+ except KeyboardInterrupt:
+ logging.debug('exiting!')
+ finally:
+ pool.stop()
+
+
+if __name__ == '__main__':
+ logging.basicConfig(
+ level=logging.DEBUG,
+ format='%(asctime)s %(threadName)-10s %(levelname)-8s %(message)s')
+ _main(sys.argv[1:])
diff --git a/lib/threadpool/schedqueue.py b/lib/threadpool/schedqueue.py
new file mode 100644
index 0000000..c2bfbae
--- /dev/null
+++ b/lib/threadpool/schedqueue.py
@@ -0,0 +1,123 @@
+#
+# 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
+#
+
+"""
+Time-aware priority queue, built like queue.Queue
+but tailored for WatchedThreadPool needs.
+"""
+
+import heapq
+import threading
+import time
+
+
+class QueueError(Exception):
+ """
+ Queue exception root.
+ """
+
+
+class Empty(QueueError):
+ """
+ Attempt to get an item from an empty queue.
+ """
+
+
+class Full(QueueError):
+ """
+ Attempt to put an item into a full queue.
+ """
+
+
+class NotYet(QueueError):
+ """
+ Attempt to get an item which is not due at the
+ present (reported) time.
+ """
+
+
+class SchedQueue(object):
+ """
+ Time-aware, not blocking (client must take care of
+ waiting/timing) priority queue.
+ """
+ def __init__(self, maxsize=0, timefunc=time.time):
+ self._timefunc = timefunc
+ self._maxsize = maxsize
+ self._queue = self._init(maxsize)
+ self._lock = threading.Lock()
+
+ def __len__(self):
+ with self._lock:
+ return self._size()
+
+ def put(self, item, delay=0):
+ """
+ Put an item into the queue, to be made ready 'delay' seconds
+ in the future.
+ """
+ with self._lock:
+ if self._maxsize > 0 and self._size() == self._maxsize:
+ raise Full
+ self._put((self._timefunc() + delay, item))
+
+ def get(self, timenow):
+ """
+ Ask for an item to be consumed.
+ """
+ with self._lock:
+ if not self._size():
+ raise Empty
+ timeval, _ = self._peek()
+ if timeval > timenow:
+ raise NotYet
+ _, item = self._get()
+ return item
+
+ def _init(self, maxsize):
+ """
+ builds the internal queue.
+ """
+ self._maxsize = maxsize # to make pylint happy
+ return []
+
+ def _size(self):
+ """
+ size of the internal queue.
+ """
+ return len(self._queue)
+
+ def _put(self, item):
+ """
+ put a new element (at the end of) in the queue.
+ """
+ heapq.heappush(self._queue, item)
+
+ def _get(self):
+ """
+ gets (and removes) the next element in queue.
+ """
+ return heapq.heappop(self._queue)
+
+ def _peek(self):
+ """
+ read-only view of the next element in queue.
+ """
+ return self._queue[0]
diff --git a/lib/threadpool/tests/test_schedqueue.py b/lib/threadpool/tests/test_schedqueue.py
new file mode 100644
index 0000000..3595589
--- /dev/null
+++ b/lib/threadpool/tests/test_schedqueue.py
@@ -0,0 +1,56 @@
+#
+# 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 unittest
+
+import schedqueue
+
+
+class SchedQueueTests(unittest.TestCase):
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def setUp(self):
+ self._ts = 0
+ self._sq = schedqueue.SchedQueue(timefunc=self._faketime)
+
+ def test_empty_on_create(self):
+ self.assertEqual(len(self._sq), 0)
+
+ def test_put_on_full(self):
+ self._sq = schedqueue.SchedQueue(maxsize=1, timefunc=self._faketime)
+ self._sq.put('foo')
+ self.assertRaises(schedqueue.Full, self._sq.put, 'bar')
+
+ def test_get_on_empty(self):
+ self.assertRaises(schedqueue.Empty, self._sq.get, 1)
+
+ def test_get_too_early(self):
+ item = 'foo'
+ self._sq.put(item, delay=5)
+ self.assertRaises(schedqueue.NotYet, self._sq.get, 0)
+
+ def test_put_get(self):
+ item = 'foo'
+ self._sq.put(item)
+ res = self._sq.get(timenow=1)
+ self.assertEqual(res, item)
diff --git a/lib/threadpool/tests/test_timetrackingthread.py b/lib/threadpool/tests/test_timetrackingthread.py
new file mode 100644
index 0000000..14a25ef
--- /dev/null
+++ b/lib/threadpool/tests/test_timetrackingthread.py
@@ -0,0 +1,64 @@
+#
+# 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 unittest
+
+import worker
+
+
+class TimeTrackingThreadTests(unittest.TestCase):
+ def _fakesleep(self, delay):
+ self._ts += max(0, delay) # avoid negative delays
+
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def setUp(self):
+ self._ts = 0
+ self._ttt = worker.TimeTrackingThread(self._faketime)
+
+ def assertIdle(self):
+ self.assertFalse(self._ttt.busy)
+ self.assertEqual(self._ttt.task, None)
+
+ def assertBusyOn(self, task):
+ self.assertTrue(self._ttt.busy)
+ self.assertEqual(self._ttt.task, task)
+
+ def test_created_free(self):
+ self.assertIdle()
+
+ def test_carry_task(self):
+ self.assertIdle()
+
+ task = 'foobar'
+ with self._ttt.track_time(task):
+ self.assertBusyOn(task)
+
+ self.assertIdle()
+
+ def test_elapsed(self):
+ task = 'barbaz'
+ delay = 5
+ with self._ttt.track_time(task):
+ self._fakesleep(delay)
+ self.assertTrue(self._ttt.elapsed >= delay)
diff --git a/lib/threadpool/tests/test_worker.py b/lib/threadpool/tests/test_worker.py
new file mode 100644
index 0000000..7396673
--- /dev/null
+++ b/lib/threadpool/tests/test_worker.py
@@ -0,0 +1,60 @@
+#
+# 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 contextlib
+import unittest
+
+import schedqueue
+import worker
+
+
+class FakeWorkQueue(object):
+ def __init__(self, items):
+ self._items = items
+
+ @contextlib.contextmanager
+ def fetch(self):
+ if not self._items:
+ raise schedqueue.Empty
+ item = self._items.pop(0)
+ if issubclass(item, Exception):
+ raise item
+ else:
+ yield item
+
+
+class WorkerThreadTests(unittest.TestCase):
+ def test__pull_empty(self):
+ wq = FakeWorkQueue([])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
+
+ def test__pull_empty_exc(self):
+ wq = FakeWorkQueue([schedqueue.Empty])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
+
+ def test__pull_notyet(self):
+ wq = FakeWorkQueue([schedqueue.NotYet])
+ wt = worker.Worker(wq)
+ with wt._pull() as item:
+ self.assertEqual(item, (None, None, None, None))
diff --git a/lib/threadpool/tests/test_workqueue.py b/lib/threadpool/tests/test_workqueue.py
new file mode 100644
index 0000000..68e982d
--- /dev/null
+++ b/lib/threadpool/tests/test_workqueue.py
@@ -0,0 +1,80 @@
+#
+# 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 unittest
+
+import schedqueue
+import watchedpool
+
+
+def _do_nothing(*args, **kwargs):
+ pass
+
+
+class WorkQueueTests(unittest.TestCase):
+ def _fakesleep(self, delay):
+ pass
+
+ def _faketime(self):
+ ret = self._ts
+ self._ts += 1
+ return ret
+
+ def assertDummyWork(self, workitem, in_tag=None):
+ out_tag, work, args, kwargs = workitem
+ if in_tag is not None:
+ self.assertEqual(in_tag, out_tag)
+ self.assertEqual(work, _do_nothing)
+ self.assertEqual(len(args), 0)
+ self.assertEqual(len(kwargs), 0)
+
+ def setUp(self):
+ self._ts = 0
+ self._wq = watchedpool.WorkQueue(
+ timefunc=self._faketime,
+ delayfunc=self._fakesleep)
+
+ def test_created_empty(self):
+ self.assertEqual(len(self._wq), 0)
+
+ def test_get_on_empty(self):
+ self.assertRaises(schedqueue.Empty, self._wq.get)
+
+ def test_post_on_empty(self):
+ self._wq.post(0, _do_nothing)
+ self.assertEqual(len(self._wq), 1)
+
+ def test_put_get(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 0, _do_nothing)
+ self.assertDummyWork(self._wq.get())
+
+ def test_remove(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 0, _do_nothing)
+ self._wq.remove(in_tag)
+ self.assertRaises(schedqueue.Empty, self._wq.get)
+
+ def test_fetch(self):
+ in_tag = 'foo'
+ self._wq.put(in_tag, 1, _do_nothing)
+ with self._wq.fetch() as work_item:
+ self.assertDummyWork(work_item)
+ self.assertEqual(len(self._wq), 1)
diff --git a/lib/threadpool/watchedpool.py b/lib/threadpool/watchedpool.py
new file mode 100644
index 0000000..6a8938f
--- /dev/null
+++ b/lib/threadpool/watchedpool.py
@@ -0,0 +1,128 @@
+#
+# 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
+#
+
+"""
+PoC of a watched thread pool implementation.
+This thread pool adds a watcher/watchman thread which detaches `rogue'
+threads from the main pool, and ask them to die as soon as possible.
+The purpose is to avoid worker threads getting stuck too much on
+busy, possible blocking operations (e.g. risky I/O).
+
+In the worst case scenario, the pool size degenerates in a
+thread-per-operation. In the common and expected case,
+the pool size is fixed and rogue threads are just spikes
+in the thread number.
+
+TODO:
+- allow the client to resize dynamically the pool [?]
+- let this code be easily mergeable with vdsm/storage/threadPool.py
+- demonstrate that this code does not leak threads.
+"""
+
+import logging
+import threading
+
+from watchman import Watchman
+from worker import Worker
+from workqueue import WorkQueue
+
+
+class WatchedThreadPool(object):
+ """
+ Straightforward thread pool, with an added Watchman.
+ """
+ def __init__(self, nworkers=1, work_interval=1):
+ self._nworkers = nworkers
+ self._work_interval = work_interval
+ self._watch_interval = 2 * self._work_interval
+ self._busy_timeout = 3 * self._watch_interval
+ self._work_queue = None
+ self._workers = None
+ self._watcher = None
+ self._running = False
+ self._lock = threading.Lock()
+
+ def start(self):
+ """
+ starts the pool. Thread safe.
+ """
+ with self._lock:
+ if not self._running:
+ self._work_queue = WorkQueue(self._work_interval)
+ self._workers = [self._make_worker()
+ for _ in range(self._nworkers)]
+ self._watcher = Watchman(self._workers,
+ self._make_worker,
+ self._watch_interval,
+ self._busy_timeout)
+ self._watcher.start()
+ logging.info('pool started')
+ self._running = True
+
+ def stop(self, wait=True):
+ """
+ terminate all pooled threads. Thread safe.
+ """
+ with self._lock:
+ if self._running:
+ self._running = False
+
+ self._watcher.stop()
+ if wait:
+ for worker in self._workers:
+ worker.join()
+ del worker
+ self._watcher.reap([]) # last chance
+ self._watcher.join()
+ del self._watcher
+
+ def submit_periodic(self, interval, work, *args, **kwargs):
+ """
+ submit a new work unit to the pool.
+ returns the work unit tag.
+ The tag is guaranteed to be unique per-pool.
+ """
+ if self._running:
+ tag = self._work_queue.post(interval, work, args, kwargs)
+ logging.debug('new work: (%s, %s, %s) -> %s',
+ str(work), str(args), str(kwargs), tag)
+ return tag # FIXME
+ else:
+ return None
+
+ def remove(self, tag):
+ """
+ remove a work unit from the pool.
+ """
+ return self._work_queue.remove(tag)
+
+ def notify_timeout(self, tag, on_timeout):
+ """
+ register a timeout callback for the given work unit.
+ """
+ self._watcher.notify_timeout(tag, on_timeout)
+
+ def _make_worker(self):
+ """
+ creates a new idle (not started) worker.
+ """
+ worker = Worker(self._work_queue)
+ logging.info('worker %s added to the pool', worker.name)
+ return worker
diff --git a/lib/threadpool/watchman.py b/lib/threadpool/watchman.py
new file mode 100644
index 0000000..53bfc11
--- /dev/null
+++ b/lib/threadpool/watchman.py
@@ -0,0 +1,209 @@
+#
+# 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
+#
+
+"""
+PoF of a watched thread pool implementation.
+This thread pool adds a watcher/watchman thread which detaches `rogue'
+threads from the main pool, and ask them to die as soon as possible.
+The purpose is to avoid worker threads getting stuck too much on
+busy, possible blocking operations (e.g. risky I/O).
+
+In the worst case scenario, the pool size degenerates in a
+thread-per-operation. In the common and expected case,
+the pool size is fixed and rogue threads are just spikes
+in the thread number.
+
+TODO:
+- allow the client to resize dynamically the pool [?]
+- let this code be easily mergeable with vdsm/storage/threadPool.py
+- demonstrate that this code does not leak threads.
+"""
+
+import logging
+import threading
+import time
+
+
+class Watchman(threading.Thread):
+ """
+ Watches the worker threads, detects the rogues, if any,
+ and possibly replace them to ensure the pool is operative
+ and not blocked.
+ """
+ def __init__(self, workers, make_worker,
+ watch_interval=5, busy_timeout=10,
+ rogues_cap=None, delayfunc=time.sleep):
+ super(Watchman, self).__init__()
+ self.daemon = True
+ self._workers = workers
+ self._rogues = []
+ self._rogues_cap = rogues_cap
+ self._watch_interval = watch_interval
+ self._make_worker = make_worker
+ self._busy_timeout = busy_timeout
+ self._timeout_cb = {}
+ self._stop = threading.Event()
+ self._running = True
+ self._delayfunc = delayfunc
+
+ def suspend(self):
+ """
+ suspend the watch, without stopping.
+ """
+ self._running = False
+
+ def resume(self):
+ """
+ resume the watch.
+ """
+ self._running = True
+
+ def start(self):
+ """
+ start the watched pool and the watcher itself.
+ """
+ for worker in self._workers:
+ worker.start()
+ super(Watchman, self).start()
+
+ def stop(self):
+ """
+ stop the watched pool, and the watcher itself.
+ """
+ self.suspend()
+ for worker in self._workers:
+ worker.stop()
+ self._stop.set()
+
+ def run(self):
+ logging.debug(
+ 'watcher starting (with %i workers)', len(self._workers))
+
+ while not self._stop.is_set():
+ self._delayfunc(self._watch_interval)
+ if self._running:
+ workers, rogues, newbies = self.verify()
+ self.enforce(workers, rogues, newbies)
+
+ logging.debug('watcher done')
+
+ def notify_timeout(self, tag, on_timeout):
+ """
+ register a timeout callback for the given work unit.
+ """
+ self._timeout_cb[tag] = on_timeout
+
+ def stuck(self, worker):
+ """
+ is a given worker stuck?
+ """
+ stuck, task = False, None
+ if worker.busy:
+ # use locals to freeze the state
+ task = worker.task
+ elapsed = worker.elapsed
+ stuck = elapsed > self._busy_timeout
+ logging.warning(
+ '%s busy on %s for the last %02i seconds (max=%02i) -> %s',
+ worker.name, task, elapsed, self._busy_timeout,
+ 'STUCK' if stuck else 'OK')
+ return stuck, task
+
+ def verify(self):
+ """
+ check the health of the worker pool, and classify
+ it into
+ - workers: well behaving worker threads
+ - rogues: workers got stuck, to be evicted from the pool
+ - newbies: new workers ready to replace rogues
+ """
+ try:
+ workers, rogues, newbies = [], [], []
+ logging.debug('verification loop started')
+ for worker in self._workers:
+ stuck, task = self.stuck(worker)
+ if not stuck:
+ workers.append(worker)
+ elif not self._room_for_rogues(rogues):
+ logging.warning(
+ 'worker %s gone rogue, but cap reached', worker.name)
+ elif not self._is_detachable(task):
+ logging.warning(
+ 'worker %s gone rogue, but task %s not detachable',
+ worker.name, task)
+ else:
+ logging.info(
+ 'worker %s gone rogue, replacing', worker.name)
+ worker.stop()
+ rogues.append(worker) # until it dies
+ newbies.append(self._make_worker())
+ logging.debug('verification loop done: %i rogues', len(rogues))
+ return workers, rogues, newbies
+ except Exception:
+ logging.exception('verify failed', exc_info=True)
+
+ def enforce(self, workers, rogues, newbies):
+ """
+ enforce the pool size by evicting rogues into a limbo,
+ and by replenishing the worker pool with the provided
+ newbies.
+ """
+ try:
+ for newbie in newbies:
+ newbie.start()
+ workers.append(newbie)
+ self._workers = workers
+ self.reap(rogues)
+ except Exception:
+ logging.exception('enforce failed', exc_info=True)
+
+ def reap(self, rogues):
+ """
+ collect the terminated rogues.
+ """
+ rogues.extend(self._rogues)
+ new_rogues = []
+ for rogue in rogues:
+ if rogue.is_alive():
+ new_rogues.append(rogue)
+ else:
+ logging.info('rogue thread %s collected', rogue.name)
+ rogue.join()
+ del rogue
+ self._rogues = new_rogues
+
+ def _room_for_rogues(self, rogues):
+ return (self._rogues_cap is None
+ or len(rogues) < self._rogues_cap)
+
+ def _is_detachable(self, task):
+ """
+ if a task is detachable, its worker thread will be detached
+ by the pool. Otherwise the thread will be kept in the pool
+ """
+ on_timeout = self._timeout_cb.get(task, _on_timeout_dummy)
+ return on_timeout(task)
+
+
+def _on_timeout_dummy(*args):
+ """
+ Dummy timeout callback. Marks every work unit as detachable.
+ """
+ return True
diff --git a/lib/threadpool/worker.py b/lib/threadpool/worker.py
new file mode 100644
index 0000000..fe63d34
--- /dev/null
+++ b/lib/threadpool/worker.py
@@ -0,0 +1,129 @@
+#
+# 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 contextlib
+import logging
+import threading
+import time
+
+import schedqueue
+
+
+class TimeTrackingThread(threading.Thread):
+ """
+ Thin mixin to help a worker thread to track the time
+ consumed by a work item.
+ """
+ def __init__(self, timefunc=time.time):
+ super(TimeTrackingThread, self).__init__()
+ self._work_started_at = None
+ self._working_on = None
+ self._timefunc = timefunc
+
+ @property
+ def busy(self):
+ """
+ is the thread doing some work or is it idle?
+ """
+ return self._work_started_at is not None
+
+ @property
+ def elapsed(self):
+ """
+ how much time since the thread started the current work unit?
+ """
+ return self._timefunc() - self._work_started_at
+
+ @property
+ def task(self):
+ """
+ what is this worker doing?
+ """
+ return self._working_on
+
+ @contextlib.contextmanager
+ def track_time(self, tag):
+ """
+ helper context manager to track the time consumed in
+ the current work unit
+ """
+ self._working_on = tag
+ self._work_started_at = self._timefunc()
+ yield
+ self._work_started_at = None
+ self._working_on = None
+
+
+class Worker(TimeTrackingThread):
+ """
+ A regular worker thread. This pool has no additional
+ requirements saved the ability of a worker thread
+ to die ASAP (see the stop() method)
+ """
+ def __init__(self, work_queue):
+ super(Worker, self).__init__()
+ self.daemon = True
+ self._work_queue = work_queue
+ self._stop = threading.Event()
+
+ def run(self):
+ logging.info('worker %s starting', self.name)
+
+ while not self._stop.is_set():
+ self.process()
+
+ logging.info('worker %s done', self.name)
+
+ def process(self):
+ """
+ execute a single try of work processing.
+ """
+ with self._pull() as (tag, work, args, kwargs):
+ if work is not None:
+ self._do_work(tag, work, args, kwargs)
+
+ def stop(self):
+ """
+ gracefully terminate ASAP, and avoid to begin more work.
+ """
+ self._stop.set()
+
+ @contextlib.contextmanager
+ def _pull(self):
+ """
+ pulls the next work item
+ """
+ try:
+ with self._work_queue.fetch() as item:
+ yield item
+ except schedqueue.NotYet:
+ yield None, None, None, None
+ except schedqueue.Empty:
+ yield None, None, None, None
+
+ def _do_work(self, tag, work, args, kwargs):
+ """
+ process a single work unit.
+ """
+ with self.track_time(tag):
+ try:
+ work(*args, **kwargs)
+ except Exception:
+ logging.exception('work failed', exc_info=True)
diff --git a/lib/threadpool/workqueue.py b/lib/threadpool/workqueue.py
new file mode 100644
index 0000000..73949b8
--- /dev/null
+++ b/lib/threadpool/workqueue.py
@@ -0,0 +1,87 @@
+#
+# 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 contextlib
+import time
+import uuid
+
+import schedqueue
+
+
+class WorkQueue(object):
+ """
+ Tiny wrapper around a plain Queue.Queue, to encapsulate
+ the 'periodic work' behaviour.
+ """
+ def __init__(self, wait_timeout=1,
+ timefunc=time.time, delayfunc=time.sleep):
+ self._queue = schedqueue.SchedQueue(timefunc=timefunc)
+ self._work = {}
+ self._wait_timeout = wait_timeout
+ self._timefunc = timefunc
+ self._delayfunc = delayfunc
+
+ def put(self, tag, interval, work, *args, **kwargs):
+ """
+ like Queue.put().
+ """
+ item = (work, args, kwargs, interval)
+ self._work[tag] = item
+ self._queue.put(tag, delay=interval)
+
+ def get(self):
+ """
+ like Queue.get().
+ """
+ tag, work, args, kwargs, _ = self._get()
+ return (tag, work, args, kwargs)
+
+ def remove(self, tag):
+ return self._work.pop(tag, None)
+
+ def post(self, interval, work, *args, **kwargs):
+ """
+ like put(), but with auto-tagging.
+ """
+ return self.put(str(uuid.uuid4()), interval, work, *args, **kwargs)
+
+ @contextlib.contextmanager
+ def fetch(self):
+ """
+ like get(), but automatically reinject the work
+ to do in the queue if the task is periodic.
+ """
+ tag, work, args, kwargs, interval = self._get()
+ yield (tag, work, args, kwargs)
+ if interval:
+ self.put(tag, interval, work, args, kwargs)
+
+ def _get(self):
+ tag, work, args, kwargs = None, None, None, None
+ while work is None:
+ self._delayfunc(self._wait_timeout)
+ tag = self._queue.get(self._timefunc())
+ item = self._work.get(tag, None)
+ if item is not None:
+ work, args, kwargs, interval = item
+ return (tag, work, args, kwargs, interval)
+
+ def __len__(self):
+ return len(self._work)
--
To view, visit http://gerrit.ovirt.org/29191
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ic06da1ba57868dc2c7db67a1868ad10087a1cff2
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani(a)redhat.com>
Amador Pahim has uploaded a new change for review.
Change subject: [RFC] Sysctl to allow iSCSI multipath with multiple NICs in the same subnet
......................................................................
[RFC] Sysctl to allow iSCSI multipath with multiple NICs in the same subnet
When two or more interfaces are in the same subnet, the default OS
behaviour won't allow the independent traffic between local NICs and
iSCSI targets.
These sysctl tweaks are needed to have iSCSI multipath working in such
topology:
eth1 (192.168.25.200) --|
|-- iSCSI Target (192.168.25.10)
eth2 (192.168.25.201) --|
The proposed changes are reccomended by the storage vendor and are described
in documents bellow:
- http://en.community.dell.com/dell-groups/dtcmedia/m/mediagallery/20371245/d…
- https://support.equallogic.com/WorkArea/DownloadAsset.aspx?id=8727
Change-Id: Ibf93d49317c76aece764e53e58e0ff28868f16b0
Bug-Url: https://bugzilla.redhat.com/show_bug.cgi?id=1112861
Signed-off-by: Amador Pahim <apahim(a)redhat.com>
---
M vdsm/vdsm-sysctl.conf
1 file changed, 11 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/29/31529/1
diff --git a/vdsm/vdsm-sysctl.conf b/vdsm/vdsm-sysctl.conf
index ecd1386..332379c 100644
--- a/vdsm/vdsm-sysctl.conf
+++ b/vdsm/vdsm-sysctl.conf
@@ -1,3 +1,14 @@
#Set dirty page parameters
vm.dirty_ratio = 5
vm.dirty_background_ratio = 2
+
+#Set ARP to avoid RST when interfaces are in the same subnet.
+net.ipv4.conf.all.arp_ignore = 1
+net.ipv4.conf.all.arp_announce = 2
+
+#Set rp_filter to "loose mode" allowing multiple iSCSI connections in
+#a multiple NIC per subnet configuration.
+net.ipv4.conf.all.rp_filter = 2
+
+#Workaround for netfilter issue marking packets as invalid under heavy load.
+net.netfilter.nf_conntrack_tcp_be_liberal = 1
--
To view, visit http://gerrit.ovirt.org/31529
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ibf93d49317c76aece764e53e58e0ff28868f16b0
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Amador Pahim <apahim(a)redhat.com>
Francesco Romani has uploaded a new change for review.
Change subject: profiling: Add support for memory profiling
......................................................................
profiling: Add support for memory profiling
There are many python memory profilers, but most of them are
tailored for interactive usage, so they are poorly suited
for a daemon, like VDSM.
This patch adds support for dowser:
<http://www.aminus.net/wiki/Dowser>
<https://pypi.python.org/pypi/dowser/0.2>
which has the following benefits which makes it ina better position
to be used with VDSM:
+ self contained, includes WEB UI
+ allows live monitoring
+ tailored for cherrypy usage, so friendly towards server applications.
+ easily portable (pure package)
It must be noted that dowser has some drawbacks as well
- significat dependencies, both in number and in size (with respect
to VDSM standards): cherrypy, PIL
- based on gc module
As per the cpu profile already added, this code is meant to be
used only as debug aid or in development environments.
The feature is controlled by a config tunable and disabled by default;
the feature disable itself if any dependency is missed.
Change-Id: Ib56b65513e0118b68cd43791bf655c928d6a26e2
Signed-off-by: Francesco Romani <fromani(a)redhat.com>
---
M debian/vdsm-python.install
M lib/vdsm/Makefile.am
A lib/vdsm/profile_mem.py
M vdsm.spec.in
M vdsm/vdsm
5 files changed, 89 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/19/32019/1
diff --git a/debian/vdsm-python.install b/debian/vdsm-python.install
index 8d11bfc..f429a91 100644
--- a/debian/vdsm-python.install
+++ b/debian/vdsm-python.install
@@ -15,6 +15,7 @@
./usr/lib/python2.7/dist-packages/vdsm/netlink/link.py
./usr/lib/python2.7/dist-packages/vdsm/netlink/route.py
./usr/lib/python2.7/dist-packages/vdsm/profile.py
+./usr/lib/python2.7/dist-packages/vdsm/profile_mem.py
./usr/lib/python2.7/dist-packages/vdsm/qemuimg.py
./usr/lib/python2.7/dist-packages/vdsm/sslutils.py
./usr/lib/python2.7/dist-packages/vdsm/tool/__init__.py
diff --git a/lib/vdsm/Makefile.am b/lib/vdsm/Makefile.am
index 4bebf28..04c8018 100644
--- a/lib/vdsm/Makefile.am
+++ b/lib/vdsm/Makefile.am
@@ -31,6 +31,7 @@
netconfpersistence.py \
netinfo.py \
profile.py \
+ profile_mem.py \
qemuimg.py \
sslutils.py \
utils.py \
diff --git a/lib/vdsm/profile_mem.py b/lib/vdsm/profile_mem.py
new file mode 100755
index 0000000..a60a833
--- /dev/null
+++ b/lib/vdsm/profile_mem.py
@@ -0,0 +1,83 @@
+#
+# 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 memory profiling.
+"""
+
+import threading
+
+from vdsm.config import config
+
+try:
+ # dowser dependencies
+ import cherrypy
+ from PIL import Image, ImageDraw
+ # dowser profiler itself
+ import dowser
+except IOError:
+ _SUPPORTED = False
+else:
+ _SUPPORTED = True
+
+Image # makes pyflakes happy
+ImageDraw # same
+
+
+_thread = None
+
+
+def memory_usage_server(port=None):
+ cherrypy.tree.mount(dowser.Root())
+
+ port = (config.getint('vars', 'profile_memory_port')
+ if port is None else port)
+
+ cherrypy.config.update({
+ 'environment': 'embedded',
+ 'server.socket_port': port})
+
+ cherrypy.engine.start()
+
+
+def start():
+ """ Starts application memory profiling """
+ if _SUPPORTED and is_enabled():
+ _start_profiling()
+
+
+def stop():
+ """ Stops application memory profiling """
+ pass # not yet needed
+
+
+def is_enabled():
+ return config.getboolean('vars', 'profile_memory_enable')
+
+
+def is_running():
+ return _thread is not None and _thread.is_alive()
+
+
+def _start_profiling():
+ global _thread
+ _thread = threading.Thread(name='memprofile', target=memory_usage_server)
+ _thread.daemon = True
+ _thread.start()
diff --git a/vdsm.spec.in b/vdsm.spec.in
index 5f9488c..233eee6 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -1184,6 +1184,7 @@
%{python_sitearch}/%{vdsm_name}/netlink/link.py*
%{python_sitearch}/%{vdsm_name}/netlink/route.py*
%{python_sitearch}/%{vdsm_name}/profile.py*
+%{python_sitearch}/%{vdsm_name}/profile_mem.py*
%{python_sitearch}/%{vdsm_name}/qemuimg.py*
%{python_sitearch}/%{vdsm_name}/netconfpersistence.py*
%{python_sitearch}/%{vdsm_name}/sslutils.py*
diff --git a/vdsm/vdsm b/vdsm/vdsm
index e575295..619e57e 100755
--- a/vdsm/vdsm
+++ b/vdsm/vdsm
@@ -34,6 +34,7 @@
from vdsm.config import config
from vdsm import libvirtconnection
from vdsm import profile
+from vdsm import profile_mem
from storage.dispatcher import Dispatcher
from storage.hsm import HSM
@@ -69,6 +70,7 @@
zombiereaper.registerSignalHandler()
profile.start()
+ profile_mem.start()
libvirtconnection.start_event_loop()
@@ -85,6 +87,7 @@
while running[0]:
signal.pause()
+ profile_mem.stop()
profile.stop()
finally:
cif.prepareForShutdown()
--
To view, visit http://gerrit.ovirt.org/32019
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ib56b65513e0118b68cd43791bf655c928d6a26e2
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani(a)redhat.com>