X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession.c;h=c5aaf2e2b15a1385a3021c789c947a0b45276c99;hb=acd2a6a32cebab70a054e9335e446798538d0524;hp=8e2b2616f580137ee70f3a3aee9fb4215fc9731a;hpb=3e350af5d3e9744a4529a28dd293b2d4601442f7;p=vpp.git

diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 8e2b2616f58..c5aaf2e2b15 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -36,15 +36,14 @@ 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);
  */
-static void
-stream_session_table_add_for_tc (u8 sst, transport_connection_t * tc,
-				 u64 value)
+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 (sst)
+  switch (tc->proto)
     {
     case SESSION_TYPE_IP4_UDP:
     case SESSION_TYPE_IP4_TCP:
@@ -72,12 +71,12 @@ stream_session_table_add (session_manager_main_t * smm, stream_session_t * s,
 
   tc = tp_vfts[s->session_type].get_connection (s->connection_index,
 						s->thread_index);
-  stream_session_table_add_for_tc (s->session_type, tc, value);
+  stream_session_table_add_for_tc (tc, value);
 }
 
 static void
-stream_session_half_open_table_add (u8 sst, transport_connection_t * tc,
-				    u64 value)
+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;
@@ -105,14 +104,13 @@ stream_session_half_open_table_add (u8 sst, transport_connection_t * tc,
     }
 }
 
-static int
-stream_session_table_del_for_tc (session_manager_main_t * smm, u8 sst,
-				 transport_connection_t * tc)
+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 (sst)
+  switch (tc->proto)
     {
     case SESSION_TYPE_IP4_UDP:
     case SESSION_TYPE_IP4_TCP:
@@ -141,7 +139,7 @@ stream_session_table_del (session_manager_main_t * smm, stream_session_t * s)
 
   ts = tp_vfts[s->session_type].get_connection (s->connection_index,
 						s->thread_index);
-  return stream_session_table_del_for_tc (smm, s->session_type, ts);
+  return stream_session_table_del_for_tc (ts);
 }
 
 static void
@@ -379,313 +377,43 @@ stream_session_lookup_transport6 (ip6_address_t * lcl, ip6_address_t * rmt,
   return 0;
 }
 
