Make tcp active open data structures thread safe

- Cleanup half-open connections and timers on the right thread
- Ensure half-open connection and transport endpoint pools are thread safe
- Enqueue TX events to the correct vpp thread in the builtin client
- Use transport proto in transport connections instead of session type

Change-Id: Id13239a206afbff6f34a38afa510fe014e4b2049
Signed-off-by: Florin Coras <fcoras@cisco.com>
Signed-off-by: Dave Barach <dave@barachs.net>

8 files changed, 59 insertions, 81 deletions

diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 48000a6f..004c7193 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -64,7 +64,8 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
   s->server_tx_fifo = server_tx_fifo;
 
   /* Initialize state machine, such as it is... */
-  s->session_type = tc->proto;
+  s->session_type = session_type_from_proto_and_ip (tc->transport_proto,
+						    tc->is_ip4);
   s->session_state = SESSION_STATE_CONNECTING;
   s->svm_segment_index = fifo_segment_index;
   s->thread_index = thread_index;
@@ -354,8 +355,7 @@ stream_session_init_fifos_pointers (transport_connection_t * tc,
 }
 
 int
-stream_session_connect_notify (transport_connection_t * tc, u8 sst,
-			       u8 is_fail)
+stream_session_connect_notify (transport_connection_t * tc, u8 is_fail)
 {
   application_t *app;
   stream_session_t *new_s = 0;
@@ -365,7 +365,7 @@ stream_session_connect_notify (transport_connection_t * tc, u8 sst,
 
   handle = stream_session_half_open_lookup_handle (&tc->lcl_ip, &tc->rmt_ip,
 						   tc->lcl_port, tc->rmt_port,
-						   tc->proto);
+						   tc->transport_proto);
   if (handle == HALF_OPEN_LOOKUP_INVALID_VALUE)
     {
       clib_warning ("This can't be good!");
@@ -391,7 +391,7 @@ stream_session_connect_notify (transport_connection_t * tc, u8 sst,
 	new_s->app_index = app->index;
     }
 
-  /* Notify client */
+  /* Notify client application */
   if (app->cb_fns.session_connected_callback (app->index, api_context, new_s,
 					      is_fail))
     {
@@ -406,7 +406,7 @@ stream_session_connect_notify (transport_connection_t * tc, u8 sst,
     }
 
   /* Cleanup session lookup */
-  stream_session_half_open_table_del (sst, tc);
+  stream_session_half_open_table_del (tc);
 
   return error;
 }
@@ -567,7 +567,7 @@ stream_session_open (u32 app_index, session_type_t st,
   handle = (((u64) app_index) << 32) | (u64) tc->c_index;
 
   /* Add to the half-open lookup table */
-  stream_session_half_open_table_add (st, tc, handle);
+  stream_session_half_open_table_add (tc, handle);
 
   *res = tc;
 
diff --git a/src/vnet/session/session.h b/src/vnet/session/session.h
index bb22f100..180b9f8a 100644
--- a/src/vnet/session/session.h
+++ b/src/vnet/session/session.h
@@ -274,8 +274,7 @@ stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
 			   u32 offset, u32 max_bytes);
 u32 stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes);
 
-int stream_session_connect_notify (transport_connection_t * tc, u8 sst,
-				   u8 is_fail);
+int stream_session_connect_notify (transport_connection_t * tc, u8 is_fail);
 void stream_session_init_fifos_pointers (transport_connection_t * tc,
 					 u32 rx_pointer, u32 tx_pointer);
 
diff --git a/src/vnet/session/session_cli.c b/src/vnet/session/session_cli.c
index 4d432977..de564ea7 100755
--- a/src/vnet/session/session_cli.c
+++ b/src/vnet/session/session_cli.c
@@ -57,7 +57,7 @@ format_stream_session (u8 * s, va_list * args)
   u8 *str = 0;
   tp_vft = session_get_transport_vft (ss->session_type);
 
-  if (verbose == 1)
+  if (verbose == 1 && ss->session_state >= SESSION_STATE_ACCEPTING)
     str = format (0, "%-10u%-10u%-10lld",
 		  svm_fifo_max_dequeue (ss->server_rx_fifo),
 		  svm_fifo_max_enqueue (ss->server_tx_fifo),
diff --git a/src/vnet/session/session_lookup.c b/src/vnet/session/session_lookup.c
index b3862ee3..1ce22f80 100644
--- a/src/vnet/session/session_lookup.c
+++ b/src/vnet/session/session_lookup.c
@@ -107,7 +107,7 @@ always_inline void
 make_v4_ss_kv_from_tc (session_kv4_t * kv, transport_connection_t * t)
 {
   return make_v4_ss_kv (kv, &t->lcl_ip.ip4, &t->rmt_ip.ip4, t->lcl_port,
-			t->rmt_port, t->proto);
+			t->rmt_port, t->transport_proto);
 }
 
 always_inline void
@@ -150,7 +150,7 @@ always_inline void
 make_v6_ss_kv_from_tc (session_kv6_t * kv, transport_connection_t * t)
 {
   make_v6_ss_kv (kv, &t->lcl_ip.ip6, &t->rmt_ip.ip6, t->lcl_port,
-		 t->rmt_port, t->proto);
+		 t->rmt_port, t->transport_proto);
 }
 
 /*
@@ -164,23 +164,17 @@ stream_session_table_add_for_tc (transport_connection_t * tc, u64 value)
   session_kv4_t kv4;
   session_kv6_t kv6;
 
-  switch (tc->proto)
+  if (tc->is_ip4)
     {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
       make_v4_ss_kv_from_tc (&kv4, tc);
       kv4.value = value;
       clib_bihash_add_del_16_8 (&sl->v4_session_hash, &kv4, 1 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
+    }
+  else
+    {
       make_v6_ss_kv_from_tc (&kv6, tc);
       kv6.value = value;
       clib_bihash_add_del_48_8 (&sl->v6_session_hash, &kv6, 1 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
     }
 }
 
@@ -195,59 +189,24 @@ stream_session_table_add (session_manager_main_t * smm, stream_session_t * s,
   stream_session_table_add_for_tc (tc, value);
 }
 
-void
-stream_session_half_open_table_add (session_type_t sst,
-				    transport_connection_t * tc, u64 value)
-{
-  session_lookup_t *sl = &session_lookup;
-  session_kv4_t kv4;
-  session_kv6_t kv6;
-
-  switch (sst)
-    {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
-      make_v4_ss_kv_from_tc (&kv4, tc);
-      kv4.value = value;
-      clib_bihash_add_del_16_8 (&sl->v4_half_open_hash, &kv4,
-				1 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
-      make_v6_ss_kv_from_tc (&kv6, tc);
-      kv6.value = value;
-      clib_bihash_add_del_48_8 (&sl->v6_half_open_hash, &kv6,
-				1 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
-    }
-}
-
 int
 stream_session_table_del_for_tc (transport_connection_t * tc)
 {
   session_lookup_t *sl = &session_lookup;
   session_kv4_t kv4;
   session_kv6_t kv6;
-  switch (tc->proto)
+
+  if (tc->is_ip4)
     {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
       make_v4_ss_kv_from_tc (&kv4, tc);
       return clib_bihash_add_del_16_8 (&sl->v4_session_hash, &kv4,
 				       0 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
+    }
+  else
+    {
       make_v6_ss_kv_from_tc (&kv6, tc);
       return clib_bihash_add_del_48_8 (&sl->v6_session_hash, &kv6,
 				       0 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
     }
 
   return 0;
@@ -262,30 +221,48 @@ stream_session_table_del (stream_session_t * s)
   return stream_session_table_del_for_tc (ts);
 }
 
+
 void
-stream_session_half_open_table_del (u8 sst, transport_connection_t * tc)
+stream_session_half_open_table_add (transport_connection_t * tc, u64 value)
 {
   session_lookup_t *sl = &session_lookup;
   session_kv4_t kv4;
   session_kv6_t kv6;
 
-  switch (sst)
+  if (tc->is_ip4)
+    {
+      make_v4_ss_kv_from_tc (&kv4, tc);
+      kv4.value = value;
+      clib_bihash_add_del_16_8 (&sl->v4_half_open_hash, &kv4,
+				1 /* is_add */ );
+    }
+  else
+    {
+      make_v6_ss_kv_from_tc (&kv6, tc);
+      kv6.value = value;
+      clib_bihash_add_del_48_8 (&sl->v6_half_open_hash, &kv6,
+				1 /* is_add */ );
+    }
+}
+
+void
+stream_session_half_open_table_del (transport_connection_t * tc)
+{
+  session_lookup_t *sl = &session_lookup;
+  session_kv4_t kv4;
+  session_kv6_t kv6;
+
+  if (tc->is_ip4)
     {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
       make_v4_ss_kv_from_tc (&kv4, tc);
       clib_bihash_add_del_16_8 (&sl->v4_half_open_hash, &kv4,
 				0 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
+    }
+  else
+    {
       make_v6_ss_kv_from_tc (&kv6, tc);
       clib_bihash_add_del_48_8 (&sl->v6_half_open_hash, &kv6,
 				0 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
     }
 }
 
diff --git a/src/vnet/session/session_lookup.h b/src/vnet/session/session_lookup.h
index 9e92dab1..cf1dc013 100644
--- a/src/vnet/session/session_lookup.h
+++ b/src/vnet/session/session_lookup.h
@@ -83,9 +83,8 @@ transport_connection_t *stream_session_half_open_lookup (ip46_address_t * lcl,
 void stream_session_table_add_for_tc (transport_connection_t * tc, u64 value);
 int stream_session_table_del_for_tc (transport_connection_t * tc);
 int stream_session_table_del (stream_session_t * s);
-void stream_session_half_open_table_del (u8 sst, transport_connection_t * tc);
-void stream_session_half_open_table_add (session_type_t sst,
-					 transport_connection_t * tc,
+void stream_session_half_open_table_del (transport_connection_t * tc);
+void stream_session_half_open_table_add (transport_connection_t * tc,
 					 u64 value);
 
 void session_lookup_init (void);
diff --git a/src/vnet/session/transport.h b/src/vnet/session/transport.h
index 3895a60a..e56be338 100644
--- a/src/vnet/session/transport.h
+++ b/src/vnet/session/transport.h
@@ -31,12 +31,12 @@ typedef struct _transport_connection
   ip46_address_t lcl_ip;	/**< Local IP */
   u16 lcl_port;			/**< Local port */
   u16 rmt_port;			/**< Remote port */
-  u8 proto;			/**< Protocol id (also session type) */
+  u8 transport_proto;		/**< Protocol id */
+  u8 is_ip4;			/**< Flag if IP4 connection */
   u32 vrf;			/**< FIB table id */
 
   u32 s_index;			/**< Parent session index */
   u32 c_index;			/**< Connection index in transport pool */
-  u8 is_ip4;			/**< Flag if IP4 connection */
   u32 thread_index;		/**< Worker-thread index */
 
   fib_node_index_t rmt_fei;	/**< FIB entry index for rmt */
@@ -56,7 +56,7 @@ typedef struct _transport_connection
 #define c_rmt_ip6 connection.rmt_ip.ip6
 #define c_lcl_port connection.lcl_port
 #define c_rmt_port connection.rmt_port
-#define c_proto connection.proto
+#define c_transport_proto connection.transport_proto
 #define c_vrf connection.vrf
 #define c_state connection.state
 #define c_s_index connection.s_index
diff --git a/src/vnet/session/transport_interface.c b/src/vnet/session/transport_interface.c
index eb12aa69..ef8d1e49 100644
--- a/src/vnet/session/transport_interface.c
+++ b/src/vnet/session/transport_interface.c
@@ -73,9 +73,12 @@ transport_endpoint_table_del (transport_endpoint_table_t * ht,
  * @param vft - virtual function table
  */
 void
-session_register_transport (u8 session_type,
+session_register_transport (transport_proto_t transport_proto, u8 is_ip4,
 			    const transport_proto_vft_t * vft)
 {
+  u8 session_type;
+  session_type = session_type_from_proto_and_ip (transport_proto, is_ip4);
+
   vec_validate (tp_vfts, session_type);
   tp_vfts[session_type] = *vft;
 
diff --git a/src/vnet/session/transport_interface.h b/src/vnet/session/transport_interface.h
index b7e86ee7..661221c4 100644
--- a/src/vnet/session/transport_interface.h
+++ b/src/vnet/session/transport_interface.h
@@ -67,7 +67,7 @@ void transport_endpoint_table_add (transport_endpoint_table_t * ht,
 void transport_endpoint_table_del (transport_endpoint_table_t * ht,
 				   transport_endpoint_t * te);
 
-void session_register_transport (u8 session_type,
+void session_register_transport (transport_proto_t transport_proto, u8 is_ip4,
 				 const transport_proto_vft_t * vft);
 transport_proto_vft_t *session_get_transport_vft (u8 session_type);