From b861999ef46968342a0b4e2a0cf475b89152435c Mon Sep 17 00:00:00 2001 From: Queue A Date: Sat, 29 Aug 2026 19:59:55 +0200 Subject: [PATCH] now with cross proc event pipeline --- asyncron/apps.py | 10 +- asyncron/gunicorn.py | 16 +- .../commands/run_asyncron_worker.py | 1 - asyncron/models.py | 11 +- asyncron/utils.py | 12 ++ asyncron/worker_client.py | 170 ++++++++++++++++++ asyncron/worker_server.py | 35 ++++ asyncron/workers.py | 148 ++++++++------- 8 files changed, 326 insertions(+), 77 deletions(-) create mode 100644 asyncron/worker_client.py create mode 100644 asyncron/worker_server.py diff --git a/asyncron/apps.py b/asyncron/apps.py index 9e3755c..830f146 100644 --- a/asyncron/apps.py +++ b/asyncron/apps.py @@ -26,11 +26,13 @@ class AsyncronConfig(AppConfig): self.load_extensions() #Init and run the asyncron worker singleton for this process if it's not already running. - from .workers import AsyncronWorker - if AsyncronWorker.IS_ACTIVE: + worker_type = os.environ.get('ASYNCRON_WORKER', 'DISABLED') + if worker_type in ("DORMANT", "ENABLED"): + from .workers import AsyncronWorker worker = AsyncronWorker() - worker.start_after_db_ready = True - worker.start( daemon = True ) + if worker_type == "ENABLED": + worker.start_after_db_ready = True + worker.start( daemon = True ) def import_per_app( self, names ): for app in apps.get_app_configs(): diff --git a/asyncron/gunicorn.py b/asyncron/gunicorn.py index 9782443..75f4a29 100644 --- a/asyncron/gunicorn.py +++ b/asyncron/gunicorn.py @@ -11,16 +11,12 @@ def post_fork( server, worker ): #worker and AsyncronWorker, pay attention! post_fork.server = server post_fork.worker = worker - #Temporary so I can work on this. - import time - for attempt in range(3): - try: from .workers import AsyncronWorker - except ImportError: time.sleep(0.1) - else: break - else: raise ImportError() - - AsyncronWorker.IS_ACTIVE = True - AsyncronWorker.register_init_callback( _patch ) + try: from .workers import AsyncronWorker + except ImportError: pass + else: ... + #AsyncronWorker.IS_ACTIVE = True + #AsyncronWorker.IS_DORMANT = True + #AsyncronWorker.register_init_callback( _patch ) def _patch( aworker ): gserver = post_fork.server diff --git a/asyncron/management/commands/run_asyncron_worker.py b/asyncron/management/commands/run_asyncron_worker.py index 42d4194..d6419cf 100644 --- a/asyncron/management/commands/run_asyncron_worker.py +++ b/asyncron/management/commands/run_asyncron_worker.py @@ -29,7 +29,6 @@ class Command(BaseCommand): help = 'Start an Asyncorn Worker' def handle( self, *arg, **kwargs ): - AsyncronWorker.IS_ACTIVE = True while True: worker = AsyncronWorker() diff --git a/asyncron/models.py b/asyncron/models.py index 33fb2b2..d0e3a6e 100644 --- a/asyncron/models.py +++ b/asyncron/models.py @@ -3,8 +3,10 @@ from django.db import models from django.db.models.constraints import UniqueConstraint, Q from django.tasks.base import TaskResult, TaskResultStatus from django.tasks import task_backends +from django.conf import settings from .base.models import BaseModel +from .worker_client import WorkerClientMixin from .utils import TIMEDELTA_PATTERN import functools, traceback, io @@ -13,7 +15,7 @@ import asyncio import os, threading -class Worker( BaseModel ): +class Worker( BaseModel, WorkerClientMixin ): id = models.UUIDField( primary_key = True, default = uuid.uuid4, editable = False ) process_id = models.IntegerField( default = os.getpid ) @@ -34,6 +36,11 @@ class Worker( BaseModel ): UniqueConstraint( fields = ['is_coordinator'], condition = Q( is_coordinator = True ), name = 'unique_coordinator' ), ] + @property + def socket_path( self ): + filename = f"asyncron-worker-{self.id}.sock" + try: return str(settings.ASYNCRON['SOCKET_DIR'] / filename) + except (KeyError, AttributeError): return f"/tmp/{filename}" class TaskSchedule( BaseModel ): name = models.CharField( default = "default", max_length = 200 ) @@ -168,7 +175,7 @@ class Trace( BaseModel ): exception_type = type(status) self.status = TaskResultStatus.FAILED self.lifetime[self.status] = desc - self.traceback = traceback.format_exception( status ) + self.traceback = ''.join( traceback.format_exception( status ) ) self.exception_class_path = f"{exception_type.__module__}.{exception_type.__qualname__}" return diff --git a/asyncron/utils.py b/asyncron/utils.py index 18bc914..d15c1b3 100644 --- a/asyncron/utils.py +++ b/asyncron/utils.py @@ -66,6 +66,18 @@ class JSONSet(Func): #Vibe Coded Class +import importlib +def class_to_path(cls) -> str: + return f"{cls.__module__}.{cls.__qualname__}" +def path_to_class(path: str): + module_path, _, qualname = path.rpartition(".") + module = importlib.import_module(module_path) + obj = module + for part in qualname.split("."): + obj = getattr(obj, part) + return obj + + #Django keeps giving: exception=OperationalError('the connection is closed') from django.db.utils import OperationalError diff --git a/asyncron/worker_client.py b/asyncron/worker_client.py new file mode 100644 index 0000000..4700e39 --- /dev/null +++ b/asyncron/worker_client.py @@ -0,0 +1,170 @@ +#Fully vibe coded file! + +import asyncio +import pickle +import socket +import struct +import threading + + +class WorkerClientMixin: + """ + Mixin providing a cached, per-process Unix socket connection to a worker, + with sync and async send/recv methods sharing the same wire protocol + (4-byte big-endian length prefix + pickled payload). + + Expects the including class to provide a `socket_path` property/attribute. + + Connections are cached per socket_path at the class level, so all + instances pointing at the same path share one connection per process + (and per event loop, for the async side). + """ + + # sync: path -> socket.socket + _sync_conns = {} + _sync_conns_lock = threading.Lock() + # sync: path -> threading.Lock (serializes send+recv on that connection) + _sync_call_locks = {} + + # async: path -> (loop, reader, writer) + _async_conns = {} + _async_conns_lock = threading.Lock() # plain lock: guards dict access, not awaited + # async: path -> asyncio.Lock (serializes send+recv on that connection) + _async_call_locks = {} + + # --- sync connection management --- + + def _get_sync_call_lock(self): + path = self.socket_path + with WorkerClientMixin._sync_conns_lock: + lock = WorkerClientMixin._sync_call_locks.get(path) + if lock is None: + lock = threading.Lock() + WorkerClientMixin._sync_call_locks[path] = lock + return lock + + def _get_sync_socket(self): + path = self.socket_path + with WorkerClientMixin._sync_conns_lock: + sock = WorkerClientMixin._sync_conns.get(path) + if sock is not None: + return sock + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.connect(path) + WorkerClientMixin._sync_conns[path] = sock + return sock + + def _drop_sync_socket(self): + path = self.socket_path + with WorkerClientMixin._sync_conns_lock: + sock = WorkerClientMixin._sync_conns.pop(path, None) + if sock is not None: + try: + sock.close() + except OSError: + pass + + # --- async connection management --- + + def _get_async_call_lock(self): + path = self.socket_path + with WorkerClientMixin._async_conns_lock: + lock = WorkerClientMixin._async_call_locks.get(path) + if lock is None: + lock = asyncio.Lock() + WorkerClientMixin._async_call_locks[path] = lock + return lock + + async def _get_async_connection(self): + path = self.socket_path + loop = asyncio.get_running_loop() + + with WorkerClientMixin._async_conns_lock: + cached = WorkerClientMixin._async_conns.get(path) + + if cached is not None: + cached_loop, reader, writer = cached + if cached_loop is loop and not writer.is_closing(): + return reader, writer + # Stale: dead/foreign loop, or writer already closing. + try: + writer.close() + except Exception: + pass + with WorkerClientMixin._async_conns_lock: + WorkerClientMixin._async_conns.pop(path, None) + + reader, writer = await asyncio.open_unix_connection(path=path) + with WorkerClientMixin._async_conns_lock: + WorkerClientMixin._async_conns[path] = (loop, reader, writer) + return reader, writer + + async def _drop_async_connection(self): + path = self.socket_path + with WorkerClientMixin._async_conns_lock: + cached = WorkerClientMixin._async_conns.pop(path, None) + if cached is not None: + _, _, writer = cached + try: + writer.close() + await writer.wait_closed() + except Exception: + pass + + # --- sync API --- + + def send_msg(self, obj): + data = pickle.dumps(obj) + with self._get_sync_call_lock(): + sock = self._get_sync_socket() + try: + sock.sendall(struct.pack(">I", len(data)) + data) + except OSError: + self._drop_sync_socket() + raise + + def recv_msg(self): + def recv_exactly(sock, n): + buf = b"" + while len(buf) < n: + chunk = sock.recv(n - len(buf)) + if not chunk: + raise ConnectionError("Connection closed before expected data was received") + buf += chunk + return buf + + with self._get_sync_call_lock(): + sock = self._get_sync_socket() + try: + header = recv_exactly(sock, 4) + (length,) = struct.unpack(">I", header) + data = recv_exactly(sock, length) + return pickle.loads(data) + except (OSError, ConnectionError): + self._drop_sync_socket() + raise + + # --- async API --- + + async def asend_msg(self, obj): + data = pickle.dumps(obj) + async with self._get_async_call_lock(): + reader, writer = await self._get_async_connection() + try: + writer.write(struct.pack(">I", len(data)) + data) + await writer.drain() + except (OSError, ConnectionError): + await self._drop_async_connection() + raise + + async def arecv_msg(self): + async with self._get_async_call_lock(): + reader, writer = await self._get_async_connection() + try: + header = await reader.readexactly(4) + (length,) = struct.unpack(">I", header) + data = await reader.readexactly(length) + return pickle.loads(data) + except (asyncio.IncompleteReadError, OSError, ConnectionError): + await self._drop_async_connection() + raise diff --git a/asyncron/worker_server.py b/asyncron/worker_server.py new file mode 100644 index 0000000..4c7593d --- /dev/null +++ b/asyncron/worker_server.py @@ -0,0 +1,35 @@ +#Inter proccess Communication server, mostly vide coded +import asyncio +import pickle, struct + +async def start_serving(queue: asyncio.Queue, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + try: + + while True: + try: + header = await reader.readexactly(4) + except asyncio.IncompleteReadError: + break # connection closed cleanly, header wasn't sent — exit loop + + (length,) = struct.unpack(">I", header) + + try: + data = await reader.readexactly(length) + except asyncio.IncompleteReadError: + break # connection closed mid-message — exit loop + + try: + obj = pickle.loads(data) + except (pickle.UnpicklingError, AttributeError, ModuleNotFoundError, ImportError, EOFError) as e: + print(f"Skipping unpicklable message: {e}") + continue # stream position still valid, safe to read next message + + queue.put_nowait( obj ) + + except ConnectionResetError: pass + finally: + try: + writer.close() + await writer.wait_closed() + except (ConnectionResetError, BrokenPipeError, OSError): + pass diff --git a/asyncron/workers.py b/asyncron/workers.py index b48f4be..15c5277 100644 --- a/asyncron/workers.py +++ b/asyncron/workers.py @@ -17,11 +17,12 @@ import collections, functools import random import humanize -from .utils import is_proc_alive, JSONSet + +from .utils import is_proc_alive, class_to_path, JSONSet from .asynctools import AsyncOriented from .singleton import Singleton from .models import Worker, TaskSchedule, Trace - +from .worker_server import start_serving class AsyncronWorker( Singleton, AsyncOriented ): @@ -35,8 +36,6 @@ class AsyncronWorker( Singleton, 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! - EXIT_SIGNALS = [ signal.SIGABRT, signal.SIGHUP, @@ -44,6 +43,7 @@ class AsyncronWorker( Singleton, AsyncOriented ): signal.SIGINT, signal.SIGTERM ] + IGNORE_MODEL_SIGNALS = [ Worker, TaskSchedule, Trace ] LOG_NAME = "asyncron.worker" LOG_FMT = r"%(asctime)s [%(process)d.Asyncron] [%(levelname)s] %(message)s", r"[%Y-%m-%d %H:%M:%S %z]" @@ -59,26 +59,65 @@ class AsyncronWorker( Singleton, AsyncOriented ): def __init__( self ): self.log #Evaluating the log property while we have the creation lock + self.is_dormant = True #if the start method runs, will turn this into false self.is_db_ready = asyncio.Event() self.is_work_over = asyncio.Event() self.model = Worker() + #Sockets Server + self.server = None #socket server for active workers + self.server_incoming_queue = asyncio.Queue() + #Parallelism self.thread = None #Once the worker starts, it'll be populated - self.compatible_task_paths = [] #We have to conver and use it in database queries a lot, might as well not use a set. - self.clearing_dead_workers = False + self.compatible_task_paths = [] #We have to convert and use it in database queries a lot, might as well not use a set. self.watching_models = collections.defaultdict( set ) # Model -> Set of key name of the tasks - - for callback in self.INIT_CALLBACKS: callback( self ) - self.register_with_exit_signals() + self.receive_django_model_signals() + #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 receive_django_model_signals( self ): + django_name_to_signals = { + name : attr + for name in ["post_save", "post_delete"] #TO Expand: dir(models.signals) + if not name.startswith("_") #Dont get private stuff + and ( attr := getattr(models.signals, name) ) #Just an assignment + and isinstance( attr, models.signals.ModelSignal ) #Is a signal related to models! + } + for name, signal in django_name_to_signals.items(): + signal.connect( functools.partial( self.on_django_model_signal, name ) ) + + #from .models import Task + #for name, task in Task.registered_tasks.items(): + # if not hasattr(task, 'watching_models'): continue + # for model in getattr(task, 'watching_models'): + # self.watching_models[ model ].add( name ) + + def on_django_model_signal( self, signal_name, sender, signal, instance, **kwargs ): + if not kwargs.get('created', False) and any( isinstance( instance, m ) for m in self.IGNORE_MODEL_SIGNALS ): return + + msg = ( self.model.id, signal_name, instance, kwargs ) + + for worker in Worker.objects.all(): + try: worker.send_msg( msg ) + except FileNotFoundError: continue + + + + async def dequeue_server_incoming( self ): + while True: + worker_id, type, *msg = await self.server_incoming_queue.get() + #print( f"Got {type} from worker {worker_id}:", msg ) + + def register_with_exit_signals( self ): + return """ Hooks this worker into EXIT_SIGNALS, without messing up other handlers downstream. resulting in self.handle_exit_signal to be called on exit signals. @@ -121,6 +160,7 @@ class AsyncronWorker( Singleton, AsyncOriented ): def start( self, daemon = False ): + self.is_dormant = False assert not self.thread, "This Worker has already been started once!" if daemon: @@ -154,7 +194,6 @@ class AsyncronWorker( Singleton, AsyncOriented ): await super().startup() self.backend = task_backends['default'] - self.task_reason_jobs_queue = asyncio.Queue() #Run tasks from other threads, safely async def cleanup( self ): try: @@ -169,6 +208,14 @@ class AsyncronWorker( Singleton, AsyncOriented ): else: if count: self.log.info(f"Marked {count} trace(s) as 'Aborted'.") + if self.server: + self.server_task.cancel() + try: await self.server_task + except asyncio.CancelledError: pass + + self.server.close() + await self.server.wait_closed() + await super().cleanup() def start_working( self, is_standalone = False ): @@ -192,11 +239,10 @@ class AsyncronWorker( Singleton, AsyncOriented ): runner.run( self.sync_task_schedules() ) - #self.create_task( self.consume_task_reason_jobs_queue() ) self.create_task( self.maintain_coordinator() ) self.create_task( self.run_tasks_on_schedule() ) + self.create_task( self.start_listening_for_other_workers() ) - self.attach_django_signals() try: runner.run( self.is_work_over.wait() ) #This is the lifetime of this worker @@ -374,49 +420,6 @@ class AsyncronWorker( Singleton, AsyncOriented ): - def attach_django_signals( self ): - django_name_to_signals = { - name : attr - for name in ["post_save", "post_delete"] #TO Expand: dir(models.signals) - if not name.startswith("_") #Dont get private stuff - and ( attr := getattr(models.signals, name) ) #Just an assignment - and isinstance( attr, models.signals.ModelSignal ) #Is a signal related to models! - } - #for name, signal in django_name_to_signals.items(): - # signal.connect( functools.partial( self.model_changed, name ) ) - - return - from .models import Task - for name, task in Task.registered_tasks.items(): - if not hasattr(task, 'watching_models'): continue - for model in getattr(task, 'watching_models'): - self.watching_models[ model ].add( name ) - - - def model_changed( self, signal_name, sender, signal, instance, **kwargs ): - from .models import Task - for name in self.watching_models[instance.__class__]: - task = Task.registered_tasks[name].task - - if task.trace_set.filter( status = "R" ).exists(): - #print("Will not run another trace of the same task to reduce the change of an infinite cycle.") - continue - - if task.worker_type not in ("AR" if self.model.is_standalone else "AD"): #If we can't run this trace - asyncio.run_coroutine_threadsafe( - task.ensure_quick_execution( reason = f"Change ({signal_name}) on {instance}" ), - self.loop - ) - continue - - #print( threading.current_thread(), signal_name, sender, signal, instance, kwargs ) - try: - asyncio.run_coroutine_threadsafe( - self.task_reason_jobs_queue.put( (task, f"Change ({signal_name}) on {instance}") ), - self.loop - ) - except Exception as e: - self.log.warning(f"Model {task} Threadsafe execution Change ({signal_name}) on {instance} failed: {e}") @@ -439,6 +442,11 @@ class AsyncronWorker( Singleton, AsyncOriented ): this_worker_as_queryset = Worker.objects.filter( id = self.model.id ) loop_wait_seconds = 5 + #On startup, explicitly check the health of the current coordinator + current_coordinator = await Worker.objects.filter( is_coordinator = True ).afirst() + if current_coordinator and not is_proc_alive( current_coordinator.process_id ): + await Worker.objects.filter( id = current_coordinator.id ).aupdate( is_coordinator = False ) + while await this_worker_as_queryset.aupdate( last_activity = timezone.now() ): try: @@ -491,11 +499,6 @@ class AsyncronWorker( Singleton, AsyncOriented ): - async def consume_task_reason_jobs_queue( self ): - while True: - task, reason = await self.task_reason_jobs_queue.get() - await self.start_task_now( task, reason ) - self.task_reason_jobs_queue.task_done() async def start_task_now( self, task, reason = "" ): @@ -505,3 +508,28 @@ class AsyncronWorker( Singleton, AsyncOriented ): trace.worker_id = self.model.id self.create_task( self.start_trace_on_time( trace ) ) + + + + + + + + + + + + + + + + + + async def start_listening_for_other_workers( self ): + self.server = await asyncio.start_unix_server( + functools.partial( start_serving, self.server_incoming_queue ), + path = self.model.socket_path + ) + self.log.info(f"Serving on {self.model.socket_path}") + self.server_task = self.create_task( self.server.serve_forever() ) + self.create_task( self.dequeue_server_incoming() )