-/**
- * Allocate vpp event queue (once) per worker thread
- */
-void
-vpp_session_event_queue_allocate (session_manager_main_t * smm,
-				  u32 thread_index)
-{
-  api_main_t *am = &api_main;
-  void *oldheap;
-
-  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);
-
-      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 */
-	);
-
-      svm_pop_heap (oldheap);
-    }
-}
-
-void
-session_manager_get_segment_info (u32 index, u8 ** name, u32 * size)
-{
-  svm_fifo_segment_private_t *s;
-  s = svm_fifo_get_segment (index);
-  *name = s->h->segment_name;
-  *size = s->ssvm.ssvm_size;
-}
-
-always_inline int
-session_manager_add_segment_i (session_manager_main_t * smm,
-			       session_manager_t * sm,
-			       u32 segment_size, u8 * segment_name)
-{
-  svm_fifo_segment_create_args_t _ca, *ca = &_ca;
-  int rv;
-
-  memset (ca, 0, sizeof (*ca));
-
-  ca->segment_name = (char *) segment_name;
-  ca->segment_size = segment_size;
-
-  rv = svm_fifo_segment_create (ca);
-  if (rv)
-    {
-      clib_warning ("svm_fifo_segment_create ('%s', %d) failed",
-		    ca->segment_name, ca->segment_size);
-      vec_free (segment_name);
-      return -1;
-    }
-
-  vec_add1 (sm->segment_indices, ca->new_segment_index);
-
-  return 0;
-}
-
-static int
-session_manager_add_segment (session_manager_main_t * smm,
-			     session_manager_t * sm)
-{
-  u8 *segment_name;
-  svm_fifo_segment_create_args_t _ca, *ca = &_ca;
-  u32 add_segment_size;
-  u32 default_segment_size = 128 << 10;
-
-  memset (ca, 0, sizeof (*ca));
-  segment_name = format (0, "%d-%d%c", getpid (),
-			 smm->unique_segment_name_counter++, 0);
-  add_segment_size =
-    sm->add_segment_size ? sm->add_segment_size : default_segment_size;
-
-  return session_manager_add_segment_i (smm, sm, add_segment_size,
-					segment_name);
-}
-
-int
-session_manager_add_first_segment (session_manager_main_t * smm,
-				   session_manager_t * sm, u32 segment_size,
-				   u8 ** segment_name)
-{
-  svm_fifo_segment_create_args_t _ca, *ca = &_ca;
-  memset (ca, 0, sizeof (*ca));
-  *segment_name = format (0, "%d-%d%c", getpid (),
-			  smm->unique_segment_name_counter++, 0);
-  return session_manager_add_segment_i (smm, sm, segment_size, *segment_name);
-}
-
-void
-session_manager_del (session_manager_main_t * smm, session_manager_t * sm)
-{
-  u32 *deleted_sessions = 0;
-  u32 *deleted_thread_indices = 0;
-  int i, j;
-
-  /* Across all fifo segments used by the server */
-  for (j = 0; j < vec_len (sm->segment_indices); j++)
-    {
-      svm_fifo_segment_private_t *fifo_segment;
-      svm_fifo_t **fifos;
-      /* Vector of fifos allocated in the segment */
-      fifo_segment = svm_fifo_get_segment (sm->segment_indices[j]);
-      fifos = (svm_fifo_t **) fifo_segment->h->fifos;
-
-      /*
-       * Remove any residual sessions from the session lookup table
-       * Don't bother deleting the individual fifos, we're going to
-       * throw away the fifo segment in a minute.
-       */
-      for (i = 0; i < vec_len (fifos); i++)
-	{
-	  svm_fifo_t *fifo;
-	  u32 session_index, thread_index;
-	  stream_session_t *session;
-
-	  fifo = fifos[i];
-	  session_index = fifo->server_session_index;
-	  thread_index = fifo->server_thread_index;
-
-	  session = pool_elt_at_index (smm->sessions[thread_index],
-				       session_index);
-
-	  /* Add to the deleted_sessions vector (once!) */
-	  if (!session->is_deleted)
-	    {
-	      session->is_deleted = 1;
-	      vec_add1 (deleted_sessions,
-			session - smm->sessions[thread_index]);
-	      vec_add1 (deleted_thread_indices, thread_index);
-	    }
-	}
-
-      for (i = 0; i < vec_len (deleted_sessions); i++)
-	{
-	  stream_session_t *session;
-
-	  session =
-	    pool_elt_at_index (smm->sessions[deleted_thread_indices[i]],
-			       deleted_sessions[i]);
-
-	  /* Instead of directly removing the session call disconnect */
-	  stream_session_disconnect (session);
-
-	  /*
-	     stream_session_table_del (smm, session);
-	     pool_put(smm->sessions[deleted_thread_indices[i]], session);
-	   */
-	}
-
-      vec_reset_length (deleted_sessions);
-      vec_reset_length (deleted_thread_indices);
-
-      /* Instead of removing the segment, test when removing the session if
-       * the segment can be removed
-       */
-      /* svm_fifo_segment_delete (fifo_segment); */
-    }
-
-  vec_free (deleted_sessions);
-  vec_free (deleted_thread_indices);
-}
-
-int
-session_manager_allocate_session_fifos (session_manager_main_t * smm,
-					session_manager_t * sm,
-					svm_fifo_t ** server_rx_fifo,
-					svm_fifo_t ** server_tx_fifo,
-					u32 * fifo_segment_index,
-					u8 * added_a_segment)
-{
-  svm_fifo_segment_private_t *fifo_segment;
-  u32 fifo_size, default_fifo_size = 1 << 16;	/* TODO config */
-  int i;
-
-  *added_a_segment = 0;
-
-  /* Allocate svm fifos */
-  ASSERT (vec_len (sm->segment_indices));
-
-again:
-  for (i = 0; i < vec_len (sm->segment_indices); i++)
-    {
-      *fifo_segment_index = sm->segment_indices[i];
-      fifo_segment = svm_fifo_get_segment (*fifo_segment_index);
-
-      fifo_size = sm->rx_fifo_size;
-      fifo_size = (fifo_size == 0) ? default_fifo_size : fifo_size;
-      *server_rx_fifo = svm_fifo_segment_alloc_fifo (fifo_segment, fifo_size);
-
-      fifo_size = sm->tx_fifo_size;
-      fifo_size = (fifo_size == 0) ? default_fifo_size : fifo_size;
-      *server_tx_fifo = svm_fifo_segment_alloc_fifo (fifo_segment, fifo_size);
-
-      if (*server_rx_fifo == 0)
-	{
-	  /* This would be very odd, but handle it... */
-	  if (*server_tx_fifo != 0)
-	    {
-	      svm_fifo_segment_free_fifo (fifo_segment, *server_tx_fifo);
-	      *server_tx_fifo = 0;
-	    }
-	  continue;
-	}
-      if (*server_tx_fifo == 0)
-	{
-	  if (*server_rx_fifo != 0)
-	    {
-	      svm_fifo_segment_free_fifo (fifo_segment, *server_rx_fifo);
-	      *server_rx_fifo = 0;
-	    }
-	  continue;
-	}
-      break;
-    }
-
-  /* See if we're supposed to create another segment */
-  if (*server_rx_fifo == 0)
-    {
-      if (sm->add_segment)
-	{
-	  if (*added_a_segment)
-	    {
-	      clib_warning ("added a segment, still cant allocate a fifo");
-	      return SESSION_ERROR_NEW_SEG_NO_SPACE;
-	    }
-
-	  if (session_manager_add_segment (smm, sm))
-	    return VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
-
-	  *added_a_segment = 1;
-	  goto again;
-	}
-      else
-	{
-	  clib_warning ("No space to allocate fifos!");
-	  return SESSION_ERROR_NO_SPACE;
-	}
-    }
-  return 0;
-}
-
 int
