X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fsvm%2Ffifo_segment.c;h=c2ba6c0eb28e04aa6d3db56989068d4a5af39c53;hb=c795b8836d83b4b062aa21423f842aa765269c2c;hp=0c3a79a613b009db039d181e26ba4e2607885552;hpb=b462418890240b2e38dbf522f9dd0196b79e0fa8;p=vpp.git

diff --git a/src/svm/fifo_segment.c b/src/svm/fifo_segment.c
index 0c3a79a613b..c2ba6c0eb28 100644
--- a/src/svm/fifo_segment.c
+++ b/src/svm/fifo_segment.c
@@ -105,13 +105,14 @@ fsh_n_active_fifos (fifo_segment_header_t * fsh)
 }
 
 static inline uword
-fsh_virtual_mem (fifo_segment_header_t * fsh)
+fs_virtual_mem (fifo_segment_t *fs)
 {
+  fifo_segment_header_t *fsh = fs->h;
   fifo_segment_slice_t *fss;
   uword total_vm = 0;
   int i;
 
-  for (i = 0; i < fsh->n_slices; i++)
+  for (i = 0; i < fs->n_slices; i++)
     {
       fss = fsh_slice_get (fsh, i);
       total_vm += clib_atomic_load_relax_n (&fss->virtual_mem);
@@ -127,31 +128,22 @@ fsh_virtual_mem_update (fifo_segment_header_t * fsh, u32 slice_index,
   fss->virtual_mem += n_bytes;
 }
 
-static inline void
-fss_chunk_freelist_lock (fifo_segment_slice_t *fss)
+static inline int
+fss_chunk_fl_index_is_valid (fifo_segment_slice_t *fss, u32 fl_index)
 {
-  u32 free = 0;
-  while (!clib_atomic_cmp_and_swap_acq_relax_n (&fss->chunk_lock, &free, 1, 0))
-    {
-      /* atomic load limits number of compare_exchange executions */
-      while (clib_atomic_load_relax_n (&fss->chunk_lock))
-	CLIB_PAUSE ();
-      /* on failure, compare_exchange writes (*p)->lock into free */
-      free = 0;
-    }
+  return (fl_index < FS_CHUNK_VEC_LEN);
 }
 
-static inline void
-fss_chunk_freelist_unlock (fifo_segment_slice_t *fss)
-{
-  /* Make sure all reads/writes are complete before releasing the lock */
-  clib_atomic_release (&fss->chunk_lock);
-}
+#define FS_CL_HEAD_MASK	 0xFFFFFFFFFFFF
+#define FS_CL_HEAD_TMASK 0xFFFF000000000000
+#define FS_CL_HEAD_TINC	 (1ULL << 48)
 
-static inline int
-fss_chunk_fl_index_is_valid (fifo_segment_slice_t * fss, u32 fl_index)
+static svm_fifo_chunk_t *
+fss_chunk_free_list_head (fifo_segment_header_t *fsh,
+			  fifo_segment_slice_t *fss, u32 fl_index)
 {
-  return (fl_index < FS_CHUNK_VEC_LEN);
+  fs_sptr_t headsp = clib_atomic_load_relax_n (&fss->free_chunks[fl_index]);
+  return fs_chunk_ptr (fsh, headsp & FS_CL_HEAD_MASK);
 }
 
 static void
@@ -159,10 +151,20 @@ fss_chunk_free_list_push (fifo_segment_header_t *fsh,
 			  fifo_segment_slice_t *fss, u32 fl_index,
 			  svm_fifo_chunk_t *c)
 {
-  fss_chunk_freelist_lock (fss);
-  c->next = fss->free_chunks[fl_index];
-  fss->free_chunks[fl_index] = fs_chunk_sptr (fsh, c);
-  fss_chunk_freelist_unlock (fss);
+  fs_sptr_t old_head, new_head, csp;
+
+  csp = fs_chunk_sptr (fsh, c);
+  ASSERT (csp <= FS_CL_HEAD_MASK);
+  old_head = clib_atomic_load_acq_n (&fss->free_chunks[fl_index]);
+
+  do
+    {
+      c->next = old_head & FS_CL_HEAD_MASK;
+      new_head = csp + ((old_head + FS_CL_HEAD_TINC) & FS_CL_HEAD_TMASK);
+    }
+  while (!__atomic_compare_exchange (&fss->free_chunks[fl_index], &old_head,
+				     &new_head, 0 /* weak */, __ATOMIC_RELEASE,
+				     __ATOMIC_ACQUIRE));
 }
 
 static void
@@ -170,34 +172,78 @@ fss_chunk_free_list_push_list (fifo_segment_header_t *fsh,
 			       fifo_segment_slice_t *fss, u32 fl_index,
 			       svm_fifo_chunk_t *head, svm_fifo_chunk_t *tail)
 {
-  fss_chunk_freelist_lock (fss);
-  tail->next = fss->free_chunks[fl_index];
-  fss->free_chunks[fl_index] = fs_chunk_sptr (fsh, head);
-  fss_chunk_freelist_unlock (fss);
+  fs_sptr_t old_head, new_head, headsp;
+
+  headsp = fs_chunk_sptr (fsh, head);
+  ASSERT (headsp <= FS_CL_HEAD_MASK);
+  old_head = clib_atomic_load_acq_n (&fss->free_chunks[fl_index]);
+
+  do
+    {
+      tail->next = old_head & FS_CL_HEAD_MASK;
+      new_head = headsp + ((old_head + FS_CL_HEAD_TINC) & FS_CL_HEAD_TMASK);
+    }
+  while (!__atomic_compare_exchange (&fss->free_chunks[fl_index], &old_head,
+				     &new_head, 0 /* weak */, __ATOMIC_RELEASE,
+				     __ATOMIC_ACQUIRE));
 }
 
 static svm_fifo_chunk_t *
 fss_chunk_free_list_pop (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss,
 			 u32 fl_index)
 {
+  fs_sptr_t old_head, new_head;
   svm_fifo_chunk_t *c;
 
   ASSERT (fss_chunk_fl_index_is_valid (fss, fl_index));
 
-  fss_chunk_freelist_lock (fss);
+  old_head = clib_atomic_load_acq_n (&fss->free_chunks[fl_index]);
 
-  if (!fss->free_chunks[fl_index])
+  /* Lock-free stacks are affected by ABA if a side allocates a chunk and
+   * shortly thereafter frees it. To circumvent that, reuse the upper bits
+   * of the head of the list shared pointer, i.e., offset to where the chunk
+   * is, as a tag. The tag is incremented with each push/pop operation and
+   * therefore collisions can only happen if an element is popped and pushed
+   * exactly after a complete wrap of the tag (16 bits). It's unlikely either
+   * of the sides will be descheduled for that long */
+  do
     {
-      fss_chunk_freelist_unlock (fss);
-      return 0;
+      if (!(old_head & FS_CL_HEAD_MASK))
+	return 0;
+      c = fs_chunk_ptr (fsh, old_head & FS_CL_HEAD_MASK);
+      new_head = c->next + ((old_head + FS_CL_HEAD_TINC) & FS_CL_HEAD_TMASK);
     }
+  while (!__atomic_compare_exchange (&fss->free_chunks[fl_index], &old_head,
+				     &new_head, 0 /* weak */, __ATOMIC_RELEASE,
+				     __ATOMIC_ACQUIRE));
 
-  c = fs_chunk_ptr (fsh, fss->free_chunks[fl_index]);
-  fss->free_chunks[fl_index] = c->next;
+  return c;
+}
 
-  fss_chunk_freelist_unlock (fss);
+static void
+fss_fifo_free_list_push (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss,
+			 svm_fifo_shared_t *sf)
+{
+  sf->next = fss->free_fifos;
+  fss->free_fifos = fs_sptr (fsh, sf);
+}
 
-  return c;
+static void
+fss_fifo_free_list_push_list (fifo_segment_header_t *fsh,
+			      fifo_segment_slice_t *fss,
+			      svm_fifo_shared_t *head, svm_fifo_shared_t *tail)
+{
+  tail->next = fss->free_fifos;
+  fss->free_fifos = fs_sptr (fsh, head);
+}
+
+svm_fifo_shared_t *
+fss_fifo_free_list_pop (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss)
+{
+  svm_fifo_shared_t *sf;
+  sf = fs_ptr (fsh, fss->free_fifos);
+  fss->free_fifos = sf->next;
+  return sf;
 }
 
 static inline void
@@ -249,7 +295,7 @@ fss_fl_chunk_bytes_sub (fifo_segment_slice_t * fss, uword size)
 int
 fifo_segment_init (fifo_segment_t * fs)
 {
-  u32 align = 8, offset = 2 * 4096, slices_sz, i;
+  u32 align = 8, offset = FIFO_SEGMENT_ALLOC_OVERHEAD, slices_sz, i;
   uword max_fifo, seg_start, seg_sz;
   fifo_segment_header_t *fsh;
   ssvm_shared_header_t *sh;
@@ -266,6 +312,7 @@ fifo_segment_init (fifo_segment_t * fs)
 
   seg_start = round_pow2_u64 (pointer_to_uword (seg_data), align);
   fsh = uword_to_pointer (seg_start, void *);
+  clib_mem_unpoison (fsh, seg_sz);
   memset (fsh, 0, sizeof (*fsh) + slices_sz);
 
   fsh->byte_index = sizeof (*fsh) + slices_sz;
@@ -275,10 +322,15 @@ fifo_segment_init (fifo_segment_t * fs)
   fsh->max_log2_fifo_size = min_log2 (max_fifo);
   fsh->n_cached_bytes = 0;
   fsh->n_reserved_bytes = fsh->byte_index;
+  fsh->start_byte_index = fsh->byte_index;
   ASSERT (fsh->max_byte_index <= sh->ssvm_size - offset);
 
   fs->max_byte_index = fsh->max_byte_index;
-  fs->h = sh->opaque[0] = fsh;
+  fs->h = fsh;
+  sh->opaque[0] = (void *) ((u8 *) fsh - (u8 *) fs->ssvm.sh);
+
+  /* Allow random offsets */
+  fs->ssvm.sh->ssvm_va = 0;
 
   vec_validate (fs->slices, fs->n_slices - 1);
   for (i = 0; i < fs->n_slices; i++)
@@ -335,10 +387,12 @@ fifo_segment_attach (fifo_segment_main_t * sm, fifo_segment_create_args_t * a)
 
   pool_get_zero (sm->segments, fs);
 
+  fs->fs_index = fs - sm->segments;
+  fs->sm_index = ~0;
   fs->ssvm.ssvm_size = a->segment_size;
   fs->ssvm.my_pid = getpid ();
   fs->ssvm.name = format (0, "%s%c", a->segment_name, 0);
-  fs->ssvm.requested_va = sm->next_baseva;
+  fs->ssvm.requested_va = 0;
   if (a->segment_type == SSVM_SEGMENT_MEMFD)
     fs->ssvm.fd = a->memfd_fd;
   else
@@ -346,17 +400,15 @@ fifo_segment_attach (fifo_segment_main_t * sm, fifo_segment_create_args_t * a)
 
   if ((rv = ssvm_client_init (&fs->ssvm, a->segment_type)))
     {
-      _vec_len (fs) = vec_len (fs) - 1;
+      pool_put (sm->segments, fs);
       return (rv);
     }
 
-  /* Fish the segment header */
-  fsh = fs->h = fs->ssvm.sh->opaque[0];
-
   /* Probably a segment without fifos */
-  if (!fsh)
+  if (!fs->ssvm.sh->opaque[0])
     goto done;
 
+  fsh = fs->h = (void *) fs->ssvm.sh + (uword) fs->ssvm.sh->opaque[0];
   fs->max_byte_index = fsh->max_byte_index;
   vec_validate (fs->slices, 0);
   fs->slices[0].fifos =
@@ -388,6 +440,14 @@ fifo_segment_get_segment (fifo_segment_main_t * sm, u32 segment_index)
   return pool_elt_at_index (sm->segments, segment_index);
 }
 
+fifo_segment_t *
+fifo_segment_get_segment_if_valid (fifo_segment_main_t *sm, u32 segment_index)
+{
+  if (pool_is_free_index (sm->segments, segment_index))
+    return 0;
+  return pool_elt_at_index (sm->segments, segment_index);
+}
+
 void
 fifo_segment_info (fifo_segment_t * seg, char **address, size_t * size)
 {
@@ -497,7 +557,7 @@ static int
 fsh_try_alloc_fifo_hdr_batch (fifo_segment_header_t * fsh,
 			      fifo_segment_slice_t * fss, u32 batch_size)
 {
-  svm_fifo_shared_t *f;
+  svm_fifo_shared_t *f, *head = 0, *tail;
   uword size;
   u8 *fmem;
   int i;
@@ -511,15 +571,18 @@ fsh_try_alloc_fifo_hdr_batch (fifo_segment_header_t * fsh,
     return -1;
 
   /* Carve fifo hdr space */
+  tail = f = (svm_fifo_shared_t *) fmem;
   for (i = 0; i < batch_size; i++)
     {
-      f = (svm_fifo_shared_t *) fmem;
-      memset (f, 0, sizeof (*f));
-      f->next = fss->free_fifos;
-      fss->free_fifos = f;
+      clib_memset (f, 0, sizeof (*f));
+      f->next = fs_sptr (fsh, head);
+      head = f;
       fmem += sizeof (*f);
+      f = (svm_fifo_shared_t *) fmem;
     }
 
+  fss_fifo_free_list_push_list (fsh, fss, head, tail);
+
   return 0;
 }
 
@@ -577,7 +640,7 @@ fs_try_alloc_fifo_batch (fifo_segment_header_t * fsh,
 static svm_fifo_shared_t *
 fsh_try_alloc_fifo_hdr (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss)
 {
-  svm_fifo_shared_t *f;
+  svm_fifo_shared_t *sf;
 
   if (!fss->free_fifos)
     {
@@ -586,10 +649,10 @@ fsh_try_alloc_fifo_hdr (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss)
 	return 0;
     }
 
-  f = fss->free_fifos;
-  fss->free_fifos = f->next;
-  memset (f, 0, sizeof (*f));
-  return f;
+  sf = fss_fifo_free_list_pop (fsh, fss);
+  clib_memset (sf, 0, sizeof (*sf));
+
+  return sf;
 }
 
 static svm_fifo_chunk_t *
@@ -637,7 +700,8 @@ free_list:
       if (data_bytes <= fss_fl_chunk_bytes (fss) + n_free)
 	{
 	  u32 min_size = FIFO_SEGMENT_MIN_FIFO_SIZE;
-
+	  if (n_free < min_size)
+	    goto done;
 	  batch = (data_bytes - fss_fl_chunk_bytes (fss)) / min_size;
 	  batch = clib_min (batch + 1, n_free / min_size);
 	  if (fsh_try_alloc_chunk_batch (fsh, fss, 0, batch))
@@ -682,8 +746,7 @@ fs_try_alloc_fifo (fifo_segment_header_t *fsh, u32 slice_index, u32 data_bytes)
   c = fsh_try_alloc_chunk (fsh, fss, min_size);
   if (!c)
     {
-      sf->next = fss->free_fifos;
-      fss->free_fifos = sf;
+      fss_fifo_free_list_push (fsh, fss, sf);
       return 0;
     }
 
@@ -718,7 +781,7 @@ fsh_slice_collect_chunks (fifo_segment_header_t * fsh,
 
   while (c)
     {
-      CLIB_MEM_UNPOISON (c, sizeof (*c));
+      clib_mem_unpoison (c, sizeof (*c));
       next = fs_chunk_ptr (fsh, c->next);
       fl_index = fs_freelist_for_size (c->length);
       fss_chunk_free_list_push (fsh, fss, fl_index, c);
@@ -751,11 +814,13 @@ fs_fifo_alloc (fifo_segment_t *fs, u32 slice_index)
 }
 
 void
-fs_fifo_free (fifo_segment_t *fs, svm_fifo_t *f)
+fs_fifo_free (fifo_segment_t *fs, svm_fifo_t *f, u32 slice_index)
 {
-  u32 slice_index = f->shr->slice_index;
   fifo_slice_private_t *pfss;
 
+  if (CLIB_DEBUG)
+    clib_memset (f, 0xfc, sizeof (*f));
+
   pfss = &fs->slices[slice_index];
   clib_mem_bulk_free (pfss->fifos, f);
 }
@@ -769,8 +834,10 @@ fifo_segment_cleanup (fifo_segment_t *fs)
   for (slice_index = 0; slice_index < fs->n_slices; slice_index++)
     clib_mem_bulk_destroy (fs->slices[slice_index].fifos);
 
+  vec_free (fs->slices);
+
   vec_foreach (fs->mqs, mq)
-    vec_free (mq->rings);
+    svm_msg_q_cleanup (mq);
 
   vec_free (fs->mqs);
 }
@@ -803,6 +870,9 @@ fifo_segment_alloc_fifo_w_slice (fifo_segment_t * fs, u32 slice_index,
 
   svm_fifo_init (f, data_bytes);
 
+  f->segment_manager = fs->sm_index;
+  f->segment_index = fs->fs_index;
+
   fss = fsh_slice_get (fsh, slice_index);
   pfss = fs_slice_private_get (fs, slice_index);
 
@@ -824,9 +894,12 @@ done:
 }
 
 svm_fifo_t *
-fifo_segment_alloc_fifo_w_shared (fifo_segment_t *fs, svm_fifo_shared_t *sf)
+fifo_segment_alloc_fifo_w_offset (fifo_segment_t *fs, uword offset)
 {
   svm_fifo_t *f = fs_fifo_alloc (fs, 0);
+  svm_fifo_shared_t *sf;
+
+  sf = (svm_fifo_shared_t *) ((u8 *) fs->h + offset);
   f->fs_hdr = fs->h;
   f->shr = sf;
 
@@ -836,6 +909,14 @@ fifo_segment_alloc_fifo_w_shared (fifo_segment_t *fs, svm_fifo_shared_t *sf)
   return f;
 }
 
+svm_fifo_t *
+fifo_segment_duplicate_fifo (fifo_segment_t *fs, svm_fifo_t *f)
+{
+  svm_fifo_t *nf = fs_fifo_alloc (fs, 0);
+  clib_memcpy (nf, f, sizeof (*f));
+  return nf;
+}
+
 /**
  * Free fifo allocated in fifo segment
  */
@@ -867,8 +948,7 @@ fifo_segment_free_fifo (fifo_segment_t * fs, svm_fifo_t * f)
   sf->head_chunk = sf->tail_chunk = 0;
 
   /* Add to free list */
-  sf->next = fss->free_fifos;
-  fss->free_fifos = sf;
+  fss_fifo_free_list_push (fsh, fss, sf);
 
   fss->virtual_mem -= svm_fifo_size (f);
 
@@ -895,59 +975,109 @@ fifo_segment_free_fifo (fifo_segment_t * fs, svm_fifo_t * f)
   f->ooo_enq = f->ooo_deq = 0;
   f->prev = 0;
 
-  fs_fifo_free (fs, f);
+  fs_fifo_free (fs, f, f->shr->slice_index);
 
   fsh_active_fifos_update (fsh, -1);
 }
 
 void
-fifo_segment_detach_fifo (fifo_segment_t * fs, svm_fifo_t * f)
+fifo_segment_free_client_fifo (fifo_segment_t *fs, svm_fifo_t *f)
+{
+  fs_fifo_free (fs, f, 0 /* clients attach fifos in slice 0 */);
+}
+
+void
+fifo_segment_detach_fifo (fifo_segment_t *fs, svm_fifo_t **f)
 {
   fifo_slice_private_t *pfss;
   fifo_segment_slice_t *fss;
-  svm_fifo_chunk_t *c;
-  u32 fl_index;
+  svm_fifo_t *of = *f;
+  u32 slice_index;
 
-  ASSERT (f->refcnt == 1);
+  slice_index = of->master_thread_index;
+  fss = fsh_slice_get (fs->h, slice_index);
+  pfss = fs_slice_private_get (fs, slice_index);
+  fss->virtual_mem -= svm_fifo_size (of);
+  if (of->flags & SVM_FIFO_F_LL_TRACKED)
+    pfss_fifo_del_active_list (pfss, of);
 
-  fss = fsh_slice_get (fs->h, f->shr->slice_index);
-  pfss = fs_slice_private_get (fs, f->shr->slice_index);
-  fss->virtual_mem -= svm_fifo_size (f);
-  if (f->flags & SVM_FIFO_F_LL_TRACKED)
-    pfss_fifo_del_active_list (pfss, f);
+  /* Collect chunks that were provided in return for those detached */
+  fsh_slice_collect_chunks (fs->h, fss, of->chunks_at_attach);
+  of->chunks_at_attach = 0;
 
-  c = fs_chunk_ptr (fs->h, f->shr->start_chunk);
-  while (c)
-    {
-      fl_index = fs_freelist_for_size (c->length);
-      clib_atomic_fetch_sub_rel (&fss->num_chunks[fl_index], 1);
-      c = fs_chunk_ptr (fs->h, c->next);
-    }
+  /* Collect hdr that was provided in return for the detached */
+  fss_fifo_free_list_push (fs->h, fss, of->hdr_at_attach);
+  of->hdr_at_attach = 0;
+
+  clib_mem_bulk_free (pfss->fifos, *f);
+  *f = 0;
 }
 
 void
-fifo_segment_attach_fifo (fifo_segment_t * fs, svm_fifo_t * f,
-			  u32 slice_index)
+fifo_segment_attach_fifo (fifo_segment_t *fs, svm_fifo_t **f, u32 slice_index)
 {
+  svm_fifo_chunk_t *c, *nc, *pc = 0;
   fifo_slice_private_t *pfss;
   fifo_segment_slice_t *fss;
-  svm_fifo_chunk_t *c;
-  u32 fl_index;
+  svm_fifo_t *nf, *of;
 
-  f->shr->slice_index = slice_index;
-  fss = fsh_slice_get (fs->h, f->shr->slice_index);
-  pfss = fs_slice_private_get (fs, f->shr->slice_index);
-  fss->virtual_mem += svm_fifo_size (f);
-  if (f->flags & SVM_FIFO_F_LL_TRACKED)
-    pfss_fifo_add_active_list (pfss, f);
+  nf = fs_fifo_alloc (fs, slice_index);
+  clib_memcpy_fast (nf, *f, sizeof (*nf));
 
-  c = fs_chunk_ptr (fs->h, f->shr->start_chunk);
-  while (c)
+  fss = fsh_slice_get (fs->h, slice_index);
+  pfss = fs_slice_private_get (fs, slice_index);
+  fss->virtual_mem += svm_fifo_size (nf);
+  nf->next = nf->prev = 0;
+  if (nf->flags & SVM_FIFO_F_LL_TRACKED)
+    pfss_fifo_add_active_list (pfss, nf);
+
+  /* Allocate shared hdr and chunks to be collected at detach in return
+   * for those that are being attached now */
+  of = *f;
+  of->hdr_at_attach = fsh_try_alloc_fifo_hdr (fs->h, fss);
+
+  c = fs_chunk_ptr (fs->h, nf->shr->start_chunk);
+  of->chunks_at_attach = pc = fsh_try_alloc_chunk (fs->h, fss, c->length);
+
+  while ((c = fs_chunk_ptr (fs->h, c->next)))
     {
-      fl_index = fs_freelist_for_size (c->length);
-      clib_atomic_fetch_add_rel (&fss->num_chunks[fl_index], 1);
-      c = fs_chunk_ptr (fs->h, c->next);
+      nc = fsh_try_alloc_chunk (fs->h, fss, c->length);
+      pc->next = fs_chunk_sptr (fs->h, nc);
+      pc = nc;
     }
+
+  nf->shr->slice_index = slice_index;
+  *f = nf;
+}
+
+uword
+fifo_segment_fifo_offset (svm_fifo_t *f)
+{
+  return (u8 *) f->shr - (u8 *) f->fs_hdr;
+}
+
+svm_fifo_chunk_t *
+fifo_segment_alloc_chunk_w_slice (fifo_segment_t *fs, u32 slice_index,
+				  u32 chunk_size)
+{
+  fifo_segment_header_t *fsh = fs->h;
+  fifo_segment_slice_t *fss;
+
+  fss = fsh_slice_get (fsh, slice_index);
+  return fsh_try_alloc_chunk (fsh, fss, chunk_size);
+}
+
+void
+fifo_segment_collect_chunk (fifo_segment_t *fs, u32 slice_index,
+			    svm_fifo_chunk_t *c)
+{
+  fsh_collect_chunks (fs->h, slice_index, c);
+}
+
+uword
+fifo_segment_chunk_offset (fifo_segment_t *fs, svm_fifo_chunk_t *c)
+{
+  return (u8 *) c - (u8 *) fs->h;
 }
 
 svm_msg_q_t *
@@ -968,6 +1098,9 @@ fifo_segment_msg_q_alloc (fifo_segment_t *fs, u32 mq_index,
 
   size = svm_msg_q_size_to_alloc (cfg);
   base = fsh_alloc_aligned (fsh, size, 8);
+  if (!base)
+    return 0;
+
   fsh->n_reserved_bytes += size;
 
   smq = svm_msg_q_init (base, cfg);
@@ -990,7 +1123,7 @@ fifo_segment_msg_q_attach (fifo_segment_t *fs, uword offset, u32 mq_index)
 
   mq = vec_elt_at_index (fs->mqs, mq_index);
 
-  if (!mq->q)
+  if (!mq->q.shr)
     {
       svm_msg_q_shared_t *smq;
       smq = (svm_msg_q_shared_t *) ((u8 *) fs->h + offset);
@@ -1002,15 +1135,48 @@ fifo_segment_msg_q_attach (fifo_segment_t *fs, uword offset, u32 mq_index)
   return mq;
 }
 
+void
+fifo_segment_msg_qs_discover (fifo_segment_t *fs, int *fds, u32 n_fds)
+{
+  svm_msg_q_shared_t *smq;
+  u32 n_mqs, size, i;
+  uword offset = 0, n_alloced;
+  svm_msg_q_t *mq;
+
+  n_mqs = fs->h->n_mqs;
+  if (n_fds && n_mqs != n_fds)
+    {
+      clib_warning ("expected %u fds got %u", n_mqs, n_fds);
+      return;
+    }
+
+  vec_validate (fs->mqs, n_mqs - 1);
+  n_alloced = fs->h->n_reserved_bytes - fs->h->start_byte_index;
+  ASSERT (n_alloced % n_mqs == 0);
+  size = n_alloced / n_mqs;
+
+  offset = fs->h->start_byte_index;
+  for (i = 0; i < n_mqs; i++)
+    {
+      mq = vec_elt_at_index (fs->mqs, i);
+      smq = (svm_msg_q_shared_t *) ((u8 *) fs->h + offset);
+      svm_msg_q_attach (mq, smq);
+      if (n_fds)
+	svm_msg_q_set_eventfd (mq, fds[i]);
+      offset += size;
+    }
+}
+
 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)
+  if (mq->q.shr == 0)
     return ~0ULL;
 
-  return (uword) ((u8 *) mq->q - (u8 *) fs->h) - sizeof (svm_msg_q_shared_t);
+  return (uword) ((u8 *) mq->q.shr - (u8 *) fs->h) -
+	 sizeof (svm_msg_q_shared_t);
 }
 
 int
@@ -1120,18 +1286,18 @@ fifo_segment_num_fifos (fifo_segment_t * fs)
 }
 
 static u32
-fs_slice_num_free_fifos (fifo_segment_slice_t * fss)
+fs_slice_num_free_fifos (fifo_segment_header_t *fsh, fifo_segment_slice_t *fss)
 {
   svm_fifo_shared_t *f;
   u32 count = 0;
 
-  f = fss->free_fifos;
+  f = fs_ptr (fsh, fss->free_fifos);
   if (f == 0)
     return 0;
 
   while (f)
     {
-      f = f->next;
+      f = fs_ptr (fsh, f->next);
       count++;
     }
   return count;
@@ -1148,7 +1314,7 @@ fifo_segment_num_free_fifos (fifo_segment_t * fs)
   for (slice_index = 0; slice_index < fs->n_slices; slice_index++)
     {
       fss = fsh_slice_get (fsh, slice_index);
-      count += fs_slice_num_free_fifos (fss);
+      count += fs_slice_num_free_fifos (fsh, fss);
     }
   return count;
 }
@@ -1166,7 +1332,7 @@ fs_slice_num_free_chunks (fifo_segment_header_t *fsh,
     {
       for (i = 0; i < FS_CHUNK_VEC_LEN; i++)
 	{
-	  c = fs_chunk_ptr (fsh, fss->free_chunks[i]);
+	  c = fss_chunk_free_list_head (fsh, fss, i);
 	  if (c == 0)
 	    continue;
 
@@ -1185,7 +1351,7 @@ fs_slice_num_free_chunks (fifo_segment_header_t *fsh,
   if (fl_index >= FS_CHUNK_VEC_LEN)
     return 0;
 
-  c = fs_chunk_ptr (fsh, fss->free_chunks[fl_index]);
+  c = fss_chunk_free_list_head (fsh, fss, fl_index);
   if (c == 0)
     return 0;
 
@@ -1219,16 +1385,16 @@ fifo_segment_size (fifo_segment_t * fs)
   return fs->h->max_byte_index - fs->h->n_reserved_bytes;
 }
 
-u8
-fsh_has_reached_mem_limit (fifo_segment_header_t * fsh)
+static u8
+fs_has_reached_mem_limit (fifo_segment_t *fs)
 {
-  return (fsh->flags & FIFO_SEGMENT_F_MEM_LIMIT) ? 1 : 0;
+  return (fs->flags & FIFO_SEGMENT_F_MEM_LIMIT) ? 1 : 0;
 }
 
-void
-fsh_reset_mem_limit (fifo_segment_header_t * fsh)
+static void
+fs_reset_mem_limit (fifo_segment_t *fs)
 {
-  fsh->flags &= ~FIFO_SEGMENT_F_MEM_LIMIT;
+  fs->flags &= ~FIFO_SEGMENT_F_MEM_LIMIT;
 }
 
 void *
@@ -1303,26 +1469,26 @@ fifo_segment_get_mem_usage (fifo_segment_t * fs)
 }
 
 fifo_segment_mem_status_t
-fifo_segment_determine_status (fifo_segment_header_t * fsh, u8 usage)
+fifo_segment_determine_status (fifo_segment_t *fs, u8 usage)
 {
-  if (!fsh->high_watermark || !fsh->low_watermark)
+  if (!fs->high_watermark || !fs->low_watermark)
     return MEMORY_PRESSURE_NO_PRESSURE;
 
   /* once the no-memory is detected, the status continues
    * until memory usage gets below the high watermark
    */
-  if (fsh_has_reached_mem_limit (fsh))
+  if (fs_has_reached_mem_limit (fs))
     {
-      if (usage >= fsh->high_watermark)
+      if (usage >= fs->high_watermark)
 	return MEMORY_PRESSURE_NO_MEMORY;
       else
-	fsh_reset_mem_limit (fsh);
+	fs_reset_mem_limit (fs);
     }
 
-  if (usage >= fsh->high_watermark)
+  if (usage >= fs->high_watermark)
     return MEMORY_PRESSURE_HIGH_PRESSURE;
 
-  else if (usage >= fsh->low_watermark)
+  else if (usage >= fs->low_watermark)
     return MEMORY_PRESSURE_LOW_PRESSURE;
 
   return MEMORY_PRESSURE_NO_PRESSURE;
@@ -1331,10 +1497,9 @@ fifo_segment_determine_status (fifo_segment_header_t * fsh, u8 usage)
 fifo_segment_mem_status_t
 fifo_segment_get_mem_status (fifo_segment_t * fs)
 {
-  fifo_segment_header_t *fsh = fs->h;
   u8 usage = fifo_segment_get_mem_usage (fs);
 
-  return fifo_segment_determine_status (fsh, usage);
+  return fifo_segment_determine_status (fs, usage);
 }
 
 u8 *
@@ -1379,22 +1544,15 @@ format_fifo_segment (u8 * s, va_list * args)
   f64 usage;
   fifo_segment_mem_status_t mem_st;
 
-  indent = format_get_indent (s) + 2;
-
-  if (fs == 0)
-    {
-      s = format (s, "%-20s%10s%15s%15s%15s%15s", "Name", "Type",
-		  "HeapSize (M)", "ActiveFifos", "FreeFifos", "Address");
-      return s;
-    }
+  indent = format_get_indent (s);
 
   fifo_segment_info (fs, &address, &size);
   active_fifos = fifo_segment_num_fifos (fs);
   free_fifos = fifo_segment_num_free_fifos (fs);
 
-  s = format (s, "%-20v%10U%15llu%15u%15u%15llx", ssvm_name (&fs->ssvm),
-	      format_fifo_segment_type, fs, size >> 20ULL, active_fifos,
-	      free_fifos, address);
+  s = format (s, "%U%v type: %U size: %U active fifos: %u", format_white_space,
+	      2, ssvm_name (&fs->ssvm), format_fifo_segment_type, fs,
+	      format_memory_size, size, active_fifos);
 
   if (!verbose)
     return s;
@@ -1403,9 +1561,8 @@ format_fifo_segment (u8 * s, va_list * args)
 
   free_chunks = fifo_segment_num_free_chunks (fs, ~0);
   if (free_chunks)
-    s =
-      format (s, "\n\n%UFree/Allocated chunks by size:\n", format_white_space,
-	      indent + 2);
+    s = format (s, "\n\n%UFree/Allocated chunks by size:\n",
+		format_white_space, indent + 2);
   else
     s = format (s, "\n");
 
@@ -1414,7 +1571,7 @@ format_fifo_segment (u8 * s, va_list * args)
       fss = fsh_slice_get (fsh, slice_index);
       for (i = 0; i < FS_CHUNK_VEC_LEN; i++)
 	{
-	  c = fs_chunk_ptr (fsh, fss->free_chunks[i]);
+	  c = fss_chunk_free_list_head (fsh, fss, i);
 	  if (c == 0 && fss->num_chunks[i] == 0)
 	    continue;
 	  count = 0;
@@ -1441,7 +1598,7 @@ format_fifo_segment (u8 * s, va_list * args)
   in_use = fifo_segment_size (fs) - est_free_seg_bytes - tracked_cached_bytes;
   usage = (100.0 * in_use) / allocated;
   mem_st = fifo_segment_get_mem_status (fs);
-  virt = fsh_virtual_mem (fsh);
+  virt = fs_virtual_mem (fs);
   reserved = fsh->n_reserved_bytes;
 
   s = format (s, "\n%Useg free bytes: %U (%lu) estimated: %U (%lu) reserved:"
@@ -1454,8 +1611,8 @@ format_fifo_segment (u8 * s, va_list * args)
 	      format_memory_size, chunk_bytes, chunk_bytes,
 	      format_memory_size, est_chunk_bytes, est_chunk_bytes,
 	      format_memory_size, tracked_cached_bytes, tracked_cached_bytes);
-  s = format (s, "%Ufifo active: %u hdr free bytes: %U (%u) \n",
-	      format_white_space, indent + 2, fsh->n_active_fifos,
+  s = format (s, "%Ufifo active: %u hdr free: %u bytes: %U (%u) \n",
+	      format_white_space, indent + 2, fsh->n_active_fifos, free_fifos,
 	      format_memory_size, fifo_hdr, fifo_hdr);
   s = format (s, "%Usegment usage: %.2f%% (%U / %U) virt: %U status: %s\n",
 	      format_white_space, indent + 2, usage, format_memory_size,