diff options
| -rw-r--r-- | src/lib/dev.c | 56 | ||||
| -rw-r--r-- | src/lib/frct.c | 52 |
2 files changed, 54 insertions, 54 deletions
diff --git a/src/lib/dev.c b/src/lib/dev.c index 5a57aa08..d950df03 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -1626,6 +1626,58 @@ bool fset_has(const struct flow_set * set, return ret; } +/* Filter fqueue events for non-data packets */ +static int fqueue_filter(struct fqueue * fq) +{ + struct shm_du_buff * sdb; + int fd; + ssize_t idx; + struct frcti * frcti; + + while (fq->next < fq->fqsize) { + if (fq->fqueue[fq->next + 1] != FLOW_PKT) + return 1; + + pthread_rwlock_rdlock(&ai.lock); + + fd = ai.ports[fq->fqueue[fq->next]].fd; + frcti = ai.flows[fd].frcti; + + if (frcti == NULL) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + if (__frcti_pdu_ready(frcti) >= 0) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + pthread_rwlock_unlock(&ai.lock); + + idx = flow_rx_sdb(&ai.flows[fd], &sdb, false, NULL); + if (idx < 0) + return 0; + + pthread_rwlock_rdlock(&ai.lock); + + sdb = shm_rdrbuff_get(ai.rdrb, idx); + + __frcti_rcv(frcti, sdb); + + if (__frcti_pdu_ready(frcti) >= 0) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + pthread_rwlock_unlock(&ai.lock); + + fq->next += 2; + } + + return fq->next < fq->fqsize; +} + int fqueue_next(struct fqueue * fq) { int fd; @@ -1638,7 +1690,7 @@ int fqueue_next(struct fqueue * fq) pthread_rwlock_rdlock(&ai.lock); - if (fq->next != 0 && frcti_filter(fq) == 0) { + if (fq->next != 0 && fqueue_filter(fq) == 0) { pthread_rwlock_unlock(&ai.lock); return -EPERM; } @@ -1692,7 +1744,7 @@ ssize_t fevent(struct flow_set * set, fq->fqsize = ret << 1; fq->next = 0; - ret = frcti_filter(fq); + ret = fqueue_filter(fq); } assert(ret); diff --git a/src/lib/frct.c b/src/lib/frct.c index 2d31e6f2..c9e45ef8 100644 --- a/src/lib/frct.c +++ b/src/lib/frct.c @@ -904,55 +904,3 @@ static void __frcti_rcv(struct frcti * frcti, shm_rdrbuff_remove(ai.rdrb, idx); return; } - -/* Filter fqueue events for non-data packets */ -int frcti_filter(struct fqueue * fq) -{ - struct shm_du_buff * sdb; - int fd; - ssize_t idx; - struct frcti * frcti; - struct shm_rbuff * rb; - - while (fq->next < fq->fqsize) { - if (fq->fqueue[fq->next + 1] != FLOW_PKT) - return 1; - - pthread_rwlock_rdlock(&ai.lock); - - fd = ai.ports[fq->fqueue[fq->next]].fd; - rb = ai.flows[fd].rx_rb; - frcti = ai.flows[fd].frcti; - - if (frcti == NULL) { - pthread_rwlock_unlock(&ai.lock); - return 1; - } - - if (__frcti_pdu_ready(frcti) >= 0) { - pthread_rwlock_unlock(&ai.lock); - return 1; - } - - idx = shm_rbuff_read(rb); - if (idx < 0) { - pthread_rwlock_unlock(&ai.lock); - return 0; - } - - sdb = shm_rdrbuff_get(ai.rdrb, idx); - - __frcti_rcv(frcti, sdb); - - if (__frcti_pdu_ready(frcti) >= 0) { - pthread_rwlock_unlock(&ai.lock); - return 1; - } - - pthread_rwlock_unlock(&ai.lock); - - fq->next += 2; - } - - return fq->next < fq->fqsize; -} |