cpghum.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758
  1. /*
  2. * Copyright (c) 2015-2017 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. #include <stdio.h>
  35. #include <stdlib.h>
  36. #include <string.h>
  37. #include <signal.h>
  38. #include <unistd.h>
  39. #include <errno.h>
  40. #include <time.h>
  41. #include <limits.h>
  42. #include <syslog.h>
  43. #include <stdarg.h>
  44. #include <sys/time.h>
  45. #include <sys/types.h>
  46. #include <sys/socket.h>
  47. #include <sys/select.h>
  48. #include <sys/uio.h>
  49. #include <sys/un.h>
  50. #include <netinet/in.h>
  51. #include <arpa/inet.h>
  52. #include <pthread.h>
  53. #include <zlib.h>
  54. #include <libgen.h>
  55. #include <corosync/corotypes.h>
  56. #include <corosync/cpg.h>
  57. static cpg_handle_t handle;
  58. static pthread_t thread;
  59. #ifndef timersub
  60. #define timersub(a, b, result) \
  61. do { \
  62. (result)->tv_sec = (a)->tv_sec - (b)->tv_sec; \
  63. (result)->tv_usec = (a)->tv_usec - (b)->tv_usec; \
  64. if ((result)->tv_usec < 0) { \
  65. --(result)->tv_sec; \
  66. (result)->tv_usec += 1000000; \
  67. } \
  68. } while (0)
  69. #endif /* timersub */
  70. static int alarm_notice;
  71. #define MAX_NODEID 65536
  72. #define ONE_MEG 1048576
  73. #define DATASIZE (ONE_MEG*20)
  74. static char data[DATASIZE];
  75. static int send_counter = 0;
  76. static int do_syslog = 0;
  77. static int quiet = 0;
  78. static int report_rtt = 0;
  79. static int abort_on_error = 0;
  80. static int machine_readable = 0;
  81. static char delimiter = ',';
  82. static int to_stderr = 0;
  83. static unsigned int g_our_nodeid;
  84. static volatile int stopped;
  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 packets_recvd1=0; /* For flood intermediates */
  92. static unsigned int send_retries=0;
  93. static unsigned int send_fails=0;
  94. static unsigned long avg_rtt=0;
  95. static unsigned long max_rtt=0;
  96. static unsigned long min_rtt=LONG_MAX;
  97. static unsigned long interim_avg_rtt=0;
  98. static unsigned long interim_max_rtt=0;
  99. static unsigned long interim_min_rtt=LONG_MAX;
  100. struct cpghum_header {
  101. unsigned int counter;
  102. unsigned int crc;
  103. unsigned int size;
  104. struct timeval timestamp;
  105. };
  106. static void cpg_bm_confchg_fn (
  107. cpg_handle_t handle_in,
  108. const struct cpg_name *group_name,
  109. const struct cpg_address *member_list, size_t member_list_entries,
  110. const struct cpg_address *left_list, size_t left_list_entries,
  111. const struct cpg_address *joined_list, size_t joined_list_entries)
  112. {
  113. }
  114. static unsigned int g_recv_count;
  115. static unsigned int g_recv_length;
  116. static int g_recv_start[MAX_NODEID+1];
  117. static int g_recv_counter[MAX_NODEID+1];
  118. static int g_recv_size[MAX_NODEID+1];
  119. static int g_log_mask = 0xFFFF;
  120. typedef enum
  121. {
  122. CPGH_LOG_INFO = 1,
  123. CPGH_LOG_PERF = 2,
  124. CPGH_LOG_RTT = 4,
  125. CPGH_LOG_STATS = 8,
  126. CPGH_LOG_ERR = 16
  127. } log_type_t;
  128. static void cpgh_print_message(int syslog_level, const char *facility_name, const char *format, va_list ap)
  129. {
  130. char msg[1024];
  131. int start = 0;
  132. if (machine_readable) {
  133. snprintf(msg, sizeof(msg), "%s%c", facility_name, delimiter);
  134. start = strlen(msg);
  135. }
  136. vsnprintf(msg+start, sizeof(msg)-start, format, ap);
  137. if (to_stderr || (syslog_level <= LOG_ERR)) {
  138. fprintf(stderr, "%s", msg);
  139. }
  140. else {
  141. printf("%s", msg);
  142. }
  143. if (do_syslog) {
  144. syslog(syslog_level, "%s", msg);
  145. }
  146. }
  147. static void cpgh_log_printf(log_type_t type, const char *format, ...)
  148. {
  149. va_list ap;
  150. if (!(type & g_log_mask)) {
  151. return;
  152. }
  153. va_start(ap, format);
  154. switch (type) {
  155. case CPGH_LOG_INFO:
  156. cpgh_print_message(LOG_INFO, "[Info]", format, ap);
  157. break;
  158. case CPGH_LOG_PERF:
  159. cpgh_print_message(LOG_INFO, "[Perf]", format, ap);
  160. break;
  161. case CPGH_LOG_RTT:
  162. cpgh_print_message(LOG_INFO, "[RTT]", format, ap);
  163. break;
  164. case CPGH_LOG_STATS:
  165. cpgh_print_message(LOG_INFO, "[Stats]", format, ap);
  166. break;
  167. case CPGH_LOG_ERR:
  168. cpgh_print_message(LOG_ERR, "[Err]", format, ap);
  169. break;
  170. default:
  171. break;
  172. }
  173. va_end(ap);
  174. }
  175. static unsigned long update_rtt(struct timeval *header_timestamp, int packet_count,
  176. unsigned long *rtt_min, unsigned long *rtt_avg, unsigned long *rtt_max)
  177. {
  178. struct timeval tv1;
  179. struct timeval rtt;
  180. unsigned long rtt_usecs;
  181. gettimeofday (&tv1, NULL);
  182. timersub(&tv1, header_timestamp, &rtt);
  183. rtt_usecs = rtt.tv_usec + rtt.tv_sec*1000000;
  184. if (rtt_usecs > *rtt_max) {
  185. *rtt_max = rtt_usecs;
  186. }
  187. if (rtt_usecs < *rtt_min) {
  188. *rtt_min = rtt_usecs;
  189. }
  190. /* Don't start the average with 0 */
  191. if (*rtt_avg == 0) {
  192. *rtt_avg = rtt_usecs;
  193. }
  194. else {
  195. *rtt_avg = ((*rtt_avg * packet_count) + rtt_usecs) / (packet_count+1);
  196. }
  197. return rtt_usecs;
  198. }
  199. static void cpg_bm_deliver_fn (
  200. cpg_handle_t handle_in,
  201. const struct cpg_name *group_name,
  202. uint32_t nodeid,
  203. uint32_t pid,
  204. void *msg,
  205. size_t msg_len)
  206. {
  207. uLong crc=0;
  208. struct cpghum_header *header = (struct cpghum_header *)msg;
  209. uLong recv_crc = header->crc & 0xFFFFFFFF;
  210. unsigned int *dataint = (unsigned int *)((char*)msg + sizeof(struct cpghum_header));
  211. unsigned int datalen;
  212. if (nodeid > MAX_NODEID) {
  213. cpgh_log_printf(CPGH_LOG_ERR, "Got message from invalid nodeid %d (too high for us). Quitting\n", nodeid);
  214. exit(1);
  215. }
  216. packets_recvd++;
  217. packets_recvd1++;
  218. g_recv_length = msg_len;
  219. datalen = header->size - sizeof(struct cpghum_header);
  220. // Report RTT first in case abort_on_error is set
  221. if (nodeid == g_our_nodeid) {
  222. unsigned long rtt_usecs;
  223. // For flood
  224. update_rtt(&header->timestamp, packets_recvd1, &interim_min_rtt, &interim_avg_rtt, &interim_max_rtt);
  225. rtt_usecs = update_rtt(&header->timestamp, g_recv_counter[nodeid], &min_rtt, &avg_rtt, &max_rtt);
  226. if (report_rtt) {
  227. if (machine_readable) {
  228. cpgh_log_printf(CPGH_LOG_RTT, "%ld%c%ld%c%ld%c%ld\n", rtt_usecs, delimiter, min_rtt, delimiter, avg_rtt, delimiter, max_rtt);
  229. }
  230. else {
  231. cpgh_log_printf(CPGH_LOG_RTT, "%s: RTT %ld uS (min/avg/max): %ld/%ld/%ld\n", group_name->value, rtt_usecs, min_rtt, avg_rtt, max_rtt);
  232. }
  233. }
  234. }
  235. // Basic check, packets should all be the right size
  236. if (msg_len != header->size) {
  237. length_errors++;
  238. cpgh_log_printf(CPGH_LOG_ERR, "%s: message sizes don't match. got %zu, expected %u from node %d\n", group_name->value, msg_len, header->size, nodeid);
  239. if (abort_on_error) {
  240. exit(2);
  241. }
  242. }
  243. g_recv_size[nodeid] = msg_len;
  244. // Sequence counters are incrementing in step?
  245. if (header->counter != g_recv_counter[nodeid]) {
  246. /* Don't report the first mismatch or a newly restarted sender, we're just catching up */
  247. if (g_recv_counter[nodeid] && header->counter) {
  248. sequence_errors++;
  249. cpgh_log_printf(CPGH_LOG_ERR, "%s: counters don't match. got %d, expected %d from node %d\n", group_name->value, header->counter, g_recv_counter[nodeid], nodeid);
  250. if (abort_on_error) {
  251. exit(2);
  252. }
  253. }
  254. else {
  255. g_recv_start[nodeid] = header->counter;
  256. }
  257. /* Catch up or we'll be printing errors for ever */
  258. g_recv_counter[nodeid] = header->counter+1;
  259. }
  260. else {
  261. g_recv_counter[nodeid]++;
  262. }
  263. /* Check crc */
  264. crc = crc32(0, NULL, 0);
  265. crc = crc32(crc, (Bytef *)dataint, datalen) & 0xFFFFFFFF;
  266. if (crc != recv_crc) {
  267. crc_errors++;
  268. cpgh_log_printf(CPGH_LOG_ERR, "%s: CRCs don't match. got %lx, expected %lx from nodeid %d\n", group_name->value, recv_crc, crc, nodeid);
  269. if (abort_on_error) {
  270. exit(2);
  271. }
  272. }
  273. g_recv_count++;
  274. }
  275. static cpg_model_v1_data_t model1_data = {
  276. .cpg_deliver_fn = cpg_bm_deliver_fn,
  277. .cpg_confchg_fn = cpg_bm_confchg_fn,
  278. };
  279. static cpg_callbacks_t callbacks = {
  280. .cpg_deliver_fn = cpg_bm_deliver_fn,
  281. .cpg_confchg_fn = cpg_bm_confchg_fn
  282. };
  283. static struct cpg_name group_name = {
  284. .value = "cpghum",
  285. .length = 7
  286. };
  287. static void set_packet(int write_size, int counter)
  288. {
  289. struct cpghum_header *header = (struct cpghum_header *)data;
  290. int i;
  291. unsigned int *dataint = (unsigned int *)(data + sizeof(struct cpghum_header));
  292. unsigned int datalen = write_size - sizeof(struct cpghum_header);
  293. struct timeval tv1;
  294. uLong crc;
  295. header->counter = counter;
  296. for (i=0; i<(datalen/4); i++) {
  297. dataint[i] = rand();
  298. }
  299. crc = crc32(0, NULL, 0);
  300. header->crc = crc32(crc, (Bytef*)&dataint[0], datalen);
  301. header->size = write_size;
  302. gettimeofday (&tv1, NULL);
  303. memcpy(&header->timestamp, &tv1, sizeof(struct timeval));
  304. }
  305. /* Basically this is cpgbench.c */
  306. static void cpg_flood (
  307. cpg_handle_t handle_in,
  308. int write_size)
  309. {
  310. struct timeval tv1, tv2, tv_elapsed;
  311. struct iovec iov;
  312. unsigned int res = CS_OK;
  313. alarm_notice = 0;
  314. iov.iov_base = data;
  315. iov.iov_len = write_size;
  316. alarm (10);
  317. packets_recvd1 = 0;
  318. interim_avg_rtt = 0;
  319. interim_max_rtt = 0;
  320. interim_min_rtt = LONG_MAX;
  321. gettimeofday (&tv1, NULL);
  322. do {
  323. if (res == CS_OK) {
  324. set_packet(write_size, send_counter);
  325. }
  326. res = cpg_mcast_joined (handle_in, CPG_TYPE_AGREED, &iov, 1);
  327. if (res == CS_OK) {
  328. /* Only increment the packet counter if it was sucessfully sent */
  329. packets_sent++;
  330. send_counter++;
  331. }
  332. else {
  333. if (res == CS_ERR_TRY_AGAIN) {
  334. send_retries++;
  335. }
  336. else {
  337. send_fails++;
  338. }
  339. }
  340. } while (!stopped && alarm_notice == 0 && (res == CS_OK || res == CS_ERR_TRY_AGAIN));
  341. gettimeofday (&tv2, NULL);
  342. timersub (&tv2, &tv1, &tv_elapsed);
  343. if (!quiet) {
  344. if (machine_readable) {
  345. cpgh_log_printf (CPGH_LOG_PERF, "%d%c%d%c%f%c%f%c%f%c%ld%c%ld%c%ld\n", packets_recvd1, delimiter, write_size, delimiter,
  346. (tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)), delimiter,
  347. ((float)packets_recvd1) / (tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)), delimiter,
  348. ((float)packets_recvd1) * ((float)write_size) / ((tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)) * 1000000.0), delimiter,
  349. interim_min_rtt, delimiter, interim_avg_rtt, delimiter, interim_max_rtt);
  350. }
  351. else {
  352. cpgh_log_printf (CPGH_LOG_PERF, "%5d messages received ", packets_recvd1);
  353. cpgh_log_printf (CPGH_LOG_PERF, "%5d bytes per write ", write_size);
  354. cpgh_log_printf (CPGH_LOG_PERF, "%7.3f Seconds runtime ",
  355. (tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)));
  356. cpgh_log_printf (CPGH_LOG_PERF, "%9.3f TP/s ",
  357. ((float)packets_recvd1) / (tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)));
  358. cpgh_log_printf (CPGH_LOG_PERF, "%7.3f MB/s ",
  359. ((float)packets_recvd1) * ((float)write_size) / ((tv_elapsed.tv_sec + (tv_elapsed.tv_usec / 1000000.0)) * 1000000.0));
  360. cpgh_log_printf (CPGH_LOG_PERF, "RTT for this size (min/avg/max) %ld/%ld/%ld\n",
  361. interim_min_rtt, interim_avg_rtt, interim_max_rtt);
  362. }
  363. }
  364. }
  365. static void cpg_test (
  366. cpg_handle_t handle_in,
  367. int write_size,
  368. int delay_time,
  369. int print_time)
  370. {
  371. struct timeval tv1, tv2, tv_elapsed;
  372. struct iovec iov;
  373. unsigned int res;
  374. alarm_notice = 0;
  375. iov.iov_base = data;
  376. iov.iov_len = write_size;
  377. g_recv_count = 0;
  378. alarm (print_time);
  379. do {
  380. send_counter++;
  381. resend:
  382. set_packet(write_size, send_counter);
  383. res = cpg_mcast_joined (handle_in, CPG_TYPE_AGREED, &iov, 1);
  384. if (res == CS_ERR_TRY_AGAIN) {
  385. usleep(10000);
  386. send_retries++;
  387. goto resend;
  388. }
  389. if (res != CS_OK) {
  390. cpgh_log_printf(CPGH_LOG_ERR, "send failed: %d\n", res);
  391. send_fails++;
  392. }
  393. else {
  394. packets_sent++;
  395. }
  396. usleep(delay_time*1000);
  397. } while (alarm_notice == 0 && (res == CS_OK || res == CS_ERR_TRY_AGAIN) && stopped == 0);
  398. gettimeofday (&tv2, NULL);
  399. timersub (&tv2, &tv1, &tv_elapsed);
  400. if (!quiet) {
  401. if (machine_readable) {
  402. cpgh_log_printf(CPGH_LOG_RTT, "%d%c%ld%c%ld%c%ld\n", 0, delimiter, min_rtt, delimiter, avg_rtt, delimiter, max_rtt);
  403. }
  404. else {
  405. cpgh_log_printf(CPGH_LOG_PERF, "%s: %5d message%s received, ", group_name.value, g_recv_count, g_recv_count==1?"":"s");
  406. cpgh_log_printf(CPGH_LOG_PERF, "%5d bytes per write. ", write_size);
  407. cpgh_log_printf(CPGH_LOG_RTT, "RTT min/avg/max: %ld/%ld/%ld\n", min_rtt, avg_rtt, max_rtt);
  408. }
  409. }
  410. }
  411. static void sigalrm_handler (int num)
  412. {
  413. alarm_notice = 1;
  414. }
  415. static void sigint_handler (int num)
  416. {
  417. stopped = 1;
  418. }
  419. static void* dispatch_thread (void *arg)
  420. {
  421. cpg_dispatch (handle, CS_DISPATCH_BLOCKING);
  422. return NULL;
  423. }
  424. static void usage(char *cmd)
  425. {
  426. fprintf(stderr, "%s [OPTIONS]\n", cmd);
  427. fprintf(stderr, "\n");
  428. fprintf(stderr, "%s sends CPG messages to all registered users of the CPG.\n", cmd);
  429. fprintf(stderr, "The messages have a sequence number and a CRC so that missing or\n");
  430. fprintf(stderr, "corrupted messages will be detected and reported.\n");
  431. fprintf(stderr, "\n");
  432. fprintf(stderr, "%s can also be asked to simply listen for (and check) packets\n", cmd);
  433. fprintf(stderr, "so that there is another node in the cluster connected to the CPG.\n");
  434. fprintf(stderr, "\n");
  435. fprintf(stderr, "Multiple copies, in different CPGs, can also be run on the same or\n");
  436. fprintf(stderr, "different nodes by using the -n option.\n");
  437. fprintf(stderr, "\n");
  438. fprintf(stderr, "%s can handle more than 1 sender in the same CPG provided they are on\n", cmd);
  439. fprintf(stderr, "different nodes.\n");
  440. fprintf(stderr, "\n");
  441. fprintf(stderr, " -w<num> Write size in Kbytes, default 4\n");
  442. fprintf(stderr, " -W<num> Write size in bytes, default 4096\n");
  443. fprintf(stderr, " -n<name> CPG name to use, default 'cpghum'\n");
  444. fprintf(stderr, " -M Write machine-readable results\n");
  445. fprintf(stderr, " -D<char> Delimiter for machine-readable results (default ',')\n");
  446. fprintf(stderr, " -E Send normal output to stderr instead of stdout\n");
  447. fprintf(stderr, " -d<num> Delay between sending packets (mS), default 1000\n");
  448. fprintf(stderr, " -r<num> Number of repetitions, default 100\n");
  449. fprintf(stderr, " -p<num> Delay between printing output (seconds), default 10s\n");
  450. fprintf(stderr, " -l Listen and check CRCs only, don't send (^C to quit)\n");
  451. fprintf(stderr, " -t Report Round Trip Times for each packet.\n");
  452. fprintf(stderr, " -m<num> cpg_initialise() model. Default 1.\n");
  453. fprintf(stderr, " -s Also send errors to syslog (for daemon log correlation).\n");
  454. fprintf(stderr, " -f Flood test CPG (cpgbench). -W starts at 64 in this case.\n");
  455. fprintf(stderr, " -a Abort on crc/length/sequence error\n");
  456. fprintf(stderr, " -q Quiet. Don't print messages every 10 seconds (see also -p)\n");
  457. fprintf(stderr, " -qq Very quiet. Don't print stats at the end\n");
  458. fprintf(stderr, "\n");
  459. fprintf(stderr, "%s exit code is 0 if no error happened, 1 on generic error and 2 on\n", cmd);
  460. fprintf(stderr, "send/crc/length/sequence error");
  461. fprintf(stderr, "\n");
  462. }
  463. int main (int argc, char *argv[]) {
  464. int i;
  465. unsigned int res;
  466. uint32_t maxsize;
  467. int opt;
  468. int bs;
  469. int write_size = 4096;
  470. int delay_time = 1000;
  471. int repetitions = 100;
  472. int print_time = 10;
  473. int have_size = 0;
  474. int listen_only = 0;
  475. int flood = 0;
  476. int model = 1;
  477. while ( (opt = getopt(argc, argv, "qlstafMEn:d:r:p:m:w:W:D:")) != -1 ) {
  478. switch (opt) {
  479. case 'w': // Write size in K
  480. bs = atoi(optarg);
  481. if (bs > 0) {
  482. write_size = bs*1024;
  483. have_size = 1;
  484. }
  485. break;
  486. case 'W': // Write size in bytes
  487. bs = atoi(optarg);
  488. if (bs > 0) {
  489. write_size = bs;
  490. have_size = 1;
  491. }
  492. break;
  493. case 'n':
  494. if (strlen(optarg) >= CPG_MAX_NAME_LENGTH) {
  495. fprintf(stderr, "CPG name too long\n");
  496. exit(1);
  497. }
  498. strcpy(group_name.value, optarg);
  499. group_name.length = strlen(group_name.value);
  500. break;
  501. case 't':
  502. report_rtt = 1;
  503. break;
  504. case 'E':
  505. to_stderr = 1;
  506. break;
  507. case 'M':
  508. machine_readable = 1;
  509. break;
  510. case 'f':
  511. flood = 1;
  512. break;
  513. case 'a':
  514. abort_on_error = 1;
  515. break;
  516. case 'd':
  517. delay_time = atoi(optarg);
  518. break;
  519. case 'D':
  520. delimiter = optarg[0];
  521. break;
  522. case 'r':
  523. repetitions = atoi(optarg);
  524. break;
  525. case 'p':
  526. print_time = atoi(optarg);
  527. break;
  528. case 'l':
  529. listen_only = 1;
  530. break;
  531. case 's':
  532. do_syslog = 1;
  533. break;
  534. case 'q':
  535. quiet++;
  536. break;
  537. case 'm':
  538. model = atoi(optarg);
  539. if (model < 0 || model > 1) {
  540. fprintf(stderr, "%s: Model must be 0-1\n", argv[0]);
  541. exit(1);
  542. }
  543. break;
  544. case '?':
  545. usage(basename(argv[0]));
  546. exit(1);
  547. }
  548. }
  549. if (!have_size && flood) {
  550. write_size = 64;
  551. }
  552. signal (SIGALRM, sigalrm_handler);
  553. signal (SIGINT, sigint_handler);
  554. switch (model) {
  555. case 0:
  556. res = cpg_initialize (&handle, &callbacks);
  557. break;
  558. case 1:
  559. res = cpg_model_initialize (&handle, CPG_MODEL_V1, (cpg_model_data_t *)&model1_data, NULL);
  560. break;
  561. default:
  562. res=999; // can't get here but it keeps the compiler happy
  563. break;
  564. }
  565. if (res != CS_OK) {
  566. cpgh_log_printf(CPGH_LOG_ERR, "cpg_initialize failed with result %d\n", res);
  567. exit (1);
  568. }
  569. cpg_local_get(handle, &g_our_nodeid);
  570. pthread_create (&thread, NULL, dispatch_thread, NULL);
  571. res = cpg_join (handle, &group_name);
  572. if (res != CS_OK) {
  573. cpgh_log_printf(CPGH_LOG_ERR, "cpg_join failed with result %d\n", res);
  574. exit (1);
  575. }
  576. if (listen_only) {
  577. int secs = 0;
  578. while (!stopped) {
  579. sleep(1);
  580. if (++secs > print_time && !quiet) {
  581. int nodes_printed = 0;
  582. if (!machine_readable) {
  583. for (i=1; i<MAX_NODEID; i++) {
  584. if (g_recv_counter[i]) {
  585. cpgh_log_printf(CPGH_LOG_INFO, "%s: %5d message%s of %d bytes received from node %d\n",
  586. group_name.value, g_recv_counter[i] - g_recv_start[i],
  587. g_recv_counter[i]==1?"":"s",
  588. g_recv_size[i], i);
  589. nodes_printed++;
  590. }
  591. }
  592. }
  593. /* Separate list of nodes if more than one */
  594. if (nodes_printed > 1) {
  595. cpgh_log_printf(CPGH_LOG_INFO, "\n");
  596. }
  597. secs = 0;
  598. }
  599. }
  600. }
  601. else {
  602. cpg_max_atomic_msgsize_get (handle, &maxsize);
  603. if (write_size > maxsize) {
  604. fprintf(stderr, "INFO: packet size (%d) is larger than the maximum atomic size (%d), libcpg will fragment\n",
  605. write_size, maxsize);
  606. }
  607. /* The main job starts here */
  608. if (flood) {
  609. for (i = 0; i < 10; i++) { /* number of repetitions - up to 50k */
  610. cpg_flood (handle, write_size);
  611. signal (SIGALRM, sigalrm_handler);
  612. write_size *= 5;
  613. if (write_size >= (ONE_MEG - 100)) {
  614. break;
  615. }
  616. }
  617. }
  618. else {
  619. send_counter = -1; /* So we start from zero to allow listeners to sync */
  620. for (i = 0; i < repetitions && !stopped; i++) {
  621. cpg_test (handle, write_size, delay_time, print_time);
  622. signal (SIGALRM, sigalrm_handler);
  623. }
  624. }
  625. }
  626. res = cpg_finalize (handle);
  627. if (res != CS_OK) {
  628. cpgh_log_printf(CPGH_LOG_ERR, "cpg_finalize failed with result %d\n", res);
  629. exit (1);
  630. }
  631. if (quiet < 2) {
  632. /* Don't print LONG_MAX for min_rtt if we don't have a value */
  633. if (min_rtt == LONG_MAX) {
  634. min_rtt = 0L;
  635. }
  636. if (machine_readable) {
  637. cpgh_log_printf(CPGH_LOG_STATS, "%d%c%d%c%d%c%d%c%d%c%d%c%d%c%ld%c%ld%c%ld\n",
  638. packets_sent, delimiter,
  639. send_fails, delimiter,
  640. send_retries, delimiter,
  641. length_errors, delimiter,
  642. packets_recvd, delimiter,
  643. sequence_errors, delimiter,
  644. crc_errors, delimiter,
  645. min_rtt, delimiter,
  646. avg_rtt, delimiter,
  647. max_rtt);
  648. }
  649. else {
  650. cpgh_log_printf(CPGH_LOG_STATS, "\n");
  651. cpgh_log_printf(CPGH_LOG_STATS, "Stats:\n");
  652. if (!listen_only) {
  653. cpgh_log_printf(CPGH_LOG_STATS, " packets sent: %d\n", packets_sent);
  654. cpgh_log_printf(CPGH_LOG_STATS, " send failures: %d\n", send_fails);
  655. cpgh_log_printf(CPGH_LOG_STATS, " send retries: %d\n", send_retries);
  656. }
  657. cpgh_log_printf(CPGH_LOG_STATS, " length errors: %d\n", length_errors);
  658. cpgh_log_printf(CPGH_LOG_STATS, " packets recvd: %d\n", packets_recvd);
  659. cpgh_log_printf(CPGH_LOG_STATS, " sequence errors: %d\n", sequence_errors);
  660. cpgh_log_printf(CPGH_LOG_STATS, " crc errors: %d\n", crc_errors);
  661. if (!listen_only) {
  662. cpgh_log_printf(CPGH_LOG_STATS, " min RTT: %ld\n", min_rtt);
  663. cpgh_log_printf(CPGH_LOG_STATS, " max RTT: %ld\n", max_rtt);
  664. cpgh_log_printf(CPGH_LOG_STATS, " avg RTT: %ld\n", avg_rtt);
  665. }
  666. cpgh_log_printf(CPGH_LOG_STATS, "\n");
  667. }
  668. }
  669. res = 0;
  670. if (send_fails > 0 || (have_size && length_errors > 0) || sequence_errors > 0 || crc_errors > 0) {
  671. res = 2;
  672. }
  673. return (res);
  674. }