session: move connects to first worker
[vpp.git] / src / vnet / session / transport.c
index 192a201..8554c73 100644 (file)
 #include <vnet/session/session.h>
 #include <vnet/fib/fib.h>
 
-typedef struct local_endpoint_
-{
-  transport_endpoint_t ep;
-  int refcnt;
-} local_endpoint_t;
-
 /**
  * Per-type vector of transport protocol virtual function tables
  */
 transport_proto_vft_t *tp_vfts;
 
-/*
- * Port allocator seed
- */
-static u32 port_allocator_seed;
-
-/*
- * Local endpoints table
- */
-static transport_endpoint_table_t local_endpoints_table;
+typedef struct local_endpoint_
+{
+  transport_endpoint_t ep;
+  transport_proto_t proto;
+  int refcnt;
+} local_endpoint_t;
 
-/*
- * Pool of local endpoints
- */
-static local_endpoint_t *local_endpoints;
+typedef struct transport_main_
+{
+  transport_endpoint_table_t local_endpoints_table;
+  local_endpoint_t *local_endpoints;
+  u32 *lcl_endpts_freelist;
+  u32 port_allocator_seed;
+  u8 lcl_endpts_cleanup_pending;
+  clib_spinlock_t local_endpoints_lock;
+} transport_main_t;
 
-/*
- * Local endpoints pool lock
- */
-static clib_spinlock_t local_endpoints_lock;
+static transport_main_t tp_main;
 
 u8 *
 format_transport_proto (u8 * s, va_list * args)
