session: api to add new transport types
[vpp.git] / src / plugins / hs_apps / sapi / vpp_echo.c
index 019860a..d6f0b28 100644 (file)
@@ -150,7 +150,11 @@ print_global_json_stats (echo_main_t * em)
   fformat (stdout, "  \"end_evt_missing\": \"%s\",\n",
           end_evt_missing ? "True" : "False");
   fformat (stdout, "  \"rx_data\": %lld,\n", em->stats.rx_total);
-  fformat (stdout, "  \"tx_rx\": %lld,\n", em->stats.tx_total);
+  fformat (stdout, "  \"tx_data\": %lld,\n", em->stats.tx_total);
+  fformat (stdout, "  \"rx_bits_per_second\": %.1f,\n",
+          em->stats.rx_total * 8 / deltat);
+  fformat (stdout, "  \"tx_bits_per_second\": %.1f,\n",
+          em->stats.tx_total * 8 / deltat);
   fformat (stdout, "  \"closing\": {\n");
   fformat (stdout, "    \"reset\": { \"q\": %d, \"s\": %d },\n",
           em->stats.reset_count.q, em->stats.reset_count.s);
@@ -158,15 +162,15 @@ print_global_json_stats (echo_main_t * em)
           em->stats.close_count.q, em->stats.close_count.s);
   fformat (stdout, "    \"send evt\": { \"q\": %d, \"s\": %d },\n",
           em->stats.active_count.q, em->stats.active_count.s);
-  fformat (stdout, "    \"clean\": { \"q\": %d, \"s\": %d }\n",
+  fformat (stdout, "    \"clean\": { \"q\": %d, \"s\": %d },\n",
           em->stats.clean_count.q, em->stats.clean_count.s);
-  fformat (stdout, "    \"accepted\": { \"q\": %d, \"s\": %d }\n",
+  fformat (stdout, "    \"accepted\": { \"q\": %d, \"s\": %d },\n",
           em->stats.accepted_count.q, em->stats.accepted_count.s);
   fformat (stdout, "    \"connected\": { \"q\": %d, \"s\": %d }\n",
           em->stats.connected_count.q, em->stats.connected_count.s);
-  fformat (stdout, "  }\n");
+  fformat (stdout, "  },\n");
   fformat (stdout, "  \"results\": {\n");
-  fformat (stdout, "    \"has_failed\": \"%d\"\n", em->has_failed);
+  fformat (stdout, "    \"has_failed\": \"%d\",\n", em->has_failed);
   fformat (stdout, "    \"fail_descr\": \"%v\"\n", em->fail_descr);
   fformat (stdout, "  }\n");
   fformat (stdout, "}\n");
@@ -244,10 +248,10 @@ echo_update_count_on_session_close (echo_main_t * em, echo_session_t * s)
            s->bytes_received, s->bytes_received + s->bytes_to_receive,
            echo_format_session, s, s->bytes_sent,
            s->bytes_sent + s->bytes_to_send);
-  clib_atomic_fetch_add (&em->stats.tx_total, s->bytes_sent);
-  clib_atomic_fetch_add (&em->stats.rx_total, s->bytes_received);
 
-  if (PREDICT_FALSE (em->stats.rx_total == em->stats.rx_expected))
+  if (PREDICT_FALSE
+      ((em->stats.rx_total == em->stats.rx_expected)
+       && (em->stats.tx_total == em->stats.tx_expected)))
     echo_notify_event (em, ECHO_EVT_LAST_BYTE);
 }
 
@@ -313,6 +317,7 @@ recv_data_chunk (echo_main_t * em, echo_session_t * s, u8 * rx_buf)
 
   s->bytes_received += n_read;
   s->bytes_to_receive -= n_read;
+  clib_atomic_fetch_add (&em->stats.rx_total, n_read);
   return n_read;
 }
 
