svm: refactor fifo

Type: refactor

Switch from a wrapped byte space to a "continuous" one wherein fifo
chunks are appended to the fifo as more data is enqueued and chunks are
removed as data is dequeued.

The fifo is still subject to a maximum size, i.e., maximum number of
bytes that can be enqueued, so the max number of chunks associated to
the fifo is also constrained.

When enqueueing data, which must fit within the available free space, if
not enough "supporting" chunk  memory is available, the fifo asks the
fifo segment for enough chunk memory to ensure that the write can
succeed. To avoid allocating large amounts of small chunks due to small
writes, if possible, the size of the chunks requested is lower capped by
min_alloc.

When dequeuing data, all the chunks that have been completely drained,
i.e., head moved beyond the chunks’ end bytes, are unlinked from the
fifo and returned to the fifo segment. The one exception to this is the
last chunk which is never unlinked.

Change-Id: I98c1dbd9135fb79650365c7e40c29238b96cd4ee
Signed-off-by: Florin Coras <fcoras@cisco.com>

1 files changed, 95 insertions, 94 deletions

diff --git a/src/svm/fifo_segment.c b/src/svm/fifo_segment.c
index bc46aba7bf6..49cd1161a4b 100644
--- a/src/svm/fifo_segment.c
+++ b/src/svm/fifo_segment.c
@@ -127,6 +127,7 @@ fifo_segment_init (fifo_segment_t * fs)
     {
       fss = fsh_slice_get (fsh, i);
       vec_validate_init_empty (fss->free_chunks, max_chunk_sz, 0);
+      clib_spinlock_init (&fss->chunk_lock);
     }
 
   ssvm_pop_heap (oldheap);
