diff options
Diffstat (limited to 'src/svm')
| -rw-r--r-- | src/svm/message_queue.c | 102 | ||||
| -rw-r--r-- | src/svm/message_queue.h | 157 | ||||
| -rw-r--r-- | src/svm/queue.c | 18 | ||||
| -rw-r--r-- | src/svm/queue.h | 8 | ||||
| -rw-r--r-- | src/svm/svm_fifo.h | 6 | ||||
| -rw-r--r-- | src/svm/test_svm_message_queue.c | 4 |
6 files changed, 255 insertions, 40 deletions
diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c index 4f3e7642740..89411143c12 100644 --- a/src/svm/message_queue.c +++ b/src/svm/message_queue.c @@ -16,6 +16,25 @@ #include <svm/message_queue.h> #include <vppinfra/mem.h> +static inline svm_msg_q_ring_t * +svm_msg_q_ring_inline (svm_msg_q_t * mq, u32 ring_index) +{ + return vec_elt_at_index (mq->rings, ring_index); +} + +svm_msg_q_ring_t * +svm_msg_q_ring (svm_msg_q_t * mq, u32 ring_index) +{ + return svm_msg_q_ring_inline (mq, ring_index); +} + +static inline void * +svm_msg_q_ring_data (svm_msg_q_ring_t * ring, u32 elt_index) +{ + ASSERT (elt_index < ring->nitems); + return (ring->data + elt_index * ring->elsize); +} + svm_msg_q_t * svm_msg_q_alloc (svm_msg_q_cfg_t * cfg) { @@ -63,6 +82,53 @@ svm_msg_q_free (svm_msg_q_t * mq) } svm_msg_q_msg_t +svm_msg_q_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index) +{ + svm_msg_q_msg_t msg = {.as_u64 = ~0 }; + svm_msg_q_ring_t *ring = svm_msg_q_ring_inline (mq, ring_index); + + ASSERT (ring->cursize != ring->nitems); + msg.ring_index = ring - mq->rings; + msg.elt_index = ring->tail; + ring->tail = (ring->tail + 1) % ring->nitems; + __sync_fetch_and_add (&ring->cursize, 1); + return msg; +} + +int +svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index, + u8 noblock, svm_msg_q_msg_t * msg) +{ + if (noblock) + { + if (svm_msg_q_try_lock (mq)) + return -1; + if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, ring_index))) + { + svm_msg_q_unlock (mq); + return -2; + } + *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index); + if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (msg))) + { + svm_msg_q_unlock (mq); + return -2; + } + } + else + { + svm_msg_q_lock (mq); + *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index); + while (svm_msg_q_msg_is_invalid (msg)) + { + svm_msg_q_wait (mq); + *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index); + } + } + return 0; +} + +svm_msg_q_msg_t svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes) { svm_msg_q_msg_t msg = {.as_u64 = ~0 }; @@ -81,23 +147,10 @@ svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes) return msg; } -static inline svm_msg_q_ring_t * -svm_msg_q_get_ring (svm_msg_q_t * mq, u32 ring_index) -{ - return vec_elt_at_index (mq->rings, ring_index); -} - -static inline void * -svm_msg_q_ring_data (svm_msg_q_ring_t * ring, u32 elt_index) -{ - ASSERT (elt_index < ring->nitems); - return (ring->data + elt_index * ring->elsize); -} - void * svm_msg_q_msg_data (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) { - svm_msg_q_ring_t *ring = svm_msg_q_get_ring (mq, msg->ring_index); + svm_msg_q_ring_t *ring = svm_msg_q_ring_inline (mq, msg->ring_index); return svm_msg_q_ring_data (ring, msg->elt_index); } @@ -131,7 +184,7 @@ svm_msq_q_msg_is_valid (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) return 0; ring = &mq->rings[msg->ring_index]; - dist1 = ((ring->nitems + msg->ring_index) - ring->head) % ring->nitems; + dist1 = ((ring->nitems + msg->elt_index) - ring->head) % ring->nitems; if (ring->tail == ring->head) dist2 = (ring->cursize == 0) ? 0 : ring->nitems; else @@ -140,10 +193,17 @@ svm_msq_q_msg_is_valid (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) } int -svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait) +svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, int nowait) +{ + ASSERT (svm_msq_q_msg_is_valid (mq, msg)); + return svm_queue_add (mq->q, (u8 *) msg, nowait); +} + +void +svm_msg_q_add_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) { - ASSERT (svm_msq_q_msg_is_valid (mq, &msg)); - return svm_queue_add (mq->q, (u8 *) & msg, nowait); + ASSERT (svm_msq_q_msg_is_valid (mq, msg)); + svm_queue_add_raw (mq->q, (u8 *) msg); } int @@ -153,6 +213,12 @@ svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, return svm_queue_sub (mq->q, (u8 *) msg, cond, time); } +void +svm_msg_q_sub_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) +{ + svm_queue_sub_raw (mq->q, (u8 *) msg); +} + /* * fd.io coding-style-patch-verification: ON * diff --git a/src/svm/message_queue.h b/src/svm/message_queue.h index 5ec8547016e..708a03d716e 100644 --- a/src/svm/message_queue.h +++ b/src/svm/message_queue.h @@ -21,14 +21,15 @@ #define SRC_SVM_MESSAGE_QUEUE_H_ #include <vppinfra/clib.h> +#include <vppinfra/error.h> #include <svm/queue.h> typedef struct svm_msg_q_ring_ { volatile u32 cursize; /**< current size of the ring */ u32 nitems; /**< max size of the ring */ - u32 head; /**< current head (for dequeue) */ - u32 tail; /**< current tail (for enqueue) */ + volatile u32 head; /**< current head (for dequeue) */ + volatile u32 tail; /**< current tail (for enqueue) */ u32 elsize; /**< size of an element */ u8 *data; /**< chunk of memory for msg data */ } svm_msg_q_ring_t; @@ -64,6 +65,7 @@ typedef union u64 as_u64; } svm_msg_q_msg_t; +#define SVM_MQ_INVALID_MSG { .as_u64 = ~0 } /** * Allocate message queue * @@ -98,6 +100,36 @@ void svm_msg_q_free (svm_msg_q_t * mq); svm_msg_q_msg_t svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes); /** + * Allocate message buffer on ring + * + * Message is allocated, on requested ring. The caller MUST check that + * the ring is not full. + * + * @param mq message queue + * @param ring_index ring on which the allocation should occur + * @return message structure pointing to the ring and position + * allocated + */ +svm_msg_q_msg_t svm_msg_q_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index); + +/** + * Lock message queue and allocate message buffer on ring + * + * This should be used when multiple writers/readers are expected to + * compete for the rings/queue. Message should be enqueued by calling + * @ref svm_msg_q_add_w_lock and the caller MUST unlock the queue once + * the message in enqueued. + * + * @param mq message queue + * @param ring_index ring on which the allocation should occur + * @param noblock flag that indicates if request should block + * @param msg pointer to message to be filled in + * @return 0 on success, negative number otherwise + */ +int svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index, + u8 noblock, svm_msg_q_msg_t * msg); + +/** * Free message buffer * * Marks message buffer on ring as free. @@ -106,6 +138,7 @@ svm_msg_q_msg_t svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes); * @param msg message to be freed */ void svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg); + /** * Producer enqueue one message to queue * @@ -117,7 +150,20 @@ void svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg); * @param nowait flag to indicate if request is blocking or not * @return success status */ -int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait); +int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, int nowait); + +/** + * Producer enqueue one message to queue with mutex held + * + * Prior to calling this, the producer should've obtained a message buffer + * from one of the rings by calling @ref svm_msg_q_alloc_msg. It assumes + * the queue mutex is held. + * + * @param mq message queue + * @param msg message (pointer to ring position) to be enqueued + * @return success status + */ +void svm_msg_q_add_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg); /** * Consumer dequeue one message from queue @@ -129,13 +175,28 @@ int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait); * @param mq message queue * @param msg pointer to structure where message is to be received * @param cond flag that indicates if request should block or not + * @param time time to wait if condition it SVM_Q_TIMEDWAIT * @return success status */ int svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, svm_q_conditional_wait_t cond, u32 time); /** - * Get data for message in queu + * Consumer dequeue one message from queue with mutex held + * + * Returns the message pointing to the data in the message rings under the + * assumption that the message queue lock is already held. The consumer is + * expected to call @ref svm_msg_q_free_msg once it finishes + * processing/copies the message data. + * + * @param mq message queue + * @param msg pointer to structure where message is to be received + * @return success status + */ +void svm_msg_q_sub_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg); + +/** + * Get data for message in queue * * @param mq message queue * @param msg message for which the data is requested @@ -143,6 +204,94 @@ int svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, */ void *svm_msg_q_msg_data (svm_msg_q_t * mq, svm_msg_q_msg_t * msg); +/** + * Get message queue ring + * + * @param mq message queue + * @param ring_index index of ring + * @return pointer to ring + */ +svm_msg_q_ring_t *svm_msg_q_ring (svm_msg_q_t * mq, u32 ring_index); + +/** + * Check if message queue is full + */ +static inline u8 +svm_msg_q_is_full (svm_msg_q_t * mq) +{ + return (mq->q->cursize == mq->q->maxsize); +} + +static inline u8 +svm_msg_q_ring_is_full (svm_msg_q_t * mq, u32 ring_index) +{ + ASSERT (ring_index < vec_len (mq->rings)); + return (mq->rings[ring_index].cursize == mq->rings[ring_index].nitems); +} + +/** + * Check if message queue is empty + */ +static inline u8 +svm_msg_q_is_empty (svm_msg_q_t * mq) +{ + return (mq->q->cursize == 0); +} + +/** + * Check length of message queue + */ +static inline u32 +svm_msg_q_size (svm_msg_q_t * mq) +{ + return mq->q->cursize; +} + +/** + * Check if message is invalid + */ +static inline u8 +svm_msg_q_msg_is_invalid (svm_msg_q_msg_t * msg) +{ + return (msg->as_u64 == (u64) ~ 0); +} + +/** + * Try locking message queue + */ +static inline int +svm_msg_q_try_lock (svm_msg_q_t * mq) +{ + return pthread_mutex_trylock (&mq->q->mutex); +} + +/** + * Lock, or block trying, the message queue + */ +static inline int +svm_msg_q_lock (svm_msg_q_t * mq) +{ + return pthread_mutex_lock (&mq->q->mutex); +} + +static inline void +svm_msg_q_wait (svm_msg_q_t * mq) +{ + pthread_cond_wait (&mq->q->condvar, &mq->q->mutex); +} + +/** + * Unlock message queue + */ +static inline void +svm_msg_q_unlock (svm_msg_q_t * mq) +{ + /* The other side of the connection is not polling */ + if (mq->q->cursize < (mq->q->maxsize / 8)) + (void) pthread_cond_broadcast (&mq->q->condvar); + pthread_mutex_unlock (&mq->q->mutex); +} + #endif /* SRC_SVM_MESSAGE_QUEUE_H_ */ /* diff --git a/src/svm/queue.c b/src/svm/queue.c index 96e40fc2aec..8e18f5832e3 100644 --- a/src/svm/queue.c +++ b/src/svm/queue.c @@ -154,26 +154,16 @@ svm_queue_add_nolock (svm_queue_t * q, u8 * elem) return 0; } -int +void svm_queue_add_raw (svm_queue_t * q, u8 * elem) { i8 *tailp; - if (PREDICT_FALSE (q->cursize == q->maxsize)) - { - while (q->cursize == q->maxsize) - ; - } - tailp = (i8 *) (&q->data[0] + q->elsize * q->tail); clib_memcpy (tailp, elem, q->elsize); - q->tail++; + q->tail = (q->tail + 1) % q->maxsize; q->cursize++; - - if (q->tail == q->maxsize) - q->tail = 0; - return 0; } @@ -414,11 +404,9 @@ 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); - q->head++; + q->head = (q->head + 1) % q->maxsize; q->cursize--; - if (q->head == q->maxsize) - q->head = 0; return 0; } diff --git a/src/svm/queue.h b/src/svm/queue.h index 856c17237d1..68a63d769b6 100644 --- a/src/svm/queue.h +++ b/src/svm/queue.h @@ -69,7 +69,13 @@ void svm_queue_unlock (svm_queue_t * q); int svm_queue_is_full (svm_queue_t * q); int svm_queue_add_nolock (svm_queue_t * q, u8 * elem); int svm_queue_sub_raw (svm_queue_t * q, u8 * elem); -int svm_queue_add_raw (svm_queue_t * q, u8 * elem); + +/** + * Add element to queue with mutex held + * @param q queue + * @param elem pointer element data to add + */ +void svm_queue_add_raw (svm_queue_t * q, u8 * elem); /* * DEPRECATED please use svm_queue_t instead diff --git a/src/svm/svm_fifo.h b/src/svm/svm_fifo.h index 0d5a08b86ae..40182901db5 100644 --- a/src/svm/svm_fifo.h +++ b/src/svm/svm_fifo.h @@ -107,6 +107,12 @@ svm_fifo_max_dequeue (svm_fifo_t * f) return f->cursize; } +static inline int +svm_fifo_is_full (svm_fifo_t * f) +{ + return (f->cursize == f->nitems); +} + static inline u32 svm_fifo_max_enqueue (svm_fifo_t * f) { diff --git a/src/svm/test_svm_message_queue.c b/src/svm/test_svm_message_queue.c index 69ffd131ac2..758163ffeab 100644 --- a/src/svm/test_svm_message_queue.c +++ b/src/svm/test_svm_message_queue.c @@ -88,9 +88,9 @@ test1 (int verbose) test1_error ("failed: msg alloc3"); *(u32 *)svm_msg_q_msg_data (mq, &msg2) = 123; - svm_msg_q_add (mq, msg2, SVM_Q_NOWAIT); + svm_msg_q_add (mq, &msg2, SVM_Q_NOWAIT); for (i = 0; i < 12; i++) - svm_msg_q_add (mq, msg[i], SVM_Q_NOWAIT); + svm_msg_q_add (mq, &msg[i], SVM_Q_NOWAIT); if (svm_msg_q_sub (mq, &msg2, SVM_Q_NOWAIT, 0)) test1_error ("failed: dequeue1"); |