Merged in dstaesse/ouroboros/be-ipcp-read-fd (pull request #258)

lib: Add non-copy reading from fd for IPCPs

3 files changed, 45 insertions, 10 deletions

diff --git a/src/ipcpd/shim-eth-llc/main.c b/src/ipcpd/shim-eth-llc/main.c
index c5e6d74d..130f3945 100644
--- a/src/ipcpd/shim-eth-llc/main.c
+++ b/src/ipcpd/shim-eth-llc/main.c
@@ -597,7 +597,7 @@ static void * eth_llc_ipcp_sdu_writer(void * o)
                 uint8_t dsap;
                 uint8_t r_addr[MAC_SIZE];
 
-                fd = ipcp_flow_read(&sdb);
+                fd = ipcp_read_shim(&sdb);
                 if (fd < 0)
                         continue;
 
diff --git a/src/ipcpd/shim-udp/main.c b/src/ipcpd/shim-udp/main.c
index 8c31e11a..12f6d82e 100644
--- a/src/ipcpd/shim-udp/main.c
+++ b/src/ipcpd/shim-udp/main.c
@@ -510,7 +510,7 @@ static void * ipcp_udp_sdu_loop(void * o)
                 int fd;
                 struct shm_du_buff * sdb;
 
-                fd = ipcp_flow_read(&sdb);
+                fd = ipcp_read_shim(&sdb);
                 if (fd < 0)
                         continue;
 
diff --git a/src/lib/dev.c b/src/lib/dev.c
index d36764ed..e20d23d4 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -1026,24 +1026,39 @@ int ipcp_flow_alloc_reply(int fd, int response)
         return ret;
 }
 
-int ipcp_flow_read(struct shm_du_buff ** sdb)
+int ipcp_flow_read(int fd, struct shm_du_buff ** sdb)
 {
-        int fd;
-        struct rb_entry * e;
-
-        e = shm_ap_rbuff_read(ai.rb);
+        int idx = -1;
+        int port_id = -1;
 
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_rdlock(&ai.flows_lock);
 
-        fd = ai.ports[e->port_id].fd;
+        if ((port_id = ai.flows[fd].port_id) < 0) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -ENOTALLOC;
+        }
 
-        *sdb = shm_rdrbuff_get(ai.rdrb, e->index);
+        pthread_rwlock_unlock(&ai.flows_lock);
+        pthread_rwlock_unlock(&ai.data_lock);
+
+        idx = shm_ap_rbuff_read_port(ai.rb, port_id);
+        if (idx < 0) {
+                pthread_rwlock_rdlock(&ai.data_lock);
+                pthread_rwlock_rdlock(&ai.flows_lock);
+                return idx;
+        }
+
+        pthread_rwlock_rdlock(&ai.data_lock);
+        pthread_rwlock_rdlock(&ai.flows_lock);
+
+        *sdb = shm_rdrbuff_get(ai.rdrb, idx);
 
         pthread_rwlock_unlock(&ai.flows_lock);
         pthread_rwlock_unlock(&ai.data_lock);
 
-        return fd;
+        return 0;
 }
 
 int ipcp_flow_write(int fd, struct shm_du_buff * sdb)
@@ -1114,6 +1129,26 @@ int local_flow_write(int fd, struct rb_entry * e)
         return 0;
 }
 
+int ipcp_read_shim(struct shm_du_buff ** sdb)
+{
+        int fd;
+        struct rb_entry * e;
+
+        e = shm_ap_rbuff_read(ai.rb);
+
+        pthread_rwlock_rdlock(&ai.data_lock);
+        pthread_rwlock_rdlock(&ai.flows_lock);
+
+        fd = ai.ports[e->port_id].fd;
+
+        *sdb = shm_rdrbuff_get(ai.rdrb, e->index);
+
+        pthread_rwlock_unlock(&ai.flows_lock);
+        pthread_rwlock_unlock(&ai.data_lock);
+
+        return fd;
+}
+
 void ipcp_flow_del(struct shm_du_buff * sdb)
 {
         shm_rdrbuff_remove(ai.rdrb, shm_du_buff_get_idx(sdb));