lib: Implement timeout on blocking write

This completes the implementation of the SNDTIMEO for a blocking
write.

Fixes #6.

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

2 files changed, 57 insertions, 34 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index 2e128d59..3564c293 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -808,10 +808,12 @@ ssize_t flow_write(int          fd,
                    const void * buf,
                    size_t       count)
 {
-        struct flow * flow;
-        ssize_t       idx;
-        int           ret;
-        int           flags;
+        struct flow *     flow;
+        ssize_t           idx;
+        int               ret;
+        int               flags;
+        struct timespec   abs;
+        struct timespec * abstime = NULL;
 
         if (buf == NULL)
                 return 0;
@@ -821,6 +823,8 @@ ssize_t flow_write(int          fd,
 
         flow = &ai.flows[fd];
 
+        clock_gettime(PTHREAD_COND_CLOCK, &abs);
+
         pthread_rwlock_rdlock(&ai.lock);
 
         if (flow->port_id < 0) {
@@ -828,6 +832,11 @@ ssize_t flow_write(int          fd,
                 return -ENOTALLOC;
         }
 
+        if (ai.flows[fd].snd_timesout) {
+                ts_add(&abs, &flow->snd_timeo, &abs);
+                abstime = &abs;
+        }
+
         flags = flow->oflags;
 
         pthread_rwlock_unlock(&ai.lock);
@@ -846,7 +855,8 @@ ssize_t flow_write(int          fd,
                                           DU_BUFF_HEADSPACE,
                                           DU_BUFF_TAILSPACE,
                                           buf,
-                                          count);
+                                          count,
+                                          abstime);
         if (idx < 0)
                 return idx;
 
@@ -879,7 +889,6 @@ ssize_t flow_read(int    fd,
         uint8_t *            sdu;
         struct shm_rbuff *   rb;
         struct shm_du_buff * sdb;
-        struct timespec      now;
         struct timespec      abs;
         struct timespec *    abstime = NULL;
         struct flow *        flow;
@@ -890,7 +899,7 @@ ssize_t flow_read(int    fd,
 
         flow = &ai.flows[fd];
 
-        clock_gettime(PTHREAD_COND_CLOCK, &now);
+        clock_gettime(PTHREAD_COND_CLOCK, &abs);
 
         pthread_rwlock_rdlock(&ai.lock);
 
@@ -903,7 +912,7 @@ ssize_t flow_read(int    fd,
         noblock = flow->oflags & FLOWFRNOBLOCK;
 
         if (ai.flows[fd].rcv_timesout) {
-                ts_add(&now, &flow->rcv_timeo, &abs);
+                ts_add(&abs, &flow->rcv_timeo, &abs);
                 abstime = &abs;
         }
 
@@ -1349,7 +1358,8 @@ int ipcp_sdb_reserve(struct shm_du_buff ** sdb,
                                   DU_BUFF_HEADSPACE,
                                   DU_BUFF_TAILSPACE,
                                   NULL,
-                                  len);
+                                  len,
+                                  NULL);
 
         if (idx < 0)
                 return -1;
diff --git a/src/lib/shm_rdrbuff.c b/src/lib/shm_rdrbuff.c
index d5526c49..d1ad5a3b 100644
--- a/src/lib/shm_rdrbuff.c
+++ b/src/lib/shm_rdrbuff.c
@@ -419,11 +419,12 @@ ssize_t shm_rdrbuff_write(struct shm_rdrbuff * rdrb,
         return sdb->idx;
 }
 
-ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
-                            size_t               headspace,
-                            size_t               tailspace,
-                            const uint8_t *      data,
-                            size_t               len)
+ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff *    rdrb,
+                            size_t                  headspace,
+                            size_t                  tailspace,
+                            const uint8_t *         data,
+                            size_t                  len,
+                            const struct timespec * abstime)
 {
         struct shm_du_buff * sdb;
         size_t               size = headspace + len + tailspace;
@@ -432,6 +433,7 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
         size_t               padblocks = 0;
 #endif
         ssize_t              sz = size + sizeof(*sdb);
+        int                  ret = 0;
 
         assert(rdrb);
 
@@ -457,39 +459,50 @@ ssize_t shm_rdrbuff_write_b(struct shm_rdrbuff * rdrb,
         if (blocks + *rdrb->head > (SHM_BUFFER_SIZE))
                 padblocks = (SHM_BUFFER_SIZE) - *rdrb->head;
 
-        while (!shm_rdrb_free(rdrb, (blocks + padblocks))) {
+        while (!shm_rdrb_free(rdrb, (blocks + padblocks)) && ret != ETIMEDOUT) {
 #else
-        while (!shm_rdrb_free(rdrb, 1)) {
+        while (!shm_rdrb_free(rdrb, 1) && ret != ETIMEDOUT) {
 #endif
                 pthread_cond_broadcast(rdrb->full);
-                pthread_cond_wait(rdrb->healthy, rdrb->lock);
+                if (abstime != NULL)
+                        ret = pthread_cond_timedwait(rdrb->healthy,
+                                                     rdrb->lock,
+                                                     abstime);
+                else
+                        ret = pthread_cond_wait(rdrb->healthy, rdrb->lock);
         }
 
+        if (ret != ETIMEDOUT) {
 #ifdef SHM_RDRB_MULTI_BLOCK
-        if (padblocks) {
-                sdb = get_head_ptr(rdrb);
-                sdb->size    = 0;
-                sdb->blocks  = padblocks;
-                sdb->flags   = SDB_NULL;
-                sdb->du_head = 0;
-                sdb->du_tail = 0;
-                sdb->idx     = *rdrb->head;
-
-                *rdrb->head = 0;
-        }
+                if (padblocks) {
+                        sdb = get_head_ptr(rdrb);
+                        sdb->size    = 0;
+                        sdb->blocks  = padblocks;
+                        sdb->flags   = SDB_NULL;
+                        sdb->du_head = 0;
+                        sdb->du_tail = 0;
+                        sdb->idx     = *rdrb->head;
+
+                        *rdrb->head = 0;
+                }
 #endif
-        sdb        = get_head_ptr(rdrb);
-        sdb->flags = SDB_VALID;
-        sdb->idx   = *rdrb->head;
+                sdb        = get_head_ptr(rdrb);
+                sdb->flags = SDB_VALID;
+                sdb->idx   = *rdrb->head;
 #ifdef SHM_RDRB_MULTI_BLOCK
-        sdb->blocks  = blocks;
+                sdb->blocks  = blocks;
 
-        *rdrb->head = (*rdrb->head + blocks) & ((SHM_BUFFER_SIZE) - 1);
+                *rdrb->head = (*rdrb->head + blocks) & ((SHM_BUFFER_SIZE) - 1);
 #else
-        *rdrb->head = (*rdrb->head + 1) & ((SHM_BUFFER_SIZE) - 1);
+                *rdrb->head = (*rdrb->head + 1) & ((SHM_BUFFER_SIZE) - 1);
 #endif
+        }
+
         pthread_cleanup_pop(true);
 
+        if (ret == ETIMEDOUT)
+                return -ETIMEDOUT;
+
         sdb->size    = size;
         sdb->du_head = headspace;
         sdb->du_tail = sdb->du_head + len;