summaryrefslogtreecommitdiff
path: root/src/lib
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib')
-rw-r--r--src/lib/dev.c17
-rw-r--r--src/lib/shm_ap_rbuff.c24
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;