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>

6 files changed, 41 insertions, 46 deletions

diff --git a/src/vnet/tcp/builtin_client.c b/src/vnet/tcp/builtin_client.c
index f8fbf28c..276beb21 100644
--- a/src/vnet/tcp/builtin_client.c
+++ b/src/vnet/tcp/builtin_client.c
@@ -62,8 +62,7 @@ send_test_chunk (tclient_main_t * tm, session_t * s)
   bytes_this_chunk = bytes_this_chunk < s->bytes_to_send
     ? bytes_this_chunk : s->bytes_to_send;
 
-  rv = svm_fifo_enqueue_nowait (s->server_tx_fifo, 0 /*pid */ ,
-				bytes_this_chunk,
+  rv = svm_fifo_enqueue_nowait (s->server_tx_fifo, bytes_this_chunk,
 				test_data + test_buf_offset);
 
   /* If we managed to enqueue data... */
@@ -95,7 +94,7 @@ send_test_chunk (tclient_main_t * tm, session_t * s)
 	{
 	  /* Fabricate TX event, send to vpp */
 	  evt.fifo = s->server_tx_fifo;
-	  evt.event_type = FIFO_EVENT_SERVER_TX;
+	  evt.event_type = FIFO_EVENT_APP_TX;
 	  evt.event_id = serial_number++;
 
 	  unix_shared_memory_queue_add (tm->vpp_event_queue, (u8 *) & evt,
@@ -113,7 +112,7 @@ receive_test_chunk (tclient_main_t * tm, session_t * s)
   /* Allow enqueuing of new event */
   // svm_fifo_unset_event (rx_fifo);
 
-  n_read = svm_fifo_dequeue_nowait (rx_fifo, 0, vec_len (tm->rx_buf),
+  n_read = svm_fifo_dequeue_nowait (rx_fifo, vec_len (tm->rx_buf),
 				    tm->rx_buf);
   if (n_read > 0)
     {
@@ -457,6 +456,8 @@ attach_builtin_test_clients ()
 
   options[SESSION_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
   options[SESSION_OPTIONS_SEGMENT_SIZE] = (2 << 30);	/*$$$$ config / arg */
+  options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
+
   a->options = options;
 
   return vnet_application_attach (a);
diff --git a/src/vnet/tcp/builtin_server.c b/src/vnet/tcp/builtin_server.c
index 8308e3d9..34682699 100644
--- a/src/vnet/tcp/builtin_server.c
+++ b/src/vnet/tcp/builtin_server.c
@@ -180,7 +180,7 @@ builtin_server_rx_callback (stream_session_t * s)
   vec_validate (bsm->rx_buf, max_transfer - 1);
   _vec_len (bsm->rx_buf) = max_transfer;
 
-  actual_transfer = svm_fifo_dequeue_nowait (rx_fifo, 0, max_transfer,
+  actual_transfer = svm_fifo_dequeue_nowait (rx_fifo, max_transfer,
 					     bsm->rx_buf);
   ASSERT (actual_transfer == max_transfer);
 
@@ -190,8 +190,7 @@ builtin_server_rx_callback (stream_session_t * s)
    * Echo back
    */
 
-  n_written =
-    svm_fifo_enqueue_nowait (tx_fifo, 0, actual_transfer, bsm->rx_buf);
+  n_written = svm_fifo_enqueue_nowait (tx_fifo, actual_transfer, bsm->rx_buf);
 
   if (n_written != max_transfer)
     clib_warning ("short trout!");
@@ -200,7 +199,7 @@ builtin_server_rx_callback (stream_session_t * s)
     {
       /* Fabricate TX event, send to vpp */
       evt.fifo = tx_fifo;
-      evt.event_type = FIFO_EVENT_SERVER_TX;
+      evt.event_type = FIFO_EVENT_APP_TX;
       evt.event_id = serial_number++;
 
       unix_shared_memory_queue_add (bsm->vpp_queue[s->thread_index],
@@ -288,6 +287,7 @@ server_attach ()
   a->options[SESSION_OPTIONS_SEGMENT_SIZE] = 128 << 20;
   a->options[SESSION_OPTIONS_RX_FIFO_SIZE] = 1 << 16;
   a->options[SESSION_OPTIONS_TX_FIFO_SIZE] = 1 << 16;
+  a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
   a->segment_name = segment_name;
   a->segment_name_length = ARRAY_LEN (segment_name);
 
diff --git a/src/vnet/tcp/tcp.c b/src/vnet/tcp/tcp.c
index 12982589..245a35ab 100644
--- a/src/vnet/tcp/tcp.c
+++ b/src/vnet/tcp/tcp.c
@@ -487,7 +487,8 @@ u8 *
 format_tcp_connection (u8 * s, va_list * args)
 {
   tcp_connection_t *tc = va_arg (*args, tcp_connection_t *);
-
+  if (!tc)
+    return s;
   if (tc->c_is_ip4)
     {
       s = format (s, "[#%d][%s] %U:%d->%U:%d", tc->c_thread_index, "T",
@@ -747,12 +748,14 @@ void
 tcp_initialize_timer_wheels (tcp_main_t * tm)
 {
   tw_timer_wheel_16t_2w_512sl_t *tw;
-  vec_foreach (tw, tm->timer_wheels)
-  {
+  /* *INDENT-OFF* */
+  foreach_vlib_main (({
+    tw = &tm->timer_wheels[ii];
     tw_timer_wheel_init_16t_2w_512sl (tw, tcp_expired_timers_dispatch,
 				      100e-3 /* timer period 100ms */ , ~0);
-    tw->last_run_time = vlib_time_now (tm->vlib_main);
-  }
+    tw->last_run_time = vlib_time_now (this_vlib_main);
+  }));
+  /* *INDENT-ON* */
 }
 
 clib_error_t *
diff --git a/src/vnet/tcp/tcp_input.c b/src/vnet/tcp/tcp_input.c
index 97679aaf..3bd53878 100644
--- a/src/vnet/tcp/tcp_input.c
+++ b/src/vnet/tcp/tcp_input.c
@@ -1011,8 +1011,8 @@ tcp_session_enqueue_ooo (tcp_connection_t * tc, vlib_buffer_t * b,
 
   clib_warning ("ooo: offset %d len %d", offset, data_len);
 
-  rv = svm_fifo_enqueue_with_offset (s0->server_rx_fifo, s0->pid, offset,
-				     data_len, vlib_buffer_get_current (b));
+  rv = svm_fifo_enqueue_with_offset (s0->server_rx_fifo, offset, data_len,
+				     vlib_buffer_get_current (b));
 
   /* Nothing written */
   if (rv)
@@ -2392,8 +2392,8 @@ tcp46_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
 	    {
 	      t0 = vlib_add_trace (vm, node, b0, sizeof (*t0));
 	      clib_memcpy (&t0->tcp_header, tcp0, sizeof (t0->tcp_header));
-	      clib_memcpy (&t0->tcp_connection, tc0,
-			   sizeof (t0->tcp_connection));
+	      if (tc0)
+		clib_memcpy (&t0->tcp_connection, tc0, sizeof (*tc0));
 	    }
 
 	  vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next,
diff --git a/src/vnet/tcp/tcp_output.c b/src/vnet/tcp/tcp_output.c
index a7be8bd5..4e1a7aa5 100644
--- a/src/vnet/tcp/tcp_output.c
+++ b/src/vnet/tcp/tcp_output.c
@@ -1558,7 +1558,6 @@ tcp46_send_reset_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
 	  vlib_buffer_t *b0;
 	  tcp_tx_trace_t *t0;
 	  tcp_header_t *th0;
-	  tcp_connection_t *tc0;
 	  u32 error0 = TCP_ERROR_RST_SENT, next0 = TCP_RESET_NEXT_IP_LOOKUP;
 
 	  bi0 = from[0];
@@ -1592,13 +1591,8 @@ tcp46_send_reset_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
 		th0 = ip4_next_header ((ip4_header_t *) th0);
 	      else
 		th0 = ip6_next_header ((ip6_header_t *) th0);
-	      tc0 =
-		tcp_connection_get (vnet_buffer (b0)->tcp.connection_index,
-				    my_thread_index);
 	      t0 = vlib_add_trace (vm, node, b0, sizeof (*t0));
 	      clib_memcpy (&t0->tcp_header, th0, sizeof (t0->tcp_header));
-	      clib_memcpy (&t0->tcp_connection, tc0,
-			   sizeof (t0->tcp_connection));
 	    }
 
 	  vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next,
diff --git a/src/vnet/tcp/tcp_test.c b/src/vnet/tcp/tcp_test.c
index 890e50b9..0146154b 100644
--- a/src/vnet/tcp/tcp_test.c
+++ b/src/vnet/tcp/tcp_test.c
@@ -351,8 +351,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
   /*
    * Enqueue an initial (un-dequeued) chunk
    */
-  rv = svm_fifo_enqueue_nowait (f, 0 /* pid */ ,
-				sizeof (u32), (u8 *) test_data);
+  rv = svm_fifo_enqueue_nowait (f, sizeof (u32), (u8 *) test_data);
   TCP_TEST ((rv == sizeof (u32)), "enqueued %d", rv);
   TCP_TEST ((f->tail == 4), "fifo tail %u", f->tail);
 
@@ -364,7 +363,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
     {
       offset = (2 * i + 1) * sizeof (u32);
       data = (u8 *) (test_data + (2 * i + 1));
-      rv = svm_fifo_enqueue_with_offset (f, 0, offset, sizeof (u32), data);
+      rv = svm_fifo_enqueue_with_offset (f, offset, sizeof (u32), data);
       if (verbose)
 	vlib_cli_output (vm, "add [%d] [%d, %d]", 2 * i + 1, offset,
 			 offset + sizeof (u32));
@@ -393,7 +392,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
     {
       offset = (2 * i + 0) * sizeof (u32);
       data = (u8 *) (test_data + (2 * i + 0));
-      rv = svm_fifo_enqueue_with_offset (f, 0, offset, sizeof (u32), data);
+      rv = svm_fifo_enqueue_with_offset (f, offset, sizeof (u32), data);
       if (verbose)
 	vlib_cli_output (vm, "add [%d] [%d, %d]", 2 * i, offset,
 			 offset + sizeof (u32));
@@ -418,8 +417,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
   /*
    * Enqueue the missing u32
    */
-  rv = svm_fifo_enqueue_nowait (f, 0 /* pid */ , sizeof (u32),
-				(u8 *) (test_data + 2));
+  rv = svm_fifo_enqueue_nowait (f, sizeof (u32), (u8 *) (test_data + 2));
   if (verbose)
     vlib_cli_output (vm, "fifo after missing link: %U", format_svm_fifo, f,
 		     1);
@@ -432,8 +430,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
    */
   for (i = 0; i < 7; i++)
     {
-      rv = svm_fifo_dequeue_nowait (f, 0 /* pid */ , sizeof (u32),
-				    (u8 *) & data_word);
+      rv = svm_fifo_dequeue_nowait (f, sizeof (u32), (u8 *) & data_word);
       if (rv != sizeof (u32))
 	{
 	  clib_warning ("bytes dequeues %u", rv);
@@ -457,7 +454,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
     {
       offset = (2 * i + 1) * sizeof (u32);
       data = (u8 *) (test_data + (2 * i + 1));
-      rv = svm_fifo_enqueue_with_offset (f, 0, offset, sizeof (u32), data);
+      rv = svm_fifo_enqueue_with_offset (f, offset, sizeof (u32), data);
       if (verbose)
 	vlib_cli_output (vm, "add [%d] [%d, %d]", 2 * i + 1, offset,
 			 offset + sizeof (u32));
@@ -468,13 +465,13 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
 	}
     }
 
-  rv = svm_fifo_enqueue_with_offset (f, 0, 8, 21, data);
+  rv = svm_fifo_enqueue_with_offset (f, 8, 21, data);
   TCP_TEST ((rv == 0), "ooo enqueued %u", rv);
   TCP_TEST ((svm_fifo_number_ooo_segments (f) == 1),
 	    "number of ooo segments %u", svm_fifo_number_ooo_segments (f));
 
   vec_validate (data_buf, vec_len (data));
-  svm_fifo_peek (f, 0, 0, vec_len (data), data_buf);
+  svm_fifo_peek (f, 0, vec_len (data), data_buf);
   if (compare_data (data_buf, data, 8, vec_len (data), &j))
     {
       TCP_TEST (0, "[%d] peeked %u expected %u", j, data_buf[j], data[j]);
@@ -491,7 +488,7 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
     {
       offset = (2 * i + 1) * sizeof (u32);
       data = (u8 *) (test_data + (2 * i + 1));
-      rv = svm_fifo_enqueue_with_offset (f, 0, offset, sizeof (u32), data);
+      rv = svm_fifo_enqueue_with_offset (f, offset, sizeof (u32), data);
       if (verbose)
 	vlib_cli_output (vm, "add [%d] [%d, %d]", 2 * i + 1, offset,
 			 offset + sizeof (u32));
@@ -502,13 +499,13 @@ tcp_test_fifo1 (vlib_main_t * vm, unformat_input_t * input)
 	}
     }
 
-  rv = svm_fifo_enqueue_nowait (f, 0, 29, data);
+  rv = svm_fifo_enqueue_nowait (f, 29, data);
   TCP_TEST ((rv == 32), "ooo enqueued %u", rv);
   TCP_TEST ((svm_fifo_number_ooo_segments (f) == 0),
 	    "number of ooo segments %u", svm_fifo_number_ooo_segments (f));
 
   vec_validate (data_buf, vec_len (data));
-  svm_fifo_peek (f, 0, 0, vec_len (data), data_buf);
+  svm_fifo_peek (f, 0, vec_len (data), data_buf);
   if (compare_data (data_buf, data, 0, vec_len (data), &j))
     {
       TCP_TEST (0, "[%d] peeked %u expected %u", j, data_buf[j], data[j]);
@@ -551,7 +548,7 @@ tcp_test_fifo2 (vlib_main_t * vm)
     {
       tp = vp + i;
       data64 = tp->offset;
-      rv = svm_fifo_enqueue_with_offset (f, 0, tp->offset, tp->len,
+      rv = svm_fifo_enqueue_with_offset (f, tp->offset, tp->len,
 					 (u8 *) & data64);
     }
 
@@ -565,7 +562,7 @@ tcp_test_fifo2 (vlib_main_t * vm)
 	    "first ooo seg length %u", ooo_seg->length);
 
   data64 = 0;
-  rv = svm_fifo_enqueue_nowait (f, 0, sizeof (u32), (u8 *) & data64);
+  rv = svm_fifo_enqueue_nowait (f, sizeof (u32), (u8 *) & data64);
   TCP_TEST ((rv == 3000), "bytes to be enqueued %u", rv);
 
   svm_fifo_free (f);
@@ -581,7 +578,7 @@ tcp_test_fifo2 (vlib_main_t * vm)
     {
       tp = &test_data[i];
       data64 = tp->offset;
-      rv = svm_fifo_enqueue_with_offset (f, 0, tp->offset, tp->len,
+      rv = svm_fifo_enqueue_with_offset (f, tp->offset, tp->len,
 					 (u8 *) & data64);
       if (rv)
 	{
@@ -599,7 +596,7 @@ tcp_test_fifo2 (vlib_main_t * vm)
 	    "first ooo seg length %u", ooo_seg->length);
 
   data64 = 0;
-  rv = svm_fifo_enqueue_nowait (f, 0, sizeof (u32), (u8 *) & data64);
+  rv = svm_fifo_enqueue_nowait (f, sizeof (u32), (u8 *) & data64);
 
   TCP_TEST ((rv == 3000), "bytes to be enqueued %u", rv);
 
@@ -755,7 +752,7 @@ tcp_test_fifo3 (vlib_main_t * vm, unformat_input_t * input)
   for (i = 0; i < vec_len (generate); i++)
     {
       tp = generate + i;
-      rv = svm_fifo_enqueue_with_offset (f, 0, fifo_initial_offset
+      rv = svm_fifo_enqueue_with_offset (f, fifo_initial_offset
 					 + tp->offset, tp->len,
 					 (u8 *) data_pattern + tp->offset);
     }
@@ -776,7 +773,7 @@ tcp_test_fifo3 (vlib_main_t * vm, unformat_input_t * input)
       u32 bytes_to_enq = 1;
       if (in_seq_all)
 	bytes_to_enq = total_size;
-      rv = svm_fifo_enqueue_nowait (f, 0, bytes_to_enq, data_pattern + 0);
+      rv = svm_fifo_enqueue_nowait (f, bytes_to_enq, data_pattern + 0);
 
       if (verbose)
 	vlib_cli_output (vm, "in-order enqueue returned %d", rv);
@@ -793,7 +790,7 @@ tcp_test_fifo3 (vlib_main_t * vm, unformat_input_t * input)
    * Test if peeked data is the same as original data
    */
   vec_validate (data_buf, vec_len (data_pattern));
-  svm_fifo_peek (f, 0, 0, vec_len (data_pattern), data_buf);
+  svm_fifo_peek (f, 0, vec_len (data_pattern), data_buf);
   if (compare_data (data_buf, data_pattern, 0, vec_len (data_pattern), &j))
     {
       TCP_TEST (0, "[%d] peeked %u expected %u", j, data_buf[j],
@@ -806,11 +803,11 @@ tcp_test_fifo3 (vlib_main_t * vm, unformat_input_t * input)
    */
   if (drop)
     {
-      svm_fifo_dequeue_drop (f, 0, vec_len (data_pattern));
+      svm_fifo_dequeue_drop (f, vec_len (data_pattern));
     }
   else
     {
-      svm_fifo_dequeue_nowait (f, 0, vec_len (data_pattern), data_buf);
+      svm_fifo_dequeue_nowait (f, vec_len (data_pattern), data_buf);
       if (compare_data
 	  (data_buf, data_pattern, 0, vec_len (data_pattern), &j))
 	{