Collapse the dispatch paths and split the large modules

CHANGELOG.md

@@ -49,6 +49,22 @@
   storms, loop churn, 60 s mixed workload with resource counters checked.
 - Guide: `docs/isolated.md`, with what each of the three modes guarantees.
 
+### Changed
+
+- The NIF side of every context request goes through one dispatcher
+  (`ctx_dispatch`, `ctx_dispatch_async` in `c_src/py_nif.c`) instead of a
+  per-request copy of the enqueue-and-wait loop; the execute functions are
+  `ctx_execute_*` and the thread functions `ctx_thread_main_*`, since both
+  serve worker and owngil contexts. Creating a process-local env and
+  applying imports or paths run on the context thread in `worker` mode too;
+  the scheduler-side copies of those paths are gone.
+- The NIF function table is assembled from one `PY_*_NIFS` macro per area,
+  defined at the end of the file that owns the NIFs.
+- `py_context` keeps the API and the reply protocol; the process body for
+  embedded modes moved to `py_context_embedded`. `py` delegates streaming,
+  virtual environments and shared dicts to `py_stream`, `py_venv` and
+  `py_shared_dict`. The public API is unchanged.
+
 ### Removed
 
 - The legacy worker API (`py_nif:worker_new/0,1`, `worker_call`, `worker_eval`,

c_src/README.md

@@ -14,7 +14,7 @@ where things are.
 | File | What it owns | Notes |
 |---|---|---|
 | `py_nif.h` | All shared types: `py_context_t` and its request queue, request types, callback and suspension state, runtime state machine, atoms, globals, declarations | 2.4k lines. The struct comments carry the locking rules; read `py_context_t` before touching threads |
-| `py_nif.c` | Runtime init and finalize, resource types, context create/destroy, the request queue, `worker_context_thread_main` and `owngil_context_thread_main`, `owngil_execute_*` (used by both thread kinds), the `nif_context_*` NIFs, process-local envs, `py_ref`, the NIF function table at the end | Sections are banner-separated; `grep -n '^ \* ===\|^/\* ==='` lists them |
+| `py_nif.c` | Runtime init and finalize, resource types, context create/destroy, the request queue, `ctx_thread_main_worker` and `ctx_thread_main_owngil`, `ctx_execute_*` (one set for both thread kinds), `ctx_dispatch` / `ctx_dispatch_async` (the only way a NIF reaches a context thread), the `nif_context_*` NIFs, process-local envs, `py_ref`, the NIF function table at the end, assembled from the `PY_*_NIFS` macros of the other files | Sections are banner-separated; `grep -n '^ \* ===\|^/\* ==='` lists them |
 | `py_convert.c` | `py_to_term`, `term_to_py`, depth limits, tagged tuples (`{bytes, B}`, `{'$py_shm', ...}`), error tuples `{error, {Type, Msg}}` | The type mapping tables in the comments are the reference for `_etf.py` |
 | `py_exec.c` | Execution mode detection (free-threaded or GIL build) | |
 | `py_callback.c` | The `erlang` Python module: `call`, `send`, `whereis`, `schedule*`, `Atom`/`Pid`/`Ref` types, callback delivery paths (suspension, blocking pipe, async pipe), channel and shared-dict methods, callback name registry | `erlang_call_impl` documents the path precedence |
@@ -29,8 +29,8 @@ where things are.
 ## Where the live paths are
 
 - `py:call/3` in worker or owngil mode: `nif_context_call_async` (`py_nif.c`)
-  enqueues; `worker_context_thread_main` or `owngil_context_thread_main`
-  dequeues and calls `owngil_execute_request`; the reply goes out as
+  enqueues; `ctx_thread_main_worker` or `ctx_thread_main_owngil`
+  dequeues and calls `ctx_execute_request`; the reply goes out as
   `{py_result, Ref, Result}`.
 - `erlang.call` from Python: `erlang_call_impl` (`py_callback.c`).
 - Interrupt: `nif_context_interrupt` (`py_nif.c`), `interrupt_mutex` rules on
@@ -57,7 +57,9 @@ where things are.
 
 1. Implement `static ERL_NIF_TERM nif_x(ErlNifEnv*, int, const ERL_NIF_TERM[])`
    next to related code.
-2. Add `{"x", Arity, nif_x, Flags}` to `nif_funcs[]` at the end of `py_nif.c`.
+2. Add `{"x", Arity, nif_x, Flags}` to the `PY_*_NIFS` macro at the end of
+   that file (or to the `py_nif.c` block of `nif_funcs[]` for NIFs that
+   live there).
 3. Add the stub and its `-spec` and doc to `src/py_nif.erl`.
 4. Cover it in a suite; `rebar3 dialyzer` and `rebar3 xref` must stay clean.
 

c_src/py_buffer.c

@@ -1103,3 +1103,10 @@ static ERL_NIF_TERM nif_py_buffer_close(ErlNifEnv *env, int argc,
 
     return ATOM_OK;
 }
+
+/* NIF table entries of this file; py_nif.c concatenates them into nif_funcs[].
+ * Flags: ERL_NIF_DIRTY_JOB_* for anything that can block or run Python. */
+#define PY_BUFFER_NIFS \
+    {"py_buffer_create", 1, nif_py_buffer_create, 0}, \
+    {"py_buffer_write", 2, nif_py_buffer_write, 0}, \
+    {"py_buffer_close", 1, nif_py_buffer_close, 0}

c_src/py_callback.c

@@ -4053,3 +4053,9 @@ static ERL_NIF_TERM nif_unregister_callback_name(ErlNifEnv *env, int argc, const
 
     return ATOM_OK;
 }
+
+/* NIF table entries of this file; py_nif.c concatenates them into nif_funcs[].
+ * Flags: ERL_NIF_DIRTY_JOB_* for anything that can block or run Python. */
+#define PY_CALLBACK_NIFS \
+    {"register_callback_name", 1, nif_register_callback_name, 0}, \
+    {"unregister_callback_name", 1, nif_unregister_callback_name, 0}

c_src/py_channel.c

@@ -1033,3 +1033,21 @@ ERL_NIF_TERM nif_byte_channel_wait_bytes(ErlNifEnv *env, int argc, const ERL_NIF
     /* Return ok - Python will await Future */
     return ATOM_OK;
 }
+
+/* NIF table entries of this file; py_nif.c concatenates them into nif_funcs[].
+ * Flags: ERL_NIF_DIRTY_JOB_* for anything that can block or run Python. */
+#define PY_CHANNEL_NIFS \
+    {"channel_create", 0, nif_channel_create, 0}, \
+    {"channel_create", 1, nif_channel_create, 0}, \
+    {"channel_send", 2, nif_channel_send, 0}, \
+    {"channel_receive", 2, nif_channel_receive, 0}, \
+    {"channel_try_receive", 1, nif_channel_try_receive, 0}, \
+    {"channel_reply", 3, nif_channel_reply, 0}, \
+    {"channel_close", 1, nif_channel_close, 0}, \
+    {"channel_info", 1, nif_channel_info, 0}, \
+    {"channel_wait", 3, nif_channel_wait, 0}, \
+    {"channel_cancel_wait", 2, nif_channel_cancel_wait, 0}, \
+    {"channel_register_sync_waiter", 1, nif_channel_register_sync_waiter, 0}, \
+    {"byte_channel_send_bytes", 2, nif_byte_channel_send_bytes, 0}, \
+    {"byte_channel_try_receive_bytes", 1, nif_byte_channel_try_receive_bytes, 0}, \
+    {"byte_channel_wait_bytes", 3, nif_byte_channel_wait_bytes, 0}

c_src/py_event_loop.c

@@ -93,10 +93,10 @@
  */
 
 /** @brief Name for the PyCapsule storing event loop pointer */
 static const char *EVENT_LOOP_CAPSULE_NAME = "erlang_python.event_loop";
 
 /** @brief Module attribute name for storing the event loop */
 static const char *EVENT_LOOP_ATTR_NAME = "_loop";
 
 /* ============================================================================
  * Module State Structure
@@ -1285,7 +1285,7 @@
         if (!enif_get_atom(env, argv[1], atom_buf, sizeof(atom_buf), ERL_NIF_LATIN1)) {
             return make_error(env, "invalid_id");
         }
         strncpy(loop->loop_id, atom_buf, sizeof(loop->loop_id) - 1);
         loop->loop_id[sizeof(loop->loop_id) - 1] = '\0';
     } else {
         size_t copy_len = id_bin.size < sizeof(loop->loop_id) - 1 ?
@@ -2471,9 +2471,9 @@
 
         if (entry->callable == NULL) {
             /* Found empty slot */
             strncpy(entry->module_name, module, CALLABLE_NAME_MAX - 1);
             entry->module_name[CALLABLE_NAME_MAX - 1] = '\0';
             strncpy(entry->func_name, func, CALLABLE_NAME_MAX - 1);
             entry->func_name[CALLABLE_NAME_MAX - 1] = '\0';
             Py_INCREF(callable);
             entry->callable = callable;
@@ -2859,7 +2859,7 @@
     }
 #endif
     (void)loop;
     PyGILState_Release(g->gstate);
 }
 
 void event_loop_detach_interpreter(erlang_event_loop_t *loop) {
@@ -3036,7 +3036,7 @@
      * PHASE 2: Process all tasks WITH GIL (Python operations)
      * ======================================================================== */
 
     loop_gil_t gil;
     if (!loop_gil_acquire(loop, &gil)) {
         return make_error(env, "interpreter_gone");
     }
@@ -5735,7 +5735,7 @@
 #ifdef HAVE_SUBINTERPRETERS
     /* OWN_GIL mode: dispatch to dedicated thread */
     if (ctx->uses_own_gil) {
-        return dispatch_reactor_read_to_owngil(env, ctx, fd, buffer);
+        return dispatch_reactor_read(env, ctx, fd, buffer);
     }
 #endif
 
@@ -5831,7 +5831,7 @@
 #ifdef HAVE_SUBINTERPRETERS
     /* OWN_GIL mode: dispatch to dedicated thread */
     if (ctx->uses_own_gil) {
-        return dispatch_reactor_write_to_owngil(env, ctx, fd);
+        return dispatch_reactor_write(env, ctx, fd);
     }
 #endif
 
@@ -5917,7 +5917,7 @@
 #ifdef HAVE_SUBINTERPRETERS
     /* OWN_GIL mode: dispatch to dedicated thread */
     if (ctx->uses_own_gil) {
-        return dispatch_reactor_init_to_owngil(env, ctx, fd, argv[2]);
+        return dispatch_reactor_init(env, ctx, fd, argv[2]);
     }
 #endif
 
@@ -8528,3 +8528,74 @@
     }
     return 0;
 }
