X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Ftests%2Fvnet%2Fsession%2Ftcp_echo.c;h=435a414d0979e756b39317da818ecd7afbaa9a34;hb=6442401c2;hp=0108b16ac7b916c20f167516225c11066cf29bb4;hpb=8f89dd01289ea9e97405432d2351a19c842dd6d5;p=vpp.git

diff --git a/src/tests/vnet/session/tcp_echo.c b/src/tests/vnet/session/tcp_echo.c
index 0108b16ac7b..435a414d097 100644
--- a/src/tests/vnet/session/tcp_echo.c
+++ b/src/tests/vnet/session/tcp_echo.c
@@ -38,23 +38,34 @@
 #include <vpp/api/vpe_all_api_h.h>
 #undef vl_printfun
 
+#define TCP_ECHO_DBG 0
+#define DBG(_fmt,_args...)			\
+    if (TCP_ECHO_DBG) 				\
+      clib_warning (_fmt, _args)
+
 typedef struct
 {
-  svm_fifo_t *server_rx_fifo;
-  svm_fifo_t *server_tx_fifo;
-
+  CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
+#define _(type, name) type name;
+  foreach_app_session_field
+#undef _
   u64 vpp_session_handle;
-  u64 bytes_received;
+  u64 bytes_sent;
+  u64 bytes_to_send;
+  volatile u64 bytes_received;
+  volatile u64 bytes_to_receive;
   f64 start;
-} session_t;
+} echo_session_t;
 
 typedef enum
 {
   STATE_START,
   STATE_ATTACHED,
+  STATE_LISTEN,
   STATE_READY,
   STATE_DISCONNECTING,
-  STATE_FAILED
+  STATE_FAILED,
+  STATE_DETACHED
 } connection_state_t;
 
 typedef struct
@@ -69,7 +80,7 @@ typedef struct
   u8 *uri;
 
   /* Session pool */
-  session_t *sessions;
+  echo_session_t *sessions;
 
   /* Hash table for disconnect processing */
   uword *session_index_by_vpp_handles;
@@ -85,13 +96,10 @@ typedef struct
   int i_am_master;
 
   /* drop all packets */
-  int drop_packets;
+  int no_return;
 
   /* Our event queue */
-  svm_queue_t *our_event_queue;
-
-  /* $$$ single thread only for the moment */
-  svm_queue_t *vpp_event_queue;
+  svm_msg_q_t *our_event_queue;
 
   u8 *socket_name;
 
@@ -113,18 +121,27 @@ typedef struct
   uword *error_string_by_error_number;
 
   u8 *connect_test_data;
-  pthread_t client_rx_thread_handle;
+  pthread_t *client_thread_handles;
+  u32 *thread_args;
   u32 client_bytes_received;
   u8 test_return_packets;
   u64 bytes_to_send;
+  u32 fifo_size;
+
+  u32 n_clients;
+  u64 tx_total;
+  u64 rx_total;
+
+  volatile u32 n_clients_connected;
+  volatile u32 n_active_clients;
+
 
   /** Flag that decides if socket, instead of svm, api is used to connect to
    * vpp. If sock api is used, shm binary api is subsequently bootstrapped
    * and all other messages are exchanged using shm IPC. */
   u8 use_sock_api;
 
-  /* convenience */
-  svm_fifo_segment_main_t *segment_main;
+  svm_fifo_segment_main_t segment_main;
 } echo_main_t;
 
 echo_main_t echo_main;
@@ -135,62 +152,6 @@ echo_main_t echo_main;
 #define NITER 4000000
 #endif
 
-const char test_srv_crt_rsa[] =
-  "-----BEGIN CERTIFICATE-----\r\n"
-  "MIID5zCCAs+gAwIBAgIJALeMYCEHrTtJMA0GCSqGSIb3DQEBCwUAMIGJMQswCQYD\r\n"
-  "VQQGEwJVUzELMAkGA1UECAwCQ0ExETAPBgNVBAcMCFNhbiBKb3NlMQ4wDAYDVQQK\r\n"
-  "DAVDaXNjbzEOMAwGA1UECwwFZmQuaW8xFjAUBgNVBAMMDXRlc3R0bHMuZmQuaW8x\r\n"
-  "IjAgBgkqhkiG9w0BCQEWE3ZwcC1kZXZAbGlzdHMuZmQuaW8wHhcNMTgwMzA1MjEx\r\n"
-  "NTEyWhcNMjgwMzAyMjExNTEyWjCBiTELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNB\r\n"
-  "MREwDwYDVQQHDAhTYW4gSm9zZTEOMAwGA1UECgwFQ2lzY28xDjAMBgNVBAsMBWZk\r\n"
-  "LmlvMRYwFAYDVQQDDA10ZXN0dGxzLmZkLmlvMSIwIAYJKoZIhvcNAQkBFhN2cHAt\r\n"
-  "ZGV2QGxpc3RzLmZkLmlvMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA\r\n"
-  "4C1k8a1DuStgggqT4o09fP9sJ2dC54bxhS/Xk2VEfaIZ222WSo4X/syRVfVy9Yah\r\n"
-  "cpI1zJ/RDxaZSFhgA+nPZBrFMsrULkrdAOpOVj8eDEp9JuWdO2ODSoFnCvLxcYWB\r\n"
-  "Yc5kHryJpEaGJl1sFQSesnzMFty/59ta0stk0Fp8r5NhIjWvSovGzPo6Bhz+VS2c\r\n"
-  "ebIZh4x1t2hHaFcgm0qJoJ6DceReWCW8w+yOVovTolGGq+bpb2Hn7MnRSZ2K2NdL\r\n"
-  "+aLXpkZbS/AODP1FF2vTO1mYL290LO7/51vJmPXNKSDYMy5EvILr5/VqtjsFCwRL\r\n"
-  "Q4jcM/+GeHSAFWx4qIv0BwIDAQABo1AwTjAdBgNVHQ4EFgQUWa1SOB37xmT53tZQ\r\n"
-  "aXuLLhRI7U8wHwYDVR0jBBgwFoAUWa1SOB37xmT53tZQaXuLLhRI7U8wDAYDVR0T\r\n"
-  "BAUwAwEB/zANBgkqhkiG9w0BAQsFAAOCAQEAoUht13W4ya27NVzQuCMvqPWL3VM4\r\n"
-  "3xbPFk02FaGz/WupPu276zGlzJAZrbuDcQowwwU1Ni1Yygxl96s1c2M5rHDTrOKG\r\n"
-  "rK0hbkSFBo+i6I8u4HiiQ4rYmG0Hv6+sXn3of0HsbtDPGgWZoipPWDljPYEURu3e\r\n"
-  "3HRe/Dtsj9CakBoSDzs8ndWaBR+f4sM9Tk1cjD46Gq2T/qpSPXqKxEUXlzhdCAn4\r\n"
-  "twub17Bq2kykHpppCwPg5M+v30tHG/R2Go15MeFWbEJthFk3TZMjKL7UFs7fH+x2\r\n"
-  "wSonXb++jY+KmCb93C+soABBizE57g/KmiR2IxQ/LMjDik01RSUIaM0lLA==\r\n"
-  "-----END CERTIFICATE-----\r\n";
-const u32 test_srv_crt_rsa_len = sizeof (test_srv_crt_rsa);
-
-const char test_srv_key_rsa[] =
-  "-----BEGIN PRIVATE KEY-----\r\n"
-  "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDgLWTxrUO5K2CC\r\n"
-  "CpPijT18/2wnZ0LnhvGFL9eTZUR9ohnbbZZKjhf+zJFV9XL1hqFykjXMn9EPFplI\r\n"
-  "WGAD6c9kGsUyytQuSt0A6k5WPx4MSn0m5Z07Y4NKgWcK8vFxhYFhzmQevImkRoYm\r\n"
-  "XWwVBJ6yfMwW3L/n21rSy2TQWnyvk2EiNa9Ki8bM+joGHP5VLZx5shmHjHW3aEdo\r\n"
-  "VyCbSomgnoNx5F5YJbzD7I5Wi9OiUYar5ulvYefsydFJnYrY10v5otemRltL8A4M\r\n"
-  "/UUXa9M7WZgvb3Qs7v/nW8mY9c0pINgzLkS8guvn9Wq2OwULBEtDiNwz/4Z4dIAV\r\n"
-  "bHioi/QHAgMBAAECggEBAMzGipP8+oT166U+NlJXRFifFVN1DvdhG9PWnOxGL+c3\r\n"
-  "ILmBBC08WQzmHshPemBvR6DZkA1H23cV5JTiLWrFtC00CvhXsLRMrE5+uWotI6yE\r\n"
-  "iofybMroHvD6/X5R510UX9hQ6MHu5ShLR5VZ9zXHz5MpTmB/60jG5dLx+jgcwBK8\r\n"
-  "LuGv2YB/WCUwT9QJ3YU2eaingnXtz/MrFbkbltrqlnBdlD+kTtw6Yac9y1XuuQXc\r\n"
-  "BPeulLNDuPolJVWbUvDBZrpt2dXTgz8ws1sv+wCNE0xwQJsqW4Nx3QkpibUL9RUr\r\n"
-  "CVbKlNfa9lopT6nGKlgX69R/uH35yh9AOsfasro6w0ECgYEA82UJ8u/+ORah+0sF\r\n"
-  "Q0FfW5MTdi7OAUHOz16pUsGlaEv0ERrjZxmAkHA/VRwpvDBpx4alCv0Hc39PFLIk\r\n"
-  "nhSsM2BEuBkTAs6/GaoNAiBtQVE/hN7awNRWVmlieS0go3Y3dzaE9IUMyj8sPOFT\r\n"
-  "5JdJ6BM69PHKCkY3dKdnnfpFEuECgYEA68mRpteunF1mdZgXs+WrN+uLlRrQR20F\r\n"
-  "ZyMYiUCH2Dtn26EzA2moy7FipIIrQcX/j+KhYNGM3e7MU4LymIO29E18mn8JODnH\r\n"
-  "sQOXzBTsf8A4yIVMkcuQD3bfb0JiUGYUPOidTp2N7IJA7+6Yc3vQOyb74lnKnJoO\r\n"
-  "gougPT2wS+cCgYAn7muzb6xFsXDhyW0Tm6YJYBfRS9yAWEuVufINobeBZPSl2cN1\r\n"
-  "Jrnw+HlrfTNbrJWuJmjtZJXUXQ6cVp2rUbjutNyRV4vG6iRwEXYQ40EJdkr1gZpi\r\n"
-  "CHQhuShuuPih2MNAy7EEbM+sXrDjTBR3bFqzuHPzu7dp+BshCFX3lRfAAQKBgGQt\r\n"
-  "K5i7IhCFDjb/+3IPLgOAK7mZvsvZ4eXD33TQ2eZgtut1PXtBtNl17/b85uv293Fm\r\n"
-  "VDISVcsk3eLNS8zIiT6afUoWlxAwXEs0v5WRfjl4radkGvgGiJpJYvyeM67877RB\r\n"
-  "EDSKc/X8ESLfOB44iGvZUEMG6zJFscx9DgN25iQZAoGAbyd+JEWwdVH9/K3IH1t2\r\n"
-  "PBkZX17kNWv+iVM1WyFjbe++vfKZCrOJiyiqhDeEqgrP3AuNMlaaduC3VRC3G5oV\r\n"
-  "Mj1tlhDWQ/qhvKdCKNdIVQYDE75nw+FRWV8yYkHAnXYW3tNoweDIwixE0hkPR1bc\r\n"
-  "oEjPLVNtx8SOj/M4rhaPT3I=\r\n" "-----END PRIVATE KEY-----\r\n";
-const u32 test_srv_key_rsa_len = sizeof (test_srv_key_rsa);
-
 static u8 *
 format_api_error (u8 * s, va_list * args)
 {
@@ -219,16 +180,22 @@ init_error_string_table (echo_main_t * em)
   hash_set (em->error_string_by_error_number, 99, "Misc");
 }
 
