# Copyright 2026 Benoit Chesneau # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ Erlang-native asyncio event loop implementation. This module provides an asyncio event loop backed by Erlang's scheduler using enif_select for I/O multiplexing. This replaces Python's polling-based event loop with true event-driven callbacks integrated into the BEAM VM. Usage: from erlang_loop import ErlangEventLoop import asyncio loop = ErlangEventLoop(nif_module) asyncio.set_event_loop(loop) async def main(): await asyncio.sleep(1.0) # Uses erlang:send_after asyncio.run(main()) """ import asyncio import time import threading import sys import socket import ssl import weakref import heapq from asyncio import events, futures, tasks, protocols, transports from asyncio import constants, coroutines, base_events from collections import deque # Event type constants (match C enum values for fast integer comparison) EVENT_TYPE_READ = 1 EVENT_TYPE_WRITE = 2 EVENT_TYPE_TIMER = 3 # Try to import selector_events for transport classes try: from asyncio import selector_events _SelectorSocketTransport = selector_events._SelectorSocketTransport _SelectorDatagramTransport = selector_events._SelectorDatagramTransport except ImportError: _SelectorSocketTransport = None _SelectorDatagramTransport = None class ErlangEventLoop(asyncio.AbstractEventLoop): """asyncio event loop backed by Erlang's scheduler. This event loop implementation delegates I/O multiplexing to Erlang via enif_select, providing: - Sub-millisecond latency (vs 10ms polling) - Zero CPU usage when idle - Full GIL release during waits - Native Erlang scheduler integration The loop works by: 1. add_reader/add_writer register fds with enif_select 2. call_later creates timers via erlang:send_after 3. _run_once waits for events (GIL released in C) 4. Callbacks are dispatched when events occur """ # Use __slots__ for faster attribute access and reduced memory __slots__ = ( '_pel', '_loop_handle', # Native loop handle (capsule) for per-loop isolation '_readers', '_writers', '_readers_by_cid', '_writers_by_cid', '_timers', '_timer_refs', '_timer_heap', '_handle_to_callback_id', '_ready', '_callback_id', '_handle_pool', '_handle_pool_max', '_running', '_stopping', '_closed', '_thread_id', '_clock_resolution', '_exception_handler', '_current_handle', '_debug', '_task_factory', '_default_executor', # Cached method references for hot paths '_ready_append', '_ready_popleft', ) def __init__(self, isolated=False): """Initialize the Erlang event loop. The event loop is backed by Erlang's scheduler via the py_event_loop C module. This module provides direct access to the event loop without going through Erlang callbacks. Args: isolated: If True, create an isolated event loop with its own pending queue. Useful for multi-threaded applications where each thread needs its own event loop. If False (default), use the shared global loop managed by Erlang. """ try: import py_event_loop as pel self._pel = pel if isolated: # Create a new isolated loop handle self._loop_handle = pel._loop_new() else: # Use shared global loop - check it's initialized if not pel._is_initialized(): raise RuntimeError("Erlang event loop not initialized. " "Make sure erlang_python application is started.") self._loop_handle = None except ImportError: # Fallback for testing without actual NIF self._pel = _MockNifModule() self._loop_handle = None # Callback management self._readers = {} # fd -> (callback, args, callback_id, fd_key) self._writers = {} # fd -> (callback, args, callback_id, fd_key) self._readers_by_cid = {} # callback_id -> fd (reverse map for O(1) lookup) self._writers_by_cid = {} # callback_id -> fd (reverse map for O(1) lookup) self._timers = {} # callback_id -> handle self._timer_refs = {} # callback_id -> timer_ref (for cancellation) self._timer_heap = [] # min-heap of (when, callback_id) for O(1) minimum lookup self._handle_to_callback_id = {} # handle -> callback_id (reverse map for O(1) cancellation) self._ready = deque() # Callbacks ready to run self._callback_id = 0 # Cache deque methods for hot path (avoids attribute lookup) self._ready_append = self._ready.append self._ready_popleft = self._ready.popleft # Handle object pool for reduced allocations # Trade-off: smaller pool = less GC (better high_concurrency) # larger pool = more reuse (better large_response) # 150 balances both workloads self._handle_pool = [] self._handle_pool_max = 150 # State self._running = False self._stopping = False self._closed = False self._thread_id = None self._clock_resolution = 1e-9 # nanoseconds # Exception handling self._exception_handler = None self._current_handle = None # Debug mode self._debug = False # Task factory self._task_factory = None # SSL context self._default_executor = None def _next_id(self): """Generate a unique callback ID.""" self._callback_id += 1 return self._callback_id # ======================================================================== # Running and stopping the event loop # ======================================================================== def run_forever(self): """Run the event loop until stop() is called.""" self._check_closed() self._check_running() self._set_coroutine_origin_tracking(self._debug) self._thread_id = threading.get_ident() self._running = True self._stopping = False # Register as the running loop so asyncio.get_running_loop() works old_running_loop = events._get_running_loop() events._set_running_loop(self) try: while not self._stopping: self._run_once() finally: events._set_running_loop(old_running_loop) self._stopping = False self._running = False self._thread_id = None self._set_coroutine_origin_tracking(False) def run_until_complete(self, future): """Run the event loop until a future is done.""" self._check_closed() self._check_running() new_task = not futures.isfuture(future) future = tasks.ensure_future(future, loop=self) if new_task: future._log_destroy_pending = False # Use a single callback reference to ensure proper removal # (two different lambdas would be different objects) def _done_callback(f): self.stop() future.add_done_callback(_done_callback) try: self.run_forever() except Exception: if new_task and future.done() and not future.cancelled(): future.exception() raise finally: future.remove_done_callback(_done_callback) if not future.done(): raise RuntimeError('Event loop stopped before Future completed.') return future.result() def stop(self): """Stop the event loop.""" self._stopping = True # Wake up the event loop if it's waiting try: self._pel._wakeup() except Exception: pass def is_running(self): """Return True if the event loop is running.""" return self._running def is_closed(self): """Return True if the event loop is closed.""" return self._closed def close(self): """Close the event loop.""" if self._running: raise RuntimeError("Cannot close a running event loop") if self._closed: return self._closed = True # Cancel all timers for callback_id, handle in list(self._timers.items()): handle.cancel() timer_ref = self._timer_refs.get(callback_id) if timer_ref is not None: try: if self._loop_handle is not None: self._pel._cancel_timer_for(self._loop_handle, timer_ref) else: self._pel._cancel_timer(timer_ref) except (AttributeError, RuntimeError): pass self._timers.clear() self._timer_refs.clear() self._timer_heap.clear() self._handle_to_callback_id.clear() # Remove all readers/writers for fd in list(self._readers.keys()): self.remove_reader(fd) for fd in list(self._writers.keys()): self.remove_writer(fd) # Destroy the native loop handle if we have one if self._loop_handle is not None: try: self._pel._loop_destroy(self._loop_handle) except (AttributeError, RuntimeError): pass self._loop_handle = None # Shutdown default executor if self._default_executor is not None: self._default_executor.shutdown(wait=False) self._default_executor = None async def shutdown_asyncgens(self): """Shutdown all active asynchronous generators.""" # Not implemented - would need tracking of async generators pass async def shutdown_default_executor(self, timeout=None): """Shutdown the default executor.""" if self._default_executor is not None: self._default_executor.shutdown(wait=True) self._default_executor = None # ======================================================================== # Scheduling callbacks # ======================================================================== def call_soon(self, callback, *args, context=None): """Schedule a callback to be called soon.""" self._check_closed() handle = events.Handle(callback, args, self, context) self._ready_append(handle) # Use cached method return handle def call_soon_threadsafe(self, callback, *args, context=None): """Thread-safe version of call_soon.""" handle = self.call_soon(callback, *args, context=context) # Wake up the event loop try: if self._loop_handle is not None: self._pel._wakeup_for(self._loop_handle) else: self._pel._wakeup() except Exception: pass return handle def call_later(self, delay, callback, *args, context=None): """Schedule a callback to be called after delay seconds.""" self._check_closed() timer = self.call_at(self.time() + delay, callback, *args, context=context) return timer def call_at(self, when, callback, *args, context=None): """Schedule a callback to be called at a specific time.""" self._check_closed() callback_id = self._next_id() handle = events.TimerHandle(when, callback, args, self, context) self._timers[callback_id] = handle self._handle_to_callback_id[id(handle)] = callback_id # Reverse map for O(1) cancellation # Push to timer heap for O(1) minimum lookup heapq.heappush(self._timer_heap, (when, callback_id)) # Schedule with Erlang's native timer system delay_ms = max(0, int((when - self.time()) * 1000)) try: if self._loop_handle is not None: timer_ref = self._pel._schedule_timer_for(self._loop_handle, delay_ms, callback_id) else: timer_ref = self._pel._schedule_timer(delay_ms, callback_id) self._timer_refs[callback_id] = timer_ref except AttributeError: pass # Fallback: mock module doesn't have _schedule_timer except RuntimeError as e: # Fail fast on initialization errors - don't silently hang raise RuntimeError(f"Timer scheduling failed: {e}") from e return handle def time(self): """Return the current time according to the event loop's clock.""" return time.monotonic() # ======================================================================== # Creating Futures and Tasks # ======================================================================== def create_future(self): """Create a Future object attached to this loop.""" return futures.Future(loop=self) def create_task(self, coro, *, name=None, context=None): """Schedule a coroutine to be executed.""" self._check_closed() if self._task_factory is None: # Python 3.9 doesn't support context parameter if sys.version_info >= (3, 11): task = tasks.Task(coro, loop=self, name=name, context=context) elif sys.version_info >= (3, 8): task = tasks.Task(coro, loop=self, name=name) else: task = tasks.Task(coro, loop=self) if name is not None: task.set_name(name) else: if sys.version_info >= (3, 11) and context is not None: task = self._task_factory(self, coro, context=context) else: task = self._task_factory(self, coro) if name is not None: task.set_name(name) return task def set_task_factory(self, factory): """Set a task factory.""" self._task_factory = factory def get_task_factory(self): """Return the task factory.""" return self._task_factory # ======================================================================== # File descriptor callbacks # ======================================================================== def add_reader(self, fd, callback, *args): """Register a reader callback for a file descriptor.""" self._check_closed() self.remove_reader(fd) callback_id = self._next_id() try: if self._loop_handle is not None: fd_key = self._pel._add_reader_for(self._loop_handle, fd, callback_id) else: fd_key = self._pel._add_reader(fd, callback_id) self._readers[fd] = (callback, args, callback_id, fd_key) self._readers_by_cid[callback_id] = fd # Reverse map for O(1) dispatch except Exception as e: raise RuntimeError(f"Failed to add reader: {e}") def remove_reader(self, fd): """Unregister a reader callback for a file descriptor.""" if fd in self._readers: entry = self._readers[fd] callback_id = entry[2] fd_key = entry[3] if len(entry) > 3 else None del self._readers[fd] self._readers_by_cid.pop(callback_id, None) # Clean up reverse map try: if fd_key is not None: if self._loop_handle is not None: self._pel._remove_reader_for(self._loop_handle, fd_key) else: self._pel._remove_reader(fd_key) except Exception: pass return True return False def add_writer(self, fd, callback, *args): """Register a writer callback for a file descriptor.""" self._check_closed() self.remove_writer(fd) callback_id = self._next_id() try: if self._loop_handle is not None: fd_key = self._pel._add_writer_for(self._loop_handle, fd, callback_id) else: fd_key = self._pel._add_writer(fd, callback_id) self._writers[fd] = (callback, args, callback_id, fd_key) self._writers_by_cid[callback_id] = fd # Reverse map for O(1) dispatch except Exception as e: raise RuntimeError(f"Failed to add writer: {e}") def remove_writer(self, fd): """Unregister a writer callback for a file descriptor.""" if fd in self._writers: entry = self._writers[fd] callback_id = entry[2] fd_key = entry[3] if len(entry) > 3 else None del self._writers[fd] self._writers_by_cid.pop(callback_id, None) # Clean up reverse map try: if fd_key is not None: if self._loop_handle is not None: self._pel._remove_writer_for(self._loop_handle, fd_key) else: self._pel._remove_writer(fd_key) except Exception: pass return True return False # ======================================================================== # Socket operations # ======================================================================== async def sock_recv(self, sock, nbytes): """Receive data from a socket.""" fut = self.create_future() def _recv(): try: data = sock.recv(nbytes) self.call_soon(fut.set_result, data) except (BlockingIOError, InterruptedError): return # Not ready, keep waiting except Exception as e: self.call_soon(fut.set_exception, e) self.remove_reader(sock.fileno()) self.add_reader(sock.fileno(), _recv) return await fut async def sock_recv_into(self, sock, buf): """Receive data from a socket into a buffer.""" fut = self.create_future() def _recv_into(): try: nbytes = sock.recv_into(buf) self.call_soon(fut.set_result, nbytes) except (BlockingIOError, InterruptedError): return except Exception as e: self.call_soon(fut.set_exception, e) self.remove_reader(sock.fileno()) self.add_reader(sock.fileno(), _recv_into) return await fut async def sock_sendall(self, sock, data): """Send data to a socket.""" fut = self.create_future() data = memoryview(data) offset = [0] def _send(): try: n = sock.send(data[offset[0]:]) offset[0] += n if offset[0] >= len(data): self.remove_writer(sock.fileno()) self.call_soon(fut.set_result, None) except (BlockingIOError, InterruptedError): return except Exception as e: self.remove_writer(sock.fileno()) self.call_soon(fut.set_exception, e) self.add_writer(sock.fileno(), _send) return await fut async def sock_connect(self, sock, address): """Connect a socket to a remote address.""" fut = self.create_future() try: sock.connect(address) fut.set_result(None) return await fut except (BlockingIOError, InterruptedError): pass def _connect(): try: err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) if err != 0: raise OSError(err, f'Connect call failed {address}') self.call_soon(fut.set_result, None) except Exception as e: self.call_soon(fut.set_exception, e) self.remove_writer(sock.fileno()) self.add_writer(sock.fileno(), _connect) return await fut async def sock_accept(self, sock): """Accept a connection on a socket.""" fut = self.create_future() def _accept(): try: conn, address = sock.accept() conn.setblocking(False) self.call_soon(fut.set_result, (conn, address)) except (BlockingIOError, InterruptedError): return except Exception as e: self.call_soon(fut.set_exception, e) self.remove_reader(sock.fileno()) self.add_reader(sock.fileno(), _accept) return await fut async def sock_sendfile(self, sock, file, offset=0, count=None, *, fallback=True): """Send a file through a socket.""" raise NotImplementedError("sock_sendfile not implemented") # ======================================================================== # High-level connection methods # ======================================================================== async def create_connection( self, protocol_factory, host=None, port=None, *, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, happy_eyeballs_delay=None, interleave=None): """Create a streaming transport connection.""" if sock is not None: # Use provided socket sock.setblocking(False) else: # Resolve address and connect infos = await self.getaddrinfo( host, port, family=family, type=socket.SOCK_STREAM, proto=proto, flags=flags) if not infos: raise OSError(f'getaddrinfo({host!r}) returned empty list') exceptions = [] for family, type_, proto, cname, address in infos: sock = socket.socket(family, type_, proto) sock.setblocking(False) try: await self.sock_connect(sock, address) break except OSError as exc: exceptions.append(exc) sock.close() sock = None if sock is None: if len(exceptions) == 1: raise exceptions[0] raise OSError(f'Multiple exceptions: {exceptions}') # Create transport and protocol protocol = protocol_factory() transport = _ErlangSocketTransport(self, sock, protocol) try: await transport._start() except Exception: transport.close() raise return transport, protocol async def create_server( self, protocol_factory, host=None, port=None, *, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True): """Create a TCP server.""" if sock is not None: sockets = [sock] else: if host == '': hosts = [None] elif isinstance(host, str): hosts = [host] else: hosts = host if host else [None] sockets = [] infos = [] for h in hosts: info = await self.getaddrinfo( h, port, family=family, type=socket.SOCK_STREAM, flags=flags) infos.extend(info) completed = set() for family, type_, proto, cname, address in infos: key = (family, address) if key in completed: continue completed.add(key) sock = socket.socket(family, type_, proto) sock.setblocking(False) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if reuse_port: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) try: sock.bind(address) except OSError: sock.close() raise sock.listen(backlog) sockets.append(sock) server = _ErlangServer(self, sockets, protocol_factory, ssl, backlog) if start_serving: server._start_serving() return server async def create_datagram_endpoint(self, protocol_factory, local_addr=None, remote_addr=None, *, family=0, proto=0, flags=0, reuse_address=None, reuse_port=None, allow_broadcast=None, sock=None): """Create datagram (UDP) connection. Args: protocol_factory: Factory function returning a DatagramProtocol local_addr: Local (host, port) tuple to bind to remote_addr: Remote (host, port) tuple to connect to (optional) family: Socket family (AF_INET or AF_INET6) proto: Socket protocol number flags: getaddrinfo flags reuse_address: Allow reuse of local address (SO_REUSEADDR) reuse_port: Allow reuse of local port (SO_REUSEPORT) allow_broadcast: Allow sending to broadcast addresses (SO_BROADCAST) sock: Pre-existing socket to use (overrides other options) Returns: (transport, protocol) tuple """ if sock is not None: # Use provided socket sock.setblocking(False) else: # Determine address family if family == 0: if local_addr: family = socket.AF_INET elif remote_addr: family = socket.AF_INET else: family = socket.AF_INET # Create UDP socket sock = socket.socket(family, socket.SOCK_DGRAM, proto) sock.setblocking(False) # Apply socket options if reuse_address: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) if reuse_port: try: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except (AttributeError, OSError): # SO_REUSEPORT not available on all platforms pass if allow_broadcast: sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) # Bind to local address if local_addr: sock.bind(local_addr) # Connect to remote address (makes it a connected UDP socket) if remote_addr: sock.connect(remote_addr) # Create transport and protocol protocol = protocol_factory() transport = _ErlangDatagramTransport( self, sock, protocol, address=remote_addr ) # Start the transport transport._start() return transport, protocol # ======================================================================== # Signal handling # ======================================================================== def add_signal_handler(self, sig, callback, *args): """Add a signal handler.""" raise NotImplementedError("Signal handlers not supported in ErlangEventLoop") def remove_signal_handler(self, sig): """Remove a signal handler.""" raise NotImplementedError("Signal handlers not supported in ErlangEventLoop") # ======================================================================== # Error handling # ======================================================================== def set_exception_handler(self, handler): """Set the exception handler.""" self._exception_handler = handler def get_exception_handler(self): """Get the exception handler.""" return self._exception_handler def default_exception_handler(self, context): """Default exception handler.""" message = context.get('message', 'Unhandled exception') exception = context.get('exception') if exception is not None: import traceback exc_info = (type(exception), exception, exception.__traceback__) tb = ''.join(traceback.format_exception(*exc_info)) print(f'{message}\n{tb}', file=sys.stderr) else: print(f'{message}', file=sys.stderr) def call_exception_handler(self, context): """Call the exception handler.""" if self._exception_handler is not None: try: self._exception_handler(self, context) except Exception: self.default_exception_handler(context) else: self.default_exception_handler(context) # ======================================================================== # Debug mode # ======================================================================== def get_debug(self): """Return the debug mode setting.""" return self._debug def set_debug(self, enabled): """Set the debug mode.""" self._debug = enabled # ======================================================================== # Internal methods # ======================================================================== def _run_once(self): """Run one iteration of the event loop.""" # Local aliases for hot-path attributes (avoids repeated lookups) ready = self._ready popleft = self._ready_popleft return_handle = self._return_handle # Run all ready callbacks (timer cleanup happens lazily during dispatch) ntodo = len(ready) for _ in range(ntodo): if not ready: break handle = popleft() if handle._cancelled: return_handle(handle) continue self._current_handle = handle try: handle._run() except Exception as e: self.call_exception_handler({ 'message': 'Exception in callback', 'exception': e, 'handle': handle, }) finally: self._current_handle = None return_handle(handle) # Calculate timeout based on next timer using heap with lazy deletion if ready or self._stopping: timeout = 0 elif self._timer_heap: # Lazy cleanup - pop stale/cancelled entries from heap timer_heap = self._timer_heap timers = self._timers while timer_heap: when, cid = timer_heap[0] handle = timers.get(cid) if handle is None or handle._cancelled: heapq.heappop(timer_heap) continue break # Found valid minimum if timer_heap: when, _ = timer_heap[0] timeout = max(0, int((when - self.time()) * 1000)) timeout = max(1, min(timeout, 1000)) # Cap at 1s, min 1ms else: # All timers cancelled - bulk cleanup on rare path timers.clear() self._timer_refs.clear() timeout = 1000 else: timeout = 1000 # 1s max wait # Use combined poll + get_pending (single NIF call, integer event types) try: # Use handle-based API when loop_handle is set, legacy otherwise if self._loop_handle is not None: pending = self._pel._run_once_native_for(self._loop_handle, timeout) else: pending = self._pel._run_once_native(timeout) dispatch = self._dispatch for callback_id, event_type in pending: dispatch(callback_id, event_type) except AttributeError: # Fallback for old NIF without _run_once_native try: num_events = self._pel._poll_events(timeout) if num_events > 0: pending = self._pel._get_pending() dispatch = self._dispatch for callback_id, event_type in pending: dispatch(callback_id, event_type) except AttributeError: pass # Mock module without these methods except RuntimeError as e: # Fail fast on initialization errors - don't silently hang raise RuntimeError(f"Event loop poll failed: {e}") from e except RuntimeError as e: # Fail fast on initialization errors - don't silently hang raise RuntimeError(f"Event loop poll failed: {e}") from e def _dispatch(self, callback_id, event_type): """Dispatch a callback based on event type. Uses O(1) reverse map lookup for fd events instead of O(n) iteration. Event types are integers for fast comparison (no string allocation). Inlined lookup: dict.get(None) returns None, so single expression is safe. """ # Integer comparison is faster than string - NIF returns integers if event_type == 1: # EVENT_TYPE_READ # Inlined lookup: _readers.get(None) returns None (safe) entry = self._readers.get(self._readers_by_cid.get(callback_id)) if entry is not None: self._ready_append(self._get_handle(entry[0], entry[1])) elif event_type == 2: # EVENT_TYPE_WRITE entry = self._writers.get(self._writers_by_cid.get(callback_id)) if entry is not None: self._ready_append(self._get_handle(entry[0], entry[1])) elif event_type == 3: # EVENT_TYPE_TIMER handle = self._timers.pop(callback_id, None) if handle is not None: self._timer_refs.pop(callback_id, None) self._handle_to_callback_id.pop(id(handle), None) # Clean up reverse map if not handle._cancelled: self._ready_append(handle) def _check_closed(self): """Raise an error if the loop is closed.""" if self._closed: raise RuntimeError('Event loop is closed') def _check_running(self): """Raise an error if the loop is already running.""" if self._running: raise RuntimeError('This event loop is already running') def _timer_handle_cancelled(self, handle): """Called when a TimerHandle is cancelled. Uses O(1) reverse map lookup instead of O(n) iteration. """ # O(1) lookup via reverse map callback_id = self._handle_to_callback_id.pop(id(handle), None) if callback_id is not None: self._timers.pop(callback_id, None) timer_ref = self._timer_refs.pop(callback_id, None) if timer_ref is not None: try: self._pel._cancel_timer(timer_ref) except (AttributeError, RuntimeError): pass def _set_coroutine_origin_tracking(self, enabled): """Enable/disable coroutine origin tracking.""" if enabled: sys.set_coroutine_origin_tracking_depth(1) else: sys.set_coroutine_origin_tracking_depth(0) # ======================================================================== # Handle pool for reduced allocations # ======================================================================== def _get_handle(self, callback, args): """Get a Handle from the pool or create a new one.""" if self._handle_pool: handle = self._handle_pool.pop() handle._callback = callback handle._args = args handle._cancelled = False return handle return events.Handle(callback, args, self, None) def _return_handle(self, handle): """Return a Handle to the pool for reuse.""" if len(self._handle_pool) < self._handle_pool_max: # Clear references to avoid keeping objects alive handle._callback = None handle._args = None self._handle_pool.append(handle) # ======================================================================== # Executor methods # ======================================================================== def run_in_executor(self, executor, func, *args): """Run a function in an executor.""" self._check_closed() if executor is None: executor = self._get_default_executor() return asyncio.wrap_future( executor.submit(func, *args), loop=self ) def _get_default_executor(self): """Get or create the default executor.""" if self._default_executor is None: from concurrent.futures import ThreadPoolExecutor self._default_executor = ThreadPoolExecutor() return self._default_executor def set_default_executor(self, executor): """Set the default executor.""" self._default_executor = executor # ======================================================================== # DNS resolution # ======================================================================== async def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0): """Resolve host/port to address info.""" return await self.run_in_executor( None, socket.getaddrinfo, host, port, family, type, proto, flags ) async def getnameinfo(self, sockaddr, flags=0): """Resolve socket address to host/port.""" return await self.run_in_executor(None, socket.getnameinfo, sockaddr, flags) class _ErlangSocketTransport(transports.Transport): """Socket transport for ErlangEventLoop.""" __slots__ = ( '_loop', '_sock', '_protocol', '_buffer', '_closing', '_conn_lost', '_write_ready', '_paused', '_extra', '_fileno', ) _buffer_factory = bytearray max_size = 256 * 1024 # 256 KB def __init__(self, loop, sock, protocol, extra=None): super().__init__(extra) self._loop = loop self._sock = sock self._protocol = protocol self._buffer = self._buffer_factory() self._closing = False self._conn_lost = 0 self._write_ready = True self._paused = False self._fileno = sock.fileno() # Cache fileno to avoid repeated calls self._extra = extra or {} self._extra['socket'] = sock try: self._extra['sockname'] = sock.getsockname() except OSError: pass try: self._extra['peername'] = sock.getpeername() except OSError: pass async def _start(self): """Start the transport.""" self._loop.call_soon(self._protocol.connection_made, self) self._loop.add_reader(self._fileno, self._read_ready) def _read_ready(self): """Called when data is available to read.""" if self._conn_lost: return try: data = self._sock.recv(self.max_size) except (BlockingIOError, InterruptedError): return except Exception as exc: self._fatal_error(exc, 'Fatal read error') return if data: self._protocol.data_received(data) else: # Connection closed self._loop.remove_reader(self._fileno) self._protocol.eof_received() def write(self, data): """Write data to the transport.""" if self._conn_lost or self._closing: return if not data: return if not self._buffer: try: n = self._sock.send(data) except (BlockingIOError, InterruptedError): n = 0 except Exception as exc: self._fatal_error(exc, 'Fatal write error') return if n == len(data): return elif n > 0: data = data[n:] self._loop.add_writer(self._fileno, self._write_ready_cb) self._buffer.extend(data) def _write_ready_cb(self): """Called when socket is ready for writing.""" if not self._buffer: self._loop.remove_writer(self._fileno) if self._closing: self._call_connection_lost(None) return try: n = self._sock.send(self._buffer) except (BlockingIOError, InterruptedError): return except Exception as exc: self._loop.remove_writer(self._fileno) self._fatal_error(exc, 'Fatal write error') return if n: del self._buffer[:n] if not self._buffer: self._loop.remove_writer(self._fileno) if self._closing: self._call_connection_lost(None) def write_eof(self): """Close the write end.""" if self._closing: return self._closing = True if not self._buffer: self._loop.remove_reader(self._fileno) self._call_connection_lost(None) def can_write_eof(self): return True def close(self): """Close the transport.""" if self._closing: return self._closing = True self._loop.remove_reader(self._fileno) if not self._buffer: self._conn_lost += 1 self._call_connection_lost(None) def _call_connection_lost(self, exc): """Call protocol.connection_lost().""" try: self._protocol.connection_lost(exc) finally: self._sock.close() def _fatal_error(self, exc, message='Fatal error'): """Handle fatal errors.""" self._loop.call_exception_handler({ 'message': message, 'exception': exc, 'transport': self, 'protocol': self._protocol, }) self.close() def get_extra_info(self, name, default=None): return self._extra.get(name, default) def is_closing(self): return self._closing def get_write_buffer_size(self): return len(self._buffer) def abort(self): """Close immediately.""" self._closing = True self._conn_lost += 1 self._loop.remove_reader(self._fileno) self._loop.remove_writer(self._fileno) self._call_connection_lost(None) class _ErlangDatagramTransport(transports.DatagramTransport): """Datagram (UDP) transport for ErlangEventLoop.""" __slots__ = ( '_loop', '_sock', '_protocol', '_address', '_buffer', '_closing', '_conn_lost', '_extra', '_fileno', ) max_size = 256 * 1024 # 256 KB def __init__(self, loop, sock, protocol, address=None, extra=None): super().__init__(extra) self._loop = loop self._sock = sock self._protocol = protocol self._address = address # Default remote address (for connected UDP) self._buffer = deque() # Deque of (data, addr) tuples for O(1) popleft self._closing = False self._conn_lost = 0 self._fileno = sock.fileno() # Cache fileno to avoid repeated calls self._extra = extra or {} self._extra['socket'] = sock try: self._extra['sockname'] = sock.getsockname() except OSError: pass if address: self._extra['peername'] = address def _start(self): """Start the transport.""" self._loop.call_soon(self._protocol.connection_made, self) self._loop.add_reader(self._fileno, self._read_ready) def _read_ready(self): """Called when data is available to read.""" if self._conn_lost: return try: data, addr = self._sock.recvfrom(self.max_size) except (BlockingIOError, InterruptedError): return except OSError as exc: self._protocol.error_received(exc) return except Exception as exc: self._fatal_error(exc, 'Fatal read error on datagram transport') return self._protocol.datagram_received(data, addr) def sendto(self, data, addr=None): """Send data to the transport.""" if self._conn_lost or self._closing: return if not data: return if addr is None: addr = self._address if not self._buffer: # Try to send immediately try: if addr: self._sock.sendto(data, addr) else: self._sock.send(data) return except (BlockingIOError, InterruptedError): # Buffer and wait for write ready self._loop.add_writer(self._fileno, self._write_ready) except OSError as exc: self._protocol.error_received(exc) return except Exception as exc: self._fatal_error(exc, 'Fatal write error on datagram transport') return # Buffer the data self._buffer.append((data, addr)) def _write_ready(self): """Called when socket is ready for writing.""" while self._buffer: data, addr = self._buffer[0] try: if addr: self._sock.sendto(data, addr) else: self._sock.send(data) except (BlockingIOError, InterruptedError): return except OSError as exc: self._buffer.popleft() self._protocol.error_received(exc) return except Exception as exc: self._fatal_error(exc, 'Fatal write error on datagram transport') return self._buffer.popleft() self._loop.remove_writer(self._fileno) if self._closing: self._call_connection_lost(None) def close(self): """Close the transport.""" if self._closing: return self._closing = True self._loop.remove_reader(self._fileno) if not self._buffer: self._conn_lost += 1 self._call_connection_lost(None) def _call_connection_lost(self, exc): """Call protocol.connection_lost().""" try: self._protocol.connection_lost(exc) finally: self._sock.close() def _fatal_error(self, exc, message='Fatal error on datagram transport'): """Handle fatal errors.""" self._loop.call_exception_handler({ 'message': message, 'exception': exc, 'transport': self, 'protocol': self._protocol, }) self.close() def get_extra_info(self, name, default=None): return self._extra.get(name, default) def is_closing(self): return self._closing def abort(self): """Close immediately.""" self._closing = True self._conn_lost += 1 self._loop.remove_reader(self._fileno) self._loop.remove_writer(self._fileno) self._buffer.clear() self._call_connection_lost(None) def get_write_buffer_size(self): """Return the current size of the write buffer.""" return sum(len(data) for data, _ in self._buffer) class _ErlangServer: """TCP server for ErlangEventLoop.""" def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog): self._loop = loop self._sockets = sockets self._protocol_factory = protocol_factory self._ssl_context = ssl_context self._backlog = backlog self._serving = False self._waiters = [] def _start_serving(self): """Start accepting connections.""" if self._serving: return self._serving = True for sock in self._sockets: self._loop.add_reader(sock.fileno(), self._accept_connection, sock) def _accept_connection(self, server_sock): """Accept a new connection.""" try: conn, addr = server_sock.accept() conn.setblocking(False) except (BlockingIOError, InterruptedError): return except OSError as exc: if exc.errno not in (errno.EMFILE, errno.ENFILE, errno.ENOBUFS, errno.ENOMEM): raise return protocol = self._protocol_factory() transport = _ErlangSocketTransport(self._loop, conn, protocol) self._loop.create_task(transport._start()) def close(self): """Stop the server.""" if not self._serving: return self._serving = False for sock in self._sockets: self._loop.remove_reader(sock.fileno()) sock.close() self._sockets.clear() async def start_serving(self): """Start serving.""" self._start_serving() async def serve_forever(self): """Serve forever.""" if not self._serving: self._start_serving() waiter = self._loop.create_future() self._waiters.append(waiter) try: await waiter finally: self._waiters.remove(waiter) def is_serving(self): return self._serving def get_loop(self): return self._loop @property def sockets(self): return tuple(self._sockets) async def __aenter__(self): return self async def __aexit__(self, *exc): self.close() await self.wait_closed() async def wait_closed(self): """Wait until server is closed.""" if self._sockets: await asyncio.sleep(0) # Import errno for _ErlangServer import errno class _MockNifModule: """Mock NIF module for testing without actual Erlang integration.""" def __init__(self): self.readers = {} self.writers = {} self.pending = [] self._counter = 0 def _is_initialized(self): return True def _poll_events(self, timeout_ms): import time time.sleep(min(timeout_ms, 10) / 1000.0) return len(self.pending) def _get_pending(self): result = list(self.pending) self.pending.clear() return result def _run_once_native(self, timeout_ms): """Combined poll + get_pending returning integer event types.""" import time time.sleep(min(timeout_ms, 10) / 1000.0) # Convert string event types to integers result = [] for callback_id, event_type in self.pending: if isinstance(event_type, str): if event_type == 'read': event_type = EVENT_TYPE_READ elif event_type == 'write': event_type = EVENT_TYPE_WRITE else: event_type = EVENT_TYPE_TIMER result.append((callback_id, event_type)) self.pending.clear() return result def _wakeup(self): pass def _add_pending(self, callback_id, type_str): self.pending.append((callback_id, type_str)) def _add_reader(self, fd, callback_id): self._counter += 1 self.readers[fd] = (callback_id, self._counter) return self._counter def _remove_reader(self, fd_key): for fd, (cid, key) in list(self.readers.items()): if key == fd_key: del self.readers[fd] break def _add_writer(self, fd, callback_id): self._counter += 1 self.writers[fd] = (callback_id, self._counter) return self._counter def _remove_writer(self, fd_key): for fd, (cid, key) in list(self.writers.items()): if key == fd_key: del self.writers[fd] break def _schedule_timer(self, delay_ms, callback_id): """Mock timer scheduling.""" return callback_id # Return callback_id as timer_ref def _cancel_timer(self, timer_ref): """Mock timer cancellation.""" pass def get_event_loop_policy(): """Get an event loop policy that uses ErlangEventLoop for the main thread. Non-main threads get the default SelectorEventLoop to avoid conflicts with the Erlang-native event loop which is designed for the main thread. """ # Capture main thread ID at policy creation time main_thread_id = threading.main_thread().ident class ErlangEventLoopPolicy(asyncio.AbstractEventLoopPolicy): def __init__(self): self._local = threading.local() def get_event_loop(self): if not hasattr(self._local, 'loop') or self._local.loop is None: self._local.loop = self.new_event_loop() return self._local.loop def set_event_loop(self, loop): self._local.loop = loop def new_event_loop(self): # Only use ErlangEventLoop for the main thread # Worker threads should use the default selector-based loop if threading.current_thread().ident == main_thread_id: return ErlangEventLoop() else: # Return default selector event loop for non-main threads return asyncio.SelectorEventLoop() return ErlangEventLoopPolicy()