session/svm: apis for fifo shrinking

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

src/plugins/unittest/svm_fifo_test.c

@@ -2020,6 +2020,74 @@ sfifo_test_fifo_segment_fifo_grow (int verbose)
   return 0;
 }
 
+static int
+sfifo_test_fifo_segment_fifo_shrink (int verbose)
+{
+  fifo_segment_main_t *sm = &segment_main;
+  fifo_segment_create_args_t _a, *a = &_a;
+  int i, rv, chunk_size = 4096, n_chunks;
+  fifo_segment_t *fs;
+  svm_fifo_t *f;
+
+  clib_memset (a, 0, sizeof (*a));
+  a->segment_name = "fifo-test1";
+  a->segment_size = 256 << 10;
+
+  rv = fifo_segment_create (sm, a);
+
+  SFIFO_TEST (!rv, "svm_fifo_segment_create returned %d", rv);
+
+  /*
+   * Alloc and grow fifo
+   */
+  fs = fifo_segment_get_segment (sm, a->new_segment_indices[0]);
+  f = fifo_segment_alloc_fifo (fs, chunk_size, FIFO_SEGMENT_RX_FIFO);
+
+  SFIFO_TEST (f != 0, "svm_fifo_segment_alloc_fifo");
+
+  for (i = 0; i < 9; i++)
+    {
+      fifo_segment_grow_fifo (fs, f, chunk_size);
+      if (f->size != (i + 2) * chunk_size)
+	SFIFO_TEST (0, "fifo size should be %u is %u",
+		    (i + 2) * chunk_size, f->size);
+    }
+
+  rv = svm_fifo_reduce_size (f, 3.5 * chunk_size, 1 /* is producer */ );
+  SFIFO_TEST (rv == 3 * chunk_size, "len expected %u is %u", 3 * chunk_size,
+	      rv);
+
+  n_chunks = fifo_segment_num_free_chunks (fs, chunk_size);
+  SFIFO_TEST (n_chunks == 0, "free chunks should be %u is %u", 0, n_chunks);
+
+  fifo_segment_collect_fifo_chunks (fs, f);
+
+  n_chunks = fifo_segment_num_free_chunks (fs, chunk_size);
+  SFIFO_TEST (n_chunks == 3, "free chunks should be %u is %u", 3, n_chunks);
+
+  rv = svm_fifo_reduce_size (f, 7 * chunk_size - 1, 1 /* is producer */ );
+  SFIFO_TEST (rv == 6 * chunk_size, "len expected %u is %u", 6 * chunk_size,
+	      rv);
+
+  fifo_segment_collect_fifo_chunks (fs, f);
+
+  n_chunks = fifo_segment_num_free_chunks (fs, chunk_size);
+  SFIFO_TEST (n_chunks == 9, "free chunks should be %u is %u", 9, n_chunks);
+  /*
+   * Free
+   */
+  fifo_segment_free_fifo (fs, f);
+  n_chunks = fifo_segment_num_free_chunks (fs, ~0);
+  SFIFO_TEST (n_chunks == 9, "free chunks should be %u is %u", 9, n_chunks);
+
+  /*
+   * Cleanup
+   */
+  fifo_segment_delete (sm, fs);
+  vec_free (a->new_segment_indices);
+  return 0;
+}
+
 static int
 sfifo_test_fifo_segment_slave (int verbose)
 {
@@ -2219,6 +2287,11 @@ sfifo_test_fifo_segment (vlib_main_t * vm, unformat_input_t * input)
 	  if ((rv = sfifo_test_fifo_segment_fifo_grow (verbose)))
 	    return -1;
 	}
+      else if (unformat (input, "shrink fifo"))
+	{
+	  if ((rv = sfifo_test_fifo_segment_fifo_shrink (verbose)))
+	    return -1;
+	}
       else if (unformat (input, "all"))
 	{
 	  if ((rv = sfifo_test_fifo_segment_hello_world (verbose)))
@@ -2227,6 +2300,8 @@ sfifo_test_fifo_segment (vlib_main_t * vm, unformat_input_t * input)
 	    return -1;
 	  if ((rv = sfifo_test_fifo_segment_fifo_grow (verbose)))
 	    return -1;
+	  if ((rv = sfifo_test_fifo_segment_fifo_shrink (verbose)))
+	    return -1;
 	  /* Pretty slow so avoid running it always
 	     if ((rv = sfifo_test_fifo_segment_master_slave (verbose)))
 	     return -1;

src/svm/fifo_segment.c

@@ -570,6 +570,35 @@ fifo_segment_grow_fifo (fifo_segment_t * fs, svm_fifo_t * f, u32 chunk_size)
   return 0;
 }
 
+int
+fifo_segment_collect_fifo_chunks (fifo_segment_t * fs, svm_fifo_t * f)
+{
+  svm_fifo_chunk_t *cur, *next;
+  ssvm_shared_header_t *sh;
+  void *oldheap;
+  int fl_index;
+
+  sh = fs->ssvm.sh;
+  ssvm_lock_non_recursive (sh, 1);
+
+  oldheap = ssvm_push_heap (sh);
+  cur = svm_fifo_collect_chunks (f);
+
+  while (cur)
+    {
+      next = cur->next;
+      fl_index = fs_free_list_for_size (cur->length);
+      cur->next = fs->h->free_chunks[fl_index];
+      fs->h->free_chunks[fl_index] = cur;
+      cur = next;
+    }
+
+  ssvm_pop_heap (oldheap);
+  ssvm_unlock_non_recursive (sh);
+
+  return 0;
+}
+
 /**
  * Get number of active fifos
  */

src/svm/fifo_segment.h

@@ -128,6 +128,15 @@ void 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);
+
+/**
+ * Collect unused chunks for fifo
+ *
+ * @param fs		fifo segment for fifo
+ * @param f		fifo whose chunks are to be collected
+ * @return		0 on success, error otherwise
+ */
+int fifo_segment_collect_fifo_chunks (fifo_segment_t * fs, svm_fifo_t * f);
 u8 fifo_segment_has_fifos (fifo_segment_t * fs);
 svm_fifo_t *fifo_segment_get_fifo_list (fifo_segment_t * fs);
 u32 fifo_segment_num_fifos (fifo_segment_t * fs);

