Integrate extended events into Python asyncio event loop

Python ErlangEventLoop now handles all extended event types:
- EVENT_TYPE_CALL_RESULT (10), EVENT_TYPE_CALL_ERROR (11)
- EVENT_TYPE_ASYNC_RESULT (12), EVENT_TYPE_ASYNC_ERROR (13)
- EVENT_TYPE_SUBPROCESS_EXIT (20), STDOUT (21), STDERR (22)
- EVENT_TYPE_SOCKET_DATA (30), CLOSED (31), ERROR (32)

Added registration methods for protocols/futures:
- _register_call_future, _register_async_future
- _register_subprocess, _register_socket
- _next_callback_id for generating unique IDs

Extended events are 3-tuples (callback_id, type, data) vs 2-tuples.

Author: benoitc

priv/erlang_loop.py

@@ -49,6 +49,18 @@ async def main():
 EVENT_TYPE_WRITE = 2
 EVENT_TYPE_TIMER = 3
 
+# Extended event types for unified event processing
+EVENT_TYPE_CALL_RESULT = 10
+EVENT_TYPE_CALL_ERROR = 11
+EVENT_TYPE_ASYNC_RESULT = 12
+EVENT_TYPE_ASYNC_ERROR = 13
+EVENT_TYPE_SUBPROCESS_EXIT = 20
+EVENT_TYPE_SUBPROCESS_STDOUT = 21
+EVENT_TYPE_SUBPROCESS_STDERR = 22
+EVENT_TYPE_SOCKET_DATA = 30
+EVENT_TYPE_SOCKET_CLOSED = 31
+EVENT_TYPE_SOCKET_ERROR = 32
+
 # Try to import selector_events for transport classes
 try:
     from asyncio import selector_events
@@ -88,6 +100,11 @@ class ErlangEventLoop(asyncio.AbstractEventLoop):
         '_debug', '_task_factory', '_default_executor',
         # Cached method references for hot paths
         '_ready_append', '_ready_popleft',
+        # Extended event callbacks
+        '_call_futures',       # callback_id -> Future for call results
+        '_async_futures',      # callback_id -> Future for async results
+        '_subprocess_cbs',     # callback_id -> (protocol, transport)
+        '_socket_cbs',         # callback_id -> (protocol, transport)
     )
 
     def __init__(self, isolated=False):
@@ -133,6 +150,12 @@ def __init__(self, isolated=False):
         self._ready = deque()  # Callbacks ready to run
         self._callback_id = 0
 
+        # Extended event callbacks
+        self._call_futures = {}      # callback_id -> Future for call results
+        self._async_futures = {}     # callback_id -> Future for async results
+        self._subprocess_cbs = {}    # callback_id -> (protocol, transport)
+        self._socket_cbs = {}        # callback_id -> (protocol, transport)
+
         # Cache deque methods for hot path (avoids attribute lookup)
         self._ready_append = self._ready.append
         self._ready_popleft = self._ready.popleft
@@ -889,8 +912,13 @@ def _run_once(self):
             else:
                 pending = self._pel._run_once_native(timeout)
             dispatch = self._dispatch
-            for callback_id, event_type in pending:
-                dispatch(callback_id, event_type)
+            for event in pending:
+                # Extended events are 3-tuples (callback_id, event_type, data)
+                # Simple events are 2-tuples (callback_id, event_type)
+                if len(event) == 3:
+                    dispatch(event[0], event[1], event[2])
+                else:
+                    dispatch(event[0], event[1])
         except AttributeError:
             # Fallback for old NIF without _run_once_native
             try:
@@ -909,12 +937,14 @@ def _run_once(self):
             # 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):
+    def _dispatch(self, callback_id, event_type, data=None):
         """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.
+
+        Extended events (10+) carry data and resolve futures or call protocols.
         """
         # Integer comparison is faster than string - NIF returns integers
         if event_type == 1:  # EVENT_TYPE_READ
@@ -933,6 +963,63 @@ def _dispatch(self, callback_id, event_type):
                 self._handle_to_callback_id.pop(id(handle), None)  # Clean up reverse map
                 if not handle._cancelled:
                     self._ready_append(handle)
