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