Cleanup/refactor session layer code
[vpp.git] / src / vnet / session / session.c
index d17c93f..48000a6 100644 (file)
  */
 
 #include <vnet/session/session.h>
+#include <vnet/session/session_debug.h>
+#include <vnet/session/application.h>
 #include <vlibmemory/api.h>
 #include <vnet/dpo/load_balance.h>
 #include <vnet/fib/ip4_fib.h>
-#include <vnet/session/application.h>
 #include <vnet/tcp/tcp.h>
-#include <vnet/session/session_debug.h>
-
-/**
- * Per-type vector of transport protocol virtual function tables
- */
-static transport_proto_vft_t *tp_vfts;
 
 session_manager_main_t session_manager_main;
-
-/*
- * Session lookup key; (src-ip, dst-ip, src-port, dst-port, session-type)
- * Value: (owner thread index << 32 | session_index);
- */
-void
-stream_session_table_add_for_tc (transport_connection_t * tc, u64 value)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  session_kv4_t kv4;
-  session_kv6_t kv6;
-
-  switch (tc->proto)
-    {
-    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 (&smm->v4_session_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 (&smm->v6_session_hash, &kv6, 1 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
-    }
-}
-
-void
-stream_session_table_add (session_manager_main_t * smm, stream_session_t * s,
-                         u64 value)
-{
-  transport_connection_t *tc;
-
-  tc = tp_vfts[s->session_type].get_connection (s->connection_index,
-                                               s->thread_index);
-  stream_session_table_add_for_tc (tc, value);
-}
-
-static void
-stream_session_half_open_table_add (session_type_t sst,
-                                   transport_connection_t * tc, u64 value)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  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 (&smm->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 (&smm->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_manager_main_t *smm = &session_manager_main;
-  session_kv4_t kv4;
-  session_kv6_t kv6;
-  switch (tc->proto)
-    {
-    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 (&smm->v4_session_hash, &kv4,
-                                      0 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
-      make_v6_ss_kv_from_tc (&kv6, tc);
-      return clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6,
-                                      0 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
-    }
-
-  return 0;
-}
-
-static int
-stream_session_table_del (session_manager_main_t * smm, stream_session_t * s)
-{
-  transport_connection_t *ts;
-
-  ts = tp_vfts[s->session_type].get_connection (s->connection_index,
-                                               s->thread_index);
-  return stream_session_table_del_for_tc (ts);
-}
-
-static void
-stream_session_half_open_table_del (session_manager_main_t * smm, u8 sst,
-                                   transport_connection_t * tc)
-{
-  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);
-      clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4,
-                               0 /* is_add */ );
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
-      make_v6_ss_kv_from_tc (&kv6, tc);
-      clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6,
-                               0 /* is_add */ );
-      break;
-    default:
-      clib_warning ("Session type not supported");
-      ASSERT (0);
-    }
-}
-
-stream_session_t *
-stream_session_lookup_listener4 (ip4_address_t * lcl, u16 lcl_port, u8 proto)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  session_kv4_t kv4;
-  int rv;
-
-  make_v4_listener_kv (&kv4, lcl, lcl_port, proto);
-  rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
-  if (rv == 0)
-    return pool_elt_at_index (smm->listen_sessions[proto], (u32) kv4.value);
-
-  /* Zero out the lcl ip */
-  kv4.key[0] = 0;
-  rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
-  if (rv == 0)
-    return pool_elt_at_index (smm->listen_sessions[proto], kv4.value);
-
-  return 0;
-}
-
-/** Looks up a session based on the 5-tuple passed as argument.
- *
- * First it tries to find an established session, if this fails, it tries
- * finding a listener session if this fails, it tries a lookup with a
- * wildcarded local source (listener bound to all interfaces)
- */
-stream_session_t *
-stream_session_lookup4 (ip4_address_t * lcl, ip4_address_t * rmt,
-                       u16 lcl_port, u16 rmt_port, u8 proto,
-                       u32 my_thread_index)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  session_kv4_t kv4;
-  int rv;
-
-  /* Lookup session amongst established ones */
-  make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
-  rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
-  if (rv == 0)
-    return stream_session_get_tsi (kv4.value, my_thread_index);
-
-  /* If nothing is found, check if any listener is available */
-  return stream_session_lookup_listener4 (lcl, lcl_port, proto);
-}
-
-stream_session_t *
-stream_session_lookup_listener6 (ip6_address_t * lcl, u16 lcl_port, u8 proto)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  session_kv6_t kv6;
-  int rv;
-
-  make_v6_listener_kv (&kv6, lcl, lcl_port, proto);
-  rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
-  if (rv == 0)
-    return pool_elt_at_index (smm->listen_sessions[proto], kv6.value);
-
-  /* Zero out the lcl ip */
-  kv6.key[0] = kv6.key[1] = 0;
-  rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
-  if (rv == 0)
-    return pool_elt_at_index (smm->listen_sessions[proto], kv6.value);
-
-  return 0;
-}
-
-/* Looks up a session based on the 5-tuple passed as argument.
- * First it tries to find an established session, if this fails, it tries
- * finding a listener session if this fails, it tries a lookup with a
- * wildcarded local source (listener bound to all interfaces) */
-stream_session_t *
-stream_session_lookup6 (ip6_address_t * lcl, ip6_address_t * rmt,
-                       u16 lcl_port, u16 rmt_port, u8 proto,
-                       u32 my_thread_index)
-{
-  session_manager_main_t *smm = vnet_get_session_manager_main ();
-  session_kv6_t kv6;
-  int rv;
-
-  make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
-  rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
-  if (rv == 0)
-    return stream_session_get_tsi (kv6.value, my_thread_index);
-
-  /* If nothing is found, check if any listener is available */
-  return stream_session_lookup_listener6 (lcl, lcl_port, proto);
-}
-
-stream_session_t *
-stream_session_lookup_listener (ip46_address_t * lcl, u16 lcl_port, u8 proto)
-{
-  switch (proto)
-    {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
-      return stream_session_lookup_listener4 (&lcl->ip4, lcl_port, proto);
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
-      return stream_session_lookup_listener6 (&lcl->ip6, lcl_port, proto);
-      break;
-    }
-  return 0;
-}
-
-static u64
-stream_session_half_open_lookup (session_manager_main_t * smm,
-                                ip46_address_t * lcl, ip46_address_t * rmt,
-                                u16 lcl_port, u16 rmt_port, u8 proto)
-{
-  session_kv4_t kv4;
-  session_kv6_t kv6;
-  int rv;
-
-  switch (proto)
-    {
-    case SESSION_TYPE_IP4_UDP:
-    case SESSION_TYPE_IP4_TCP:
-      make_v4_ss_kv (&kv4, &lcl->ip4, &rmt->ip4, lcl_port, rmt_port, proto);
-      rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
-
-      if (rv == 0)
-       return kv4.value;
-
-      return (u64) ~ 0;
-      break;
-    case SESSION_TYPE_IP6_UDP:
-    case SESSION_TYPE_IP6_TCP:
-      make_v6_ss_kv (&kv6, &lcl->ip6, &rmt->ip6, lcl_port, rmt_port, proto);
-      rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
-
-      if (rv == 0)
-       return kv6.value;
-
-      return (u64) ~ 0;
-      break;
-    }
-  return 0;
-}
-
-transport_connection_t *
-stream_session_lookup_transport4 (ip4_address_t * lcl, ip4_address_t * rmt,
-                                 u16 lcl_port, u16 rmt_port, u8 proto,
-                                 u32 my_thread_index)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  session_kv4_t kv4;
-  stream_session_t *s;
-  int rv;
-
-  /* Lookup session amongst established ones */
-  make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
-  rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
-  if (rv == 0)
-    {
-      s = stream_session_get_tsi (kv4.value, my_thread_index);
-
-      return tp_vfts[s->session_type].get_connection (s->connection_index,
-                                                     my_thread_index);
-    }
-
-  /* If nothing is found, check if any listener is available */
-  s = stream_session_lookup_listener4 (lcl, lcl_port, proto);
-  if (s)
-    return tp_vfts[s->session_type].get_listener (s->connection_index);
-
-  /* Finally, try half-open connections */
-  rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
-  if (rv == 0)
-    return tp_vfts[proto].get_half_open (kv4.value & 0xFFFFFFFF);
-
-  return 0;
-}
-
-transport_connection_t *
-stream_session_lookup_transport6 (ip6_address_t * lcl, ip6_address_t * rmt,
-                                 u16 lcl_port, u16 rmt_port, u8 proto,
-                                 u32 my_thread_index)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  stream_session_t *s;
-  session_kv6_t kv6;
-  int rv;
-
-  make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
-  rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
-  if (rv == 0)
-    {
-      s = stream_session_get_tsi (kv6.value, my_thread_index);
-
-      return tp_vfts[s->session_type].get_connection (s->connection_index,
-                                                     my_thread_index);
-    }
-
-  /* If nothing is found, check if any listener is available */
-  s = stream_session_lookup_listener6 (lcl, lcl_port, proto);
-  if (s)
-    return tp_vfts[s->session_type].get_listener (s->connection_index);
-
-  /* Finally, try half-open connections */
-  rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
-  if (rv == 0)
-    return tp_vfts[proto].get_half_open (kv6.value & 0xFFFFFFFF);
-
-  return 0;
-}
+extern transport_proto_vft_t *tp_vfts;
 
 int
 stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
