run_electrum: have daemon manage Plugins object, and call Plugins.stop

Plugins.stop was never called, so the Plugins thread only stopped
because of the is_running() check in run(), which triggers too late:
the Plugins thread was stopping after the main thread stopped.

E.g. playing around in the qt wizard with wallet creation for a Trezor,
and closing the wizard (only window):
``` 24.85 | E | p/plugin.Plugins |
Traceback (most recent call last):
  File "/home/user/wspace/electrum/electrum/util.py", line 386, in run_jobs
    job.run()
  File "/home/user/wspace/electrum/electrum/plugin.py", line 430, in run
    client.timeout(cutoff)
  File "/home/user/wspace/electrum/electrum/plugin.py", line 363, in wrapper
    return run_in_hwd_thread(partial(func, *args, **kwargs))
  File "/home/user/wspace/electrum/electrum/plugin.py", line 355, in run_in_hwd_thread
    fut = _hwd_comms_executor.submit(func)
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 167, in submit
    raise RuntimeError('cannot schedule new futures after shutdown')
RuntimeError: cannot schedule new futures after shutdown
```

electrum/daemon.py

@@ -411,6 +411,7 @@ class Daemon(Logger):
         if 'wallet_path' in config.cmdline_options:
             self.logger.warning("Ignoring parameter 'wallet_path' for daemon. "
                                 "Use the load_wallet command instead.")
+        self._plugins = None  # type: Optional[Plugins]
         self.asyncio_loop = util.get_asyncio_loop()
         if not self.config.NETWORK_OFFLINE:
             self.network = Network(config, daemon=self)
@@ -560,6 +561,9 @@ class Daemon(Logger):
         return True
 
     def run_daemon(self):
+        # init plugins
+        self._plugins = Plugins(self.config, 'cmdline')
+        # block until we are stopping
         try:
             self._stopping_soon_or_errored.wait()
         except KeyboardInterrupt:
@@ -590,6 +594,11 @@ class Daemon(Logger):
                     if self.network:
                         await group.spawn(self.network.stop(full_shutdown=True))
                     await group.spawn(self.taskgroup.cancel_remaining())
+            if self._plugins:
+                self.logger.info("stopping plugins")
+                self._plugins.stop()
+                async with ignore_after(1):
+                    await self._plugins.stopped_event_async.wait()
         finally:
             if self.listen_jsonrpc:
                 self.logger.info("removing lockfile")
@@ -597,18 +606,21 @@ class Daemon(Logger):
             self.logger.info("stopped")
             self._stopped_event.set()
 
-    def run_gui(self, config: 'SimpleConfig', plugins: 'Plugins'):
+    def run_gui(self) -> None:
+        assert self.config
+        assert self._plugins
         threading.current_thread().name = 'GUI'
-        gui_name = config.GUI_NAME
+        gui_name = self.config.GUI_NAME
         if gui_name in ['lite', 'classic']:
             gui_name = 'qt'
+        self._plugins = Plugins(self.config, gui_name)  # init plugins
         self.logger.info(f'launching GUI: {gui_name}')
         try:
             try:
                 gui = __import__('electrum.gui.' + gui_name, fromlist=['electrum'])
             except GuiImportError as e:
                 sys.exit(str(e))
-            self.gui_object = gui.ElectrumGui(config=config, daemon=self, plugins=plugins)
+            self.gui_object = gui.ElectrumGui(config=self.config, daemon=self, plugins=self._plugins)
             if not self._stop_entered:
                 self.gui_object.main()
             else:

electrum/util.py

@@ -370,7 +370,8 @@ class DaemonThread(threading.Thread, Logger):
         self.running_lock = threading.Lock()
         self.job_lock = threading.Lock()
         self.jobs = []
-        self.stopped_event = threading.Event()  # set when fully stopped
+        self.stopped_event = threading.Event()        # set when fully stopped
+        self.stopped_event_async = asyncio.Event()    # set when fully stopped
 
     def add_jobs(self, jobs):
         with self.job_lock:
@@ -412,6 +413,8 @@ class DaemonThread(threading.Thread, Logger):
             self.logger.info("jnius detach")
         self.logger.info("stopped")
         self.stopped_event.set()
+        loop = get_asyncio_loop()
+        loop.call_soon_threadsafe(self.stopped_event_async.set)
 
 
 def print_stderr(*args):

run_electrum

@@ -451,10 +451,9 @@ def handle_cmd(*, cmdname: str, config: 'SimpleConfig', config_options: dict):
         configure_logging(config)
         fd = daemon.get_file_descriptor(config)
         if fd is not None:
-            plugins = init_plugins(config, config.GUI_NAME)
             d = daemon.Daemon(config, fd, start_network=False)
             try:
-                d.run_gui(config, plugins)
+                d.run_gui()
             except BaseException as e:
                 _logger.exception('daemon.run_gui errored')
                 sys_exit(1)
@@ -469,7 +468,6 @@ def handle_cmd(*, cmdname: str, config: 'SimpleConfig', config_options: dict):
         fd = daemon.get_file_descriptor(config)
         if fd is not None:
             # run daemon
-            init_plugins(config, 'cmdline')
             d = daemon.Daemon(config, fd)
             d.run_daemon()
             sys_exit(0)
@@ -515,7 +513,11 @@ def handle_cmd(*, cmdname: str, config: 'SimpleConfig', config_options: dict):
             coro = run_offline_command(config, config_options, plugins)
             fut = asyncio.run_coroutine_threadsafe(coro, loop)
             try:
-                result = fut.result()
+                try:
+                    result = fut.result()
+                finally:
+                    plugins.stop()
+                    plugins.stopped_event.wait(1)
             except Exception as e:
                 print_stderr(str(e) or repr(e))
                 sys_exit(1)