X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fsvm%2Fqueue.c;h=3577c783dba00b5d042a8c69140596d09c8d9171;hb=1c17e2e;hp=0fa1fe9b230ec9d3b6e0785530f1f1d5f5006078;hpb=c470e22f12a68f06990f57f12f551fee50b6bb0d;p=vpp.git diff --git a/src/svm/queue.c b/src/svm/queue.c index 0fa1fe9b230..3577c783dba 100644 --- a/src/svm/queue.c +++ b/src/svm/queue.c @@ -27,6 +27,7 @@ #include #include #include +#include svm_queue_t * svm_queue_init (void *base, int nels, int elsize) @@ -36,15 +37,15 @@ svm_queue_init (void *base, int nels, int elsize) pthread_condattr_t cattr; q = (svm_queue_t *) base; - memset (q, 0, sizeof (*q)); + clib_memset (q, 0, sizeof (*q)); q->elsize = elsize; q->maxsize = nels; q->producer_evtfd = -1; q->consumer_evtfd = -1; - memset (&attr, 0, sizeof (attr)); - memset (&cattr, 0, sizeof (cattr)); + clib_memset (&attr, 0, sizeof (attr)); + clib_memset (&cattr, 0, sizeof (cattr)); if (pthread_mutexattr_init (&attr)) clib_unix_warning ("mutexattr_init"); @@ -74,7 +75,7 @@ svm_queue_alloc_and_init (int nels, int elsize, int consumer_pid) q = clib_mem_alloc_aligned (sizeof (svm_queue_t) + nels * elsize, CLIB_CACHE_LINE_BYTES); - memset (q, 0, sizeof (*q)); + clib_memset (q, 0, sizeof (*q)); q = svm_queue_init (q, nels, elsize); q->consumer_pid = consumer_pid; @@ -121,12 +122,69 @@ svm_queue_send_signal (svm_queue_t * q, u8 is_prod) { int __clib_unused rv, fd; u64 data = 1; - ASSERT (q->consumer_evtfd != -1); + ASSERT (q->consumer_evtfd > 0 && q->producer_evtfd > 0); fd = is_prod ? q->producer_evtfd : q->consumer_evtfd; rv = write (fd, &data, sizeof (data)); } } +static inline void +svm_queue_wait_inline (svm_queue_t * q) +{ + if (q->producer_evtfd == -1) + { + pthread_cond_wait (&q->condvar, &q->mutex); + } + else + { + /* Fake a wait for event. We could use epoll but that would mean + * using yet another fd. Should do for now */ + u32 cursize = q->cursize; + pthread_mutex_unlock (&q->mutex); + while (q->cursize == cursize) + CLIB_PAUSE (); + pthread_mutex_lock (&q->mutex); + } +} + +void +svm_queue_wait (svm_queue_t * q) +{ + svm_queue_wait_inline (q); +} + +static inline int +svm_queue_timedwait_inline (svm_queue_t * q, double timeout) +{ + struct timespec ts; + ts.tv_sec = unix_time_now () + (u32) timeout; + ts.tv_nsec = (timeout - (u32) timeout) * 1e9; + + if (q->producer_evtfd == -1) + { + return pthread_cond_timedwait (&q->condvar, &q->mutex, &ts); + } + else + { + double max_time = unix_time_now () + timeout; + u32 cursize = q->cursize; + int rv; + + pthread_mutex_unlock (&q->mutex); + while (q->cursize == cursize && unix_time_now () < max_time) + CLIB_PAUSE (); + rv = unix_time_now () < max_time ? 0 : ETIMEDOUT; + pthread_mutex_lock (&q->mutex); + return rv; + } +} + +int +svm_queue_timedwait (svm_queue_t * q, double timeout) +{ + return svm_queue_timedwait_inline (q, timeout); +} + /* * svm_queue_add_nolock */ @@ -139,13 +197,11 @@ svm_queue_add_nolock (svm_queue_t * q, u8 * elem) if (PREDICT_FALSE (q->cursize == q->maxsize)) { while (q->cursize == q->maxsize) - { - (void) pthread_cond_wait (&q->condvar, &q->mutex); - } + svm_queue_wait_inline (q); } tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); - clib_memcpy (tailp, elem, q->elsize); + clib_memcpy_fast (tailp, elem, q->elsize); q->tail++; q->cursize++; @@ -166,10 +222,13 @@ svm_queue_add_raw (svm_queue_t * q, u8 * elem) i8 *tailp; tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); - clib_memcpy (tailp, elem, q->elsize); + clib_memcpy_fast (tailp, elem, q->elsize); q->tail = (q->tail + 1) % q->maxsize; q->cursize++; + + if (q->cursize == 1) + svm_queue_send_signal (q, 1); } @@ -201,13 +260,11 @@ svm_queue_add (svm_queue_t * q, u8 * elem, int nowait) return (-2); } while (q->cursize == q->maxsize) - { - (void) pthread_cond_wait (&q->condvar, &q->mutex); - } + svm_queue_wait_inline (q); } tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); - clib_memcpy (tailp, elem, q->elsize); + clib_memcpy_fast (tailp, elem, q->elsize); q->tail++; q->cursize++; @@ -253,13 +310,11 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) return (-2); } while (q->cursize + 1 == q->maxsize) - { - (void) pthread_cond_wait (&q->condvar, &q->mutex); - } + svm_queue_wait_inline (q); } tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); - clib_memcpy (tailp, elem, q->elsize); + clib_memcpy_fast (tailp, elem, q->elsize); q->tail++; q->cursize++; @@ -270,7 +325,7 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) need_broadcast = (q->cursize == 1); tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); - clib_memcpy (tailp, elem2, q->elsize); + clib_memcpy_fast (tailp, elem2, q->elsize); q->tail++; q->cursize++; @@ -317,13 +372,9 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, } else if (cond == SVM_Q_TIMEDWAIT) { - struct timespec ts; - ts.tv_sec = unix_time_now () + time; - ts.tv_nsec = 0; while (q->cursize == 0 && rc == 0) - { - rc = pthread_cond_timedwait (&q->condvar, &q->mutex, &ts); - } + rc = svm_queue_timedwait_inline (q, time); + if (rc == ETIMEDOUT) { pthread_mutex_unlock (&q->mutex); @@ -333,14 +384,12 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, else { while (q->cursize == 0) - { - (void) pthread_cond_wait (&q->condvar, &q->mutex); - } + svm_queue_wait_inline (q); } } headp = (i8 *) (&q->data[0] + q->elsize * q->head); - clib_memcpy (elem, headp, q->elsize); + clib_memcpy_fast (elem, headp, q->elsize); q->head++; /* $$$$ JFC shouldn't this be == 0? */ @@ -374,7 +423,7 @@ svm_queue_sub2 (svm_queue_t * q, u8 * elem) } headp = (i8 *) (&q->data[0] + q->elsize * q->head); - clib_memcpy (elem, headp, q->elsize); + clib_memcpy_fast (elem, headp, q->elsize); q->head++; need_broadcast = (q->cursize == q->maxsize / 2); @@ -402,7 +451,7 @@ svm_queue_sub_raw (svm_queue_t * q, u8 * elem) } headp = (i8 *) (&q->data[0] + q->elsize * q->head); - clib_memcpy (elem, headp, q->elsize); + clib_memcpy_fast (elem, headp, q->elsize); q->head = (q->head + 1) % q->maxsize; q->cursize--;