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 <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

1 files changed, 10 insertions, 11 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index 80d7e9ad..e8989a48 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -429,9 +429,6 @@ static void init(int     argc,
         if (pthread_rwlock_init(&ai.lock, NULL))
                 goto fail_lock;
 
-        if (rxmwheel_init())
-                goto fail_rxmwheel;
-
         ai.fqset = shm_flow_set_open(getpid());
         if (ai.fqset == NULL)
                 goto fail_fqset;
@@ -439,8 +436,6 @@ static void init(int     argc,
         return;
 
  fail_fqset:
-        rxmwheel_fini();
- fail_rxmwheel:
         pthread_rwlock_destroy(&ai.lock);
  fail_lock:
         for (i = 0; i < SYS_MAX_FLOWS; ++i)
@@ -474,8 +469,6 @@ static void fini(void)
         if (ai.fds == NULL)
                 return;
 
-        rxmwheel_fini();
-
         if (ai.prog != NULL)
                 free(ai.prog);
 
@@ -1080,15 +1073,16 @@ ssize_t flow_read(int    fd,
 
         flow = &ai.flows[fd];
 
+        clock_gettime(PTHREAD_COND_CLOCK, &abs);
+
+        pthread_rwlock_rdlock(&ai.lock);
+
         if (flow->part_idx == DONE_PART) {
+                pthread_rwlock_unlock(&ai.lock);
                 flow->part_idx = NO_PART;
                 return 0;
         }
 
-        clock_gettime(PTHREAD_COND_CLOCK, &abs);
-
-        pthread_rwlock_rdlock(&ai.lock);
-
         if (flow->flow_id < 0) {
                 pthread_rwlock_unlock(&ai.lock);
                 return -ENOTALLOC;
@@ -1141,8 +1135,13 @@ ssize_t flow_read(int    fd,
         if (n <= (ssize_t) count) {
                 memcpy(buf, packet, n);
                 shm_rdrbuff_remove(ai.rdrb, idx);
+
+                pthread_rwlock_wrlock(&ai.lock);
+
                 flow->part_idx = (partrd && n == (ssize_t) count) ?
                         DONE_PART : NO_PART;
+
+                pthread_rwlock_unlock(&ai.lock);
                 return n;
         } else {
                 if (partrd) {