X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fsvm%2Fqueue.c;h=78444d8ede48695d4780a81214955d4966d6b696;hb=4f30bc74577959efddfe4bc544cea4ec5c7019da;hp=0fa1fe9b230ec9d3b6e0785530f1f1d5f5006078;hpb=c470e22f12a68f06990f57f12f551fee50b6bb0d;p=vpp.git diff --git a/src/svm/queue.c b/src/svm/queue.c index 0fa1fe9b230..78444d8ede4 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,6 +27,7 @@ #include #include #include +#include svm_queue_t * svm_queue_init (void *base, int nels, int elsize) @@ -36,20 +37,22 @@ 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"); if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED)) clib_unix_warning ("pthread_mutexattr_setpshared"); + if (pthread_mutexattr_setrobust (&attr, PTHREAD_MUTEX_ROBUST)) + clib_unix_warning ("setrobust"); if (pthread_mutex_init (&q->mutex, &attr)) clib_unix_warning ("mutex_init"); if (pthread_mutexattr_destroy (&attr)) @@ -74,7 +77,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; @@ -95,7 +98,18 @@ svm_queue_free (svm_queue_t * q) void svm_queue_lock (svm_queue_t * q) { - pthread_mutex_lock (&q->mutex); + int rv = pthread_mutex_lock (&q->mutex); + if (PREDICT_FALSE (rv == EOWNERDEAD)) + pthread_mutex_consistent (&q->mutex); +} + +static int +svm_queue_trylock (svm_queue_t * q) +{ + int rv = pthread_mutex_trylock (&q->mutex); + if (PREDICT_FALSE (rv == EOWNERDEAD)) + rv = pthread_mutex_consistent (&q->mutex); + return rv; } void @@ -111,7 +125,7 @@ svm_queue_is_full (svm_queue_t * q) } static inline void -svm_queue_send_signal (svm_queue_t * q, u8 is_prod) +svm_queue_send_signal_inline (svm_queue_t * q, u8 is_prod) { if (q->producer_evtfd == -1) { @@ -121,10 +135,75 @@ 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)); + 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); } /* @@ -139,13 +218,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++; @@ -156,7 +233,7 @@ svm_queue_add_nolock (svm_queue_t * q, u8 * elem) q->tail = 0; if (need_broadcast) - svm_queue_send_signal (q, 1); + svm_queue_send_signal_inline (q, 1); return 0; } @@ -166,10 +243,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,29 +265,27 @@ svm_queue_add (svm_queue_t * q, u8 * elem, int nowait) if (nowait) { /* zero on success */ - if (pthread_mutex_trylock (&q->mutex)) + if (svm_queue_trylock (q)) { return (-1); } } 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++; @@ -218,9 +296,9 @@ svm_queue_add (svm_queue_t * q, u8 * elem, int nowait) q->tail = 0; if (need_broadcast) - svm_queue_send_signal (q, 1); + svm_queue_send_signal_inline (q, 1); - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return 0; } @@ -237,29 +315,27 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) if (nowait) { /* zero on success */ - if (pthread_mutex_trylock (&q->mutex)) + if (svm_queue_trylock (q)) { return (-1); } } else - pthread_mutex_lock (&q->mutex); + svm_queue_lock (q); - if (PREDICT_FALSE (q->cursize + 1 == q->maxsize)) + 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); - } + while (q->cursize + 1 >= q->maxsize) + 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 +346,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++; @@ -279,9 +355,9 @@ svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait) q->tail = 0; if (need_broadcast) - svm_queue_send_signal (q, 1); + svm_queue_send_signal_inline (q, 1); - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return 0; } @@ -300,47 +376,41 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, if (cond == SVM_Q_NOWAIT) { /* zero on success */ - if (pthread_mutex_trylock (&q->mutex)) + if (svm_queue_trylock (q)) { return (-1); } } 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 +423,9 @@ svm_queue_sub (svm_queue_t * q, u8 * elem, svm_q_conditional_wait_t cond, q->head = 0; if (need_broadcast) - svm_queue_send_signal (q, 0); + svm_queue_send_signal_inline (q, 0); - pthread_mutex_unlock (&q->mutex); + svm_queue_unlock (q); return 0; } @@ -366,15 +436,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 +452,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) - svm_queue_send_signal (q, 0); + svm_queue_send_signal_inline (q, 0); return 0; } @@ -393,6 +463,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,11 +473,16 @@ 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; }