From ba963d6367c7262f93e2bda94957fb5d4ef5c2bc Mon Sep 17 00:00:00 2001 From: Queue A Date: Thu, 20 Aug 2026 08:15:01 +0200 Subject: [PATCH] before using django tasks --- asyncron/singleton.py | 31 +++++++++++++++++ asyncron/workers.py | 81 +++++++++++++++++-------------------------- 2 files changed, 63 insertions(+), 49 deletions(-) create mode 100644 asyncron/singleton.py diff --git a/asyncron/singleton.py b/asyncron/singleton.py new file mode 100644 index 0000000..994548a --- /dev/null +++ b/asyncron/singleton.py @@ -0,0 +1,31 @@ +# +# Last Change: 2025-10-25 +# + +import threading + +class Singleton: + """ + A thread lock protected singleton, + makes any inherting class into a singlton. + + The SINGLETON_LOCK object is shared between all inherited children, + unless they initiate their own using the 'decouple_lock_from_parent_class' method. + """ + + SINGLETON_INSTANCE = None + SINGLETON_LOCK = threading.Lock() + + @classmethod + def decouple_lock_from_parent_class( cls ): + if cls.SINGLETON_LOCK is not super().SINGLETON_LOCK: return + cls.SINGLETON_LOCK = threading.Lock() + + def __new__( cls, *args, **kwargs ): + if cls.SINGLETON_INSTANCE: return cls.SINGLETON_INSTANCE + + with cls.SINGLETON_LOCK: + if cls.SINGLETON_INSTANCE: return cls.SINGLETON_INSTANCE + cls.SINGLETON_INSTANCE = super().__new__( cls, *args, **kwargs ) + + return cls.SINGLETON_INSTANCE diff --git a/asyncron/workers.py b/asyncron/workers.py index 2269956..54d485b 100644 --- a/asyncron/workers.py +++ b/asyncron/workers.py @@ -17,8 +17,9 @@ import random from .utils import retry_on_db_error, ignore_on_db_error from .asynctools import AsyncOriented +from .singleton import Singleton -class AsyncronWorker( AsyncOriented ): +class AsyncronWorker( Singleton, AsyncOriented ): """ The asyncron worker for a process, limited to one per proccess. Other threads on the same process, can offload their work to this thread and it's single async event loop. @@ -30,10 +31,6 @@ 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 singleton is created, it'll run through the callbacks with itself as the first arg. - - THREAD_LOCK = threading.Lock() #for initial singleton creation. EXIT_SIGNALS = [ signal.SIGABRT, @@ -47,28 +44,35 @@ class AsyncronWorker( AsyncOriented ): LOG_FMT = r"%(asctime)s [%(process)d.Asyncron] [%(levelname)s] %(message)s", r"[%Y-%m-%d %H:%M:%S %z]" LOG_LEVEL = logging.DEBUG + INIT_CALLBACKS = [] #Once the singleton is created, it'll run through the callbacks with itself as the first arg. + @classmethod def register_init_callback( cls, callback ): - assert not cls.INSTANCE, "Cannot register new callbacks after the worker is created!" + assert not cls.SINGLETON_INSTANCE, "Cannot register new callbacks after the worker is created!" cls.INIT_CALLBACKS.append( callback ) - def __new__( cls, *args, **kwargs ): - """ - Thread safe singleton logic - """ + def __init__( self ): + self.log #Evaluating the log property while we have the creation lock - if cls.INSTANCE: return cls.INSTANCE - with cls.THREAD_LOCK: - if cls.INSTANCE: return cls.INSTANCE - cls.INSTANCE = super().__new__( cls, *args, **kwargs ) - for callback in cls.INIT_CALLBACKS: callback( cls.INSTANCE ) - cls.INSTANCE.log #Evaluating the log property while we have the lock - cls.INSTANCE.register_with_exit_signals() + self.is_db_ready_event = asyncio.Event() + self.is_stopping_event = asyncio.Event() - django_signals.connection_created.connect( cls.INSTANCE.handle_new_db_connection ) - cls.INSTANCE.log.debug("Worker created for this process.") - return cls.INSTANCE + #Just so that the asyncron.apps.ready doens't trigger the django warning + self.start_after_db_ready = False + #Parallelism + self.thread = None #Once the worker starts, it'll be populated + + self.clearing_dead_workers = False + self.watching_models = collections.defaultdict( set ) # Model -> Set of key name of the tasks + + self.database_unreachable = False + + for callback in self.INIT_CALLBACKS: callback( self ) + self.register_with_exit_signals() + + django_signals.connection_created.connect( self.handle_new_db_connection ) + self.log.debug("Worker created for this process.") def register_with_exit_signals( self ): @@ -103,14 +107,14 @@ class AsyncronWorker( AsyncOriented ): self.stop(f"Signal {signal.strsignal(signum)}") def handle_new_db_connection( self, sender, **kwargs ): - if self.is_db_ready: return + if self.is_db_ready_event.is_set(): return self.log.debug(f"First DB connection: {sender}") - self.is_db_ready = True + self.is_db_ready_event.set() django_signals.connection_created.disconnect( self.handle_new_db_connection ) - if not self.loop: return #We're not inside an async runner in this(?) or another thread. - self.loop.call_soon_threadsafe( self.is_db_ready_event.set ) + #if not self.loop: return #We're not inside an async runner in this(?) or another thread. + #self.loop.call_soon_threadsafe( self.is_db_ready_event.set ) def start( self, daemon = False ): @@ -127,13 +131,13 @@ class AsyncronWorker( AsyncOriented ): def stop( self, reason = None ): - if self.is_stopping: return #TODO: Insisting on exiting faster should probably be managed in the signal handler + if self.is_stopping_event.is_set(): return #TODO: Insisting on exiting faster should probably be managed in the signal handler self.log.info( f"Stopping Worker: {reason}" ) - self.is_stopping = True + self.is_stopping_event.set() - if not self.loop: return - self.loop.call_soon_threadsafe( self.is_stopping_event.set ) + #if not self.loop: return + #self.loop.call_soon_threadsafe( self.is_stopping_event.set ) ## @@ -143,30 +147,9 @@ class AsyncronWorker( AsyncOriented ): ## - def __init__( self ): - - #These booleans have asyncio.Event counterparts in the loop context - self.is_db_ready = False - self.is_stopping = False - - #Just so that the asyncron.apps.ready doens't trigger the django warning - self.start_after_db_ready = False - - #Parallelism - self.thread = None #Once the worker starts, it'll be populated - - self.clearing_dead_workers = False - self.watching_models = collections.defaultdict( set ) # Model -> Set of key name of the tasks - - self.database_unreachable = False - async def startup( self ): await super().startup() - #These asyncio.Events have booleans counterparts in the __init__ section for main thread non async logic - self.is_db_ready_event = asyncio.Event() - self.is_stopping_event = asyncio.Event() - self.task_reason_jobs_queue = asyncio.Queue() #Run tasks from other threads, safely