-int
+static void handle_mq_event (session_event_t * e);
+
+static int
 wait_for_state_change (echo_main_t * em, connection_state_t state)
 {
+  svm_msg_q_msg_t msg;
+  session_event_t *e;
+  f64 timeout;
+
 #if CLIB_DEBUG > 0
 #define TIMEOUT 600.0
 #else
 #define TIMEOUT 600.0
 #endif
 
-  f64 timeout = clib_time_now (&em->clib_time) + TIMEOUT;
+  timeout = clib_time_now (&em->clib_time) + TIMEOUT;
 
   while (clib_time_now (&em->clib_time) < timeout)
     {
@@ -238,8 +205,16 @@ wait_for_state_change (echo_main_t * em, connection_state_t state)
 	return -1;
       if (em->time_to_stop == 1)
 	return 0;
+      if (!em->our_event_queue || em->state < STATE_ATTACHED)
+	continue;
+
+      if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_NOWAIT, 0))
+	continue;
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
+      handle_mq_event (e);
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
     }
-  clib_warning ("timeout waiting for STATE_READY");
+  clib_warning ("timeout waiting for state %d", state);
   return -1;
 }
 
@@ -250,42 +225,42 @@ application_send_attach (echo_main_t * em)
   vl_api_application_tls_cert_add_t *cert_mp;
   vl_api_application_tls_key_add_t *key_mp;
 
-  u32 fifo_size = 4 << 20;
   bmp = vl_msg_api_alloc (sizeof (*bmp));
-  memset (bmp, 0, sizeof (*bmp));
+  clib_memset (bmp, 0, sizeof (*bmp));
 
   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_ATTACH);
   bmp->client_index = em->my_client_index;
   bmp->context = ntohl (0xfeedface);
-  bmp->options[APP_OPTIONS_FLAGS] =
-    APP_OPTIONS_FLAGS_ACCEPT_REDIRECT | APP_OPTIONS_FLAGS_ADD_SEGMENT;
+  bmp->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_ACCEPT_REDIRECT;
+  bmp->options[APP_OPTIONS_FLAGS] |= APP_OPTIONS_FLAGS_ADD_SEGMENT;
   bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16;
-  bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = fifo_size;
-  bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = fifo_size;
+  bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = em->fifo_size;
+  bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = em->fifo_size;
   bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
   bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
+  bmp->options[APP_OPTIONS_EVT_QUEUE_SIZE] = 256;
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 
   cert_mp = vl_msg_api_alloc (sizeof (*cert_mp) + test_srv_crt_rsa_len);
-  memset (cert_mp, 0, sizeof (*cert_mp));
+  clib_memset (cert_mp, 0, sizeof (*cert_mp));
   cert_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_CERT_ADD);
   cert_mp->client_index = em->my_client_index;
   cert_mp->context = ntohl (0xfeedface);
   cert_mp->cert_len = clib_host_to_net_u16 (test_srv_crt_rsa_len);
-  clib_memcpy (cert_mp->cert, test_srv_crt_rsa, test_srv_crt_rsa_len);
+  clib_memcpy_fast (cert_mp->cert, test_srv_crt_rsa, test_srv_crt_rsa_len);
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cert_mp);
 
   key_mp = vl_msg_api_alloc (sizeof (*key_mp) + test_srv_key_rsa_len);
-  memset (key_mp, 0, sizeof (*key_mp) + test_srv_key_rsa_len);
+  clib_memset (key_mp, 0, sizeof (*key_mp) + test_srv_key_rsa_len);
   key_mp->_vl_msg_id = ntohs (VL_API_APPLICATION_TLS_KEY_ADD);
   key_mp->client_index = em->my_client_index;
   key_mp->context = ntohl (0xfeedface);
   key_mp->key_len = clib_host_to_net_u16 (test_srv_key_rsa_len);
