Add try_pull for ring_mt

eclipse-zenoh · Apr 8, 2024 · e729237 · e729237
1 parent 63cc4b7
commit e729237
Show file tree

Hide file tree

Showing 4 changed files with 34 additions and 5 deletions.
diff --git a/include/zenoh-pico/api/handlers.h b/include/zenoh-pico/api/handlers.h
@@ -85,7 +85,7 @@ z_owned_sample_t *_z_sample_to_owned_ptr(const _z_sample_t *src);
 
 // z_owned_sample_ring_channel_t
 _Z_CHANNEL_DEFINE(sample_ring_channel, closure_sample, closure_owned_sample, z_sample_t, z_owned_sample_t, _z_ring_mt_t,
-                  _z_ring_mt_new, _z_ring_mt_free, _z_ring_mt_push, _z_ring_mt_pull, _z_ring_mt_pull,
+                  _z_ring_mt_new, _z_ring_mt_free, _z_ring_mt_push, _z_ring_mt_pull, _z_ring_mt_try_pull,
                   _z_owned_sample_move, _z_sample_to_owned_ptr, z_sample_drop)
 
 // z_owned_sample_fifo_channel_t

diff --git a/include/zenoh-pico/collections/ring_mt.h b/include/zenoh-pico/collections/ring_mt.h
@@ -25,6 +25,7 @@ typedef struct {
     _z_ring_t _ring;
 #if Z_FEATURE_MULTI_THREAD == 1
     zp_mutex_t _mutex;
+    zp_condvar_t _cv_not_empty;
 #endif
 } _z_ring_mt_t;
 
@@ -37,5 +38,6 @@ void _z_ring_mt_free(_z_ring_mt_t *ring, z_element_free_f free_f);
 int8_t _z_ring_mt_push(const void *src, void *context, z_element_free_f element_free);
 
 int8_t _z_ring_mt_pull(void *dst, void *context, z_element_move_f element_move);
+int8_t _z_ring_mt_try_pull(void *dst, void *context, z_element_move_f element_move);
 
 #endif  // ZENOH_PICO_COLLECTIONS_RING_MT_H
diff --git a/src/collections/ring_mt.c b/src/collections/ring_mt.c
@@ -24,6 +24,7 @@ int8_t _z_ring_mt_init(_z_ring_mt_t *ring, size_t capacity) {
 
 #if Z_FEATURE_MULTI_THREAD == 1
     _Z_RETURN_IF_ERR(zp_mutex_init(&ring->_mutex))
+    _Z_RETURN_IF_ERR(zp_condvar_init(&ring->_cv_not_empty))
 #endif
     return _Z_RES_OK;
 }
@@ -47,6 +48,7 @@ _z_ring_mt_t *_z_ring_mt_new(size_t capacity) {
 void _z_ring_mt_clear(_z_ring_mt_t *ring, z_element_free_f free_f) {
 #if Z_FEATURE_MULTI_THREAD == 1
     zp_mutex_free(&ring->_mutex);
+    zp_condvar_free(&ring->_cv_not_empty);
 #endif
 
     _z_ring_clear(&ring->_ring, free_f);
@@ -72,6 +74,7 @@ int8_t _z_ring_mt_push(const void *elem, void *context, z_element_free_f element
     _z_ring_push_force_drop(&r->_ring, (void *)elem, element_free);
 
 #if Z_FEATURE_MULTI_THREAD == 1
+    _Z_RETURN_IF_ERR(zp_condvar_signal(&r->_cv_not_empty))
     _Z_RETURN_IF_ERR(zp_mutex_unlock(&r->_mutex))
 #endif
     return _Z_RES_OK;
@@ -80,6 +83,30 @@ int8_t _z_ring_mt_push(const void *elem, void *context, z_element_free_f element
 int8_t _z_ring_mt_pull(void *dst, void *context, z_element_move_f element_move) {
     _z_ring_mt_t *r = (_z_ring_mt_t *)context;
 
+#if Z_FEATURE_MULTI_THREAD == 1
+    void *src = NULL;
+    _Z_RETURN_IF_ERR(zp_mutex_lock(&r->_mutex))
+    while (src == NULL) {
+        src = _z_ring_pull(&r->_ring);
+        if (src == NULL) {
+            _Z_RETURN_IF_ERR(zp_condvar_wait(&r->_cv_not_empty, &r->_mutex))
+        }
+    }
+    _Z_RETURN_IF_ERR(zp_mutex_unlock(&r->_mutex))
+    element_move(dst, src);
+#else   // Z_FEATURE_MULTI_THREAD == 1
+    void *src = _z_ring_pull(&r->_ring);
+    if (src != NULL) {
+        element_move(dst, src);
+    }
+#endif  // Z_FEATURE_MULTI_THREAD == 1
+
+    return _Z_RES_OK;
+}
+
+int8_t _z_ring_mt_try_pull(void *dst, void *context, z_element_move_f element_move) {
+    _z_ring_mt_t *r = (_z_ring_mt_t *)context;
+
 #if Z_FEATURE_MULTI_THREAD == 1
     _Z_RETURN_IF_ERR(zp_mutex_lock(&r->_mutex))
 #endif

diff --git a/tests/z_channels_test.c b/tests/z_channels_test.c
@@ -100,7 +100,7 @@ void sample_ring_channel_test_in_size(void) {
 
     char buf[100];
 
-    RECV(channel, buf)
+    TRY_RECV(channel, buf)
     assert(strcmp(buf, "") == 0);
 
     SEND(channel, "v1")
@@ -116,7 +116,7 @@ void sample_ring_channel_test_in_size(void) {
     assert(strcmp(buf, "v333") == 0);
     RECV(channel, buf)
     assert(strcmp(buf, "v4444") == 0);
-    RECV(channel, buf)
+    TRY_RECV(channel, buf)
     assert(strcmp(buf, "") == 0);
 
     z_drop(z_move(channel));
@@ -126,7 +126,7 @@ void sample_ring_channel_test_over_size(void) {
     z_owned_sample_ring_channel_t channel = z_sample_ring_channel_new(3);
 
     char buf[100];
-    RECV(channel, buf)
+    TRY_RECV(channel, buf)
     assert(strcmp(buf, "") == 0);
 
     SEND(channel, "v1")
@@ -140,7 +140,7 @@ void sample_ring_channel_test_over_size(void) {
     assert(strcmp(buf, "v333") == 0);
     RECV(channel, buf)
     assert(strcmp(buf, "v4444") == 0);
-    RECV(channel, buf)
+    TRY_RECV(channel, buf)
     assert(strcmp(buf, "") == 0);
 
     z_drop(z_move(channel));