tcp_echo: support multiple connections

Change-Id: I6d8e1351e088728f7363550a0fc117256cae2841
Signed-off-by: Florin Coras <fcoras@cisco.com>

1 files changed, 250 insertions, 218 deletions

diff --git a/src/tests/vnet/session/tcp_echo.c b/src/tests/vnet/session/tcp_echo.c
index d3d4c0ad0a6..e8868ede265 100644
--- a/src/tests/vnet/session/tcp_echo.c
+++ b/src/tests/vnet/session/tcp_echo.c
@@ -38,13 +38,23 @@
 #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;
 
+  svm_queue_t *vpp_evt_q;
+
   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;
 
@@ -54,7 +64,8 @@ typedef enum
   STATE_ATTACHED,
   STATE_READY,
   STATE_DISCONNECTING,
-  STATE_FAILED
+  STATE_FAILED,
+  STATE_DETACHED
 } connection_state_t;
 
 typedef struct
@@ -113,12 +124,21 @@ 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. */
@@ -240,7 +260,7 @@ wait_for_state_change (echo_main_t * em, connection_state_t state)
       if (em->time_to_stop == 1)
 	return 0;
     }
-  clib_warning ("timeout waiting for STATE_READY");
+  clib_warning ("timeout waiting for state %d", state);
   return -1;
 }
 
@@ -264,6 +284,7 @@ application_send_attach (echo_main_t * em)
   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);
@@ -309,7 +330,7 @@ application_detach (echo_main_t * em)
   bmp->context = ntohl (0xfeedface);
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
 
-  clib_warning ("Sent detach");
+  DBG ("%s", "Sent detach");
 }
 
 static int
@@ -415,6 +436,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
@@ -593,107 +615,164 @@ vl_api_reset_session_t_handler (vl_api_reset_session_t * mp)
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & rmp);
 }
 
-void
-client_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
+static void
+test_recv_bytes (session_t * s, u8 * rx_buf, u32 n_read)
 {
-  svm_fifo_t *rx_fifo;
-  int n_read, bytes, i;
+  int i;
+  for (i = 0; i < n_read; i++)
+    {
+      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));
+	}
+    }
+}
 
-  rx_fifo = e->fifo;
+static void
+recv_test_chunk (echo_main_t * em, session_t * s, u8 * rx_buf)
+{
+  svm_fifo_t *rx_fifo = s->server_rx_fifo;
+  u32 n_read_now, n_to_read;
+  int n_read;
 
-  bytes = svm_fifo_max_dequeue (rx_fifo);
-  /* Allow enqueuing of new event */
+  n_to_read = svm_fifo_max_dequeue (rx_fifo);
   svm_fifo_unset_event (rx_fifo);
 
-  /* 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;
-	    }
-	}
+      n_read_now = clib_min (vec_len (rx_buf), n_to_read);
+      n_read = svm_fifo_dequeue_nowait (rx_fifo, n_read_now, rx_buf);
+      if (n_read <= 0)
+	break;
 
+      if (n_read_now != n_read)
+	clib_warning ("huh?");
+
+      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;
     }
-  while (bytes > 0);
+  while (n_to_read > 0);
 }
 
 void
-client_handle_event_queue (echo_main_t * em)
+client_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e,
+			     u8 * rx_buf)
 {
-  session_fifo_event_t _e, *e = &_e;;
+  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;
+  s = pool_elt_at_index (em->sessions, e->fifo->client_session_index);
+  recv_test_chunk (em, s, rx_buf);
+}
 
-    case FIFO_EVENT_DISCONNECT:
-      return;
+static void
+send_test_chunk (echo_main_t * em, session_t * s)
+{
+  u64 test_buf_len, bytes_this_chunk, test_buf_offset;
+  svm_fifo_t *tx_fifo = s->server_tx_fifo;
+  u8 *test_data = em->connect_test_data;
+  u32 enq_space, min_chunk = 16 << 10;
+  session_fifo_event_t evt;
+  int written;
+
+  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);
+  enq_space = svm_fifo_max_enqueue (tx_fifo);
+  if (enq_space < clib_min (bytes_this_chunk, min_chunk))
+    return;
 
-    default:
-      clib_warning ("unknown event type %d", e->event_type);
-      break;
+  bytes_this_chunk = clib_min (bytes_this_chunk, enq_space);
+  written = svm_fifo_enqueue_nowait (tx_fifo, bytes_this_chunk,
+				     test_data + test_buf_offset);
+
+  if (written > 0)
+    {
+      s->bytes_to_send -= written;
+      s->bytes_sent += written;
+
+      if (svm_fifo_set_event (tx_fifo))
+	{
+	  /* Fabricate TX event, send to vpp */
+	  evt.fifo = tx_fifo;
+	  evt.event_type = FIFO_EVENT_APP_TX;
+	  svm_queue_add (s->vpp_evt_q, (u8 *) & evt, 0 /* wait for mutex */ );
+	}
     }
 }
 
