From fafed58498bd0ce1151589bf9924bed135efc2b6 Mon Sep 17 00:00:00 2001
From: David Lehman <dlehman@redhat.com>
Date: Tue, 14 Apr 2015 14:47:34 -0500
Subject: [PATCH 26/51] Use udev to determine when an operation has completed.

Use StorageEventSynchronizer to manage synchronization with the
uevent handler thread.
---
 blivet/devices/btrfs.py     |   3 +
 blivet/devices/container.py |  24 +++++++
 blivet/devices/dm.py        |   1 +
 blivet/devices/file.py      |   1 +
 blivet/devices/luks.py      |   4 ++
 blivet/devices/lvm.py       |  22 +++++-
 blivet/devices/md.py        |  10 ++-
 blivet/devices/nfs.py       |   1 +
 blivet/devices/nodev.py     |   1 +
 blivet/devices/partition.py |  16 ++++-
 blivet/devices/storage.py   | 163 ++++++++++++++++++++++++++++++++++++++++++--
 blivet/formats/__init__.py  |  80 ++++++++++++++++++++--
 blivet/formats/disklabel.py |  37 +++++++++-
 blivet/formats/fs.py        |  34 ++++++++-
 blivet/formats/luks.py      |   4 ++
 blivet/formats/lvmpv.py     |  22 ++++--
 blivet/formats/mdraid.py    |   6 ++
 blivet/formats/prepboot.py  |  21 +++---
 18 files changed, 411 insertions(+), 39 deletions(-)

diff --git a/blivet/devices/btrfs.py b/blivet/devices/btrfs.py
index 987a86b..31e975b 100644
--- a/blivet/devices/btrfs.py
+++ b/blivet/devices/btrfs.py
@@ -581,6 +581,9 @@ class BTRFSSubVolumeDevice(BTRFSDevice, RaidDevice):
         if self not in self.volume.subvolumes:
             self.volume._addSubVolume(self)
 
+    controlSync = property(lambda d: d.volume.controlSync)
+    modifySync = property(lambda d: d.volume.modifySync)
+
     def populateKSData(self, data):
         super(BTRFSSubVolumeDevice, self).populateKSData(data)
         data.subvol = True
diff --git a/blivet/devices/container.py b/blivet/devices/container.py
index a625e46..e00ef81 100644
--- a/blivet/devices/container.py
+++ b/blivet/devices/container.py
@@ -26,6 +26,7 @@ from six import add_metaclass
 from .. import errors
 from ..storage_log import log_method_call
 from ..formats import get_device_format_class
+from ..synchronizer import StorageEventSynchronizerSet
 from ..threads import SynchronizedABCMeta
 
 import logging
@@ -61,6 +62,7 @@ class ContainerDevice(StorageDevice):
         if not self.formatClass:
             raise errors.StorageError("cannot find '%s' class" % self._formatClassName)
 
+        self._parentSyncSet = None
         super(ContainerDevice, self).__init__(*args, **kwargs)
 
     def _verifyMemberUuid(self, member, expect_equality=True, require_existence=True):
@@ -123,6 +125,11 @@ class ContainerDevice(StorageDevice):
             raise ValueError("cannot add member with mismatched UUID")
 
         super(ContainerDevice, self)._addParent(member)
+        self._clearEventSyncs()
+
+    def _removeParent(self, member):
+        super(ContainerDevice, self)._removeParent(member)
+        self._clearEventSyncs()
 
     @abc.abstractmethod
     def _add(self, member):
@@ -191,3 +198,20 @@ class ContainerDevice(StorageDevice):
 
     def updateSize(self, newsize=None):
         pass
+
+    # Container device operations manifest as events on member devices. We use
+    # the parents' controlSync since it monitors events on the device itself
+    # instead of delegating to parents whenever possible (IFF the device has a
+    # device node).
+    def _getParentSyncSet(self):
+        if self._parentSyncSet is None:
+            synchronizers = [p.controlSync for p in self.parents]
+            self._parentSyncSet = StorageEventSynchronizerSet(synchronizers)
+
+        return self._parentSyncSet
+
+    def _clearEventSyncs(self):
+        self._parentSyncSet = None
+
+    controlSync = property(lambda c: c._getParentSyncSet())
+    modifySync = property(lambda c: c._getParentSyncSet())
diff --git a/blivet/devices/dm.py b/blivet/devices/dm.py
index f980f64..b94211e 100644
--- a/blivet/devices/dm.py
+++ b/blivet/devices/dm.py
@@ -207,6 +207,7 @@ class DMLinearDevice(DMDevice):
             return
 
         log.debug("not tearing down dm-linear device %s", self.name)
+        self.controlSync.reset()
 
     @property
     def description(self):
diff --git a/blivet/devices/file.py b/blivet/devices/file.py
index f104568..70639ea 100644
--- a/blivet/devices/file.py
+++ b/blivet/devices/file.py
@@ -39,6 +39,7 @@ class FileDevice(StorageDevice):
     """
     _type = "file"
     _devDir = ""
+    _udev = False
 
     def __init__(self, path, fmt=None, size=None,
                  exists=False, parents=None):
diff --git a/blivet/devices/luks.py b/blivet/devices/luks.py
index 197926f..87970e4 100644
--- a/blivet/devices/luks.py
+++ b/blivet/devices/luks.py
@@ -67,6 +67,10 @@ class LUKSDevice(DMCryptDevice):
             size = self.currentSize
         return size
 
+    def _setup(self, orig=False):
+        # this makes the post-setup sync happen
+        super(LUKSDevice, self)._setup(orig=orig)
+
     def _postCreate(self):
         self.name = self.slave.format.mapName
         StorageDevice._postCreate(self)
diff --git a/blivet/devices/lvm.py b/blivet/devices/lvm.py
index bc3b8d1..acc51e1 100644
--- a/blivet/devices/lvm.py
+++ b/blivet/devices/lvm.py
@@ -53,6 +53,9 @@ class LVMVolumeGroupDevice(ContainerDevice):
     _formatClassName = property(lambda s: "lvmpv")
     _formatUUIDAttr = property(lambda s: "vgUuid")
     _formatImmutable = True
+    # Sometimes a vg teardown does not generate a uevent on any pv, so we have
+    # to set a timeout.
+    _postTeardownWaitTimeout = 0.2
 
     @staticmethod
     def get_supported_pe_sizes():
@@ -694,8 +697,19 @@ class LVMLogicalVolumeDevice(DMDevice):
         if self.format.exists:
             self.format.teardown()
 
-        udev.settle()
-        blockdev.lvm.lvresize(self.vg.name, self._name, self.size)
+        if not flags.uevents:
+            udev.settle()
+
+        try:
+            blockdev.lvm.lvresize(self.vg.name, self._name, self.size)
+        except Exception:
+            raise
+        else:
+            event_sync.set_ready()
+            event_sync.wait()
+        finally:
+            event_sync.reset()
+            event_sync.notify()
 
     @property
     def isleaf(self):
@@ -751,6 +765,10 @@ class LVMLogicalVolumeDevice(DMDevice):
         if self not in self.vg.lvs:
             self.vg._addLogVol(self)
 
+    # lv create/destroy manifest as events on PVs
+    modifySync = property(lambda d: d.vg.modifySync,
+                          doc="control synchronization for create/destroy")
+
     def populateKSData(self, data):
         super(LVMLogicalVolumeDevice, self).populateKSData(data)
         data.vgname = self.vg.name
diff --git a/blivet/devices/md.py b/blivet/devices/md.py
index ec9aa64..f7499cf 100644
--- a/blivet/devices/md.py
+++ b/blivet/devices/md.py
@@ -533,6 +533,10 @@ class MDRaidArrayDevice(ContainerDevice, RaidDevice):
     def model(self):
         return self.description
 
+    # md arrays use events on the members for create/destroy and events on the
+    # array device for setup/teardown
+    controlSync = property(lambda d: d._getEventSync())
+
     def dracutSetupArgs(self):
         return set(["rd.md.uuid=%s" % self.mdadmFormatUUID])
 
@@ -579,12 +583,13 @@ class MDContainerDevice(MDRaidArrayDevice):
         log_method_call(self, self.name, status=self.status,
                         controllable=self.controllable)
         # we don't really care about the return value of _preTeardown here.
-        # see comment just above md_deactivate call
+        # see comment just above mddeactivate call
         self._preTeardown(recursive=recursive)
 
         # Since BIOS RAID sets (containers in mdraid terminology) never change
         # there is no need to stop them and later restart them. Not stopping
         # (and thus also not starting) them also works around bug 523334
+        self.controlSync.reset()
         return
 
     @property
@@ -637,10 +642,11 @@ class MDBiosRaidArrayDevice(MDRaidArrayDevice):
         log_method_call(self, self.name, status=self.status,
                         controllable=self.controllable)
         # we don't really care about the return value of _preTeardown here.
-        # see comment just above md_deactivate call
+        # see comment just above mddeactivate call
         self._preTeardown(recursive=recursive)
 
         # Since BIOS RAID sets (containers in mdraid terminology) never change
         # there is no need to stop them and later restart them. Not stopping
         # (and thus also not starting) them also works around bug 523334
+        self.controlSync.reset()
         return
diff --git a/blivet/devices/nfs.py b/blivet/devices/nfs.py
index f85a05d..d8840ea 100644
--- a/blivet/devices/nfs.py
+++ b/blivet/devices/nfs.py
@@ -31,6 +31,7 @@ class NFSDevice(StorageDevice, NetworkStorageDevice):
     """ An NFS device """
     _type = "nfs"
     _packages = ["dracut-network"]
+    _udev = False
 
     def __init__(self, device, fmt=None, parents=None):
         """
