loop: replace use of asyncio.get_event_loop

The behavior of asyncio.get_event_loop() will change in a future Python
version. It is deprecated starting Python 3.10.

The functions that we can use instead are:

 * asyncio.new_event_loop() - which creates a new event loop
 * asyncio.get_running_loop() - which returns the event loop only if it
   is already running

Signed-off-by: Olivier Gayot <olivier.gayot@canonical.com>

subiquity/common/errorreport.py

@@ -62,7 +62,7 @@ class Upload(metaclass=urwid.MetaSignals):
     def start(self):
         self.pipe_r, self.pipe_w = os.pipe()
         fcntl.fcntl(self.pipe_r, fcntl.F_SETFL, os.O_NONBLOCK)
-        asyncio.get_event_loop().add_reader(self.pipe_r, self._progress)
+        asyncio.get_running_loop().add_reader(self.pipe_r, self._progress)
 
     def _progress(self):
         os.read(self.pipe_r, 4096)
@@ -75,7 +75,7 @@ class Upload(metaclass=urwid.MetaSignals):
         os.write(self.pipe_w, b'x')
 
     def stop(self):
-        asyncio.get_event_loop().remove_reader(self.pipe_r)
+        asyncio.get_running_loop().remove_reader(self.pipe_r)
         os.close(self.pipe_w)
         os.close(self.pipe_r)
 
@@ -174,7 +174,7 @@ class ErrorReport(metaclass=urwid.MetaSignals):
                 _bg_add_info()
                 context.description = "written to " + self.path
         else:
-            self._info_task = asyncio.get_event_loop().create_task(add_info())
+            self._info_task = asyncio.get_running_loop().create_task(add_info())
 
     async def load(self):
         with self._context.child("load"):
@@ -432,7 +432,7 @@ class ErrorReporter(object):
         if report is not None:
             return report
 
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
 
         await self.client.errors.wait.GET(error_ref)
 

subiquity/models/subiquity.py

@@ -268,7 +268,7 @@ class SubiquityModel:
 
     async def wait_confirmation(self):
         if self._confirmation_task is None:
-            self._confirmation_task = asyncio.get_event_loop().create_task(
+            self._confirmation_task = asyncio.get_running_loop().create_task(
                 self._confirmation.wait())
         try:
             await self._confirmation_task

subiquity/server/controllers/filesystem.py

@@ -658,11 +658,11 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         await self._probe_task.start()
 
     def start_listening_udev(self):
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         loop.add_reader(self._monitor.fileno(), self._udev_event)
 
     def stop_listening_udev(self):
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         loop.remove_reader(self._monitor.fileno())
 
     def _udev_event(self):
@@ -670,7 +670,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         if cp.returncode != 0:
             log.debug("waiting 0.1 to let udev event queue settle")
             self.stop_listening_udev()
-            loop = asyncio.get_event_loop()
+            loop = asyncio.get_running_loop()
             loop.call_later(0.1, self.start_listening_udev)
             return
         # Drain the udev events in the queue -- if we stopped listening to

subiquity/server/curtin.py

@@ -104,7 +104,7 @@ class _CurtinCommand:
 
     async def start(self, context, **opts):
         self._fd = journald_listen(
-            asyncio.get_event_loop(), [self._event_syslog_id], self._event)
+            asyncio.get_running_loop(), [self._event_syslog_id], self._event)
         # Yield to the event loop before starting curtin to avoid missing the
         # first couple of events.
         await asyncio.sleep(0)
@@ -120,7 +120,7 @@ class _CurtinCommand:
             waited += 0.1
             log.debug("waited %s seconds for events to drain", waited)
         self._event_contexts.pop('', None)
-        asyncio.get_event_loop().remove_reader(self._fd)
+        asyncio.get_running_loop().remove_reader(self._fd)
         return result
 
     async def run(self, context):

subiquitycore/async_helpers.py

@@ -29,7 +29,7 @@ def _done(fut):
 
 
 def schedule_task(coro, propagate_errors=True):
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
     if asyncio.iscoroutine(coro):
         task = asyncio.Task(coro)
     else:
@@ -41,7 +41,7 @@ def schedule_task(coro, propagate_errors=True):
 
 
 async def run_in_thread(func, *args):
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
     try:
         return await loop.run_in_executor(None, func, *args)
     except concurrent.futures.CancelledError:

subiquitycore/controllers/network.py

@@ -162,7 +162,7 @@ class BaseNetworkController(BaseController):
     def stop_watching(self):
         if not self._watching:
             return
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         for fd in self._observer_fds:
             loop.remove_reader(fd)
         self._watching = False
@@ -170,7 +170,7 @@ class BaseNetworkController(BaseController):
     def start_watching(self):
         if self._watching:
             return
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         for fd in self._observer_fds:
             loop.add_reader(fd, self._data_ready, fd)
         self._watching = True
@@ -180,7 +180,7 @@ class BaseNetworkController(BaseController):
         if cp.returncode != 0:
             log.debug("waiting 0.1 to let udev event queue settle")
             self.stop_watching()
-            loop = asyncio.get_event_loop()
+            loop = asyncio.get_running_loop()
             loop.call_later(0.1, self.start_watching)
             return
         self.observer.data_ready(fd)

subiquitycore/core.py

@@ -70,7 +70,7 @@ class Application:
             os.environ.get('SUBIQUITY_REPLAY_TIMESCALE', "1"))
         self.updated = os.path.exists(self.state_path('updating'))
         self.hub = MessageHub()
-        self.aio_loop = asyncio.get_event_loop()
+        self.aio_loop = asyncio.new_event_loop()
         self.aio_loop.set_exception_handler(self._exception_handler)
         self.load_controllers(self.controllers)
         self.context = Context.new(self)

subiquitycore/pubsub.py

@@ -36,5 +36,5 @@ class MessageHub:
                 await v
 
     def broadcast(self, channel, *args, **kwargs):
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         return loop.create_task(self.abroadcast(channel, *args, **kwargs))