svm: basic support for fifo shrinking

As opposed to growing, this is not a bulk operation, instead dependent
on how the producer/consumer advance head and tail, the fifo will shrink
in one or multiple steps.

Only once the fifo's nitems and size are reduced to their appropriate
values, equal or larger to what was requested, can the fifo chunks be
collected by the owner. Chunk collection must be done with the segment
heap pushed.

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

1 files changed, 55 insertions, 24 deletions

diff --git a/src/svm/svm_fifo.h b/src/svm/svm_fifo.h
index 390e1170e27..417b0ecb672 100644
--- a/src/svm/svm_fifo.h
+++ b/src/svm/svm_fifo.h
@@ -64,9 +64,11 @@ typedef struct svm_fifo_chunk_
 
 typedef enum svm_fifo_flag_
 {
-  SVM_FIFO_F_SIZE_UPDATE = 1 << 0,
-  SVM_FIFO_F_MULTI_CHUNK = 1 << 1,
-  SVM_FIFO_F_LL_TRACKED = 1 << 2,
+  SVM_FIFO_F_MULTI_CHUNK = 1 << 0,
+  SVM_FIFO_F_GROW = 1 << 1,
+  SVM_FIFO_F_SHRINK = 1 << 2,
+  SVM_FIFO_F_COLLECT_CHUNKS = 1 << 3,
+  SVM_FIFO_F_LL_TRACKED = 1 << 4,
 } svm_fifo_flag_t;
 
 typedef struct _svm_fifo
@@ -86,12 +88,13 @@ typedef struct _svm_fifo
   u32 client_session_index;	/**< app session index */
   u8 master_thread_index;	/**< session layer thread index */
   u8 client_thread_index;	/**< app worker index */
+  i8 refcnt;			/**< reference count  */
   u32 segment_manager;		/**< session layer segment manager index */
   u32 segment_index;		/**< segment index in segment manager */
   u32 freelist_index;		/**< aka log2(allocated_size) - const. */
-  i8 refcnt;			/**< reference count  */
   struct _svm_fifo *next;	/**< next in freelist/active chain */
   struct _svm_fifo *prev;	/**< prev in active chain */
+  u32 size_decrement;		/**< bytes to remove from fifo */
 
     CLIB_CACHE_LINE_ALIGN_MARK (consumer);
   u32 head;			/**< fifo head position/byte */
@@ -172,7 +175,8 @@ f_load_head_tail_prod (svm_fifo_t * f, u32 * head, u32 * tail)
   *head = clib_atomic_load_acq_n (&f->head);
 }
 
-/* Load head and tail independent of producer/consumer role
+/**
+ * Load head and tail independent of producer/consumer role
  *
  * Internal function.
  */
@@ -230,6 +234,13 @@ f_free_count (svm_fifo_t * f, u32 head, u32 tail)
 }
 
 /**
+ * Try to shrink fifo size.
+ *
+ * Internal function.
+ */
+void svm_fifo_try_shrink (svm_fifo_t * f, u32 head, u32 tail);
+
+/**
  * Create fifo of requested size
  *
  * Allocates fifo on current heap.
@@ -267,6 +278,31 @@ svm_fifo_chunk_t *svm_fifo_chunk_alloc (u32 size);
  */
 void svm_fifo_add_chunk (svm_fifo_t * f, svm_fifo_chunk_t * c);
 /**
+ * Request to reduce fifo size by amount of bytes
+ *
+ * Because the producer might be enqueuing data when this is called, the
+ * actual size update is only applied when producer tries to enqueue new
+ * data, unless @param try_shrink is set.
+ *
+ * @param f		fifo
+ * @param len		number of bytes to remove from fifo. The actual number
+ * 			of bytes to be removed will be less or equal to this
+ * 			value.
+ * @param try_shrink	flg to indicate if it's safe to try to shrink fifo
+ * 			size. It should be set only if this is called by the
+ * 			producer of if the producer is not using the fifo
+ * @return		actual length fifo size will be reduced by
+ */
+int svm_fifo_reduce_size (svm_fifo_t * f, u32 len, u8 try_shrink);
+/**
+ * Removes chunks that are after fifo end byte
+ *
+ * Needs to be called with segment heap pushed.
+ *
+ * @param f fifo
+ */
+svm_fifo_chunk_t *svm_fifo_collect_chunks (svm_fifo_t * f);
+/**
  * Free fifo and associated state
  *
  * @param f	fifo
@@ -333,6 +369,16 @@ int svm_fifo_enqueue (svm_fifo_t * f, u32 len, const u8 * src);
  */
 int svm_fifo_enqueue_with_offset (svm_fifo_t * f, u32 offset, u32 len,
 				  u8 * src);
+
+/**
+ * Advance tail pointer
+ *
+ * Useful for moving tail pointer after external enqueue.
+ *
+ * @param f		fifo
+ * @param len		number of bytes to add to tail
+ */
+void svm_fifo_enqueue_nocopy (svm_fifo_t * f, u32 len);
 /**
  * Overwrite fifo head with new data
  *
@@ -551,6 +597,8 @@ svm_fifo_max_enqueue_prod (svm_fifo_t * f)
 {
   u32 head, tail;
   f_load_head_tail_prod (f, &head, &tail);
+  if (PREDICT_FALSE (f->flags & SVM_FIFO_F_SHRINK))
+    svm_fifo_try_shrink (f, head, tail);
   return f_free_count (f, head, tail);
 }
 
@@ -565,6 +613,8 @@ svm_fifo_max_enqueue (svm_fifo_t * f)
 {
   u32 head, tail;
   f_load_head_tail_all_acq (f, &head, &tail);
+  if (PREDICT_FALSE (f->flags & SVM_FIFO_F_SHRINK))
+    svm_fifo_try_shrink (f, head, tail);
   return f_free_count (f, head, tail);
 }
 
@@ -590,25 +640,6 @@ svm_fifo_max_write_chunk (svm_fifo_t * f)
   return tail > head ? f->size - tail : f_free_count (f, head, tail);
 }
 
-/**
- * Advance tail pointer
- *
- * Useful for moving tail pointer after external enqueue.
- *
- * @param f		fifo
- * @param len		number of bytes to add to tail
- */
-static inline void
-svm_fifo_enqueue_nocopy (svm_fifo_t * f, u32 len)
-{
-  ASSERT (len <= svm_fifo_max_enqueue_prod (f));
-  /* load-relaxed: producer owned index */
-  u32 tail = f->tail;
-  tail = (tail + len) % f->size;
-  /* store-rel: producer owned index (paired with load-acq in consumer) */
-  clib_atomic_store_rel_n (&f->tail, tail);
-}
-
 static inline u8 *
 svm_fifo_head (svm_fifo_t * f)
 {