diff --git a/blivet/devices/nodev.py b/blivet/devices/nodev.py
index 6d7570a..094d338 100644
--- a/blivet/devices/nodev.py
+++ b/blivet/devices/nodev.py
@@ -30,6 +30,7 @@ from .storage import StorageDevice
 class NoDevice(StorageDevice):
     """ A nodev device for nodev filesystems like tmpfs. """
     _type = "nodev"
+    _udev = False
 
     def __init__(self, fmt=None):
         """
diff --git a/blivet/devices/partition.py b/blivet/devices/partition.py
index d980479..a292e5a 100644
--- a/blivet/devices/partition.py
+++ b/blivet/devices/partition.py
@@ -568,8 +568,9 @@ class PartitionDevice(StorageDevice):
         self.disk.format.addPartition(self.partedPartition.geometry.start,
                                       self.partedPartition.geometry.end,
                                       self.partedPartition.type)
-
+        self.modifySync.reset()
         self._wipe()
+        self.modifySync.creating = True
         try:
             self.disk.format.commit()
         except errors.DiskLabelCommitError:
@@ -647,7 +648,18 @@ class PartitionDevice(StorageDevice):
                                         start=geometry.start,
                                         end=geometry.end)
 
-        self.disk.format.commit()
+        event_sync = self.modifySync
+        event_sync.resizing = True
+        try:
+            self.disk.format.commit()
+        except Exception:
+            raise
+        else:
+            event_sync.wait()
+        finally:
+            event_sync.reset()
+            event_sync.notify()
+
         self.updateSize()
 
     def _preDestroy(self):
diff --git a/blivet/devices/storage.py b/blivet/devices/storage.py
index dd9dd82..3c47818 100644
--- a/blivet/devices/storage.py
+++ b/blivet/devices/storage.py
@@ -31,6 +31,7 @@ from ..storage_log import log_method_call
 from .. import udev
 from ..formats import getFormat
 from ..size import Size
+from ..synchronizer import StorageEventSynchronizer
 
 import logging
 log = logging.getLogger("blivet")
@@ -57,6 +58,12 @@ class StorageDevice(Device):
     _isDisk = False
     _encrypted = False
 
+    _udev = True
+    """ Whether udev manages device nodes for devices of this type. """
+
+    _postTeardownWaitTimeout = None
+    """ Timeout value for wait call following teardown. """
+
     def __init__(self, name, fmt=None, uuid=None,
                  size=None, major=None, minor=None,
                  sysfsPath='', parents=None, exists=False, serial=None,
@@ -128,6 +135,9 @@ class StorageDevice(Device):
 
         self.deviceLinks = []
 
+        """ Used to synchronize access with uevent handlers. """
+        self._eventSync = None # use the same instance for all operations
+
         if self.exists and self.status:
             self.updateSize()
 
@@ -343,6 +353,7 @@ class StorageDevice(Device):
             return False
 
         self.setupParents(orig=orig)
+        self.controlSync.starting = True
         return True
 
     def _setup(self, orig=False):
@@ -356,14 +367,27 @@ class StorageDevice(Device):
         if not self._preSetup(orig=orig):
             return
 
-        self._setup(orig=orig)
+        try:
+            self._setup(orig=orig)
+        except Exception:
+            # don't cause a deadlock if the failure generated an event
+            self.controlSync.wait(timeout=1)
+            self.controlSync.notify()
+            self.controlSync.reset()
+            raise
         self._postSetup()
 
     def _postSetup(self):
         """ Perform post-setup operations. """
+        event_sync = self.controlSync
+        if self.__class__._setup != StorageDevice._setup:
+            event_sync.wait()
+
+        event_sync.reset()
+        event_sync.notify()
         udev.settle()
         self.updateSysfsPath()
-        # the device may not be set up when we want information about it
+        # the device may not be set up when we want information about it later
         if self._size == Size(0):
             self.updateSize()
 
@@ -385,7 +409,10 @@ class StorageDevice(Device):
             self.originalFormat.teardown()
         if self.format.exists:
             self.format.teardown()
+
         udev.settle()
+
+        self.controlSync.stopping = True
         return True
 
     def _teardown(self, recursive=None):
@@ -399,11 +426,27 @@ class StorageDevice(Device):
         if not self._preTeardown(recursive=recursive):
             return
 
-        self._teardown(recursive=recursive)
+        try:
+            self._teardown(recursive=recursive)
+        except Exception:
+            # don't cause a deadlock if the failure generated an event
+            self.controlSync.wait(timeout=1)
+            self.controlSync.notify()
+            self.controlSync.reset()
+            raise
+
         self._postTeardown(recursive=recursive)
 
     def _postTeardown(self, recursive=None):
         """ Perform post-teardown operations. """
+        event_sync = self.controlSync
+        if self.__class__._teardown != StorageDevice._teardown:
+            event_sync.wait(timeout=self._postTeardownWaitTimeout)
+
+        event_sync.reset()
+        event_sync.notify()
+        self.sysfsPath = ""
+
         if recursive:
             self.teardownParents(recursive=recursive)
 
@@ -416,6 +459,7 @@ class StorageDevice(Device):
             raise errors.DeviceError("device has already been created", self.name)
 
         self.setupParents()
+        self.modifySync.creating = True
 
     def _create(self):
         """ Perform device-specific create operations. """
@@ -425,21 +469,56 @@ class StorageDevice(Device):
         """ Create the device. """
         log_method_call(self, self.name, status=self.status)
         self._preCreate()
-        self._create()
+        try:
+            self._create()
+        except Exception:
+            # don't cause a deadlock if the failure generated an event
+            self.modifySync.notify()
+            self.modifySync.wait(timeout=1)
+            self.modifySync.notify()
+            self.modifySync.reset()
+            raise
         self._postCreate()
 
     def _postCreate(self):
         """ Perform post-create operations. """
+        # XXX what about devices that don't get started automatically when
+        #     they are created
+        event_sync = self.modifySync
+        if self.__class__._create != StorageDevice._create:
+            # There is an extra notify/wait cycle for device creation. This is
+            # because some device types (partition) have so much pre/post code
+            # that it becomes likely that the event handler will notify the
+            # event sync before the device is waiting for that notification,
+            # which leads to a deadlock when the device waits for notification.
+            #
+            # It seemed better to add an extra wait with a short (~1s) timeout
+            # in the event handler than to use a timeout here and potentially
+            # run along before any event is received.
+            #
+            # If the timeout is here, we're assuming that reaching this point
+            # means an event has been generated and is being handled -- the
+            # second part there is the tricky one.
+            # If the timeout is there, we're assuming that reaching that point
+            # means that the device will be ready for post-processing within the
+            # timeout.
+            # The latter seems like a safer assumption to me.
+            event_sync.notify() # notify the handler we're ready to post-process
+            event_sync.wait() # wait for notification the event was received
+
+        event_sync.reset()
         self.exists = True
+        event_sync.notify() # notify event handler we're done post-processing
+
         self.setup()
         self.updateSysfsPath()
         udev.settle()
 
+        self._updateNetDevMountOption()
+
         # make sure that targetSize is updated to reflect the actual size
         self.updateSize()
 
-        self._updateNetDevMountOption()
-
     #
     # destroy
     #
@@ -452,6 +531,7 @@ class StorageDevice(Device):
             raise errors.DeviceError("Cannot destroy non-leaf device", self.name)
 
         self.teardown()
+        self.modifySync.destroying = True
 
     def _destroy(self):
         """ Perform device-specific destruction operations. """
@@ -461,12 +541,30 @@ class StorageDevice(Device):
         """ Destroy the device. """
         log_method_call(self, self.name, status=self.status)
         self._preDestroy()
-        self._destroy()
+        try:
+            self._destroy()
+        except Exception:
+            # don't cause a deadlock if the failure generated an event
+            self.modifySync.notify()
+            self.modifySync.wait(timeout=1)
+            self.modifySync.notify()
+            self.modifySync.reset()
+            raise
         self._postDestroy()
 
     def _postDestroy(self):
         """ Perform post-destruction operations. """
+        event_sync = self.modifySync
+        if self.__class__._destroy != StorageDevice._destroy:
+            # See comment in _postCreate, above for explanation of the extra
+            # notify here as compared to setup and teardown.
+            event_sync.notify() # notify the handler we're ready to post-process
+            event_sync.wait() # wait for notification that event was received
+
+        event_sync.reset()
         self.exists = False
+        event_sync.notify() # notify the handler we're done post-processing
+        self.sysfsPath = ""
 
     def setupParents(self, orig=False):
         """ Run setup method of all parent devices. """
@@ -719,6 +817,57 @@ class StorageDevice(Device):
         if not new:
             map(lambda p: p.addChild(), self.parents)
 
+    def _getEventSync(self):
+        if self._eventSync is None:
+            self._eventSync = StorageEventSynchronizer(passthrough=not self._udev)
+
+        return self._eventSync
+
+    ##
+    ## Event synchronization managers
+    ##
+    # These manage synchronization between the device methods that initiate
+    # actions on the underlying device and uevent handlers. This allows us to
+    # know exactly when newly created devices appear, for example.
+    #
+    # There are two of them because of variations in behavior among different
+    # device types:
+    #
+    # * Disks and partitions only need one synchronization manager because all
+    #   operations on those devices manifest as uevents on the device itself.
+    # * LVM VGs are different in that there is no block device to represent the
+    #   VG. This means that actions on a VG manifest as uevents on the VG's PVs.
+    # * LVM LVs have a combination of the two aforementioned models. Controlling
+    #   actions (setup, teardown) manifest as uevents on the LV device, while
+    #   modifying actions (create, destroy) events manifest as uevents on the
+    #   PVs.
+    # * MD arrays are similar to LVM LVs except that there is no intermediate
+    #   VG layer (which is really only a passthrough anyway).
+    # * BTRFS volumes behave the same way as LVM VGs.
+    # * BTRFS subvolumes use their top-level volume's synchronization manager,
+    #   meaning they also rely on uevents on the physical member devices.
+    #
+    # It is counter-intuitive to use the controlSync for "changing" actions when
+    # "changing" is synonymous with "modify" (as in "modifySync"). We do this
+    # because we want "changing" to use events on the actual device if that
+    # device has a device node. In every case except for lvm vg and btrfs, the
+    # controlSync is connected to the device itself and is not delegating
+    # to parent devices.
+    controlSync = property(lambda d: d._getEventSync(),
+                           doc="controls synchronization for setup/teardown")
+    modifySync = property(lambda d: d._getEventSync(),
+                          doc="controls synchronization for create/destroy")
+
+    @property
+    def delegateModifyEvents(self):
+        """ Do modify actions for this device manifest on parent devices? """
+        return self.modifySync != self._getEventSync()
+
+    @property
+    def delegateControlEvents(self):
+        """ Do control actions for this device manifest on parent devices? """
+        return self.controlSync != self._getEventSync()
+
     def populateKSData(self, data):
         # the common pieces are basically the formatting
         self.format.populateKSData(data)
diff --git a/blivet/formats/__init__.py b/blivet/formats/__init__.py
index 86ba415..40f92a5 100644
--- a/blivet/formats/__init__.py
+++ b/blivet/formats/__init__.py
@@ -23,6 +23,7 @@
 
 import os
 import importlib
+import stat
 from gi.repository import BlockDev as blockdev
 from six import add_metaclass
 
@@ -34,6 +35,7 @@ from ..storage_log import log_method_call
 from ..errors import DeviceFormatError, FormatCreateError, FormatDestroyError, FormatSetupError
 from ..i18n import N_
 from ..size import Size
+from ..synchronizer import StorageEventSynchronizer, KEY_ABSENT, KEY_PRESENT
 from ..threads import SynchronizedMeta
 
 import logging
@@ -191,6 +193,9 @@ class DeviceFormat(ObjectID):
         self.exists = kwargs.get("exists", False)
         self.options = kwargs.get("options")
 
+        self.eventSync = StorageEventSynchronizer()
+        self.updateSyncPassthrough()
+
     def __repr__(self):
         s = ("%(classname)s instance (%(id)s) object id %(object_id)d--\n"
              "  type = %(type)s  name = %(name)s  status = %(status)s\n"
@@ -312,6 +317,8 @@ class DeviceFormat(ObjectID):
         if error_msg:
             raise ValueError(error_msg)
         self._device = devspec
+        if hasattr(self, "eventSync"):
+            self.updateSyncPassthrough()
 
     def _getDevice(self):
         return self._device
@@ -328,6 +335,15 @@ class DeviceFormat(ObjectID):
     def type(self):
         return self._type
 
+    def updateSyncPassthrough(self):
+        self.eventSync.passthrough = (self.device and
+                               (not os.path.exists(self.device) or
+                                not stat.S_ISBLK(os.stat(self.device).st_mode)))
+
+    @property
+    def createGeneratesEvent(self):
+        return self.__class__._create != DeviceFormat._create
+
     def create(self, **kwargs):
         """ Write the formatting to the specified block device.
 
