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

lib: Stabilize fast flow deallocation over local IPCP

2 files changed, 47 insertions, 33 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);
diff --git a/src/lib/shm_ap_rbuff.c b/src/lib/shm_ap_rbuff.c
index ede0b7f7..5cbf5bd0 100644
--- a/src/lib/shm_ap_rbuff.c
+++ b/src/lib/shm_ap_rbuff.c
@@ -249,18 +249,14 @@ void shm_ap_rbuff_open_port(struct shm_ap_rbuff * rb, int port_id)
                 pthread_mutex_consistent(rb->lock);
         }
 #endif
-
-#ifdef OUROBOROS_CONFIG_DEBUG
-        if (!rb->acl[port_id])
-                LOG_DBG("Trying to open open port.");
-#endif
         rb->acl[port_id] = 0; /* open */
 
         pthread_mutex_unlock(rb->lock);
 }
 
-void shm_ap_rbuff_close_port(struct shm_ap_rbuff * rb, int port_id)
+int shm_ap_rbuff_close_port(struct shm_ap_rbuff * rb, int port_id)
 {
+        int ret = 0;
 
         assert(rb);
 
@@ -272,13 +268,14 @@ void shm_ap_rbuff_close_port(struct shm_ap_rbuff * rb, int port_id)
                 pthread_mutex_consistent(rb->lock);
         }
 #endif
-#ifdef OUROBOROS_CONFIG_DEBUG
-        if (rb->acl[port_id])
-                LOG_DBG("Trying to close closed port.");
-#endif
         rb->acl[port_id] = -1;
 
+        if (rb->cntrs[port_id] > 0)
+                ret = -EBUSY;
+
         pthread_mutex_unlock(rb->lock);
+
+        return ret;
 }
 
 void shm_ap_rbuff_destroy(struct shm_ap_rbuff * rb)
@@ -551,6 +548,8 @@ ssize_t shm_ap_rbuff_read_port(struct shm_ap_rbuff * rb, int port_id)
 {
         ssize_t idx = -1;
 
+        assert(rb);
+
 #ifdef __APPLE__
         pthread_mutex_lock(rb->lock);
 #else
@@ -559,11 +558,6 @@ ssize_t shm_ap_rbuff_read_port(struct shm_ap_rbuff * rb, int port_id)
                 pthread_mutex_consistent(rb->lock);
         }
 #endif
-        if (rb->acl[port_id]) {
-                pthread_mutex_unlock(rb->lock);
-                return -ENOTALLOC;
-        }
-
         if (shm_rbuff_empty(rb) || tail_el_ptr(rb)->port_id != port_id) {
                 pthread_mutex_unlock(rb->lock);
                 return -1;
@@ -597,11 +591,6 @@ ssize_t shm_ap_rbuff_read_port_b(struct shm_ap_rbuff *   rb,
                 pthread_mutex_consistent(rb->lock);
         }
 #endif
-        if (rb->acl[port_id]) {
-                pthread_mutex_unlock(rb->lock);
-                return -ENOTALLOC;
-        }
-
         if (timeout != NULL) {
                 idx = -ETIMEDOUT;
                 clock_gettime(PTHREAD_COND_CLOCK, &abstime);
@@ -650,7 +639,7 @@ ssize_t shm_ap_rbuff_read_port_b(struct shm_ap_rbuff *   rb,
 
         if (ret != ETIMEDOUT) {
                 idx = tail_el_ptr(rb)->index;
-                  --rb->cntrs[port_id];
+                --rb->cntrs[port_id];
                 *rb->tail = (*rb->tail + 1) & (SHM_BUFFER_SIZE -1);
 
                 pthread_cond_broadcast(rb->del);