aliq/asyncron
3
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

before using django tasks

Browse Source
This commit is contained in:
Queue A
2026-08-20 08:15:01 +02:00
parent 4bf684f4b5
commit ba963d6367
2 changed files with 63 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
asyncron/singleton.py
+31
View File
@@ -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
asyncron/workers.py
+32 -49
View File
@@ -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