libvirtio: lazy create the Event object in drain()

The drain method uses an asyncio.Event object to be notified when other
coroutines have removed all registered callbacks. The Event object needs
to be associated with the coroutine that the event loop is running with
and currently this is achieved by passing in the 'loop' parameter.

Unfortunately Python 3.10 has removed the 'loop' parameter and now the
object is associated implicitly with the current thread's event loop.
At the time the virEventAsyncIOImpl constructor is called, however,
there is no guarantee that an event loop has been set for the thread.
The explicitly passed in 'loop' parameter would handle this scenario.

For portability with Python >= 3.10 we need to delay creation of the
Event object until we have a guarantee that there is a loop associated
with the current thread. This is achieved by lazily creating the Event
object inside the 'drain' method, which is expected to be invoked from
coroutine context and thus ensure a loop is associated.

Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>

1 files changed, 11 insertions, 5 deletions

diff --git a/libvirtaio.py b/libvirtaio.py
index 31a8f48..8fa6efe 100644
--- a/libvirtaio.py
+++ b/libvirtaio.py
@@ -271,10 +271,11 @@ 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()
+        # Transient asyncio.Event instance dynamically created
+        # and destroyed by drain()
+        # NOTE invariant: _finished.is_set() iff _pending == 0
+        self._finished = None
 
     def __repr__(self) -> str:
         return '<{} callbacks={} descriptors={}>'.format(
@@ -283,13 +284,14 @@ class virEventAsyncIOImpl(object):
     def _pending_inc(self) -> None:
         '''Increase the count of pending affairs. Do not use directly.'''
         self._pending += 1
-        self._finished.clear()
+        if self._finished is not None:
+            self._finished.clear()
 
     def _pending_dec(self) -> None:
         '''Decrease the count of pending affairs. Do not use directly.'''
         assert self._pending > 0
         self._pending -= 1
-        if self._pending == 0:
+        if self._pending == 0 and self._finished is not None:
             self._finished.set()
 
     def register(self) -> "virEventAsyncIOImpl":
@@ -321,7 +323,11 @@ class virEventAsyncIOImpl(object):
         '''
         self.log.debug('drain()')
         if self._pending:
+            assert self._finished is None
+            self._finished = asyncio.Event()
             await self._finished.wait()
+            self._finished = None
+            assert self._pending == 0
         self.log.debug('drain ended')
 
     def is_idle(self) -> bool: