This was a harmless bug, without any impact, but it is wrong to manage the collection of callbacks from it's members. Signed-off-by: Wojtek Porczyk <woju(a)invisiblethingslab.com&gt; --- libvirtaio.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/libvirtaio.py b/libvirtaio.py index d962e64..239561d 100644 --- a/libvirtaio.py +++ b/libvirtaio.py @@ -63,11 +63,6 @@ class Callback(object): self.cb = cb self.opaque = opaque - assert self.iden not in self.impl.callbacks, \ - 'found {} callback: {!r}'.format( - self.iden, self.impl.callbacks[self.iden]) - self.impl.callbacks[self.iden] = self - def __repr__(self): return '<{} iden={}>'.format(self.__class__.__name__, self.iden) @@ -324,6 +319,8 @@ class virEventAsyncIOImpl(object): ''' callback = FDCallback(self, cb, opaque, descriptor=self.descriptors[fd], event=event) + assert callback.iden not in self.callbacks + self.log.debug('add_handle(fd=%d, event=%d, cb=..., opaque=...) = %d', fd, event, callback.iden) self.callbacks[callback.iden] = callback @@ -376,6 +373,8 @@ class virEventAsyncIOImpl(object): https://libvirt.org/html/libvirt-libvirt-event.html#virEventAddTimeoutFunc ''' callback = TimeoutCallback(self, cb, opaque) + assert callback.iden not in self.callbacks + self.log.debug('add_timeout(timeout=%d, cb=..., opaque=...) = %d', timeout, callback.iden) self.callbacks[callback.iden] = callback -- 2.9.4

- Descriptor.close() was a dead code, never used. - TimeoutCallback.close(), as a cleanup function, should have called super() as last statement, not first Signed-off-by: Wojtek Porczyk <woju(a)invisiblethingslab.com&gt; --- libvirtaio.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/libvirtaio.py b/libvirtaio.py index 239561d..a7ede41 100644 --- a/libvirtaio.py +++ b/libvirtaio.py @@ -154,11 +154,6 @@ class Descriptor(object): self.update() return callback - def close(self): - '''''' - self.callbacks.clear() - self.update() - class DescriptorDict(dict): '''Descriptors collection @@ -249,8 +244,8 @@ class TimeoutCallback(Callback): def close(self): '''Stop the timer and call ff callback''' - super(TimeoutCallback, self).close() self.update(timeout=-1) + super(TimeoutCallback, self).close() # # main implementation -- 2.9.4

The intended use is to ensure that the implementation is empty, which is one way to ensure that all connections were properly closed and file descriptors reclaimed. Signed-off-by: Wojtek Porczyk <woju(a)invisiblethingslab.com&gt; --- libvirtaio.py | 36 ++++++++++++++++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) diff --git a/libvirtaio.py b/libvirtaio.py index 97a7f6c..1c432dd 100644 --- a/libvirtaio.py +++ b/libvirtaio.py @@ -269,10 +269,27 @@ class virEventAsyncIOImpl(object): self.descriptors = DescriptorDict(self) self.log = logging.getLogger(self.__class__.__name__) + # NOTE invariant: _finished.is_set() iff _pending == 0 + self._pending = 0 + self._finished = asyncio.Event(loop=loop) + self._finished.set() + def __repr__(self): return '<{} callbacks={} descriptors={}>'.format( type(self).__name__, self.callbacks, self.descriptors) + def _pending_inc(self): + '''Increase the count of pending affairs. Do not use directly.''' + self._pending += 1 + self._finished.clear() + + def _pending_dec(self): + '''Decrease the count of pending affairs. Do not use directly.''' + assert self._pending > 0 + self._pending -= 1 + if self._pending == 0: + self._finished.set() + def register(self): '''Register this instance as event loop implementation''' # pylint: disable=bad-whitespace @@ -293,7 +310,20 @@ class virEventAsyncIOImpl(object): This is a coroutine. ''' self.log.debug('ff_callback(iden=%d, opaque=...)', iden) - return libvirt.virEventInvokeFreeCallback(opaque) + ret = libvirt.virEventInvokeFreeCallback(opaque) + self._pending_dec() + return ret + + @asyncio.coroutine + def drain(self): + '''Wait for the implementation to become idle. + + This is a coroutine. + ''' + self.log.debug('drain()') + if self._pending: + yield from self._finished.wait() + self.log.debug('drain ended') def is_idle(self): '''Returns False if there are leftovers from a connection @@ -301,7 +331,7 @@ class virEventAsyncIOImpl(object): Those may happen if there are sematical problems while closing a connection. For example, not deregistered events before .close(). ''' - return not self.callbacks + return not self.callbacks and not self._pending def _add_handle(self, fd, event, cb, opaque): '''Register a callback for monitoring file handle events @@ -324,6 +354,7 @@ class virEventAsyncIOImpl(object): fd, event, callback.iden) self.callbacks[callback.iden] = callback self.descriptors[fd].add_handle(callback) + self._pending_inc() return callback.iden def _update_handle(self, watch, event): @@ -378,6 +409,7 @@ class virEventAsyncIOImpl(object): timeout, callback.iden) self.callbacks[callback.iden] = callback callback.update(timeout=timeout) + self._pending_inc() return callback.iden def _update_timeout(self, timer, timeout): -- 2.9.4

When the callback causes something that results in changes wrt registered handles, python aborts iteration. Relevant error message: Exception in callback None() handle: <Handle cancelled> Traceback (most recent call last): File "/usr/lib64/python3.5/asyncio/events.py", line 126, in _run self._callback(*self._args) File "/usr/lib64/python3.5/site-packages/libvirtaio.py", line 99, in _handle for callback in self.callbacks.values(): RuntimeError: dictionary changed size during iteration QubesOS/qubes-issues#2805 Signed-off-by: Wojtek Porczyk <woju(a)invisiblethingslab.com&gt; --- libvirtaio.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libvirtaio.py b/libvirtaio.py index 46de9ab..d962e64 100644 --- a/libvirtaio.py +++ b/libvirtaio.py @@ -96,7 +96,7 @@ class Descriptor(object): :param int event: The event (from libvirt's constants) being dispatched ''' - for callback in self.callbacks.values(): + for callback in list(self.callbacks.values()): if callback.event is not None and callback.event & event: callback.cb(callback.iden, self.fd, event, callback.opaque) -- 2.9.4

