diff options
Diffstat (limited to 'src/lib')
| -rw-r--r-- | src/lib/dev.c | 17 | ||||
| -rw-r--r-- | src/lib/shm_ap_rbuff.c | 24 |
2 files changed, 30 insertions, 11 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c index c365a17b..fab37bbf 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -588,18 +588,21 @@ ssize_t flow_read(int fd, void * buf, size_t count) rw_lock_rdlock(&_ap_instance->flows_lock); if (_ap_instance->flows[fd].oflags & FLOW_O_NONBLOCK) { - e = shm_ap_rbuff_read(_ap_instance->rb); - } else { - - /* FIXME: this will throw away packets for other fd's */ - while (e == NULL || - e->port_id != _ap_instance->flows[fd].port_id) { - e = shm_ap_rbuff_read(_ap_instance->rb); + if (shm_ap_rbuff_peek(_ap_instance->rb) + != _ap_instance->flows[fd].port_id) { + rw_lock_unlock(&_ap_instance->flows_lock); + rw_lock_unlock(&_ap_instance->data_lock); + return -1; } + } else { /* block */ + while (shm_ap_rbuff_peek(_ap_instance->rb) + != _ap_instance->flows[fd].port_id) + ; } rw_lock_unlock(&_ap_instance->flows_lock); + e = shm_ap_rbuff_read(_ap_instance->rb); if (e == NULL) { rw_lock_unlock(&_ap_instance->data_lock); return -1; diff --git a/src/lib/shm_ap_rbuff.c b/src/lib/shm_ap_rbuff.c index 71a7e733..18fedc88 100644 --- a/src/lib/shm_ap_rbuff.c +++ b/src/lib/shm_ap_rbuff.c @@ -219,8 +219,6 @@ void shm_ap_rbuff_destroy(struct shm_ap_rbuff * rb) int shm_ap_rbuff_write(struct shm_ap_rbuff * rb, struct rb_entry * e) { - struct rb_entry * pos; - if (rb == NULL || e == NULL) return -1; @@ -231,8 +229,7 @@ int shm_ap_rbuff_write(struct shm_ap_rbuff * rb, struct rb_entry * e) return -1; } - pos = rb->shm_base + *rb->ptr_head; - *pos = *e; + *(rb->shm_base + *rb->ptr_head) = *e; *rb->ptr_head = (*rb->ptr_head + 1) & (SHM_RBUFF_SIZE -1); pthread_mutex_unlock(rb->shm_mutex); @@ -240,6 +237,25 @@ int shm_ap_rbuff_write(struct shm_ap_rbuff * rb, struct rb_entry * e) return 0; } + +int shm_ap_rbuff_peek(struct shm_ap_rbuff * rb) +{ + int port_id = -1; + + pthread_mutex_lock(rb->shm_mutex); + + if (shm_rbuff_used(rb) == 0) { + pthread_mutex_unlock(rb->shm_mutex); + return -7; /* -EAGAIN */ + } + + port_id = (rb->shm_base + *rb->ptr_tail)->port_id; + + pthread_mutex_unlock(rb->shm_mutex); + + return port_id; +} + struct rb_entry * shm_ap_rbuff_read(struct shm_ap_rbuff * rb) { struct rb_entry * e = NULL; |