+        # Extended events - call results
+        elif event_type == 10:  # EVENT_TYPE_CALL_RESULT
+            fut = self._call_futures.pop(callback_id, None)
+            if fut is not None and not fut.done():
+                fut.set_result(data)
+        elif event_type == 11:  # EVENT_TYPE_CALL_ERROR
+            fut = self._call_futures.pop(callback_id, None)
+            if fut is not None and not fut.done():
+                fut.set_exception(Exception(data) if isinstance(data, str) else data)
+        # Extended events - async results
+        elif event_type == 12:  # EVENT_TYPE_ASYNC_RESULT
+            fut = self._async_futures.pop(callback_id, None)
+            if fut is not None and not fut.done():
+                fut.set_result(data)
+        elif event_type == 13:  # EVENT_TYPE_ASYNC_ERROR
+            fut = self._async_futures.pop(callback_id, None)
+            if fut is not None and not fut.done():
+                fut.set_exception(Exception(data) if isinstance(data, str) else data)
+        # Extended events - subprocess
+        elif event_type == 20:  # EVENT_TYPE_SUBPROCESS_EXIT
+            entry = self._subprocess_cbs.pop(callback_id, None)
+            if entry is not None:
+                protocol, transport = entry
+                self._ready_append(self._get_handle(
+                    protocol.process_exited, ()))
+        elif event_type == 21:  # EVENT_TYPE_SUBPROCESS_STDOUT
+            entry = self._subprocess_cbs.get(callback_id)
+            if entry is not None:
+                protocol, transport = entry
+                self._ready_append(self._get_handle(
+                    protocol.pipe_data_received, (1, data)))  # stdout = fd 1
+        elif event_type == 22:  # EVENT_TYPE_SUBPROCESS_STDERR
+            entry = self._subprocess_cbs.get(callback_id)
+            if entry is not None:
+                protocol, transport = entry
+                self._ready_append(self._get_handle(
+                    protocol.pipe_data_received, (2, data)))  # stderr = fd 2
+        # Extended events - socket/network
+        elif event_type == 30:  # EVENT_TYPE_SOCKET_DATA
+            entry = self._socket_cbs.get(callback_id)
+            if entry is not None:
+                protocol, transport = entry
+                self._ready_append(self._get_handle(
+                    protocol.data_received, (data,)))
+        elif event_type == 31:  # EVENT_TYPE_SOCKET_CLOSED
+            entry = self._socket_cbs.pop(callback_id, None)
+            if entry is not None:
+                protocol, transport = entry
+                self._ready_append(self._get_handle(
+                    protocol.connection_lost, (None,)))
+        elif event_type == 32:  # EVENT_TYPE_SOCKET_ERROR
+            entry = self._socket_cbs.pop(callback_id, None)
+            if entry is not None:
+                protocol, transport = entry
+                exc = Exception(data) if isinstance(data, str) else data
+                self._ready_append(self._get_handle(
+                    protocol.connection_lost, (exc,)))
 
     def _check_closed(self):
         """Raise an error if the loop is closed."""
@@ -960,6 +1047,47 @@ def _timer_handle_cancelled(self, handle):
                 except (AttributeError, RuntimeError):
                     pass
 
+    # ========================================================================
+    # Extended event registration (call, async, subprocess, socket)
+    # ========================================================================
+
+    def _register_call_future(self, callback_id, future):
+        """Register a Future for receiving call_result/call_error events."""
+        self._call_futures[callback_id] = future
+
+    def _unregister_call_future(self, callback_id):
+        """Unregister a call future."""
+        self._call_futures.pop(callback_id, None)
+
+    def _register_async_future(self, callback_id, future):
+        """Register a Future for receiving async_result/async_error events."""
+        self._async_futures[callback_id] = future
+
+    def _unregister_async_future(self, callback_id):
+        """Unregister an async future."""
+        self._async_futures.pop(callback_id, None)
+
+    def _register_subprocess(self, callback_id, protocol, transport):
+        """Register a subprocess protocol for receiving subprocess events."""
+        self._subprocess_cbs[callback_id] = (protocol, transport)
+
+    def _unregister_subprocess(self, callback_id):
+        """Unregister a subprocess."""
+        self._subprocess_cbs.pop(callback_id, None)
+
+    def _register_socket(self, callback_id, protocol, transport):
+        """Register a socket protocol for receiving socket events."""
+        self._socket_cbs[callback_id] = (protocol, transport)
+
+    def _unregister_socket(self, callback_id):
+        """Unregister a socket."""
+        self._socket_cbs.pop(callback_id, None)
+
+    def _next_callback_id(self):
+        """Generate the next unique callback ID."""
+        self._callback_id += 1
+        return self._callback_id
+
     def _set_coroutine_origin_tracking(self, enabled):
         """Enable/disable coroutine origin tracking."""
         if enabled: