svm: store normalized head/tail for fifo

If head/tail are stored as "absolute" values that are normalized to [0,
fifo_size] interval, when fifo is shrunk/grown the consumer and producer
have to independently update to the new fifo size and fix head and tail,
respectively.

If the head and tail are stored as normalized values, under the right
conditions, they don't need to be fixed when fifo size changes.

This reverts one of the changes in gerrit 18223.

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

diff --git a/src/plugins/unittest/svm_fifo_test.c b/src/plugins/unittest/svm_fifo_test.c
index 99f0e50..d2afc36 100644 (file)
--- a/src/plugins/unittest/svm_fifo_test.c
+++ b/src/plugins/unittest/svm_fifo_test.c
@@ -806,7 +806,7 @@ sfifo_test_fifo4 (vlib_main_t * vm, unformat_input_t * input)
 static u32
 fifo_pos (svm_fifo_t * f, u32 pos)
 {
-  return pos;
+  return pos % f->size;
 }
 
 /* Avoids exposing svm_fifo.c internal function */
@@ -1194,6 +1194,7 @@ sfifo_test_fifo_grow (vlib_main_t * vm, unformat_input_t * input)
   svm_fifo_chunk_t *c, *next, *prev;
   u8 *test_data = 0, *data_buf = 0;
   svm_fifo_t *f;
