Add shared router for per-loop created loops

Per-loop created loops (via _loop_new()) now automatically use a shared
router, enabling full asyncio support including timers and FD operations.

Implementation (Option B - Shared Router with Loop Identity):
- Add global shared router storage in C (g_shared_router)
- Add set_shared_router/1 NIF to set the shared router
- py_loop_new() automatically assigns shared router to new loops
- Timer messages now include LoopRef for correct dispatch
- Router stores LoopRef with each timer, dispatches to correct loop

Test:
- Add test_two_loops_with_timers verifying timer callbacks work
- All 5 integration tests pass
- All existing event loop tests pass

Author: benoitc

c_src/py_event_loop.c

@@ -92,6 +92,10 @@ static erlang_event_loop_t *g_python_event_loop;
 /* Global flag for isolation mode - set by Erlang via NIF */
 static volatile int g_isolation_mode = 0;  /* 0 = global, 1 = per_loop */
 
+/* Global shared router PID - set during init, used by all loops in per_loop mode */
+static ErlNifPid g_shared_router;
+static volatile int g_shared_router_valid = 0;
+
 /**
  * Get the py_event_loop module for the current interpreter.
  * MUST be called with GIL held.
@@ -2329,6 +2333,21 @@ ERL_NIF_TERM nif_set_isolation_mode(ErlNifEnv *env, int argc,
     return make_error(env, "invalid_mode");
 }
 
+/**
+ * Set the shared router PID for per-loop created loops.
+ * This router will be used by all loops created via _loop_new().
+ */
+ERL_NIF_TERM nif_set_shared_router(ErlNifEnv *env, int argc,
+                                    const ERL_NIF_TERM argv[]) {
+    (void)argc;
+
+    if (!enif_get_local_pid(env, argv[0], &g_shared_router)) {
+        return make_error(env, "invalid_pid");
+    }
+    g_shared_router_valid = 1;
+    return ATOM_OK;
+}
+
 /* Python function: _poll_events(timeout_ms) -> num_events */
 static PyObject *py_poll_events(PyObject *self, PyObject *args) {
     (void)self;
@@ -2952,6 +2971,12 @@ static PyObject *py_loop_new(PyObject *self, PyObject *args) {
     loop->event_freelist = NULL;
     loop->freelist_count = 0;
 
+    /* Use shared router if available (for per-loop mode) */
+    if (g_shared_router_valid) {
+        loop->router_pid = g_shared_router;
+        loop->has_router = true;
+    }
+
     /* Create a capsule wrapping the loop pointer */
     PyObject *capsule = PyCapsule_New(loop, LOOP_CAPSULE_NAME, loop_capsule_destructor);
     if (capsule == NULL) {
@@ -3260,9 +3285,13 @@ static PyObject *py_schedule_timer_for(PyObject *self, PyObject *args) {
         return NULL;
     }
 
-    ERL_NIF_TERM msg = enif_make_tuple4(
+    /* Include loop resource in message so router dispatches to correct loop */
+    ERL_NIF_TERM loop_term = enif_make_resource(msg_env, loop);
+
+    ERL_NIF_TERM msg = enif_make_tuple5(
         msg_env,
         ATOM_START_TIMER,
+        loop_term,
         enif_make_int(msg_env, delay_ms),
         enif_make_uint64(msg_env, callback_id),
         enif_make_uint64(msg_env, timer_ref_id)

c_src/py_nif.c

@@ -1924,6 +1924,7 @@ static ErlNifFunc nif_funcs[] = {
     /* Python event loop integration */
     {"set_python_event_loop", 1, nif_set_python_event_loop, 0},
     {"set_isolation_mode", 1, nif_set_isolation_mode, 0},
+    {"set_shared_router", 1, nif_set_shared_router, 0},
 
     /* ASGI optimizations */
     {"asgi_build_scope", 1, nif_asgi_build_scope, ERL_NIF_DIRTY_JOB_IO_BOUND},

src/py_event_loop.erl

@@ -98,6 +98,9 @@ init([]) ->
         {ok, LoopRef} ->
             {ok, RouterPid} = py_event_router:start_link(LoopRef),
             ok = py_nif:event_loop_set_router(LoopRef, RouterPid),
+            %% Set shared router for per-loop created loops
+            %% All loops created via _loop_new() in Python will use this router
+            ok = py_nif:set_shared_router(RouterPid),
             %% Make the event loop available to Python
             ok = py_nif:set_python_event_loop(LoopRef),
             %% Set ErlangEventLoop as the default asyncio policy

src/py_event_router.erl

@@ -42,8 +42,9 @@
 
 -record(state, {
     loop_ref :: reference(),
-    %% Map of TimerRef -> {ErlangTimerRef, CallbackId}
-    timers = #{} :: #{reference() => {reference(), non_neg_integer()}}
+    %% Map of TimerRef -> {LoopRef, ErlangTimerRef, CallbackId}
+    %% LoopRef is included to dispatch to the correct loop for per-loop timers
+    timers = #{} :: #{non_neg_integer() => {reference(), reference(), non_neg_integer()}}
 }).
 
 %% ============================================================================
@@ -100,12 +101,21 @@ handle_info({select, FdRes, _Ref, ready_output}, State) ->
     py_nif:reselect_writer_fd(FdRes),
     {noreply, State};
 
-%% Handle timer start request from call_later NIF
-handle_info({start_timer, DelayMs, CallbackId, TimerRef}, State) ->
+%% Handle timer start request from call_later NIF (new format with LoopRef)
+handle_info({start_timer, LoopRef, DelayMs, CallbackId, TimerRef}, State) ->
     #state{timers = Timers} = State,
     %% Create the actual Erlang timer
-    ErlTimerRef = erlang:send_after(DelayMs, self(), {timeout, TimerRef, CallbackId}),
-    NewTimers = maps:put(TimerRef, {ErlTimerRef, CallbackId}, Timers),
+    ErlTimerRef = erlang:send_after(DelayMs, self(), {timeout, TimerRef}),
+    %% Store LoopRef so we dispatch to the correct loop
+    NewTimers = maps:put(TimerRef, {LoopRef, ErlTimerRef, CallbackId}, Timers),
+    {noreply, State#state{timers = NewTimers}};
+
+%% Handle timer start request (legacy format without LoopRef - uses state's loop_ref)
+handle_info({start_timer, DelayMs, CallbackId, TimerRef}, State) ->
+    #state{loop_ref = LoopRef, timers = Timers} = State,
+    %% Create the actual Erlang timer
+    ErlTimerRef = erlang:send_after(DelayMs, self(), {timeout, TimerRef}),
+    NewTimers = maps:put(TimerRef, {LoopRef, ErlTimerRef, CallbackId}, Timers),
     {noreply, State#state{timers = NewTimers}};
 
 %% Handle timer cancellation
@@ -114,20 +124,26 @@ handle_info({cancel_timer, TimerRef}, State) ->
     case maps:get(TimerRef, Timers, undefined) of
         undefined ->
             {noreply, State};
-        {ErlTimerRef, _CallbackId} ->
+        {_LoopRef, ErlTimerRef, _CallbackId} ->
             erlang:cancel_timer(ErlTimerRef),
             NewTimers = maps:remove(TimerRef, Timers),
             {noreply, State#state{timers = NewTimers}}
     end;
 
 %% Handle timer expiration
-handle_info({timeout, TimerRef, CallbackId}, State) ->
-    #state{loop_ref = LoopRef, timers = Timers} = State,
-    %% Dispatch the timer callback
-    py_nif:dispatch_timer(LoopRef, CallbackId),
-    %% Remove from active timers
-    NewTimers = maps:remove(TimerRef, Timers),
-    {noreply, State#state{timers = NewTimers}};
+handle_info({timeout, TimerRef}, State) ->
+    #state{timers = Timers} = State,
+    case maps:get(TimerRef, Timers, undefined) of
+        undefined ->
+            %% Timer was cancelled
+            {noreply, State};
+        {LoopRef, _ErlTimerRef, CallbackId} ->
+            %% Dispatch the timer callback to the correct loop
+            py_nif:dispatch_timer(LoopRef, CallbackId),
+            %% Remove from active timers
+            NewTimers = maps:remove(TimerRef, Timers),
+            {noreply, State#state{timers = NewTimers}}
+    end;
 
 %% Handle select stop notifications
 handle_info({select, _FdRes, _Ref, cancelled}, State) ->
@@ -139,7 +155,7 @@ handle_info(_Info, State) ->
 
 terminate(_Reason, #state{timers = Timers}) ->
     %% Cancel all pending timers
-    maps:foreach(fun(_TimerRef, {ErlTimerRef, _CallbackId}) ->
+    maps:foreach(fun(_TimerRef, {_LoopRef, ErlTimerRef, _CallbackId}) ->
         erlang:cancel_timer(ErlTimerRef)
     end, Timers),
     ok.

src/py_nif.erl

@@ -120,6 +120,7 @@
     %% Python event loop integration
     set_python_event_loop/1,
     set_isolation_mode/1,
+    set_shared_router/1,
     %% ASGI optimizations
     asgi_build_scope/1,
     asgi_run/5,
@@ -771,6 +772,13 @@ set_python_event_loop(_LoopRef) ->
 set_isolation_mode(_Mode) ->
     ?NIF_STUB.
 
+%% @doc Set the shared router PID for per-loop created loops.
+%% All loops created via _loop_new() in Python will use this router
+%% for FD monitoring and timer operations.
+-spec set_shared_router(pid()) -> ok | {error, term()}.
+set_shared_router(_RouterPid) ->
+    ?NIF_STUB.
+
 %%% ============================================================================
 %%% ASGI Optimizations
 %%% ============================================================================

test/py_multi_loop_integration_SUITE.erl

@@ -21,15 +21,17 @@
     test_per_loop_mode_enabled/1,
     test_two_loops_concurrent_sleep/1,
     test_two_loops_concurrent_gather/1,
-    test_isolated_loop_tcp_echo/1
+    test_isolated_loop_tcp_echo/1,
+    test_two_loops_with_timers/1
 ]).
 
 all() ->
     [
         test_per_loop_mode_enabled,
         test_two_loops_concurrent_sleep,
         test_two_loops_concurrent_gather,
-        test_isolated_loop_tcp_echo
+        test_isolated_loop_tcp_echo,
+        test_two_loops_with_timers
     ].
 
 init_per_suite(Config) ->
@@ -256,3 +258,59 @@ for loop in loops:
     assert loop.is_closed(), 'Loop not closed'
 ">>),
     ok.
+
+%% ============================================================================
+%% Test: Two loops with timer callbacks (call_later)
+%% ============================================================================
+%%
+%% Tests that per-loop created loops can use timers (via shared router).
+%% This verifies that the shared router is properly set up and working.
+
+test_two_loops_with_timers(_Config) ->
+    ok = py:exec(<<"
+import time
+from erlang_loop import ErlangEventLoop
+
+# Test timer in a single loop first
+loop = ErlangEventLoop()
+timer_fired = []
+
+def timer_callback():
+    timer_fired.append(time.time())
+
+# Schedule a 50ms timer
+handle = loop.call_later(0.05, timer_callback)
+
+# Run the loop to process the timer
+start = time.time()
+deadline = start + 0.5  # 500ms timeout
+
+while time.time() < deadline and not timer_fired:
+    loop._run_once()
+    time.sleep(0.01)
+
+assert len(timer_fired) > 0, f'Timer did not fire within timeout, elapsed={time.time()-start:.3f}s'
+
+loop.close()
+
+# Now test two loops sequentially
+for loop_id in ['loop_a', 'loop_b']:
+    loop = ErlangEventLoop()
+    timer_result = []
+
+    def make_cb(lid):
+        def cb():
+            timer_result.append(lid)
+        return cb
+
+    loop.call_later(0.03, make_cb(loop_id))
+
+    start = time.time()
+    while time.time() < start + 0.3 and not timer_result:
+        loop._run_once()
+        time.sleep(0.01)
+
+    assert timer_result == [loop_id], f'Loop {loop_id} timer failed: {timer_result}'
+    loop.close()
+">>),
+    ok.