1 /*
2 * Copyright (c) 2015 Red Hat, Inc.
3 *
4 * All rights reserved.
5 *
6 * Author: Christine Caulfield <ccaulfie@redhat.com>
7 *
8 * This software licensed under BSD license, the text of which follows:
9 *
10 * Redistribution and use in source and binary forms, with or without
11 * modification, are permitted provided that the following conditions are met:
12 *
13 * - Redistributions of source code must retain the above copyright notice,
14 * this list of conditions and the following disclaimer.
15 * - Redistributions in binary form must reproduce the above copyright notice,
16 * this list of conditions and the following disclaimer in the documentation
17 * and/or other materials provided with the distribution.
18 * - Neither the name of the MontaVista Software, Inc. nor the names of its
19 * contributors may be used to endorse or promote products derived from this
20 * software without specific prior written permission.
21 *
22 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
23 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
24 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
25 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
26 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
27 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
28 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
29 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
30 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
31 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
32 * THE POSSIBILITY OF SUCH DAMAGE.
33 */
34
35 #include <stdio.h>
36 #include <stdlib.h>
37 #include <string.h>
38 #include <signal.h>
39 #include <unistd.h>
40 #include <errno.h>
41 #include <time.h>
42 #include <sys/time.h>
43 #include <sys/types.h>
44 #include <sys/socket.h>
45 #include <sys/select.h>
46 #include <sys/uio.h>
47 #include <sys/un.h>
48 #include <netinet/in.h>
49 #include <arpa/inet.h>
50 #include <pthread.h>
51 #include <zlib.h>
52 #include <libgen.h>
53
54 #include <qb/qblog.h>
55 #include <qb/qbutil.h>
56
57 #include <corosync/corotypes.h>
58 #include <corosync/cpg.h>
59
60 static cpg_handle_t handle;
61
62 static pthread_t thread;
63
64 #ifndef timersub
65 #define timersub(a, b, result) \
66 do { \
67 (result)->tv_sec = (a)->tv_sec - (b)->tv_sec; \
68 (result)->tv_usec = (a)->tv_usec - (b)->tv_usec; \
69 if ((result)->tv_usec < 0) { \
70 --(result)->tv_sec; \
71 (result)->tv_usec += 1000000; \
72 } \
73 } while (0)
74 #endif /* timersub */
75
76 static int alarm_notice;
77 #define ONE_MEG 1048576
78 #define DATASIZE (ONE_MEG*20)
79 static char data[DATASIZE];
80 static int send_counter = 0;
81 static int do_syslog = 0;
82 static int quiet = 0;
83 static volatile int stopped;
84
85 // stats
86 static unsigned int length_errors=0;
87 static unsigned int crc_errors=0;
88 static unsigned int sequence_errors=0;
89 static unsigned int packets_sent=0;
90 static unsigned int packets_recvd=0;
91 static unsigned int send_retries=0;
92 static unsigned int send_fails=0;
93
cpg_bm_confchg_fn(cpg_handle_t handle_in,const struct cpg_name * group_name,const struct cpg_address * member_list,size_t member_list_entries,const struct cpg_address * left_list,size_t left_list_entries,const struct cpg_address * joined_list,size_t joined_list_entries)94 static void cpg_bm_confchg_fn (
95 cpg_handle_t handle_in,
96 const struct cpg_name *group_name,
97 const struct cpg_address *member_list, size_t member_list_entries,
98 const struct cpg_address *left_list, size_t left_list_entries,
99 const struct cpg_address *joined_list, size_t joined_list_entries)
100 {
101 }
102
103 static unsigned int g_recv_count;
104 static unsigned int g_recv_length;
105 static unsigned int g_write_size;
106 static int g_recv_counter = 0;
107
cpg_bm_deliver_fn(cpg_handle_t handle_in,const struct cpg_name * group_name,uint32_t nodeid,uint32_t pid,void * msg,size_t msg_len)108 static void cpg_bm_deliver_fn (
109 cpg_handle_t handle_in,
110 const struct cpg_name *group_name,
111 uint32_t nodeid,
112 uint32_t pid,
113 void *msg,
114 size_t msg_len)
115 {
116 int *value = msg;
117 uLong crc=0;
118 uLong recv_crc = value[1] & 0xFFFFFFFF;
119
120 packets_recvd++;
121 g_recv_length = msg_len;
122
123 // Basic check, packets should all be the right size
124 if (g_write_size && (msg_len != g_write_size)) {
125 length_errors++;
126 fprintf(stderr, "%s: message sizes don't match. got %zu, expected %u\n", group_name->value, msg_len, g_write_size);
127 if (do_syslog) {
128 syslog(LOG_ERR, "%s: message sizes don't match. got %zu, expected %u\n", group_name->value, msg_len, g_write_size);
129 }
130 }
131
132 // Sequence counters are incrementing in step?
133 if (*value != g_recv_counter) {
134 sequence_errors++;
135 fprintf(stderr, "%s: counters don't match. got %d, expected %d\n", group_name->value, *value, g_recv_counter);
136 if (do_syslog) {
137 syslog(LOG_ERR, "%s: counters don't match. got %d, expected %d\n", group_name->value, *value, g_recv_counter);
138 }
139 // Catch up or we'll be printing errors for ever
140 g_recv_counter = *value +1;
141 } else {
142 g_recv_counter++;
143 }
144
145 // Check crc
146 crc = crc32(0, NULL, 0);
147 crc = crc32(crc, (Bytef *)&value[2], msg_len-sizeof(int)*2) & 0xFFFFFFFF;
148 if (crc != recv_crc) {
149 crc_errors++;
150 fprintf(stderr, "%s: CRCs don't match. got %lx, expected %lx\n", group_name->value, recv_crc, crc);
151 if (do_syslog) {
152 syslog(LOG_ERR, "%s: CRCs don't match. got %lx, expected %lx\n", group_name->value, recv_crc, crc);
153 }
154 }
155
156 g_recv_count++;
157
158 }
159
160 static cpg_model_v1_data_t model1_data = {
161 .cpg_deliver_fn = cpg_bm_deliver_fn,
162 .cpg_confchg_fn = cpg_bm_confchg_fn,
163 };
164
165 static cpg_callbacks_t callbacks = {
166 .cpg_deliver_fn = cpg_bm_deliver_fn,
167 .cpg_confchg_fn = cpg_bm_confchg_fn
168 };
169
170 static struct cpg_name group_name = {
171 .value = "cpghum",
172 .length = 7
173 };
174
cpg_test(cpg_handle_t handle_in,int write_size,int delay_time,int print_time)175 static void cpg_test (
176 cpg_handle_t handle_in,
177 int write_size,
178 int delay_time,
179 int print_time)
180 {
181 struct timeval tv1, tv2, tv_elapsed;
182 struct iovec iov;
183 unsigned int res;
184 int i;
185 unsigned int *dataint = (unsigned int *)data;
186 uLong crc;
187
188 alarm_notice = 0;
189 iov.iov_base = data;
190 iov.iov_len = write_size;
191
192 g_recv_count = 0;
193 alarm (print_time);
194
195 gettimeofday (&tv1, NULL);
196 do {
197 dataint[0] = send_counter++;
198 for (i=2; i<(DATASIZE-sizeof(int)*2)/4; i++) {
199 dataint[i] = rand();
200 }
201 crc = crc32(0, NULL, 0);
202 dataint[1] = crc32(crc, (Bytef*)&dataint[2], write_size-sizeof(int)*2);
203 resend:
204 res = cpg_mcast_joined (handle_in, CPG_TYPE_AGREED, &iov, 1);
205 if (res == CS_ERR_TRY_AGAIN) {
206 usleep(10000);
207 send_retries++;
208 goto resend;
209 }
210 if (res != CS_OK) {
211 fprintf(stderr, "send failed: %d\n", res);
212 send_fails++;
213 }
214 else {
215 packets_sent++;
216 }
217 usleep(delay_time*1000);
218 } while (alarm_notice == 0 && (res == CS_OK || res == CS_ERR_TRY_AGAIN) && stopped == 0);
219 gettimeofday (&tv2, NULL);
220 timersub (&tv2, &tv1, &tv_elapsed);
221
222 if (!quiet) {
223 printf ("%s: %5d message%s received, ", group_name.value, g_recv_count, g_recv_count==1?"":"s");
224 printf ("%5d bytes per write\n", write_size);
225 }
226
227 }
228
sigalrm_handler(int num)229 static void sigalrm_handler (int num)
230 {
231 alarm_notice = 1;
232 }
233
sigint_handler(int num)234 static void sigint_handler (int num)
235 {
236 stopped = 1;
237 }
238
dispatch_thread(void * arg)239 static void* dispatch_thread (void *arg)
240 {
241 cpg_dispatch (handle, CS_DISPATCH_BLOCKING);
242 return NULL;
243 }
244
usage(char * cmd)245 static void usage(char *cmd)
246 {
247 fprintf(stderr, "%s [OPTIONS]\n", cmd);
248 fprintf(stderr, "\n");
249 fprintf(stderr, "%s sends CPG messages to all registered users of the CPG.\n", cmd);
250 fprintf(stderr, "The messages have a sequence number and a CRC so that missing or\n");
251 fprintf(stderr, "corrupted messages will be detected and reported.\n");
252 fprintf(stderr, "\n");
253 fprintf(stderr, "%s can also be asked to simply listen for (and check) packets\n", cmd);
254 fprintf(stderr, "so that there is another node in the cluster connected to the CPG.\n");
255 fprintf(stderr, "\n");
256 fprintf(stderr, "When -l is present, packet size is only checked if specified by -w or -W\n");
257 fprintf(stderr, "and it, obviously, must match that of the sender.\n");
258 fprintf(stderr, "\n");
259 fprintf(stderr, "Multiple copies, in different CPGs, can also be run on the same or\n");
260 fprintf(stderr, "different nodes by using the -n option.\n");
261 fprintf(stderr, "\n");
262 fprintf(stderr, "%s can't handle more than 1 sender in the same CPG as it messes with the\n", cmd);
263 fprintf(stderr, "sequence numbers.\n");
264 fprintf(stderr, "\n");
265 fprintf(stderr, " -w Write size in Kbytes, default 4\n");
266 fprintf(stderr, " -W Write size in bytes, default 4096\n");
267 fprintf(stderr, " -n CPG name to use, default 'cpghum'\n");
268 fprintf(stderr, " -d Delay between sending packets (mS), default 1000\n");
269 fprintf(stderr, " -r Number of repetitions, default 100\n");
270 fprintf(stderr, " -p Delay between printing output(S), default 10s\n");
271 fprintf(stderr, " -l Listen and check CRCs only, don't send (^C to quit)\n");
272 fprintf(stderr, " -m cpg_initialise() model. Default 1.\n");
273 fprintf(stderr, " -s Also send errors to syslog (for daemon log correlation).\n");
274 fprintf(stderr, " -q Quiet. Don't print messages every 10 seconds (see also -p)\n");
275 fprintf(stderr, "\n");
276 }
277
main(int argc,char * argv[])278 int main (int argc, char *argv[]) {
279 int i;
280 unsigned int res;
281 uint32_t maxsize;
282 int opt;
283 int bs;
284 int write_size = 4096;
285 int delay_time = 1000;
286 int repetitions = 100;
287 int print_time = 10;
288 int have_size = 0;
289 int listen_only = 0;
290 int model = 1;
291
292 while ( (opt = getopt(argc, argv, "qlsn:d:r:p:m:w:W:")) != -1 ) {
293 switch (opt) {
294 case 'w': // Write size in K
295 bs = atoi(optarg);
296 if (bs > 0) {
297 write_size = bs*1024;
298 have_size = 1;
299 }
300 break;
301 case 'W': // Write size in bytes
302 bs = atoi(optarg);
303 if (bs > 0) {
304 write_size = bs;
305 have_size = 1;
306 }
307 break;
308 case 'n':
309 if (strlen(optarg) >= CPG_MAX_NAME_LENGTH) {
310 fprintf(stderr, "CPG name too long\n");
311 exit(1);
312 }
313
314 strcpy(group_name.value, optarg);
315 group_name.length = strlen(group_name.value);
316 break;
317 case 'd':
318 delay_time = atoi(optarg);
319 break;
320 case 'r':
321 repetitions = atoi(optarg);
322 break;
323 case 'p':
324 print_time = atoi(optarg);
325 break;
326 case 'l':
327 listen_only = 1;
328 break;
329 case 's':
330 do_syslog = 1;
331 break;
332 case 'q':
333 quiet = 1;
334 break;
335 case 'm':
336 model = atoi(optarg);
337 if (model < 0 || model > 1) {
338 fprintf(stderr, "%s: Model must be 0-1\n", argv[0]);
339 exit(1);
340 }
341 break;
342 case '?':
343 usage(basename(argv[0]));
344 exit(0);
345 }
346 }
347
348 qb_log_init("cpghum", LOG_USER, LOG_EMERG);
349 qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_FALSE);
350 qb_log_filter_ctl(QB_LOG_STDERR, QB_LOG_FILTER_ADD,
351 QB_LOG_FILTER_FILE, "*", LOG_DEBUG);
352 qb_log_ctl(QB_LOG_STDERR, QB_LOG_CONF_ENABLED, QB_TRUE);
353
354 g_write_size = write_size;
355
356 signal (SIGALRM, sigalrm_handler);
357 signal (SIGINT, sigint_handler);
358 switch (model) {
359 case 0:
360 res = cpg_initialize (&handle, &callbacks);
361 break;
362 case 1:
363 res = cpg_model_initialize (&handle, CPG_MODEL_V1, (cpg_model_data_t *)&model1_data, NULL);
364 break;
365 default:
366 res=999; // can't get here but it keeps the compiler happy
367 break;
368 }
369
370 if (res != CS_OK) {
371 printf ("cpg_initialize failed with result %d\n", res);
372 exit (1);
373 }
374 pthread_create (&thread, NULL, dispatch_thread, NULL);
375
376 res = cpg_join (handle, &group_name);
377 if (res != CS_OK) {
378 printf ("cpg_join failed with result %d\n", res);
379 exit (1);
380 }
381
382 if (listen_only) {
383 int secs = 0;
384 if (!quiet) {
385 printf("-- Listening on CPG %s\n", group_name.value);
386 printf("-- Ignore any starting \"counters don't match\" error while we catch up\n");
387 }
388
389 /* Only check packet size if specified on the command-line */
390 if (!have_size) {
391 g_write_size = 0;
392 }
393
394 while (!stopped) {
395 sleep(1);
396 if (++secs > print_time && !quiet) {
397 printf ("%s: %5d message%s received. %d bytes\n", group_name.value, g_recv_count, g_recv_count==1?"":"s", g_recv_length);
398 secs = 0;
399 g_recv_count = 0;
400 }
401 }
402 }
403 else {
404 cpg_max_atomic_msgsize_get (handle, &maxsize);
405 if ( write_size > maxsize) {
406 fprintf(stderr, "INFO: packet size (%d) is larger than the maximum atomic size (%d), libcpg will fragment\n",
407 write_size, maxsize);
408 }
409 for (i = 0; i < repetitions && !stopped; i++) {
410 cpg_test (handle, write_size, delay_time, print_time);
411 signal (SIGALRM, sigalrm_handler);
412 }
413 }
414
415 res = cpg_finalize (handle);
416 if (res != CS_OK) {
417 printf ("cpg_finalize failed with result %d\n", res);
418 exit (1);
419 }
420
421 printf("\n");
422 printf("Stats:\n");
423 if (!listen_only) {
424 printf(" packets sent: %d\n", packets_sent);
425 printf(" send failures: %d\n", send_fails);
426 printf(" send retries: %d\n", send_retries);
427 }
428 if (have_size) {
429 printf(" length errors: %d\n", length_errors);
430 }
431 printf(" packets recvd: %d\n", packets_recvd);
432 printf(" sequence errors: %d\n", sequence_errors);
433 printf(" crc errors: %d\n", crc_errors);
434 printf("\n");
435 return (0);
436 }
437