Session layer improvements

Among others:
- Moved app event queue to shared memory segment
- Use private memory segment for builtin apps
- Remove pid from svm fifo
- Protect session fifo (de)allocation
- Use fifo event for session disconnects
- Have session queue node poll in all wk threads

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

5 files changed, 82 insertions, 61 deletions

diff --git a/src/svm/svm_fifo.c b/src/svm/svm_fifo.c
index f428d3eced7..8f2ed0c9d95 100644
--- a/src/svm/svm_fifo.c
+++ b/src/svm/svm_fifo.c
@@ -57,7 +57,7 @@ format_svm_fifo (u8 * s, va_list * args)
   if (verbose > 1)
     s = format
       (s, "server session %d thread %d client session %d thread %d\n",
-       f->server_session_index, f->server_thread_index,
+       f->master_session_index, f->master_thread_index,
        f->client_session_index, f->client_thread_index);
 
   if (verbose)
@@ -353,8 +353,7 @@ ooo_segment_try_collect (svm_fifo_t * f, u32 n_bytes_enqueued)
 }
 
 static int
-svm_fifo_enqueue_internal (svm_fifo_t * f,
-			   int pid, u32 max_bytes, u8 * copy_from_here)
+svm_fifo_enqueue_internal (svm_fifo_t * f, u32 max_bytes, u8 * copy_from_here)
 {
   u32 total_copy_bytes, first_copy_bytes, second_copy_bytes;
   u32 cursize, nitems;
@@ -411,10 +410,9 @@ svm_fifo_enqueue_internal (svm_fifo_t * f,
 }
 
 int
-svm_fifo_enqueue_nowait (svm_fifo_t * f,
-			 int pid, u32 max_bytes, u8 * copy_from_here)
+svm_fifo_enqueue_nowait (svm_fifo_t * f, u32 max_bytes, u8 * copy_from_here)
 {
-  return svm_fifo_enqueue_internal (f, pid, max_bytes, copy_from_here);
+  return svm_fifo_enqueue_internal (f, max_bytes, copy_from_here);
 }
 
 /**
@@ -426,7 +424,6 @@ svm_fifo_enqueue_nowait (svm_fifo_t * f,
  */
 static int
 svm_fifo_enqueue_with_offset_internal (svm_fifo_t * f,
-				       int pid,
 				       u32 offset,
 				       u32 required_bytes,
 				       u8 * copy_from_here)
@@ -439,7 +436,7 @@ svm_fifo_enqueue_with_offset_internal (svm_fifo_t * f,
   /* Users would do well to avoid this */
   if (PREDICT_FALSE (f->tail == (offset % f->nitems)))
     {
-      rv = svm_fifo_enqueue_internal (f, pid, required_bytes, copy_from_here);
+      rv = svm_fifo_enqueue_internal (f, required_bytes, copy_from_here);
       if (rv > 0)
 	return 0;
       return -1;
@@ -484,18 +481,16 @@ svm_fifo_enqueue_with_offset_internal (svm_fifo_t * f,
 
 int
 svm_fifo_enqueue_with_offset (svm_fifo_t * f,
-			      int pid,
 			      u32 offset,
 			      u32 required_bytes, u8 * copy_from_here)
 {
-  return svm_fifo_enqueue_with_offset_internal
-    (f, pid, offset, required_bytes, copy_from_here);
+  return svm_fifo_enqueue_with_offset_internal (f, offset, required_bytes,
+						copy_from_here);
 }
 
 
 static int
-svm_fifo_dequeue_internal (svm_fifo_t * f,
-			   int pid, u32 max_bytes, u8 * copy_here)
+svm_fifo_dequeue_internal (svm_fifo_t * f, u32 max_bytes, u8 * copy_here)
 {
   u32 total_copy_bytes, first_copy_bytes, second_copy_bytes;
   u32 cursize, nitems;
@@ -545,14 +540,13 @@ svm_fifo_dequeue_internal (svm_fifo_t * f,
 }
 
 int
-svm_fifo_dequeue_nowait (svm_fifo_t * f,
-			 int pid, u32 max_bytes, u8 * copy_here)
+svm_fifo_dequeue_nowait (svm_fifo_t * f, u32 max_bytes, u8 * copy_here)
 {
-  return svm_fifo_dequeue_internal (f, pid, max_bytes, copy_here);
+  return svm_fifo_dequeue_internal (f, max_bytes, copy_here);
 }
 
 int
-svm_fifo_peek (svm_fifo_t * f, int pid, u32 relative_offset, u32 max_bytes,
+svm_fifo_peek (svm_fifo_t * f, u32 relative_offset, u32 max_bytes,
 	       u8 * copy_here)
 {
   u32 total_copy_bytes, first_copy_bytes, second_copy_bytes;
@@ -590,7 +584,7 @@ svm_fifo_peek (svm_fifo_t * f, int pid, u32 relative_offset, u32 max_bytes,
 }
 
 int
-svm_fifo_dequeue_drop (svm_fifo_t * f, int pid, u32 max_bytes)
+svm_fifo_dequeue_drop (svm_fifo_t * f, u32 max_bytes)
 {
   u32 total_drop_bytes, first_drop_bytes, second_drop_bytes;
   u32 cursize, nitems;
diff --git a/src/svm/svm_fifo.h b/src/svm/svm_fifo.h
index 0fff2577517..d67237c69b0 100644
--- a/src/svm/svm_fifo.h
+++ b/src/svm/svm_fifo.h
@@ -23,13 +23,6 @@
 #include <vppinfra/format.h>
 #include <pthread.h>
 
-typedef enum
-{
-  SVM_FIFO_TAG_NOT_HELD = 0,
-  SVM_FIFO_TAG_DEQUEUE,
-  SVM_FIFO_TAG_ENQUEUE,
-} svm_lock_tag_t;
-
 /** Out-of-order segment */
 typedef struct
 {
@@ -37,7 +30,7 @@ typedef struct
   u32 prev;	/**< Previous linked-list element pool index */
 
   u32 start;	/**< Start of segment, normalized*/
-  u32 length;		/**< Length of segment */
+  u32 length;	/**< Length of segment */
 } ooo_segment_t;
 
 format_function_t format_ooo_segment;
@@ -52,12 +45,11 @@ typedef struct
     CLIB_CACHE_LINE_ALIGN_MARK (end_cursize);
 
   volatile u8 has_event;	/**< non-zero if deq event exists */
-  u32 owner_pid;
 
   /* Backpointers */
-  u32 server_session_index;
+  u32 master_session_index;
   u32 client_session_index;
-  u8 server_thread_index;
+  u8 master_thread_index;
   u8 client_thread_index;
   u32 segment_manager;
     CLIB_CACHE_LINE_ALIGN_MARK (end_shared);
@@ -117,19 +109,14 @@ svm_fifo_unset_event (svm_fifo_t * f)
 svm_fifo_t *svm_fifo_create (u32 data_size_in_bytes);
 void svm_fifo_free (svm_fifo_t * f);
 
-int svm_fifo_enqueue_nowait (svm_fifo_t * f, int pid, u32 max_bytes,
+int svm_fifo_enqueue_nowait (svm_fifo_t * f, u32 max_bytes,
 			     u8 * copy_from_here);
+int svm_fifo_enqueue_with_offset (svm_fifo_t * f, u32 offset,
+				  u32 required_bytes, u8 * copy_from_here);
+int svm_fifo_dequeue_nowait (svm_fifo_t * f, u32 max_bytes, u8 * copy_here);
 
-int svm_fifo_enqueue_with_offset (svm_fifo_t * f, int pid,
-				  u32 offset, u32 required_bytes,
-				  u8 * copy_from_here);
-
-int svm_fifo_dequeue_nowait (svm_fifo_t * f, int pid, u32 max_bytes,
-			     u8 * copy_here);
-
-int svm_fifo_peek (svm_fifo_t * f, int pid, u32 offset, u32 max_bytes,
-		   u8 * copy_here);
-int svm_fifo_dequeue_drop (svm_fifo_t * f, int pid, u32 max_bytes);
+int svm_fifo_peek (svm_fifo_t * f, u32 offset, u32 max_bytes, u8 * copy_here);
+int svm_fifo_dequeue_drop (svm_fifo_t * f, u32 max_bytes);
 u32 svm_fifo_number_ooo_segments (svm_fifo_t * f);
 ooo_segment_t *svm_fifo_first_ooo_segment (svm_fifo_t * f);
 
diff --git a/src/svm/svm_fifo_segment.c b/src/svm/svm_fifo_segment.c
index acabb3bd95c..281fae27997 100644
--- a/src/svm/svm_fifo_segment.c
+++ b/src/svm/svm_fifo_segment.c
@@ -70,6 +70,44 @@ svm_fifo_segment_create (svm_fifo_segment_create_args_t * a)
   return (0);
 }
 
+/** Create an svm fifo segment in process-private memory */
+int
+svm_fifo_segment_create_process_private (svm_fifo_segment_create_args_t * a)
+{
+  svm_fifo_segment_private_t *s;
+  svm_fifo_segment_main_t *sm = &svm_fifo_segment_main;
+  ssvm_shared_header_t *sh;
+  svm_fifo_segment_header_t *fsh;
+
+  /* Allocate a fresh segment */
+  pool_get (sm->segments, s);
+  memset (s, 0, sizeof (*s));
+
+  s->ssvm.ssvm_size = ~0;
+  s->ssvm.i_am_master = 1;
+  s->ssvm.my_pid = getpid ();
+  s->ssvm.name = (u8 *) a->segment_name;
+  s->ssvm.requested_va = ~0;
+
+  /* Allocate a [sic] shared memory header, in process memory... */
+  sh = clib_mem_alloc_aligned (sizeof (*sh), CLIB_CACHE_LINE_BYTES);
+  s->ssvm.sh = sh;
+
+  memset (sh, 0, sizeof (*sh));
+  sh->heap = clib_mem_get_heap ();
+
+  /* Set up svm_fifo_segment shared header */
+  fsh = clib_mem_alloc (sizeof (*fsh));
+  memset (fsh, 0, sizeof (*fsh));
+  sh->opaque[0] = fsh;
+  s->h = fsh;
+  fsh->segment_name = format (0, "%s%c", a->segment_name, 0);
+
+  sh->ready = 1;
+  a->new_segment_index = s - sm->segments;
+  return (0);
+}
+
 /** (slave) attach to an svm fifo segment */
 int
 svm_fifo_segment_attach (svm_fifo_segment_create_args_t * a)
@@ -82,7 +120,6 @@ svm_fifo_segment_attach (svm_fifo_segment_create_args_t * a)
 
   /* Allocate a fresh segment */
   pool_get (sm->segments, s);
-
   memset (s, 0, sizeof (*s));
 
   s->ssvm.ssvm_size = a->segment_size;
@@ -126,19 +163,22 @@ svm_fifo_segment_alloc_fifo (svm_fifo_segment_private_t * s,
 
   sh = s->ssvm.sh;
   fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
+
+  ssvm_lock (sh, 1, 0);
   oldheap = ssvm_push_heap (sh);
 
   /* Note: this can fail, in which case: create another segment */
   f = svm_fifo_create (data_size_in_bytes);
-  if (f == 0)
+  if (PREDICT_FALSE (f == 0))
     {
       ssvm_pop_heap (oldheap);
+      ssvm_unlock (sh);
       return (0);
     }
 
   vec_add1 (fsh->fifos, f);
-
   ssvm_pop_heap (oldheap);
+  ssvm_unlock (sh);
   return (f);
 }
 
@@ -152,8 +192,9 @@ svm_fifo_segment_free_fifo (svm_fifo_segment_private_t * s, svm_fifo_t * f)
 
   sh = s->ssvm.sh;
   fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
-  oldheap = ssvm_push_heap (sh);
 
+  ssvm_lock (sh, 1, 0);
+  oldheap = ssvm_push_heap (sh);
   for (i = 0; i < vec_len (fsh->fifos); i++)
     {
       if (fsh->fifos[i] == f)
@@ -167,6 +208,7 @@ svm_fifo_segment_free_fifo (svm_fifo_segment_private_t * s, svm_fifo_t * f)
 found:
   clib_mem_free (f);
   ssvm_pop_heap (oldheap);
+  ssvm_unlock (sh);
 }
 
 void
diff --git a/src/svm/svm_fifo_segment.h b/src/svm/svm_fifo_segment.h
index 9ab47a4c161..4218013a9e4 100644
--- a/src/svm/svm_fifo_segment.h
+++ b/src/svm/svm_fifo_segment.h
@@ -17,6 +17,7 @@
 
 #include <svm/svm_fifo.h>
 #include <svm/ssvm.h>
+#include <vppinfra/lock.h>
 
 typedef struct
 {
@@ -32,6 +33,8 @@ typedef struct
 
 typedef struct
 {
+  volatile u32 lock;
+
   /** pool of segments */
   svm_fifo_segment_private_t *segments;
   /* Where to put the next one */
@@ -78,6 +81,8 @@ typedef enum
 } ssvm_fifo_segment_api_error_enum_t;
 
 int svm_fifo_segment_create (svm_fifo_segment_create_args_t * a);
+int svm_fifo_segment_create_process_private (svm_fifo_segment_create_args_t
+					     * a);
 int svm_fifo_segment_attach (svm_fifo_segment_create_args_t * a);
 void svm_fifo_segment_delete (svm_fifo_segment_private_t * s);
 
diff --git a/src/svm/test_svm_fifo1.c b/src/svm/test_svm_fifo1.c
index 355653df42a..398dd6d7c38 100644
--- a/src/svm/test_svm_fifo1.c
+++ b/src/svm/test_svm_fifo1.c
@@ -25,7 +25,6 @@ hello_world (int verbose)
   u8 *test_data;
   u8 *retrieved_data = 0;
   clib_error_t *error = 0;
-  int pid = getpid ();
 
   memset (a, 0, sizeof (*a));
 
@@ -48,18 +47,16 @@ hello_world (int verbose)
   vec_validate (retrieved_data, vec_len (test_data) - 1);
 
   while (svm_fifo_max_enqueue (f) >= vec_len (test_data))
-    svm_fifo_enqueue_nowait (f, pid, vec_len (test_data), test_data);
+    svm_fifo_enqueue_nowait (f, vec_len (test_data), test_data);
 
   while (svm_fifo_max_dequeue (f) >= vec_len (test_data))
-    svm_fifo_dequeue_nowait (f, pid, vec_len (retrieved_data),
-			     retrieved_data);
+    svm_fifo_dequeue_nowait (f, vec_len (retrieved_data), retrieved_data);
 
   while (svm_fifo_max_enqueue (f) >= vec_len (test_data))
-    svm_fifo_enqueue_nowait (f, pid, vec_len (test_data), test_data);
+    svm_fifo_enqueue_nowait (f, vec_len (test_data), test_data);
 
   while (svm_fifo_max_dequeue (f) >= vec_len (test_data))
-    svm_fifo_dequeue_nowait (f, pid, vec_len (retrieved_data),
-			     retrieved_data);
+    svm_fifo_dequeue_nowait (f, vec_len (retrieved_data), retrieved_data);
 
   if (!memcmp (retrieved_data, test_data, vec_len (test_data)))
     error = clib_error_return (0, "data test OK, got '%s'", retrieved_data);
@@ -81,7 +78,6 @@ master (int verbose)
   u8 *test_data;
   u8 *retrieved_data = 0;
   int i;
-  int pid = getpid ();
 
   memset (a, 0, sizeof (*a));
 
@@ -104,7 +100,7 @@ master (int verbose)
   vec_validate (retrieved_data, vec_len (test_data) - 1);
 
   for (i = 0; i < 1000; i++)
-    svm_fifo_enqueue_nowait (f, pid, vec_len (test_data), test_data);
+    svm_fifo_enqueue_nowait (f, vec_len (test_data), test_data);
 
   return clib_error_return (0, "master (enqueue) done");
 }
@@ -176,7 +172,6 @@ offset (int verbose)
   u32 *test_data = 0;
   u32 *recovered_data = 0;
   int i;
-  int pid = getpid ();
 
   memset (a, 0, sizeof (*a));
 
@@ -199,19 +194,19 @@ offset (int verbose)
     vec_add1 (test_data, i);
 
   /* Enqueue the first 1024 u32's */
-  svm_fifo_enqueue_nowait (f, pid, 4096 /* bytes to enqueue */ ,
+  svm_fifo_enqueue_nowait (f, 4096 /* bytes to enqueue */ ,
 			   (u8 *) test_data);
 
   /* Enqueue the third 1024 u32's 2048 ahead of the current tail */
-  svm_fifo_enqueue_with_offset (f, pid, 4096, 4096, (u8 *) & test_data[2048]);
+  svm_fifo_enqueue_with_offset (f, 4096, 4096, (u8 *) & test_data[2048]);
 
   /* Enqueue the second 1024 u32's at the current tail */
-  svm_fifo_enqueue_nowait (f, pid, 4096 /* bytes to enqueue */ ,
+  svm_fifo_enqueue_nowait (f, 4096 /* bytes to enqueue */ ,
 			   (u8 *) & test_data[1024]);
 
   vec_validate (recovered_data, (3 * 1024) - 1);
 
-  svm_fifo_dequeue_nowait (f, pid, 3 * 4096, (u8 *) recovered_data);
+  svm_fifo_dequeue_nowait (f, 3 * 4096, (u8 *) recovered_data);
 
   for (i = 0; i < (3 * 1024); i++)
     {
@@ -237,7 +232,6 @@ slave (int verbose)
   int rv;
   u8 *test_data;
   u8 *retrieved_data = 0;
-  int pid = getpid ();
   int i;
 
   memset (a, 0, sizeof (*a));
@@ -262,8 +256,7 @@ slave (int verbose)
 
   for (i = 0; i < 1000; i++)
     {
-      svm_fifo_dequeue_nowait (f, pid, vec_len (retrieved_data),
-			       retrieved_data);
+      svm_fifo_dequeue_nowait (f, vec_len (retrieved_data), retrieved_data);
       if (memcmp (retrieved_data, test_data, vec_len (retrieved_data)))
 	return clib_error_return (0, "retrieved data incorrect, '%s'",
 				  retrieved_data);