1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
</*
* mc_test.c: test program for vlib mc
*
* Copyright (c) 2010 Cisco and/or its affiliates.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at:
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include <vlib/vlib.h>
#include <vlib/unix/mc_socket.h>
#include <vppinfra/random.h>
typedef struct
{
u32 min_n_msg_bytes;
u32 max_n_msg_bytes;
u32 tx_serial;
u32 rx_serial;
u32 seed;
u32 verbose;
u32 validate;
u32 window_size;
f64 min_delay, max_delay;
f64 n_packets_to_send;
} mc_test_main_t;
always_inline u32
choose_msg_size (mc_test_main_t * tm)
{
u32 r = tm->min_n_msg_bytes;
if (tm->max_n_msg_bytes > tm->min_n_msg_bytes)
r +=
random_u32 (&tm->seed) % (1 + tm->max_n_msg_bytes -
tm->min_n_msg_bytes);
return r;
}
static mc_test_main_t mc_test_main;
static void
serialize_test_msg (serialize_main_t * m, va_list * va)
{
mc_test_main_t *tm = &mc_test_main;
u32 n_bytes = choose_msg_size (tm);
u8 *msg;
int i;
serialize_integer (m, n_bytes, sizeof (n_bytes));
msg = serialize_get (m, n_bytes);
for (i = 0; i < n_bytes; i++)
msg[i] = i + tm->tx_serial;
tm->tx_serial += n_bytes;
}
static void
unserialize_test_msg (serialize_main_t * m, va_list * va)
{
mc_test_main_t *tm = &mc_test_main;
u32 i, n_bytes, dump_msg = tm->verbose;
u8 *p;
unserialize_integer (m, &n_bytes, sizeof (n_bytes));
p = unserialize_get (m, n_bytes);
if (tm->validate)
for (i = 0; i < n_bytes; i++)
if (p[i] != ((tm->rx_serial + i) & 0xff))
{
clib_warning ("corrupt msg at offset %d", i);
dump_msg = 1;
break;
}
if (dump_msg)
clib_warning ("got %d bytes, %U", n_bytes, format_hex_bytes, p, n_bytes);
tm->rx_serial += n_bytes;
}
MC_SERIALIZE_MSG (test_msg, static) =
{
.name = "test_msg",.serialize = serialize_test_msg,.unserialize =
unserialize_test_msg,};
#define SERIALIZE 1
#define EVENT_JOIN_STREAM 10
#define EVENT_SEND_DATA 11
static void
test_rx_callback (mc_main_t * mcm,
mc_stream_t * stream,
mc_peer_id_t peer_id, u32 buffer_index)
{
if (SERIALIZE)
{
return mc_unserialize (mcm, stream, buffer_index);
}
else
{
#if DEBUG > 1
vlib_main_t *vm = mcm->vlib_main;
vlib_buffer_t *b = vlib_get_buffer (vm, buffer_index);
u8 *dp = vlib_buffer_get_current (b);
fformat (stdout, "RX from %U %U\n",
stream->transport->format_peer_id, peer_id,
format_hex_bytes, dp, tm->n_msg_bytes);
#endif
}
}
static u8 *
test_snapshot_callback (mc_main_t * mcm,
u8 * data_vector, u32 last_global_sequence_processed)
{
if (SERIALIZE)
{
serialize_main_t m;
/* Append serialized data to data vector. */
serialize_open_vector (&m, data_vector);
m.stream.current_buffer_index = vec_len (data_vector);
return serialize_close_vector (&m);
}
else
return format (data_vector,
"snapshot, last global seq 0x%x",
last_global_sequence_processed);
}
static void
test_handle_snapshot_callback (mc_main_t * mcm, u8 * data, u32 n_data_bytes)
{
if (SERIALIZE)
{
serialize_main_t s;
unserialize_open_data (&s, data, n_data_bytes);
}
else
clib_warning ("snapshot `%*s'", n_data_bytes, data);
}
static mc_socket_main_t mc_socket_main;
static uword
mc_test_process (vlib_main_t * vm,
vlib_node_runtime_t * node, vlib_frame_t * f)
{
mc_test_main_t *tm = &mc_test_main;
mc_socket_main_t *msm = &mc_socket_main;
mc_main_t *mcm = &msm->mc_main;
uword event_type, *event_data = 0;
u32 data_serial = 0, stream_index;
f64 delay;
mc_stream_config_t config;
clib_error_t *error;
int i;
char *intfcs[] = { "eth1", "eth0", "ce" };
memset (&config, 0, sizeof (config));
config.name = "test";
config.window_size = tm->window_size;
config.rx_buffer = test_rx_callback;
config.catchup_snapshot = test_snapshot_callback;
config.catchup = test_handle_snapshot_callback;
stream_index = ~0;
msm->multicast_tx_ip4_address_host_byte_order = 0xefff0100;
msm->base_multicast_udp_port_host_byte_order = 0xffab;
error = mc_socket_main_init (&mc_socket_main, intfcs, ARRAY_LEN (intfcs));
if (error)
{
clib_error_report (error);
exit (1);
}
mcm->we_can_be_relay_master = 1;
while (1)
{
vlib_process_wait_for_event (vm);
event_type = vlib_process_get_events (vm, &event_data);
switch (event_type)
{
case EVENT_JOIN_STREAM:
stream_index = mc_stream_join (mcm, &config);
break;
case EVENT_SEND_DATA:
{
f64 times[2];
if (stream_index == ~0)
stream_index = mc_stream_join (mcm, &config);
times[0] = vlib_time_now (vm);
for (i = 0; i < event_data[0]; i++)
{
u32 bi;
if (SERIALIZE)
{
mc_serialize_stream (mcm, stream_index, &test_msg,
data_serial);
}
else
{
u8 *mp;
mp = mc_get_vlib_buffer (vm, sizeof (mp[0]), &bi);
mp[0] = data_serial;
mc_stream_send (mcm, stream_index, bi);
}
if (tm->min_delay > 0)
{
delay =
tm->min_delay +
random_f64 (&tm->seed) * (tm->max_delay -
tm->min_delay);
vlib_process_suspend (vm, delay);
}
data_serial++;
}
times[1] = vlib_time_now (vm);
clib_warning ("done sending %d; %.4e per sec",
event_data[0],
(f64) event_data[0] / (times[1] - times[0]));
break;
}
default:
clib_warning ("bug");
break;
}
if (event_data)
_vec_len (event_data) = 0;
}
}
/* *INDENT-OFF* */
VLIB_REGISTER_NODE (mc_test_process_node, static) =
{
.function = mc_test_process,.type = VLIB_NODE_TYPE_PROCESS,.name =
"mc-test-process",};
/* *INDENT-ON* */
static clib_error_t *
mc_test_command (vlib_main_t * vm,
unformat_input_t * input, vlib_cli_command_t * cmd)
{
f64 npkts = 10;
if (unformat (input, "join"))
{
vlib_cli_output (vm, "Join stream...\n");
vlib_process_signal_event (vm, mc_test_process_node.index,
EVENT_JOIN_STREAM, 0);
return 0;
}
else if (unformat (input, "send %f", &npkts) || unformat (input, "send"))
{
vlib_process_signal_event (vm, mc_test_process_node.index,
EVENT_SEND_DATA, (uword) npkts);
vlib_cli_output (vm, "Send %.0f pkts...\n", npkts);
return 0;
}
else
return unformat_parse_error (input);
}
/* *INDENT-OFF* */
VLIB_CLI_COMMAND (test_mc_command, static) =
{
.path = "test mc",.short_help = "Test mc command",.function =
mc_test_command,};
/* *INDENT-ON* */
static clib_error_t *
mc_show_command (vlib_main_t * vm,
unformat_input_t * input, vlib_cli_command_t * cmd)
{
mc_main_t *mcm = &mc_socket_main.mc_main;
vlib_cli_output (vm, "%U", format_mc_main, mcm);
return 0;
}
/* *INDENT-OFF* */
VLIB_CLI_COMMAND (show_mc_command, static) =
{
.path = "show mc",.short_help = "Show mc command",.function =
mc_show_command,};
/* *INDENT-ON* */
static clib_error_t *
mc_clear_command (vlib_main_t * vm,
unformat_input_t * input, vlib_cli_command_t * cmd)
{
mc_main_t *mcm = &mc_socket_main.mc_main;
mc_clear_stream_stats (mcm);
return 0;
}
/* *INDENT-OFF* */
VLIB_CLI_COMMAND (clear_mc_command, static) =
{
.path = "clear mc",.short_help = "Clear mc command",.function =
mc_clear_command,};
/* *INDENT-ON* */
static clib_error_t *
mc_config (vlib_main_t * vm, unformat_input_t * input)
{
mc_test_main_t *tm = &mc_test_main;
mc_socket_main_t *msm = &mc_socket_main;
clib_error_t *error = 0;
tm->min_n_msg_bytes = 4;
tm->max_n_msg_bytes = 4;
tm->window_size = 8;
tm->seed = getpid ();
tm->verbose = 0;
tm->validate = 1;
tm->min_delay = 10e-6;
tm->max_delay = 10e-3;
tm->n_packets_to_send = 0;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
if (unformat (input, "interface %s", &msm->multicast_interface_name))
;
else if (unformat (input, "n-bytes %d", &tm->max_n_msg_bytes))
tm->min_n_msg_bytes = tm->max_n_msg_bytes;
else if (unformat (input, "max-n-bytes %d", &tm->max_n_msg_bytes))
;
else if (unformat (input, "min-n-bytes %d", &tm->min_n_msg_bytes))
;
else if (unformat (input, "seed %d", &tm->seed))
;
else if (unformat (input, "window %d", &tm->window_size))
;
else if (unformat (input, "verbose"))
tm->verbose = 1;
else if (unformat (input, "no-validate"))
tm->validate = 0;
else if (unformat (input, "min-delay %f", &tm->min_delay))
;
else if (unformat (input, "max-delay %f", &tm->max_delay))
;
else if (unformat (input, "no-delay"))
tm->min_delay = tm->max_delay = 0;
else if (unformat (input, "n-packets %f", &tm->n_packets_to_send))
;
else
return clib_error_return (0, "unknown input `%U'",
format_unformat_error, input);
}
if (tm->n_packets_to_send > 0)
vlib_process_signal_event (vm, mc_test_process_node.index,
EVENT_SEND_DATA,
(uword) tm->n_packets_to_send);
return error;
}
VLIB_CONFIG_FUNCTION (mc_config, "mc");
/*
* fd.io coding-style-patch-verification: ON
*
* Local Variables:
* eval: (c-set-style "gnu")
* End:
*/
|
