diff options
Diffstat (limited to 'src/svm')
| -rw-r--r-- | src/svm/fifo_segment.c | 69 | ||||
| -rw-r--r-- | src/svm/fifo_segment.h | 33 | ||||
| -rw-r--r-- | src/svm/fifo_types.h | 1 | ||||
| -rw-r--r-- | src/svm/message_queue.c | 136 | ||||
| -rw-r--r-- | src/svm/message_queue.h | 28 | ||||
| -rw-r--r-- | src/svm/svm_fifo.c | 1 |
6 files changed, 198 insertions, 70 deletions
diff --git a/src/svm/fifo_segment.c b/src/svm/fifo_segment.c index 636f223cde1..0c3a79a613b 100644 --- a/src/svm/fifo_segment.c +++ b/src/svm/fifo_segment.c @@ -764,9 +764,15 @@ void fifo_segment_cleanup (fifo_segment_t *fs) { int slice_index; + svm_msg_q_t *mq = 0; for (slice_index = 0; slice_index < fs->n_slices; slice_index++) clib_mem_bulk_destroy (fs->slices[slice_index].fifos); + + vec_foreach (fs->mqs, mq) + vec_free (mq->rings); + + vec_free (fs->mqs); } /** @@ -944,6 +950,69 @@ fifo_segment_attach_fifo (fifo_segment_t * fs, svm_fifo_t * f, } } +svm_msg_q_t * +fifo_segment_msg_q_alloc (fifo_segment_t *fs, u32 mq_index, + svm_msg_q_cfg_t *cfg) +{ + fifo_segment_header_t *fsh = fs->h; + svm_msg_q_shared_t *smq; + svm_msg_q_t *mq; + void *base; + u32 size; + + if (!fs->mqs) + { + u32 n_mqs = clib_max (fs->h->n_mqs, 1); + vec_validate (fs->mqs, n_mqs - 1); + } + + size = svm_msg_q_size_to_alloc (cfg); + base = fsh_alloc_aligned (fsh, size, 8); + fsh->n_reserved_bytes += size; + + smq = svm_msg_q_init (base, cfg); + mq = vec_elt_at_index (fs->mqs, mq_index); + svm_msg_q_attach (mq, smq); + + return mq; +} + +svm_msg_q_t * +fifo_segment_msg_q_attach (fifo_segment_t *fs, uword offset, u32 mq_index) +{ + svm_msg_q_t *mq; + + if (!fs->mqs) + { + u32 n_mqs = clib_max (fs->h->n_mqs, 1); + vec_validate (fs->mqs, n_mqs - 1); + } + + mq = vec_elt_at_index (fs->mqs, mq_index); + + if (!mq->q) + { + svm_msg_q_shared_t *smq; + smq = (svm_msg_q_shared_t *) ((u8 *) fs->h + offset); + svm_msg_q_attach (mq, smq); + } + + ASSERT (fifo_segment_msg_q_offset (fs, mq_index) == offset); + + return mq; +} + +uword +fifo_segment_msg_q_offset (fifo_segment_t *fs, u32 mq_index) +{ + svm_msg_q_t *mq = vec_elt_at_index (fs->mqs, mq_index); + + if (mq->q == 0) + return ~0ULL; + + return (uword) ((u8 *) mq->q - (u8 *) fs->h) - sizeof (svm_msg_q_shared_t); +} + int fifo_segment_prealloc_fifo_hdrs (fifo_segment_t * fs, u32 slice_index, u32 batch_size) diff --git a/src/svm/fifo_segment.h b/src/svm/fifo_segment.h index 006ffc40f63..195869a25dd 100644 --- a/src/svm/fifo_segment.h +++ b/src/svm/fifo_segment.h @@ -17,6 +17,7 @@ #include <svm/ssvm.h> #include <svm/fifo_types.h> +#include <svm/message_queue.h> #include <svm/svm_fifo.h> typedef enum @@ -70,6 +71,7 @@ typedef struct uword max_byte_index; u8 n_slices; /**< number of fifo segment slices */ fifo_slice_private_t *slices; /**< private slice information */ + svm_msg_q_t *mqs; /**< private vec of attached mqs */ } fifo_segment_t; typedef struct @@ -130,6 +132,37 @@ void fifo_segment_attach_fifo (fifo_segment_t * fs, svm_fifo_t * f, u32 slice_index); /** + * Allocate message queue on segment + * + * @param fs fifo segment for mq + * @param mq_index index in private mqs vector to use to attach + * @param cfg configuration for mq + * @return attached message queue + */ +svm_msg_q_t *fifo_segment_msg_q_alloc (fifo_segment_t *fs, u32 mq_index, + svm_msg_q_cfg_t *cfg); + +/** + * Attach message queue at fifo segment offset + * + * @param fs fifo segment for mq + * @param offset offset for shared mq on the segment + * @param mq_index index in private mqs vector to use to attach + * @return attached message queue + */ +svm_msg_q_t *fifo_segment_msg_q_attach (fifo_segment_t *fs, uword offset, + u32 mq_index); + +/** + * Message queue offset on segment + * + * @param fs fifo segment for mq + * @param mq_index index of mq in private mqs vector + * @return offset of the shared mq the private mq is attached to + */ +uword fifo_segment_msg_q_offset (fifo_segment_t *fs, u32 mq_index); + +/** * Try to preallocate fifo headers * * Tries to preallocate fifo headers and adds them to freelist. diff --git a/src/svm/fifo_types.h b/src/svm/fifo_types.h index 85e67bb039c..bfd1a41a17b 100644 --- a/src/svm/fifo_types.h +++ b/src/svm/fifo_types.h @@ -146,6 +146,7 @@ struct fifo_segment_header_ u8 high_watermark; /**< Memory pressure watermark high */ u8 low_watermark; /**< Memory pressure watermark low */ u8 pct_first_alloc; /**< Pct of fifo size to alloc */ + u8 n_mqs; /**< Num mqs for mqs segment */ CLIB_CACHE_LINE_ALIGN_MARK (allocator); uword byte_index; uword max_byte_index; diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c index e586841cdb1..0ebce702430 100644 --- a/src/svm/message_queue.c +++ b/src/svm/message_queue.c @@ -34,45 +34,34 @@ 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); + return (ring->shr->data + elt_index * ring->elsize); } -svm_msg_q_t * +svm_msg_q_shared_t * svm_msg_q_init (void *base, svm_msg_q_cfg_t *cfg) { - svm_msg_q_ring_t *ring; - vec_header_t *vh; - svm_msg_q_t *mq; - u8 *rings_ptr; - u32 q_sz; + svm_msg_q_ring_shared_t *ring; + svm_msg_q_shared_t *smq; + u32 q_sz, offset; int i; q_sz = sizeof (svm_queue_t) + cfg->q_nitems * sizeof (svm_msg_q_msg_t); - mq = (svm_msg_q_t *) base; - mq->q = svm_queue_init (base + sizeof (svm_msg_q_t), cfg->q_nitems, - sizeof (svm_msg_q_msg_t)); - mq->q->consumer_pid = cfg->consumer_pid; - vh = (vec_header_t *) ((u8 *) mq->q + q_sz); - vh->len = cfg->n_rings; - mq->rings = (svm_msg_q_ring_t *) (vh + 1); - rings_ptr = (u8 *) mq->rings + sizeof (svm_msg_q_ring_t) * cfg->n_rings; + smq = (svm_msg_q_shared_t *) base; + svm_queue_init (&smq->q, cfg->q_nitems, sizeof (svm_msg_q_msg_t)); + smq->q->consumer_pid = cfg->consumer_pid; + smq->n_rings = cfg->n_rings; + ring = (void *) ((u8 *) smq->q + q_sz); for (i = 0; i < cfg->n_rings; i++) { - ring = &mq->rings[i]; ring->elsize = cfg->ring_cfgs[i].elsize; ring->nitems = cfg->ring_cfgs[i].nitems; ring->cursize = ring->head = ring->tail = 0; - if (cfg->ring_cfgs[i].data) - ring->data = cfg->ring_cfgs[i].data; - else - { - ring->data = rings_ptr; - rings_ptr += (uword) ring->nitems * ring->elsize; - } + offset = sizeof (*ring) + ring->nitems * ring->elsize; + ring = (void *) ((u8 *) ring + offset); } - return mq; + return smq; } uword @@ -80,12 +69,12 @@ svm_msg_q_size_to_alloc (svm_msg_q_cfg_t *cfg) { svm_msg_q_ring_cfg_t *ring_cfg; uword rings_sz = 0, mq_sz; - u32 vec_sz, q_sz; + u32 q_sz; int i; ASSERT (cfg); - vec_sz = vec_header_bytes (0) + sizeof (svm_msg_q_ring_t) * cfg->n_rings; + rings_sz = sizeof (svm_msg_q_ring_shared_t) * cfg->n_rings; for (i = 0; i < cfg->n_rings; i++) { if (cfg->ring_cfgs[i].data) @@ -95,33 +84,18 @@ svm_msg_q_size_to_alloc (svm_msg_q_cfg_t *cfg) } q_sz = sizeof (svm_queue_t) + cfg->q_nitems * sizeof (svm_msg_q_msg_t); - mq_sz = sizeof (svm_msg_q_t) + vec_sz + rings_sz + q_sz; + mq_sz = sizeof (svm_msg_q_shared_t) + q_sz + rings_sz; return mq_sz; } -svm_msg_q_t * +svm_msg_q_shared_t * svm_msg_q_alloc (svm_msg_q_cfg_t *cfg) { - svm_msg_q_ring_cfg_t *ring_cfg; - uword rings_sz = 0, mq_sz; - u32 vec_sz, q_sz; + uword mq_sz; u8 *base; - int i; - - ASSERT (cfg); - vec_sz = vec_header_bytes (0) + sizeof (svm_msg_q_ring_t) * cfg->n_rings; - for (i = 0; i < cfg->n_rings; i++) - { - if (cfg->ring_cfgs[i].data) - continue; - ring_cfg = &cfg->ring_cfgs[i]; - rings_sz += (uword) ring_cfg->nitems * ring_cfg->elsize; - } - - q_sz = sizeof (svm_queue_t) + cfg->q_nitems * sizeof (svm_msg_q_msg_t); - mq_sz = sizeof (svm_msg_q_t) + vec_sz + rings_sz + q_sz; + mq_sz = svm_msg_q_size_to_alloc (cfg); base = clib_mem_alloc_aligned (mq_sz, CLIB_CACHE_LINE_BYTES); if (!base) return 0; @@ -130,6 +104,29 @@ svm_msg_q_alloc (svm_msg_q_cfg_t *cfg) } void +svm_msg_q_attach (svm_msg_q_t *mq, void *smq_base) +{ + svm_msg_q_ring_shared_t *ring; + svm_msg_q_shared_t *smq; + u32 i, n_rings, q_sz, offset; + + smq = (svm_msg_q_shared_t *) smq_base; + mq->q = smq->q; + n_rings = smq->n_rings; + vec_validate (mq->rings, n_rings - 1); + q_sz = sizeof (svm_queue_t) + mq->q->maxsize * sizeof (svm_msg_q_msg_t); + ring = (void *) ((u8 *) smq->q + q_sz); + for (i = 0; i < n_rings; i++) + { + mq->rings[i].nitems = ring->nitems; + mq->rings[i].elsize = ring->elsize; + mq->rings[i].shr = ring; + offset = sizeof (*ring) + ring->nitems * ring->elsize; + ring = (void *) ((u8 *) ring + offset); + } +} + +void svm_msg_q_free (svm_msg_q_t * mq) { svm_queue_free (mq->q); @@ -139,14 +136,18 @@ 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_ring_shared_t *sr; + svm_msg_q_ring_t *ring; svm_msg_q_msg_t msg; - svm_msg_q_ring_t *ring = svm_msg_q_ring_inline (mq, ring_index); - ASSERT (ring->cursize < ring->nitems); + ring = svm_msg_q_ring_inline (mq, ring_index); + sr = ring->shr; + + ASSERT (sr->cursize < ring->nitems); msg.ring_index = ring - mq->rings; - msg.elt_index = ring->tail; - ring->tail = (ring->tail + 1) % ring->nitems; - clib_atomic_fetch_add (&ring->cursize, 1); + msg.elt_index = sr->tail; + sr->tail = (sr->tail + 1) % ring->nitems; + clib_atomic_fetch_add (&sr->cursize, 1); return msg; } @@ -181,16 +182,18 @@ 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 }; + svm_msg_q_ring_shared_t *sr; svm_msg_q_ring_t *ring; vec_foreach (ring, mq->rings) { - if (ring->elsize < nbytes || ring->cursize == ring->nitems) + sr = ring->shr; + if (ring->elsize < nbytes || sr->cursize == ring->nitems) continue; msg.ring_index = ring - mq->rings; - msg.elt_index = ring->tail; - ring->tail = (ring->tail + 1) % ring->nitems; - clib_atomic_fetch_add (&ring->cursize, 1); + msg.elt_index = sr->tail; + sr->tail = (sr->tail + 1) % ring->nitems; + clib_atomic_fetch_add (&sr->cursize, 1); break; } return msg; @@ -206,14 +209,16 @@ svm_msg_q_msg_data (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) void svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) { + svm_msg_q_ring_shared_t *sr; svm_msg_q_ring_t *ring; int need_signal; ASSERT (vec_len (mq->rings) > msg->ring_index); - ring = &mq->rings[msg->ring_index]; - if (msg->elt_index == ring->head) + ring = svm_msg_q_ring_inline (mq, msg->ring_index); + sr = ring->shr; + if (msg->elt_index == sr->head) { - ring->head = (ring->head + 1) % ring->nitems; + sr->head = (sr->head + 1) % ring->nitems; } else { @@ -222,8 +227,8 @@ svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) ASSERT (0); } - need_signal = ring->cursize == ring->nitems; - clib_atomic_fetch_sub (&ring->cursize, 1); + need_signal = sr->cursize == ring->nitems; + clib_atomic_fetch_sub (&sr->cursize, 1); if (PREDICT_FALSE (need_signal)) svm_queue_send_signal (mq->q, 0); @@ -233,17 +238,20 @@ static int svm_msq_q_msg_is_valid (svm_msg_q_t * mq, svm_msg_q_msg_t * msg) { u32 dist1, dist2, tail, head; + svm_msg_q_ring_shared_t *sr; svm_msg_q_ring_t *ring; if (vec_len (mq->rings) <= msg->ring_index) return 0; - ring = &mq->rings[msg->ring_index]; - tail = ring->tail; - head = ring->head; + + ring = svm_msg_q_ring_inline (mq, msg->ring_index); + sr = ring->shr; + tail = sr->tail; + head = sr->head; dist1 = ((ring->nitems + msg->elt_index) - head) % ring->nitems; if (tail == head) - dist2 = (ring->cursize == 0) ? 0 : ring->nitems; + dist2 = (sr->cursize == 0) ? 0 : ring->nitems; else dist2 = ((ring->nitems + tail) - head) % ring->nitems; return (dist1 < dist2); @@ -316,7 +324,7 @@ format_svm_msg_q (u8 * s, va_list * args) s = format (s, " [Q:%d/%d]", mq->q->cursize, mq->q->maxsize); for (u32 i = 0; i < vec_len (mq->rings); i++) { - s = format (s, " [R%d:%d/%d]", i, mq->rings[i].cursize, + s = format (s, " [R%d:%d/%d]", i, mq->rings[i].shr->cursize, mq->rings[i].nitems); } return s; diff --git a/src/svm/message_queue.h b/src/svm/message_queue.h index 50f79fb0c19..4b314b837e9 100644 --- a/src/svm/message_queue.h +++ b/src/svm/message_queue.h @@ -24,16 +24,30 @@ #include <vppinfra/error.h> #include <svm/queue.h> -typedef struct svm_msg_q_ring_ +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; /**< chunk of memory for msg data */ + 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_queue_t q[0]; /**< queue for exchanging messages */ +} __clib_packed svm_msg_q_shared_t; + typedef struct svm_msg_q_ { svm_queue_t *q; /**< queue for exchanging messages */ @@ -77,10 +91,12 @@ typedef union * ring configs * @return message queue */ -svm_msg_q_t *svm_msg_q_alloc (svm_msg_q_cfg_t * cfg); -svm_msg_q_t *svm_msg_q_init (void *base, svm_msg_q_cfg_t *cfg); +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); + /** * Free message queue * @@ -267,8 +283,8 @@ svm_msg_q_is_full (svm_msg_q_t * mq) 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); + svm_msg_q_ring_t *ring = vec_elt_at_index (mq->rings, ring_index); + return (ring->shr->cursize >= ring->nitems); } /** diff --git a/src/svm/svm_fifo.c b/src/svm/svm_fifo.c index f79f37a4687..14729710aac 100644 --- a/src/svm/svm_fifo.c +++ b/src/svm/svm_fifo.c @@ -398,6 +398,7 @@ svm_fifo_init (svm_fifo_t * f, u32 size) { c->start_byte = prev->start_byte + prev->length; c->enq_rb_index = c->deq_rb_index = RBTREE_TNIL_INDEX; + ASSERT (c->length >= 1 << FS_MIN_LOG2_CHUNK_SZ); prev = c; c = f_cptr (f, c->next); } |