@@ -243,6 +244,8 @@ fifo_segment_main_init (fifo_segment_main_t * sm, u64 baseva,
 static inline u32
 fs_freelist_for_size (u32 size)
 {
+  if (PREDICT_FALSE (size < FIFO_SEGMENT_MIN_FIFO_SIZE))
+    return 0;
   return max_log2 (size) - FIFO_SEGMENT_MIN_LOG2_FIFO_SIZE;
 }
 
@@ -278,9 +281,8 @@ fs_try_alloc_fifo_freelist (fifo_segment_slice_t * fss,
 
   fss->free_fifos = f->next;
   fss->free_chunks[fl_index] = c->next;
-  c->next = c;
+  c->next = 0;
   c->start_byte = 0;
-  c->length = data_bytes;
   memset (f, 0, sizeof (*f));
   f->start_chunk = c;
   f->end_chunk = c;
@@ -331,7 +333,6 @@ fs_try_alloc_fifo_freelist_multi_chunk (fifo_segment_header_t * fsh,
 	  c->next = first;
 	  first = c;
 	  n_alloc += fl_size;
-	  c->length = clib_min (fl_size, data_bytes);
 	  data_bytes -= c->length;
 	}
       else
@@ -370,7 +371,6 @@ fs_try_alloc_fifo_freelist_multi_chunk (fifo_segment_header_t * fsh,
 
   f->start_chunk = first;
   f->end_chunk = last;
-  last->next = first;
   fss->n_fl_chunk_bytes -= n_alloc;
   return f;
 }
@@ -412,7 +412,8 @@ fs_try_alloc_fifo_batch (fifo_segment_header_t * fsh,
       c = (svm_fifo_chunk_t *) (fmem + sizeof (*f));
       c->start_byte = 0;
       c->length = rounded_data_size;
-      c->rb_index = RBTREE_TNIL_INDEX;
+      c->enq_rb_index = RBTREE_TNIL_INDEX;
+      c->deq_rb_index = RBTREE_TNIL_INDEX;
       c->next = fss->free_chunks[fl_index];
       fss->free_chunks[fl_index] = c;
       fmem += hdrs + rounded_data_size;
@@ -466,7 +467,7 @@ fs_try_alloc_fifo (fifo_segment_header_t * fsh, fifo_segment_slice_t * fss,
   if (fifo_sz <= n_free_bytes)
     {
       void *oldheap = ssvm_push_heap (fsh->ssvm_sh);
-      f = svm_fifo_create (data_bytes);
+      f = svm_fifo_alloc (data_bytes);
       ssvm_pop_heap (oldheap);
       if (f)
 	{
@@ -479,9 +480,87 @@ fs_try_alloc_fifo (fifo_segment_header_t * fsh, fifo_segment_slice_t * fss,
 
 done:
 
+  if (f)
+    f->fs_hdr = fsh;
   return f;
 }
 
+svm_fifo_chunk_t *
+fsh_alloc_chunk (fifo_segment_header_t * fsh, u32 slice_index, u32 chunk_size)
+{
+  fifo_segment_slice_t *fss;
+  svm_fifo_chunk_t *c;
+  void *oldheap;
+  int fl_index;
+
+  fl_index = fs_freelist_for_size (chunk_size);
+  fss = fsh_slice_get (fsh, slice_index);
+
+  clib_spinlock_lock (&fss->chunk_lock);
+  c = fss->free_chunks[fl_index];
+
+  if (!c)
+    {
+      fsh_check_mem (fsh);
+      chunk_size = fs_freelist_index_to_size (fl_index);
+      if (fsh_n_free_bytes (fsh) < chunk_size)
+	goto done;
+
+      oldheap = ssvm_push_heap (fsh->ssvm_sh);
+      c = svm_fifo_chunk_alloc (chunk_size);
+      ssvm_pop_heap (oldheap);
+
+      if (!c)
+	goto done;
+
+      fsh_free_bytes_sub (fsh, chunk_size + sizeof (*c));
+    }
+  else
+    {
+      fss->free_chunks[fl_index] = c->next;
+      c->next = 0;
+      fss->n_fl_chunk_bytes -= fs_freelist_index_to_size (fl_index);
+    }
+
+done:
+
+  clib_spinlock_unlock (&fss->chunk_lock);
+
+  return c;
+}
+
+static void
+fsh_slice_collect_chunks (fifo_segment_slice_t * fss, svm_fifo_chunk_t * cur)
+{
+  svm_fifo_chunk_t *next;
+  int fl_index;
+
+  clib_spinlock_lock (&fss->chunk_lock);
+
+  while (cur)
+    {
+      next = cur->next;
+      fl_index = fs_freelist_for_size (cur->length);
+      cur->next = fss->free_chunks[fl_index];
+      cur->enq_rb_index = RBTREE_TNIL_INDEX;
+      cur->deq_rb_index = RBTREE_TNIL_INDEX;
+      fss->free_chunks[fl_index] = cur;
+      fss->n_fl_chunk_bytes += fs_freelist_index_to_size (fl_index);
+      cur = next;
+    }
+
+  clib_spinlock_unlock (&fss->chunk_lock);
+}
+
+void
+fsh_collect_chunks (fifo_segment_header_t * fsh, u32 slice_index,
+		    svm_fifo_chunk_t * cur)
+{
+  fifo_segment_slice_t *fss;
+  fss = fsh_slice_get (fsh, slice_index);
+  fsh_slice_collect_chunks (fss, cur);
+}
+
 /**
  * Allocate fifo in fifo segment
  */
@@ -502,13 +581,8 @@ fifo_segment_alloc_fifo_w_slice (fifo_segment_t * fs, u32 slice_index,
 
   f->slice_index = slice_index;
 
-  /* (re)initialize the fifo, as in svm_fifo_create */
   svm_fifo_init (f, data_bytes);
 
-  /* Initialize chunks and rbtree for multi-chunk fifos */
-  if (f->start_chunk->next != f->start_chunk)
-    svm_fifo_init_chunks (f);
-
   /* If rx fifo type add to active fifos list. When cleaning up segment,
    * we need a list of active sessions that should be disconnected. Since
    * both rx and tx fifos keep pointers to the session, it's enough to track
@@ -522,7 +596,14 @@ fifo_segment_alloc_fifo_w_slice (fifo_segment_t * fs, u32 slice_index,
 	}
       fss->fifos = f;
       f->flags |= SVM_FIFO_F_LL_TRACKED;
+
+      svm_fifo_init_ooo_lookup (f, 0 /* ooo enq */ );
+    }
+  else
+    {
+      svm_fifo_init_ooo_lookup (f, 1 /* ooo deq */ );
     }
+
   fsh_active_fifos_update (fsh, 1);
 
 done:
@@ -536,9 +617,7 @@ void
 fifo_segment_free_fifo (fifo_segment_t * fs, svm_fifo_t * f)
 {
   fifo_segment_header_t *fsh = fs->h;
-  svm_fifo_chunk_t *cur, *next;
   fifo_segment_slice_t *fss;
-  int fl_index;
 
   ASSERT (f->refcnt > 0);
 
@@ -565,26 +644,13 @@ fifo_segment_free_fifo (fifo_segment_t * fs, svm_fifo_t * f)
   fss->free_fifos = f;
 
   /* Free fifo chunks */
-  cur = f->start_chunk;
-  do
-    {
-      next = cur->next;
-      fl_index = fs_freelist_for_size (cur->length);
-      ASSERT (fl_index < vec_len (fss->free_chunks));
-      cur->next = fss->free_chunks[fl_index];
-      cur->rb_index = RBTREE_TNIL_INDEX;
-      fss->free_chunks[fl_index] = cur;
-      fss->n_fl_chunk_bytes += fs_freelist_index_to_size (fl_index);
-      cur = next;
-    }
-  while (cur != f->start_chunk);
+  fsh_slice_collect_chunks (fss, f->start_chunk);
 
-  f->start_chunk = f->end_chunk = f->new_chunks = 0;
+  f->start_chunk = f->end_chunk = 0;
   f->head_chunk = f->tail_chunk = f->ooo_enq = f->ooo_deq = 0;
 
-  svm_fifo_free_chunk_lookup (f);
-
   /* not allocated on segment heap */
+  svm_fifo_free_chunk_lookup (f);
   svm_fifo_free_ooo_data (f);
 
   if (CLIB_DEBUG)
@@ -751,71 +817,6 @@ fifo_segment_preallocate_fifo_pairs (fifo_segment_t * fs,
     }
 }
 
-int
-fifo_segment_grow_fifo (fifo_segment_t * fs, svm_fifo_t * f, u32 chunk_size)
-{
-  fifo_segment_header_t *fsh = fs->h;
-  fifo_segment_slice_t *fss;
-  svm_fifo_chunk_t *c;
-  void *oldheap;
-  int fl_index;
-
-  fl_index = fs_freelist_for_size (chunk_size);
-  fss = fsh_slice_get (fsh, f->slice_index);
-
-  c = fss->free_chunks[fl_index];
-
-  if (!c)
-    {
-      fsh_check_mem (fsh);
-      if (fsh_n_free_bytes (fsh) < chunk_size)
-	return -1;
-
-      oldheap = ssvm_push_heap (fsh->ssvm_sh);
-      c = svm_fifo_chunk_alloc (chunk_size);
-      ssvm_pop_heap (oldheap);
-
-      if (!c)
-	return -1;
-
-      fsh_free_bytes_sub (fsh, chunk_size + sizeof (*c));
-    }
-  else
-    {
-      fss->free_chunks[fl_index] = c->next;
-      c->next = 0;
-      fss->n_fl_chunk_bytes -= fs_freelist_index_to_size (fl_index);
-    }
-
-  svm_fifo_add_chunk (f, c);
-
-  return 0;
-}
-
-int
-fifo_segment_collect_fifo_chunks (fifo_segment_t * fs, svm_fifo_t * f)
-{
-  fifo_segment_header_t *fsh = fs->h;
-  svm_fifo_chunk_t *cur, *next;
-  fifo_segment_slice_t *fss;
-  int fl_index;
-
-  cur = svm_fifo_collect_chunks (f);
-
-  fss = fsh_slice_get (fsh, f->slice_index);
-
-  while (cur)
-    {
-      next = cur->next;
-      fl_index = fs_freelist_for_size (cur->length);
-      cur->next = fss->free_chunks[fl_index];
-      fss->free_chunks[fl_index] = cur;
-      cur = next;
-    }
-
-  return 0;
-}
-
 /**
  * Get number of active fifos
  */