1 // SPDX-License-Identifier: GPL-2.0 2 3 #define _GNU_SOURCE 4 5 #include <errno.h> 6 #include <limits.h> 7 #include <fcntl.h> 8 #include <string.h> 9 #include <stdarg.h> 10 #include <stdbool.h> 11 #include <stdint.h> 12 #include <stdio.h> 13 #include <stdlib.h> 14 #include <strings.h> 15 #include <signal.h> 16 #include <unistd.h> 17 #include <time.h> 18 19 #include <sys/ioctl.h> 20 #include <sys/poll.h> 21 #include <sys/sendfile.h> 22 #include <sys/stat.h> 23 #include <sys/socket.h> 24 #include <sys/types.h> 25 #include <sys/mman.h> 26 27 #include <netdb.h> 28 #include <netinet/in.h> 29 30 #include <linux/tcp.h> 31 #include <linux/time_types.h> 32 #include <linux/sockios.h> 33 34 extern int optind; 35 36 #ifndef IPPROTO_MPTCP 37 #define IPPROTO_MPTCP 262 38 #endif 39 #ifndef TCP_ULP 40 #define TCP_ULP 31 41 #endif 42 43 static int poll_timeout = 10 * 1000; 44 static bool listen_mode; 45 static bool quit; 46 47 enum cfg_mode { 48 CFG_MODE_POLL, 49 CFG_MODE_MMAP, 50 CFG_MODE_SENDFILE, 51 }; 52 53 enum cfg_peek { 54 CFG_NONE_PEEK, 55 CFG_WITH_PEEK, 56 CFG_AFTER_PEEK, 57 }; 58 59 static enum cfg_mode cfg_mode = CFG_MODE_POLL; 60 static enum cfg_peek cfg_peek = CFG_NONE_PEEK; 61 static const char *cfg_host; 62 static const char *cfg_port = "12000"; 63 static int cfg_sock_proto = IPPROTO_MPTCP; 64 static int pf = AF_INET; 65 static int cfg_sndbuf; 66 static int cfg_rcvbuf; 67 static bool cfg_join; 68 static bool cfg_remove; 69 static unsigned int cfg_time; 70 static unsigned int cfg_do_w; 71 static int cfg_wait; 72 static uint32_t cfg_mark; 73 static char *cfg_input; 74 static int cfg_repeat = 1; 75 76 struct cfg_cmsg_types { 77 unsigned int cmsg_enabled:1; 78 unsigned int timestampns:1; 79 unsigned int tcp_inq:1; 80 }; 81 82 struct cfg_sockopt_types { 83 unsigned int transparent:1; 84 }; 85 86 struct tcp_inq_state { 87 unsigned int last; 88 bool expect_eof; 89 }; 90 91 static struct tcp_inq_state tcp_inq; 92 93 static struct cfg_cmsg_types cfg_cmsg_types; 94 static struct cfg_sockopt_types cfg_sockopt_types; 95 96 static void die_usage(void) 97 { 98 fprintf(stderr, "Usage: mptcp_connect [-6] [-c cmsg] [-i file] [-I num] [-j] [-l] " 99 "[-m mode] [-M mark] [-o option] [-p port] [-P mode] [-j] [-l] [-r num] " 100 "[-s MPTCP|TCP] [-S num] [-r num] [-t num] [-T num] [-u] [-w sec] connect_address\n"); 101 fprintf(stderr, "\t-6 use ipv6\n"); 102 fprintf(stderr, "\t-c cmsg -- test cmsg type <cmsg>\n"); 103 fprintf(stderr, "\t-i file -- read the data to send from the given file instead of stdin"); 104 fprintf(stderr, "\t-I num -- repeat the transfer 'num' times. In listen mode accepts num " 105 "incoming connections, in client mode, disconnect and reconnect to the server\n"); 106 fprintf(stderr, "\t-j -- add additional sleep at connection start and tear down " 107 "-- for MPJ tests\n"); 108 fprintf(stderr, "\t-l -- listens mode, accepts incoming connection\n"); 109 fprintf(stderr, "\t-m [poll|mmap|sendfile] -- use poll(default)/mmap+write/sendfile\n"); 110 fprintf(stderr, "\t-M mark -- set socket packet mark\n"); 111 fprintf(stderr, "\t-o option -- test sockopt <option>\n"); 112 fprintf(stderr, "\t-p num -- use port num\n"); 113 fprintf(stderr, 114 "\t-P [saveWithPeek|saveAfterPeek] -- save data with/after MSG_PEEK form tcp socket\n"); 115 fprintf(stderr, "\t-t num -- set poll timeout to num\n"); 116 fprintf(stderr, "\t-T num -- set expected runtime to num ms\n"); 117 fprintf(stderr, "\t-r num -- enable slow mode, limiting each write to num bytes " 118 "-- for remove addr tests\n"); 119 fprintf(stderr, "\t-R num -- set SO_RCVBUF to num\n"); 120 fprintf(stderr, "\t-s [MPTCP|TCP] -- use mptcp(default) or tcp sockets\n"); 121 fprintf(stderr, "\t-S num -- set SO_SNDBUF to num\n"); 122 fprintf(stderr, "\t-w num -- wait num sec before closing the socket\n"); 123 exit(1); 124 } 125 126 static void xerror(const char *fmt, ...) 127 { 128 va_list ap; 129 130 va_start(ap, fmt); 131 vfprintf(stderr, fmt, ap); 132 va_end(ap); 133 exit(1); 134 } 135 136 static void handle_signal(int nr) 137 { 138 quit = true; 139 } 140 141 static const char *getxinfo_strerr(int err) 142 { 143 if (err == EAI_SYSTEM) 144 return strerror(errno); 145 146 return gai_strerror(err); 147 } 148 149 static void xgetnameinfo(const struct sockaddr *addr, socklen_t addrlen, 150 char *host, socklen_t hostlen, 151 char *serv, socklen_t servlen) 152 { 153 int flags = NI_NUMERICHOST | NI_NUMERICSERV; 154 int err = getnameinfo(addr, addrlen, host, hostlen, serv, servlen, 155 flags); 156 157 if (err) { 158 const char *errstr = getxinfo_strerr(err); 159 160 fprintf(stderr, "Fatal: getnameinfo: %s\n", errstr); 161 exit(1); 162 } 163 } 164 165 static void xgetaddrinfo(const char *node, const char *service, 166 const struct addrinfo *hints, 167 struct addrinfo **res) 168 { 169 int err = getaddrinfo(node, service, hints, res); 170 171 if (err) { 172 const char *errstr = getxinfo_strerr(err); 173 174 fprintf(stderr, "Fatal: getaddrinfo(%s:%s): %s\n", 175 node ? node : "", service ? service : "", errstr); 176 exit(1); 177 } 178 } 179 180 static void set_rcvbuf(int fd, unsigned int size) 181 { 182 int err; 183 184 err = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &size, sizeof(size)); 185 if (err) { 186 perror("set SO_RCVBUF"); 187 exit(1); 188 } 189 } 190 191 static void set_sndbuf(int fd, unsigned int size) 192 { 193 int err; 194 195 err = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &size, sizeof(size)); 196 if (err) { 197 perror("set SO_SNDBUF"); 198 exit(1); 199 } 200 } 201 202 static void set_mark(int fd, uint32_t mark) 203 { 204 int err; 205 206 err = setsockopt(fd, SOL_SOCKET, SO_MARK, &mark, sizeof(mark)); 207 if (err) { 208 perror("set SO_MARK"); 209 exit(1); 210 } 211 } 212 213 static void set_transparent(int fd, int pf) 214 { 215 int one = 1; 216 217 switch (pf) { 218 case AF_INET: 219 if (-1 == setsockopt(fd, SOL_IP, IP_TRANSPARENT, &one, sizeof(one))) 220 perror("IP_TRANSPARENT"); 221 break; 222 case AF_INET6: 223 if (-1 == setsockopt(fd, IPPROTO_IPV6, IPV6_TRANSPARENT, &one, sizeof(one))) 224 perror("IPV6_TRANSPARENT"); 225 break; 226 } 227 } 228 229 static int do_ulp_so(int sock, const char *name) 230 { 231 return setsockopt(sock, IPPROTO_TCP, TCP_ULP, name, strlen(name)); 232 } 233 234 #define X(m) xerror("%s:%u: %s: failed for proto %d at line %u", __FILE__, __LINE__, (m), proto, line) 235 static void sock_test_tcpulp(int sock, int proto, unsigned int line) 236 { 237 socklen_t buflen = 8; 238 char buf[8] = ""; 239 int ret = getsockopt(sock, IPPROTO_TCP, TCP_ULP, buf, &buflen); 240 241 if (ret != 0) 242 X("getsockopt"); 243 244 if (buflen > 0) { 245 if (strcmp(buf, "mptcp") != 0) 246 xerror("unexpected ULP '%s' for proto %d at line %u", buf, proto, line); 247 ret = do_ulp_so(sock, "tls"); 248 if (ret == 0) 249 X("setsockopt"); 250 } else if (proto == IPPROTO_MPTCP) { 251 ret = do_ulp_so(sock, "tls"); 252 if (ret != -1) 253 X("setsockopt"); 254 } 255 256 ret = do_ulp_so(sock, "mptcp"); 257 if (ret != -1) 258 X("setsockopt"); 259 260 #undef X 261 } 262 263 #define SOCK_TEST_TCPULP(s, p) sock_test_tcpulp((s), (p), __LINE__) 264 265 static int sock_listen_mptcp(const char * const listenaddr, 266 const char * const port) 267 { 268 int sock = -1; 269 struct addrinfo hints = { 270 .ai_protocol = IPPROTO_TCP, 271 .ai_socktype = SOCK_STREAM, 272 .ai_flags = AI_PASSIVE | AI_NUMERICHOST 273 }; 274 275 hints.ai_family = pf; 276 277 struct addrinfo *a, *addr; 278 int one = 1; 279 280 xgetaddrinfo(listenaddr, port, &hints, &addr); 281 hints.ai_family = pf; 282 283 for (a = addr; a; a = a->ai_next) { 284 sock = socket(a->ai_family, a->ai_socktype, cfg_sock_proto); 285 if (sock < 0) 286 continue; 287 288 SOCK_TEST_TCPULP(sock, cfg_sock_proto); 289 290 if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one, 291 sizeof(one))) 292 perror("setsockopt"); 293 294 if (cfg_sockopt_types.transparent) 295 set_transparent(sock, pf); 296 297 if (bind(sock, a->ai_addr, a->ai_addrlen) == 0) 298 break; /* success */ 299 300 perror("bind"); 301 close(sock); 302 sock = -1; 303 } 304 305 freeaddrinfo(addr); 306 307 if (sock < 0) { 308 fprintf(stderr, "Could not create listen socket\n"); 309 return sock; 310 } 311 312 SOCK_TEST_TCPULP(sock, cfg_sock_proto); 313 314 if (listen(sock, 20)) { 315 perror("listen"); 316 close(sock); 317 return -1; 318 } 319 320 SOCK_TEST_TCPULP(sock, cfg_sock_proto); 321 322 return sock; 323 } 324 325 static int sock_connect_mptcp(const char * const remoteaddr, 326 const char * const port, int proto, 327 struct addrinfo **peer) 328 { 329 struct addrinfo hints = { 330 .ai_protocol = IPPROTO_TCP, 331 .ai_socktype = SOCK_STREAM, 332 }; 333 struct addrinfo *a, *addr; 334 int sock = -1; 335 336 hints.ai_family = pf; 337 338 xgetaddrinfo(remoteaddr, port, &hints, &addr); 339 for (a = addr; a; a = a->ai_next) { 340 sock = socket(a->ai_family, a->ai_socktype, proto); 341 if (sock < 0) { 342 perror("socket"); 343 continue; 344 } 345 346 SOCK_TEST_TCPULP(sock, proto); 347 348 if (cfg_mark) 349 set_mark(sock, cfg_mark); 350 351 if (connect(sock, a->ai_addr, a->ai_addrlen) == 0) { 352 *peer = a; 353 break; /* success */ 354 } 355 356 perror("connect()"); 357 close(sock); 358 sock = -1; 359 } 360 361 freeaddrinfo(addr); 362 if (sock != -1) 363 SOCK_TEST_TCPULP(sock, proto); 364 return sock; 365 } 366 367 static size_t do_rnd_write(const int fd, char *buf, const size_t len) 368 { 369 static bool first = true; 370 unsigned int do_w; 371 ssize_t bw; 372 373 do_w = rand() & 0xffff; 374 if (do_w == 0 || do_w > len) 375 do_w = len; 376 377 if (cfg_join && first && do_w > 100) 378 do_w = 100; 379 380 if (cfg_remove && do_w > cfg_do_w) 381 do_w = cfg_do_w; 382 383 bw = write(fd, buf, do_w); 384 if (bw < 0) 385 perror("write"); 386 387 /* let the join handshake complete, before going on */ 388 if (cfg_join && first) { 389 usleep(200000); 390 first = false; 391 } 392 393 if (cfg_remove) 394 usleep(200000); 395 396 return bw; 397 } 398 399 static size_t do_write(const int fd, char *buf, const size_t len) 400 { 401 size_t offset = 0; 402 403 while (offset < len) { 404 size_t written; 405 ssize_t bw; 406 407 bw = write(fd, buf + offset, len - offset); 408 if (bw < 0) { 409 perror("write"); 410 return 0; 411 } 412 413 written = (size_t)bw; 414 offset += written; 415 } 416 417 return offset; 418 } 419 420 static void process_cmsg(struct msghdr *msgh) 421 { 422 struct __kernel_timespec ts; 423 bool inq_found = false; 424 bool ts_found = false; 425 unsigned int inq = 0; 426 struct cmsghdr *cmsg; 427 428 for (cmsg = CMSG_FIRSTHDR(msgh); cmsg ; cmsg = CMSG_NXTHDR(msgh, cmsg)) { 429 if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SO_TIMESTAMPNS_NEW) { 430 memcpy(&ts, CMSG_DATA(cmsg), sizeof(ts)); 431 ts_found = true; 432 continue; 433 } 434 if (cmsg->cmsg_level == IPPROTO_TCP && cmsg->cmsg_type == TCP_CM_INQ) { 435 memcpy(&inq, CMSG_DATA(cmsg), sizeof(inq)); 436 inq_found = true; 437 continue; 438 } 439 440 } 441 442 if (cfg_cmsg_types.timestampns) { 443 if (!ts_found) 444 xerror("TIMESTAMPNS not present\n"); 445 } 446 447 if (cfg_cmsg_types.tcp_inq) { 448 if (!inq_found) 449 xerror("TCP_INQ not present\n"); 450 451 if (inq > 1024) 452 xerror("tcp_inq %u is larger than one kbyte\n", inq); 453 tcp_inq.last = inq; 454 } 455 } 456 457 static ssize_t do_recvmsg_cmsg(const int fd, char *buf, const size_t len) 458 { 459 char msg_buf[8192]; 460 struct iovec iov = { 461 .iov_base = buf, 462 .iov_len = len, 463 }; 464 struct msghdr msg = { 465 .msg_iov = &iov, 466 .msg_iovlen = 1, 467 .msg_control = msg_buf, 468 .msg_controllen = sizeof(msg_buf), 469 }; 470 int flags = 0; 471 unsigned int last_hint = tcp_inq.last; 472 int ret = recvmsg(fd, &msg, flags); 473 474 if (ret <= 0) { 475 if (ret == 0 && tcp_inq.expect_eof) 476 return ret; 477 478 if (ret == 0 && cfg_cmsg_types.tcp_inq) 479 if (last_hint != 1 && last_hint != 0) 480 xerror("EOF but last tcp_inq hint was %u\n", last_hint); 481 482 return ret; 483 } 484 485 if (tcp_inq.expect_eof) 486 xerror("expected EOF, last_hint %u, now %u\n", 487 last_hint, tcp_inq.last); 488 489 if (msg.msg_controllen && !cfg_cmsg_types.cmsg_enabled) 490 xerror("got %lu bytes of cmsg data, expected 0\n", 491 (unsigned long)msg.msg_controllen); 492 493 if (msg.msg_controllen == 0 && cfg_cmsg_types.cmsg_enabled) 494 xerror("%s\n", "got no cmsg data"); 495 496 if (msg.msg_controllen) 497 process_cmsg(&msg); 498 499 if (cfg_cmsg_types.tcp_inq) { 500 if ((size_t)ret < len && last_hint > (unsigned int)ret) { 501 if (ret + 1 != (int)last_hint) { 502 int next = read(fd, msg_buf, sizeof(msg_buf)); 503 504 xerror("read %u of %u, last_hint was %u tcp_inq hint now %u next_read returned %d/%m\n", 505 ret, (unsigned int)len, last_hint, tcp_inq.last, next); 506 } else { 507 tcp_inq.expect_eof = true; 508 } 509 } 510 } 511 512 return ret; 513 } 514 515 static ssize_t do_rnd_read(const int fd, char *buf, const size_t len) 516 { 517 int ret = 0; 518 char tmp[16384]; 519 size_t cap = rand(); 520 521 cap &= 0xffff; 522 523 if (cap == 0) 524 cap = 1; 525 else if (cap > len) 526 cap = len; 527 528 if (cfg_peek == CFG_WITH_PEEK) { 529 ret = recv(fd, buf, cap, MSG_PEEK); 530 ret = (ret < 0) ? ret : read(fd, tmp, ret); 531 } else if (cfg_peek == CFG_AFTER_PEEK) { 532 ret = recv(fd, buf, cap, MSG_PEEK); 533 ret = (ret < 0) ? ret : read(fd, buf, cap); 534 } else if (cfg_cmsg_types.cmsg_enabled) { 535 ret = do_recvmsg_cmsg(fd, buf, cap); 536 } else { 537 ret = read(fd, buf, cap); 538 } 539 540 return ret; 541 } 542 543 static void set_nonblock(int fd, bool nonblock) 544 { 545 int flags = fcntl(fd, F_GETFL); 546 547 if (flags == -1) 548 return; 549 550 if (nonblock) 551 fcntl(fd, F_SETFL, flags | O_NONBLOCK); 552 else 553 fcntl(fd, F_SETFL, flags & ~O_NONBLOCK); 554 } 555 556 static void shut_wr(int fd) 557 { 558 /* Close our write side, ev. give some time 559 * for address notification and/or checking 560 * the current status 561 */ 562 if (cfg_wait) 563 usleep(cfg_wait); 564 565 shutdown(fd, SHUT_WR); 566 } 567 568 static int copyfd_io_poll(int infd, int peerfd, int outfd, bool *in_closed_after_out) 569 { 570 struct pollfd fds = { 571 .fd = peerfd, 572 .events = POLLIN | POLLOUT, 573 }; 574 unsigned int woff = 0, wlen = 0; 575 char wbuf[8192]; 576 577 set_nonblock(peerfd, true); 578 579 for (;;) { 580 char rbuf[8192]; 581 ssize_t len; 582 583 if (fds.events == 0) 584 break; 585 586 switch (poll(&fds, 1, poll_timeout)) { 587 case -1: 588 if (errno == EINTR) 589 continue; 590 perror("poll"); 591 return 1; 592 case 0: 593 fprintf(stderr, "%s: poll timed out (events: " 594 "POLLIN %u, POLLOUT %u)\n", __func__, 595 fds.events & POLLIN, fds.events & POLLOUT); 596 return 2; 597 } 598 599 if (fds.revents & POLLIN) { 600 len = do_rnd_read(peerfd, rbuf, sizeof(rbuf)); 601 if (len == 0) { 602 /* no more data to receive: 603 * peer has closed its write side 604 */ 605 fds.events &= ~POLLIN; 606 607 if ((fds.events & POLLOUT) == 0) { 608 *in_closed_after_out = true; 609 /* and nothing more to send */ 610 break; 611 } 612 613 /* Else, still have data to transmit */ 614 } else if (len < 0) { 615 perror("read"); 616 return 3; 617 } 618 619 do_write(outfd, rbuf, len); 620 } 621 622 if (fds.revents & POLLOUT) { 623 if (wlen == 0) { 624 woff = 0; 625 wlen = read(infd, wbuf, sizeof(wbuf)); 626 } 627 628 if (wlen > 0) { 629 ssize_t bw; 630 631 bw = do_rnd_write(peerfd, wbuf + woff, wlen); 632 if (bw < 0) 633 return 111; 634 635 woff += bw; 636 wlen -= bw; 637 } else if (wlen == 0) { 638 /* We have no more data to send. */ 639 fds.events &= ~POLLOUT; 640 641 if ((fds.events & POLLIN) == 0) 642 /* ... and peer also closed already */ 643 break; 644 645 shut_wr(peerfd); 646 } else { 647 if (errno == EINTR) 648 continue; 649 perror("read"); 650 return 4; 651 } 652 } 653 654 if (fds.revents & (POLLERR | POLLNVAL)) { 655 fprintf(stderr, "Unexpected revents: " 656 "POLLERR/POLLNVAL(%x)\n", fds.revents); 657 return 5; 658 } 659 } 660 661 /* leave some time for late join/announce */ 662 if (cfg_remove) 663 usleep(cfg_wait); 664 665 return 0; 666 } 667 668 static int do_recvfile(int infd, int outfd) 669 { 670 ssize_t r; 671 672 do { 673 char buf[16384]; 674 675 r = do_rnd_read(infd, buf, sizeof(buf)); 676 if (r > 0) { 677 if (write(outfd, buf, r) != r) 678 break; 679 } else if (r < 0) { 680 perror("read"); 681 } 682 } while (r > 0); 683 684 return (int)r; 685 } 686 687 static int do_mmap(int infd, int outfd, unsigned int size) 688 { 689 char *inbuf = mmap(NULL, size, PROT_READ, MAP_SHARED, infd, 0); 690 ssize_t ret = 0, off = 0; 691 size_t rem; 692 693 if (inbuf == MAP_FAILED) { 694 perror("mmap"); 695 return 1; 696 } 697 698 rem = size; 699 700 while (rem > 0) { 701 ret = write(outfd, inbuf + off, rem); 702 703 if (ret < 0) { 704 perror("write"); 705 break; 706 } 707 708 off += ret; 709 rem -= ret; 710 } 711 712 munmap(inbuf, size); 713 return rem; 714 } 715 716 static int get_infd_size(int fd) 717 { 718 struct stat sb; 719 ssize_t count; 720 int err; 721 722 err = fstat(fd, &sb); 723 if (err < 0) { 724 perror("fstat"); 725 return -1; 726 } 727 728 if ((sb.st_mode & S_IFMT) != S_IFREG) { 729 fprintf(stderr, "%s: stdin is not a regular file\n", __func__); 730 return -2; 731 } 732 733 count = sb.st_size; 734 if (count > INT_MAX) { 735 fprintf(stderr, "File too large: %zu\n", count); 736 return -3; 737 } 738 739 return (int)count; 740 } 741 742 static int do_sendfile(int infd, int outfd, unsigned int count) 743 { 744 while (count > 0) { 745 ssize_t r; 746 747 r = sendfile(outfd, infd, NULL, count); 748 if (r < 0) { 749 perror("sendfile"); 750 return 3; 751 } 752 753 count -= r; 754 } 755 756 return 0; 757 } 758 759 static int copyfd_io_mmap(int infd, int peerfd, int outfd, 760 unsigned int size, bool *in_closed_after_out) 761 { 762 int err; 763 764 if (listen_mode) { 765 err = do_recvfile(peerfd, outfd); 766 if (err) 767 return err; 768 769 err = do_mmap(infd, peerfd, size); 770 } else { 771 err = do_mmap(infd, peerfd, size); 772 if (err) 773 return err; 774 775 shut_wr(peerfd); 776 777 err = do_recvfile(peerfd, outfd); 778 *in_closed_after_out = true; 779 } 780 781 return err; 782 } 783 784 static int copyfd_io_sendfile(int infd, int peerfd, int outfd, 785 unsigned int size, bool *in_closed_after_out) 786 { 787 int err; 788 789 if (listen_mode) { 790 err = do_recvfile(peerfd, outfd); 791 if (err) 792 return err; 793 794 err = do_sendfile(infd, peerfd, size); 795 } else { 796 err = do_sendfile(infd, peerfd, size); 797 if (err) 798 return err; 799 800 shut_wr(peerfd); 801 802 err = do_recvfile(peerfd, outfd); 803 *in_closed_after_out = true; 804 } 805 806 return err; 807 } 808 809 static int copyfd_io(int infd, int peerfd, int outfd, bool close_peerfd) 810 { 811 bool in_closed_after_out = false; 812 struct timespec start, end; 813 int file_size; 814 int ret; 815 816 if (cfg_time && (clock_gettime(CLOCK_MONOTONIC, &start) < 0)) 817 xerror("can not fetch start time %d", errno); 818 819 switch (cfg_mode) { 820 case CFG_MODE_POLL: 821 ret = copyfd_io_poll(infd, peerfd, outfd, &in_closed_after_out); 822 break; 823 824 case CFG_MODE_MMAP: 825 file_size = get_infd_size(infd); 826 if (file_size < 0) 827 return file_size; 828 ret = copyfd_io_mmap(infd, peerfd, outfd, file_size, &in_closed_after_out); 829 break; 830 831 case CFG_MODE_SENDFILE: 832 file_size = get_infd_size(infd); 833 if (file_size < 0) 834 return file_size; 835 ret = copyfd_io_sendfile(infd, peerfd, outfd, file_size, &in_closed_after_out); 836 break; 837 838 default: 839 fprintf(stderr, "Invalid mode %d\n", cfg_mode); 840 841 die_usage(); 842 return 1; 843 } 844 845 if (ret) 846 return ret; 847 848 if (close_peerfd) 849 close(peerfd); 850 851 if (cfg_time) { 852 unsigned int delta_ms; 853 854 if (clock_gettime(CLOCK_MONOTONIC, &end) < 0) 855 xerror("can not fetch end time %d", errno); 856 delta_ms = (end.tv_sec - start.tv_sec) * 1000 + (end.tv_nsec - start.tv_nsec) / 1000000; 857 if (delta_ms > cfg_time) { 858 xerror("transfer slower than expected! runtime %d ms, expected %d ms", 859 delta_ms, cfg_time); 860 } 861 862 /* show the runtime only if this end shutdown(wr) before receiving the EOF, 863 * (that is, if this end got the longer runtime) 864 */ 865 if (in_closed_after_out) 866 fprintf(stderr, "%d", delta_ms); 867 } 868 869 return 0; 870 } 871 872 static void check_sockaddr(int pf, struct sockaddr_storage *ss, 873 socklen_t salen) 874 { 875 struct sockaddr_in6 *sin6; 876 struct sockaddr_in *sin; 877 socklen_t wanted_size = 0; 878 879 switch (pf) { 880 case AF_INET: 881 wanted_size = sizeof(*sin); 882 sin = (void *)ss; 883 if (!sin->sin_port) 884 fprintf(stderr, "accept: something wrong: ip connection from port 0"); 885 break; 886 case AF_INET6: 887 wanted_size = sizeof(*sin6); 888 sin6 = (void *)ss; 889 if (!sin6->sin6_port) 890 fprintf(stderr, "accept: something wrong: ipv6 connection from port 0"); 891 break; 892 default: 893 fprintf(stderr, "accept: Unknown pf %d, salen %u\n", pf, salen); 894 return; 895 } 896 897 if (salen != wanted_size) 898 fprintf(stderr, "accept: size mismatch, got %d expected %d\n", 899 (int)salen, wanted_size); 900 901 if (ss->ss_family != pf) 902 fprintf(stderr, "accept: pf mismatch, expect %d, ss_family is %d\n", 903 (int)ss->ss_family, pf); 904 } 905 906 static void check_getpeername(int fd, struct sockaddr_storage *ss, socklen_t salen) 907 { 908 struct sockaddr_storage peerss; 909 socklen_t peersalen = sizeof(peerss); 910 911 if (getpeername(fd, (struct sockaddr *)&peerss, &peersalen) < 0) { 912 perror("getpeername"); 913 return; 914 } 915 916 if (peersalen != salen) { 917 fprintf(stderr, "%s: %d vs %d\n", __func__, peersalen, salen); 918 return; 919 } 920 921 if (memcmp(ss, &peerss, peersalen)) { 922 char a[INET6_ADDRSTRLEN]; 923 char b[INET6_ADDRSTRLEN]; 924 char c[INET6_ADDRSTRLEN]; 925 char d[INET6_ADDRSTRLEN]; 926 927 xgetnameinfo((struct sockaddr *)ss, salen, 928 a, sizeof(a), b, sizeof(b)); 929 930 xgetnameinfo((struct sockaddr *)&peerss, peersalen, 931 c, sizeof(c), d, sizeof(d)); 932 933 fprintf(stderr, "%s: memcmp failure: accept %s vs peername %s, %s vs %s salen %d vs %d\n", 934 __func__, a, c, b, d, peersalen, salen); 935 } 936 } 937 938 static void check_getpeername_connect(int fd) 939 { 940 struct sockaddr_storage ss; 941 socklen_t salen = sizeof(ss); 942 char a[INET6_ADDRSTRLEN]; 943 char b[INET6_ADDRSTRLEN]; 944 945 if (getpeername(fd, (struct sockaddr *)&ss, &salen) < 0) { 946 perror("getpeername"); 947 return; 948 } 949 950 xgetnameinfo((struct sockaddr *)&ss, salen, 951 a, sizeof(a), b, sizeof(b)); 952 953 if (strcmp(cfg_host, a) || strcmp(cfg_port, b)) 954 fprintf(stderr, "%s: %s vs %s, %s vs %s\n", __func__, 955 cfg_host, a, cfg_port, b); 956 } 957 958 static void maybe_close(int fd) 959 { 960 unsigned int r = rand(); 961 962 if (!(cfg_join || cfg_remove || cfg_repeat > 1) && (r & 1)) 963 close(fd); 964 } 965 966 int main_loop_s(int listensock) 967 { 968 struct sockaddr_storage ss; 969 struct pollfd polls; 970 socklen_t salen; 971 int remotesock; 972 int fd = 0; 973 974 again: 975 polls.fd = listensock; 976 polls.events = POLLIN; 977 978 switch (poll(&polls, 1, poll_timeout)) { 979 case -1: 980 perror("poll"); 981 return 1; 982 case 0: 983 fprintf(stderr, "%s: timed out\n", __func__); 984 close(listensock); 985 return 2; 986 } 987 988 salen = sizeof(ss); 989 remotesock = accept(listensock, (struct sockaddr *)&ss, &salen); 990 if (remotesock >= 0) { 991 maybe_close(listensock); 992 check_sockaddr(pf, &ss, salen); 993 check_getpeername(remotesock, &ss, salen); 994 995 if (cfg_input) { 996 fd = open(cfg_input, O_RDONLY); 997 if (fd < 0) 998 xerror("can't open %s: %d", cfg_input, errno); 999 } 1000 1001 SOCK_TEST_TCPULP(remotesock, 0); 1002 1003 copyfd_io(fd, remotesock, 1, true); 1004 } else { 1005 perror("accept"); 1006 return 1; 1007 } 1008 1009 if (--cfg_repeat > 0) { 1010 if (cfg_input) 1011 close(fd); 1012 goto again; 1013 } 1014 1015 return 0; 1016 } 1017 1018 static void init_rng(void) 1019 { 1020 int fd = open("/dev/urandom", O_RDONLY); 1021 unsigned int foo; 1022 1023 if (fd > 0) { 1024 int ret = read(fd, &foo, sizeof(foo)); 1025 1026 if (ret < 0) 1027 srand(fd + foo); 1028 close(fd); 1029 } 1030 1031 srand(foo); 1032 } 1033 1034 static void xsetsockopt(int fd, int level, int optname, const void *optval, socklen_t optlen) 1035 { 1036 int err; 1037 1038 err = setsockopt(fd, level, optname, optval, optlen); 1039 if (err) { 1040 perror("setsockopt"); 1041 exit(1); 1042 } 1043 } 1044 1045 static void apply_cmsg_types(int fd, const struct cfg_cmsg_types *cmsg) 1046 { 1047 static const unsigned int on = 1; 1048 1049 if (cmsg->timestampns) 1050 xsetsockopt(fd, SOL_SOCKET, SO_TIMESTAMPNS_NEW, &on, sizeof(on)); 1051 if (cmsg->tcp_inq) 1052 xsetsockopt(fd, IPPROTO_TCP, TCP_INQ, &on, sizeof(on)); 1053 } 1054 1055 static void parse_cmsg_types(const char *type) 1056 { 1057 char *next = strchr(type, ','); 1058 unsigned int len = 0; 1059 1060 cfg_cmsg_types.cmsg_enabled = 1; 1061 1062 if (next) { 1063 parse_cmsg_types(next + 1); 1064 len = next - type; 1065 } else { 1066 len = strlen(type); 1067 } 1068 1069 if (strncmp(type, "TIMESTAMPNS", len) == 0) { 1070 cfg_cmsg_types.timestampns = 1; 1071 return; 1072 } 1073 1074 if (strncmp(type, "TCPINQ", len) == 0) { 1075 cfg_cmsg_types.tcp_inq = 1; 1076 return; 1077 } 1078 1079 fprintf(stderr, "Unrecognized cmsg option %s\n", type); 1080 exit(1); 1081 } 1082 1083 static void parse_setsock_options(const char *name) 1084 { 1085 char *next = strchr(name, ','); 1086 unsigned int len = 0; 1087 1088 if (next) { 1089 parse_setsock_options(next + 1); 1090 len = next - name; 1091 } else { 1092 len = strlen(name); 1093 } 1094 1095 if (strncmp(name, "TRANSPARENT", len) == 0) { 1096 cfg_sockopt_types.transparent = 1; 1097 return; 1098 } 1099 1100 fprintf(stderr, "Unrecognized setsockopt option %s\n", name); 1101 exit(1); 1102 } 1103 1104 void xdisconnect(int fd, int addrlen) 1105 { 1106 struct sockaddr_storage empty; 1107 int msec_sleep = 10; 1108 int queued = 1; 1109 int i; 1110 1111 shutdown(fd, SHUT_WR); 1112 1113 /* while until the pending data is completely flushed, the later 1114 * disconnect will bypass/ignore/drop any pending data. 1115 */ 1116 for (i = 0; ; i += msec_sleep) { 1117 if (ioctl(fd, SIOCOUTQ, &queued) < 0) 1118 xerror("can't query out socket queue: %d", errno); 1119 1120 if (!queued) 1121 break; 1122 1123 if (i > poll_timeout) 1124 xerror("timeout while waiting for spool to complete"); 1125 usleep(msec_sleep * 1000); 1126 } 1127 1128 memset(&empty, 0, sizeof(empty)); 1129 empty.ss_family = AF_UNSPEC; 1130 if (connect(fd, (struct sockaddr *)&empty, addrlen) < 0) 1131 xerror("can't disconnect: %d", errno); 1132 } 1133 1134 int main_loop(void) 1135 { 1136 int fd, ret, fd_in = 0; 1137 struct addrinfo *peer; 1138 1139 /* listener is ready. */ 1140 fd = sock_connect_mptcp(cfg_host, cfg_port, cfg_sock_proto, &peer); 1141 if (fd < 0) 1142 return 2; 1143 1144 again: 1145 check_getpeername_connect(fd); 1146 1147 SOCK_TEST_TCPULP(fd, cfg_sock_proto); 1148 1149 if (cfg_rcvbuf) 1150 set_rcvbuf(fd, cfg_rcvbuf); 1151 if (cfg_sndbuf) 1152 set_sndbuf(fd, cfg_sndbuf); 1153 if (cfg_cmsg_types.cmsg_enabled) 1154 apply_cmsg_types(fd, &cfg_cmsg_types); 1155 1156 if (cfg_input) { 1157 fd_in = open(cfg_input, O_RDONLY); 1158 if (fd < 0) 1159 xerror("can't open %s:%d", cfg_input, errno); 1160 } 1161 1162 /* close the client socket open only if we are not going to reconnect */ 1163 ret = copyfd_io(fd_in, fd, 1, cfg_repeat == 1); 1164 if (ret) 1165 return ret; 1166 1167 if (--cfg_repeat > 0) { 1168 xdisconnect(fd, peer->ai_addrlen); 1169 1170 /* the socket could be unblocking at this point, we need the 1171 * connect to be blocking 1172 */ 1173 set_nonblock(fd, false); 1174 if (connect(fd, peer->ai_addr, peer->ai_addrlen)) 1175 xerror("can't reconnect: %d", errno); 1176 if (cfg_input) 1177 close(fd_in); 1178 goto again; 1179 } 1180 return 0; 1181 } 1182 1183 int parse_proto(const char *proto) 1184 { 1185 if (!strcasecmp(proto, "MPTCP")) 1186 return IPPROTO_MPTCP; 1187 if (!strcasecmp(proto, "TCP")) 1188 return IPPROTO_TCP; 1189 1190 fprintf(stderr, "Unknown protocol: %s\n.", proto); 1191 die_usage(); 1192 1193 /* silence compiler warning */ 1194 return 0; 1195 } 1196 1197 int parse_mode(const char *mode) 1198 { 1199 if (!strcasecmp(mode, "poll")) 1200 return CFG_MODE_POLL; 1201 if (!strcasecmp(mode, "mmap")) 1202 return CFG_MODE_MMAP; 1203 if (!strcasecmp(mode, "sendfile")) 1204 return CFG_MODE_SENDFILE; 1205 1206 fprintf(stderr, "Unknown test mode: %s\n", mode); 1207 fprintf(stderr, "Supported modes are:\n"); 1208 fprintf(stderr, "\t\t\"poll\" - interleaved read/write using poll()\n"); 1209 fprintf(stderr, "\t\t\"mmap\" - send entire input file (mmap+write), then read response (-l will read input first)\n"); 1210 fprintf(stderr, "\t\t\"sendfile\" - send entire input file (sendfile), then read response (-l will read input first)\n"); 1211 1212 die_usage(); 1213 1214 /* silence compiler warning */ 1215 return 0; 1216 } 1217 1218 int parse_peek(const char *mode) 1219 { 1220 if (!strcasecmp(mode, "saveWithPeek")) 1221 return CFG_WITH_PEEK; 1222 if (!strcasecmp(mode, "saveAfterPeek")) 1223 return CFG_AFTER_PEEK; 1224 1225 fprintf(stderr, "Unknown: %s\n", mode); 1226 fprintf(stderr, "Supported MSG_PEEK mode are:\n"); 1227 fprintf(stderr, 1228 "\t\t\"saveWithPeek\" - recv data with flags 'MSG_PEEK' and save the peek data into file\n"); 1229 fprintf(stderr, 1230 "\t\t\"saveAfterPeek\" - read and save data into file after recv with flags 'MSG_PEEK'\n"); 1231 1232 die_usage(); 1233 1234 /* silence compiler warning */ 1235 return 0; 1236 } 1237 1238 static int parse_int(const char *size) 1239 { 1240 unsigned long s; 1241 1242 errno = 0; 1243 1244 s = strtoul(size, NULL, 0); 1245 1246 if (errno) { 1247 fprintf(stderr, "Invalid sndbuf size %s (%s)\n", 1248 size, strerror(errno)); 1249 die_usage(); 1250 } 1251 1252 if (s > INT_MAX) { 1253 fprintf(stderr, "Invalid sndbuf size %s (%s)\n", 1254 size, strerror(ERANGE)); 1255 die_usage(); 1256 } 1257 1258 return (int)s; 1259 } 1260 1261 static void parse_opts(int argc, char **argv) 1262 { 1263 int c; 1264 1265 while ((c = getopt(argc, argv, "6c:hi:I:jlm:M:o:p:P:r:R:s:S:t:T:w:")) != -1) { 1266 switch (c) { 1267 case 'j': 1268 cfg_join = true; 1269 cfg_mode = CFG_MODE_POLL; 1270 break; 1271 case 'r': 1272 cfg_remove = true; 1273 cfg_mode = CFG_MODE_POLL; 1274 cfg_wait = 400000; 1275 cfg_do_w = atoi(optarg); 1276 if (cfg_do_w <= 0) 1277 cfg_do_w = 50; 1278 break; 1279 case 'i': 1280 cfg_input = optarg; 1281 break; 1282 case 'I': 1283 cfg_repeat = atoi(optarg); 1284 break; 1285 case 'l': 1286 listen_mode = true; 1287 break; 1288 case 'p': 1289 cfg_port = optarg; 1290 break; 1291 case 's': 1292 cfg_sock_proto = parse_proto(optarg); 1293 break; 1294 case 'h': 1295 die_usage(); 1296 break; 1297 case '6': 1298 pf = AF_INET6; 1299 break; 1300 case 't': 1301 poll_timeout = atoi(optarg) * 1000; 1302 if (poll_timeout <= 0) 1303 poll_timeout = -1; 1304 break; 1305 case 'T': 1306 cfg_time = atoi(optarg); 1307 break; 1308 case 'm': 1309 cfg_mode = parse_mode(optarg); 1310 break; 1311 case 'S': 1312 cfg_sndbuf = parse_int(optarg); 1313 break; 1314 case 'R': 1315 cfg_rcvbuf = parse_int(optarg); 1316 break; 1317 case 'w': 1318 cfg_wait = atoi(optarg)*1000000; 1319 break; 1320 case 'M': 1321 cfg_mark = strtol(optarg, NULL, 0); 1322 break; 1323 case 'P': 1324 cfg_peek = parse_peek(optarg); 1325 break; 1326 case 'c': 1327 parse_cmsg_types(optarg); 1328 break; 1329 case 'o': 1330 parse_setsock_options(optarg); 1331 break; 1332 } 1333 } 1334 1335 if (optind + 1 != argc) 1336 die_usage(); 1337 cfg_host = argv[optind]; 1338 1339 if (strchr(cfg_host, ':')) 1340 pf = AF_INET6; 1341 } 1342 1343 int main(int argc, char *argv[]) 1344 { 1345 init_rng(); 1346 1347 signal(SIGUSR1, handle_signal); 1348 parse_opts(argc, argv); 1349 1350 if (listen_mode) { 1351 int fd = sock_listen_mptcp(cfg_host, cfg_port); 1352 1353 if (fd < 0) 1354 return 1; 1355 1356 if (cfg_rcvbuf) 1357 set_rcvbuf(fd, cfg_rcvbuf); 1358 if (cfg_sndbuf) 1359 set_sndbuf(fd, cfg_sndbuf); 1360 if (cfg_mark) 1361 set_mark(fd, cfg_mark); 1362 if (cfg_cmsg_types.cmsg_enabled) 1363 apply_cmsg_types(fd, &cfg_cmsg_types); 1364 1365 return main_loop_s(fd); 1366 } 1367 1368 return main_loop(); 1369 } 1370