-  clib_memcpy (key_mp->key, test_srv_key_rsa, test_srv_key_rsa_len);
+  clib_memcpy_fast (key_mp->key, test_srv_key_rsa, test_srv_key_rsa_len);
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & key_mp);
 }
 
-int
+static int
 application_attach (echo_main_t * em)
 {
   application_send_attach (em);
@@ -302,62 +277,37 @@ application_detach (echo_main_t * em)
 {
   vl_api_application_detach_t *bmp;
   bmp = vl_msg_api_alloc (sizeof (*bmp));
-  memset (bmp, 0, sizeof (*bmp));
+  clib_memset (bmp, 0, sizeof (*bmp));
 
   bmp->_vl_msg_id = ntohs (VL_API_APPLICATION_DETACH);
   bmp->client_index = em->my_client_index;
   bmp->context = ntohl (0xfeedface);
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 
-  clib_warning ("Sent detach");
-}
-
-static int
-memfd_segment_attach (void)
-{
-  ssvm_private_t _ssvm = { 0 }, *ssvm = &_ssvm;
-  clib_error_t *error;
-  int rv;
-
-  if ((error = vl_socket_client_recv_fd_msg (&ssvm->fd, 5)))
-    {
-      clib_error_report (error);
-      return -1;
-    }
-
-  if ((rv = ssvm_slave_init_memfd (ssvm)))
-    return rv;
-
-  return 0;
+  DBG ("%s", "Sent detach");
 }
 
 static int
-fifo_segment_attach (char *name, u32 size, ssvm_segment_type_t type)
+ssvm_segment_attach (char *name, ssvm_segment_type_t type, int fd)
 {
   svm_fifo_segment_create_args_t _a, *a = &_a;
-  clib_error_t *error;
+  svm_fifo_segment_main_t *sm = &echo_main.segment_main;
   int rv;
 
-  memset (a, 0, sizeof (*a));
+  clib_memset (a, 0, sizeof (*a));
   a->segment_name = (char *) name;
-  a->segment_size = size;
   a->segment_type = type;
 
   if (type == SSVM_SEGMENT_MEMFD)
-    {
-      if ((error = vl_socket_client_recv_fd_msg (&a->memfd_fd, 5)))
-	{
-	  clib_error_report (error);
-	  return -1;
-	}
-    }
+    a->memfd_fd = fd;
 
-  if ((rv = svm_fifo_segment_attach (a)))
+  if ((rv = svm_fifo_segment_attach (sm, a)))
     {
       clib_warning ("svm_fifo_segment_attach ('%s') failed", name);
       return rv;
     }
 
+  vec_reset_length (a->new_segment_indices);
   return 0;
 }
 
@@ -366,47 +316,57 @@ vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
 					   mp)
 {
   echo_main_t *em = &echo_main;
-  ssvm_segment_type_t seg_type;
+  int *fds = 0;
+  u32 n_fds = 0;
 
   if (mp->retval)
     {
       clib_warning ("attach failed: %U", format_api_error,
 		    clib_net_to_host_u32 (mp->retval));
-      em->state = STATE_FAILED;
-      return;
+      goto failed;
     }
 
   if (mp->segment_name_length == 0)
     {
       clib_warning ("segment_name_length zero");
-      return;
+      goto failed;
     }
 
-  seg_type = em->use_sock_api ? SSVM_SEGMENT_MEMFD : SSVM_SEGMENT_SHM;
+  ASSERT (mp->app_event_queue_address);
+  em->our_event_queue = uword_to_pointer (mp->app_event_queue_address,
+					  svm_msg_q_t *);
 
-  /* Attach to fifo segment */
-  if (fifo_segment_attach ((char *) mp->segment_name, mp->segment_size,
-			   seg_type))
+  if (mp->n_fds)
     {
-      em->state = STATE_FAILED;
-      return;
-    }
+      vec_validate (fds, mp->n_fds);
+      vl_socket_client_recv_fd_msg (fds, mp->n_fds, 5);
+
+      if (mp->fd_flags & SESSION_FD_F_VPP_MQ_SEGMENT)
+	if (ssvm_segment_attach (0, SSVM_SEGMENT_MEMFD, fds[n_fds++]))
+	  goto failed;
+
+      if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
+	if (ssvm_segment_attach ((char *) mp->segment_name,
+				 SSVM_SEGMENT_MEMFD, fds[n_fds++]))
+	  goto failed;
 
-  /* If we're using memfd segments, read and attach to event qs segment */
-  if (seg_type == SSVM_SEGMENT_MEMFD)
+      if (mp->fd_flags & SESSION_FD_F_MQ_EVENTFD)
+	svm_msg_q_set_consumer_eventfd (em->our_event_queue, fds[n_fds++]);
+
+      vec_free (fds);
+    }
+  else
     {
-      if (memfd_segment_attach ())
-	{
-	  clib_warning ("failed to attach to evt q segment");
-	  em->state = STATE_FAILED;
-	  return;
-	}
+      if (ssvm_segment_attach ((char *) mp->segment_name, SSVM_SEGMENT_SHM,
+			       -1))
+	goto failed;
     }
 
-  ASSERT (mp->app_event_queue_address);
-  em->our_event_queue = uword_to_pointer (mp->app_event_queue_address,
-					  svm_queue_t *);
   em->state = STATE_ATTACHED;
+  return;
+failed:
+  em->state = STATE_FAILED;
+  return;
 }
 
 static void
@@ -415,6 +375,7 @@ vl_api_application_detach_reply_t_handler (vl_api_application_detach_reply_t *
 {
   if (mp->retval)
     clib_warning ("detach returned with err: %d", mp->retval);
+  echo_main.state = STATE_DETACHED;
 }
 
 static void
@@ -464,7 +425,7 @@ connect_to_vpp (char *name)
 	  return -1;
 	}
 
