lib: Have flow_event_wait return non-zero value

Flow_event_wait will either return -EINVAL, -ETIMEDOUT or a positive
integer indicating the number of SDUs in the fqueue. This allows to
call the function as the condition for a non-terminating while loop.

2 files changed, 26 insertions, 12 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index ec1cd696..4b97428e 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -1002,12 +1002,16 @@ int flow_event_wait(struct flow_set *       set,
         assert(!fq->next);
 
         ret = shm_flow_set_wait(ai.fqset, set->idx, fq->fqueue, timeout);
-        if (ret == -ETIMEDOUT)
+        if (ret == -ETIMEDOUT) {
+                fq->fqsize = 0;
                 return -ETIMEDOUT;
+        }
 
         fq->fqsize = ret;
 
-        return 0;
+        assert(ret);
+
+        return ret;
 }
 
 /* ipcp-dev functions */
@@ -1295,6 +1299,12 @@ int ipcp_flow_write(int fd, struct shm_du_buff * sdb)
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_rdlock(&ai.flows_lock);
 
+        if (ai.flows[fd].port_id < 0) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -ENOTALLOC;
+        }
+
         if ((ai.flows[fd].oflags & FLOW_O_ACCMODE) == FLOW_O_RDONLY) {
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -1356,7 +1366,11 @@ int local_flow_write(int fd, size_t idx)
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_rdlock(&ai.flows_lock);
 
-        assert(ai.flows[fd].tx_rb);
+        if (ai.flows[fd].port_id < 0) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -ENOTALLOC;
+        }
 
         shm_rbuff_write(ai.flows[fd].tx_rb, idx);
 
diff --git a/src/lib/shm_flow_set.c b/src/lib/shm_flow_set.c
index 3b1af83f..6cc94573 100644
--- a/src/lib/shm_flow_set.c
+++ b/src/lib/shm_flow_set.c
@@ -376,22 +376,20 @@ ssize_t shm_flow_set_wait(const struct shm_flow_set * shm_set,
 
         while (shm_set->heads[idx] == 0 && ret != -ETIMEDOUT) {
                 if (timeout != NULL)
-                        ret = pthread_cond_timedwait(shm_set->conds + idx,
-                                                     shm_set->lock,
-                                                     &abstime);
+                        ret = -pthread_cond_timedwait(shm_set->conds + idx,
+                                                      shm_set->lock,
+                                                      &abstime);
                 else
-                        ret = pthread_cond_wait(shm_set->conds + idx,
-                                                shm_set->lock);
+                        ret = -pthread_cond_wait(shm_set->conds + idx,
+                                                 shm_set->lock);
 #ifndef __APPLE__
-                if (ret == EOWNERDEAD) {
+                if (ret == -EOWNERDEAD) {
                         LOG_DBG("Recovering dead mutex.");
                         pthread_mutex_consistent(shm_set->lock);
                 }
 #endif
-                if (ret == ETIMEDOUT) {
-                        ret = -ETIMEDOUT;
+                if (ret == -ETIMEDOUT)
                         break;
-                }
         }
 
         if (ret != -ETIMEDOUT) {
@@ -404,5 +402,7 @@ ssize_t shm_flow_set_wait(const struct shm_flow_set * shm_set,
 
         pthread_cleanup_pop(true);
 
+        assert(ret);
+
         return ret;
 }