@@ -344,9 +360,21 @@ class DeviceFormat(ObjectID):
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
         self._preCreate(**kwargs)
-        self._create(**kwargs)
+        try:
+            self._create(**kwargs)
+        except Exception:
+            if self.createGeneratesEvent:
+                self.eventSync.wait(timeout=2)
+                self.eventSync.reset()
+                self.eventSync.notify()
+            raise
         self._postCreate(**kwargs)
 
+    def _setCreateEventInfo(self):
+        """ Specify requirements for udev info to confirm success. """
+        self.eventSync.info_update(ID_FS_TYPE=self._udevTypes[0],
+                                   ID_FS_UUID=KEY_PRESENT)
+
     def _preCreate(self, **kwargs):
         """ Perform checks and setup prior to creating the format. """
         # allow late specification of device path
@@ -365,25 +393,57 @@ class DeviceFormat(ObjectID):
 
     # pylint: disable=unused-argument
     def _create(self, **kwargs):
-        """ Type-specific create method. """
-        pass
+        """ Type-specific create method.
+
+            All this version does is set up event synchronization. Subclasses
+            should either call this when overriding it or set up the event
+            synchronizer themselves.
+        """
+        if self.createGeneratesEvent:
+            self.updateSyncPassthrough()
+            self._setCreateEventInfo()
+            self.eventSync.creating = True
 
     # pylint: disable=unused-argument
     def _postCreate(self, **kwargs):