-      if (vl_socket_client_init_shm (0))
+      if (vl_socket_client_init_shm (0, 1 /* want_pthread */ ))
 	{
 	  clib_warning ("init shm api failed");
 	  return -1;
@@ -495,14 +456,15 @@ disconnect_from_vpp (echo_main_t * em)
 static void
 vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 {
+  svm_fifo_segment_main_t *sm = &echo_main.segment_main;
   svm_fifo_segment_create_args_t _a, *a = &_a;
   int rv;
 
-  memset (a, 0, sizeof (*a));
+  clib_memset (a, 0, sizeof (*a));
   a->segment_name = (char *) mp->segment_name;
   a->segment_size = mp->segment_size;
   /* Attach to the segment vpp created */
-  rv = svm_fifo_segment_attach (a);
+  rv = svm_fifo_segment_attach (sm, a);
   if (rv)
     {
       clib_warning ("svm_fifo_segment_attach ('%s') failed",
@@ -514,7 +476,7 @@ vl_api_map_another_segment_t_handler (vl_api_map_another_segment_t * mp)
 }
 
 static void
-session_print_stats (echo_main_t * em, session_t * session)
+session_print_stats (echo_main_t * em, echo_session_t * session)
 {
   f64 deltat;
   u64 bytes;
@@ -526,204 +488,223 @@ session_print_stats (echo_main_t * em, session_t * session)
 }
 
 static void
-vl_api_disconnect_session_t_handler (vl_api_disconnect_session_t * mp)
+test_recv_bytes (echo_session_t * s, u8 * rx_buf, u32 n_read)
 {
-  echo_main_t *em = &echo_main;
-  session_t *session = 0;
-  vl_api_disconnect_session_reply_t *rmp;
-  uword *p;
-  int rv = 0;
-
-  p = hash_get (em->session_index_by_vpp_handles, mp->handle);
-
-  if (p)
-    {
-      session = pool_elt_at_index (em->sessions, p[0]);
-      hash_unset (em->session_index_by_vpp_handles, mp->handle);
-      pool_put (em->sessions, session);
-    }
-  else
+  int i;
+  for (i = 0; i < n_read; i++)
     {
-      clib_warning ("couldn't find session key %llx", mp->handle);
-      rv = -11;
+      if (rx_buf[i] != ((s->bytes_received + i) & 0xff))
+	{
+	  clib_warning ("error at byte %lld, 0x%x not 0x%x",
+			s->bytes_received + i, rx_buf[i],
+			((s->bytes_received + i) & 0xff));
+	}
     }
-
-//  em->time_to_stop = 1;
-
-  rmp = vl_msg_api_alloc (sizeof (*rmp));
-  memset (rmp, 0, sizeof (*rmp));
-
-  rmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION_REPLY);
-  rmp->retval = rv;
-  rmp->handle = mp->handle;
-  rmp->context = mp->context;
-  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
-
-  if (session)
-    session_print_stats (em, session);
 }
 
 static void
-vl_api_reset_session_t_handler (vl_api_reset_session_t * mp)
+recv_data_chunk (echo_main_t * em, echo_session_t * s, u8 * rx_buf)
 {
-  echo_main_t *em = &echo_main;
-  vl_api_reset_session_reply_t *rmp;
-  uword *p;
-  int rv = 0;
+  int n_to_read, n_read;
 
-  p = hash_get (em->session_index_by_vpp_handles, mp->handle);
+  n_to_read = svm_fifo_max_dequeue (s->rx_fifo);
+  if (!n_to_read)
+    return;
 
-  if (p)
-    {
-      clib_warning ("got reset");
-      /* Cleanup later */
-      em->time_to_stop = 1;
-    }
-  else
+  do
     {
-      clib_warning ("couldn't find session key %llx", mp->handle);
-      rv = -11;
-    }
+      n_read = app_recv_stream ((app_session_t *) s, rx_buf,
+				vec_len (rx_buf));
 
-  rmp = vl_msg_api_alloc (sizeof (*rmp));
-  memset (rmp, 0, sizeof (*rmp));
-  rmp->_vl_msg_id = ntohs (VL_API_RESET_SESSION_REPLY);
-  rmp->retval = rv;
-  rmp->handle = mp->handle;
-  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
+      if (n_read > 0)
+	{
+	  if (em->test_return_packets)
+	    test_recv_bytes (s, rx_buf, n_read);
+
+	  n_to_read -= n_read;
+
+	  s->bytes_received += n_read;
+	  s->bytes_to_receive -= n_read;
+	}
+      else
+	break;
+    }
+  while (n_to_read > 0);
 }
 
 void
-client_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
+client_handle_rx (echo_main_t * em, session_event_t * e, u8 * rx_buf)
 {
-  svm_fifo_t *rx_fifo;
-  int n_read, bytes, i;
+  echo_session_t *s;
 
-  rx_fifo = e->fifo;
+  s = pool_elt_at_index (em->sessions, e->fifo->client_session_index);
+  recv_data_chunk (em, s, rx_buf);
+}
 
-  bytes = svm_fifo_max_dequeue (rx_fifo);
-  /* Allow enqueuing of new event */
-  svm_fifo_unset_event (rx_fifo);
+static void
+send_data_chunk (echo_main_t * em, echo_session_t * s)
+{
+  u64 test_buf_len, bytes_this_chunk, test_buf_offset;
+  u8 *test_data = em->connect_test_data;
+  int n_sent;
 
-  /* Read the bytes */
-  do
-    {
-      n_read = svm_fifo_dequeue_nowait (rx_fifo,
-					clib_min (vec_len (em->rx_buf),
-						  bytes), em->rx_buf);
-      if (n_read > 0)
-	{
-	  bytes -= n_read;
-	  if (em->test_return_packets)
-	    {
-	      for (i = 0; i < n_read; i++)
-		{
-		  if (em->rx_buf[i]
-		      != ((em->client_bytes_received + i) & 0xff))
-		    {
-		      clib_warning ("error at byte %lld, 0x%x not 0x%x",
-				    em->client_bytes_received + i,
-				    em->rx_buf[i],
-				    ((em->client_bytes_received + i) & 0xff));
-		    }
-		}
-	    }
-	  em->client_bytes_received += n_read;
-	}
-      else
-	{
-	  if (n_read == -2)
-	    {
-//            clib_warning ("weird!");
-	      break;
-	    }
-	}
+  test_buf_len = vec_len (test_data);
+  test_buf_offset = s->bytes_sent % test_buf_len;
+  bytes_this_chunk = clib_min (test_buf_len - test_buf_offset,
+			       s->bytes_to_send);
+
+  n_sent = app_send_stream ((app_session_t *) s, test_data + test_buf_offset,
+			    bytes_this_chunk, 0);
 
+  if (n_sent > 0)
+    {
+      s->bytes_to_send -= n_sent;
+      s->bytes_sent += n_sent;
     }
-  while (bytes > 0);
 }
 
-void
-client_handle_event_queue (echo_main_t * em)
+/*
+ * Rx/Tx polling thread per connection
+ */
+static void *
+client_thread_fn (void *arg)
 {
-  session_fifo_event_t _e, *e = &_e;;
+  echo_main_t *em = &echo_main;
+  static u8 *rx_buf = 0;
+  u32 session_index = *(u32 *) arg;
+  echo_session_t *s;
 
-  svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
-  switch (e->event_type)
-    {
-    case FIFO_EVENT_APP_RX:
-      client_handle_fifo_event_rx (em, e);
-      break;
+  vec_validate (rx_buf, 1 << 20);
 
-    case FIFO_EVENT_DISCONNECT:
-      return;
+  while (!em->time_to_stop && em->state != STATE_READY)
+    ;
 
-    default:
-      clib_warning ("unknown event type %d", e->event_type);
-      break;
+  s = pool_elt_at_index (em->sessions, session_index);
+  while (!em->time_to_stop)
+    {
+      send_data_chunk (em, s);
+      recv_data_chunk (em, s, rx_buf);
+      if (!s->bytes_to_send && !s->bytes_to_receive)
+	break;
     }
+
+  DBG ("session %d done", session_index);
+  em->tx_total += s->bytes_sent;
+  em->rx_total += s->bytes_received;
+  em->n_active_clients--;
+
+  pthread_exit (0);
 }
 
-static void *
+/*
+ * Rx thread that handles all connections.
+ *
+ * Not used.
+ */
+void *
 client_rx_thread_fn (void *arg)
 {
-  session_fifo_event_t _e, *e = &_e;
+  session_event_t _e, *e = &_e;
   echo_main_t *em = &echo_main;
+  static u8 *rx_buf = 0;
+  svm_msg_q_msg_t msg;
 
-  em->client_bytes_received = 0;
-  while (1)
+  vec_validate (rx_buf, 1 << 20);
+
+  while (!em->time_to_stop && em->state != STATE_READY)
+    ;
+
+  while (!em->time_to_stop)
     {
-      svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
+      svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
-	  client_handle_fifo_event_rx (em, e);
+	  client_handle_rx (em, e, rx_buf);
 	  break;
-
-	case FIFO_EVENT_DISCONNECT:
-	  return 0;
 	default:
 	  clib_warning ("unknown event type %d", e->event_type);
 	  break;
 	}
-
-      if (PREDICT_FALSE (em->time_to_stop == 1))
-	break;
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
     }
   pthread_exit (0);
 }
 
+void
+client_send_connect (echo_main_t * em)
+{
+  vl_api_connect_uri_t *cmp;
+  cmp = vl_msg_api_alloc (sizeof (*cmp));
+  clib_memset (cmp, 0, sizeof (*cmp));
+
+  cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
+  cmp->client_index = em->my_client_index;
+  cmp->context = ntohl (0xfeedface);
+  memcpy (cmp->uri, em->connect_uri, vec_len (em->connect_uri));
+  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
+}
+
+void
+client_send_disconnect (echo_main_t * em, echo_session_t * s)
+{
+  vl_api_disconnect_session_t *dmp;
+  dmp = vl_msg_api_alloc (sizeof (*dmp));
+  clib_memset (dmp, 0, sizeof (*dmp));
+  dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
+  dmp->client_index = em->my_client_index;
+  dmp->handle = s->vpp_session_handle;
+  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
+}
+
+int
+client_disconnect (echo_main_t * em, echo_session_t * s)
+{
+  client_send_disconnect (em, s);
+  pool_put (em->sessions, s);
+  clib_memset (s, 0xfe, sizeof (*s));
+  return 0;
+}
 
 static void
-vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
+session_bound_handler (session_bound_msg_t * mp)
 {
   echo_main_t *em = &echo_main;
-  session_t *session;
-  u32 session_index;
-  svm_fifo_t *rx_fifo, *tx_fifo;
-  int rv;
 
   if (mp->retval)
     {
-      clib_warning ("connection failed with code: %U", format_api_error,
+      clib_warning ("bind failed: %U", format_api_error,
 		    clib_net_to_host_u32 (mp->retval));
       em->state = STATE_FAILED;
       return;
     }
-  else
-    {
-      clib_warning ("connected with local ip %U port %d", format_ip46_address,
-		    mp->lcl_ip, mp->is_ip4,
-		    clib_net_to_host_u16 (mp->lcl_port));
-    }
 
-  em->vpp_event_queue =
-    uword_to_pointer (mp->vpp_event_queue_address, svm_queue_t *);
+  clib_warning ("listening on %U:%u", format_ip46_address, mp->lcl_ip,
+		mp->lcl_is_ip4 ? IP46_TYPE_IP4 : IP46_TYPE_IP6, mp->lcl_port);
+  em->state = STATE_READY;
+}
+
+static void
+session_accepted_handler (session_accepted_msg_t * mp)
+{
+  app_session_evt_t _app_evt, *app_evt = &_app_evt;
+  session_accepted_reply_msg_t *rmp;
+  svm_fifo_t *rx_fifo, *tx_fifo;
+  echo_main_t *em = &echo_main;
+  echo_session_t *session;
+  static f64 start_time;
+  u32 session_index;
+  u8 *ip_str;
 
-  /*
-   * Setup session
-   */
+  if (start_time == 0.0)
+    start_time = clib_time_now (&em->clib_time);
 
+  ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4);
+  clib_warning ("Accepted session from: %s:%d", ip_str,
+		clib_net_to_host_u16 (mp->port));
+
+  /* Allocate local session and set it up */
   pool_get (em->sessions, session);
   session_index = session - em->sessions;
 
@@ -732,194 +713,274 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
   tx_fifo->client_session_index = session_index;
 
-  session->server_rx_fifo = rx_fifo;
-  session->server_tx_fifo = tx_fifo;
-  session->vpp_session_handle = mp->handle;
-  session->start = clib_time_now (&em->clib_time);
-
-  /* Save handle */
-  em->connected_session_index = session_index;
-  em->state = STATE_READY;
+  session->rx_fifo = rx_fifo;
+  session->tx_fifo = tx_fifo;
+  session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
+					 svm_msg_q_t *);
 
   /* Add it to lookup table */
   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 
-  /* Start RX thread */
-  rv = pthread_create (&em->client_rx_thread_handle,
-		       NULL /*attr */ , client_rx_thread_fn, 0);
-  if (rv)
+  em->state = STATE_READY;
+
+  /* Stats printing */
+  if (pool_elts (em->sessions) && (pool_elts (em->sessions) % 20000) == 0)
     {
-      clib_warning ("pthread_create returned %d", rv);
-      rv = VNET_API_ERROR_SYSCALL_ERROR_1;
+      f64 now = clib_time_now (&em->clib_time);
+      fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
+	       pool_elts (em->sessions), now - start_time,
+	       (f64) pool_elts (em->sessions) / (now - start_time));
     }
+
+  /*
+   * Send accept reply to vpp
+   */
+  app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
+			     SESSION_CTRL_EVT_ACCEPTED_REPLY);
+  rmp = (session_accepted_reply_msg_t *) app_evt->evt->data;
+  rmp->handle = mp->handle;
+  rmp->context = mp->context;
+  app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
+
+  session->bytes_received = 0;
+  session->start = clib_time_now (&em->clib_time);
 }
 
 static void
