def _startUnderlyingMigration(self, startTime):
        if self.hibernating:
            hooks.before_vm_hibernate(self._vm._dom.XMLDesc(0), self._vm.conf)
            try:
                self._vm._vmStats.pause()
                fname = self._vm.cif.prepareVolumePath(self._dst)
                try:
                    self._vm._dom.save(fname)
                finally:
                    self._vm.cif.teardownVolumePath(self._dst)
            except Exception:
                self._vm._vmStats.cont()
                raise
        else:
            for dev in self._vm._customDevices():
                hooks.before_device_migrate_source(
                    dev._deviceXML, self._vm.conf, dev.custom)
            hooks.before_vm_migrate_source(self._vm._dom.XMLDesc(0),
                                           self._vm.conf)

            # Do not measure the time spent for creating the VM on the
            # destination. In some cases some expensive operations can cause
            # the migration to get cancelled right after the transfer started.
            destCreateStartTime = time.time()
            result = self._destServer.migrationCreate(self._machineParams)
            destCreationTime = time.time() - destCreateStartTime
            startTime += destCreationTime
            self.log.info('Creation of destination VM took: %d seconds',
                          destCreationTime)

            if result['status']['code']:
                self.status = result
                raise RuntimeError('migration destination error: ' +
                                   result['status']['message'])
            if config.getboolean('vars', 'ssl'):
                transport = 'tls'
            else:
                transport = 'tcp'
            duri = 'qemu+%s://%s/system' % (transport, self.remoteHost)
            if self._vm.conf['_migrationParams']['dstqemu']:
                muri = 'tcp://%s' % \
                       self._vm.conf['_migrationParams']['dstqemu']
            else:
                muri = 'tcp://%s' % self.remoteHost

            self._vm.log.info('starting migration to %s '
                              'with miguri %s', duri, muri)

            downtimeThread = DowntimeThread(self._vm, int(self._downtime))
            self._monitorThread = MonitorThread(self._vm, startTime)
            with utils.running(downtimeThread):
                with utils.running(self._monitorThread):
                    # we need to support python 2.6, so two nested with-s.
                    self._perform_migration(duri, muri)

            self.log.info("migration took %d seconds to complete",
                          (time.time() - startTime) + destCreationTime)

    def _perform_with_downtime_thread(self, duri, muri):
        self._vm.log.debug('performing migration with downtime thread')
        downtimeThread = DowntimeThread(
            self._vm,
            int(self._downtime),
            config.getint('vars', 'migration_downtime_steps'))

        with utils.running(downtimeThread):
            with utils.running(self._monitorThread):
                # we need to support python 2.6, so two nested with-s.
                self._perform_migration(duri, muri)

 def test_wakeup_blocked_reader(self):
     lock = RWLock()
     writer = LockingThread(lock.exclusive)
     with utils.running(writer):
         if not writer.acquired.wait(2):
             raise RuntimeError("Timeout waiting for writer thread")
         reader = LockingThread(lock.shared)
         with utils.running(reader):
             if not reader.ready.wait(2):
                 raise RuntimeError("Timeout waiting for reader thread")
             self.assertFalse(reader.acquired.wait(1))
             writer.done.set()
             self.assertTrue(reader.acquired.wait(2))

 def test_release_other_thread_read_lock(self):
     lock = RWLock()
     reader = LockingThread(lock.shared)
     with utils.running(reader):
         if not reader.acquired.wait(2):
             raise RuntimeError("Timeout waiting for reader thread")
         self.assertRaises(RuntimeError, lock.release)

 def test_release_other_thread_write_lock(self):
     lock = RWLock()
     writer = LockingThread(lock.exclusive)
     with utils.running(writer):
         if not writer.acquired.wait(2):
             raise RuntimeError("Timeout waiting for writer thread")
         self.assertRaises(RuntimeError, lock.release)

def test_loopback_event(tmpdir):
    listener = udev.MultipathListener()
    received = threading.Event()
    devices = []

    def callback(device):
        pprint.pprint({k: device[k] for k in device})
        devices.append(device)
        received.set()

    listener._callback = callback
    with running(listener):
        # Create a backing file
        filename = str(tmpdir.join("file"))
        with open(filename, "wb") as f:
            f.truncate(1024**2 * 10)

        # Create and remove a loop device
        with loopback.Device(filename) as loop:
            print("Created a loop device at %r" % loop.path)
            if not received.wait(1):
                raise RuntimeError("Timeout receiving event")

            # We expect an event about our loop device
            assert devices[0].get("DEVNAME") == loop.path