src/svm/svm_fifo.c

@@ -677,7 +677,10 @@ svm_fifo_reduce_size (svm_fifo_t * f, u32 len, u8 try_shrink)
   svm_fifo_chunk_t *cur;
   u32 actual_len = 0;
 
-  if (len > f->nitems)
+  /* Abort if trying to reduce by more than fifo size or if
+   * fifo is undergoing resizing already */
+  if (len >= f->size || f->size > f->nitems + 1
+      || (f->flags & SVM_FIFO_F_SHRINK) || (f->flags & SVM_FIFO_F_GROW))
     return 0;
 
   /* last chunk that will not be removed */

src/vnet/session/segment_manager.c

@@ -682,6 +682,37 @@ segment_manager_grow_fifo (segment_manager_t * sm, svm_fifo_t * f, u32 size)
   return rv;
 }
 
+int
+segment_manager_collect_fifo_chunks (segment_manager_t * sm, svm_fifo_t * f)
+{
+  fifo_segment_t *fs;
+  int rv;
+
+  fs = segment_manager_get_segment_w_lock (sm, f->segment_index);
+  rv = fifo_segment_collect_fifo_chunks (fs, f);
+  segment_manager_segment_reader_unlock (sm);
+
+  return rv;
+}
+
+int
+segment_manager_shrink_fifo (segment_manager_t * sm, svm_fifo_t * f, u32 size,
+			     u8 is_producer)
+{
+  int rv;
+
+  rv = svm_fifo_reduce_size (f, size, is_producer);
+
+  /* Nothing to collect at this point */
+  if (!is_producer)
+    return rv;
+
+  if (f->flags & SVM_FIFO_F_COLLECT_CHUNKS)
+    segment_manager_collect_fifo_chunks (sm, f);
+
+  return rv;
+}
+
 u32
 segment_manager_evt_q_expected_size (u32 q_len)
 {

src/vnet/session/segment_manager.h

@@ -125,6 +125,40 @@ void segment_manager_dealloc_fifos (svm_fifo_t * rx_fifo,
  */
 int segment_manager_grow_fifo (segment_manager_t * sm, svm_fifo_t * f,
 			       u32 size);
+
+/**
+ * Request to shrink fifo owned by segment manager
+ *
+ * If this is not called by the producer, no attempt is made to reduce the
+ * size until the producer tries to enqueue more data. To collect the chunks
+ * that are to be removed call @ref segment_manager_collect_fifo_chunks
+ *
+ * Size reduction does not affect fifo chunk boundaries. Therefore chunks are
+ * not split and the amount of bytes to be removed can be equal to or less
+ * than what was requested.
+ *
+ * @param sm		segment manager that owns the fifo
+ * @param f		fifo to be shrunk
+ * @param size		amount of bytes to remove from fifo
+ * @param is_producer	flag that indicates is caller is the producer for the
+ * 			fifo.
+ * @return		actual number of bytes to be removed
+ */
+int segment_manager_shrink_fifo (segment_manager_t * sm, svm_fifo_t * f,
+				 u32 size, u8 is_producer);
+
+/**
+ * Collect fifo chunks that are no longer used
+ *
+ * This should not be called unless SVM_FIFO_F_COLLECT_CHUNKS is set for
+ * the fifo. The chunks are returned to the fifo segment freelist.
+ *
+ * @param sm		segment manager that owns the fifo
+ * @param f		fifo whose chunks are to be collected
+ * @return		0 on success, error otherwise
+ */
+int segment_manager_collect_fifo_chunks (segment_manager_t * sm,
+					 svm_fifo_t * f);
 u8 segment_manager_has_fifos (segment_manager_t * sm);
 
 svm_msg_q_t *segment_manager_alloc_queue (fifo_segment_t * fs,