4 files changed, 131 insertions, 87 deletions

diff --git a/src/vnet/session/application_local.c b/src/vnet/session/application_local.c
index 25ff780eab2..58d3789cfee 100644
--- a/src/vnet/session/application_local.c
+++ b/src/vnet/session/application_local.c
@@ -16,66 +16,56 @@
 #include <vnet/session/application_local.h>
 #include <vnet/session/session.h>
 
-static ct_connection_t *connections;
-
-static void
-ct_enable_disable_main_pre_input_node (u8 is_add)
+typedef struct ct_main_
 {
-  u32 n_conns;
-
-  if (!vlib_num_workers ())
-    return;
-
-  n_conns = pool_elts (connections);
-  if (n_conns > 2)
-    return;
-
-  if (n_conns > 0 && is_add)
-    vlib_node_set_state (vlib_get_main (),
-			 session_queue_pre_input_node.index,
-			 VLIB_NODE_STATE_POLLING);
-  else if (n_conns == 0)
-    vlib_node_set_state (vlib_get_main (),
-			 session_queue_pre_input_node.index,
-			 VLIB_NODE_STATE_DISABLED);
-}
+  ct_connection_t **connections;	/**< Per-worker connection pools */
+  u32 n_workers;			/**< Number of vpp workers */
+  u32 n_sessions;			/**< Cumulative sessions counter */
+} ct_main_t;
+
+static ct_main_t ct_main;
 
 static ct_connection_t *
-ct_connection_alloc (void)
+ct_connection_alloc (u32 thread_index)
 {
   ct_connection_t *ct;
 
-  pool_get_zero (connections, ct);
-  ct->c_c_index = ct - connections;
-  ct->c_thread_index = 0;
+  pool_get_zero (ct_main.connections[thread_index], ct);
+  ct->c_c_index = ct - ct_main.connections[thread_index];
+  ct->c_thread_index = thread_index;
   ct->client_wrk = ~0;
   ct->server_wrk = ~0;
   return ct;
 }
 
 static ct_connection_t *
-ct_connection_get (u32 ct_index)
+ct_connection_get (u32 ct_index, u32 thread_index)
 {
-  if (pool_is_free_index (connections, ct_index))
+  if (pool_is_free_index (ct_main.connections[thread_index], ct_index))
     return 0;
-  return pool_elt_at_index (connections, ct_index);
+  return pool_elt_at_index (ct_main.connections[thread_index], ct_index);
 }
 
 static void
 ct_connection_free (ct_connection_t * ct)
 {
   if (CLIB_DEBUG)
-    memset (ct, 0xfc, sizeof (*ct));
-  pool_put (connections, ct);
+    {
+      u32 thread_index = ct->c_thread_index;
+      memset (ct, 0xfc, sizeof (*ct));
+      pool_put (ct_main.connections[thread_index], ct);
+      return;
+    }
+  pool_put (ct_main.connections[ct->c_thread_index], ct);
 }
 
 session_t *
 ct_session_get_peer (session_t * s)
 {
   ct_connection_t *ct, *peer_ct;
-  ct = ct_connection_get (s->connection_index);
-  peer_ct = ct_connection_get (ct->peer_index);
-  return session_get (peer_ct->c_s_index, 0);
+  ct = ct_connection_get (s->connection_index, s->thread_index);
+  peer_ct = ct_connection_get (ct->peer_index, s->thread_index);
+  return session_get (peer_ct->c_s_index, s->thread_index);
 }
 
 void
@@ -92,16 +82,17 @@ ct_session_endpoint (session_t * ll, session_endpoint_t * sep)
 int
 ct_session_connect_notify (session_t * ss)
 {
+  u32 ss_index, opaque, thread_index;
   ct_connection_t *sct, *cct;
   app_worker_t *client_wrk;
   segment_manager_t *sm;
-  u32 ss_index, opaque;
   fifo_segment_t *seg;
   u64 segment_handle;
   int err = 0;
   session_t *cs;
 
   ss_index = ss->session_index;
+  thread_index = ss->thread_index;
   sct = (ct_connection_t *) session_get_transport (ss);
   client_wrk = app_worker_get (sct->client_wrk);
   opaque = sct->client_opaque;
@@ -124,10 +115,10 @@ ct_session_connect_notify (session_t * ss)
     }
 
   /* Alloc client session */