@@ -390,13 +41,15 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
   u32 thread_index = tc->thread_index;
   int rv;
 
+  ASSERT (thread_index == vlib_get_thread_index ());
+
   if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo,
                                                 &server_tx_fifo,
                                                 &fifo_segment_index)))
     return rv;
 
   /* Create the session */
-  pool_get (smm->sessions[thread_index], s);
+  pool_get_aligned (smm->sessions[thread_index], s, CLIB_CACHE_LINE_BYTES);
   memset (s, 0, sizeof (*s));
 
   /* Initialize backpointers */
@@ -424,7 +77,7 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
   tc->s_index = s->session_index;
 
   /* Add to the main lookup table */
-  value = (((u64) thread_index) << 32) | (u64) s->session_index;
+  value = stream_session_handle (s);
   stream_session_table_add_for_tc (tc, value);
 
   *ret_s = s;
@@ -432,33 +85,97 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
   return 0;
 }
 
+/** Enqueue buffer chain tail */
+always_inline int
+session_enqueue_chain_tail (stream_session_t * s, vlib_buffer_t * b,
+                           u32 offset, u8 is_in_order)
+{
+  vlib_buffer_t *chain_b;
+  u32 chain_bi = b->next_buffer;
+  vlib_main_t *vm = vlib_get_main ();
+  u8 *data, len;
+  u16 written = 0;
+  int rv = 0;
+
+  do
+    {
+      chain_b = vlib_get_buffer (vm, chain_bi);
+      data = vlib_buffer_get_current (chain_b);
+      len = chain_b->current_length;
+      if (is_in_order)
+       {
+         rv = svm_fifo_enqueue_nowait (s->server_rx_fifo, len, data);
+         if (rv < len)
+           {
+             return (rv > 0) ? (written + rv) : written;
+           }
+         written += rv;
+       }
+      else
+       {
+         rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset, len,
+                                            data);
+         if (rv)
+           return -1;
+         offset += len;
+       }
+    }
+  while ((chain_bi = (chain_b->flags & VLIB_BUFFER_NEXT_PRESENT)
+         ? chain_b->next_buffer : 0));
+
+  if (is_in_order)
+    return written;
+
+  return 0;
+}
+
 /*
  * Enqueue data for delivery to session peer. Does not notify peer of enqueue
  * event but on request can queue notification events for later delivery by
  * calling stream_server_flush_enqueue_events().
  *
  * @param tc Transport connection which is to be enqueued data
- * @param data Data to be enqueued
- * @param len Length of data to be enqueued
+ * @param b Buffer to be enqueued
+ * @param offset Offset at which to start enqueueing if out-of-order
  * @param queue_event Flag to indicate if peer is to be notified or if event
  *                    is to be queued. The former is useful when more data is
  *                    enqueued and only one event is to be generated.
+ * @param is_in_order Flag to indicate if data is in order
  * @return Number of bytes enqueued or a negative value if enqueueing failed.
  */
 int