-stream_session_create_i (session_manager_main_t * smm, application_t * app,
-			 transport_connection_t * tc,
+stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc,
 			 stream_session_t ** ret_s)
 {
-  int rv;
+  session_manager_main_t *smm = &session_manager_main;
   svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
   u32 fifo_segment_index;
-  u32 pool_index, seg_size;
+  u32 pool_index;
   stream_session_t *s;
   u64 value;
   u32 thread_index = tc->thread_index;
-  session_manager_t *sm;
-  u8 segment_added;
-  u8 *seg_name;
-
-  sm = session_manager_get (app->session_manager_index);
-
-  /* Check the API queue */
-  if (app->mode == APP_SERVER && application_api_queue_is_full (app))
-    return SESSION_ERROR_API_QUEUE_FULL;
+  int rv;
 
-  if ((rv = session_manager_allocate_session_fifos (smm, sm, &server_rx_fifo,
-						    &server_tx_fifo,
-						    &fifo_segment_index,
-						    &segment_added)))
+  if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo,
+						 &server_tx_fifo,
+						 &fifo_segment_index)))
     return rv;
 
-  if (segment_added && app->mode == APP_SERVER)
-    {
-      /* Send an API message to the external server, to map new segment */
-      ASSERT (app->cb_fns.add_segment_callback);
-
-      session_manager_get_segment_info (fifo_segment_index, &seg_name,
-					&seg_size);
-      if (app->cb_fns.add_segment_callback (app->api_client_index, seg_name,
-					    seg_size))
-	return VNET_API_ERROR_URI_FIFO_CREATE_FAILED;
-    }
-
   /* 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 */
   pool_index = s - smm->sessions[thread_index];
-  server_rx_fifo->server_session_index = pool_index;
-  server_rx_fifo->server_thread_index = thread_index;
+  server_rx_fifo->master_session_index = pool_index;
+  server_rx_fifo->master_thread_index = thread_index;
 
-  server_tx_fifo->server_session_index = pool_index;
-  server_tx_fifo->server_thread_index = thread_index;
+  server_tx_fifo->master_session_index = pool_index;
+  server_tx_fifo->master_thread_index = thread_index;
 
   s->server_rx_fifo = server_rx_fifo;
   s->server_tx_fifo = server_tx_fifo;
 
   /* Initialize state machine, such as it is... */
-  s->session_type = app->session_type;
+  s->session_type = tc->proto;
   s->session_state = SESSION_STATE_CONNECTING;
-  s->app_index = application_get_index (app);
-  s->server_segment_index = fifo_segment_index;
+  s->svm_segment_index = fifo_segment_index;
   s->thread_index = thread_index;
   s->session_index = pool_index;
 