+  u32 old_tail;
 
   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
     {
@@ -1288,10 +1289,14 @@ sfifo_test_fifo_grow (vlib_main_t * vm, unformat_input_t * input)
   SFIFO_TEST (f->size == fifo_size + 200, "size expected %u is %u",
              fifo_size + 200, f->size);
 
-  svm_fifo_dequeue (f, 201, data_buf);
+  old_tail = f->tail;
+  svm_fifo_dequeue (f, 101, data_buf);
 
   SFIFO_TEST (f->size == fifo_size + 200 + 10 * 100, "size expected %u is %u",
              fifo_size + 200 + 10 * 100, f->size);
+  SFIFO_TEST (f->tail == old_tail, "new tail expected %u is %u", old_tail,
+             f->tail);
+
   /*
    * Enqueue/dequeue tests
    */

diff --git a/src/svm/svm_fifo.c b/src/svm/svm_fifo.c
index 3512c72..d563877 100644 (file)
--- a/src/svm/svm_fifo.c
+++ b/src/svm/svm_fifo.c
@@ -125,7 +125,7 @@ position_diff (svm_fifo_t * f, u32 a, u32 b, u32 tail)
 static inline u32
 ooo_segment_end_pos (svm_fifo_t * f, ooo_segment_t * s)
 {
-  return s->start + s->length;
+  return (s->start + s->length) % f->size;
 }
 
 void
@@ -202,8 +202,8 @@ ooo_segment_add (svm_fifo_t * f, u32 offset, u32 head, u32 tail, u32 length)
 
   ASSERT (offset + length <= f_distance_to (f, head, tail) || head == tail);
 
-  offset_pos = tail + offset;
-  offset_end_pos = tail + offset + length;
+  offset_pos = (tail + offset) % f->size;
+  offset_end_pos = (tail + offset + length) % f->size;
 
   f->ooos_newest = OOO_SEGMENT_INVALID_INDEX;
 
@@ -350,7 +350,7 @@ ooo_segment_try_collect (svm_fifo_t * f, u32 n_bytes_enqueued, u32 * tail)
       if (s->length > diff)
        {
          bytes = s->length - diff;
-         *tail = *tail + bytes;
+         *tail = (*tail + bytes) % f->size;
          ooo_segment_free (f, s_index);
          break;
        }
@@ -460,7 +460,7 @@ svm_fifo_size_update (svm_fifo_t * f, svm_fifo_chunk_t * c)
 static void
 svm_fifo_try_size_update (svm_fifo_t * f, u32 new_head)
 {
-  if (new_head % f->size > f->tail % f->size)
+  if (new_head > f->tail)
     return;
 
   svm_fifo_size_update (f, f->new_chunks);
@@ -603,8 +603,7 @@ svm_fifo_overwrite_head (svm_fifo_t * f, u8 * src, u32 len)
 
   f_load_head_tail_cons (f, &head, &tail);
   c = f->head_chunk;
-  head_idx = head % f->size;
-  head_idx -= c->start_byte;
+  head_idx = head - c->start_byte;
   n_chunk = c->length - head_idx;
   if (len <= n_chunk)
     clib_memcpy_fast (&c->data[head_idx], src, len);
@@ -632,9 +631,8 @@ svm_fifo_enqueue (svm_fifo_t * f, u32 len, const u8 * src)
 
   /* number of bytes we're going to copy */
   len = clib_min (free_count, len);
-  svm_fifo_copy_to_chunk (f, f->tail_chunk, tail % f->size, src, len,
-                         &f->tail_chunk);
-  tail += len;
+  svm_fifo_copy_to_chunk (f, f->tail_chunk, tail, src, len, &f->tail_chunk);
+  tail = (tail + len) % f->size;
 
   svm_fifo_trace_add (f, head, len, 2);
 
@@ -672,7 +670,7 @@ svm_fifo_enqueue_with_offset (svm_fifo_t * f, u32 offset, u32 len, u8 * src)
   f->ooos_newest = OOO_SEGMENT_INVALID_INDEX;
   svm_fifo_trace_add (f, offset, len, 1);
   ooo_segment_add (f, offset, head, tail, len);
-  tail_idx = (tail % f->size + offset) % f->size;
+  tail_idx = (tail + offset) % f->size;
 
   if (!svm_fifo_chunk_includes_pos (f->ooo_enq, tail_idx))
     f->ooo_enq = svm_fifo_find_chunk (f, tail_idx);
@@ -696,9 +694,8 @@ svm_fifo_dequeue (svm_fifo_t * f, u32 len, u8 * dst)
     return SVM_FIFO_EEMPTY;
 
   len = clib_min (cursize, len);
-  svm_fifo_copy_from_chunk (f, f->head_chunk, head % f->size, dst, len,
-                           &f->head_chunk);
-  head += len;
+  svm_fifo_copy_from_chunk (f, f->head_chunk, head, dst, len, &f->head_chunk);
+  head = (head + len) % f->size;
 
   if (PREDICT_FALSE (f->flags & SVM_FIFO_F_SIZE_UPDATE))
     svm_fifo_try_size_update (f, head);
@@ -723,7 +720,7 @@ svm_fifo_peek (svm_fifo_t * f, u32 offset, u32 len, u8 * dst)
     return SVM_FIFO_EEMPTY;
 
   len = clib_min (cursize - offset, len);
-  head_idx = (head % f->size + offset) % f->size;
+  head_idx = (head + offset) % f->size;
   if (!svm_fifo_chunk_includes_pos (f->ooo_deq, head_idx))
     f->ooo_deq = svm_fifo_find_chunk (f, head_idx);
 
@@ -749,7 +746,7 @@ svm_fifo_dequeue_drop (svm_fifo_t * f, u32 len)
   total_drop_bytes = clib_min (cursize, len);
 
   /* move head */
-  head += total_drop_bytes;
+  head = (head + total_drop_bytes) % f->size;
 
   /* store-rel: consumer owned index (paired with load-acq in producer) */
   clib_atomic_store_rel_n (&f->head, head);
@@ -779,7 +776,7 @@ svm_fifo_segments (svm_fifo_t * f, svm_fifo_seg_t * fs)
   if (PREDICT_FALSE (cursize == 0))
     return SVM_FIFO_EEMPTY;
 
-  head_idx = head % f->size;
+  head_idx = head;
 
   if (tail < head)
     {
@@ -801,14 +798,13 @@ svm_fifo_segments (svm_fifo_t * f, svm_fifo_seg_t * fs)
 void
 svm_fifo_segments_free (svm_fifo_t * f, svm_fifo_seg_t * fs)
 {
-  u32 head, head_idx;
+  u32 head;
 
   /* consumer owned index */
   head = f->head;
-  head_idx = head % f->size;
 
-  ASSERT (fs[0].data == f->head_chunk->data + head_idx);
-  head += fs[0].len + fs[1].len;
+  ASSERT (fs[0].data == f->head_chunk->data + head);
+  head = (head + fs[0].len + fs[1].len) % f->size;
   /* store-rel: consumer owned index (paired with load-acq in producer) */
   clib_atomic_store_rel_n (&f->head, head);
 }
@@ -849,15 +845,17 @@ svm_fifo_first_ooo_segment (svm_fifo_t * f)
 void
 svm_fifo_init_pointers (svm_fifo_t * f, u32 head, u32 tail)
 {
+  head = head % f->size;
+  tail = tail % f->size;
   clib_atomic_store_rel_n (&f->head, head);
   clib_atomic_store_rel_n (&f->tail, tail);
   if (f->flags & SVM_FIFO_F_MULTI_CHUNK)
     {
       svm_fifo_chunk_t *c;
-      c = svm_fifo_find_chunk (f, head % f->size);
+      c = svm_fifo_find_chunk (f, head);
       ASSERT (c != 0);
       f->head_chunk = f->ooo_deq = c;
-      c = svm_fifo_find_chunk (f, tail % f->size);
+      c = svm_fifo_find_chunk (f, tail);
       ASSERT (c != 0);
       f->tail_chunk = f->ooo_enq = c;
     }

diff --git a/src/svm/svm_fifo.h b/src/svm/svm_fifo.h
index 96ca3ee..390e117 100644 (file)
--- a/src/svm/svm_fifo.h
+++ b/src/svm/svm_fifo.h
@@ -186,47 +186,47 @@ f_load_head_tail_all_acq (svm_fifo_t * f, u32 * head, u32 * tail)
 }
 
 /**
- * Fifo free bytes, i.e., number of free bytes
+ * Distance to a from b, i.e., a - b in the fifo
  *
- * Internal function
+ * Internal function.
  */
 static inline u32
-f_free_count (svm_fifo_t * f, u32 head, u32 tail)
+f_distance_to (svm_fifo_t * f, u32 a, u32 b)
 {
-  return (f->nitems + head - tail);
+  return ((f->size + a - b) % f->size);
 }
 
 /**
- * Fifo current size, i.e., number of bytes enqueued
+ * Distance from a to b, i.e., b - a in the fifo
  *
  * Internal function.
  */
 static inline u32
-f_cursize (svm_fifo_t * f, u32 head, u32 tail)
+f_distance_from (svm_fifo_t * f, u32 a, u32 b)
 {
-  return (f->nitems - f_free_count (f, head, tail));
+  return ((f->size + b - a) % f->size);
 }
 
 /**
- * Distance to a from b, i.e., a - b in the fifo
+ * Fifo current size, i.e., number of bytes enqueued
  *
  * Internal function.
  */
 static inline u32
-f_distance_to (svm_fifo_t * f, u32 a, u32 b)
+f_cursize (svm_fifo_t * f, u32 head, u32 tail)
 {
-  return ((f->size + a - b) % f->size);
+  return (head <= tail ? tail - head : f->size + tail - head);
 }
 
 /**
- * Distance from a to b, i.e., b - a in the fifo
+ * Fifo free bytes, i.e., number of free bytes
  *
- * Internal function.
+ * Internal function
  */
 static inline u32
-f_distance_from (svm_fifo_t * f, u32 a, u32 b)
+f_free_count (svm_fifo_t * f, u32 head, u32 tail)
 {
-  return ((f->size + b - a) % f->size);
+  return (f->nitems - f_cursize (f, head, tail));
 }
 
 /**
@@ -535,7 +535,7 @@ svm_fifo_is_wrapped (svm_fifo_t * f)
 {
   u32 head, tail;
   f_load_head_tail_all_acq (f, &head, &tail);
-  return head % f->size > tail % f->size;
+  return head > tail;
 }
 
 /**
@@ -575,11 +575,8 @@ static inline u32
 svm_fifo_max_read_chunk (svm_fifo_t * f)
 {
   u32 head, tail;
-  u32 head_idx, tail_idx;
   f_load_head_tail_cons (f, &head, &tail);
-  head_idx = head % f->size;
-  tail_idx = tail % f->size;
-  return tail_idx > head_idx ? (tail_idx - head_idx) : (f->size - head_idx);
+  return tail >= head ? (tail - head) : (f->size - head);
 }
 
 /**
@@ -589,11 +586,8 @@ static inline u32
 svm_fifo_max_write_chunk (svm_fifo_t * f)
 {
   u32 head, tail;
-  u32 head_idx, tail_idx;
   f_load_head_tail_prod (f, &head, &tail);
-  head_idx = head % f->size;
-  tail_idx = tail % f->size;
-  return tail_idx >= head_idx ? (f->size - tail_idx) : (head_idx - tail_idx);
+  return tail > head ? f->size - tail : f_free_count (f, head, tail);
 }
 
 /**
@@ -610,7 +604,7 @@ 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 += len;
+  tail = (tail + len) % f->size;
   /* store-rel: producer owned index (paired with load-acq in consumer) */
   clib_atomic_store_rel_n (&f->tail, tail);
 }
@@ -619,16 +613,14 @@ static inline u8 *
 svm_fifo_head (svm_fifo_t * f)
 {
   /* load-relaxed: consumer owned index */
-  return (f->head_chunk->data
-         + ((f->head % f->size) - f->head_chunk->start_byte));
+  return (f->head_chunk->data + (f->head - f->head_chunk->start_byte));
 }
 
 static inline u8 *
 svm_fifo_tail (svm_fifo_t * f)
 {
   /* load-relaxed: producer owned index */
-  return (f->tail_chunk->data
-         + ((f->tail % f->size) - f->tail_chunk->start_byte));
+  return (f->tail_chunk->data + (f->tail - f->tail_chunk->start_byte));
 }
 
 static inline u8