now with cross proc event pipeline

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():

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

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()

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
 

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

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

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

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() )