This commit is contained in:
John Lancaster
2024-10-21 03:15:59 +00:00
parent bc2224a919
commit de496509a9
3 changed files with 26 additions and 14 deletions

View File

@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
class AppDaemonRunContext: class AppDaemonRunContext:
_stack: ExitStack = field(default_factory=ExitStack) _stack: ExitStack = field(default_factory=ExitStack)
stop_event: Event = field(default_factory=Event) stop_event: Event = field(default_factory=Event)
shutdown_grace_period: float = 0.75 shutdown_grace_period: float = 0.1
loop: asyncio.AbstractEventLoop = field(init=False) loop: asyncio.AbstractEventLoop = field(init=False)
executor: ThreadPoolExecutor = field(init=False) executor: ThreadPoolExecutor = field(init=False)
@@ -37,10 +37,10 @@ class AppDaemonRunContext:
logger.debug(f'Closing context from {self.__class__.__name__}') logger.debug(f'Closing context from {self.__class__.__name__}')
self._stack.close() self._stack.close()
def get_running_tasks(self, exclude_current: bool = True) -> list[asyncio.Task]: def get_running_tasks(self) -> list[asyncio.Task]:
return [ return [
t for t in asyncio.all_tasks(self.loop) t for t in asyncio.all_tasks(self.loop)
if exclude_current and t is not asyncio.current_task() if t is not asyncio.current_task()
] ]
async def shutdown(self, signal=signal.SIGTERM): async def shutdown(self, signal=signal.SIGTERM):
@@ -59,15 +59,17 @@ class AppDaemonRunContext:
asyncio.wait_for(t, timeout=self.shutdown_grace_period) asyncio.wait_for(t, timeout=self.shutdown_grace_period)
for t in tasks for t in tasks
) )
logger.debug(f'Allowing graceful shutdown from stop event for { logger.debug(
self.shutdown_grace_period}s') 'Allowing graceful shutdown from stop event for %ss',
self.shutdown_grace_period
)
await asyncio.gather(*graceful, return_exceptions=True) await asyncio.gather(*graceful, return_exceptions=True)
for task in tasks: for task in tasks:
if task.cancelled(): if task.cancelled():
logger.warning(f'Cancelled {task.get_name()}') logger.warning(f'Cancelled {task.get_name()}')
logger.info("Stopping asyncio event loop") logger.info("Calling stop() for asyncio event loop")
self.loop.stop() self.loop.stop()
else: else:
logger.warning('Already started shutdown') logger.warning('Already started shutdown')

View File

@@ -43,6 +43,7 @@ class ADMain:
if hasattr(self, 'ad'): if hasattr(self, 'ad'):
self.ad.logger.info('Running asyncio event loop indefinitely...') self.ad.logger.info('Running asyncio event loop indefinitely...')
self.run_context.loop.run_forever() self.run_context.loop.run_forever()
self.ad.logger.debug('Gracefully stopped event loop')
else: else:
logging.error('Running ADMain without context manager') logging.error('Running ADMain without context manager')
@@ -59,7 +60,7 @@ if __name__ == '__main__':
{ {
'version': 1, 'version': 1,
'disable_existing_loggers': False, 'disable_existing_loggers': False,
'formatters': {'basic': {'style': '{', 'format': '[yellow]{name}[/] {message}'}}, 'formatters': {'basic': {'style': '{', 'format': '[yellow]{name:20}[/] {message}'}},
'handlers': { 'handlers': {
'rich': { 'rich': {
'()': 'rich.logging.RichHandler', '()': 'rich.logging.RichHandler',
@@ -69,9 +70,10 @@ if __name__ == '__main__':
'markup': True 'markup': True
} }
}, },
'root': {'level': 'INFO', 'handlers': ['rich']}, 'root': {'level': 'DEBUG', 'handlers': ['rich']},
} }
) )
with ADMain('/conf/ad-test/conf/appdaemon.yaml') as main: with ADMain('/conf/ad-test/conf/appdaemon.yaml') as main:
main.run() main.run()
main.ad.logger.debug('Exiting main context from with statement')

View File

@@ -34,40 +34,46 @@ class ADSubsystem:
self.logger.debug(f'Starting {self.__class__.__name__}') self.logger.debug(f'Starting {self.__class__.__name__}')
def __exit__(self, exc_type, exc_value, traceback): def __exit__(self, exc_type, exc_value, traceback):
self.logger.debug(f'Exiting {self.__class__.__name__}') pass
# self.logger.debug(f'Exiting {self.__class__.__name__}')
@property @property
def stopping(self) -> bool: def stopping(self) -> bool:
return self.stop.is_set() return self.stop.is_set()
async def sleep(self, delay: float): async def sleep(self, delay: float):
"""Wrapper function for asyncio.sleep that suppresses and logs a task cancellation"""
try: try:
if not self.stopping: if not self.stopping:
await asyncio.sleep(delay) await asyncio.sleep(delay)
else:
self.logger.debug('Skipping sleep due to stop event')
except asyncio.CancelledError: except asyncio.CancelledError:
self.logger.debug('Cancelled during sleep') self.logger.debug('Cancelled during sleep')
@dataclass @dataclass
class Utility(ADSubsystem): class Utility(ADSubsystem):
loop_rate: float = 0.25 loop_rate: float = 1.0
def __enter__(self): def __enter__(self):
super().__enter__() super().__enter__()
self.create_task(self.loop(), 'Utility loop', critical=True) self.create_task(self.loop(), 'utility loop', critical=True)
return self return self
async def loop(self): async def loop(self):
while not self.stopping: while not self.stopping:
self.logger.debug('Looping...') self.logger.debug('Looping...')
await self.sleep(self.loop_rate) await self.sleep(self.loop_rate)
self.logger.debug('Stopped utility loop gracefully')
task_name = asyncio.current_task().get_name()
self.logger.debug(f'Gracefully stopped {task_name} task')
@dataclass @dataclass
class Plugin(ADSubsystem): class Plugin(ADSubsystem):
state: dict[str, int] = field(default_factory=dict) state: dict[str, int] = field(default_factory=dict)
update_rate: float = 2.0 update_rate: float = 5.0
def __post_init__(self) -> None: def __post_init__(self) -> None:
super().__post_init__() super().__post_init__()
@@ -100,7 +106,9 @@ class Plugin(ADSubsystem):
# raise ValueError('fake error') # raise ValueError('fake error')
await self.sleep(self.update_rate) await self.sleep(self.update_rate)
self.logger.debug('Stopped plugin updates gracefully')
task_name = asyncio.current_task().get_name()
self.logger.debug(f'Gracefully stopped {task_name} task')
@dataclass @dataclass