+        if self.createGeneratesEvent:
+            self.eventSync.wait()
+            self.eventSync.reset()
+            self.eventSync.notify()
+
         self.exists = True
 
+    @property
+    def destroyGeneratesEvent(self):
+        return True
+
     def destroy(self, **kwargs):
         """ Remove the formatting from the associated block device.
 
+            :keyword bool notify: whether to ensure the operation via udev
             :raises: FormatDestroyError
             :returns: None.
         """
+        # pylint: disable=unused-argument
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
         self._preDestroy(**kwargs)
-        self._destroy(**kwargs)
+        try:
+            self._destroy()
+        except Exception:
+            if self.destroyGeneratesEvent:
+                self.eventSync.wait(timeout=2)
+                self.eventSync.reset()
+                self.eventSync.notify()
+            raise
+
         self._postDestroy(**kwargs)
 
+    def _setDestroyEventInfo(self):
+        """ Specify requirements for udev info to confirm success. """
+        self.eventSync.info_update(ID_FS_TYPE=KEY_ABSENT,
+                                   ID_FS_UUID=KEY_ABSENT)
+
     # pylint: disable=unused-argument
     def _preDestroy(self, **kwargs):
         if not self.exists:
@@ -396,6 +456,12 @@ class DeviceFormat(ObjectID):
             raise DeviceFormatError("device path does not exist or is not writable")
 
     def _destroy(self, **kwargs):
