hbc proper termination, hbd config reloadable

hbd/client/main.py

@@ -28,12 +28,13 @@ from .plugin import PluginRegistry, PluginLoader, InfoPlugin, MonitorPlugin
 # Constants
 PORT = 50003
 INTERVAL = 10
-VER = 6
 MAXRECV = 32767
 
 # Global state
 running = True
 dorestart = False
+shutdown_event: Optional[asyncio.Event] = None
+active_tasks: List[asyncio.Task] = []
 
 
 class AsyncConnection:
@@ -101,7 +102,6 @@ class AsyncConnection:
         # Add standard fields
         msg["name"] = shortname(self.name)
         msg["id"] = self.conn_id
-        msg["ver"] = VER
         msg["time"] = time.time()
         
         # Encode message
@@ -278,9 +278,25 @@ async def heartbeat_sender(conn: AsyncConnection, interval: int):
             
         except Exception as e:
             logger.error(f"Error sending heartbeat: {e}", exc_info=True)
+        except asyncio.CancelledError:
+            logger.debug("Heartbeat sender cancelled")
+            raise
         
-        # Wait for next interval
-        await asyncio.sleep(interval)
+        # Wait for next interval or shutdown event
+        try:
+            if shutdown_event:
+                await asyncio.wait_for(
+                    shutdown_event.wait(), 
+                    timeout=interval
+                )
+                break
+            else:
+                await asyncio.sleep(interval)
+        except asyncio.TimeoutError:
+            pass  # Normal timeout, continue loop
+        except asyncio.CancelledError:
+            logger.debug("Heartbeat sender cancelled during sleep")
+            raise
 
 
 async def plugin_collector(conn: AsyncConnection, registry: PluginRegistry):
@@ -324,7 +340,14 @@ async def plugin_collector(conn: AsyncConnection, registry: PluginRegistry):
     
     # Wait for all tasks
     if tasks:
-        await asyncio.gather(*tasks)
+        try:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        except asyncio.CancelledError:
+            logger.debug("Plugin collector cancelled, cancelling sub-tasks")
+            for task in tasks:
+                if not task.done():
+                    task.cancel()
+            raise
 
 
 async def plugin_collector_interval(
@@ -350,13 +373,30 @@ async def plugin_collector_interval(
                     plugin_msg = {"plugin": plugin.name, **data}
                     await conn.sendto(plugin_msg, "PLG")
                     logger.debug(f"Sent {plugin.name} data")
+            except asyncio.CancelledError:
+                logger.debug("Plugin collector cancelled")
+                raise
             except Exception as e:
                 logger.error(
                     f"Error collecting {plugin.name}: {e}",
                     exc_info=True
                 )
         
-        await asyncio.sleep(interval)
+        # Wait for next interval or shutdown event
+        try:
+            if shutdown_event:
+                await asyncio.wait_for(
+                    shutdown_event.wait(),
+                    timeout=interval
+                )
+                break
+            else:
+                await asyncio.sleep(interval)
+        except asyncio.TimeoutError:
+            pass  # Normal timeout, continue loop
+        except asyncio.CancelledError:
+            logger.debug("Plugin collector cancelled during sleep")
+            raise
 
 
 def shortname(name: str) -> str:
@@ -368,6 +408,15 @@ def stop():
     """Stop the event loop."""
     global running
     running = False
+    
+    # Set shutdown event to wake up sleeping tasks
+    if shutdown_event:
+        shutdown_event.set()
+    
+    # Cancel all active tasks
+    for task in active_tasks:
+        if not task.done():
+            task.cancel()
 
 
 async def cleanup(connections: List[AsyncConnection]):
@@ -393,7 +442,11 @@ async def cleanup(connections: List[AsyncConnection]):
 
 async def async_main(args, config):
     """Async main function."""
-    global running
+    global running, shutdown_event, active_tasks
+    
+    # Create shutdown event
+    shutdown_event = asyncio.Event()
+    active_tasks = []
     
     logger = logging.getLogger("hbc.main")
     
@@ -464,31 +517,30 @@ async def async_main(args, config):
     else:
         logger.warning(f"Plugin directory not found: {plugin_dir}")
     
-    # Start async tasks
-    tasks = []
-    
-    # Heartbeat senders (one per connection)
-    for conn in connections:
-        task = asyncio.create_task(heartbeat_sender(conn, interval))
-        tasks.append(task)
-    
-    # Plugin collector (uses all connections, but we'll use first one)
-    if connections and registry.get_enabled():
-        task = asyncio.create_task(plugin_collector(connections[0], registry))
-        tasks.append(task)
-    
     # Setup signal handlers
     loop = asyncio.get_event_loop()
     for sig in (signal.SIGTERM, signal.SIGINT):
         loop.add_signal_handler(sig, stop)
     
+    # Start async tasks
+    # Heartbeat senders (one per connection)
+    for conn in connections:
+        task = asyncio.create_task(heartbeat_sender(conn, interval))
+        active_tasks.append(task)
+    
+    # Plugin collector (uses all connections, but we'll use first one)
+    if connections and registry.get_enabled():
+        task = asyncio.create_task(plugin_collector(connections[0], registry))
+        active_tasks.append(task)
+    
     # Wait for stop or tasks to complete
     try:
-        await asyncio.gather(*tasks)
+        await asyncio.gather(*active_tasks, return_exceptions=True)
     except asyncio.CancelledError:
-        pass
+        logger.info("Tasks cancelled")
     
     # Cleanup
+    logger.info("Shutting down...")
     await cleanup(connections)
     await loader.unload_all()