+/*
+ * Rx/Tx polling thread per connection
+ */
 static void *
+client_thread_fn (void *arg)
+{
+  echo_main_t *em = &echo_main;
+  static u8 *rx_buf = 0;
+  u32 session_index = *(u32 *) arg;
+  session_t *s;
+
+  vec_validate (rx_buf, 1 << 20);
+
+  while (!em->time_to_stop && em->state != STATE_READY)
+    ;
+
+  s = pool_elt_at_index (em->sessions, session_index);
+  while (!em->time_to_stop)
+    {
+      send_test_chunk (em, s);
+      recv_test_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);
+}
+
+/*
+ * Rx thread that handles all connections.
+ *
+ * Not used.
+ */
+void *
 client_rx_thread_fn (void *arg)
 {
   session_fifo_event_t _e, *e = &_e;
   echo_main_t *em = &echo_main;
+  static u8 *rx_buf = 0;
 
-  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);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
-	  client_handle_fifo_event_rx (em, e);
+	  client_handle_fifo_event_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;
     }
   pthread_exit (0);
 }
 
-
 static void
 vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
 {
@@ -710,21 +789,13 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
       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 *);
 
   /*
    * Setup session
    */
 
   pool_get (em->sessions, session);
+  memset (session, 0, sizeof (*session));
   session_index = session - em->sessions;
 
   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
@@ -736,110 +807,28 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
   session->server_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_queue_t *);
 