@@ -697,40 +425,104 @@ stream_session_create_i (session_manager_main_t * smm, application_t * app,
 
   /* Add to the main lookup table */
   value = (((u64) thread_index) << 32) | (u64) s->session_index;
-  stream_session_table_add_for_tc (app->session_type, tc, value);
+  stream_session_table_add_for_tc (tc, value);
 
   *ret_s = s;
 
   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, s->pid, 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)
     {
@@ -748,7 +540,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 */
@@ -772,14 +567,14 @@ stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
 			   u32 offset, u32 max_bytes)
 {
   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
-  return svm_fifo_peek (s->server_tx_fifo, s->pid, offset, max_bytes, buffer);
+  return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
 }
 
 u32
 stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
 {
   stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index);
-  return svm_fifo_dequeue_drop (s->server_tx_fifo, s->pid, max_bytes);
+  return svm_fifo_dequeue_drop (s->server_tx_fifo, max_bytes);
 }
 
 /**
@@ -813,7 +608,7 @@ stream_session_enqueue_notify (stream_session_t * s, u8 block)
     {
       /* Fabricate event */
       evt.fifo = s->server_rx_fifo;
-      evt.event_type = FIFO_EVENT_SERVER_RX;
+      evt.event_type = FIFO_EVENT_APP_RX;
       evt.event_id = serial_number++;
 
       /* Add event to server's event queue */
@@ -881,92 +676,19 @@ session_manager_flush_enqueue_events (u32 thread_index)
   return errors;
 }
 
-/*
- * Start listening on server's ip/port pair for requested transport.
+/**
+ * Init fifo tail and head pointers
  *
- * Creates a 'dummy' stream session with state LISTENING to be used in session
- * lookups, prior to establishing connection. Requests transport to build
- * it's own specific listening connection.
+ * Useful if transport uses absolute offsets for tracking ooo segments.
  */
-int
-stream_session_start_listen (u32 server_index, ip46_address_t * ip, u16 port)
-{
-  session_manager_main_t *smm = &session_manager_main;
-  stream_session_t *s;
-  transport_connection_t *tc;
-  application_t *srv;
-  u32 tci;
-
-  srv = application_get (server_index);
-
-  pool_get (smm->listen_sessions[srv->session_type], s);
-  memset (s, 0, sizeof (*s));
-
-  s->session_type = srv->session_type;
-  s->session_state = SESSION_STATE_LISTENING;
-  s->session_index = s - smm->listen_sessions[srv->session_type];
-  s->app_index = srv->index;
-
-  /* Transport bind/listen  */
-  tci = tp_vfts[srv->session_type].bind (s->session_index, ip, port);
-
-  /* Attach transport to session */
-  s->connection_index = tci;
-  tc = tp_vfts[srv->session_type].get_listener (tci);
-
-  srv->session_index = s->session_index;
-
-  /* Add to the main lookup table */
-  stream_session_table_add_for_tc (s->session_type, tc, s->session_index);
-
-  return 0;
-}
-
 void
-stream_session_stop_listen (u32 server_index)
+stream_session_init_fifos_pointers (transport_connection_t * tc,
+				    u32 rx_pointer, u32 tx_pointer)
 {
-  session_manager_main_t *smm = &session_manager_main;
-  stream_session_t *listener;
-  transport_connection_t *tc;
-  application_t *srv;
-
-  srv = application_get (server_index);
-  listener = pool_elt_at_index (smm->listen_sessions[srv->session_type],
-				srv->session_index);
-
-  tc = tp_vfts[srv->session_type].get_listener (listener->connection_index);
-  stream_session_table_del_for_tc (smm, listener->session_type, tc);
-
-  tp_vfts[srv->session_type].unbind (listener->connection_index);
-  pool_put (smm->listen_sessions[srv->session_type], listener);
-}
-
-int
-connect_server_add_segment_cb (application_t * ss, char *segment_name,
-			       u32 segment_size)
-{
-  /* Does exactly nothing, but die */
-  ASSERT (0);
-  return 0;
-}
-
-void
-connects_session_manager_init (session_manager_main_t * smm, u8 session_type)
-{
-  session_manager_t *sm;
-  u32 connect_fifo_size = 256 << 10;	/* Config? */
-  u32 default_segment_size = 1 << 20;
-
-  pool_get (smm->session_managers, sm);
-  memset (sm, 0, sizeof (*sm));
-
-  sm->add_segment_size = default_segment_size;
-  sm->rx_fifo_size = connect_fifo_size;
-  sm->tx_fifo_size = connect_fifo_size;
-  sm->add_segment = 1;
-
-  session_manager_add_segment (smm, sm);
-  smm->connect_manager_index[session_type] = sm - smm->session_managers;
+  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);
 }
 
 void
