diff options
Diffstat (limited to 'src/svm/message_queue.h')
| -rw-r--r-- | src/svm/message_queue.h | 157 |
1 files changed, 153 insertions, 4 deletions
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_ */ /* |