Add synchronous blocking channel receive

Implement blocking receive for channels that suspends Python while
waiting for data and releases the dirty scheduler worker.

- Add sync_waiter_pid and has_sync_waiter fields to py_channel_t
- Add channel_register_sync_waiter NIF to register calling process
- Modify channel_send to notify sync waiter via Erlang message
- Modify channel_close to notify sync waiter of channel closure
- Implement blocking handle_receive using Erlang receive to wait
- Add tests for immediate, delayed, and closed channel cases

Author: benoitc

c_src/py_channel.c

@@ -82,6 +82,8 @@ py_channel_t *channel_alloc(size_t max_size) {
     channel->waiter_loop = NULL;
     channel->waiter_callback_id = 0;
     channel->has_waiter = false;
+    channel->has_sync_waiter = false;
+    memset(&channel->sync_waiter_pid, 0, sizeof(ErlNifPid));
     channel->closed = false;
     channel->channel_id = atomic_fetch_add(&g_channel_id_counter, 1);
 
@@ -140,6 +142,16 @@ int channel_send(py_channel_t *channel, const unsigned char *data, size_t size)
         /* Note: Keep the reference until after dispatch */
     }
 
+    /* Check if there's a sync waiter to notify */
+    ErlNifPid sync_waiter;
+    bool notify_sync = false;
+
+    if (channel->has_sync_waiter) {
+        sync_waiter = channel->sync_waiter_pid;
+        notify_sync = true;
+        channel->has_sync_waiter = false;
+    }
+
     pthread_mutex_unlock(&channel->mutex);
 
     /* Resume happens outside the lock to avoid deadlocks */
@@ -154,6 +166,14 @@ int channel_send(py_channel_t *channel, const unsigned char *data, size_t size)
         enif_release_resource(loop_to_wake);
     }
 
+    /* Notify sync waiter via Erlang message */
+    if (notify_sync) {
+        ErlNifEnv *msg_env = enif_alloc_env();
+        enif_send(NULL, &sync_waiter, msg_env,
+                  enif_make_atom(msg_env, "channel_data_ready"));
+        enif_free_env(msg_env);
+    }
+
     return 0;
 }
 
@@ -198,6 +218,16 @@ int channel_send_owned_binary(py_channel_t *channel, ErlNifBinary *bin) {
         channel->waiter_loop = NULL;
     }
 
+    /* Check if there's a sync waiter to notify */
+    ErlNifPid sync_waiter;
+    bool notify_sync = false;
+
+    if (channel->has_sync_waiter) {
+        sync_waiter = channel->sync_waiter_pid;
+        notify_sync = true;
+        channel->has_sync_waiter = false;
+    }
+
     pthread_mutex_unlock(&channel->mutex);
 
     if (should_resume) {
@@ -209,6 +239,14 @@ int channel_send_owned_binary(py_channel_t *channel, ErlNifBinary *bin) {
         enif_release_resource(loop_to_wake);
     }
 
+    /* Notify sync waiter via Erlang message */
+    if (notify_sync) {
+        ErlNifEnv *msg_env = enif_alloc_env();
+        enif_send(NULL, &sync_waiter, msg_env,
+                  enif_make_atom(msg_env, "channel_data_ready"));
+        enif_free_env(msg_env);
+    }
+
     return 0;
 }
 
@@ -273,6 +311,16 @@ void channel_close(py_channel_t *channel) {
         channel->waiter_loop = NULL;
     }
 