-  /* Save handle */
-  em->connected_session_index = session_index;
-  em->state = STATE_READY;
-
-  /* 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);
+  /*
+   * 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);
-      rv = VNET_API_ERROR_SYSCALL_ERROR_1;
-    }
-}
-
-static void
-send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
-{
-  u32 bytes_to_snd, enq_space, min_chunk = 16 << 10;
-  u8 *test_data = em->connect_test_data;
-  u64 bytes_sent = 0;
-  int test_buf_offset = 0;
-  session_fifo_event_t evt;
-  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)
-    {
-      enq_space = svm_fifo_max_enqueue (tx_fifo);
-      if (enq_space < clib_min (bytes_to_snd, min_chunk))
-	continue;
-      rv = svm_fifo_enqueue_nowait (tx_fifo,
-				    clib_min (bytes_to_snd, enq_space),
-				    test_data + test_buf_offset);
-      if (rv > 0)
-	{
-	  bytes_to_snd -= rv;
-	  test_buf_offset += rv;
-	  bytes_sent += rv;
-
-	  if (svm_fifo_set_event (tx_fifo))
-	    {
-	      /* Fabricate TX event, send to vpp */
-	      evt.fifo = tx_fifo;
-	      evt.event_type = FIFO_EVENT_APP_TX;
-	      svm_queue_add (em->vpp_event_queue, (u8 *) & evt,
-			     0 /* do wait for mutex */ );
-	    }
-	}
-    }
-}
-
-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 = pool_elt_at_index (em->sessions, em->connected_session_index);
-  tx_fifo = session->server_tx_fifo;
-
-  ASSERT (vec_len (test_data) > 0);
-
-  vec_validate (em->rx_buf, vec_len (test_data) - 1);
-  n_iterations = em->bytes_to_send / vec_len (test_data);
-
-  for (i = 0; i < n_iterations; i++)
-    {
-      send_test_chunk (em, tx_fifo, mypid, 0);
-      if (em->time_to_stop)
-	break;
-    }
-
-  leftover = em->bytes_to_send % vec_len (test_data);
-  if (leftover)
-    send_test_chunk (em, tx_fifo, mypid, leftover);
-
-  if (!em->no_return)
-    {
-      f64 timeout = clib_time_now (&em->clib_time) + 10;
-
-      /* 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;
-	    }
-	}
+      return;
     }
 
-  em->time_to_stop = 1;
+  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));
 }
 
 void
@@ -856,70 +845,97 @@ client_send_connect (echo_main_t * em)
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & cmp);
 }
 
-int
-client_connect (echo_main_t * em)
-{
-  client_send_connect (em);
-  if (wait_for_state_change (em, STATE_READY))
-    {
-      clib_warning ("Connect failed");
-      return -1;
-    }
-  return 0;
-}
-
 void
-client_send_disconnect (echo_main_t * em)
+client_send_disconnect (echo_main_t * em, session_t * s)
 {
-  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;
+  dmp->handle = s->vpp_session_handle;
   vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & dmp);
 }
 
 int
-client_disconnect (echo_main_t * em)
+client_disconnect (echo_main_t * em, session_t * s)
 {
-  client_send_disconnect (em);
-  clib_warning ("Sent disconnect");
-  if (wait_for_state_change (em, STATE_START))
-    {
-      clib_warning ("Disconnect failed");
-      return -1;
-    }
+  client_send_disconnect (em, s);
+  pool_put (em->sessions, s);
+  memset (s, 0xfe, sizeof (*s));
   return 0;
 }
 
 static void
-client_run (echo_main_t * em)
+clients_run (echo_main_t * em)
 {
+  session_fifo_event_t _e, *e = &_e;
+  f64 start_time, deltat, timeout = 100.0;
+  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)
+    ;
+
+  if (em->n_clients_connected != em->n_clients)
     {
-      application_detach (em);
+      clib_warning ("failed to initialize all connections");
       return;
     }
 
-  /* 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;
+  /*
+   * 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)
+    svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_NOWAIT, 0);
+
+
+  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);
 }
 
@@ -1106,17 +1122,15 @@ server_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
 
   rx_buf_len = vec_len (em->rx_buf);
   rx_fifo = e->fifo;
-  session = &em->sessions[rx_fifo->client_session_index];
+  session = pool_elt_at_index (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;
-    }
+  if (PREDICT_FALSE (!max_dequeue))
+    return;
 
   /* Read the max_dequeue */
   do
@@ -1127,6 +1141,7 @@ server_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
 	{
 	  max_dequeue -= n_read;
 	  session->bytes_received += n_read;
+	  session->bytes_to_receive -= n_read;
 	}
 
       /* Reflect if a non-drop session */
@@ -1281,18 +1296,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
@@ -1364,12 +1387,14 @@ main (int argc, char **argv)
 
   vec_validate (em->rx_buf, 128 << 10);
 
+  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);
@@ -1410,6 +1435,8 @@ main (int argc, char **argv)
 	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");
@@ -1436,6 +1463,8 @@ main (int argc, char **argv)
   em->test_return_packets = test_return_packets;
   em->bytes_to_send = bytes_to_send;
   em->time_to_stop = 0;
+  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);
@@ -1449,10 +1478,13 @@ main (int argc, char **argv)
     }
 
   if (i_am_server == 0)
-    client_run (em);
+    clients_run (em);
   else
     server_run (em);
 
+  /* Make sure detach finishes */
+  wait_for_state_change (em, STATE_DETACHED);
+
   disconnect_from_vpp (em);
   exit (0);
 }