X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fsvm%2Fqueue.c;h=8d70dda61a3be6a9b9e5f0c79f6a0b21acd154a0;hb=2de9c0f92;hp=8e18f5832e395d6089fcbf18d70acea3ba056f6d;hpb=3c2fed5145d9e40a9ecd178c2866c813eddc6203;p=vpp.git diff --git a/src/svm/queue.c b/src/svm/queue.c index 8e18f5832e3..8d70dda61a3 100644 --- a/src/svm/queue.c +++ b/src/svm/queue.c @@ -2,7 +2,7 @@ *------------------------------------------------------------------ * svm_queue.c - unidirectional shared-memory queues * - * Copyright (c) 2009 Cisco and/or its affiliates. + * Copyright (c) 2009-2019 Cisco and/or its affiliates. * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at: @@ -27,45 +27,25 @@ #include #include #include -#include +#include -/* - * svm_queue_init - * - * nels = number of elements on the queue - * elsize = element size, presumably 4 and cacheline-size will - * be popular choices. - * pid = consumer pid - * - * The idea is to call this function in the queue consumer, - * and e-mail the queue pointer to the producer(s). - * - * The vpp process / main thread allocates one of these - * at startup; its main input queue. The vpp main input queue - * has a pointer to it in the shared memory segment header. - * - * You probably want to be on an svm data heap before calling this - * function. - */ svm_queue_t * -svm_queue_init (int nels, - int elsize, int consumer_pid, int signal_when_queue_non_empty) +svm_queue_init (void *base, int nels, int elsize) { svm_queue_t *q; pthread_mutexattr_t attr; pthread_condattr_t cattr; - q = clib_mem_alloc_aligned (sizeof (svm_queue_t) - + nels * elsize, CLIB_CACHE_LINE_BYTES); - memset (q, 0, sizeof (*q)); + q = (svm_queue_t *) base; + clib_memset (q, 0, sizeof (*q)); q->elsize = elsize; q->maxsize = nels; - q->consumer_pid = consumer_pid; - q->signal_when_queue_non_empty = signal_when_queue_non_empty; + 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"); @@ -88,6 +68,20 @@ svm_queue_init (int nels, return (q); } +svm_queue_t * +svm_queue_alloc_and_init (int nels, int elsize, int consumer_pid) +{ + svm_queue_t *q; + + q = clib_mem_alloc_aligned (sizeof (svm_queue_t) + + nels * elsize, CLIB_CACHE_LINE_BYTES); + clib_memset (q, 0, sizeof (*q)); + q = svm_queue_init (q, nels, elsize); + q->consumer_pid = consumer_pid; + + return q; +} + /* * svm_queue_free */ @@ -117,6 +111,88 @@ svm_queue_is_full (svm_queue_t * q) return q->cursize == q->maxsize; } +static inline void +svm_queue_send_signal_inline (svm_queue_t * q, u8 is_prod) +{ + if (q->producer_evtfd == -1) + { + (void) pthread_cond_broadcast (&q->condvar); + } + else + { + int __clib_unused rv, fd; + u64 data = 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)); + if (PREDICT_FALSE (rv < 0)) + clib_unix_warning ("signal write on %d returned %d", fd, rv); + } +} + +void +svm_queue_send_signal (svm_queue_t * q, u8 is_prod) +{ + svm_queue_send_signal_inline (q, is_prod); +} + +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; + svm_queue_unlock (q); + while (q->cursize == cursize) + CLIB_PAUSE (); + svm_queue_lock (q); + } +} + +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; + + svm_queue_unlock (q); + while (q->cursize == cursize && unix_time_now () < max_time) + CLIB_PAUSE (); + rv = unix_time_now () < max_time ? 0 : ETIMEDOUT; + svm_queue_lock (q); + return rv; + } +} + +int +svm_queue_timedwait (svm_queue_t * q, double timeout) +{ + return svm_queue_timedwait_inline (q, timeout); +} + /* * svm_queue_add_nolock */ @@ -129,13 +205,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++; @@ -146,11 +220,7 @@ svm_queue_add_nolock (svm_queue_t * q, u8 * elem) q->tail = 0; if (need_broadcast) - { - (void) pthread_cond_broadcast (&q->condvar); - if (q->signal_when_queue_non_empty) - kill (q->consumer_pid, q->signal_when_queue_non_empty); - } + svm_queue_send_signal_inline (q, 1); return 0; } @@ -160,10 +230,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_inline (q, 1); } @@ -185,23 +258,21 @@ svm_queue_add (svm_queue_t * q, u8 * elem, int nowait) } } else - pthread_mutex_lock (&q->mutex); + svm_queue_lock (q); if (PREDICT_FALSE (q->cursize == q->maxsize)) { if (nowait) { - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); 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++; @@ -212,12 +283,9 @@ svm_queue_add (svm_queue_t * q, u8 * elem, int nowait) q->tail = 0; if (need_broadcast) - { - (void) pthread_cond_broadcast (&q->condvar); - if (q->signal_when_queue_non_empty) - kill (q->consumer_pid, q->signal_when_queue_non_empty); - } - pthread_mutex_unlock (&q->mutex); + svm_queue_send_signal_inline (q, 1); + + svm_queue_unlock (q); return 0; } @@ -240,23 +308,21 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) } } else - pthread_mutex_lock (&q->mutex); + svm_queue_lock (q); if (PREDICT_FALSE (q->cursize + 1 == q->maxsize)) { if (nowait) { - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); 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++; @@ -267,7 +333,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++; @@ -276,12 +342,9 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) q->tail = 0; if (need_broadcast) - { - (void) pthread_cond_broadcast (&q->condvar); - if (q->signal_when_queue_non_empty) - kill (q->consumer_pid, q->signal_when_queue_non_empty); - } - pthread_mutex_unlock (&q->mutex); + svm_queue_send_signal_inline (q, 1); + + svm_queue_unlock (q); return 0; } @@ -306,41 +369,35 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, } } else - pthread_mutex_lock (&q->mutex); + svm_queue_lock (q); if (PREDICT_FALSE (q->cursize == 0)) { if (cond == SVM_Q_NOWAIT) { - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return (-2); } 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); + svm_queue_unlock (q); return ETIMEDOUT; } } 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? */ @@ -353,9 +410,9 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, q->head = 0; if (need_broadcast) - (void) pthread_cond_broadcast (&q->condvar); + svm_queue_send_signal_inline (q, 0); - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return 0; } @@ -366,15 +423,15 @@ svm_queue_sub2 (svm_queue_t * q, u8 * elem) int need_broadcast; i8 *headp; - pthread_mutex_lock (&q->mutex); + svm_queue_lock (q); if (q->cursize == 0) { - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return -1; } 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); @@ -382,10 +439,10 @@ svm_queue_sub2 (svm_queue_t * q, u8 * elem) if (PREDICT_FALSE (q->head == q->maxsize)) q->head = 0; - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); if (need_broadcast) - (void) pthread_cond_broadcast (&q->condvar); + svm_queue_send_signal_inline (q, 0); return 0; } @@ -393,6 +450,7 @@ svm_queue_sub2 (svm_queue_t * q, u8 * elem) int svm_queue_sub_raw (svm_queue_t * q, u8 * elem) { + int need_broadcast; i8 *headp; if (PREDICT_FALSE (q->cursize == 0)) @@ -402,14 +460,31 @@ 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); + + need_broadcast = q->cursize == q->maxsize; q->head = (q->head + 1) % q->maxsize; q->cursize--; + if (PREDICT_FALSE (need_broadcast)) + svm_queue_send_signal_inline (q, 0); + return 0; } +void +svm_queue_set_producer_event_fd (svm_queue_t * q, int fd) +{ + q->producer_evtfd = fd; +} + +void +svm_queue_set_consumer_event_fd (svm_queue_t * q, int fd) +{ + q->consumer_evtfd = fd; +} + /* * fd.io coding-style-patch-verification: ON *