-send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
+session_connected_handler (session_connected_msg_t * mp)
 {
-  u8 *test_data = em->connect_test_data;
-  u64 bytes_sent = 0;
-  int test_buf_offset = 0;
-  u32 bytes_to_snd;
-  u32 queue_max_chunk = 128 << 10, actual_write;
-  session_fifo_event_t evt;
+  echo_main_t *em = &echo_main;
+  echo_session_t *session;
+  u32 session_index;
+  svm_fifo_t *rx_fifo, *tx_fifo;
   int rv;
 
-  bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
-  if (bytes_to_snd > vec_len (test_data))
-    bytes_to_snd = vec_len (test_data);
-
-  while (bytes_to_snd > 0 && !em->time_to_stop)
+  if (mp->retval)
     {
-      actual_write = (bytes_to_snd > queue_max_chunk) ?
-	queue_max_chunk : bytes_to_snd;
-      rv = svm_fifo_enqueue_nowait (tx_fifo, actual_write,
-				    test_data + test_buf_offset);
+      clib_warning ("connection failed with code: %U", format_api_error,
+		    clib_net_to_host_u32 (mp->retval));
+      em->state = STATE_FAILED;
+      return;
+    }
 
-      if (rv > 0)
-	{
-	  bytes_to_snd -= rv;
-	  test_buf_offset += rv;
-	  bytes_sent += rv;
+  /*
+   * Setup session
+   */
 
-	  if (svm_fifo_set_event (tx_fifo))
-	    {
-	      /* Fabricate TX event, send to vpp */
-	      evt.fifo = tx_fifo;
-	      evt.event_type = FIFO_EVENT_APP_TX;
+  pool_get (em->sessions, session);
+  clib_memset (session, 0, sizeof (*session));
+  session_index = session - em->sessions;
 
-	      svm_queue_add (em->vpp_event_queue,
-			     (u8 *) & evt, 0 /* do wait for mutex */ );
-	    }
-	}
-    }
-}
+  rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
+  rx_fifo->client_session_index = session_index;
+  tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
+  tx_fifo->client_session_index = session_index;
 
-void
-client_send_data (echo_main_t * em)
-{
-  u8 *test_data = em->connect_test_data;
-  int mypid = getpid ();
-  session_t *session;
-  svm_fifo_t *tx_fifo;
-  u32 n_iterations, leftover;
-  int i;
+  session->rx_fifo = rx_fifo;
+  session->tx_fifo = tx_fifo;
+  session->vpp_session_handle = mp->handle;
+  session->start = clib_time_now (&em->clib_time);
+  session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
+					 svm_msg_q_t *);
 
-  session = pool_elt_at_index (em->sessions, em->connected_session_index);
-  tx_fifo = session->server_tx_fifo;
+  hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
+
+  /*
+   * Start RX thread
+   */
+  em->thread_args[em->n_clients_connected] = session_index;
+  rv = pthread_create (&em->client_thread_handles[em->n_clients_connected],
+		       NULL /*attr */ , client_thread_fn,
+		       (void *) &em->thread_args[em->n_clients_connected]);
+  if (rv)
+    {
+      clib_warning ("pthread_create returned %d", rv);
+      return;
+    }
 
-  ASSERT (vec_len (test_data) > 0);
+  em->n_clients_connected += 1;
+  clib_warning ("session %u (0x%llx) connected with local ip %U port %d",
+		session_index, mp->handle, format_ip46_address, mp->lcl_ip,
+		mp->is_ip4, clib_net_to_host_u16 (mp->lcl_port));
+}
 
-  vec_validate (em->rx_buf, vec_len (test_data) - 1);
-  n_iterations = em->bytes_to_send / vec_len (test_data);
+static void
+session_disconnected_handler (session_disconnected_msg_t * mp)
+{
+  app_session_evt_t _app_evt, *app_evt = &_app_evt;
+  session_disconnected_reply_msg_t *rmp;
+  echo_main_t *em = &echo_main;
+  echo_session_t *session = 0;
+  uword *p;
+  int rv = 0;
 
-  for (i = 0; i < n_iterations; i++)
+  p = hash_get (em->session_index_by_vpp_handles, mp->handle);
+  if (!p)
     {
-      send_test_chunk (em, tx_fifo, mypid, 0);
-      if (em->time_to_stop)
-	break;
+      clib_warning ("couldn't find session key %llx", mp->handle);
+      return;
     }
 
-  leftover = em->bytes_to_send % vec_len (test_data);
-  if (leftover)
-    send_test_chunk (em, tx_fifo, mypid, leftover);
+  session = pool_elt_at_index (em->sessions, p[0]);
+  hash_unset (em->session_index_by_vpp_handles, mp->handle);
+  pool_put (em->sessions, session);
 
-  if (!em->drop_packets)
-    {
-      f64 timeout = clib_time_now (&em->clib_time) + 10;
+  app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
+			     SESSION_CTRL_EVT_DISCONNECTED_REPLY);
+  rmp = (session_disconnected_reply_msg_t *) app_evt->evt->data;
+  rmp->retval = rv;
+  rmp->handle = mp->handle;
+  rmp->context = mp->context;
+  app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 
-      /* Wait for the outstanding packets */
-      while (em->client_bytes_received <
-	     vec_len (test_data) * n_iterations + leftover)
-	{
-	  if (clib_time_now (&em->clib_time) > timeout)
-	    {
-	      clib_warning ("timed out waiting for the missing packets");
-	      break;
-	    }
-	}
-    }
-  em->time_to_stop = 1;
+  session_print_stats (em, session);
 }
 
-void
-client_send_connect (echo_main_t * em)
+static void
+session_reset_handler (session_reset_msg_t * mp)
 {
-  vl_api_connect_uri_t *cmp;
-  cmp = vl_msg_api_alloc (sizeof (*cmp));
-  memset (cmp, 0, sizeof (*cmp));
+  app_session_evt_t _app_evt, *app_evt = &_app_evt;
+  echo_main_t *em = &echo_main;
+  session_reset_reply_msg_t *rmp;
+  echo_session_t *session = 0;
+  uword *p;
+  int rv = 0;
 
-  cmp->_vl_msg_id = ntohs (VL_API_CONNECT_URI);
-  cmp->client_index = em->my_client_index;
-  cmp->context = ntohl (0xfeedface);
-  memcpy (cmp->uri, em->connect_uri, vec_len (em->connect_uri));
-  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
-}
+  p = hash_get (em->session_index_by_vpp_handles, mp->handle);
 
-int
-client_connect (echo_main_t * em)
-{
-  client_send_connect (em);
-  if (wait_for_state_change (em, STATE_READY))
+  if (p)
     {
-      clib_warning ("Connect failed");
-      return -1;
+      session = pool_elt_at_index (em->sessions, p[0]);
+      clib_warning ("got reset");
+      /* Cleanup later */
+      em->time_to_stop = 1;
+    }
+  else
+    {
+      clib_warning ("couldn't find session key %llx", mp->handle);
+      return;
     }
-  return 0;
-}
 
-void
-client_send_disconnect (echo_main_t * em)
-{
-  session_t *connected_session;
-  vl_api_disconnect_session_t *dmp;
-  connected_session = pool_elt_at_index (em->sessions,
-					 em->connected_session_index);
-  dmp = vl_msg_api_alloc (sizeof (*dmp));
-  memset (dmp, 0, sizeof (*dmp));
-  dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
-  dmp->client_index = em->my_client_index;
-  dmp->handle = connected_session->vpp_session_handle;
-  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
+  app_alloc_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt,
+			     SESSION_CTRL_EVT_RESET_REPLY);
+  rmp = (session_reset_reply_msg_t *) app_evt->evt->data;
+  rmp->retval = rv;
+  rmp->handle = mp->handle;
+  app_send_ctrl_evt_to_vpp (session->vpp_evt_q, app_evt);
 }
 