-  cct = ct_connection_get (sct->peer_index);
+  cct = ct_connection_get (sct->peer_index, thread_index);
 
-  cs = session_alloc (0);
-  ss = session_get (ss_index, 0);
+  cs = session_alloc (thread_index);
+  ss = session_get (ss_index, thread_index);
   cs->session_type = ss->session_type;
   cs->connection_index = sct->c_c_index;
   cs->listener_handle = SESSION_INVALID_HANDLE;
@@ -162,7 +153,7 @@ ct_session_connect_notify (session_t * ss)
       return -1;
     }
 
-  cs = session_get (cct->c_s_index, 0);
+  cs = session_get (cct->c_s_index, cct->c_thread_index);
   cs->session_state = SESSION_STATE_READY;
 
   return 0;
@@ -173,8 +164,9 @@ error:
 }
 
 static int
-ct_init_local_session (app_worker_t * client_wrk, app_worker_t * server_wrk,
-		       ct_connection_t * ct, session_t * ls, session_t * ll)
+ct_init_accepted_session (app_worker_t * server_wrk,
+			  ct_connection_t * ct, session_t * ls,
+			  session_t * ll)
 {
   u32 round_rx_fifo_sz, round_tx_fifo_sz, sm_index, seg_size;
   segment_manager_props_t *props;
@@ -241,50 +233,58 @@ ct_init_local_session (app_worker_t * client_wrk, app_worker_t * server_wrk,
   return 0;
 
 failed:
-  segment_manager_del_segment (sm, seg);
+  segment_manager_lock_and_del_segment (sm, seg_index);
   return rv;
 }
 
-static int
-ct_connect (app_worker_t * client_wrk, session_t * ll,
-	    session_endpoint_cfg_t * sep)
+typedef struct ct_accept_rpc_args
+{
+  u32 ll_s_index;
+  u32 thread_index;
+  ip46_address_t ip;
+  u16 port;
+  u8 is_ip4;
+  u32 opaque;
+  u32 client_wrk_index;
+} ct_accept_rpc_args_t;
+
+static void
+ct_accept_rpc_wrk_handler (void *accept_args)
 {
-  u32 cct_index, ll_index, ll_ct_index;
+  ct_accept_rpc_args_t *args = (ct_accept_rpc_args_t *) accept_args;
   ct_connection_t *sct, *cct, *ll_ct;
   app_worker_t *server_wrk;
-  session_t *ss;
+  session_t *ss, *ll;
+  u32 cct_index;
 
-  ll_index = ll->session_index;
-  ll_ct_index = ll->connection_index;
+  ll = listen_session_get (args->ll_s_index);
 
-  cct = ct_connection_alloc ();
+  cct = ct_connection_alloc (args->thread_index);
   cct_index = cct->c_c_index;
-  sct = ct_connection_alloc ();
-  ll_ct = ct_connection_get (ll_ct_index);
+  sct = ct_connection_alloc (args->thread_index);
+  ll_ct = ct_connection_get (ll->connection_index, 0 /* listener thread */ );
 
   /*
    * Alloc and init client transport
    */
-  cct = ct_connection_get (cct_index);
-  cct->c_thread_index = 0;
-  cct->c_rmt_port = sep->port;
+  cct = ct_connection_get (cct_index, args->thread_index);
+  cct->c_rmt_port = args->port;
   cct->c_lcl_port = 0;
-  cct->c_is_ip4 = sep->is_ip4;
-  clib_memcpy (&cct->c_rmt_ip, &sep->ip, sizeof (sep->ip));
+  cct->c_is_ip4 = args->is_ip4;
+  clib_memcpy (&cct->c_rmt_ip, &args->ip, sizeof (args->ip));
   cct->actual_tp = ll_ct->actual_tp;
   cct->is_client = 1;
 
   /*
    * Init server transport
    */
-  sct->c_thread_index = 0;
   sct->c_rmt_port = 0;
   sct->c_lcl_port = ll_ct->c_lcl_port;
-  sct->c_is_ip4 = sep->is_ip4;
+  sct->c_is_ip4 = args->is_ip4;
   clib_memcpy (&sct->c_lcl_ip, &ll_ct->c_lcl_ip, sizeof (ll_ct->c_lcl_ip));
-  sct->client_wrk = client_wrk->wrk_index;
+  sct->client_wrk = args->client_wrk_index;
   sct->c_proto = TRANSPORT_PROTO_NONE;
-  sct->client_opaque = sep->opaque;
+  sct->client_opaque = args->opaque;
   sct->actual_tp = ll_ct->actual_tp;
 
   sct->peer_index = cct->c_c_index;
@@ -294,8 +294,8 @@ ct_connect (app_worker_t * client_wrk, session_t * ll,
    * Accept server session. Client session is created only after
    * server confirms accept.
    */
-  ss = session_alloc (0);
-  ll = listen_session_get (ll_index);
+  ss = session_alloc (args->thread_index);
+  ll = listen_session_get (args->ll_s_index);
   ss->session_type = session_type_from_proto_and_ip (TRANSPORT_PROTO_NONE,
 						     sct->c_is_ip4);
   ss->connection_index = sct->c_c_index;
@@ -308,11 +308,11 @@ ct_connect (app_worker_t * client_wrk, session_t * ll,
   sct->c_s_index = ss->session_index;
   sct->server_wrk = ss->app_wrk_index;
 
-  if (ct_init_local_session (client_wrk, server_wrk, sct, ss, ll))
+  if (ct_init_accepted_session (server_wrk, sct, ss, ll))
     {
       ct_connection_free (sct);
       session_free (ss);
-      return -1;
+      return;
     }
 
   ss->session_state = SESSION_STATE_ACCEPTING;
@@ -321,11 +321,40 @@ ct_connect (app_worker_t * client_wrk, session_t * ll,
       ct_connection_free (sct);
       segment_manager_dealloc_fifos (ss->rx_fifo, ss->tx_fifo);
       session_free (ss);
-      return -1;
+      return;
     }
 
   cct->segment_handle = sct->segment_handle;
-  ct_enable_disable_main_pre_input_node (1 /* is_add */ );
+
+  clib_mem_free (args);
+}
+
+static int
+ct_connect (app_worker_t * client_wrk, session_t * ll,
+	    session_endpoint_cfg_t * sep)
+{
+  ct_accept_rpc_args_t *args;
+  ct_main_t *cm = &ct_main;
+  u32 thread_index;
+
+  /* Simple round-robin policy for spreading sessions over workers. We skip
+   * thread index 0, i.e., offset the index by 1, when we have workers as it
+   * is the one dedicated to main thread. Note that n_workers does not include
+   * main thread */
+  cm->n_sessions += 1;
+  thread_index = cm->n_workers ? (cm->n_sessions % cm->n_workers) + 1 : 0;
+
+  args = clib_mem_alloc (sizeof (*args));
+  args->ll_s_index = ll->session_index;
+  args->thread_index = thread_index;
+  clib_memcpy_fast (&args->ip, &sep->ip, sizeof (ip46_address_t));
+  args->port = sep->port;
+  args->is_ip4 = sep->is_ip4;
+  args->opaque = sep->opaque;
+  args->client_wrk_index = client_wrk->wrk_index;
+
+  session_send_rpc_evt_to_thread (thread_index, ct_accept_rpc_wrk_handler,
+				  args);
   return 0;
 }
 
@@ -336,14 +365,13 @@ ct_start_listen (u32 app_listener_index, transport_endpoint_t * tep)
   ct_connection_t *ct;
 
   sep = (session_endpoint_cfg_t *) tep;
-  ct = ct_connection_alloc ();
+  ct = ct_connection_alloc (0);
   ct->server_wrk = sep->app_wrk_index;
   ct->c_is_ip4 = sep->is_ip4;
   clib_memcpy (&ct->c_lcl_ip, &sep->ip, sizeof (sep->ip));
   ct->c_lcl_port = sep->port;
   ct->c_s_index = app_listener_index;
   ct->actual_tp = sep->transport_proto;
-  ct_enable_disable_main_pre_input_node (1 /* is_add */ );
   return ct->c_c_index;
 }
 
@@ -351,16 +379,15 @@ static u32
 ct_stop_listen (u32 ct_index)
 {
   ct_connection_t *ct;
-  ct = ct_connection_get (ct_index);
+  ct = ct_connection_get (ct_index, 0);
   ct_connection_free (ct);
-  ct_enable_disable_main_pre_input_node (0 /* is_add */ );
   return 0;
 }
 
 static transport_connection_t *
 ct_listener_get (u32 ct_index)
 {
-  return (transport_connection_t *) ct_connection_get (ct_index);
+  return (transport_connection_t *) ct_connection_get (ct_index, 0);
 }
 
 static int
@@ -433,15 +460,15 @@ ct_session_close (u32 ct_index, u32 thread_index)
   app_worker_t *app_wrk;
   session_t *s;
 
-  ct = ct_connection_get (ct_index);
-  peer_ct = ct_connection_get (ct->peer_index);
+  ct = ct_connection_get (ct_index, thread_index);
+  peer_ct = ct_connection_get (ct->peer_index, thread_index);
   if (peer_ct)
     {
       peer_ct->peer_index = ~0;
       session_transport_closing_notify (&peer_ct->connection);
     }
 
-  s = session_get (ct->c_s_index, 0);
+  s = session_get (ct->c_s_index, ct->c_thread_index);
   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
   if (app_wrk)
     app_worker_del_segment_notify (app_wrk, ct->segment_handle);
@@ -450,13 +477,13 @@ ct_session_close (u32 ct_index, u32 thread_index)
     segment_manager_dealloc_fifos (ct->client_rx_fifo, ct->client_tx_fifo);
 
   ct_connection_free (ct);
-  ct_enable_disable_main_pre_input_node (0 /* is_add */ );
 }
 
 static transport_connection_t *
 ct_session_get (u32 ct_index, u32 thread_index)
 {
-  return (transport_connection_t *) ct_connection_get (ct_index);
+  return (transport_connection_t *) ct_connection_get (ct_index,
+						       thread_index);
 }
 
 static u8 *
@@ -503,7 +530,7 @@ ct_app_rx_evt (transport_connection_t * tc)
   ct_connection_t *ct = (ct_connection_t *) tc, *peer_ct;
   session_t *ps;
 
-  peer_ct = ct_connection_get (ct->peer_index);
+  peer_ct = ct_connection_get (ct->peer_index, tc->thread_index);
   if (!peer_ct)
     return -1;
   ps = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
@@ -516,7 +543,7 @@ format_ct_listener (u8 * s, va_list * args)
   u32 tc_index = va_arg (*args, u32);
   u32 __clib_unused thread_index = va_arg (*args, u32);
   u32 __clib_unused verbose = va_arg (*args, u32);
-  ct_connection_t *ct = ct_connection_get (tc_index);
+  ct_connection_t *ct = ct_connection_get (tc_index, 0);
   s = format (s, "%-" SESSION_CLI_ID_LEN "U", format_ct_connection_id, ct);
   if (verbose)
     s = format (s, "%-" SESSION_CLI_STATE_LEN "s", "LISTEN");
@@ -547,11 +574,11 @@ static u8 *
 format_ct_session (u8 * s, va_list * args)
 {
   u32 ct_index = va_arg (*args, u32);
-  u32 __clib_unused thread_index = va_arg (*args, u32);
+  u32 thread_index = va_arg (*args, u32);
   u32 verbose = va_arg (*args, u32);
   ct_connection_t *ct;
 
-  ct = ct_connection_get (ct_index);
+  ct = ct_connection_get (ct_index, thread_index);
   if (!ct)
     {
       s = format (s, "empty\n");
@@ -562,8 +589,17 @@ format_ct_session (u8 * s, va_list * args)
   return s;
 }
 
+clib_error_t *
+ct_enable_disable (vlib_main_t * vm, u8 is_en)
+{
+  ct_main.n_workers = vlib_num_workers ();
+  vec_validate (ct_main.connections, ct_main.n_workers);
+  return 0;
+}
+
 /* *INDENT-OFF* */
 static const transport_proto_vft_t cut_thru_proto = {
+  .enable = ct_enable_disable,
   .start_listen = ct_start_listen,
   .stop_listen = ct_stop_listen,
   .get_listener = ct_listener_get,
@@ -590,10 +626,10 @@ ct_session_tx (session_t * s)
   session_t *peer_s;
 
   ct = (ct_connection_t *) session_get_transport (s);
-  peer_ct = ct_connection_get (ct->peer_index);
+  peer_ct = ct_connection_get (ct->peer_index, ct->c_thread_index);
   if (!peer_ct)
     return 0;
-  peer_s = session_get (peer_ct->c_s_index, 0);
+  peer_s = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
   if (peer_s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
     return 0;
   session_enqueue_notify (peer_s);
diff --git a/src/vnet/session/segment_manager.c b/src/vnet/session/segment_manager.c
index e00a7616448..1d648f9e410 100644
--- a/src/vnet/session/segment_manager.c
+++ b/src/vnet/session/segment_manager.c
@@ -231,7 +231,7 @@ segment_manager_get_segment_if_valid (segment_manager_t * sm,
  * Removes segment after acquiring writer lock
  */
 static inline void
-segment_manager_lock_and_del_segment (segment_manager_t * sm, u32 fs_index)
+sm_lock_and_del_segment_inline (segment_manager_t * sm, u32 fs_index)
 {
   fifo_segment_t *fs;
   u8 is_prealloc;
@@ -252,6 +252,12 @@ done:
   clib_rwlock_writer_unlock (&sm->segments_rwlock);
 }
 
+void
+segment_manager_lock_and_del_segment (segment_manager_t * sm, u32 fs_index)
+{
+  sm_lock_and_del_segment_inline (sm, fs_index);
+}
+
 /**
  * Reads a segment from the segment manager's pool without lock
  */
@@ -774,7 +780,7 @@ segment_manager_dealloc_fifos (svm_fifo_t * rx_fifo, svm_fifo_t * tx_fifo)
 
       /* Remove segment if it holds no fifos or first but not protected */
       if (segment_index != 0 || !sm->first_is_protected)
-	segment_manager_lock_and_del_segment (sm, segment_index);
+	sm_lock_and_del_segment_inline (sm, segment_index);
 
       /* Remove segment manager if no sessions and detached from app */
       if (segment_manager_app_detached (sm)
diff --git a/src/vnet/session/segment_manager.h b/src/vnet/session/segment_manager.h
index f4dacdde849..1036ca2a99a 100644
--- a/src/vnet/session/segment_manager.h
+++ b/src/vnet/session/segment_manager.h
@@ -110,6 +110,8 @@ u32 segment_manager_index (segment_manager_t * sm);
 int segment_manager_add_segment (segment_manager_t * sm, uword segment_size);
 void segment_manager_del_segment (segment_manager_t * sm,
 				  fifo_segment_t * fs);
+void segment_manager_lock_and_del_segment (segment_manager_t * sm,
+					   u32 fs_index);
 fifo_segment_t *segment_manager_get_segment (segment_manager_t * sm,
 					     u32 segment_index);
 fifo_segment_t *segment_manager_get_segment_w_handle (u64 sh);
diff --git a/src/vnet/session/session_api.c b/src/vnet/session/session_api.c
index 12aea496b88..6a609eb92fa 100644
--- a/src/vnet/session/session_api.c
+++ b/src/vnet/session/session_api.c
@@ -180,7 +180,7 @@ mq_send_session_accepted_cb (session_t * s)
       mp->rmt.is_ip4 = session_type_is_ip4 (listener->session_type);
       mp->rmt.port = ct->c_rmt_port;
       mp->handle = session_handle (s);
-      vpp_queue = session_main_get_vpp_event_queue (0);
+      vpp_queue = session_main_get_vpp_event_queue (s->thread_index);
       mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
     }
   svm_msg_q_add_and_unlock (app_mq, msg);
@@ -318,7 +318,7 @@ mq_send_session_connected_cb (u32 app_wrk_index, u32 api_context,
       mp->handle = session_handle (s);
       mp->lcl.port = cct->c_lcl_port;
       mp->lcl.is_ip4 = cct->c_is_ip4;
-      vpp_mq = session_main_get_vpp_event_queue (0);
+      vpp_mq = session_main_get_vpp_event_queue (s->thread_index);
       mp->vpp_event_queue_address = pointer_to_uword (vpp_mq);
       mp->server_rx_fifo = pointer_to_uword (s->rx_fifo);
       mp->server_tx_fifo = pointer_to_uword (s->tx_fifo);