+        notify = kwargs.pop("notify", True)
+        if notify and self.destroyGeneratesEvent:
+            self.updateSyncPassthrough()
+            self._setDestroyEventInfo()
+            self.eventSync.destroying = True
+
         rc = 0
         err = ""
         try:
@@ -411,7 +477,13 @@ class DeviceFormat(ObjectID):
             raise FormatDestroyError(msg)
 
     def _postDestroy(self, **kwargs):
+        notify = kwargs.pop("notify", True)
+        if notify and self.destroyGeneratesEvent:
+            self.eventSync.wait()
         self.exists = False
+        self.eventSync.reset()
+        if notify and self.destroyGeneratesEvent:
+            self.eventSync.notify()
 
     def setup(self, **kwargs):
         """ Activate the formatting.
diff --git a/blivet/formats/disklabel.py b/blivet/formats/disklabel.py
index 42f9b4c..5481a40 100644
--- a/blivet/formats/disklabel.py
+++ b/blivet/formats/disklabel.py
@@ -33,6 +33,7 @@ from ..flags import flags
 from ..i18n import _, N_
 from . import DeviceFormat, register_device_format
 from ..size import Size
+from ..synchronizer import KEY_ABSENT, KEY_PRESENT
 
 import logging
 log = logging.getLogger("blivet")
@@ -235,38 +236,68 @@ class DiskLabel(DeviceFormat):
         """ Device status. """
         return False
 
+    def _setCreateEventInfo(self):
+        self.eventSync.info_update(ID_FS_TYPE=KEY_ABSENT,
+                                   ID_PART_TABLE_TYPE=self.labelType,
+                                   ID_PART_TABLE_UUID=KEY_PRESENT)
+
     def _create(self, **kwargs):
         """ Create the device. """
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
+        super(DiskLabel, self)._create(**kwargs)
+
         # We're relying on someone having called resetPartedDisk -- we
         # could ensure a fresh disklabel by setting self._partedDisk to
         # None right before calling self.commit(), but that might hide
         # other problems.
-        self.commit()
+        self.commit(notify=False)
+
+    def _setDestroyEventInfo(self):
+        self.eventSync.info_update(ID_PART_TABLE_TYPE=KEY_ABSENT,
+                                   ID_PART_TABLE_UUID=KEY_ABSENT)
 
-    def commit(self):
-        """ Commit the current partition table to disk and notify the OS. """
+    def commit(self, notify=True):
+        """ Commit the current partition table to disk and notify the OS.
+
+            :keyword bool notify: whether to ensure the operation via udev
+        """
         log_method_call(self, device=self.device,
                         numparts=len(self.partitions))
+        if notify:
+            self.eventSync.changing = True
+        err = False
         try:
             self.partedDisk.commit()
         except parted.DiskException as msg:
+            err = True
             raise DiskLabelCommitError(msg)
         else:
             self.updateOrigPartedDisk()
+        finally:
             udev.settle()
+            if notify:
+                timeout = 2 if err else None
+                self.eventSync.wait(timeout=timeout)
+                self.eventSync.reset()
+                self.eventSync.notify()
 
     def commitToDisk(self):
         """ Commit the current partition table to disk. """
         log_method_call(self, device=self.device,
                         numparts=len(self.partitions))
+        self.eventSync.changing = True
         try:
             self.partedDisk.commitToDevice()
         except parted.DiskException as msg:
             raise DiskLabelCommitError(msg)
         else:
             self.updateOrigPartedDisk()
+            udev.settle()
+            self.eventSync.wait()
+        finally:
+            self.eventSync.reset()
+            self.eventSync.notify()
 
     def addPartition(self, start, end, ptype=None):
         """ Add a partition to the disklabel.