-stream_session_enqueue_data (transport_connection_t * tc, u8 * data, u16 len,
-                            u8 queue_event)
+stream_session_enqueue_data (transport_connection_t * tc, vlib_buffer_t * b,
+                            u32 offset, u8 queue_event, u8 is_in_order)
 {
   stream_session_t *s;
-  int enqueued;
+  int enqueued = 0, rv;
 
   s = stream_session_get (tc->s_index, tc->thread_index);
 
-  /* Make sure there's enough space left. We might've filled the pipes */
-  if (PREDICT_FALSE (len > svm_fifo_max_enqueue (s->server_rx_fifo)))
-    return -1;
-
-  enqueued = svm_fifo_enqueue_nowait (s->server_rx_fifo, len, data);
+  if (is_in_order)
+    {
+      enqueued =
+       svm_fifo_enqueue_nowait (s->server_rx_fifo, b->current_length,
+                                vlib_buffer_get_current (b));
+      if (PREDICT_FALSE
+         ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && enqueued > 0))
+       {
+         rv = session_enqueue_chain_tail (s, b, 0, 1);
+         if (rv <= 0)
+           return enqueued;
+         enqueued += rv;
+       }
+    }
+  else
+    {
+      rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset,
+                                        b->current_length,
+                                        vlib_buffer_get_current (b));
+      if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && !rv))
+       rv = session_enqueue_chain_tail (s, b, offset + b->current_length, 0);
+      if (rv)
+       return -1;
+    }
 
   if (queue_event)
     {
@@ -476,7 +193,10 @@ stream_session_enqueue_data (transport_connection_t * tc, u8 * data, u16 len,
        }
     }
 
