This commit is contained in:
Queue A
2026-06-03 06:38:18 +02:00
parent 31402d3121
commit 7061937488
2 changed files with 8 additions and 27 deletions

View File

@@ -1,5 +1,5 @@
#
# Last Change: 2025-10-25
# Last Change: 2026-04-04
#
import functools
import traceback
@@ -18,8 +18,7 @@ class AsyncOriented:
- They will track all their internal tasks if self.create_task is used,
and will perform cleanup on tasks that are done,
once the task counter has passed a certain number since the last cleanup,
or the cleanup function is explicitly called. (The cleanup is done in a sync manner)
once a task is finished, it'll remove itself from the tracked_tasks set.
Notable Fields / Methods:
@@ -39,7 +38,6 @@ class AsyncOriented:
- async self.cleanup()
"""
TRACKED_TASKS_CLEANUP_AFTER = 64
LOG_NAME = __name__
LOG_FMT = r"%(asctime)s [%(process)d] [%(levelname)s] %(message)s", r"[%Y-%m-%d %H:%M:%S %z]"
@@ -69,17 +67,14 @@ class AsyncOriented:
async def startup( self ):
assert not hasattr(self, "loop"), f"This AsyncOriented object's startup has already been called once."
self.loop = asyncio.get_running_loop()
self.tracked_tasks = []
self.tracked_tasks_next_cleanup = self.TRACKED_TASKS_CLEANUP_AFTER
self.tracked_tasks = set()
async def cleanup( self ):
self.garbage_collect_tasks()
msg = "Object's lifetime is over, cleanup is called."
for task in self.tracked_tasks:
task.cancel( msg = msg )
await asyncio.gather( *self.tracked_tasks, return_exceptions = True )
self.garbage_collect_tasks()
def wrap_coro_to_log_exceptions( self, coro, name = None ):
"""
@@ -113,27 +108,13 @@ class AsyncOriented:
if debug: coro = self.wrap_coro_to_log_exceptions( coro, name = name )
task = self.loop.create_task( coro, name = name, context = context )
self.tracked_tasks.append( task )
if self.tracked_tasks_next_cleanup <= len(self.tracked_tasks):
self.garbage_collect_tasks()
self.tracked_tasks.add( task )
task.add_done_callback( self.tracked_tasks.discard )
return task
def garbage_collect_tasks( self ):
task_count = len(self.tracked_tasks) #For the log
for t in list(self.tracked_tasks):
if t.done(): self.tracked_tasks.remove(t)
self.tracked_tasks_next_cleanup = self.TRACKED_TASKS_CLEANUP_AFTER + len(self.tracked_tasks)
self.log.debug(
f"Cleaned up {task_count - len(self.tracked_tasks)} of {task_count} tasks,"
f" next cleanup at: {self.tracked_tasks_next_cleanup}"
)

View File

@@ -31,9 +31,9 @@ class AsyncronWorker( AsyncOriented ):
IS_ACTIVE = False #Whether this proccess needs an AsyncronWorker at all, this prevents running the worker on all the things that cause asyncron.app.ready to run!
INSTANCE = None #Singelton
INIT_CALLBACKS = [] #Once the singelton is created, it'll run through the callbacks with itself as the first arg.
INIT_CALLBACKS = [] #Once the singleton is created, it'll run through the callbacks with itself as the first arg.
THREAD_LOCK = threading.Lock() #for initial singelton creation.
THREAD_LOCK = threading.Lock() #for initial singleton creation.
EXIT_SIGNALS = [
signal.SIGABRT,
@@ -54,7 +54,7 @@ class AsyncronWorker( AsyncOriented ):
def __new__( cls, *args, **kwargs ):
"""
Thread safe singelton logic
Thread safe singleton logic
"""
if cls.INSTANCE: return cls.INSTANCE