-int
-client_disconnect (echo_main_t * em)
+static void
+handle_mq_event (session_event_t * e)
 {
-  client_send_disconnect (em);
-  clib_warning ("Sent disconnect");
-  if (wait_for_state_change (em, STATE_START))
+  switch (e->event_type)
     {
-      clib_warning ("Disconnect failed");
-      return -1;
+    case SESSION_CTRL_EVT_BOUND:
+      session_bound_handler ((session_bound_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_ACCEPTED:
+      session_accepted_handler ((session_accepted_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_CONNECTED:
+      session_connected_handler ((session_connected_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_DISCONNECTED:
+      session_disconnected_handler ((session_disconnected_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_RESET:
+      session_reset_handler ((session_reset_msg_t *) e->data);
+      break;
+    default:
+      clib_warning ("unhandled %u", e->event_type);
     }
-  return 0;
 }
 
 static void
-client_run (echo_main_t * em)
+clients_run (echo_main_t * em)
 {
+  f64 start_time, deltat, timeout = 100.0;
+  svm_msg_q_msg_t msg;
+  session_event_t *e;
+  echo_session_t *s;
   int i;
 
+  /* Init test data */
+  vec_validate (em->connect_test_data, 1024 * 1024 - 1);
+  for (i = 0; i < vec_len (em->connect_test_data); i++)
+    em->connect_test_data[i] = i & 0xff;
+
+  /*
+   * Attach and connect the clients
+   */
   if (application_attach (em))
     return;
 
-  if (client_connect (em))
+  for (i = 0; i < em->n_clients; i++)
+    client_send_connect (em);
+
+  start_time = clib_time_now (&em->clib_time);
+  while (em->n_clients_connected < em->n_clients
+	 && (clib_time_now (&em->clib_time) - start_time < timeout)
+	 && em->state != STATE_FAILED)
+
     {
-      application_detach (em);
+      svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
+      handle_mq_event (e);
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
+    }
+
+  if (em->n_clients_connected != em->n_clients)
+    {
+      clib_warning ("failed to initialize all connections");
       return;
     }
 
-  /* Init test data */
-  vec_validate (em->connect_test_data, 128 * 1024 - 1);
-  for (i = 0; i < vec_len (em->connect_test_data); i++)
-    em->connect_test_data[i] = i & 0xff;
+  /*
+   * Initialize connections
+   */
+  for (i = 0; i < em->n_clients; i++)
+    {
+      s = pool_elt_at_index (em->sessions, i);
+      s->bytes_to_send = em->bytes_to_send;
+      if (!em->no_return)
+	s->bytes_to_receive = em->bytes_to_send;
+    }
+  em->n_active_clients = em->n_clients_connected;
+
+  /*
+   * Wait for client threads to send the data
+   */
+  start_time = clib_time_now (&em->clib_time);
+  em->state = STATE_READY;
+  while (em->n_active_clients)
+    if (!svm_msg_q_is_empty (em->our_event_queue))
+      {
+	if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0))
+	  {
+	    clib_warning ("svm msg q returned");
+	    continue;
+	  }
+	e = svm_msg_q_msg_data (em->our_event_queue, &msg);
+	if (e->event_type != FIFO_EVENT_APP_RX)
+	  handle_mq_event (e);
+	svm_msg_q_free_msg (em->our_event_queue, &msg);
+      }
+
+  for (i = 0; i < em->n_clients; i++)
+    {
+      s = pool_elt_at_index (em->sessions, i);
+      client_disconnect (em, s);
+    }
 
-  /* Start send */
-  client_send_data (em);
+  /*
+   * Stats and detach
+   */
+  deltat = clib_time_now (&em->clib_time) - start_time;
+  fformat (stdout, "%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds\n",
+	   em->tx_total, em->tx_total / (1ULL << 20),
+	   em->tx_total / (1ULL << 30), deltat);
+  fformat (stdout, "%.4f Gbit/second\n", (em->tx_total * 8.0) / deltat / 1e9);
 
-  /* Disconnect and detach */
-  client_disconnect (em);
   application_detach (em);
 }
 
@@ -1034,151 +1095,77 @@ format_ip46_address (u8 * s, va_list * args)
 }
 
 static void
-vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
+server_handle_rx (echo_main_t * em, session_event_t * e)
 {
-  echo_main_t *em = &echo_main;
-  vl_api_accept_session_reply_t *rmp;
-  svm_fifo_t *rx_fifo, *tx_fifo;
-  session_t *session;
-  static f64 start_time;
-  u32 session_index;
-  u8 *ip_str;
-
-  if (start_time == 0.0)
-    start_time = clib_time_now (&em->clib_time);
-
-  ip_str = format (0, "%U", format_ip46_address, &mp->ip, mp->is_ip4);
-  clib_warning ("Accepted session from: %s:%d", ip_str,
-		clib_net_to_host_u16 (mp->port));
-  em->vpp_event_queue =
-    uword_to_pointer (mp->vpp_event_queue_address, svm_queue_t *);
-
-  /* Allocate local session and set it up */
-  pool_get (em->sessions, session);
-  session_index = session - em->sessions;
-
-  rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
-  rx_fifo->client_session_index = session_index;
-  tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
-  tx_fifo->client_session_index = session_index;
-
-  session->server_rx_fifo = rx_fifo;
-  session->server_tx_fifo = tx_fifo;
+  int n_read, max_dequeue, n_sent;
+  u32 offset, to_dequeue;
+  echo_session_t *s;
 
-  /* Add it to lookup table */
-  hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
+  s = pool_elt_at_index (em->sessions, e->fifo->client_session_index);
 
-  em->state = STATE_READY;
+  /* Clear event only once. Otherwise, if we do it in the loop by calling
+   * app_recv_stream, we may end up with a lot of unhandled rx events on the
+   * message queue */
+  svm_fifo_unset_event (s->rx_fifo);
 
-  /* Stats printing */
-  if (pool_elts (em->sessions) && (pool_elts (em->sessions) % 20000) == 0)
-    {
-      f64 now = clib_time_now (&em->clib_time);
-      fformat (stdout, "%d active sessions in %.2f seconds, %.2f/sec...\n",
-	       pool_elts (em->sessions), now - start_time,
-	       (f64) pool_elts (em->sessions) / (now - start_time));
-    }
-
-  /*
-   * Send accept reply to vpp
-   */
-  rmp = vl_msg_api_alloc (sizeof (*rmp));
-  memset (rmp, 0, sizeof (*rmp));
-  rmp->_vl_msg_id = ntohs (VL_API_ACCEPT_SESSION_REPLY);
-  rmp->handle = mp->handle;
-  rmp->context = mp->context;
-  vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
-
-  session->bytes_received = 0;
-  session->start = clib_time_now (&em->clib_time);
-}
-
-void
-server_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
-{
-  svm_fifo_t *rx_fifo, *tx_fifo;
-  int n_read;
-  session_fifo_event_t evt;
-  svm_queue_t *q;
-  session_t *session;
-  int rv;
-  u32 max_dequeue, offset, max_transfer, rx_buf_len;
-
-  rx_buf_len = vec_len (em->rx_buf);
-  rx_fifo = e->fifo;
-  session = &em->sessions[rx_fifo->client_session_index];
-  tx_fifo = session->server_tx_fifo;
-
-  max_dequeue = svm_fifo_max_dequeue (rx_fifo);
-  /* Allow enqueuing of a new event */
-  svm_fifo_unset_event (rx_fifo);
-
-  if (PREDICT_FALSE (max_dequeue == 0))
-    {
-      return;
-    }
+  max_dequeue = svm_fifo_max_dequeue (s->rx_fifo);
+  if (PREDICT_FALSE (!max_dequeue))
+    return;
 
-  /* Read the max_dequeue */
   do
     {
-      max_transfer = clib_min (rx_buf_len, max_dequeue);
-      n_read = svm_fifo_dequeue_nowait (rx_fifo, max_transfer, em->rx_buf);
+      /* The options here are to limit ourselves to max_dequeue or read
+       * even the data that was enqueued while we were dequeueing and which
+       * now has an rx event in the mq. Either of the two work. */
+      to_dequeue = clib_min (max_dequeue, vec_len (em->rx_buf));
+      n_read = app_recv_stream_raw (s->rx_fifo, em->rx_buf, to_dequeue,
+				    0 /* clear evt */ , 0 /* peek */ );
       if (n_read > 0)
 	{
 	  max_dequeue -= n_read;
-	  session->bytes_received += n_read;
+	  s->bytes_received += n_read;
 	}
+      else
+	break;
 
       /* Reflect if a non-drop session */
-      if (!em->drop_packets && n_read > 0)
+      if (!em->no_return && n_read > 0)
 	{
 	  offset = 0;
 	  do
 	    {
-	      rv = svm_fifo_enqueue_nowait (tx_fifo, n_read,
-					    &em->rx_buf[offset]);
-	      if (rv > 0)
+	      n_sent = app_send_stream ((app_session_t *) s,
+					&em->rx_buf[offset],
+					n_read, SVM_Q_WAIT);
+	      if (n_sent > 0)
 		{
-		  n_read -= rv;
-		  offset += rv;
+		  n_read -= n_sent;
+		  offset += n_sent;
 		}
 	    }
-	  while ((rv <= 0 || n_read > 0) && !em->time_to_stop);
-
-	  /* If event wasn't set, add one */
-	  if (svm_fifo_set_event (tx_fifo))
-	    {
-	      /* Fabricate TX event, send to vpp */
-	      evt.fifo = tx_fifo;
-	      evt.event_type = FIFO_EVENT_APP_TX;
-
-	      q = em->vpp_event_queue;
-	      svm_queue_add (q, (u8 *) & evt, 1 /* do wait for mutex */ );
-	    }
+	  while ((n_sent <= 0 || n_read > 0) && !em->time_to_stop);
 	}
     }
-  while ((n_read < 0 || max_dequeue > 0) && !em->time_to_stop);
+  while (max_dequeue > 0 && !em->time_to_stop);
 }
 
-void
-server_handle_event_queue (echo_main_t * em)
+static void
+server_handle_mq (echo_main_t * em)
 {
-  session_fifo_event_t _e, *e = &_e;
+  svm_msg_q_msg_t msg;
+  session_event_t *e;
 
   while (1)
     {
-      svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
+      svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
-	  server_handle_fifo_event_rx (em, e);
+	  server_handle_rx (em, e);
 	  break;
-
-	case FIFO_EVENT_DISCONNECT:
-	  return;
-
 	default:
-	  clib_warning ("unknown event type %d", e->event_type);
+	  handle_mq_event (e);
 	  break;
 	}
       if (PREDICT_FALSE (em->time_to_stop == 1))
@@ -1188,6 +1175,7 @@ server_handle_event_queue (echo_main_t * em)
 	  em->time_to_print_stats = 0;
 	  fformat (stdout, "%d connections\n", pool_elts (em->sessions));
 	}
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
     }
 }
 
@@ -1196,7 +1184,7 @@ server_send_listen (echo_main_t * em)
 {
   vl_api_bind_uri_t *bmp;
   bmp = vl_msg_api_alloc (sizeof (*bmp));
-  memset (bmp, 0, sizeof (*bmp));
+  clib_memset (bmp, 0, sizeof (*bmp));
 
   bmp->_vl_msg_id = ntohs (VL_API_BIND_URI);
   bmp->client_index = em->my_client_index;
@@ -1223,7 +1211,7 @@ server_send_unbind (echo_main_t * em)
   vl_api_unbind_uri_t *ump;
 
   ump = vl_msg_api_alloc (sizeof (*ump));
-  memset (ump, 0, sizeof (*ump));
+  clib_memset (ump, 0, sizeof (*ump));
 
   ump->_vl_msg_id = ntohs (VL_API_UNBIND_URI);
   ump->client_index = em->my_client_index;
@@ -1246,14 +1234,14 @@ server_unbind (echo_main_t * em)
 void
 server_run (echo_main_t * em)
 {
-  session_t *session;
+  echo_session_t *session;
   int i;
 
   /* $$$$ hack preallocation */
   for (i = 0; i < 200000; i++)
     {
       pool_get (em->sessions, session);
-      memset (session, 0, sizeof (*session));
+      clib_memset (session, 0, sizeof (*session));
     }
   for (i = 0; i < 200000; i++)
     pool_put_index (em->sessions, i);
@@ -1266,7 +1254,7 @@ server_run (echo_main_t * em)
     return;
 
   /* Enter handle event loop */
-  server_handle_event_queue (em);
+  server_handle_mq (em);
 
   /* Cleanup */
   server_send_unbind (em);
@@ -1281,18 +1269,26 @@ vl_api_disconnect_session_reply_t_handler (vl_api_disconnect_session_reply_t *
 					   mp)
 {
   echo_main_t *em = &echo_main;
-  session_t *session;
+  uword *p;
 
   if (mp->retval)
     {
       clib_warning ("vpp complained about disconnect: %d",
 		    ntohl (mp->retval));
+      return;
     }
 
   em->state = STATE_START;
-  session = pool_elt_at_index (em->sessions, em->connected_session_index);
-  if (session)
-    session_print_stats (em, session);
+
+  p = hash_get (em->session_index_by_vpp_handles, mp->handle);
+  if (p)
+    {
+      hash_unset (em->session_index_by_vpp_handles, mp->handle);
+    }
+  else
+    {
+      clib_warning ("couldn't find session key %llx", mp->handle);
+    }
 }
 
 static void
@@ -1314,11 +1310,7 @@ static void
 #define foreach_tcp_echo_msg                            		\
 _(BIND_URI_REPLY, bind_uri_reply)                       		\
 _(UNBIND_URI_REPLY, unbind_uri_reply)                   		\
-_(ACCEPT_SESSION, accept_session)                       		\
-_(CONNECT_SESSION_REPLY, connect_session_reply)         		\
-_(DISCONNECT_SESSION, disconnect_session)               		\
 _(DISCONNECT_SESSION_REPLY, disconnect_session_reply)   		\
-_(RESET_SESSION, reset_session)                         		\
 _(APPLICATION_ATTACH_REPLY, application_attach_reply)   		\
 _(APPLICATION_DETACH_REPLY, application_detach_reply)			\
 _(MAP_ANOTHER_SEGMENT, map_another_segment)				\
@@ -1342,38 +1334,32 @@ tcp_echo_api_hookup (echo_main_t * em)
 int
 main (int argc, char **argv)
 {
-  int i_am_master = 1, drop_packets = 0, test_return_packets = 0;
+  int i_am_server = 1, test_return_packets = 0;
   echo_main_t *em = &echo_main;
+  svm_fifo_segment_main_t *sm = &em->segment_main;
   unformat_input_t _argv, *a = &_argv;
   u8 *chroot_prefix;
-  u8 *heap, *uri = 0;
+  u8 *uri = 0;
   u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
-  u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234";
+  u8 *connect_uri = (u8 *) "tcp://6.0.1.1/1234";
   u64 bytes_to_send = 64 << 10, mbytes;
   char *app_name;
   u32 tmp;
-  mheap_t *h;
 
-  clib_mem_init (0, 256 << 20);
-
-  heap = clib_mem_get_per_cpu_heap ();
-  h = mheap_header (heap);
-
-  /* make the main heap thread-safe */
-  h->flags |= MHEAP_FLAG_THREAD_SAFE;
-
-  vec_validate (em->rx_buf, 128 << 10);
+  clib_mem_init_thread_safe (0, 256 << 20);
 
+  clib_memset (em, 0, sizeof (*em));
   em->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
-
   em->my_pid = getpid ();
   em->configured_segment_size = 1 << 20;
   em->socket_name = 0;
   em->use_sock_api = 1;
+  em->fifo_size = 64 << 10;
+  em->n_clients = 1;
 
   clib_time_init (&em->clib_time);
   init_error_string_table (em);
-  svm_fifo_segment_main_init (0x200000000ULL, 20);
+  svm_fifo_segment_main_init (sm, HIGH_SEGMENT_BASEVA, 20);
   unformat_init_command_line (a, argv);
 
   while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
@@ -1388,12 +1374,12 @@ main (int argc, char **argv)
 	em->configured_segment_size = tmp << 20;
       else if (unformat (a, "segment-size %dG", &tmp))
 	em->configured_segment_size = tmp << 30;
-      else if (unformat (a, "master"))
-	i_am_master = 1;
-      else if (unformat (a, "slave"))
-	i_am_master = 0;
-      else if (unformat (a, "drop"))
-	drop_packets = 1;
+      else if (unformat (a, "server"))
+	i_am_server = 1;
+      else if (unformat (a, "client"))
+	i_am_server = 0;
+      else if (unformat (a, "no-return"))
+	em->no_return = 1;
       else if (unformat (a, "test"))
 	test_return_packets = 1;
       else if (unformat (a, "mbytes %lld", &mbytes))
@@ -1408,9 +1394,13 @@ main (int argc, char **argv)
 	;
       else if (unformat (a, "use-svm-api"))
 	em->use_sock_api = 0;
+      else if (unformat (a, "fifo-size %d", &tmp))
+	em->fifo_size = tmp << 10;
+      else if (unformat (a, "nclients %d", &em->n_clients))
+	;
       else
 	{
-	  fformat (stderr, "%s: usage [master|slave]\n");
+	  fformat (stderr, "%s: usage [master|slave]\n", argv[0]);
 	  exit (1);
 	}
     }
@@ -1429,17 +1419,18 @@ main (int argc, char **argv)
       em->connect_uri = format (0, "%s%c", connect_uri, 0);
     }
 
-  em->i_am_master = i_am_master;
-  em->segment_main = &svm_fifo_segment_main;
-  em->drop_packets = drop_packets;
+  em->i_am_master = i_am_server;
   em->test_return_packets = test_return_packets;
   em->bytes_to_send = bytes_to_send;
   em->time_to_stop = 0;
+  vec_validate (em->rx_buf, 4 << 20);
+  vec_validate (em->client_thread_handles, em->n_clients - 1);
+  vec_validate (em->thread_args, em->n_clients - 1);
 
   setup_signal_handlers ();
   tcp_echo_api_hookup (em);
 
-  app_name = i_am_master ? "tcp_echo_server" : "tcp_echo_client";
+  app_name = i_am_server ? "tcp_echo_server" : "tcp_echo_client";
   if (connect_to_vpp (app_name) < 0)
     {
       svm_region_exit ();
@@ -1447,11 +1438,14 @@ main (int argc, char **argv)
       exit (1);
     }
 
-  if (i_am_master == 0)
-    client_run (em);
+  if (i_am_server == 0)
+    clients_run (em);
   else
     server_run (em);
 
+  /* Make sure detach finishes */
+  wait_for_state_change (em, STATE_DETACHED);
+
   disconnect_from_vpp (em);
   exit (0);
 }