From 84971fe7d44a899f79caa7ed12a97b075f566e3b Mon Sep 17 00:00:00 2001
From: David Lehman <dlehman@redhat.com>
Date: Wed, 8 Apr 2015 11:07:33 -0500
Subject: [PATCH 11/51] Add a framework for event management.

There are three root classes: Event, EventQueue, EventManager.

Event can be subclassed to abstract various types of events. Initially,
the only subclass is UdevEvent.

EventQueue is meant to be a generic event queue and thus hopefully will
not need to be subclassed.

EventManager provides means to enable/disable event monitoring, set a
handler function, and dispatch received events to the configured handler.
---
 blivet/errors.py     |   9 ++
 blivet/event.py      | 291 +++++++++++++++++++++++++++++++++++++++++++++++++++
 blivet/flags.py      |   5 +
 tests/events_test.py |  88 ++++++++++++++++
 4 files changed, 393 insertions(+)
 create mode 100644 blivet/event.py
 create mode 100644 tests/events_test.py

diff --git a/blivet/errors.py b/blivet/errors.py
index d9e5c2a..a146041 100644
--- a/blivet/errors.py
+++ b/blivet/errors.py
@@ -194,3 +194,12 @@ class PopulatorLookupError(PopulatorError):
 # factories
 class DeviceFactoryError(StorageError):
     pass
