Merged in dstaesse/ouroboros/be-fp (pull request #265)

lib: Stabilize fast flow deallocation over local IPCP

1 files changed, 37 insertions, 12 deletions

diff --git a/src/lib/dev.c b/src/lib/dev.c
index 577fa7a7..77c2d06a 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -548,6 +548,12 @@ int flow_dealloc(int fd)
                 return -ENOTALLOC;
         }
 
+        if (shm_ap_rbuff_close_port(ai.rb, ai.flows[fd].port_id) == -EBUSY) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -EBUSY;
+        }
+
         msg.port_id = ai.flows[fd].port_id;
 
         port_destroy(&ai.ports[msg.port_id]);
@@ -555,6 +561,7 @@ int flow_dealloc(int fd)
         ai.flows[fd].port_id = -1;
         shm_ap_rbuff_close(ai.flows[fd].rb);
         ai.flows[fd].rb = NULL;
+        ai.flows[fd].oflags = 0;
         ai.flows[fd].api = -1;
         if (ai.flows[fd].timeout != NULL) {
                 free(ai.flows[fd].timeout);
@@ -563,8 +570,6 @@ int flow_dealloc(int fd)
 
         bmp_release(ai.fds, fd);
 
-        shm_ap_rbuff_close_port(ai.rb, msg.port_id);
-
         pthread_rwlock_unlock(&ai.flows_lock);
         pthread_rwlock_unlock(&ai.data_lock);
 
@@ -598,6 +603,10 @@ int flow_cntl(int fd, int cmd, int oflags)
                 return old;
         case FLOW_F_SETFL: /* SET FLOW FLAGS */
                 ai.flows[fd].oflags = oflags;
+                if (oflags & FLOW_O_WRONLY)
+                        shm_ap_rbuff_close_port(ai.rb, ai.flows[fd].port_id);
+                if (oflags & FLOW_O_RDWR)
+                        shm_ap_rbuff_open_port(ai.rb, ai.flows[fd].port_id);
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
                 return old;
@@ -628,6 +637,12 @@ ssize_t flow_write(int fd, void * buf, size_t count)
                 return -ENOTALLOC;
         }
 
+        if (ai.flows[fd].oflags & FLOW_O_RDONLY) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -EPERM;
+        }
+
         if (ai.flows[fd].oflags & FLOW_O_NONBLOCK) {
                 idx = shm_rdrbuff_write(ai.rdrb,
                                        ai.flows[fd].api,
@@ -1074,6 +1089,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].oflags & FLOW_O_RDONLY) {
+                pthread_rwlock_unlock(&ai.flows_lock);
+                pthread_rwlock_unlock(&ai.data_lock);
+                return -EPERM;
+        }
+
         if (ai.flows[fd].rb == NULL) {
                 pthread_rwlock_unlock(&ai.flows_lock);
                 pthread_rwlock_unlock(&ai.data_lock);
@@ -1091,26 +1112,32 @@ int ipcp_flow_write(int fd, struct shm_du_buff * sdb)
         return 0;
 }
 
-int local_flow_read(struct rb_entry * e)
+struct rb_entry * local_flow_read(int fd)
 {
-        int fd;
-
-        *e = *(shm_ap_rbuff_read(ai.rb));
+        int port_id;
+        struct rb_entry * e = NULL;
 
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_rdlock(&ai.flows_lock);
 
-        fd = ai.ports[e->port_id].fd;
+        port_id = ai.flows[fd].port_id;
 
         pthread_rwlock_unlock(&ai.flows_lock);
         pthread_rwlock_unlock(&ai.data_lock);
 
-        return fd;
+        if (port_id != -1) {
+                e = malloc(sizeof(*e));
+                if (e == NULL)
+                        return NULL;
+                e->index = shm_ap_rbuff_read_port(ai.rb, port_id);
+        }
+
+        return e;
 }
 
 int local_flow_write(int fd, struct rb_entry * e)
 {
-        if (e == NULL)
+        if (e == NULL || fd < 0)
                 return -EINVAL;
 
         pthread_rwlock_rdlock(&ai.data_lock);
@@ -1135,9 +1162,7 @@ int local_flow_write(int fd, struct rb_entry * e)
 int ipcp_read_shim(struct shm_du_buff ** sdb)
 {
         int fd;
-        struct rb_entry * e;
-
-        e = shm_ap_rbuff_read(ai.rb);
+        struct rb_entry * e = shm_ap_rbuff_read(ai.rb);
 
         pthread_rwlock_rdlock(&ai.data_lock);
         pthread_rwlock_rdlock(&ai.flows_lock);