-  return enqueued;
+  if (is_in_order)
+    return enqueued;
+
+  return 0;
 }
 
 /** Check if we have space in rx fifo to push more bytes */
@@ -484,7 +204,7 @@ u8
 stream_session_no_space (transport_connection_t * tc, u32 thread_index,
                         u16 data_len)
 {
-  stream_session_t *s = stream_session_get (tc->c_index, thread_index);
+  stream_session_t *s = stream_session_get (tc->s_index, thread_index);
 
   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY))
     return 1;
@@ -496,6 +216,15 @@ stream_session_no_space (transport_connection_t * tc, u32 thread_index,
 }
 
 u32
+stream_session_tx_fifo_max_dequeue (transport_connection_t * tc)
+{
+  stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
+  if (s->session_state != SESSION_STATE_READY)
+    return 0;
+  return svm_fifo_max_dequeue (s->server_tx_fifo);
+}
+
+int
 stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
                           u32 offset, u32 max_bytes)
 {
@@ -609,23 +338,38 @@ session_manager_flush_enqueue_events (u32 thread_index)
   return errors;
 }
 
+/**
+ * Init fifo tail and head pointers
+ *
+ * Useful if transport uses absolute offsets for tracking ooo segments.
+ */
 void
+stream_session_init_fifos_pointers (transport_connection_t * tc,
+                                   u32 rx_pointer, u32 tx_pointer)
+{
+  stream_session_t *s;
+  s = stream_session_get (tc->s_index, tc->thread_index);
+  svm_fifo_init_pointers (s->server_rx_fifo, rx_pointer);
+  svm_fifo_init_pointers (s->server_tx_fifo, tx_pointer);
+}
+
+int
 stream_session_connect_notify (transport_connection_t * tc, u8 sst,
                               u8 is_fail)
 {
-  session_manager_main_t *smm = &session_manager_main;
   application_t *app;
   stream_session_t *new_s = 0;
   u64 handle;
   u32 api_context = 0;
+  int error = 0;
 
-  handle = stream_session_half_open_lookup (smm, &tc->lcl_ip, &tc->rmt_ip,
-                                           tc->lcl_port, tc->rmt_port,
-                                           tc->proto);
+  handle = stream_session_half_open_lookup_handle (&tc->lcl_ip, &tc->rmt_ip,
+                                                  tc->lcl_port, tc->rmt_port,
+                                                  tc->proto);
   if (handle == HALF_OPEN_LOOKUP_INVALID_VALUE)
     {
       clib_warning ("This can't be good!");
-      return;
+      return -1;
     }
 
   /* Get the app's index from the handle we stored when opening connection */
@@ -639,17 +383,32 @@ stream_session_connect_notify (transport_connection_t * tc, u8 sst,
 
       /* Create new session (svm segments are allocated if needed) */
       if (stream_session_create_i (sm, tc, &new_s))
-       return;
-
-      new_s->app_index = app->index;
+       {
+         is_fail = 1;
+         error = -1;
+       }
+      else
+       new_s->app_index = app->index;
     }
 
   /* Notify client */
-  app->cb_fns.session_connected_callback (app->index, api_context, new_s,
-                                         is_fail);
+  if (app->cb_fns.session_connected_callback (app->index, api_context, new_s,
+                                             is_fail))
+    {
+      clib_warning ("failed to notify app");
+      if (!is_fail)
+       stream_session_disconnect (new_s);
+    }
+  else
+    {
+      if (!is_fail)
+       new_s->session_state = SESSION_STATE_READY;
+    }
 
   /* Cleanup session lookup */
-  stream_session_half_open_table_del (smm, sst, tc);
+  stream_session_half_open_table_del (sst, tc);
+
+  return error;
 }
 
 void