+    /* Check if there's a sync waiter to notify */
+    ErlNifPid sync_waiter;
+    bool notify_sync = false;
+
+    if (channel->has_sync_waiter) {
+        sync_waiter = channel->sync_waiter_pid;
+        notify_sync = true;
+        channel->has_sync_waiter = false;
+    }
+
     pthread_mutex_unlock(&channel->mutex);
 
     if (should_resume) {
@@ -284,6 +332,14 @@ void channel_close(py_channel_t *channel) {
         event_loop_add_pending(loop_to_wake, EVENT_TYPE_TIMER, callback_id, -1);
         enif_release_resource(loop_to_wake);
     }
+
+    /* Notify sync waiter that channel is closed */
+    if (notify_sync) {
+        ErlNifEnv *msg_env = enif_alloc_env();
+        enif_send(NULL, &sync_waiter, msg_env,
+                  enif_make_atom(msg_env, "channel_closed"));
+        enif_free_env(msg_env);
+    }
 }
 
 void channel_resume_waiting(py_channel_t *channel) {
@@ -897,3 +953,46 @@ ERL_NIF_TERM nif_channel_cancel_wait(ErlNifEnv *env, int argc, const ERL_NIF_TER
     pthread_mutex_unlock(&channel->mutex);
     return ATOM_OK;
 }
+
+/**
+ * @brief Register a sync waiter for blocking receive
+ *
+ * nif_channel_register_sync_waiter(ChannelRef) -> ok | {error, Reason}
+ *
+ * Registers the calling process as a sync waiter. When data arrives,
+ * the waiter receives a 'channel_data_ready' message. When the channel
+ * closes, receives 'channel_closed'.
+ */
+ERL_NIF_TERM nif_channel_register_sync_waiter(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
+    (void)argc;
+    py_channel_t *channel;
+
+    if (!enif_get_resource(env, argv[0], CHANNEL_RESOURCE_TYPE, (void **)&channel)) {
+        return make_error(env, "invalid_channel");
+    }
+
+    pthread_mutex_lock(&channel->mutex);
+
+    /* Check if channel is closed */
+    if (channel->closed) {
+        pthread_mutex_unlock(&channel->mutex);
+        return enif_make_tuple2(env, ATOM_ERROR, enif_make_atom(env, "closed"));
+    }
+
+    /* Check if another sync waiter is already registered */
+    if (channel->has_sync_waiter) {
+        pthread_mutex_unlock(&channel->mutex);
+        return enif_make_tuple2(env, ATOM_ERROR, enif_make_atom(env, "waiter_exists"));
+    }
+
+    /* Get calling process PID */
+    if (!enif_self(env, &channel->sync_waiter_pid)) {
+        pthread_mutex_unlock(&channel->mutex);
+        return make_error(env, "no_calling_process");
+    }
+
+    channel->has_sync_waiter = true;
+
+    pthread_mutex_unlock(&channel->mutex);
+    return ATOM_OK;
+}

c_src/py_channel.h

@@ -105,6 +105,12 @@ typedef struct {
     /** @brief Flag: async waiter is registered */
     bool has_waiter;
 
+    /** @brief Sync waiter Erlang PID (for blocking receive) */
+    ErlNifPid sync_waiter_pid;
+
+    /** @brief Flag: sync waiter is registered */
+    bool has_sync_waiter;
+
     /** @brief Flag: channel is closed */
     bool closed;
 
@@ -265,4 +271,16 @@ ERL_NIF_TERM nif_channel_wait(ErlNifEnv *env, int argc,
 ERL_NIF_TERM nif_channel_cancel_wait(ErlNifEnv *env, int argc,
                                       const ERL_NIF_TERM argv[]);
 
+/**
+ * @brief Register a sync waiter for blocking receive
+ *
+ * NIF: channel_register_sync_waiter(ChannelRef) -> ok | {error, Reason}
+ *
+ * Registers the calling process as a sync waiter. When data arrives via
+ * channel_send, the waiter receives a 'channel_data_ready' message.
+ * When the channel closes, receives 'channel_closed'.
+ */
+ERL_NIF_TERM nif_channel_register_sync_waiter(ErlNifEnv *env, int argc,
+                                               const ERL_NIF_TERM argv[]);
+
 #endif /* PY_CHANNEL_H */

c_src/py_nif.c

@@ -3928,7 +3928,8 @@ static ErlNifFunc nif_funcs[] = {
     {"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_cancel_wait", 2, nif_channel_cancel_wait, 0},
+    {"channel_register_sync_waiter", 1, nif_channel_register_sync_waiter, 0}
 };
 
 ERL_NIF_INIT(py_nif, nif_funcs, load, NULL, upgrade, unload)

src/py_channel.erl

@@ -155,10 +155,52 @@ register_callbacks() ->
 
 %% @private
 %% Handle blocking receive from Python.
+%% This blocks until data is available by registering as a sync waiter
+%% and blocking on Erlang receive. This releases the Python worker while waiting.
 %% Args: [ChannelRef]
 -spec handle_receive([term()]) -> term().
 handle_receive([ChannelRef]) ->
-    py_nif:channel_try_receive(ChannelRef).
+    case py_nif:channel_try_receive(ChannelRef) of
+        {ok, Data} ->
+            {ok, Data};
+        {error, closed} ->
+            {error, closed};
+        {error, empty} ->
+            %% Channel is empty, register as sync waiter and block
+            case py_nif:channel_register_sync_waiter(ChannelRef) of
+                ok ->
+                    wait_for_channel_data(ChannelRef);
+                {error, Reason} ->
+                    {error, Reason}
+            end
+    end.
+
+%% @private
+%% Wait for channel data to arrive via Erlang message passing.
+%% This function blocks using native Erlang receive, which releases
+%% the Python dirty scheduler worker while waiting.
+-spec wait_for_channel_data(reference()) -> {ok, term()} | {error, term()}.
+wait_for_channel_data(ChannelRef) ->
+    receive
+        channel_data_ready ->
+            case py_nif:channel_try_receive(ChannelRef) of
+                {ok, Data} ->
+                    {ok, Data};
+                {error, empty} ->
+                    %% Race condition: data was consumed by another waiter.
+                    %% Re-register and wait again.
+                    case py_nif:channel_register_sync_waiter(ChannelRef) of
+                        ok ->
+                            wait_for_channel_data(ChannelRef);
+                        {error, Reason} ->
+                            {error, Reason}
+                    end;
+                {error, closed} ->
+                    {error, closed}
+            end;
+        channel_closed ->
+            {error, closed}
+    end.
 
 %% @private
 %% Handle non-blocking receive from Python.

src/py_nif.erl

@@ -206,7 +206,8 @@
     channel_close/1,
     channel_info/1,
     channel_wait/3,
-    channel_cancel_wait/2
+    channel_cancel_wait/2,
+    channel_register_sync_waiter/1
 ]).
 
 -on_load(load_nif/0).
@@ -1712,3 +1713,15 @@ channel_wait(_ChannelRef, _CallbackId, _LoopRef) ->
 -spec channel_cancel_wait(reference(), non_neg_integer()) -> ok.
 channel_cancel_wait(_ChannelRef, _CallbackId) ->
     ?NIF_STUB.
+
+%% @doc Register a sync waiter for blocking receive.
+%%
+%% Registers the calling process as a sync waiter. When data arrives,
+%% the waiter receives a 'channel_data_ready' message. When the channel
+%% closes, receives 'channel_closed'.
+%%
+%% @param ChannelRef Channel reference
+%% @returns ok | {error, closed} | {error, waiter_exists}
+-spec channel_register_sync_waiter(reference()) -> ok | {error, term()}.
+channel_register_sync_waiter(_ChannelRef) ->
+    ?NIF_STUB.