lib: Allow disabling partial read

This allows disabling partial reads. It adds a flag FLOWFRNOPART that
disables partial reads. Partial read is different from partial
delivery (FRCTFPARTIAL), which allows delivery of fragments of an
incomplete packet and thus potentially corrupted data. FLOWFRNOPART
will never deliver corrupted data (unless FRCTFPARTIAL is also set).
If FLOWFRNOPART is set and the buffer provided to flow_read is too
small for the SDU, that SDU will be discarded and -EMSGSIZE is
returned;

Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>
Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>

2 files changed, 20 insertions, 8 deletions

diff --git a/src/ipcpd/normal/pol/link_state.c b/src/ipcpd/normal/pol/link_state.c
index 7aa7faf7..cfe4faff 100644
--- a/src/ipcpd/normal/pol/link_state.c
+++ b/src/ipcpd/normal/pol/link_state.c
@@ -29,6 +29,7 @@
 #include <ouroboros/endian.h>
 #include <ouroboros/dev.h>
 #include <ouroboros/errno.h>
+#include <ouroboros/fccntl.h>
 #include <ouroboros/fqueue.h>
 #include <ouroboros/list.h>
 #include <ouroboros/logs.h>
@@ -655,7 +656,7 @@ static void * lsreader(void * o)
 {
         fqueue_t *   fq;
         int          ret;
-        uint8_t      buf[sizeof(struct lsa) + 1];
+        uint8_t      buf[sizeof(struct lsa)];
         int          fd;
         qosspec_t    qs;
         struct lsa * msg;
@@ -680,7 +681,7 @@ static void * lsreader(void * o)
                 }
 
                 while ((fd = fqueue_next(fq)) >= 0) {
-                        len = flow_read(fd, buf, sizeof(*msg) + 1);
+                        len = flow_read(fd, buf, sizeof(*msg));
                         if (len <= 0 || len != sizeof(*msg))
                                 continue;
 
@@ -724,6 +725,7 @@ static void handle_event(void *       self,
         /* FIXME: Apply correct QoS on graph */
         struct conn * c;
         qosspec_t     qs;
+        int           flags;
 
         (void) self;
 
@@ -759,6 +761,8 @@ static void handle_event(void *       self,
                 flow_event(c->flow_info.fd, false);
                 break;
         case NOTIFY_MGMT_CONN_ADD:
+                fccntl(c->flow_info.fd, FLOWGFLAGS, &flags);
+                fccntl(c->flow_info.fd, FLOWSFLAGS, flags | FLOWFRNOPART);
                 fset_add(ls.mgmt_set, c->flow_info.fd);
                 if (lsdb_add_nb(c->conn_info.addr, c->flow_info.fd, NB_MGMT))
                         log_warn("Failed to add mgmt neighbor to LSDB.");
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 115cd565..d0766783 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -897,6 +897,7 @@ ssize_t flow_read(int    fd,
         struct timespec *    abstime = NULL;
         struct flow *        flow;
         bool                 noblock;
+        bool                 partrd;
 
         if (fd < 0 || fd > PROG_MAX_FLOWS)
                 return -EBADF;
@@ -919,6 +920,7 @@ ssize_t flow_read(int    fd,
 
         rb   = flow->rx_rb;
         noblock = flow->oflags & FLOWFRNOBLOCK;
+        partrd = !(flow->oflags & FLOWFRNOPART);
 
         if (ai.flows[fd].rcv_timesout) {
                 ts_add(&abs, &flow->rcv_timeo, &abs);
@@ -948,14 +950,20 @@ ssize_t flow_read(int    fd,
         if (n <= (ssize_t) count) {
                 memcpy(buf, sdu, n);
                 shm_rdrbuff_remove(ai.rdrb, idx);
-                flow->part_idx = (n == (ssize_t) count) ? DONE_PART : NO_PART;
+                flow->part_idx = (partrd && n == (ssize_t) count) ?
+                        DONE_PART : NO_PART;
                 return n;
         } else {
-                memcpy(buf, sdu, count);
-                sdb = shm_rdrbuff_get(ai.rdrb, idx);
-                shm_du_buff_head_release(sdb, n);
-                flow->part_idx = idx;
-                return count;
+                if (partrd) {
+                        memcpy(buf, sdu, count);
+                        sdb = shm_rdrbuff_get(ai.rdrb, idx);
+                        shm_du_buff_head_release(sdb, n);
+                        flow->part_idx = idx;
+                        return count;
+                } else {
+                        shm_rdrbuff_remove(ai.rdrb, idx);
+                        return -EMSGSIZE;
+                }
         }
 }