From ccb61361fc71a8e5aaaafbee999842dd67d8b5cd Mon Sep 17 00:00:00 2001
From: David Lehman <dlehman@redhat.com>
Date: Mon, 2 Feb 2015 18:41:52 -0600
Subject: [PATCH 21/21] Create an event handling class hierarchy.

---
 blivet/devicetree.py | 103 +++++++++++++++++++-----------------------------
 blivet/errors.py     |   3 ++
 blivet/event.py      | 109 +++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 153 insertions(+), 62 deletions(-)
 create mode 100644 blivet/event.py

diff --git a/blivet/devicetree.py b/blivet/devicetree.py
index 21c75c1..262d10b 100644
--- a/blivet/devicetree.py
+++ b/blivet/devicetree.py
@@ -27,9 +27,9 @@ import shutil
 import pprint
 import copy
 from six import add_metaclass
-import pyudev
+import time
 
-from .errors import CryptoError, DeviceError, DeviceTreeError, DiskLabelCommitError, FSError, InvalidDiskLabelError, LUKSError, MDRaidError, StorageError, UnusableConfigurationError
+from .errors import CryptoError, DeviceError, DeviceTreeError, DiskLabelCommitError, EventQueueEmptyError, FSError, InvalidDiskLabelError, LUKSError, MDRaidError, StorageError, UnusableConfigurationError
 from .devices import BTRFSDevice, BTRFSSubVolumeDevice, BTRFSVolumeDevice, BTRFSSnapShotDevice
 from .devices import DASDDevice, DMDevice, DMLinearDevice, DMRaidArrayDevice, DiskDevice
 from .devices import FcoeDiskDevice, FileDevice, LoopDevice, LUKSDevice
@@ -58,6 +58,7 @@ from .storage_log import log_exception_info, log_method_call, log_method_return
 from .i18n import _
 from .size import Size
 from .threads import blivet_lock, SynchronizedMeta
+from .event import UdevEventHandler
 
 import logging
 log = logging.getLogger("blivet")
@@ -82,7 +83,7 @@ class DeviceTree(object):
         :class:`~.deviceaction.DeviceAction` instances can only be registered
         for leaf devices, except for resize actions.
     """
-    _unsynchronized_methods = ['processActions']
+    _unsynchronized_methods = ['processActions']#, 'ueventCB']
 
     def __init__(self, conf=None, passphrase=None, luksDict=None,
                  iscsi=None, dasd=None):
@@ -153,8 +154,7 @@ class DeviceTree(object):
 
         self._cleanup = False
 
-        self.ueventNotifyCB = None
-        self._pyudev_observer = None
+        self.eventHandler = UdevEventHandler(handler=self.handleUevent)
 
     def setDiskImages(self, images):
         """ Set the disk images and reflect them in exclusiveDisks.