@@ -976,34 +698,36 @@ stream_session_connect_notify (transport_connection_t * tc, u8 sst,
   session_manager_main_t *smm = &session_manager_main;
   application_t *app;
   stream_session_t *new_s = 0;
-  u64 value;
+  u64 handle;
+  u32 api_context = 0;
 
-  value = stream_session_half_open_lookup (smm, &tc->lcl_ip, &tc->rmt_ip,
-					   tc->lcl_port, tc->rmt_port,
-					   tc->proto);
-  if (value == HALF_OPEN_LOOKUP_INVALID_VALUE)
+  handle = stream_session_half_open_lookup (smm, &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;
     }
 
-  app = application_get (value >> 32);
+  /* Get the app's index from the handle we stored when opening connection */
+  app = application_get (handle >> 32);
+  api_context = tc->s_index;
 
   if (!is_fail)
     {
-      /* Create new session (server segments are allocated if needed) */
-      if (stream_session_create_i (smm, app, tc, &new_s))
-	return;
+      segment_manager_t *sm;
+      sm = application_get_connect_segment_manager (app);
 
-      app->session_index = stream_session_get_index (new_s);
-      app->thread_index = new_s->thread_index;
+      /* Create new session (svm segments are allocated if needed) */
+      if (stream_session_create_i (sm, tc, &new_s))
+	return;
 
-      /* Allocate vpp event queue for this thread if needed */
-      vpp_session_event_queue_allocate (smm, tc->thread_index);
+      new_s->app_index = app->index;
     }
 
   /* Notify client */
-  app->cb_fns.session_connected_callback (app->api_client_index, new_s,
+  app->cb_fns.session_connected_callback (app->index, api_context, new_s,
 					  is_fail);
 
   /* Cleanup session lookup */
@@ -1046,48 +770,13 @@ void
 stream_session_delete (stream_session_t * s)
 {
   session_manager_main_t *smm = vnet_get_session_manager_main ();
-  svm_fifo_segment_private_t *fifo_segment;
-  application_t *app;
 
   /* Delete from the main lookup table. */
   stream_session_table_del (smm, s);
 
   /* Cleanup fifo segments */
-  fifo_segment = svm_fifo_get_segment (s->server_segment_index);
-  svm_fifo_segment_free_fifo (fifo_segment, s->server_rx_fifo);
-  svm_fifo_segment_free_fifo (fifo_segment, s->server_tx_fifo);
-
-  app = application_get_if_valid (s->app_index);
-
-  /* No app. A possibility: after disconnect application called unbind */
-  if (!app)
-    return;
-
-  if (app->mode == APP_CLIENT)
-    {
-      /* Cleanup app if client */
-      application_del (app);
-    }
-  else if (app->mode == APP_SERVER)
-    {
-      session_manager_t *sm;
-      svm_fifo_segment_private_t *fifo_segment;
-      svm_fifo_t **fifos;
-      u32 fifo_index;
-
-      /* For server, see if any segments can be removed */
-      sm = session_manager_get (app->session_manager_index);
-
-      /* Delete fifo */
-      fifo_segment = svm_fifo_get_segment (s->server_segment_index);
-      fifos = (svm_fifo_t **) fifo_segment->h->fifos;
-
-      fifo_index = svm_fifo_segment_index (fifo_segment);
-
-      /* Remove segment only if it holds no fifos and not the first */
-      if (sm->segment_indices[0] != fifo_index && vec_len (fifos) == 0)
-	svm_fifo_segment_delete (fifo_segment);
-    }
+  segment_manager_dealloc_fifos (s->svm_segment_index, s->server_rx_fifo,
+				 s->server_tx_fifo);
 
   pool_put (smm->sessions[s->thread_index], s);
 }
@@ -1134,21 +823,22 @@ int
 stream_session_accept (transport_connection_t * tc, u32 listener_index,
 		       u8 sst, u8 notify)
 {
-  session_manager_main_t *smm = &session_manager_main;
   application_t *server;
   stream_session_t *s, *listener;
+  segment_manager_t *sm;
 
   int rv;
 
   /* Find the server */
-  listener = pool_elt_at_index (smm->listen_sessions[sst], listener_index);
+  listener = listen_session_get (sst, listener_index);
   server = application_get (listener->app_index);
 
-  if ((rv = stream_session_create_i (smm, server, tc, &s)))
+  sm = application_get_listen_segment_manager (server, listener);
+  if ((rv = stream_session_create_i (sm, tc, &s)))
     return rv;
 
-  /* Allocate vpp event queue for this thread if needed */
-  vpp_session_event_queue_allocate (smm, tc->thread_index);
+  s->app_index = server->index;
+  s->listener_index = listener_index;
 
   /* Shoulder-tap the server */
   if (notify)
@@ -1159,68 +849,150 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index,
   return 0;
 }
 
+/**
+ * Ask transport to open connection to remote transport endpoint.
+ *
+ * Stores handle for matching request with reply since the call can be
+ * asynchronous. For instance, for TCP the 3-way handshake must complete
+ * before reply comes. Session is only created once connection is established.
+ *
+ * @param app_index Index of the application requesting the connect
+ * @param st Session type requested.
+ * @param tep Remote transport endpoint
+ * @param res Resulting transport connection .
+ */
 int
-stream_session_open (u8 sst, ip46_address_t * addr, u16 port_host_byte_order,
-		     u32 app_index)
+stream_session_open (u32 app_index, session_type_t st,
+		     transport_endpoint_t * tep,
+		     transport_connection_t ** res)
 {
   transport_connection_t *tc;
-  u32 tci;
-  u64 value;
   int rv;
+  u64 handle;
 
-  /* Ask transport to open connection */
-  rv = tp_vfts[sst].open (addr, port_host_byte_order);
+  rv = tp_vfts[st].open (&tep->ip, tep->port);
   if (rv < 0)
     {
       clib_warning ("Transport failed to open connection.");
       return VNET_API_ERROR_SESSION_CONNECT_FAIL;
     }
 
-  tci = rv;
-
-  /* Get transport connection */
-  tc = tp_vfts[sst].get_half_open (tci);
+  tc = tp_vfts[st].get_half_open ((u32) rv);
 
-  /* Store api_client_index and transport connection index */
-  value = (((u64) app_index) << 32) | (u64) tc->c_index;
+  /* Save app and tc index. The latter is needed to help establish the
+   * connection while the former is needed when the connect notify comes
+   * and we have to notify the external app */
+  handle = (((u64) app_index) << 32) | (u64) tc->c_index;
 
   /* Add to the half-open lookup table */
-  stream_session_half_open_table_add (sst, tc, value);
+  stream_session_half_open_table_add (st, tc, handle);
+
+  *res = tc;
 
   return 0;
 }
 
+/**
+ * Ask transport to listen on local transport endpoint.
+ *
+ * @param s Session for which listen will be called. Note that unlike
+ * 	    established sessions, listen sessions are not associated to a
+ * 	    thread.
+ * @param tep Local endpoint to be listened on.
+ */
+int
+stream_session_listen (stream_session_t * s, transport_endpoint_t * tep)
+{
+  transport_connection_t *tc;
+  u32 tci;
+
+  /* Transport bind/listen  */
+  tci = tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port);
+
+  if (tci == (u32) ~ 0)
+    return -1;
+
+  /* Attach transport to session */
+  s->connection_index = tci;
+  tc = tp_vfts[s->session_type].get_listener (tci);
+
+  /* Weird but handle it ... */
+  if (tc == 0)
+    return -1;
+
+  /* Add to the main lookup table */
+  stream_session_table_add_for_tc (tc, s->session_index);
+
+  return 0;
+}
+
+/**
+ * Ask transport to stop listening on local transport endpoint.
+ *
+ * @param s Session to stop listening on. It must be in state LISTENING.
+ */
+int
+stream_session_stop_listen (stream_session_t * s)
+{
+  transport_connection_t *tc;
+
+  if (s->session_state != SESSION_STATE_LISTENING)
+    {
+      clib_warning ("not a listening session");
+      return -1;
+    }
+
+  tc = tp_vfts[s->session_type].get_listener (s->connection_index);
+  if (!tc)
+    {
+      clib_warning ("no transport");
+      return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
+    }
+
+  stream_session_table_del_for_tc (tc);
+  tp_vfts[s->session_type].unbind (s->connection_index);
+  return 0;
+}
+
+void
+session_send_session_evt_to_thread (u64 session_handle,
+				    fifo_event_type_t evt_type,
+				    u32 thread_index)
+{
+  static u16 serial_number = 0;
+  session_fifo_event_t evt;
+  unix_shared_memory_queue_t *q;
+
+  /* Fabricate event */
+  evt.session_handle = session_handle;
+  evt.event_type = evt_type;
+  evt.event_id = serial_number++;
+
+  q = session_manager_get_vpp_event_queue (thread_index);
+
+  /* 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 */ );
+  else
+    {
+      clib_warning ("queue full");
+      return;
+    }
+}
+
 /**
  * Disconnect session and propagate to transport. This should eventually
  * result in a delete notification that allows us to cleanup session state.
  * Called for both active/passive disconnects.
+ *
+ * Should be called from the session's thread.
  */
 void
 stream_session_disconnect (stream_session_t * s)
 {
-//  session_fifo_event_t evt;
-
   s->session_state = SESSION_STATE_CLOSED;
-  /* RPC to vpp evt queue in the right thread */
-
   tp_vfts[s->session_type].close (s->connection_index, s->thread_index);
-
-//  {
-//  /* Fabricate event */
-//  evt.fifo = s->server_rx_fifo;
-//  evt.event_type = FIFO_EVENT_SERVER_RX;
-//  evt.event_id = serial_number++;
-//
-//  /* Based on request block (or not) for lack of space */
-//  if (PREDICT_TRUE(q->cursize < q->maxsize))
-//    unix_shared_memory_queue_add (app->event_queue, (u8 *) &evt,
-//                                0 /* do wait for mutex */);
-//  else
-//    {
-//      clib_warning("fifo full");
-//      return -1;
-//    }
-//  }
 }
 
 /**
@@ -1267,6 +1039,33 @@ session_get_transport_vft (u8 type)
   return &tp_vfts[type];
 }
 
+/**
+ * Allocate vpp event queue (once) per worker thread
+ */
+void
+session_vpp_event_queue_allocate (session_manager_main_t * smm,
+				  u32 thread_index)
+{
+  api_main_t *am = &api_main;
+  void *oldheap;
+
+  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);
+
+      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 */
+	);
+
+      svm_pop_heap (oldheap);
+    }
+}
+
 static clib_error_t *
 session_manager_main_enable (vlib_main_t * vm)
 {
@@ -1288,20 +1087,32 @@ 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
 
+  /* Allocate vpp event queues */
+  for (i = 0; i < vec_len (smm->vpp_event_queues); i++)
+    session_vpp_event_queue_allocate (smm, i);
+
   /* $$$$ preallocate hack config parameter */
   for (i = 0; i < 200000; i++)
     {
       stream_session_t *ss;
-      pool_get (smm->sessions[0], ss);
+      pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES);
       memset (ss, 0, sizeof (*ss));
     }
 