@@ -426,52 +419,113 @@ transport_connection_attribute (transport_proto_t tp, u32 conn_index,
 void
 transport_endpoint_free (u32 tepi)
 {
-  pool_put_index (local_endpoints, tepi);
+  transport_main_t *tm = &tp_main;
+  pool_put_index (tm->local_endpoints, tepi);
 }
 
 always_inline local_endpoint_t *
 transport_endpoint_alloc (void)
 {
+  transport_main_t *tm = &tp_main;
   local_endpoint_t *lep;
 
   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
-  pool_get_aligned_safe (local_endpoints, lep, 0);
+
+  pool_get_aligned_safe (tm->local_endpoints, lep, 0);
   return lep;
 }
 
+static void
+transport_cleanup_freelist (void)
+{
+  transport_main_t *tm = &tp_main;
+  local_endpoint_t *lep;
+  u32 *lep_indexp;
+
+  clib_spinlock_lock (&tm->local_endpoints_lock);
+
+  vec_foreach (lep_indexp, tm->lcl_endpts_freelist)
+    {
+      lep = pool_elt_at_index (tm->local_endpoints, *lep_indexp);
+
+      /* Port re-shared after attempt to cleanup */
+      if (lep->refcnt > 0)
+       continue;
+
+      transport_endpoint_table_del (&tm->local_endpoints_table, lep->proto,
+                                   &lep->ep);
+      transport_endpoint_free (*lep_indexp);
+    }
+
+  vec_reset_length (tm->lcl_endpts_freelist);
+
+  tm->lcl_endpts_cleanup_pending = 0;
+
+  clib_spinlock_unlock (&tm->local_endpoints_lock);
+}
+
 void
-transport_endpoint_cleanup (u8 proto, ip46_address_t * lcl_ip, u16 port)
+transport_program_endpoint_cleanup (u32 lepi)
+{
+  transport_main_t *tm = &tp_main;
+  u8 flush_fl = 0;
+
+  /* All workers can free connections. Synchronize access to freelist */
+  clib_spinlock_lock (&tm->local_endpoints_lock);
+
+  vec_add1 (tm->lcl_endpts_freelist, lepi);
+
+  /* Avoid accumulating lots of endpoints for cleanup */
+  if (!tm->lcl_endpts_cleanup_pending &&
+      vec_len (tm->lcl_endpts_freelist) > 32)
+    {
+      tm->lcl_endpts_cleanup_pending = 1;
+      flush_fl = 1;
+    }
+
+  clib_spinlock_unlock (&tm->local_endpoints_lock);
+
+  if (flush_fl)
+    session_send_rpc_evt_to_thread_force (transport_cl_thread (),
+                                         transport_cleanup_freelist, 0);
+}
+
+int
+transport_release_local_endpoint (u8 proto, ip46_address_t *lcl_ip, u16 port)
 {
+  transport_main_t *tm = &tp_main;
   local_endpoint_t *lep;
   u32 lepi;
 
-  /* Cleanup local endpoint if this was an active connect */
-  lepi = transport_endpoint_lookup (&local_endpoints_table, proto, lcl_ip,
+  lepi = transport_endpoint_lookup (&tm->local_endpoints_table, proto, lcl_ip,
                                    clib_net_to_host_u16 (port));
   if (lepi == ENDPOINT_INVALID_INDEX)
-    return;
+    return -1;
+
+  lep = pool_elt_at_index (tm->local_endpoints, lepi);
 
-  lep = pool_elt_at_index (local_endpoints, lepi);
+  /* Local endpoint no longer in use, program cleanup */
   if (!clib_atomic_sub_fetch (&lep->refcnt, 1))
     {
-      transport_endpoint_table_del (&local_endpoints_table, proto, &lep->ep);
-
-      /* All workers can free connections. Synchronize access to pool */
-      clib_spinlock_lock (&local_endpoints_lock);
-      transport_endpoint_free (lepi);
-      clib_spinlock_unlock (&local_endpoints_lock);
+      transport_program_endpoint_cleanup (lepi);
+      return 0;
     }
+
+  /* Not an error, just in idication that endpoint was not cleaned up */
+  return -1;
 }
 
 static int
 transport_endpoint_mark_used (u8 proto, ip46_address_t *ip, u16 port)
 {
+  transport_main_t *tm = &tp_main;
   local_endpoint_t *lep;
   u32 tei;
 
   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
 
-  tei = transport_endpoint_lookup (&local_endpoints_table, proto, ip, port);
+  tei =
+    transport_endpoint_lookup (&tm->local_endpoints_table, proto, ip, port);
   if (tei != ENDPOINT_INVALID_INDEX)
     return SESSION_E_PORTINUSE;
 
@@ -479,10 +533,11 @@ transport_endpoint_mark_used (u8 proto, ip46_address_t *ip, u16 port)
   lep = transport_endpoint_alloc ();
   clib_memcpy_fast (&lep->ep.ip, ip, sizeof (*ip));
   lep->ep.port = port;
+  lep->proto = proto;
   lep->refcnt = 1;
 
-  transport_endpoint_table_add (&local_endpoints_table, proto, &lep->ep,
-                               lep - local_endpoints);
+  transport_endpoint_table_add (&tm->local_endpoints_table, proto, &lep->ep,
+                               lep - tm->local_endpoints);
 
   return 0;
 }
@@ -490,14 +545,18 @@ transport_endpoint_mark_used (u8 proto, ip46_address_t *ip, u16 port)
 void
 transport_share_local_endpoint (u8 proto, ip46_address_t * lcl_ip, u16 port)
 {
+  transport_main_t *tm = &tp_main;
   local_endpoint_t *lep;
   u32 lepi;
 
-  lepi = transport_endpoint_lookup (&local_endpoints_table, proto, lcl_ip,
+  /* Active opens should call this only from a control thread, which are also
+   * used to allocate and free ports. So, pool has only one writer and
+   * potentially many readers. Listeners are allocated with barrier */
+  lepi = transport_endpoint_lookup (&tm->local_endpoints_table, proto, lcl_ip,
                                    clib_net_to_host_u16 (port));
   if (lepi != ENDPOINT_INVALID_INDEX)
     {
-      lep = pool_elt_at_index (local_endpoints, lepi);
+      lep = pool_elt_at_index (tm->local_endpoints, lepi);
       clib_atomic_add_fetch (&lep->refcnt, 1);
     }
 }
@@ -510,6 +569,7 @@ int
 transport_alloc_local_port (u8 proto, ip46_address_t * ip)
 {
   u16 min = 1024, max = 65535; /* XXX configurable ? */
+  transport_main_t *tm = &tp_main;
   int tries, limit;
 
   limit = max - min;
@@ -517,6 +577,10 @@ transport_alloc_local_port (u8 proto, ip46_address_t * ip)
   /* Only support active opens from one of ctrl threads */
   ASSERT (vlib_get_thread_index () <= transport_cl_thread ());
 
+  /* Cleanup freelist if need be */
+  if (vec_len (tm->lcl_endpts_freelist))
+    transport_cleanup_freelist ();
+
   /* Search for first free slot */
   for (tries = 0; tries < limit; tries++)
     {
@@ -525,7 +589,7 @@ transport_alloc_local_port (u8 proto, ip46_address_t * ip)
       /* Find a port in the specified range */
       while (1)
        {
-         port = random_u32 (&port_allocator_seed) & PORT_MASK;
+         port = random_u32 (&tm->port_allocator_seed) & PORT_MASK;
          if (PREDICT_TRUE (port >= min && port < max))
            break;
        }
@@ -846,6 +910,7 @@ transport_init (void)
 {
   vlib_thread_main_t *vtm = vlib_get_thread_main ();
   session_main_t *smm = vnet_get_session_main ();
+  transport_main_t *tm = &tp_main;
   u32 num_threads;
 
   if (smm->local_endpoints_table_buckets == 0)
@@ -854,12 +919,12 @@ transport_init (void)
     smm->local_endpoints_table_memory = 512 << 20;
 
   /* Initialize [port-allocator] random number seed */
-  port_allocator_seed = (u32) clib_cpu_time_now ();
+  tm->port_allocator_seed = (u32) clib_cpu_time_now ();
 
-  clib_bihash_init_24_8 (&local_endpoints_table, "local endpoints table",
+  clib_bihash_init_24_8 (&tm->local_endpoints_table, "local endpoints table",
                         smm->local_endpoints_table_buckets,
                         smm->local_endpoints_table_memory);
-  clib_spinlock_init (&local_endpoints_lock);
+  clib_spinlock_init (&tm->local_endpoints_lock);
 
   num_threads = 1 /* main thread */  + vtm->n_threads;
   if (num_threads > 1)