@@ -358,10 +358,11 @@ class DeviceTree(object):
         :type callbacks: :class:`~.callbacks.DoItCallbacks`
 
         """
-
+        self.eventHandler.enable()
         self._preProcessActions()
 
         for action in self._actions[:]:
+            print action
             log.info("executing action: %s", action)
             if dryRun:
                 continue
@@ -389,61 +390,38 @@ class DeviceTree(object):
 
                 self._completed_actions.append(self._actions.pop(0))
 
+            # give up some cycles to handle events
+            print "sleeping..."
+            time.sleep(1)
+            print "time to make the donuts"
+
         self._postProcessActions()
 
     ##
     ## uevent handling
     ##
-    def ueventCB(self, action, info):
-        print action, udev.device_get_name(info)
-        notify = False
-        device = None
-        if action == "add":
-            device = self.deviceAddedCB(info)
-            notify = True
+    def handleUevent(self):
+        """ Handle the next uevent in the queue. """
+        log_method_call(self)
+        try:
+            event = self.eventHandler.queue.dequeue()
+        except EventQueueEmptyError:
+            log.debug("uevent queue is empty")
+            return
+
+        log.debug("event: %s", event)
+        if event.action == "add":
+            self.deviceAddedCB(event.info)
             # TODO: update self.roots if in installer_mode
-        elif action == "remove":
-            device = self.deviceRemovedCB(info)
-            notify = True
+        elif event.action == "remove":
+            self.deviceRemovedCB(event.info)
             # TODO: update self.roots
-        elif action == "change":
-            device = self.deviceChangedCB(info)
-            notify = True
+        elif event.action == "change":
+            self.deviceChangedCB(event.info)
         else:
-            log.info("unknown uevent action: %s (%s)", action, info.sys_name)
-
-        if notify and self.ueventNotifyCB:
-            self.ueventNotifyCB(action, device)
-
-    def enableUeventMonitoring(self):
-        """ Enable monitoring and handling of block device uevents. """
-        monitor = pyudev.Monitor.from_netlink(udev.global_udev)
-        monitor.filter_by("block")
-        self._pyudev_observer = pyudev.MonitorObserver(monitor, self.ueventCB)
-        self._pyudev_observer.start()
-
-    def disableUeventMonitoring(self):
-        """ Disable monitoring and handling of block device uevents. """
-        if self._pyudev_observer:
-            self._pyudev_observer.stop()
-            self._pyudev_observer = None
-
-    def registerUeventNotifyCB(self, cb):
-        """ Register a uevent notification callback.
-
-            The callback will be run after processing the uevent, and will be
-            passed the action string and the StorageDevice instance that was
-            operated on.
-
-            Note that in the event of a remove action the returned device will
-            no longer be in the devicetree. Also, it is possible that the the
-            device will be hidden in the event of an add or remove action,
-            depending on the disk filtering settings of the devicetree.
-        """
-        if not callable(cb) or cb.func_code.co_argcount < 2:
-            raise ValueError("callback function must accept at least two args")
+            log.info("unknown event: %s", event)
 
-        self.ueventNotifyCB = cb
+        print "done"
 
     def deviceAddedCB(self, info, force=False):
         """ Handle an "add" uevent on a block device.
@@ -475,8 +453,6 @@ class DeviceTree(object):
         if device:
             self._hideIgnoredDisks()
 
-        return device
-
     def deviceChangedCB(self, info):
         """ Handle a "changed" uevent on a block device. """
         log.debug("device changed: %s", udev.device_get_sysfs_path(info))
@@ -495,6 +471,10 @@ class DeviceTree(object):
             (udev.device_is_md(new_info) or udev.device_is_dm(new_info))):
             return self.deviceAddedCB(info, force=True)
 
+        if not device:
+            log.warning("device not found: %s", udev.device_get_name(info))
+            return
+
         if not device.format.exists:
             #
             # A policy must be decided upon for external events that conflict
@@ -503,7 +483,7 @@ class DeviceTree(object):
             log.warning("ignoring change uevent on device with reformat scheduled")
             return
 
-        with device.lock:
+        with blivet_lock:
             #
             # Check for changes to the device itself.
             #
@@ -514,6 +494,7 @@ class DeviceTree(object):
                 device.name = name
 
             # resize
+            # FIXME: only do this if the size seems to have changed
             device.updateSize()
 
             #
@@ -522,10 +503,11 @@ class DeviceTree(object):
             uuid = udev.device_get_uuid(new_info)
             label = udev.device_get_label(new_info)
 
+            # FIXME: set an attr name in the conditional and calculate only once
             if device.format.type != "btrfs":
-                uuid_changed = (device.format.uuid != uuid)
+                uuid_changed = device.format.uuid and (device.format.uuid != uuid)
             else:
-                uuid_changed = (device.format.volUUID != uuid)
+                uuid_changed = device.format.volUUID and (device.format.volUUID != uuid)
             partitioned = (device.partitionable and
                            new_info.get("ID_PART_TABLE_TYPE"))
             new_type = getFormat(udev.device_get_format(new_info)).type
@@ -663,8 +645,6 @@ class DeviceTree(object):
 
                 # TODO: update mountopts?
 