diff --git a/blivet/formats/fs.py b/blivet/formats/fs.py
index 884b4c2..fa756f1 100644
--- a/blivet/formats/fs.py
+++ b/blivet/formats/fs.py
@@ -40,6 +40,7 @@ from ..size import B, KiB, MiB, GiB, KB, MB, GB
 from ..i18n import _, N_
 from .. import udev
 from ..mounts import mountsCache
+from ..synchronizer import KEY_PRESENT
 
 import logging
 log = logging.getLogger("blivet")
@@ -391,6 +392,10 @@ class FS(DeviceFormat):
         if not self.mkfsProg:
             return
 
+    def _setCreateEventInfo(self):
+        self.eventSync.info_update(ID_FS_TYPE=self.mountType,
+                                   ID_FS_UUID=KEY_PRESENT)
+
     def _create(self, **kwargs):
         """ Create the filesystem.
 
@@ -501,10 +506,17 @@ class FS(DeviceFormat):
         else:
             self.targetSize = rounded
 
+        self.eventSync.resizing = True
+
         try:
             ret = util.run_program([self.resizefsProg] + self.resizeArgs)
         except OSError as e:
             raise FSResizeError(e, self.device)
+        else:
+            self.eventSync.wait()
+        finally:
+            self.eventSync.reset()
+            self.eventSync.notify()
 
         if ret:
             raise FSResizeError("resize failed: %s" % ret, self.device)
@@ -541,10 +553,18 @@ class FS(DeviceFormat):
         if not os.path.exists(self.device):
             raise FSError("device does not exist")
 
+        self.eventSync.changing = True
+
         try:
             ret = util.run_program([self.fsckProg] + self._getCheckArgs())
         except OSError as e:
             raise FSError("filesystem check failed: %s" % e)
+        else:
+            # it may not have been opened r/w, so just wait briefly
+            self.eventSync.wait(timeout=2)
+        finally:
+            self.eventSync.reset()
+            self.eventSync.notify()
 
         if self._fsckFailed(ret):
             hdr = _("%(type)s filesystem check failure on %(device)s: ") % \
@@ -793,7 +813,19 @@ class FS(DeviceFormat):
         if not os.path.exists(self.device):
             raise FSError("device does not exist")
 
-        rc = util.run_program(self._labelfs.label_app.setLabelCommand(self))
+        self.eventSync.info_update(ID_FS_LABEL=self.label)
+        self.eventSync.changing = True
+        try:
+            rc = util.run_program(self._labelfs.label_app.setLabelCommand(self))
+        except Exception:
+            raise
+        else:
+            timeout = 2 if rc else None
+            self.eventSync.wait(timeout=timeout)
+        finally:
+            self.eventSync.reset()
+            self.eventSync.notify()
+
         if rc:
             raise FSError("label failed")
 
diff --git a/blivet/formats/luks.py b/blivet/formats/luks.py
index c855649..2460da8 100644
--- a/blivet/formats/luks.py
+++ b/blivet/formats/luks.py
@@ -178,7 +178,11 @@ class LUKS(DeviceFormat):
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
         log.debug("unmapping %s", self.mapName)
+        self.eventSync.stopping = True
         blockdev.crypto.luks_close(self.mapName)
+        self.eventSync.wait(timeout=2)
+        self.eventSync.reset()
+        self.eventSync.notify()
 
     def _preCreate(self, **kwargs):
         super(LUKS, self)._preCreate(**kwargs)
diff --git a/blivet/formats/lvmpv.py b/blivet/formats/lvmpv.py
index 82cdfea..5cfdd2b 100644
--- a/blivet/formats/lvmpv.py
+++ b/blivet/formats/lvmpv.py
@@ -94,23 +94,35 @@ class LVMPhysicalVolume(DeviceFormat):
                   "peStart": self.peStart, "dataAlignment": self.dataAlignment})
         return d
 
+    def _setCreateEventInfo(self):
+        # We can't seem to count on ID_FS_TYPE being set in the info we get
+        # with the change event generated by pvcreate.
+        pass
+
     def _create(self, **kwargs):
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
-
         # Consider use of -Z|--zero
         # -f|--force or -y|--yes may be required
         blockdev.lvm.pvscan(self.device)
-        blockdev.lvm.pvcreate(self.device, data_alignment=self.dataAlignment)
-        blockdev.lvm.pvscan(self.device)
+        super(LVMPhysicalVolume, self)._create(**kwargs)
+        try:
+            blockdev.lvm.pvcreate(self.device, data_alignment=self.dataAlignment)
+        finally:
+            blockdev.lvm.pvscan(self.device)
 
     def _destroy(self, **kwargs):
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
+        self.updateSyncPassthrough()
+        self._setDestroyEventInfo()
+        self.eventSync.destroying = True
+
         try:
             blockdev.lvm.pvremove(self.device)
-        except blockdev.LVMError:
-            DeviceFormat.destroy(self, **kwargs)
+        except GLib.GError:
+            self.eventSync.reset()
+            super(LVMPhysicalVolume, self)._destroy(**kwargs)
         finally:
             blockdev.lvm.pvscan(self.device)
 
diff --git a/blivet/formats/mdraid.py b/blivet/formats/mdraid.py
index 1bef9ce..77e2558 100644
--- a/blivet/formats/mdraid.py
+++ b/blivet/formats/mdraid.py
@@ -27,6 +27,7 @@ from parted import PARTITION_RAID
 from . import DeviceFormat, register_device_format
 from ..flags import flags
 from ..i18n import N_
+from ..synchronizer import KEY_ABSENT
 
 import logging
 log = logging.getLogger("blivet")
@@ -74,6 +75,11 @@ class MDRaidMember(DeviceFormat):
         d.update({"mdUUID": self.mdUuid, "biosraid": self.biosraid})
         return d
 
+    def _setDestroyEventInfo(self):
+        self.eventSync.info_update(ID_FS_TYPE=KEY_ABSENT,
+                                   ID_FS_UUID=KEY_ABSENT,
+                                   ID_FS_UUID_SUB=KEY_ABSENT)
+
     def _destroy(self, **kwargs):
         blockdev.md.destroy(self.device)
 
diff --git a/blivet/formats/prepboot.py b/blivet/formats/prepboot.py
index 16bd95b..d450e95 100644
--- a/blivet/formats/prepboot.py
+++ b/blivet/formats/prepboot.py
@@ -20,13 +20,16 @@
 # Red Hat Author(s): Dave Lehman <dlehman@redhat.com>
 #
 
+import os
+from parted import PARTITION_PREP
+
 from ..size import Size
 from .. import platform
 from .. import util
 from ..i18n import N_
 from . import DeviceFormat, register_device_format
-from parted import PARTITION_PREP
-import os
+from ..synchronizer import KEY_ABSENT
+
 import logging
 log = logging.getLogger("blivet")
 
@@ -57,18 +60,10 @@ class PPCPRePBoot(DeviceFormat):
         """
         DeviceFormat.__init__(self, **kwargs)
 
-    def _create(self, **kwargs):
-        """ Write the formatting to the specified block device.
-
-            :keyword device: path to device node
-            :type device: str
-            :returns: None.
-
-            .. :note::
+    def _setCreateEventInfo(self):
+        self.eventSync.info_update(ID_FS_TYPE=KEY_ABSENT)
 
-                If a device node path is passed to this method it will overwrite
-                any previously set value of this instance's "device" attribute.
-        """
+    def _create(self, **kwargs):
         super(PPCPRePBoot, self)._create(**kwargs)
         try:
             fd = util.eintr_retry_call(os.open, self.device, os.O_RDWR)
-- 
2.1.0