+
+class EventManagerError(StorageError):
+    pass
+
+class EventQueueEmptyError(StorageError):
+    pass
+
+class EventParamError(StorageError):
+    pass
diff --git a/blivet/event.py b/blivet/event.py
new file mode 100644
index 0000000..479b702
--- /dev/null
+++ b/blivet/event.py
@@ -0,0 +1,291 @@
+# event.py
+# Event management classes.
+#
+# Copyright (C) 2015  Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU Lesser General Public License v.2, or (at your option) any later
+# version. This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY expressed or implied, including the implied
+# warranties of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
+# the GNU Lesser General Public License for more details.  You should have
+# received a copy of the GNU Lesser General Public License along with this
+# program; if not, write to the Free Software Foundation, Inc., 51 Franklin
+# Street, Fifth Floor, Boston, MA 02110-1301, USA.  Any Red Hat trademarks
+# that are incorporated in the source code or documentation are not subject
+# to the GNU Lesser General Public License and may only be used or
+# replicated with the express permission of Red Hat, Inc.
+#
+# Red Hat Author(s): David Lehman <dlehman@redhat.com>
+#
+
+import abc
+import copy
+from collections import deque
+from threading import RLock, Timer, currentThread
+import pyudev
+from six import add_metaclass
+
+from . import udev
+from . import util
+from .errors import EventManagerError, EventParamError, EventQueueEmptyError
+from .flags import flags
+
+import logging
+log = logging.getLogger("blivet")
+
+##
+## Event
+##
+@add_metaclass(abc.ABCMeta)
+class Event(util.ObjectID):
+    def __init__(self, action, info):
+        """
+            :param str action: a string describing the type of event
+            :param info: information about the device
+
+            The info parameter can be of any type provided that the subclass
+            using it does so appropriately.
+        """
+        self.action = action
+        self.info = info
+
+    @abc.abstractproperty
+    def device(self):
+        """ Basename (friendly) of device this event acted upon. """
+        return None
+
+    def __str__(self):
+        return "%s %s [%d]" % (self.action, self.device, self.id)
+
+class UdevEvent(Event):
+    def __init__(self, action, info):
+        """
+            :param str action: a string describing the type of event
+            :param :class:`pyudev.Device` info: udev db entry
+        """
+        super(UdevEvent, self).__init__(action, info)
+
+    @property
+    def device(self):
+        return udev.device_get_name(self.info)
+
+##
+## EventQueue
+##
+class EventQueue(object):
+    def __init__(self):
+        self._queue = deque()
+        self._lock = RLock()
+
+        # list of (device, action, count) tuples
+        self._mask_list = []
+
+    def _mask_list_match(self, event):
+        """ Return True if this event should be ignored """
+        match = None
+        for (device, action, count) in self._mask_list:
+            if ((device is None or event.device == device) and
+                (action is None or event.action == action)):
+                match = (device, action, count)
+                break
+
+        return match
+
+    def _mask_list_update(self, entry):
+        """ Update event mask after a hit at the specified index. """
+        count = entry[2]
+        if count == 1:
+            self._mask_list.remove(entry)
+        elif count > 1:
+            idx = self._mask_list.index(entry)
+            self._mask_list[idx] = (entry[0], entry[1], count - 1)
+
+    def enqueue(self, event):
+        with self._lock:
+            bl_entry = self._mask_list_match(event)
+            if bl_entry is None:
+                self._queue.append(event)
+            else:
+                self._mask_list_update(bl_entry)
+
+    def dequeue(self):
+        """ Dequeue and return the next event.
+
+            :returns: the next uevent
+            :rtype: :class:`~.Event`
+            :raises class:`~.errors.EventQueueEmptyError` if the queue is empty
+        """
+        with self._lock:
+            if not self._queue:
+                raise EventQueueEmptyError()
+
+            return self._queue.popleft()
+
+    def mask_add(self, device=None, action=None, count=1):
+        """ Ignore future events.
+
+            :keyword str device: ignore events on the named device
+            :keyword str action: ignore events of the specified type
+            :keyword int count: number of events to ignore
+
+            device of None means mask events on all devices
+            action of None means mask all event types
+            count of 0 means permanently mask this device/action pair
+        """
+        self._mask_list.append((device, action, count))
+
+    def __list__(self):
+        return list(self._queue)
+
+    def __iter__(self):
+        return iter(self._queue)
+
+    def __deepcopy__(self, memo):
+        new = self.__class__.__new__(self.__class__)
+        memo[id(self)] = new
+        for (attr, value) in self.__dict__.items():
+            if attr == "_lock":
+                setattr(new, attr, RLock())
+            else:
+                setattr(new, attr, copy.deepcopy(value, memo))
+
+        return new
+
+
+##
+## EventManager
+##
+@add_metaclass(abc.ABCMeta)
+class EventManager(object):
+    _event_queue_class = EventQueue
+
+    def __init__(self, handler_cb=None, notify_cb=None):
+        self._handler_cb = None
+        self._notify_cb = None
+
+        if handler_cb is not None:
+            self.handler_cb = handler_cb
+
+        if notify_cb is not None:
+            self.notify_cb = notify_cb
+
+        self._queue = self._event_queue_class()
+
+    #
+    # handler_cb is the main event handler
+    #
+    def _set_handler_cb(self, cb):
+        if not callable(cb):
+            raise EventParamError("handler must be callable")
+
+        self._handler_cb = cb
+
+    def _get_handler_cb(self):
+        return self._handler_cb
+
+    handler_cb = property(lambda h: h._get_handler_cb(),
+                          lambda h, cb: h._set_handler_cb(cb))
+
+    #
+    # notify_cb is a notification handler that runs after the main handler
+    #
+    def _set_notify_cb(self, cb):
+        if not callable(cb) or cb.func_code.argcount < 1:
+            raise EventParamError("callback function must accept at least one arg")
+
+        self._notify_cb = cb
+
+    def _get_notify_cb(self):
+        return self._notify_cb
+
+    notify_cb = property(lambda h: h._get_notify_cb(),
+                         lambda h, cb: h._set_notify_cb(cb))
+
+    @abc.abstractproperty
+    def enabled(self):
+        return False
+
+    @abc.abstractmethod
+    def enable(self):
+        """ Enable monitoring and handling of events.
+
+            :raises: :class:`~.errors.EventManagerError` if no callback defined
+        """
+        if self.handler_cb is None:
+            raise EventManagerError("cannot enable handler with no callback")
+
+    @abc.abstractmethod
+    def disable(self):
+        """ Disable monitoring and handling of events. """
+        pass
+
+    def mask_event(self, device=None, action=None, count=1):
+        """ Ignore future events.
+
+            :keyword str device: ignore events on the named device
+            :keyword str action: ignore events of the specified type
+            :keyword int count: number of events to ignore
+        """
+        self._queue.mask_add(device=device, action=action, count=count)
+
+    def next_event(self):
+        return self._queue.dequeue()
+
+    @property
+    def events_pending(self):
+        return len(self._queue._queue)
+
+    @abc.abstractmethod
+    def enqueue_event(self, *args, **kwargs):
+        """ Convert an event into :class:`~.Event` and enqueue it. """
+        pass
+
+    def handle_event(self, *args, **kwargs):
+        """ Enqueue an event and call the configured handler. """
+        self.enqueue_event(*args, **kwargs)
+        Timer(0, self.handler_cb).start()
+
+class UdevEventManager(EventManager):
+    def __init__(self, handler_cb=None, notify_cb=None):
+        super(UdevEventManager, self).__init__(handler_cb=handler_cb,
+                                               notify_cb=notify_cb)
+        self._pyudev_observer = None
+
+    def __deepcopy__(self, memo):
+        return util.variable_copy(self, memo, shallow=('_pyudev_observer'))
+
+    @property
+    def enabled(self):
+        return self._pyudev_observer and self._pyudev_observer.monitor.started
+
+    def enable(self):
+        """ Enable monitoring and handling of block device uevents. """
+        super(UdevEventManager, self).enable()
+        monitor = pyudev.Monitor.from_netlink(udev.global_udev)
+        monitor.filter_by("block")
+        self._pyudev_observer = pyudev.MonitorObserver(monitor,
+                                                       self.handle_event)
+        self._pyudev_observer.start()
+        flags.uevents = True
+
+    def disable(self):
+        """ Disable monitoring and handling of block device uevents. """
+        if self._pyudev_observer:
+            self._pyudev_observer.stop()
+            self._pyudev_observer = None
+
+        flags.uevents = False
+
+    def enqueue_event(self, *args, **kwargs):
+        event = UdevEvent(args[0], args[1])
+        log.debug("-> %s", event)
+        self._queue.enqueue(event)
+
+    def handle_event(self, *args, **kwargs):
+        """ Enqueue a uevent and call the configured handler. """
+        current_thread = currentThread()
+        current_thread.name = current_thread.name.replace("Thread-",
+                                                          "EventManager")
+        super(UdevEventManager, self).handle_event(args[0], args[1])
diff --git a/blivet/flags.py b/blivet/flags.py
index 2e5e47e..3d7fb6d 100644
--- a/blivet/flags.py
+++ b/blivet/flags.py
@@ -31,6 +31,11 @@ class Flags(object):
         self.installer_mode = False
 
         #