@@ -321,6 +326,8 @@ send_data_chunk (echo_session_t * s, u8 * tx_buf, int offset, int len)
 {
   int n_sent;
   int bytes_this_chunk = clib_min (s->bytes_to_send, len - offset);
+  echo_main_t *em = &echo_main;
+
   if (!bytes_this_chunk)
     return 0;
   n_sent = app_send ((app_session_t *) s, tx_buf + offset,
@@ -329,6 +336,7 @@ send_data_chunk (echo_session_t * s, u8 * tx_buf, int offset, int len)
     return 0;
   s->bytes_to_send -= n_sent;
   s->bytes_sent += n_sent;
+  clib_atomic_fetch_add (&em->stats.tx_total, n_sent);
   return n_sent;
 }
 
@@ -413,7 +421,7 @@ echo_handle_data (echo_main_t * em, echo_session_t * s, u8 * rx_buf)
     {
       if (n_sent || n_read)
        s->idle_cycles = 0;
-      else if (s->idle_cycles++ == 1e7)
+      else if (s->idle_cycles++ == LOG_EVERY_N_IDLE_CYCLES)
        {
          s->idle_cycles = 0;
          ECHO_LOG (2, "Idle client TX:%dB RX:%dB", s->bytes_to_send,
@@ -494,6 +502,8 @@ session_unlisten_handler (session_unlisten_reply_msg_t * mp)
   echo_main_t *em = &echo_main;
 
   ls = echo_get_session_from_handle (em, mp->handle);
+  if (!ls)
+    return;
   em->proto_cb_vft->cleanup_cb (ls, 0 /* parent_died */ );
   ls->session_state = ECHO_SESSION_STATE_CLOSED;
   if (--em->listen_session_cnt == 0)
@@ -527,6 +537,21 @@ session_bound_handler (session_bound_msg_t * mp)
     em->proto_cb_vft->bound_uri_cb (mp, listen_session);
 }
 
+static int
+echo_segment_is_not_mapped (u64 segment_handle)
+{
+  echo_main_t *em = &echo_main;
+  uword *segment_present;
+  ECHO_LOG (3, "Check if segment mapped 0x%lx...", segment_handle);
+  clib_spinlock_lock (&em->segment_handles_lock);
+  segment_present = hash_get (em->shared_segment_handles, segment_handle);
+  clib_spinlock_unlock (&em->segment_handles_lock);
+  if (segment_present != 0)
+    return 0;
+  ECHO_LOG (2, "Segment not mapped (0x%lx)", segment_handle);
+  return -1;
+}
+
 static void
 session_accepted_handler (session_accepted_msg_t * mp)
 {
@@ -542,7 +567,7 @@ session_accepted_handler (session_accepted_msg_t * mp)
                 "Unknown listener handle 0x%lx", mp->listener_handle);
       return;
     }
-  if (wait_for_segment_allocation (mp->segment_handle))
+  if (echo_segment_is_not_mapped (mp->segment_handle))
     {
       ECHO_FAIL (ECHO_FAIL_ACCEPTED_WAIT_FOR_SEG_ALLOC,
                 "accepted wait_for_segment_allocation errored");
@@ -611,7 +636,7 @@ session_connected_handler (session_connected_msg_t * mp)
     }
 
   session = echo_session_new (em);
-  if (wait_for_segment_allocation (mp->segment_handle))
+  if (echo_segment_is_not_mapped (mp->segment_handle))
     {
       ECHO_FAIL (ECHO_FAIL_CONNECTED_WAIT_FOR_SEG_ALLOC,
                 "connected wait_for_segment_allocation errored");
@@ -704,6 +729,72 @@ session_reset_handler (session_reset_msg_t * mp)
   app_send_ctrl_evt_to_vpp (s->vpp_evt_q, app_evt);
 }
 
+static void
+add_segment_handler (session_app_add_segment_msg_t * mp)
+{
+  fifo_segment_main_t *sm = &echo_main.segment_main;
+  fifo_segment_create_args_t _a, *a = &_a;
+  echo_main_t *em = &echo_main;
+  int *fds = 0, i;
+  char *seg_name = (char *) mp->segment_name;
+  u64 segment_handle = mp->segment_handle;
+
+  if (mp->fd_flags & SESSION_FD_F_MEMFD_SEGMENT)
+    {
+      vec_validate (fds, 1);
+      if (vl_socket_client_recv_fd_msg (fds, 1, 5))
+       {
+         ECHO_FAIL (ECHO_FAIL_VL_API_RECV_FD_MSG,
+                    "vl_socket_client_recv_fd_msg failed");
+         goto failed;
+       }
+
+      if (echo_ssvm_segment_attach (seg_name, SSVM_SEGMENT_MEMFD, fds[0]))
+       {
+         ECHO_FAIL (ECHO_FAIL_VL_API_SVM_FIFO_SEG_ATTACH,
+                    "svm_fifo_segment_attach ('%s') "
+                    "failed on SSVM_SEGMENT_MEMFD", seg_name);
+         goto failed;
+       }
+      vec_free (fds);
+    }
+  else
+    {
+      clib_memset (a, 0, sizeof (*a));
+      a->segment_name = seg_name;
+      a->segment_size = mp->segment_size;
+      /* Attach to the segment vpp created */
+      if (fifo_segment_attach (sm, a))
+       {
+         ECHO_FAIL (ECHO_FAIL_VL_API_FIFO_SEG_ATTACH,
+                    "fifo_segment_attach ('%s') failed", seg_name);
+         goto failed;
+       }
+    }
+  echo_segment_handle_add_del (em, segment_handle, 1 /* add */ );
+  ECHO_LOG (2, "Mapped segment 0x%lx", segment_handle);
+  return;
+
+failed:
+  for (i = 0; i < vec_len (fds); i++)
+    close (fds[i]);
+  vec_free (fds);
+}
+
+static void
+del_segment_handler (session_app_del_segment_msg_t * mp)
+{
+  echo_main_t *em = &echo_main;
+  echo_segment_handle_add_del (em, mp->segment_handle, 0 /* add */ );
+  ECHO_LOG (2, "Unmaped segment 0x%lx", mp->segment_handle);
+}
+
+static void
+cleanup_handler (session_cleanup_msg_t * mp)
+{
+  ECHO_LOG (1, "Cleanup confirmed for 0x%lx", mp->handle);
+}
+
 static void
 handle_mq_event (session_event_t * e)
 {
@@ -723,6 +814,15 @@ handle_mq_event (session_event_t * e)
     case SESSION_CTRL_EVT_UNLISTEN_REPLY:
       return session_unlisten_handler ((session_unlisten_reply_msg_t *)
                                       e->data);
+    case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
+      add_segment_handler ((session_app_add_segment_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
+      del_segment_handler ((session_app_del_segment_msg_t *) e->data);
+      break;
+    case SESSION_CTRL_EVT_CLEANUP:
+      cleanup_handler ((session_cleanup_msg_t *) e->data);
+      break;
     case SESSION_IO_EVT_RX:
       break;
     default:
@@ -753,6 +853,34 @@ echo_process_rpcs (echo_main_t * em)
     }
 }
 
+static inline void
+echo_print_periodic_stats (echo_main_t * em)
+{
+  f64 delta, now = clib_time_now (&em->clib_time);
+  echo_stats_t _st, *st = &_st;
+  echo_stats_t *lst = &em->last_stat_sampling;
+  delta = now - em->last_stat_sampling_ts;
+  if (delta < em->periodic_stats_delta)
+    return;
+
+  clib_memcpy_fast (st, &em->stats, sizeof (*st));
+  if (st->rx_total - lst->rx_total)
+    clib_warning ("RX: %U", echo_format_bytes_per_sec,
+                 (st->rx_total - lst->rx_total) / delta);
+  if (st->tx_total - lst->tx_total)
+    clib_warning ("TX: %U", echo_format_bytes_per_sec,
+                 (st->tx_total - lst->tx_total) / delta);
+  if (st->connected_count.q - lst->connected_count.q)
+    clib_warning ("conn: %d/s",
+                 st->connected_count.q - lst->connected_count.q);
+  if (st->accepted_count.q - lst->accepted_count.q)
+    clib_warning ("accept: %d/s",
+                 st->accepted_count.q - lst->accepted_count.q);
+
+  clib_memcpy_fast (lst, st, sizeof (*st));
+  em->last_stat_sampling_ts = now;
+}
+
 static void *
 echo_mq_thread_fn (void *arg)
 {
@@ -776,6 +904,9 @@ echo_mq_thread_fn (void *arg)
 
   while (em->state < STATE_DETACHED && !em->time_to_stop)
     {
+      if (em->periodic_stats_delta)
+       echo_print_periodic_stats (em);
+
       svm_msg_q_lock (mq);
       if (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 1))
        {
@@ -913,6 +1044,7 @@ print_usage_and_exit (void)
           "  rx-results-diff     Rx results different to pass test\n"
           "  tx-results-diff     Tx results different to pass test\n"
           "  json                Output global stats in json\n"
+          "  stats N             Output stats evry N secs\n"
           "  log=N               Set the log level to [0: no output, 1:errors, 2:log]\n"
           "  crypto [engine]     Set the crypto engine [openssl, vpp, picotls, mbedtls]\n"
           "\n"
@@ -920,7 +1052,7 @@ print_usage_and_exit (void)
           "  nthreads N          Use N busy loop threads for data [in addition to main & msg queue]\n"
           "  TX=1337[K|M|G]|RX   Send 1337 [K|M|G]bytes, use TX=RX to reflect the data\n"
           "  RX=1337[K|M|G]      Expect 1337 [K|M|G]bytes\n" "\n");
-  for (i = 0; i < TRANSPORT_N_PROTO; i++)
+  for (i = 0; i < vec_len (em->available_proto_cb_vft); i++)
     {
       echo_proto_cb_vft_t *vft = em->available_proto_cb_vft[i];
       if (vft && vft->print_usage_cb)
@@ -937,7 +1069,7 @@ echo_process_each_proto_opts (unformat_input_t * a)
 {
   echo_main_t *em = &echo_main;
   int i, rv;
-  for (i = 0; i < TRANSPORT_N_PROTO; i++)
+  for (i = 0; i < vec_len (em->available_proto_cb_vft); i++)
     {
       echo_proto_cb_vft_t *vft = em->available_proto_cb_vft[i];
       if (vft && vft->process_opts_cb)
@@ -951,7 +1083,7 @@ static void
 echo_set_each_proto_defaults_before_opts (echo_main_t * em)
 {
   int i;
-  for (i = 0; i < TRANSPORT_N_PROTO; i++)
+  for (i = 0; i < vec_len (em->available_proto_cb_vft); i++)
     {
       echo_proto_cb_vft_t *vft = em->available_proto_cb_vft[i];
       if (vft && vft->set_defaults_before_opts_cb)
@@ -1047,6 +1179,8 @@ echo_process_opts (int argc, char **argv)
        em->tx_results_diff = 1;
       else if (unformat (a, "json"))
        em->output_json = 1;
+      else if (unformat (a, "stats %d", &em->periodic_stats_delta))
+       ;
       else if (unformat (a, "wait-for-gdb"))
        em->wait_for_gdb = 1;
       else if (unformat (a, "log=%d", &em->log_lvl))
@@ -1311,6 +1445,7 @@ exit_on_error:
   else
     print_global_stats (em);
   vec_free (em->fail_descr);
+  vec_free (em->available_proto_cb_vft);
   exit (em->has_failed);
 }