2 * builtin_client.c - vpp built-in tcp client/connect code
4 * Copyright (c) 2017 by Cisco and/or its affiliates.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at:
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
18 #include <vnet/vnet.h>
19 #include <vnet/plugin/plugin.h>
20 #include <vnet/tcp/builtin_client.h>
22 #include <vlibapi/api.h>
23 #include <vlibmemory/api.h>
24 #include <vlibsocket/api.h>
25 #include <vpp/app/version.h>
27 #define TCP_BUILTIN_CLIENT_DBG (0)
30 signal_evt_to_cli_i (int *code)
32 tclient_main_t *tm = &tclient_main;
33 ASSERT (vlib_get_thread_index () == 0);
34 vlib_process_signal_event (tm->vlib_main, tm->cli_node_index, *code, 0);
38 signal_evt_to_cli (int code)
40 if (vlib_get_thread_index () != 0)
41 vl_api_rpc_call_main_thread (signal_evt_to_cli_i, (u8 *) & code,
44 signal_evt_to_cli_i (&code);
48 send_test_chunk (tclient_main_t * tm, session_t * s)
50 u8 *test_data = tm->connect_test_data;
53 session_fifo_event_t evt;
54 static int serial_number = 0;
58 ASSERT (vec_len (test_data) > 0);
60 test_buf_offset = s->bytes_sent % vec_len (test_data);
61 bytes_this_chunk = vec_len (test_data) - test_buf_offset;
63 bytes_this_chunk = bytes_this_chunk < s->bytes_to_send
64 ? bytes_this_chunk : s->bytes_to_send;
66 txf = s->server_tx_fifo;
67 rv = svm_fifo_enqueue_nowait (txf, bytes_this_chunk,
68 test_data + test_buf_offset);
70 /* If we managed to enqueue data... */
73 /* Account for it... */
74 s->bytes_to_send -= rv;
77 if (TCP_BUILTIN_CLIENT_DBG)
80 ELOG_TYPE_DECLARE (e) =
82 .format = "tx-enq: xfer %d bytes, sent %u remain %u",
83 .format_args = "i4i4i4",
90 ed = ELOG_DATA (&vlib_global_main.elog_main, e);
92 ed->data[1] = s->bytes_sent;
93 ed->data[2] = s->bytes_to_send;
96 /* Poke the session layer */
97 if (svm_fifo_set_event (txf))
99 /* Fabricate TX event, send to vpp */
101 evt.event_type = FIFO_EVENT_APP_TX;
102 evt.event_id = serial_number++;
104 if (unix_shared_memory_queue_add
105 (tm->vpp_event_queue[txf->master_thread_index], (u8 *) & evt,
106 0 /* do wait for mutex */ ))
107 clib_warning ("could not enqueue event");
113 receive_test_chunk (tclient_main_t * tm, session_t * s)
115 svm_fifo_t *rx_fifo = s->server_rx_fifo;
116 int n_read, test_bytes = 0;
117 u32 my_thread_index = vlib_get_thread_index ();
119 /* Allow enqueuing of new event */
120 // svm_fifo_unset_event (rx_fifo);
124 n_read = svm_fifo_dequeue_nowait (rx_fifo,
125 vec_len (tm->rx_buf[my_thread_index]),
126 tm->rx_buf[my_thread_index]);
130 n_read = svm_fifo_max_dequeue (rx_fifo);
131 svm_fifo_dequeue_drop (rx_fifo, n_read);
136 if (TCP_BUILTIN_CLIENT_DBG)
139 ELOG_TYPE_DECLARE (e) =
141 .format = "rx-deq: %d bytes",
149 ed = ELOG_DATA (&vlib_global_main.elog_main, e);
150 ed->data[0] = n_read;
156 for (i = 0; i < n_read; i++)
158 if (tm->rx_buf[my_thread_index][i]
159 != ((s->bytes_received + i) & 0xff))
161 clib_warning ("read %d error at byte %lld, 0x%x not 0x%x",
162 n_read, s->bytes_received + i,
163 tm->rx_buf[my_thread_index][i],
164 ((s->bytes_received + i) & 0xff));
168 s->bytes_to_receive -= n_read;
169 s->bytes_received += n_read;
174 builtin_client_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
175 vlib_frame_t * frame)
177 tclient_main_t *tm = &tclient_main;
178 int my_thread_index = vlib_get_thread_index ();
182 u32 *connection_indices;
183 u32 *connections_this_batch;
184 u32 nconnections_this_batch;
186 connection_indices = tm->connection_index_by_thread[my_thread_index];
187 connections_this_batch =
188 tm->connections_this_batch_by_thread[my_thread_index];
190 if ((tm->run_test == 0) ||
191 ((vec_len (connection_indices) == 0)
192 && vec_len (connections_this_batch) == 0))
195 /* Grab another pile of connections */
196 if (PREDICT_FALSE (vec_len (connections_this_batch) == 0))
198 nconnections_this_batch =
199 clib_min (tm->connections_per_batch, vec_len (connection_indices));
201 ASSERT (nconnections_this_batch > 0);
202 vec_validate (connections_this_batch, nconnections_this_batch - 1);
203 clib_memcpy (connections_this_batch,
204 connection_indices + vec_len (connection_indices)
205 - nconnections_this_batch,
206 nconnections_this_batch * sizeof (u32));
207 _vec_len (connection_indices) -= nconnections_this_batch;
210 if (PREDICT_FALSE (tm->prev_conns != tm->connections_per_batch
211 && tm->prev_conns == vec_len (connections_this_batch)))
214 tm->prev_conns = vec_len (connections_this_batch);
215 if (tm->repeats == 500000)
217 clib_warning ("stuck clients");
222 tm->prev_conns = vec_len (connections_this_batch);
226 for (i = 0; i < vec_len (connections_this_batch); i++)
230 sp = pool_elt_at_index (tm->sessions, connections_this_batch[i]);
232 if (sp->bytes_to_send > 0)
234 send_test_chunk (tm, sp);
237 if (sp->bytes_to_receive > 0)
239 receive_test_chunk (tm, sp);
242 if (PREDICT_FALSE (delete_session == 1))
244 u32 index, thread_index;
247 __sync_fetch_and_add (&tm->tx_total, sp->bytes_sent);
248 __sync_fetch_and_add (&tm->rx_total, sp->bytes_received);
250 stream_session_parse_handle (sp->vpp_session_handle,
251 &index, &thread_index);
252 s = stream_session_get_if_valid (index, thread_index);
256 vnet_disconnect_args_t _a, *a = &_a;
257 a->handle = stream_session_handle (s);
258 a->app_index = tm->app_index;
259 vnet_disconnect_session (a);
261 vec_delete (connections_this_batch, 1, i);
263 __sync_fetch_and_add (&tm->ready_connections, -1);
266 clib_warning ("session AWOL?");
268 /* Kick the debug CLI process */
269 if (tm->ready_connections == 0)
271 signal_evt_to_cli (2);
276 tm->connection_index_by_thread[my_thread_index] = connection_indices;
277 tm->connections_this_batch_by_thread[my_thread_index] =
278 connections_this_batch;
283 VLIB_REGISTER_NODE (builtin_client_node) =
285 .function = builtin_client_node_fn,
286 .name = "builtin-tcp-client",
287 .type = VLIB_NODE_TYPE_INPUT,
288 .state = VLIB_NODE_STATE_DISABLED,
293 create_api_loopback (tclient_main_t * tm)
295 api_main_t *am = &api_main;
296 vl_shmem_hdr_t *shmem_hdr;
298 shmem_hdr = am->shmem_hdr;
299 tm->vl_input_queue = shmem_hdr->vl_input_queue;
300 tm->my_client_index =
301 vl_api_memclnt_create_internal ("tcp_test_client", tm->vl_input_queue);
306 tcp_test_clients_init (vlib_main_t * vm)
308 tclient_main_t *tm = &tclient_main;
309 vlib_thread_main_t *vtm = vlib_get_thread_main ();
313 if (create_api_loopback (tm))
316 num_threads = 1 /* main thread */ + vtm->n_threads;
318 /* Init test data. Big buffer */
319 vec_validate (tm->connect_test_data, 1024 * 1024 - 1);
320 for (i = 0; i < vec_len (tm->connect_test_data); i++)
321 tm->connect_test_data[i] = i & 0xff;
323 vec_validate (tm->rx_buf, num_threads - 1);
324 for (i = 0; i < num_threads; i++)
325 vec_validate (tm->rx_buf[i], vec_len (tm->connect_test_data) - 1);
329 vec_validate (tm->connection_index_by_thread, vtm->n_vlib_mains);
330 vec_validate (tm->connections_this_batch_by_thread, vtm->n_vlib_mains);
331 vec_validate (tm->vpp_event_queue, vtm->n_vlib_mains);
337 builtin_session_connected_callback (u32 app_index, u32 api_context,
338 stream_session_t * s, u8 is_fail)
340 tclient_main_t *tm = &tclient_main;
343 u8 thread_index = vlib_get_thread_index ();
347 clib_warning ("connection %d failed!", api_context);
348 signal_evt_to_cli (-1);
352 ASSERT (s->thread_index == thread_index);
354 if (!tm->vpp_event_queue[thread_index])
355 tm->vpp_event_queue[thread_index] =
356 session_manager_get_vpp_event_queue (thread_index);
361 clib_spinlock_lock_if_init (&tm->sessions_lock);
362 pool_get (tm->sessions, session);
363 clib_spinlock_unlock_if_init (&tm->sessions_lock);
365 memset (session, 0, sizeof (*session));
366 session_index = session - tm->sessions;
367 session->bytes_to_send = tm->bytes_to_send;
368 session->bytes_to_receive = tm->no_return ? 0ULL : tm->bytes_to_send;
369 session->server_rx_fifo = s->server_rx_fifo;
370 session->server_rx_fifo->client_session_index = session_index;
371 session->server_tx_fifo = s->server_tx_fifo;
372 session->server_tx_fifo->client_session_index = session_index;
373 session->vpp_session_handle = stream_session_handle (s);
375 vec_add1 (tm->connection_index_by_thread[thread_index], session_index);
376 __sync_fetch_and_add (&tm->ready_connections, 1);
377 if (tm->ready_connections == tm->expected_connections)
380 /* Signal the CLI process that the action is starting... */
381 signal_evt_to_cli (1);
388 builtin_session_reset_callback (stream_session_t * s)
390 if (s->session_state == SESSION_STATE_READY)
391 clib_warning ("Reset active connection %U", format_stream_session, s, 2);
392 stream_session_cleanup (s);
397 builtin_session_create_callback (stream_session_t * s)
403 builtin_session_disconnect_callback (stream_session_t * s)
405 tclient_main_t *tm = &tclient_main;
406 vnet_disconnect_args_t _a, *a = &_a;
407 a->handle = stream_session_handle (s);
408 a->app_index = tm->app_index;
409 vnet_disconnect_session (a);
414 builtin_server_rx_callback (stream_session_t * s)
420 static session_cb_vft_t builtin_clients = {
421 .session_reset_callback = builtin_session_reset_callback,
422 .session_connected_callback = builtin_session_connected_callback,
423 .session_accept_callback = builtin_session_create_callback,
424 .session_disconnect_callback = builtin_session_disconnect_callback,
425 .builtin_server_rx_callback = builtin_server_rx_callback
430 attach_builtin_test_clients_app (void)
432 tclient_main_t *tm = &tclient_main;
433 vnet_app_attach_args_t _a, *a = &_a;
434 u8 segment_name[128];
435 u32 segment_name_length, prealloc_fifos;
438 segment_name_length = ARRAY_LEN (segment_name);
440 memset (a, 0, sizeof (*a));
441 memset (options, 0, sizeof (options));
443 a->api_client_index = tm->my_client_index;
444 a->segment_name = segment_name;
445 a->segment_name_length = segment_name_length;
446 a->session_cb_vft = &builtin_clients;
448 prealloc_fifos = tm->prealloc_fifos ? tm->expected_connections : 1;
450 options[SESSION_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
451 options[SESSION_OPTIONS_SEGMENT_SIZE] = (2ULL << 32);
452 options[SESSION_OPTIONS_RX_FIFO_SIZE] = tm->fifo_size;
453 options[SESSION_OPTIONS_TX_FIFO_SIZE] = tm->fifo_size;
454 options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = tm->private_segment_count;
455 options[APP_OPTIONS_PRIVATE_SEGMENT_SIZE] = tm->private_segment_size;
456 options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = prealloc_fifos;
458 options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
460 a->options = options;
462 if (vnet_application_attach (a))
465 tm->app_index = a->app_index;
470 tclient_thread_fn (void *arg)
475 /** Start a transmit thread */
477 start_tx_pthread (tclient_main_t * tm)
479 if (tm->client_thread_handle == 0)
481 int rv = pthread_create (&tm->client_thread_handle,
483 tclient_thread_fn, 0);
486 tm->client_thread_handle = 0;
494 clients_connect (vlib_main_t * vm, u8 * uri, u32 n_clients)
496 tclient_main_t *tm = &tclient_main;
497 vnet_connect_args_t _a, *a = &_a;
499 for (i = 0; i < n_clients; i++)
501 memset (a, 0, sizeof (*a));
503 a->uri = (char *) uri;
505 a->app_index = tm->app_index;
507 vnet_connect_uri (a);
509 /* Crude pacing for call setups */
511 vlib_process_suspend (vm, 10e-6);
515 static clib_error_t *
516 test_tcp_clients_command_fn (vlib_main_t * vm,
517 unformat_input_t * input,
518 vlib_cli_command_t * cmd)
520 tclient_main_t *tm = &tclient_main;
521 vlib_thread_main_t *thread_main = vlib_get_thread_main ();
522 uword *event_data = 0, event_type;
523 u8 *default_connect_uri = (u8 *) "tcp://6.0.1.1/1234", *uri;
524 u64 tmp, total_bytes;
525 f64 test_timeout = 20.0, syn_timeout = 20.0, delta;
526 f64 time_before_connects;
528 int preallocate_sessions = 0;
532 tm->bytes_to_send = 8192;
534 tm->fifo_size = 64 << 10;
535 tm->connections_per_batch = 1000;
536 tm->private_segment_count = 0;
537 tm->private_segment_size = 0;
539 if (thread_main->n_vlib_mains > 1)
540 clib_spinlock_init (&tm->sessions_lock);
541 vec_free (tm->connect_uri);
543 while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
545 if (unformat (input, "nclients %d", &n_clients))
547 else if (unformat (input, "mbytes %lld", &tmp))
548 tm->bytes_to_send = tmp << 20;
549 else if (unformat (input, "gbytes %lld", &tmp))
550 tm->bytes_to_send = tmp << 30;
551 else if (unformat (input, "bytes %lld", &tm->bytes_to_send))
553 else if (unformat (input, "uri %s", &tm->connect_uri))
555 else if (unformat (input, "test-timeout %f", &test_timeout))
557 else if (unformat (input, "syn-timeout %f", &syn_timeout))
559 else if (unformat (input, "no-return"))
561 else if (unformat (input, "fifo-size %d", &tm->fifo_size))
562 tm->fifo_size <<= 10;
563 else if (unformat (input, "private-segment-count %d",
564 &tm->private_segment_count))
566 else if (unformat (input, "private-segment-size %U",
567 unformat_memory_size, &tmp))
569 if (tmp >= 0x100000000ULL)
570 return clib_error_return
571 (0, "private segment size %lld (%llu) too large", tmp, tmp);
572 tm->private_segment_size = tmp;
574 else if (unformat (input, "preallocate-fifos"))
575 tm->prealloc_fifos = 1;
576 else if (unformat (input, "preallocate-sessions"))
577 preallocate_sessions = 1;
579 if (unformat (input, "client-batch %d", &tm->connections_per_batch))
582 return clib_error_return (0, "unknown input `%U'",
583 format_unformat_error, input);
586 /* Store cli process node index for signalling */
587 tm->cli_node_index = vlib_get_current_process (vm)->node_runtime.node_index;
589 if (tm->is_init == 0)
591 if (tcp_test_clients_init (vm))
592 return clib_error_return (0, "failed init");
596 tm->ready_connections = 0;
597 tm->expected_connections = n_clients;
601 uri = default_connect_uri;
603 uri = tm->connect_uri;
605 #if TCP_BUILTIN_CLIENT_PTHREAD
609 vlib_worker_thread_barrier_sync (vm);
610 vnet_session_enable_disable (vm, 1 /* turn on TCP, etc. */ );
611 vlib_worker_thread_barrier_release (vm);
613 if (tm->test_client_attached == 0)
615 if (attach_builtin_test_clients_app ())
617 return clib_error_return (0, "app attach failed");
620 tm->test_client_attached = 1;
622 /* Turn on the builtin client input nodes */
623 for (i = 0; i < thread_main->n_vlib_mains; i++)
624 vlib_node_set_state (vlib_mains[i], builtin_client_node.index,
625 VLIB_NODE_STATE_POLLING);
627 if (preallocate_sessions)
629 session_t *sp __attribute__ ((unused));
630 for (i = 0; i < n_clients; i++)
631 pool_get (tm->sessions, sp);
632 for (i = 0; i < n_clients; i++)
633 pool_put_index (tm->sessions, i);
636 /* Fire off connect requests */
637 time_before_connects = vlib_time_now (vm);
638 clients_connect (vm, uri, n_clients);
640 /* Park until the sessions come up, or ten seconds elapse... */
641 vlib_process_wait_for_event_or_clock (vm, syn_timeout);
642 event_type = vlib_process_get_events (vm, &event_data);
646 vlib_cli_output (vm, "Timeout with only %d sessions active...",
647 tm->ready_connections);
651 delta = vlib_time_now (vm) - time_before_connects;
656 (vm, "%d three-way handshakes in %.2f seconds, %.2f/sec",
657 n_clients, delta, ((f64) n_clients) / delta);
660 tm->test_start_time = vlib_time_now (tm->vlib_main);
661 vlib_cli_output (vm, "Test started at %.6f", tm->test_start_time);
665 vlib_cli_output (vm, "unexpected event(1): %d", event_type);
669 /* Now wait for the sessions to finish... */
670 vlib_process_wait_for_event_or_clock (vm, test_timeout);
671 event_type = vlib_process_get_events (vm, &event_data);
675 vlib_cli_output (vm, "Timeout with %d sessions still active...",
676 tm->ready_connections);
680 tm->test_end_time = vlib_time_now (vm);
681 vlib_cli_output (vm, "Test finished at %.6f", tm->test_end_time);
685 vlib_cli_output (vm, "unexpected event(2): %d", event_type);
689 delta = tm->test_end_time - tm->test_start_time;
693 total_bytes = (tm->no_return ? tm->tx_total : tm->rx_total);
694 transfer_type = tm->no_return ? "half-duplex" : "full-duplex";
696 "%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds",
697 total_bytes, total_bytes / (1ULL << 20),
698 total_bytes / (1ULL << 30), delta);
699 vlib_cli_output (vm, "%.2f bytes/second %s",
700 ((f64) total_bytes) / (delta), transfer_type);
701 vlib_cli_output (vm, "%.4f gbit/second %s",
702 (((f64) total_bytes * 8.0) / delta / 1e9),
706 vlib_cli_output (vm, "zero delta-t?");
710 for (i = 0; i < vec_len (tm->connection_index_by_thread); i++)
712 vec_reset_length (tm->connection_index_by_thread[i]);
713 vec_reset_length (tm->connections_this_batch_by_thread[i]);
716 pool_free (tm->sessions);
718 /* Detach the application, so we can use different fifo sizes next time */
719 if (tm->test_client_attached)
721 vnet_app_detach_args_t _da, *da = &_da;
724 da->app_index = tm->app_index;
726 rv = vnet_application_detach (da);
728 vlib_cli_output (vm, "WARNING: app detach failed...");
729 tm->test_client_attached = 0;
736 VLIB_CLI_COMMAND (test_clients_command, static) =
738 .path = "test tcp clients",
739 .short_help = "test tcp clients [nclients %d] [[m|g]bytes <bytes>] "
740 "[test-timeout <time>][syn-timeout <time>][no-return][fifo-size <size>]"
741 "[private-segment-count <count>][private-segment-size <bytes>[m|g]]"
742 "[preallocate-fifos][preallocate-sessions][client-batch <batch-size>]"
743 "[uri <tcp://ip/port>]",
744 .function = test_tcp_clients_command_fn,
750 tcp_test_clients_main_init (vlib_main_t * vm)
752 tclient_main_t *tm = &tclient_main;
757 VLIB_INIT_FUNCTION (tcp_test_clients_main_init);
760 * fd.io coding-style-patch-verification: ON
763 * eval: (c-set-style "gnu")