This logging is helpful for tracing problems with unclosed connections and leaking file descriptors. Signed-off-by: Wojtek Porczyk <woju(a)invisiblethingslab.com&gt; --- libvirtaio.py | 33 +++++++++++++++++++++++++-------- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/libvirtaio.py b/libvirtaio.py index 7c8c396..46de9ab 100644 --- a/libvirtaio.py +++ b/libvirtaio.py @@ -74,7 +74,7 @@ class Callback(object): def close(self): '''Schedule *ff* callback''' self.impl.log.debug('callback %d close(), scheduling ff', self.iden) - self.impl.schedule_ff_callback(self.opaque) + self.impl.schedule_ff_callback(self.iden, self.opaque) # # file descriptors @@ -275,6 +275,10 @@ class virEventAsyncIOImpl(object): self.descriptors = DescriptorDict(self) self.log = logging.getLogger(self.__class__.__name__) + def __repr__(self): + return '<{} callbacks={} descriptors={}>'.format( + type(self).__name__, self.callbacks, self.descriptors) + def register(self): '''Register this instance as event loop implementation''' # pylint: disable=bad-whitespace @@ -284,9 +288,18 @@ class virEventAsyncIOImpl(object): self._add_timeout, self._update_timeout, self._remove_timeout) return self - def schedule_ff_callback(self, opaque): + def schedule_ff_callback(self, iden, opaque): '''Schedule a ff callback from one of the handles or timers''' - self.loop.call_soon(libvirt.virEventInvokeFreeCallback, opaque) + ensure_future(self._ff_callback(iden, opaque), loop=self.loop) + + @asyncio.coroutine + def _ff_callback(self, iden, opaque): + '''Directly free the opaque object + + This is a coroutine. + ''' + self.log.debug('ff_callback(iden=%d, opaque=...)', iden) + return libvirt.virEventInvokeFreeCallback(opaque) def is_idle(self): '''Returns False if there are leftovers from a connection @@ -309,10 +322,10 @@ class virEventAsyncIOImpl(object): .. seealso:: https://libvirt.org/html/libvirt-libvirt-event.html#virEventAddHandleFunc... ''' - self.log.debug('add_handle(fd=%d, event=%d, cb=%r, opaque=%r)', - fd, event, cb, opaque) callback = FDCallback(self, cb, opaque, descriptor=self.descriptors[fd], event=event) + self.log.debug('add_handle(fd=%d, event=%d, cb=..., opaque=...) = %d', + fd, event, callback.iden) self.callbacks[callback.iden] = callback self.descriptors[fd].add_handle(callback) return callback.iden @@ -339,7 +352,11 @@ class virEventAsyncIOImpl(object): https://libvirt.org/html/libvirt-libvirt-event.html#virEventRemoveHandleFunc ''' self.log.debug('remove_handle(watch=%d)', watch) - callback = self.callbacks.pop(watch) + try: + callback = self.callbacks.pop(watch) + except KeyError as err: + self.log.warning('remove_handle(): no such handle: %r', err.args[0]) + raise fd = callback.descriptor.fd assert callback is self.descriptors[fd].remove_handle(watch) if len(self.descriptors[fd].callbacks) == 0: @@ -358,9 +375,9 @@ class virEventAsyncIOImpl(object): .. seealso:: https://libvirt.org/html/libvirt-libvirt-event.html#virEventAddTimeoutFunc ''' - self.log.debug('add_timeout(timeout=%d, cb=%r, opaque=%r)', - timeout, cb, opaque) callback = TimeoutCallback(self, cb, opaque) + self.log.debug('add_timeout(timeout=%d, cb=..., opaque=...) = %d', + timeout, callback.iden) self.callbacks[callback.iden] = callback callback.update(timeout=timeout) return callback.iden -- 2.9.4

On Mon, Sep 25, 2017 at 02:43:56PM +0100, Daniel P. Berrange wrote: 1) There is no technical reason whatsoever. This is an artifact from the previous iteration. 2) In fact ensure_future() does not make it incompatible with python<3.4 because of the try/except import at the beginning of the file (in python<3.4.3 the function is called asyncio.async, imported as ensure_future). -- pozdrawiam / best regards _.-._ Wojtek Porczyk .-^' '^-. Invisible Things Lab |'-.-^-.-'| | | | | I do not fear computers, | '-.-' | I fear lack of them. '-._ : ,-' -- Isaac Asimov `^-^-_>