-        return device
-
     def deviceRemovedCB(self, info):
         """ Handle a "remove" uevent on a block device.
 
@@ -2315,6 +2295,7 @@ class DeviceTree(object):
                                      sysfsPath=loop_sysfs,
                                      exists=True)
                 loopdev.setup()
+                loopdev.updateSize()
                 log.debug("%s", loopdev)
                 dmdev = DMLinearDevice(name,
                                        dmUuid="ANACONDA-%s" % name,
@@ -2322,6 +2303,7 @@ class DeviceTree(object):
                                        exists=True)
                 dmdev.setup()
                 dmdev.updateSysfsPath()
+                dmdev.updateSize()
                 log.debug("%s", dmdev)
             except (ValueError, DeviceError) as e:
                 log.error("failed to set up disk image: %s", e)
@@ -2978,9 +2960,6 @@ class DeviceTree(object):
                 device.format._mountpoint = mountpoint  # active mountpoint
                 device.format.mountopts = options
 
-    def __deepcopy__(self, memo):
-        return util.variable_copy(self, memo, shallow=('_pyudev_observer'))
-
     def __str__(self):
         done = []
         def get_depth(device):
diff --git a/blivet/errors.py b/blivet/errors.py
index 34d6fd1..e41fe76 100644
--- a/blivet/errors.py
+++ b/blivet/errors.py
@@ -192,3 +192,6 @@ class UnknownSourceDeviceError(StorageError):
 # factories
 class DeviceFactoryError(StorageError):
     pass
+
+class EventQueueEmptyError(StorageError):
+    pass
diff --git a/blivet/event.py b/blivet/event.py
new file mode 100644
index 0000000..a5cbb3b
--- /dev/null
+++ b/blivet/event.py
@@ -0,0 +1,109 @@
+
+from collections import deque
+from threading import RLock, Timer
+import pyudev
+
+from . import udev
+from . import util
+from .errors import EventQueueEmptyError
+
+""" delay in seconds between receiving an event and calling the handler """
+EVENT_HANDLER_DELAY = 1
+
+class Event(object):
+    def __init__(self, action, info):
+        self.action = action
+        self.info = info
+
+    def __str__(self):
+        return "%s %s" % (self.action, self.device)
+
+    @property
+    def device(self):
+        """ The (friendly) basename of the device this event operated on. """
+        return udev.device_get_name(self.info)
+
+class EventQueue(object):
+    def __init__(self):
+        self._queue = deque()
+        self._lock = RLock()
+
+    def enqueue(self, event):
+        with self._lock:
+            self._queue.append(event)
+
+    def dequeue(self):
+        """ Dequeue and return the next event, skipping over repeated events.
+
+            :returns: the next uevent
+            :rtype: :class:`~.Event`
+            :raises class:`~.errors.EventQueueEmptyError` if the queue is empty
+
+            When a series of consecutive duplicate events is found, only the
+            last event of the sequence is processed.
+        """
+        with self._lock:
+            if not self._queue:
+                raise EventQueueEmptyError()
+
+            #return self._queue.popleft()
+            ret = None
+            while self._queue:
+                next_event = self._queue[0]
+                if (ret is not None and
+                    (next_event.action != ret.action or
+                     next_event.device != ret.device)):
+                    break
+
+                ret = self._queue.popleft()
+
+            return ret
+
+class UdevEventHandler(object):
+    def __init__(self, handler, delay=EVENT_HANDLER_DELAY):
+        self.handler = handler
+        self.delay = delay
+        self._notify_cb = None
+        self._pyudev_observer = None
+        self._queue = EventQueue()
+
+    def __deepcopy__(self, memo):
+        return util.variable_copy(self, memo, shallow=('_pyudev_observer'))
+
+    @property
+    def queue(self):
+        return self._queue
+
+    @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. """
+        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()
+
+    def disable(self):
+        """ Disable monitoring and handling of block device uevents. """
+        if self._pyudev_observer:
+            self._pyudev_observer.stop()
+            self._pyudev_observer = None
+
+    def register_notify_cb(self, cb):
+        """ Register an event notification callback.
+
+            The callback will be run after processing the event, and will be
+            passed an :class:`~.Event` instance.
+        """
+        if not callable(cb) or cb.func_code.co_argcount < 1:
+            raise ValueError("callback function must accept at least one arg")
+
+        self._notify_cb = cb
+
+    def handle_event(self, action, info):
+        """ Enqueue a uevent and handle it as soon as we can get the lock. """
+        self._queue.enqueue(Event(action, info))
+        Timer(self.delay, self.handler).start()
-- 
1.9.3