@@ -688,15 +447,19 @@ void
 stream_session_delete (stream_session_t * s)
 {
   session_manager_main_t *smm = vnet_get_session_manager_main ();
+  int rv;
 
   /* Delete from the main lookup table. */
-  stream_session_table_del (smm, s);
+  if ((rv = stream_session_table_del (s)))
+    clib_warning ("hash delete error, rv %d", rv);
 
   /* Cleanup fifo segments */
   segment_manager_dealloc_fifos (s->svm_segment_index, s->server_rx_fifo,
                                 s->server_tx_fifo);
 
   pool_put (smm->sessions[s->thread_index], s);
+  if (CLIB_DEBUG)
+    memset (s, 0xFA, sizeof (*s));
 }
 
 /**
@@ -757,6 +520,7 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index,
 
   s->app_index = server->index;
   s->listener_index = listener_index;
+  s->session_state = SESSION_STATE_ACCEPTING;
 
   /* Shoulder-tap the server */
   if (notify)
@@ -781,14 +545,14 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index,
  */
 int
 stream_session_open (u32 app_index, session_type_t st,
-                    transport_endpoint_t * tep,
+                    transport_endpoint_t * rmt,
                     transport_connection_t ** res)
 {
   transport_connection_t *tc;
   int rv;
   u64 handle;
 
-  rv = tp_vfts[st].open (&tep->ip, tep->port);
+  rv = tp_vfts[st].open (rmt);
   if (rv < 0)
     {
       clib_warning ("Transport failed to open connection.");
@@ -825,7 +589,7 @@ stream_session_listen (stream_session_t * s, transport_endpoint_t * tep)
   u32 tci;
 
   /* Transport bind/listen  */
-  tci = tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port);
+  tci = tp_vfts[s->session_type].bind (s->session_index, tep);
 
   if (tci == (u32) ~ 0)
     return -1;
@@ -890,8 +654,13 @@ session_send_session_evt_to_thread (u64 session_handle,
 
   /* Based on request block (or not) for lack of space */
   if (PREDICT_TRUE (q->cursize < q->maxsize))
-    unix_shared_memory_queue_add (q, (u8 *) & evt,
-                                 0 /* do wait for mutex */ );
+    {
+      if (unix_shared_memory_queue_add (q, (u8 *) & evt,
+                                       1 /* do wait for mutex */ ))
+       {
+         clib_warning ("failed to enqueue evt");
+       }
+    }
   else
     {
       clib_warning ("queue full");
@@ -922,41 +691,18 @@ stream_session_disconnect (stream_session_t * s)
 void
 stream_session_cleanup (stream_session_t * s)
 {
-  session_manager_main_t *smm = &session_manager_main;
   int rv;
 
   s->session_state = SESSION_STATE_CLOSED;
 
   /* Delete from the main lookup table to avoid more enqueues */
-  rv = stream_session_table_del (smm, s);
+  rv = stream_session_table_del (s);
   if (rv)
     clib_warning ("hash delete error, rv %d", rv);
 
   tp_vfts[s->session_type].cleanup (s->connection_index, s->thread_index);
 }
 
-void
-session_register_transport (u8 type, const transport_proto_vft_t * vft)
-{
-  session_manager_main_t *smm = vnet_get_session_manager_main ();
-
-  vec_validate (tp_vfts, type);
-  tp_vfts[type] = *vft;
-
-  /* If an offset function is provided, then peek instead of dequeue */
-  smm->session_tx_fns[type] =
-    (vft->tx_fifo_offset) ? session_tx_fifo_peek_and_snd :
-    session_tx_fifo_dequeue_and_snd;
-}
-
-transport_proto_vft_t *
-session_get_transport_vft (u8 type)
-{
-  if (type >= vec_len (tp_vfts))
-    return 0;
-  return &tp_vfts[type];
-}
-
 /**
  * Allocate vpp event queue (once) per worker thread
  */
@@ -966,24 +712,47 @@ session_vpp_event_queue_allocate (session_manager_main_t * smm,
 {
   api_main_t *am = &api_main;
   void *oldheap;
+  u32 event_queue_length = 2048;
 
   if (smm->vpp_event_queues[thread_index] == 0)
     {
       /* Allocate event fifo in the /vpe-api shared-memory segment */
       oldheap = svm_push_data_heap (am->vlib_rp);
 
+      if (smm->configured_event_queue_length)
+       event_queue_length = smm->configured_event_queue_length;
+
       smm->vpp_event_queues[thread_index] =
-       unix_shared_memory_queue_init (2048 /* nels $$$$ config */ ,
-                                      sizeof (session_fifo_event_t),
-                                      0 /* consumer pid */ ,
-                                      0
-                                      /* (do not) send signal when queue non-empty */
-       );
+       unix_shared_memory_queue_init
+       (event_queue_length,
+        sizeof (session_fifo_event_t), 0 /* consumer pid */ ,
+        0 /* (do not) send signal when queue non-empty */ );
 
       svm_pop_heap (oldheap);
     }
 }
 
+session_type_t
+session_type_from_proto_and_ip (transport_proto_t proto, u8 is_ip4)
+{
+  if (proto == TRANSPORT_PROTO_TCP)
+    {
+      if (is_ip4)
+       return SESSION_TYPE_IP4_TCP;
+      else
+       return SESSION_TYPE_IP6_TCP;
+    }
+  else
+    {
+      if (is_ip4)
+       return SESSION_TYPE_IP4_UDP;
+      else
+       return SESSION_TYPE_IP6_UDP;
+    }
+
+  return SESSION_N_TYPES;
+}
+
 static clib_error_t *
 session_manager_main_enable (vlib_main_t * vm)
 {
@@ -1005,11 +774,19 @@ session_manager_main_enable (vlib_main_t * vm)
   vec_validate (smm->sessions, num_threads - 1);
   vec_validate (smm->session_indices_to_enqueue_by_thread, num_threads - 1);
   vec_validate (smm->tx_buffers, num_threads - 1);
-  vec_validate (smm->fifo_events, num_threads - 1);
-  vec_validate (smm->evts_partially_read, num_threads - 1);
+  vec_validate (smm->pending_event_vector, num_threads - 1);
+  vec_validate (smm->free_event_vector, num_threads - 1);
   vec_validate (smm->current_enqueue_epoch, num_threads - 1);
   vec_validate (smm->vpp_event_queues, num_threads - 1);
 
+  for (i = 0; i < num_threads; i++)
+    {
+      vec_validate (smm->free_event_vector[i], 0);
+      _vec_len (smm->free_event_vector[i]) = 0;
+      vec_validate (smm->pending_event_vector[i], 0);
+      _vec_len (smm->pending_event_vector[i]) = 0;
+    }
+
 #if SESSION_DBG
   vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
 #endif
@@ -1019,29 +796,16 @@ session_manager_main_enable (vlib_main_t * vm)
     session_vpp_event_queue_allocate (smm, i);
 
   /* $$$$ preallocate hack config parameter */
-  for (i = 0; i < 200000; i++)
+  for (i = 0; i < smm->preallocated_sessions; i++)
     {
-      stream_session_t *ss;
-      pool_get (smm->sessions[0], ss);
-      memset (ss, 0, sizeof (*ss));
+      stream_session_t *ss __attribute__ ((unused));
+      pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES);
     }
 
-  for (i = 0; i < 200000; i++)
+  for (i = 0; i < smm->preallocated_sessions; i++)
     pool_put_index (smm->sessions[0], i);
 
-  clib_bihash_init_16_8 (&smm->v4_session_hash, "v4 session table",
-                        200000 /* $$$$ config parameter nbuckets */ ,
-                        (64 << 20) /*$$$ config parameter table size */ );
-  clib_bihash_init_48_8 (&smm->v6_session_hash, "v6 session table",
-                        200000 /* $$$$ config parameter nbuckets */ ,
-                        (64 << 20) /*$$$ config parameter table size */ );
-
-  clib_bihash_init_16_8 (&smm->v4_half_open_hash, "v4 half-open table",
-                        200000 /* $$$$ config parameter nbuckets */ ,
-                        (64 << 20) /*$$$ config parameter table size */ );
-  clib_bihash_init_48_8 (&smm->v6_half_open_hash, "v6 half-open table",
-                        200000 /* $$$$ config parameter nbuckets */ ,
-                        (64 << 20) /*$$$ config parameter table size */ );
+  session_lookup_init ();
 
   smm->is_enabled = 1;
 
@@ -1088,15 +852,39 @@ clib_error_t *
 session_manager_main_init (vlib_main_t * vm)
 {
   session_manager_main_t *smm = &session_manager_main;
-
-  smm->vlib_main = vm;
-  smm->vnet_main = vnet_get_main ();
   smm->is_enabled = 0;
+  return 0;
+}
+
+VLIB_INIT_FUNCTION (session_manager_main_init);
+
+static clib_error_t *
+session_config_fn (vlib_main_t * vm, unformat_input_t * input)
+{
+  session_manager_main_t *smm = &session_manager_main;
+  u32 nitems;
 
+  while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
+    {
+      if (unformat (input, "event-queue-length %d", &nitems))
+       {
+         if (nitems >= 2048)
+           smm->configured_event_queue_length = nitems;
+         else
+           clib_warning ("event queue length %d too small, ignored", nitems);
+       }
+      if (unformat (input, "preallocated-sessions %d",
+                   &smm->preallocated_sessions))
+       ;
+      else
+       return clib_error_return (0, "unknown input `%U'",
+                                 format_unformat_error, input);
+    }
   return 0;
 }
 
-VLIB_INIT_FUNCTION (session_manager_main_init)
+VLIB_CONFIG_FUNCTION (session_config_fn, "session");
+
 /*
  * fd.io coding-style-patch-verification: ON
  *