lib: Create an rxmwheel per flow

The single retransmission wheel caused locking headaches as the calls
for different flows could block on the same rxmwheel. This stabilizes
the stack, but if the rdrbuff gets full there can now be big delays.

Signed-off-by: Dimitri Staessens <[email protected]>
Signed-off-by: Sander Vrijders <[email protected]>

1 files changed, 9 insertions, 2 deletions

diff --git a/src/lib/shm_rbuff_pthr.c b/src/lib/shm_rbuff_pthr.c
index 00ffd583..91eb8b5f 100644
--- a/src/lib/shm_rbuff_pthr.c
+++ b/src/lib/shm_rbuff_pthr.c
@@ -109,7 +109,9 @@ int shm_rbuff_write_b(struct shm_rbuff *      rb,
         pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
                              (void *) rb->lock);
 
-        while (!shm_rbuff_free(rb) && ret != -ETIMEDOUT) {
+        while (!shm_rbuff_free(rb)
+               && ret != -ETIMEDOUT
+               && !(*rb->acl & ACL_FLOWDOWN)) {
                 if (abstime != NULL)
                         ret = -pthread_cond_timedwait(rb->del,
                                                       rb->lock,
@@ -187,7 +189,9 @@ ssize_t shm_rbuff_read_b(struct shm_rbuff *      rb,
         pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
                              (void *) rb->lock);
 
-        while (shm_rbuff_empty(rb) && (idx != -ETIMEDOUT)) {
+        while (shm_rbuff_empty(rb)
+               && (idx != -ETIMEDOUT)
+               && !(*rb->acl & ACL_FLOWDOWN)) {
                 if (abstime != NULL)
                         idx = -pthread_cond_timedwait(rb->add,
                                                       rb->lock,
@@ -224,6 +228,9 @@ void shm_rbuff_set_acl(struct shm_rbuff * rb,
 #endif
         *rb->acl = (size_t) flags;
 
+        pthread_cond_broadcast(rb->del);
+        pthread_cond_broadcast(rb->add);
+
         pthread_mutex_unlock(rb->lock);
 }