Merge pull request #50 from benoitc/fix/c-audit-findings

Fix C audit findings

Author: benoitc

c_src/py_callback.c

@@ -2682,8 +2682,8 @@ static PyObject *erlang_channel_try_receive_impl(PyObject *self, PyObject *args)
  * Returns: Python object when data available
  * Raises: RuntimeError if channel closed, TimeoutError if timeout
  *
- * This function releases the GIL while waiting, allowing other Python
- * threads to run. Uses polling with short sleeps.
+ * This function releases the GIL while waiting on a condition variable,
+ * allowing other Python threads to run efficiently without polling.
  */
 static PyObject *erlang_channel_receive_impl(PyObject *self, PyObject *args) {
     (void)self;
@@ -2709,52 +2709,19 @@ static PyObject *erlang_channel_receive_impl(PyObject *self, PyObject *args) {
     size_t size = 0;
     int result;
 
-    /* First try without blocking */
-    result = channel_try_receive(channel, &data, &size);
-    if (result == 0) {
-        goto decode_and_return;
-    } else if (result == -1) {
-        PyErr_SetString(PyExc_RuntimeError, "channel closed");
-        return NULL;
-    }
-
-    /* Need to wait - release GIL and poll */
-    {
-        long elapsed_us = 0;
-        const long poll_interval_us = 100;  /* 100 microseconds */
-        const long timeout_us = timeout_ms >= 0 ? timeout_ms * 1000 : -1;
-
-        Py_BEGIN_ALLOW_THREADS
-
-        while (1) {
-            result = channel_try_receive(channel, &data, &size);
-            if (result != 1) {
-                break;  /* Got data or closed */
-            }
-
-            /* Check timeout */
-            if (timeout_us >= 0 && elapsed_us >= timeout_us) {
-                result = 2;  /* Timeout */
-                break;
-            }
-
-            /* Sleep briefly */
-            usleep(poll_interval_us);
-            elapsed_us += poll_interval_us;
-        }
+    /* Block on condition variable until data available, closed, or timeout */
+    Py_BEGIN_ALLOW_THREADS
+    result = channel_receive_blocking(channel, &data, &size, timeout_ms);
+    Py_END_ALLOW_THREADS
 
-        Py_END_ALLOW_THREADS
-    }
-
-    if (result == 2) {
+    if (result == 1) {
         PyErr_SetString(PyExc_TimeoutError, "channel receive timeout");
         return NULL;
     } else if (result == -1) {
         PyErr_SetString(PyExc_RuntimeError, "channel closed");
         return NULL;
     }
 
-decode_and_return:
     /* Decode from Erlang external term format to Python */
     {
         ErlNifEnv *tmp_env = enif_alloc_env();
@@ -2945,8 +2912,8 @@ static PyObject *erlang_byte_channel_try_receive_bytes_impl(PyObject *self, PyOb
  * Returns: bytes when data available
  * Raises: RuntimeError if channel closed, TimeoutError if timeout
  *
- * This function releases the GIL while waiting, allowing other Python
- * threads to run. Uses polling with short sleeps.
+ * This function releases the GIL while waiting on a condition variable,
+ * allowing other Python threads to run efficiently without polling.
  */
 static PyObject *erlang_byte_channel_receive_bytes_impl(PyObject *self, PyObject *args) {
     (void)self;
@@ -2972,47 +2939,12 @@ static PyObject *erlang_byte_channel_receive_bytes_impl(PyObject *self, PyObject
     size_t size = 0;
     int result;
 
-    /* First try without blocking */
-    result = channel_try_receive(channel, &data, &size);
-    if (result == 0) {
-        /* Got data - return raw bytes */
-        PyObject *bytes = PyBytes_FromStringAndSize((char *)data, size);
-        enif_free(data);
-        return bytes;
-    } else if (result == -1) {
-        PyErr_SetString(PyExc_RuntimeError, "channel closed");
-        return NULL;
-    }
-
-    /* Need to wait - release GIL and poll */
-    {
-        long elapsed_us = 0;
-        const long poll_interval_us = 100;  /* 100 microseconds */
-        const long timeout_us = timeout_ms >= 0 ? timeout_ms * 1000 : -1;
-
-        Py_BEGIN_ALLOW_THREADS
-
-        while (1) {
-            result = channel_try_receive(channel, &data, &size);
-            if (result != 1) {
-                break;  /* Got data or closed */
-            }
-
-            /* Check timeout */
-            if (timeout_us >= 0 && elapsed_us >= timeout_us) {
-                result = 2;  /* Timeout */
-                break;
-            }
-
-            /* Sleep briefly */
-            usleep(poll_interval_us);
-            elapsed_us += poll_interval_us;
-        }
-
-        Py_END_ALLOW_THREADS
-    }
+    /* Block on condition variable until data available, closed, or timeout */
+    Py_BEGIN_ALLOW_THREADS
+    result = channel_receive_blocking(channel, &data, &size, timeout_ms);
+    Py_END_ALLOW_THREADS
 
-    if (result == 2) {
+    if (result == 1) {
         PyErr_SetString(PyExc_TimeoutError, "channel receive timeout");
         return NULL;
     } else if (result == -1) {

c_src/py_channel.c

@@ -69,6 +69,9 @@ void channel_resource_dtor(ErlNifEnv *env, void *obj) {
         channel->queue = NULL;
     }
 
+    /* Wake any threads waiting on the condition before destroying */
+    pthread_cond_broadcast(&channel->data_cond);
+    pthread_cond_destroy(&channel->data_cond);
     pthread_mutex_destroy(&channel->mutex);
 }
 
@@ -91,8 +94,6 @@ py_channel_t *channel_alloc(size_t max_size) {
 
     channel->max_size = max_size;
     channel->current_size = 0;
-    channel->waiting = NULL;
-    channel->waiting_callback_id = 0;
     channel->waiter_loop = NULL;
     channel->waiter_callback_id = 0;
     channel->has_waiter = false;
@@ -107,6 +108,13 @@ py_channel_t *channel_alloc(size_t max_size) {
         return NULL;
     }
 
+    if (pthread_cond_init(&channel->data_cond, NULL) != 0) {
+        pthread_mutex_destroy(&channel->mutex);
+        enif_ioq_destroy(channel->queue);
+        enif_release_resource(channel);
+        return NULL;
+    }
+
     return channel;
 }
 
@@ -141,9 +149,6 @@ int channel_send(py_channel_t *channel, const unsigned char *data, size_t size)
 
     channel->current_size += size;
 
-    /* Check if there's a waiting context to resume */
-    bool should_resume = (channel->waiting != NULL);
-
     /* Check if there's an async waiter to dispatch.
      * IMPORTANT: Clear waiter state BEFORE releasing mutex to avoid race condition.
      * With task_ready notification, the callback can fire before we re-acquire the mutex.
@@ -171,12 +176,10 @@ int channel_send(py_channel_t *channel, const unsigned char *data, size_t size)
         channel->has_sync_waiter = false;
     }
 
-    pthread_mutex_unlock(&channel->mutex);
+    /* Signal any threads waiting on the condition variable */
+    pthread_cond_signal(&channel->data_cond);
 
-    /* Resume happens outside the lock to avoid deadlocks */
-    if (should_resume) {
-        channel_resume_waiting(channel);
-    }
+    pthread_mutex_unlock(&channel->mutex);
 
     /* Dispatch async waiter via timer dispatch (same path as timers) */
     if (loop_to_wake != NULL) {
@@ -225,9 +228,6 @@ int channel_send_owned_binary(py_channel_t *channel, ErlNifBinary *bin) {
 
     channel->current_size += msg_size;
 
-    /* Check if there's a waiting context to resume */
-    bool should_resume = (channel->waiting != NULL);
-
     /* Check if there's an async waiter to dispatch.
      * IMPORTANT: Clear waiter state BEFORE releasing mutex to avoid race condition.
      * With task_ready notification, the callback can fire before we re-acquire the mutex.
@@ -255,11 +255,10 @@ int channel_send_owned_binary(py_channel_t *channel, ErlNifBinary *bin) {
         channel->has_sync_waiter = false;
     }
 
-    pthread_mutex_unlock(&channel->mutex);
+    /* Signal any threads waiting on the condition variable */
+    pthread_cond_signal(&channel->data_cond);
 
-    if (should_resume) {
-        channel_resume_waiting(channel);
-    }
+    pthread_mutex_unlock(&channel->mutex);
 
     /* Dispatch async waiter via timer dispatch (same path as timers) */
     if (loop_to_wake != NULL) {
@@ -326,10 +325,80 @@ int channel_try_receive(py_channel_t *channel, unsigned char **out_data, size_t
     return 0;
 }
 
+int channel_receive_blocking(py_channel_t *channel, unsigned char **out_data,
+                             size_t *out_size, long timeout_ms) {
+    pthread_mutex_lock(&channel->mutex);
+
+    /* Calculate deadline for timed wait */
+    struct timespec deadline;
+    if (timeout_ms > 0) {
+        clock_gettime(CLOCK_REALTIME, &deadline);
+        deadline.tv_sec += timeout_ms / 1000;
+        deadline.tv_nsec += (timeout_ms % 1000) * 1000000;
+        if (deadline.tv_nsec >= 1000000000) {
+            deadline.tv_sec++;
+            deadline.tv_nsec -= 1000000000;
+        }
+    }
+
+    /* Wait for data or close */
+    while (enif_ioq_size(channel->queue) == 0 && !channel->closed) {
+        if (timeout_ms == 0) {
+            /* Non-blocking: return immediately if no data */
+            pthread_mutex_unlock(&channel->mutex);
+            return 1;  /* Empty/timeout */
+        } else if (timeout_ms < 0) {
+            /* Infinite wait */
+            pthread_cond_wait(&channel->data_cond, &channel->mutex);
+        } else {
+            /* Timed wait */
+            int rc = pthread_cond_timedwait(&channel->data_cond, &channel->mutex, &deadline);
+            if (rc == ETIMEDOUT) {
+                pthread_mutex_unlock(&channel->mutex);
+                return 1;  /* Timeout */
+            }
+        }
+    }
+
+    /* Check if closed with no data */
+    if (channel->closed && enif_ioq_size(channel->queue) == 0) {
+        pthread_mutex_unlock(&channel->mutex);
+        return -1;  /* Closed */
+    }
+
+    /* We have data - dequeue it */
+    SysIOVec *iov;
+    int iovcnt;
+    iov = enif_ioq_peek(channel->queue, &iovcnt);
+
+    if (iovcnt == 0 || iov == NULL || iov[0].iov_len == 0) {
+        pthread_mutex_unlock(&channel->mutex);
+        return 1;  /* Spurious wakeup, no data */
+    }
+
+    size_t msg_size = iov[0].iov_len;
+
+    /* Allocate output buffer */
+    *out_data = enif_alloc(msg_size);
+    if (*out_data == NULL) {
+        pthread_mutex_unlock(&channel->mutex);
+        return -1;  /* Allocation error */
+    }
+
+    /* Copy data and dequeue */
+    memcpy(*out_data, iov[0].iov_base, msg_size);
+    *out_size = msg_size;
+
+    enif_ioq_deq(channel->queue, msg_size, NULL);
+    channel->current_size -= msg_size;
+
+    pthread_mutex_unlock(&channel->mutex);
+    return 0;
+}
+
 void channel_close(py_channel_t *channel) {
     pthread_mutex_lock(&channel->mutex);
     channel->closed = true;
-    bool should_resume = (channel->waiting != NULL);
 
     /* Check if there's an async waiter to dispatch.
      * For close, we unconditionally clear the waiter since the channel
@@ -354,11 +423,10 @@ void channel_close(py_channel_t *channel) {
         channel->has_sync_waiter = false;
     }
 
-    pthread_mutex_unlock(&channel->mutex);
+    /* Wake all threads waiting on the condition variable */
+    pthread_cond_broadcast(&channel->data_cond);
 
-    if (should_resume) {
-        channel_resume_waiting(channel);
-    }
+    pthread_mutex_unlock(&channel->mutex);
 
     /* Dispatch async waiter to signal closure */
     if (loop_to_wake != NULL) {
@@ -377,13 +445,6 @@ void channel_close(py_channel_t *channel) {
     }
 }
 
-void channel_resume_waiting(py_channel_t *channel) {
-    /* This function would trigger resume of the suspended context.
-     * For now, the actual resume logic is handled in the NIF receive function
-     * by checking if data is available before suspending. */
-    (void)channel;
-}
-
 int channel_init(ErlNifEnv *env) {
     /* Initialize channel-specific atoms */
     ATOM_BUSY = enif_make_atom(env, "busy");

c_src/py_channel.h

@@ -86,11 +86,8 @@ typedef struct {
     /** @brief Mutex for thread-safe queue access */
     pthread_mutex_t mutex;
 
-    /** @brief Suspended Python context waiting on this channel */
-    struct suspended_context_state *waiting;
-
-    /** @brief Callback ID for the waiting context */
-    uint64_t waiting_callback_id;
+    /** @brief Condition variable for blocking receives */
+    pthread_cond_t data_cond;
 
     /** @brief Event loop for async waiter (ref-counted via enif_keep_resource) */
     struct erlang_event_loop *waiter_loop;
@@ -175,22 +172,28 @@ int channel_send_owned_binary(py_channel_t *channel, ErlNifBinary *bin);
 int channel_try_receive(py_channel_t *channel, unsigned char **out_data, size_t *out_size);
 
 /**
- * @brief Close a channel
+ * @brief Receive a message from a channel (blocking with timeout)
  *
- * Closes the channel and wakes any waiting receivers with StopIteration.
+ * Blocks on a condition variable until data is available, channel is closed,
+ * or timeout expires. This is more efficient than polling.
  *
- * @param channel Channel to close
+ * @param channel Channel to receive from
+ * @param out_data Output: message data (caller must free with enif_free)
+ * @param out_size Output: message size
+ * @param timeout_ms Timeout in milliseconds (-1 for infinite, 0 for non-blocking)
+ * @return 0 on success, 1 if timeout, -1 if closed
  */
-void channel_close(py_channel_t *channel);
+int channel_receive_blocking(py_channel_t *channel, unsigned char **out_data,
+                             size_t *out_size, long timeout_ms);
 
 /**
- * @brief Resume a waiting Python context with channel data
+ * @brief Close a channel
  *
- * Called when data becomes available for a suspended receive.
+ * Closes the channel and wakes any waiting receivers with StopIteration.
  *
- * @param channel Channel with waiting context
+ * @param channel Channel to close
  */
-void channel_resume_waiting(py_channel_t *channel);
+void channel_close(py_channel_t *channel);
 
 /* ============================================================================
  * Channel NIF Declarations