@@ -1322,9 +1133,6 @@ session_manager_main_enable (vlib_main_t * vm)
 			 200000 /* $$$$ config parameter nbuckets */ ,
 			 (64 << 20) /*$$$ config parameter table size */ );
 
-  for (i = 0; i < SESSION_N_TYPES; i++)
-    smm->connect_manager_index[i] = INVALID_INDEX;
-
   smm->is_enabled = 1;
 
   /* Enable TCP transport */
@@ -1333,6 +1141,18 @@ session_manager_main_enable (vlib_main_t * vm)
   return 0;
 }
 
+void
+session_node_enable_disable (u8 is_en)
+{
+  u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
+  /* *INDENT-OFF* */
+  foreach_vlib_main (({
+    vlib_node_set_state (this_vlib_main, session_queue_node.index,
+                         state);
+  }));
+  /* *INDENT-ON* */
+}
+
 clib_error_t *
 vnet_session_enable_disable (vlib_main_t * vm, u8 is_en)
 {
@@ -1341,16 +1161,14 @@ vnet_session_enable_disable (vlib_main_t * vm, u8 is_en)
       if (session_manager_main.is_enabled)
 	return 0;
 
-      vlib_node_set_state (vm, session_queue_node.index,
-			   VLIB_NODE_STATE_POLLING);
+      session_node_enable_disable (is_en);
 
       return session_manager_main_enable (vm);
     }
   else
     {
       session_manager_main.is_enabled = 0;
-      vlib_node_set_state (vm, session_queue_node.index,
-			   VLIB_NODE_STATE_DISABLED);
+      session_node_enable_disable (is_en);
     }
 
   return 0;