+
+/* NIF table entries of this file; py_nif.c concatenates them into nif_funcs[].
+ * Flags: ERL_NIF_DIRTY_JOB_* for anything that can block or run Python. */
+#define PY_EVENT_LOOP_NIFS \
+    {"set_event_loop_priv_dir", 1, nif_set_event_loop_priv_dir, 0}, \
+    {"event_loop_new", 0, nif_event_loop_new, 0}, \
+    {"event_loop_destroy", 1, nif_event_loop_destroy, 0}, \
+    {"event_loop_set_router", 2, nif_event_loop_set_router, 0}, \
+    {"event_loop_set_worker", 2, nif_event_loop_set_worker, 0}, \
+    {"event_loop_set_id", 2, nif_event_loop_set_id, 0}, \
+    {"event_loop_wakeup", 1, nif_event_loop_wakeup, 0}, \
+    {"event_loop_run_async", 7, nif_event_loop_run_async, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"submit_task", 7, nif_submit_task, 0}, \
+    {"submit_task_with_env", 8, nif_submit_task_with_env, 0}, \
+    {"process_ready_tasks", 1, nif_process_ready_tasks, ERL_NIF_DIRTY_JOB_CPU_BOUND}, \
+    {"event_loop_set_py_loop", 2, nif_event_loop_set_py_loop, 0}, \
+    {"event_loop_exec", 2, nif_event_loop_exec, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"event_loop_eval", 2, nif_event_loop_eval, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"add_reader", 3, nif_add_reader, 0}, \
+    {"remove_reader", 2, nif_remove_reader, 0}, \
+    {"add_writer", 3, nif_add_writer, 0}, \
+    {"remove_writer", 2, nif_remove_writer, 0}, \
+    {"call_later", 3, nif_call_later, 0}, \
+    {"cancel_timer", 2, nif_cancel_timer, 0}, \
+    {"poll_events", 2, nif_poll_events, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"get_pending", 1, nif_get_pending, 0}, \
+    {"dispatch_callback", 3, nif_dispatch_callback, 0}, \
+    {"dispatch_timer", 2, nif_dispatch_timer, 0}, \
+    {"get_fd_callback_id", 2, nif_get_fd_callback_id, 0}, \
+    {"reselect_reader", 2, nif_reselect_reader, 0}, \
+    {"reselect_writer", 2, nif_reselect_writer, 0}, \
+    {"reselect_reader_fd", 1, nif_reselect_reader_fd, 0}, \
+    {"reselect_writer_fd", 1, nif_reselect_writer_fd, 0}, \
+    {"handle_fd_event", 2, nif_handle_fd_event, 0}, \
+    {"handle_fd_event_and_reselect", 2, nif_handle_fd_event_and_reselect, 0}, \
+    {"fd_arm", 2, nif_fd_arm, 0}, \
+    {"stop_reader", 1, nif_stop_reader, 0}, \
+    {"start_reader", 1, nif_start_reader, 0}, \
+    {"stop_writer", 1, nif_stop_writer, 0}, \
+    {"start_writer", 1, nif_start_writer, 0}, \
+    {"close_fd", 1, nif_close_fd, 0}, \
+    {"create_test_pipe", 0, nif_create_test_pipe, 0}, \
+    {"close_test_fd", 1, nif_close_test_fd, 0}, \
+    {"dup_fd", 1, nif_dup_fd, 0}, \
+    {"write_test_fd", 2, nif_write_test_fd, 0}, \
+    {"read_test_fd", 2, nif_read_test_fd, 0}, \
+    {"create_test_tcp_listener", 1, nif_create_test_tcp_listener, 0}, \
+    {"accept_test_tcp", 1, nif_accept_test_tcp, 0}, \
+    {"connect_test_tcp", 2, nif_connect_test_tcp, 0}, \
+    {"create_test_udp_socket", 1, nif_create_test_udp_socket, 0}, \
+    {"recvfrom_test_udp", 2, nif_recvfrom_test_udp, 0}, \
+    {"sendto_test_udp", 4, nif_sendto_test_udp, 0}, \
+    {"set_udp_broadcast", 2, nif_set_udp_broadcast, 0}, \
+    {"set_python_event_loop", 1, nif_set_python_event_loop, 0}, \
+    {"set_isolation_mode", 1, nif_set_isolation_mode, 0}, \
+    {"set_shared_worker", 1, nif_set_shared_worker, 0}, \
+    {"context_get_event_loop", 1, nif_context_get_event_loop, 0}, \
+    {"reactor_register_fd", 3, nif_reactor_register_fd, 0}, \
+    {"reactor_reselect_read", 1, nif_reactor_reselect_read, 0}, \
+    {"reactor_select_write", 1, nif_reactor_select_write, 0}, \
+    {"get_fd_from_resource", 1, nif_get_fd_from_resource, 0}, \
+    {"reactor_on_read_ready", 2, nif_reactor_on_read_ready, ERL_NIF_DIRTY_JOB_CPU_BOUND}, \
+    {"reactor_on_write_ready", 2, nif_reactor_on_write_ready, ERL_NIF_DIRTY_JOB_CPU_BOUND}, \
+    {"reactor_init_connection", 3, nif_reactor_init_connection, ERL_NIF_DIRTY_JOB_CPU_BOUND}, \
+    {"reactor_close_fd", 2, nif_reactor_close_fd, 0}, \
+    {"fd_read", 2, nif_fd_read, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"fd_write", 2, nif_fd_write, ERL_NIF_DIRTY_JOB_IO_BOUND}, \
+    {"fd_select_read", 1, nif_fd_select_read, 0}, \
+    {"fd_select_write", 1, nif_fd_select_write, 0}, \
+    {"fd_close", 1, nif_fd_close, 0}, \
+    {"socketpair", 0, nif_socketpair, 0}

c_src/py_logging.c

@@ -453,3 +453,11 @@ static ERL_NIF_TERM nif_clear_trace_receiver(ErlNifEnv *env, int argc, const ERL
 
     return ATOM_OK;
 }
+
+/* NIF table entries of this file; py_nif.c concatenates them into nif_funcs[].
+ * Flags: ERL_NIF_DIRTY_JOB_* for anything that can block or run Python. */
+#define PY_LOGGING_NIFS \
+    {"set_log_receiver", 2, nif_set_log_receiver, 0}, \
+    {"clear_log_receiver", 0, nif_clear_log_receiver, 0}, \
+    {"set_trace_receiver", 1, nif_set_trace_receiver, 0}, \
+    {"clear_trace_receiver", 0, nif_clear_trace_receiver, 0}