+        # minor modes
+        #
+        self.uevents = False
+
+        #
         # minor modes (installer-specific)
         #
         self.automated_install = False
diff --git a/tests/events_test.py b/tests/events_test.py
new file mode 100644
index 0000000..c4a0878
--- /dev/null
+++ b/tests/events_test.py
@@ -0,0 +1,88 @@
+
+import unittest
+
+from blivet.event import Event, EventQueue
+
+class FakeEvent(Event):
+    @property
+    def device(self):
+        return self.info
+
+class EventQueueTest(unittest.TestCase):
+    def testEventQueue(self):
+        q = EventQueue()
+
+        # should be initially empty
+        self.assertEqual(list(q), [])
+
+        e1 = FakeEvent('add', 'sdc')
+        q.enqueue(e1)
+        self.assertEqual(list(q), [e1])
+
+        e2 = FakeEvent('add', 'sdd')
+        q.enqueue(e2)
+        self.assertEqual(list(q), [e1, e2])
+
+        e1d = q.dequeue()
+        self.assertEqual(e1d, e1)
+        self.assertEqual(list(q), [e2])
+
+        q.mask_add(device='sdc', action='change', count=1)
+        e3 = FakeEvent('change', 'sdc')
+
+        # should get enqueued since it isn't a blacklist match (action)
+        e4 = FakeEvent('add', 'sdc')
+        q.enqueue(e4)
+        self.assertEqual(list(q), [e2, e4])
+
+        # should get enqueued since it isn't a blacklist match (device)
+        e5 = FakeEvent('change', 'sde')
+        q.enqueue(e5)
+        self.assertEqual(list(q), [e2, e4, e5])
+
+        # enqueuing e3 should be a no-op
+        q.enqueue(e3)
+        self.assertEqual(list(q), [e2, e4, e5])
+
+        # now that the blacklist was hit once, enqueuing e3 should work normally
+        q.enqueue(e3)
+        self.assertEqual(list(q), [e2, e4, e5, e3])
+
+        # omitting device or action should mean "any device" or "any action",
+        # respectively
+        q.mask_add(device='sdc', count=2)
+        q.enqueue(e1)
+        self.assertEqual(list(q), [e2, e4, e5, e3])
+        q.enqueue(e1)
+        self.assertEqual(list(q), [e2, e4, e5, e3])
+        # the blacklist entry should be dropped after being hit twice
+        q.enqueue(e1)
+        self.assertEqual(list(q), [e2, e4, e5, e3, e1])
+
+        q.mask_add(action='remove', count=1)
+        e6 = FakeEvent('remove', 'sdd2')
+        # enqueue will be a no-op since event.action == "remove"
+        q.enqueue(e6)
+        self.assertEqual(list(q), [e2, e4, e5, e3, e1])
+        # now the blacklist entry should be removed since it was hit once,
+        # making the next enqueue work as expected (no blacklist)
+        q.enqueue(e6)
+        self.assertEqual(list(q), [e2, e4, e5, e3, e1, e6])
+
+        # shorten the queue to save some typing below
+        _ignoreme = q.dequeue()
+        _ignoreme = q.dequeue()
+        _ignoreme = q.dequeue()
+        _ignoreme = q.dequeue()
+        self.assertEqual(list(q), [e1, e6])
+
+        # blacklist entries with a count of zero are permanent
+        e7 = FakeEvent('add', 'sdg')
+        q.mask_add(device='sdg', action='add', count=0)
+        for _i in range(300):
+            q.enqueue(e7)
+            self.assertEqual(list(q), [e1, e6])
+
+if __name__ == "__main__":
+    unittest.main()
+
-- 
2.1.0

