/* * Copyright (c) 2018-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: * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ /** * @file * @brief Unidirectional shared-memory multi-ring message queue */ #ifndef SRC_SVM_MESSAGE_QUEUE_H_ #define SRC_SVM_MESSAGE_QUEUE_H_ #include #include #include #include typedef struct svm_msg_q_shr_queue_ { pthread_mutex_t mutex; /* 8 bytes */ pthread_cond_t condvar; /* 8 bytes */ u32 head; u32 tail; volatile u32 cursize; u32 maxsize; u32 elsize; u32 pad; u8 data[0]; } svm_msg_q_shared_queue_t; typedef struct svm_msg_q_queue_ { svm_msg_q_shared_queue_t *shr; /**< pointer to shared queue */ int evtfd; /**< producer/consumer eventfd */ clib_spinlock_t lock; /**< private lock for multi-producer */ } svm_msg_q_queue_t; typedef struct svm_msg_q_ring_shared_ { volatile u32 cursize; /**< current size of the ring */ u32 nitems; /**< max size of the ring */ volatile u32 head; /**< current head (for dequeue) */ volatile u32 tail; /**< current tail (for enqueue) */ u32 elsize; /**< size of an element */ u8 data[0]; /**< chunk of memory for msg data */ } svm_msg_q_ring_shared_t; typedef struct svm_msg_q_ring_ { u32 nitems; /**< max size of the ring */ u32 elsize; /**< size of an element */ svm_msg_q_ring_shared_t *shr; /**< ring in shared memory */ } __clib_packed svm_msg_q_ring_t; typedef struct svm_msg_q_shared_ { u32 n_rings; /**< number of rings after q */ u32 pad; /**< 8 byte alignment for q */ svm_msg_q_shared_queue_t q[0]; /**< queue for exchanging messages */ } __clib_packed svm_msg_q_shared_t; typedef struct svm_msg_q_ { svm_msg_q_queue_t q; /**< queue for exchanging messages */ svm_msg_q_ring_t *rings; /**< rings with message data*/ } __clib_packed svm_msg_q_t; typedef struct svm_msg_q_ring_cfg_ { u32 nitems; u32 elsize; void *data; } svm_msg_q_ring_cfg_t; typedef struct svm_msg_q_cfg_ { int consumer_pid; /**< pid of msg consumer */ u32 q_nitems; /**< msg queue size (not rings) */ u32 n_rings; /**< number of msg rings */ svm_msg_q_ring_cfg_t *ring_cfgs; /**< array of ring cfgs */ } svm_msg_q_cfg_t; typedef union { struct { u32 ring_index; /**< ring index, could be u8 */ u32 elt_index; /**< index in ring */ }; u64 as_u64; } svm_msg_q_msg_t; #define SVM_MQ_INVALID_MSG { .as_u64 = ~0 } typedef enum svm_msg_q_wait_type_ { SVM_MQ_WAIT_EMPTY, SVM_MQ_WAIT_FULL } svm_msg_q_wait_type_t; /** * Allocate message queue * * Allocates a message queue on the heap. Based on the configuration options, * apart from the message queue this also allocates (one or multiple) * shared-memory rings for the messages. * * @param cfg configuration options: queue len, consumer pid, * ring configs * @return message queue */ svm_msg_q_shared_t *svm_msg_q_alloc (svm_msg_q_cfg_t *cfg); svm_msg_q_shared_t *svm_msg_q_init (void *base, svm_msg_q_cfg_t *cfg); uword svm_msg_q_size_to_alloc (svm_msg_q_cfg_t *cfg); void svm_msg_q_attach (svm_msg_q_t *mq, void *smq_base); /** * Cleanup mq's private data */ void svm_msg_q_cleanup (svm_msg_q_t *mq); /** * Free message queue * * @param mq message queue to be freed */ void svm_msg_q_free (svm_msg_q_t * mq); /** * Allocate message buffer * * Message is allocated on the first available ring capable of holding * the requested number of bytes. * * @param mq message queue * @param nbytes number of bytes needed for message * @return message structure pointing to the ring and position * allocated */ 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