[PATCH v3 mptcp-next 0/4] TCP_INQ support

[PATCH v3 mptcp-next 0/4] TCP_INQ support

[Florian Westphal]

No major changes since v2, see individual patches for details.

This adds TCP_INQ for mptcp and extends the selftest infra.
Patch 3 adds ioctls to retrieve amount of in-sequence bytes
ready to for read() (older version of TCP_INQ) and counters for
unsent and unacked byte count.

Last patch is new in v2, its a more specific TCP_INQ test case.
See individual patches for changes since v1.

Florian Westphal (4):
  mptcp: add TCP_INQ cmsg support
  selftests: mptcp: add TCP_INQ support
  mptcp: add SIOCINQ, OUTQ and OUTQNSD ioctls
  selftests: mptcp: add inq test case

 net/mptcp/protocol.c                          |  89 ++-
 net/mptcp/protocol.h                          |   1 +
 net/mptcp/sockopt.c                           |  37 ++
 tools/testing/selftests/net/mptcp/.gitignore  |   1 +
 tools/testing/selftests/net/mptcp/Makefile    |   2 +-
 .../selftests/net/mptcp/mptcp_connect.c       |  58 +-
 tools/testing/selftests/net/mptcp/mptcp_inq.c | 603 ++++++++++++++++++
 .../selftests/net/mptcp/mptcp_sockopt.sh      |  44 +-
 8 files changed, 830 insertions(+), 5 deletions(-)
 create mode 100644 tools/testing/selftests/net/mptcp/mptcp_inq.c

-- 
2.32.0

[PATCH mptcp-next v3 1/4] mptcp: add TCP_INQ cmsg support

[Florian Westphal]

Support the TCP_INQ setsockopt.

This is a boolean that tells recvmsg path to include the remaining
in-sequence bytes in the cmsg data.

v2: do not use CB(skb)->offset, increment map_seq instead (Paolo Abeni)
v3: adjust CB(skb)->map_seq when taking skb from ofo queue (Paolo Abeni)

Closes: https://github.com/multipath-tcp/mptcp_net-next/issues/224
Signed-off-by: Florian Westphal <fw@strlen.de>
---
 net/mptcp/protocol.c | 36 +++++++++++++++++++++++++++++++++++-
 net/mptcp/protocol.h |  1 +
 net/mptcp/sockopt.c  | 37 +++++++++++++++++++++++++++++++++++++
 3 files changed, 73 insertions(+), 1 deletion(-)

diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c
index b0bfe20d6bb0..4aaa458d846c 100644
--- a/net/mptcp/protocol.c
+++ b/net/mptcp/protocol.c
@@ -46,6 +46,7 @@ struct mptcp_skb_cb {
 
 enum {
 	MPTCP_CMSG_TS = BIT(0),
+	MPTCP_CMSG_INQ = BIT(1),
 };
 
 static struct percpu_counter mptcp_sockets_allocated;
@@ -738,6 +739,7 @@ static bool __mptcp_ofo_queue(struct mptcp_sock *msk)
 				 MPTCP_SKB_CB(skb)->map_seq, msk->ack_seq,
 				 delta);
 			MPTCP_SKB_CB(skb)->offset += delta;
+			MPTCP_SKB_CB(skb)->map_seq += delta;
 			__skb_queue_tail(&sk->sk_receive_queue, skb);
 		}
 		msk->ack_seq = end_seq;
@@ -1825,8 +1827,10 @@ static int __mptcp_recvmsg_mskq(struct mptcp_sock *msk,
 		copied += count;
 
 		if (count < data_len) {
-			if (!(flags & MSG_PEEK))
+			if (!(flags & MSG_PEEK)) {
 				MPTCP_SKB_CB(skb)->offset += count;
+				MPTCP_SKB_CB(skb)->map_seq += count;
+			}
 			break;
 		}
 
@@ -2006,6 +2010,27 @@ static bool __mptcp_move_skbs(struct mptcp_sock *msk)
 	return !skb_queue_empty(&msk->receive_queue);
 }
 
+static unsigned int mptcp_inq_hint(const struct sock *sk)
+{
+	const struct mptcp_sock *msk = mptcp_sk(sk);
+	const struct sk_buff *skb;
+
+	skb = skb_peek(&msk->receive_queue);
+	if (skb) {
+		u64 hint_val = msk->ack_seq - MPTCP_SKB_CB(skb)->map_seq;
+
+		if (hint_val >= INT_MAX)
+			return INT_MAX;
+
+		return (unsigned int)hint_val;
+	}
+
+	if (sk->sk_state == TCP_CLOSE || (sk->sk_shutdown & RCV_SHUTDOWN))
+		return 1;
+
+	return 0;
+}
+
 static int mptcp_recvmsg(struct sock *sk, struct msghdr *msg, size_t len,
 			 int nonblock, int flags, int *addr_len)
 {
@@ -2030,6 +2055,9 @@ static int mptcp_recvmsg(struct sock *sk, struct msghdr *msg, size_t len,
 	len = min_t(size_t, len, INT_MAX);
 	target = sock_rcvlowat(sk, flags & MSG_WAITALL, len);
 
+	if (unlikely(msk->recvmsg_inq))
+		cmsg_flags = MPTCP_CMSG_INQ;
+
 	while (copied < len) {
 		int bytes_read;
 
@@ -2103,6 +2131,12 @@ static int mptcp_recvmsg(struct sock *sk, struct msghdr *msg, size_t len,
 	if (cmsg_flags && copied >= 0) {
 		if (cmsg_flags & MPTCP_CMSG_TS)
 			tcp_recv_timestamp(msg, sk, &tss);
+
+		if (cmsg_flags & MPTCP_CMSG_INQ) {
+			unsigned int inq = mptcp_inq_hint(sk);
+
+			put_cmsg(msg, SOL_TCP, TCP_CM_INQ, sizeof(inq), &inq);
+		}
 	}
 
 	pr_debug("msk=%p rx queue empty=%d:%d copied=%d",
diff --git a/net/mptcp/protocol.h b/net/mptcp/protocol.h
index 906509c6cde5..e77de7662df0 100644
--- a/net/mptcp/protocol.h
+++ b/net/mptcp/protocol.h
@@ -250,6 +250,7 @@ struct mptcp_sock {
 	bool		use_64bit_ack; /* Set when we received a 64-bit DSN */
 	bool		csum_enabled;
 	bool		allow_infinite_fallback;
+	u8		recvmsg_inq:1;
 	spinlock_t	join_list_lock;
 	struct work_struct work;
 	struct sk_buff  *ooo_last_skb;
diff --git a/net/mptcp/sockopt.c b/net/mptcp/sockopt.c
index fb43e145cb57..0295d6d9207e 100644
--- a/net/mptcp/sockopt.c
+++ b/net/mptcp/sockopt.c
@@ -557,6 +557,7 @@ static bool mptcp_supported_sockopt(int level, int optname)
 		case TCP_TIMESTAMP:
 		case TCP_NOTSENT_LOWAT:
 		case TCP_TX_DELAY:
+		case TCP_INQ:
 			return true;
 		}
 
@@ -698,7 +699,21 @@ static int mptcp_setsockopt_v4(struct mptcp_sock *msk, int optname,
 static int mptcp_setsockopt_sol_tcp(struct mptcp_sock *msk, int optname,
 				    sockptr_t optval, unsigned int optlen)
 {
+	struct sock *sk = (void *)msk;
+	int ret, val;
+
 	switch (optname) {
+	case TCP_INQ:
+		ret = mptcp_get_int_option(msk, optval, optlen, &val);
+		if (ret)
+			return ret;
+		if (val < 0 || val > 1)
+			return -EINVAL;
+
+		lock_sock(sk);
+		msk->recvmsg_inq = !!val;
+		release_sock(sk);
+		return 0;
 	case TCP_ULP:
 		return -EOPNOTSUPP;
 	case TCP_CONGESTION:
@@ -1032,6 +1047,26 @@ static int mptcp_getsockopt_subflow_addrs(struct mptcp_sock *msk, char __user *o
 	return 0;
 }
 
+static int mptcp_put_int_option(struct mptcp_sock *msk, char __user *optval,
+				int __user *optlen, int val)
+{
+	int len;
+
+	if (get_user(len, optlen))
+		return -EFAULT;
+
+	len = min_t(unsigned int, len, sizeof(int));
+	if (len < 0)
+		return -EINVAL;
+
+	if (put_user(len, optlen))
+		return -EFAULT;
+	if (copy_to_user(optval, &val, len))
+		return -EFAULT;
+
+	return 0;
+}
+
 static int mptcp_getsockopt_sol_tcp(struct mptcp_sock *msk, int optname,
 				    char __user *optval, int __user *optlen)
 {
@@ -1042,6 +1077,8 @@ static int mptcp_getsockopt_sol_tcp(struct mptcp_sock *msk, int optname,
 	case TCP_CC_INFO:
 		return mptcp_getsockopt_first_sf_only(msk, SOL_TCP, optname,
 						      optval, optlen);
+	case TCP_INQ:
+		return mptcp_put_int_option(msk, optval, optlen, msk->recvmsg_inq);
 	}
 	return -EOPNOTSUPP;
 }
-- 
2.32.0

[PATCH mptcp-next v3 2/4] selftests: mptcp: add TCP_INQ support

[Florian Westphal]

Do checks on the returned inq counter.

Fail on:
1. Huge value (> 1 kbyte, test case files are 1 kb)
2. last hint larger than returned bytes when read was short
3. erronenous indication of EOF.

3) happens when a hint of X bytes reads X-1 on next call
   but next recvmsg returns more data (instead of EOF).

Signed-off-by: Florian Westphal <fw@strlen.de>
---
 .../selftests/net/mptcp/mptcp_connect.c       | 58 ++++++++++++++++++-
 .../selftests/net/mptcp/mptcp_sockopt.sh      |  4 +-
 2 files changed, 59 insertions(+), 3 deletions(-)

diff --git a/tools/testing/selftests/net/mptcp/mptcp_connect.c b/tools/testing/selftests/net/mptcp/mptcp_connect.c
index ada9b80774d4..0e396af4545a 100644
--- a/tools/testing/selftests/net/mptcp/mptcp_connect.c
+++ b/tools/testing/selftests/net/mptcp/mptcp_connect.c
@@ -73,12 +73,20 @@ static uint32_t cfg_mark;
 struct cfg_cmsg_types {
 	unsigned int cmsg_enabled:1;
 	unsigned int timestampns:1;
+	unsigned int tcp_inq:1;
 };
 
 struct cfg_sockopt_types {
 	unsigned int transparent:1;
 };
 
+struct tcp_inq_state {
+	unsigned int last;
+	bool expect_eof;
+};
+
+static struct tcp_inq_state tcp_inq;
+
 static struct cfg_cmsg_types cfg_cmsg_types;
 static struct cfg_sockopt_types cfg_sockopt_types;
 
@@ -389,7 +397,9 @@ static size_t do_write(const int fd, char *buf, const size_t len)
 static void process_cmsg(struct msghdr *msgh)
 {
 	struct __kernel_timespec ts;
+	bool inq_found = false;
 	bool ts_found = false;
+	unsigned int inq = 0;
 	struct cmsghdr *cmsg;
 
 	for (cmsg = CMSG_FIRSTHDR(msgh); cmsg ; cmsg = CMSG_NXTHDR(msgh, cmsg)) {
@@ -398,12 +408,27 @@ static void process_cmsg(struct msghdr *msgh)
 			ts_found = true;
 			continue;
 		}
+		if (cmsg->cmsg_level == IPPROTO_TCP && cmsg->cmsg_type == TCP_CM_INQ) {
+			memcpy(&inq, CMSG_DATA(cmsg), sizeof(inq));
+			inq_found = true;
+			continue;
+		}
+
 	}
 
 	if (cfg_cmsg_types.timestampns) {
 		if (!ts_found)
 			xerror("TIMESTAMPNS not present\n");
 	}
+
+	if (cfg_cmsg_types.tcp_inq) {
+		if (!inq_found)
+			xerror("TCP_INQ not present\n");
+
+		if (inq > 1024)
+			xerror("tcp_inq %u is larger than one kbyte\n", inq);
+		tcp_inq.last = inq;
+	}
 }
 
 static ssize_t do_recvmsg_cmsg(const int fd, char *buf, const size_t len)
@@ -420,10 +445,23 @@ static ssize_t do_recvmsg_cmsg(const int fd, char *buf, const size_t len)
 		.msg_controllen = sizeof(msg_buf),
 	};
 	int flags = 0;
+	unsigned int last_hint = tcp_inq.last;
 	int ret = recvmsg(fd, &msg, flags);
 
-	if (ret <= 0)
+	if (ret <= 0) {
+		if (ret == 0 && tcp_inq.expect_eof)
+			return ret;
+
+		if (ret == 0 && cfg_cmsg_types.tcp_inq)
+			if (last_hint != 1 && last_hint != 0)
+				xerror("EOF but last tcp_inq hint was %u\n", last_hint);
+
 		return ret;
+	}
+
+	if (tcp_inq.expect_eof)
+		xerror("expected EOF, last_hint %u, now %u\n",
+			last_hint, tcp_inq.last);
 
 	if (msg.msg_controllen && !cfg_cmsg_types.cmsg_enabled)
 		xerror("got %lu bytes of cmsg data, expected 0\n",
@@ -435,6 +473,17 @@ static ssize_t do_recvmsg_cmsg(const int fd, char *buf, const size_t len)
 	if (msg.msg_controllen)
 		process_cmsg(&msg);
 
+	if (cfg_cmsg_types.tcp_inq) {
+		if ((size_t)ret < len && last_hint > (unsigned int)ret) {
+			if (ret + 1 != (int)last_hint) {
+				int next = read(fd, msg_buf, sizeof(msg_buf));
+
+				xerror("read %u of %u, last_hint was %u tcp_inq hint now %u next_read returned %d/%m\n", ret, (unsigned int)len, last_hint, tcp_inq.last, next);
+			} else
+				tcp_inq.expect_eof = true;
+		}
+	}
+
 	return ret;
 }
 
@@ -944,6 +993,8 @@ static void apply_cmsg_types(int fd, const struct cfg_cmsg_types *cmsg)
 
 	if (cmsg->timestampns)
 		xsetsockopt(fd, SOL_SOCKET, SO_TIMESTAMPNS_NEW, &on, sizeof(on));
+	if (cmsg->tcp_inq)
+		xsetsockopt(fd, IPPROTO_TCP, TCP_INQ, &on, sizeof(on));
 }
 
 static void parse_cmsg_types(const char *type)
@@ -965,6 +1016,11 @@ static void parse_cmsg_types(const char *type)
 		return;
 	}
 
+	if (strncmp(type, "TCPINQ", len) == 0) {
+		cfg_cmsg_types.tcp_inq = 1;
+		return;
+	}
+
 	fprintf(stderr, "Unrecognized cmsg option %s\n", type);
 	exit(1);
 }
diff --git a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
index 41de643788b8..c8c364369599 100755
--- a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
+++ b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
@@ -178,7 +178,7 @@ do_transfer()
 
 	timeout ${timeout_test} \
 		ip netns exec ${listener_ns} \
-			$mptcp_connect -t ${timeout_poll} -l -M 1 -p $port -s ${srv_proto} -c TIMESTAMPNS \
+			$mptcp_connect -t ${timeout_poll} -l -M 1 -p $port -s ${srv_proto} -c TIMESTAMPNS,TCPINQ \
 				${local_addr} < "$sin" > "$sout" &
 	spid=$!
 
@@ -186,7 +186,7 @@ do_transfer()
 
 	timeout ${timeout_test} \
 		ip netns exec ${connector_ns} \
-			$mptcp_connect -t ${timeout_poll} -M 2 -p $port -s ${cl_proto} -c TIMESTAMPNS \
+			$mptcp_connect -t ${timeout_poll} -M 2 -p $port -s ${cl_proto} -c TIMESTAMPNS,TCPINQ \
 				$connect_addr < "$cin" > "$cout" &
 
 	cpid=$!
-- 
2.32.0

[PATCH mptcp-next v3 3/4] mptcp: add SIOCINQ, OUTQ and OUTQNSD ioctls

[Florian Westphal]

Allows to query in-sequence data ready for read(), total bytes in
write queue and total bytes in write queue that have not yet been sent.

v2: remove unneeded READ_ONCE() (Paolo Abeni)
v3: check for new data unconditionally in SIOCINQ ioctl (Mat Martineau)

Signed-off-by: Florian Westphal <fw@strlen.de>
---
 net/mptcp/protocol.c | 53 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 53 insertions(+)

diff --git a/net/mptcp/protocol.c b/net/mptcp/protocol.c
index 4aaa458d846c..3ab6628ea2b2 100644
--- a/net/mptcp/protocol.c
+++ b/net/mptcp/protocol.c
@@ -22,6 +22,7 @@
 #endif
 #include <net/mptcp.h>
 #include <net/xfrm.h>
+#include <asm/ioctls.h>
 #include "protocol.h"
 #include "mib.h"
 
@@ -3261,6 +3262,57 @@ static int mptcp_forward_alloc_get(const struct sock *sk)
 	return sk->sk_forward_alloc + mptcp_sk(sk)->rmem_fwd_alloc;
 }
 
+static int mptcp_ioctl_outq(const struct mptcp_sock *msk, u64 v)
+{
+	const struct sock *sk = (void *)msk;
+	u64 delta;
+
+	if (sk->sk_state == TCP_LISTEN)
+		return -EINVAL;
+
+	if ((1 << sk->sk_state) & (TCPF_SYN_SENT | TCPF_SYN_RECV))
+		return 0;
+
+	delta = msk->write_seq - v;
+	if (delta > INT_MAX)
+		delta = INT_MAX;
+
+	return (int)delta;
+}
+
+static int mptcp_ioctl(struct sock *sk, int cmd, unsigned long arg)
+{
+	struct mptcp_sock *msk = mptcp_sk(sk);
+	bool slow;
+	int answ;
+
+	switch (cmd) {
+	case SIOCINQ:
+		if (sk->sk_state == TCP_LISTEN)
+			return -EINVAL;
+
+		lock_sock(sk);
+		__mptcp_move_skbs(msk);
+		answ = mptcp_inq_hint(sk);
+		release_sock(sk);
+		break;
+	case SIOCOUTQ:
+		slow = lock_sock_fast(sk);
+		answ = mptcp_ioctl_outq(msk, READ_ONCE(msk->snd_una));
+		unlock_sock_fast(sk, slow);
+		break;
+	case SIOCOUTQNSD:
+		slow = lock_sock_fast(sk);
+		answ = mptcp_ioctl_outq(msk, msk->snd_nxt);
+		unlock_sock_fast(sk, slow);
+		break;
+	default:
+		return -ENOIOCTLCMD;
+	}
+
+	return put_user(answ, (int __user *)arg);
+}
+
 static struct proto mptcp_prot = {
 	.name		= "MPTCP",
 	.owner		= THIS_MODULE,
@@ -3273,6 +3325,7 @@ static struct proto mptcp_prot = {
 	.shutdown	= mptcp_shutdown,
 	.destroy	= mptcp_destroy,
 	.sendmsg	= mptcp_sendmsg,
+	.ioctl		= mptcp_ioctl,
 	.recvmsg	= mptcp_recvmsg,
 	.release_cb	= mptcp_release_cb,
 	.hash		= mptcp_hash,
-- 
2.32.0

[PATCH mptcp-next v3 4/4] selftests: mptcp: add inq test case

[Florian Westphal]

client & server use a unix socket connection to communicate
outside of the mptcp connection.

This allows the consumer to know in advance how many bytes have been
(or will be) sent by the peer.
This allows stricter checks on the bytecounts reported by TCP_INQ cmsg.

Suggested-by: Mat Martineau <mathew.j.martineau@linux.intel.com>
Signed-off-by: Florian Westphal <fw@strlen.de>
---
 tools/testing/selftests/net/mptcp/.gitignore  |   1 +
 tools/testing/selftests/net/mptcp/Makefile    |   2 +-
 tools/testing/selftests/net/mptcp/mptcp_inq.c | 603 ++++++++++++++++++
 .../selftests/net/mptcp/mptcp_sockopt.sh      |  40 ++
 4 files changed, 645 insertions(+), 1 deletion(-)
 create mode 100644 tools/testing/selftests/net/mptcp/mptcp_inq.c

diff --git a/tools/testing/selftests/net/mptcp/.gitignore b/tools/testing/selftests/net/mptcp/.gitignore
index 7569d892967a..49daae73c41e 100644
--- a/tools/testing/selftests/net/mptcp/.gitignore
+++ b/tools/testing/selftests/net/mptcp/.gitignore
@@ -1,5 +1,6 @@
 # SPDX-License-Identifier: GPL-2.0-only
 mptcp_connect
+mptcp_inq
 mptcp_sockopt
 pm_nl_ctl
 *.pcap
diff --git a/tools/testing/selftests/net/mptcp/Makefile b/tools/testing/selftests/net/mptcp/Makefile
index bbf4e448bad9..0356c4501c99 100644
--- a/tools/testing/selftests/net/mptcp/Makefile
+++ b/tools/testing/selftests/net/mptcp/Makefile
@@ -8,7 +8,7 @@ CFLAGS =  -Wall -Wl,--no-as-needed -O2 -g  -I$(top_srcdir)/usr/include
 TEST_PROGS := mptcp_connect.sh pm_netlink.sh mptcp_join.sh diag.sh \
 	      simult_flows.sh mptcp_sockopt.sh
 
-TEST_GEN_FILES = mptcp_connect pm_nl_ctl mptcp_sockopt
+TEST_GEN_FILES = mptcp_connect pm_nl_ctl mptcp_sockopt mptcp_inq
 
 TEST_FILES := settings
 
diff --git a/tools/testing/selftests/net/mptcp/mptcp_inq.c b/tools/testing/selftests/net/mptcp/mptcp_inq.c
new file mode 100644
index 000000000000..62ee2dd63b5f
--- /dev/null
+++ b/tools/testing/selftests/net/mptcp/mptcp_inq.c
@@ -0,0 +1,603 @@
+// SPDX-License-Identifier: GPL-2.0
+
+#define _GNU_SOURCE
+
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <limits.h>
+#include <string.h>
+#include <stdarg.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <inttypes.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <strings.h>
+#include <unistd.h>
+#include <time.h>
+
+#include <sys/ioctl.h>
+#include <sys/ioctl.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include <netdb.h>
+#include <netinet/in.h>
+
+#include <linux/tcp.h>
+#include <linux/sockios.h>
+
+#ifndef IPPROTO_MPTCP
+#define IPPROTO_MPTCP 262
+#endif
+#ifndef SOL_MPTCP
+#define SOL_MPTCP 284
+#endif
+
+static int pf = AF_INET;
+static int proto_tx = IPPROTO_MPTCP;
+static int proto_rx = IPPROTO_MPTCP;
+
+static void die_perror(const char *msg)
+{
+	perror(msg);
+	exit(1);
+}
+
+static void die_usage(int r)
+{
+	fprintf(stderr, "Usage: mptcp_inq [-6] [ -t tcp|mptcp ] [ -r tcp|mptcp]\n");
+	exit(r);
+}
+
+static void xerror(const char *fmt, ...)
+{
+	va_list ap;
+
+	va_start(ap, fmt);
+	vfprintf(stderr, fmt, ap);
+	va_end(ap);
+	fputc('\n', stderr);
+	exit(1);
+}
+
+static const char *getxinfo_strerr(int err)
+{
+	if (err == EAI_SYSTEM)
+		return strerror(errno);
+
+	return gai_strerror(err);
+}
+
+static void xgetaddrinfo(const char *node, const char *service,
+			 const struct addrinfo *hints,
+			 struct addrinfo **res)
+{
+	int err = getaddrinfo(node, service, hints, res);
+
+	if (err) {
+		const char *errstr = getxinfo_strerr(err);
+
+		fprintf(stderr, "Fatal: getaddrinfo(%s:%s): %s\n",
+			node ? node : "", service ? service : "", errstr);
+		exit(1);
+	}
+}
+
+static int sock_listen_mptcp(const char * const listenaddr,
+			     const char * const port)
+{
+	int sock;
+	struct addrinfo hints = {
+		.ai_protocol = IPPROTO_TCP,
+		.ai_socktype = SOCK_STREAM,
+		.ai_flags = AI_PASSIVE | AI_NUMERICHOST
+	};
+
+	hints.ai_family = pf;
+
+	struct addrinfo *a, *addr;
+	int one = 1;
+
+	xgetaddrinfo(listenaddr, port, &hints, &addr);
+	hints.ai_family = pf;
+
+	for (a = addr; a; a = a->ai_next) {
+		sock = socket(a->ai_family, a->ai_socktype, proto_rx);
+		if (sock < 0)
+			continue;
+
+		if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one,
+				     sizeof(one)))
+			perror("setsockopt");
+
+		if (bind(sock, a->ai_addr, a->ai_addrlen) == 0)
+			break; /* success */
+
+		perror("bind");
+		close(sock);
+		sock = -1;
+	}
+
+	freeaddrinfo(addr);
+
+	if (sock < 0)
+		xerror("could not create listen socket");
+
+	if (listen(sock, 20))
+		die_perror("listen");
+
+	return sock;
+}
+
+static int sock_connect_mptcp(const char * const remoteaddr,
+			      const char * const port, int proto)
+{
+	struct addrinfo hints = {
+		.ai_protocol = IPPROTO_TCP,
+		.ai_socktype = SOCK_STREAM,
+	};
+	struct addrinfo *a, *addr;
+	int sock = -1;
+
+	hints.ai_family = pf;
+
+	xgetaddrinfo(remoteaddr, port, &hints, &addr);
+	for (a = addr; a; a = a->ai_next) {
+		sock = socket(a->ai_family, a->ai_socktype, proto);
+		if (sock < 0)
+			continue;
+
+		if (connect(sock, a->ai_addr, a->ai_addrlen) == 0)
+			break; /* success */
+
+		die_perror("connect");
+	}
+
+	if (sock < 0)
+		xerror("could not create connect socket");
+
+	freeaddrinfo(addr);
+	return sock;
+}
+
+static int protostr_to_num(const char *s)
+{
+	if (strcasecmp(s, "tcp") == 0)
+		return IPPROTO_TCP;
+	if (strcasecmp(s, "mptcp") == 0)
+		return IPPROTO_MPTCP;
+
+	die_usage(1);
+	return 0;
+}
+
+static void parse_opts(int argc, char **argv)
+{
+	int c;
+
+	while ((c = getopt(argc, argv, "h6t:r:")) != -1) {
+		switch (c) {
+		case 'h':
+			die_usage(0);
+			break;
+		case '6':
+			pf = AF_INET6;
+			break;
+		case 't':
+			proto_tx = protostr_to_num(optarg);
+			break;
+		case 'r':
+			proto_rx = protostr_to_num(optarg);
+			break;
+		default:
+			die_usage(1);
+			break;
+		}
+	}
+}
+
+/* wait up to timeout milliseconds */
+static void wait_for_ack(int fd, int timeout, size_t total)
+{
+	int i;
+
+	for (i = 0; i < timeout; i++) {
+		int nsd, ret, queued = -1;
+		struct timespec req;
+
+		ret = ioctl(fd, TIOCOUTQ, &queued);
+		if (ret < 0)
+			die_perror("TIOCOUTQ");
+
+		ret = ioctl(fd, SIOCOUTQNSD, &nsd);
+		if (ret < 0)
+			die_perror("SIOCOUTQNSD");
+
+		if ((size_t)queued > total)
+			xerror("TIOCOUTQ %u, but only %zu expected\n", queued, total);
+		assert(nsd <= queued);
+
+		if (queued == 0)
+			return;
+
+		/* wait for peer to ack rx of all data */
+		req.tv_sec = 0;
+		req.tv_nsec = 1 * 1000 * 1000ul; /* 1ms */
+		nanosleep(&req, NULL);
+	}
+
+	xerror("still tx data queued after %u ms\n", timeout);
+}
+
+static void connect_one_server(int fd, int unixfd)
+{
+	size_t len, i, total, sent;
+	char buf[4096], buf2[4096];
+	int queued, on = 1;
+	ssize_t ret;
+
+	len = rand() % (sizeof(buf) - 1);
+
+	if (len < 128)
+		len = 128;
+
+	for (i = 0; i < len ; i++) {
+		buf[i] = rand() % 26;
+		buf[i] += 'A';
+	}
+
+	buf[i] = '\n';
+
+	/* un-block server */
+	ret = read(unixfd, buf2, 4);
+	assert(ret == 4);
+
+	assert(strncmp(buf2, "xmit", 4) == 0);
+
+	ret = write(unixfd, &len, sizeof(len));
+	assert(ret == (ssize_t)sizeof(len));
+
+	ret = write(fd, buf, len);
+	if (ret < 0)
+		die_perror("write");
+
+	if (ret != (ssize_t)len)
+		xerror("short write");
+
+	ret = read(unixfd, buf2, 4);
+	assert(strncmp(buf2, "huge", 4) == 0);
+
+	total = rand() % (16 * 1024 * 1024);
+	total += (1 * 1024 * 1024);
+	sent = total;
+
+	ret = write(unixfd, &total, sizeof(total));
+	assert(ret == (ssize_t)sizeof(total));
+
+	wait_for_ack(fd, 5000, len);
+
+	while (total > 0) {
+		if (total > sizeof(buf))
+			len = sizeof(buf);
+		else
+			len = total;
+
+		ret = write(fd, buf, len);
+		if (ret < 0)
+			die_perror("write");
+		total -= ret;
+
+		/* we don't have to care about buf content, only
+		 * number of total bytes sent
+		 */
+	}
+
+	ret = read(unixfd, buf2, 4);
+	assert(ret == 4);
+	assert(strncmp(buf2, "shut", 4) == 0);
+
+	wait_for_ack(fd, 5000, sent);
+
+	write(fd, buf, 1);
+	close(fd);
+	ret = write(unixfd, "closed", 6);
+	assert(ret == 6);
+
+	close(unixfd);
+}
+
+static void get_tcp_inq(struct msghdr *msgh, unsigned int *inqv)
+{
+	struct cmsghdr *cmsg;
+
+	for (cmsg = CMSG_FIRSTHDR(msgh); cmsg ; cmsg = CMSG_NXTHDR(msgh, cmsg)) {
+		if (cmsg->cmsg_level == IPPROTO_TCP && cmsg->cmsg_type == TCP_CM_INQ) {
+			memcpy(inqv, CMSG_DATA(cmsg), sizeof(*inqv));
+			return;
+		}
+	}
+
+	xerror("could not find TCP_CM_INQ cmsg type");
+}
+
+static void process_one_client(int fd, int unixfd)
+{
+	unsigned int tcp_inq;
+	size_t expect_len;
+	char msg_buf[4096];
+	char buf[4096];
+	char tmp[16];
+	struct iovec iov = {
+		.iov_base = buf,
+		.iov_len = 1,
+	};
+	struct msghdr msg = {
+		.msg_iov = &iov,
+		.msg_iovlen = 1,
+		.msg_control = msg_buf,
+		.msg_controllen = sizeof(msg_buf),
+	};
+	ssize_t ret, tot;
+
+	ret = write(unixfd, "xmit", 4);
+	assert(ret == 4);
+
+	ret = read(unixfd, &expect_len, sizeof(expect_len));
+	assert(ret == (ssize_t)sizeof(expect_len));
+
+	if (expect_len > sizeof(buf))
+		xerror("expect len %zu exceeds buffer size", expect_len);
+
+	for (;;) {
+		struct timespec req;
+		unsigned int queued;
+
+		ret = ioctl(fd, FIONREAD, &queued);
+		if (ret < 0)
+			die_perror("FIONREAD");
+		if (queued > expect_len)
+			xerror("FIONREAD returned %u, but only %zu expected\n",
+					queued, expect_len);
+		if (queued == expect_len)
+			break;
+
+		req.tv_sec = 0;
+		req.tv_nsec = 1000 * 1000ul;
+		nanosleep(&req, NULL);
+	}
+
+	/* read one byte, expect cmsg to return expected - 1 */
+	ret = recvmsg(fd, &msg, 0);
+	if (ret < 0)
+		die_perror("recvmsg");
+
+	if (msg.msg_controllen == 0)
+		xerror("msg_controllen is 0");
+
+	get_tcp_inq(&msg, &tcp_inq);
+
+	assert((size_t)tcp_inq == (expect_len - 1));
+
+	iov.iov_len = sizeof(buf);
+	ret = recvmsg(fd, &msg, 0);
+	if (ret < 0)
+		die_perror("recvmsg");
+
+	/* should have gotten exact remainder of all pending data */
+	assert(ret == (ssize_t)tcp_inq);
+
+	/* should be 0, all drained */
+	get_tcp_inq(&msg, &tcp_inq);
+	assert(tcp_inq == 0);
+
+	/* request a large swath of data. */
+	ret = write(unixfd, "huge", 4);
+	assert(ret == 4);
+
+	ret = read(unixfd, &expect_len, sizeof(expect_len));
+	assert(ret == (ssize_t)sizeof(expect_len));
+
+	/* peer should send us a few mb of data */
+	if (expect_len <= sizeof(buf))
+		xerror("expect len %zu too small\n", expect_len);
+
+	tot = 0;
+	do {
+		iov.iov_len = sizeof(buf);
+		ret = recvmsg(fd, &msg, 0);
+		if (ret < 0)
+			die_perror("recvmsg");
+
+		tot += ret;
+
+		get_tcp_inq(&msg, &tcp_inq);
+
+		if (tcp_inq > expect_len - tot)
+			xerror("inq %d, remaining %d total_len %d\n",
+					tcp_inq, expect_len - tot, (int)expect_len);
+
+		assert(tcp_inq <= expect_len - tot);
+	} while ((size_t)tot < expect_len);
+
+	ret = write(unixfd, "shut", 4);
+	assert(ret == 4);
+
+	/* wait for hangup. Should have received one more byte of data. */
+	ret = read(unixfd, tmp, sizeof(tmp));
+	assert(ret == 6);
+	assert(strncmp(tmp, "closed", 6) == 0);
+
+	sleep(1);
+
+	iov.iov_len = 1;
+	ret = recvmsg(fd, &msg, 0);
+	if (ret < 0)
+		die_perror("recvmsg");
+	assert(ret == 1);
+
+	get_tcp_inq(&msg, &tcp_inq);
+
+	/* tcp_inq should be 1 due to received fin. */
+	assert(tcp_inq == 1);
+
+	iov.iov_len = 1;
+	ret = recvmsg(fd, &msg, 0);
+	if (ret < 0)
+		die_perror("recvmsg");
+
+	/* expect EOF */
+	assert(ret == 0);
+	get_tcp_inq(&msg, &tcp_inq);
+	assert(tcp_inq == 1);
+
+	close(fd);
+}
+
+static int xaccept(int s)
+{
+	int fd = accept(s, NULL, 0);
+
+	if (fd < 0)
+		die_perror("accept");
+
+	return fd;
+}
+
+static int server(int unixfd)
+{
+	int fd = -1, r, on = 1;
+
+	switch (pf) {
+	case AF_INET:
+		fd = sock_listen_mptcp("127.0.0.1", "15432");
+		break;
+	case AF_INET6:
+		fd = sock_listen_mptcp("::1", "15432");
+		break;
+	default:
+		xerror("Unknown pf %d\n", pf);
+		break;
+	}
+
+	r = write(unixfd, "conn", 4);
+	assert(r == 4);
+
+	alarm(15);
+	r = xaccept(fd);
+
+	if (-1 == setsockopt(r, IPPROTO_TCP, TCP_INQ, &on, sizeof(on)))
+		die_perror("setsockopt");
+
+	process_one_client(r, unixfd);
+
+	return 0;
+}
+
+static int client(int unixfd)
+{
+	int fd = -1;
+
+	alarm(15);
+
+	switch (pf) {
+	case AF_INET:
+		fd = sock_connect_mptcp("127.0.0.1", "15432", proto_tx);
+		break;
+	case AF_INET6:
+		fd = sock_connect_mptcp("::1", "15432", proto_tx);
+		break;
+	default:
+		xerror("Unknown pf %d\n", pf);
+	}
+
+	connect_one_server(fd, unixfd);
+
+	return 0;
+}
+
+static void init_rng(void)
+{
+	int fd = open("/dev/urandom", O_RDONLY);
+	unsigned int foo;
+
+	if (fd > 0) {
+		int ret = read(fd, &foo, sizeof(foo));
+
+		if (ret < 0)
+			srand(fd + foo);
+		close(fd);
+	}
+
+	srand(foo);
+}
+
+static pid_t xfork(void)
+{
+	pid_t p = fork();
+
+	if (p < 0)
+		die_perror("fork");
+	else if (p == 0)
+		init_rng();
+
+	return p;
+}
+
+static int rcheck(int wstatus, const char *what)
+{
+	if (WIFEXITED(wstatus)) {
+		if (WEXITSTATUS(wstatus) == 0)
+			return 0;
+		fprintf(stderr, "%s exited, status=%d\n", what, WEXITSTATUS(wstatus));
+		return WEXITSTATUS(wstatus);
+	} else if (WIFSIGNALED(wstatus)) {
+		xerror("%s killed by signal %d\n", what, WTERMSIG(wstatus));
+	} else if (WIFSTOPPED(wstatus)) {
+		xerror("%s stopped by signal %d\n", what, WSTOPSIG(wstatus));
+	}
+
+	return 111;
+}
+
+int main(int argc, char *argv[])
+{
+	int e1, e2, wstatus;
+	pid_t s, c, ret;
+	int unixfds[2];
+
+	parse_opts(argc, argv);
+
+	e1 = socketpair(AF_UNIX, SOCK_DGRAM, 0, unixfds);
+	if (e1 < 0)
+		die_perror("pipe");
+
+	s = xfork();
+	if (s == 0)
+		return server(unixfds[1]);
+
+	close(unixfds[1]);
+
+	/* wait until server bound a socket */
+	e1 = read(unixfds[0], &e1, 4);
+	assert(e1 == 4);
+
+	c = xfork();
+	if (c == 0)
+		return client(unixfds[0]);
+
+	close(unixfds[0]);
+
+	ret = waitpid(s, &wstatus, 0);
+	if (ret == -1)
+		die_perror("waitpid");
+	e1 = rcheck(wstatus, "server");
+	ret = waitpid(c, &wstatus, 0);
+	if (ret == -1)
+		die_perror("waitpid");
+	e2 = rcheck(wstatus, "client");
+
+	return e1 ? e1 : e2;
+}
diff --git a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
index c8c364369599..0879da915014 100755
--- a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
+++ b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
@@ -279,6 +279,45 @@ run_tests()
 	fi
 }
 
+do_tcpinq_test()
+{
+	ip netns exec "$ns1" ./mptcp_inq "$@"
+	lret=$?
+	if [ $lret -ne 0 ];then
+		ret=$lret
+		echo "FAIL: mptcp_inq $@" 1>&2
+		return $lret
+	fi
+
+	echo "PASS: TCP_INQ cmsg/ioctl $@"
+	return $lret
+}
+
+do_tcpinq_tests()
+{
+	local lret=0
+
+	ip netns exec "$ns1" iptables -F
+	ip netns exec "$ns1" ip6tables -F
+
+	for args in "-t tcp" "-r tcp"; do
+		do_tcpinq_test $args
+		lret=$?
+		if [ $lret -ne 0 ] ; then
+			return $lret
+		fi
+		do_tcpinq_test -6 $args
+		lret=$?
+		if [ $lret -ne 0 ] ; then
+			return $lret
+		fi
+	done
+
+	do_tcpinq_test -r tcp -t tcp
+
+	return $?
+}
+
 sin=$(mktemp)
 sout=$(mktemp)
 cin=$(mktemp)
@@ -300,4 +339,5 @@ if [ $ret -eq 0 ];then
 	echo "PASS: SOL_MPTCP getsockopt has expected information"
 fi
 
+do_tcpinq_tests
 exit $ret
-- 
2.32.0

Re: [PATCH mptcp-next v3 4/4] selftests: mptcp: add inq test case

[Mat Martineau]

On Tue, 16 Nov 2021, Florian Westphal wrote:

> client & server use a unix socket connection to communicate
> outside of the mptcp connection.
>
> This allows the consumer to know in advance how many bytes have been
> (or will be) sent by the peer.
> This allows stricter checks on the bytecounts reported by TCP_INQ cmsg.
>
> Suggested-by: Mat Martineau <mathew.j.martineau@linux.intel.com>
> Signed-off-by: Florian Westphal <fw@strlen.de>
> ---
> tools/testing/selftests/net/mptcp/.gitignore  |   1 +
> tools/testing/selftests/net/mptcp/Makefile    |   2 +-
> tools/testing/selftests/net/mptcp/mptcp_inq.c | 603 ++++++++++++++++++
> .../selftests/net/mptcp/mptcp_sockopt.sh      |  40 ++
> 4 files changed, 645 insertions(+), 1 deletion(-)
> create mode 100644 tools/testing/selftests/net/mptcp/mptcp_inq.c
>
> diff --git a/tools/testing/selftests/net/mptcp/.gitignore b/tools/testing/selftests/net/mptcp/.gitignore
> index 7569d892967a..49daae73c41e 100644
> --- a/tools/testing/selftests/net/mptcp/.gitignore
> +++ b/tools/testing/selftests/net/mptcp/.gitignore
> @@ -1,5 +1,6 @@
> # SPDX-License-Identifier: GPL-2.0-only
> mptcp_connect
> +mptcp_inq
> mptcp_sockopt
> pm_nl_ctl
> *.pcap
> diff --git a/tools/testing/selftests/net/mptcp/Makefile b/tools/testing/selftests/net/mptcp/Makefile
> index bbf4e448bad9..0356c4501c99 100644
> --- a/tools/testing/selftests/net/mptcp/Makefile
> +++ b/tools/testing/selftests/net/mptcp/Makefile
> @@ -8,7 +8,7 @@ CFLAGS =  -Wall -Wl,--no-as-needed -O2 -g  -I$(top_srcdir)/usr/include
> TEST_PROGS := mptcp_connect.sh pm_netlink.sh mptcp_join.sh diag.sh \
> 	      simult_flows.sh mptcp_sockopt.sh
>
> -TEST_GEN_FILES = mptcp_connect pm_nl_ctl mptcp_sockopt
> +TEST_GEN_FILES = mptcp_connect pm_nl_ctl mptcp_sockopt mptcp_inq
>
> TEST_FILES := settings
>
> diff --git a/tools/testing/selftests/net/mptcp/mptcp_inq.c b/tools/testing/selftests/net/mptcp/mptcp_inq.c
> new file mode 100644
> index 000000000000..62ee2dd63b5f
> --- /dev/null
> +++ b/tools/testing/selftests/net/mptcp/mptcp_inq.c
> @@ -0,0 +1,603 @@
> +// SPDX-License-Identifier: GPL-2.0
> +
> +#define _GNU_SOURCE
> +
> +#include <assert.h>
> +#include <errno.h>
> +#include <fcntl.h>
> +#include <limits.h>
> +#include <string.h>
> +#include <stdarg.h>
> +#include <stdbool.h>
> +#include <stdint.h>
> +#include <inttypes.h>
> +#include <stdio.h>
> +#include <stdlib.h>
> +#include <strings.h>
> +#include <unistd.h>
> +#include <time.h>
> +
> +#include <sys/ioctl.h>
> +#include <sys/ioctl.h>
> +#include <sys/socket.h>
> +#include <sys/types.h>
> +#include <sys/wait.h>
> +
> +#include <netdb.h>
> +#include <netinet/in.h>
> +
> +#include <linux/tcp.h>
> +#include <linux/sockios.h>
> +
> +#ifndef IPPROTO_MPTCP
> +#define IPPROTO_MPTCP 262
> +#endif
> +#ifndef SOL_MPTCP
> +#define SOL_MPTCP 284
> +#endif
> +
> +static int pf = AF_INET;
> +static int proto_tx = IPPROTO_MPTCP;
> +static int proto_rx = IPPROTO_MPTCP;
> +
> +static void die_perror(const char *msg)
> +{
> +	perror(msg);
> +	exit(1);
> +}
> +
> +static void die_usage(int r)
> +{
> +	fprintf(stderr, "Usage: mptcp_inq [-6] [ -t tcp|mptcp ] [ -r tcp|mptcp]\n");
> +	exit(r);
> +}
> +
> +static void xerror(const char *fmt, ...)
> +{
> +	va_list ap;
> +
> +	va_start(ap, fmt);
> +	vfprintf(stderr, fmt, ap);
> +	va_end(ap);
> +	fputc('\n', stderr);
> +	exit(1);
> +}
> +
> +static const char *getxinfo_strerr(int err)
> +{
> +	if (err == EAI_SYSTEM)
> +		return strerror(errno);
> +
> +	return gai_strerror(err);
> +}
> +
> +static void xgetaddrinfo(const char *node, const char *service,
> +			 const struct addrinfo *hints,
> +			 struct addrinfo **res)
> +{
> +	int err = getaddrinfo(node, service, hints, res);
> +
> +	if (err) {
> +		const char *errstr = getxinfo_strerr(err);
> +
> +		fprintf(stderr, "Fatal: getaddrinfo(%s:%s): %s\n",
> +			node ? node : "", service ? service : "", errstr);
> +		exit(1);
> +	}
> +}
> +
> +static int sock_listen_mptcp(const char * const listenaddr,
> +			     const char * const port)
> +{
> +	int sock;
> +	struct addrinfo hints = {
> +		.ai_protocol = IPPROTO_TCP,
> +		.ai_socktype = SOCK_STREAM,
> +		.ai_flags = AI_PASSIVE | AI_NUMERICHOST
> +	};
> +
> +	hints.ai_family = pf;
> +
> +	struct addrinfo *a, *addr;
> +	int one = 1;
> +
> +	xgetaddrinfo(listenaddr, port, &hints, &addr);
> +	hints.ai_family = pf;
> +
> +	for (a = addr; a; a = a->ai_next) {
> +		sock = socket(a->ai_family, a->ai_socktype, proto_rx);
> +		if (sock < 0)
> +			continue;
> +
> +		if (-1 == setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one,
> +				     sizeof(one)))
> +			perror("setsockopt");
> +
> +		if (bind(sock, a->ai_addr, a->ai_addrlen) == 0)
> +			break; /* success */
> +
> +		perror("bind");
> +		close(sock);
> +		sock = -1;
> +	}
> +
> +	freeaddrinfo(addr);
> +
> +	if (sock < 0)
> +		xerror("could not create listen socket");
> +
> +	if (listen(sock, 20))
> +		die_perror("listen");
> +
> +	return sock;
> +}
> +
> +static int sock_connect_mptcp(const char * const remoteaddr,
> +			      const char * const port, int proto)
> +{
> +	struct addrinfo hints = {
> +		.ai_protocol = IPPROTO_TCP,
> +		.ai_socktype = SOCK_STREAM,
> +	};
> +	struct addrinfo *a, *addr;
> +	int sock = -1;
> +
> +	hints.ai_family = pf;
> +
> +	xgetaddrinfo(remoteaddr, port, &hints, &addr);
> +	for (a = addr; a; a = a->ai_next) {
> +		sock = socket(a->ai_family, a->ai_socktype, proto);
> +		if (sock < 0)
> +			continue;
> +
> +		if (connect(sock, a->ai_addr, a->ai_addrlen) == 0)
> +			break; /* success */
> +
> +		die_perror("connect");
> +	}
> +
> +	if (sock < 0)
> +		xerror("could not create connect socket");
> +
> +	freeaddrinfo(addr);
> +	return sock;
> +}
> +
> +static int protostr_to_num(const char *s)
> +{
> +	if (strcasecmp(s, "tcp") == 0)
> +		return IPPROTO_TCP;
> +	if (strcasecmp(s, "mptcp") == 0)
> +		return IPPROTO_MPTCP;
> +
> +	die_usage(1);
> +	return 0;
> +}
> +
> +static void parse_opts(int argc, char **argv)
> +{
> +	int c;
> +
> +	while ((c = getopt(argc, argv, "h6t:r:")) != -1) {
> +		switch (c) {
> +		case 'h':
> +			die_usage(0);
> +			break;
> +		case '6':
> +			pf = AF_INET6;
> +			break;
> +		case 't':
> +			proto_tx = protostr_to_num(optarg);
> +			break;
> +		case 'r':
> +			proto_rx = protostr_to_num(optarg);
> +			break;
> +		default:
> +			die_usage(1);
> +			break;
> +		}
> +	}
> +}
> +
> +/* wait up to timeout milliseconds */
> +static void wait_for_ack(int fd, int timeout, size_t total)
> +{
> +	int i;
> +
> +	for (i = 0; i < timeout; i++) {
> +		int nsd, ret, queued = -1;
> +		struct timespec req;
> +
> +		ret = ioctl(fd, TIOCOUTQ, &queued);
> +		if (ret < 0)
> +			die_perror("TIOCOUTQ");
> +
> +		ret = ioctl(fd, SIOCOUTQNSD, &nsd);
> +		if (ret < 0)
> +			die_perror("SIOCOUTQNSD");
> +
> +		if ((size_t)queued > total)
> +			xerror("TIOCOUTQ %u, but only %zu expected\n", queued, total);
> +		assert(nsd <= queued);
> +
> +		if (queued == 0)
> +			return;
> +
> +		/* wait for peer to ack rx of all data */
> +		req.tv_sec = 0;
> +		req.tv_nsec = 1 * 1000 * 1000ul; /* 1ms */
> +		nanosleep(&req, NULL);
> +	}
> +
> +	xerror("still tx data queued after %u ms\n", timeout);
> +}
> +
> +static void connect_one_server(int fd, int unixfd)
> +{
> +	size_t len, i, total, sent;
> +	char buf[4096], buf2[4096];
> +	int queued, on = 1;

Still getting the 'unused variable' warnings on this line, but we can 
squash a fix for that (or wait to apply just this last patch?).

- Mat

> +	ssize_t ret;
> +
> +	len = rand() % (sizeof(buf) - 1);
> +
> +	if (len < 128)
> +		len = 128;
> +
> +	for (i = 0; i < len ; i++) {
> +		buf[i] = rand() % 26;
> +		buf[i] += 'A';
> +	}
> +
> +	buf[i] = '\n';
> +
> +	/* un-block server */
> +	ret = read(unixfd, buf2, 4);
> +	assert(ret == 4);
> +
> +	assert(strncmp(buf2, "xmit", 4) == 0);
> +
> +	ret = write(unixfd, &len, sizeof(len));
> +	assert(ret == (ssize_t)sizeof(len));
> +
> +	ret = write(fd, buf, len);
> +	if (ret < 0)
> +		die_perror("write");
> +
> +	if (ret != (ssize_t)len)
> +		xerror("short write");
> +
> +	ret = read(unixfd, buf2, 4);
> +	assert(strncmp(buf2, "huge", 4) == 0);
> +
> +	total = rand() % (16 * 1024 * 1024);
> +	total += (1 * 1024 * 1024);
> +	sent = total;
> +
> +	ret = write(unixfd, &total, sizeof(total));
> +	assert(ret == (ssize_t)sizeof(total));
> +
> +	wait_for_ack(fd, 5000, len);
> +
> +	while (total > 0) {
> +		if (total > sizeof(buf))
> +			len = sizeof(buf);
> +		else
> +			len = total;
> +
> +		ret = write(fd, buf, len);
> +		if (ret < 0)
> +			die_perror("write");
> +		total -= ret;
> +
> +		/* we don't have to care about buf content, only
> +		 * number of total bytes sent
> +		 */
> +	}
> +
> +	ret = read(unixfd, buf2, 4);
> +	assert(ret == 4);
> +	assert(strncmp(buf2, "shut", 4) == 0);
> +
> +	wait_for_ack(fd, 5000, sent);
> +
> +	write(fd, buf, 1);
> +	close(fd);
> +	ret = write(unixfd, "closed", 6);
> +	assert(ret == 6);
> +
> +	close(unixfd);
> +}
> +
> +static void get_tcp_inq(struct msghdr *msgh, unsigned int *inqv)
> +{
> +	struct cmsghdr *cmsg;
> +
> +	for (cmsg = CMSG_FIRSTHDR(msgh); cmsg ; cmsg = CMSG_NXTHDR(msgh, cmsg)) {
> +		if (cmsg->cmsg_level == IPPROTO_TCP && cmsg->cmsg_type == TCP_CM_INQ) {
> +			memcpy(inqv, CMSG_DATA(cmsg), sizeof(*inqv));
> +			return;
> +		}
> +	}
> +
> +	xerror("could not find TCP_CM_INQ cmsg type");
> +}
> +
> +static void process_one_client(int fd, int unixfd)
> +{
> +	unsigned int tcp_inq;
> +	size_t expect_len;
> +	char msg_buf[4096];
> +	char buf[4096];
> +	char tmp[16];
> +	struct iovec iov = {
> +		.iov_base = buf,
> +		.iov_len = 1,
> +	};
> +	struct msghdr msg = {
> +		.msg_iov = &iov,
> +		.msg_iovlen = 1,
> +		.msg_control = msg_buf,
> +		.msg_controllen = sizeof(msg_buf),
> +	};
> +	ssize_t ret, tot;
> +
> +	ret = write(unixfd, "xmit", 4);
> +	assert(ret == 4);
> +
> +	ret = read(unixfd, &expect_len, sizeof(expect_len));
> +	assert(ret == (ssize_t)sizeof(expect_len));
> +
> +	if (expect_len > sizeof(buf))
> +		xerror("expect len %zu exceeds buffer size", expect_len);
> +
> +	for (;;) {
> +		struct timespec req;
> +		unsigned int queued;
> +
> +		ret = ioctl(fd, FIONREAD, &queued);
> +		if (ret < 0)
> +			die_perror("FIONREAD");
> +		if (queued > expect_len)
> +			xerror("FIONREAD returned %u, but only %zu expected\n",
> +					queued, expect_len);
> +		if (queued == expect_len)
> +			break;
> +
> +		req.tv_sec = 0;
> +		req.tv_nsec = 1000 * 1000ul;
> +		nanosleep(&req, NULL);
> +	}
> +
> +	/* read one byte, expect cmsg to return expected - 1 */
> +	ret = recvmsg(fd, &msg, 0);
> +	if (ret < 0)
> +		die_perror("recvmsg");
> +
> +	if (msg.msg_controllen == 0)
> +		xerror("msg_controllen is 0");
> +
> +	get_tcp_inq(&msg, &tcp_inq);
> +
> +	assert((size_t)tcp_inq == (expect_len - 1));
> +
> +	iov.iov_len = sizeof(buf);
> +	ret = recvmsg(fd, &msg, 0);
> +	if (ret < 0)
> +		die_perror("recvmsg");
> +
> +	/* should have gotten exact remainder of all pending data */
> +	assert(ret == (ssize_t)tcp_inq);
> +
> +	/* should be 0, all drained */
> +	get_tcp_inq(&msg, &tcp_inq);
> +	assert(tcp_inq == 0);
> +
> +	/* request a large swath of data. */
> +	ret = write(unixfd, "huge", 4);
> +	assert(ret == 4);
> +
> +	ret = read(unixfd, &expect_len, sizeof(expect_len));
> +	assert(ret == (ssize_t)sizeof(expect_len));
> +
> +	/* peer should send us a few mb of data */
> +	if (expect_len <= sizeof(buf))
> +		xerror("expect len %zu too small\n", expect_len);
> +
> +	tot = 0;
> +	do {
> +		iov.iov_len = sizeof(buf);
> +		ret = recvmsg(fd, &msg, 0);
> +		if (ret < 0)
> +			die_perror("recvmsg");
> +
> +		tot += ret;
> +
> +		get_tcp_inq(&msg, &tcp_inq);
> +
> +		if (tcp_inq > expect_len - tot)
> +			xerror("inq %d, remaining %d total_len %d\n",
> +					tcp_inq, expect_len - tot, (int)expect_len);
> +
> +		assert(tcp_inq <= expect_len - tot);
> +	} while ((size_t)tot < expect_len);
> +
> +	ret = write(unixfd, "shut", 4);
> +	assert(ret == 4);
> +
> +	/* wait for hangup. Should have received one more byte of data. */
> +	ret = read(unixfd, tmp, sizeof(tmp));
> +	assert(ret == 6);
> +	assert(strncmp(tmp, "closed", 6) == 0);
> +
> +	sleep(1);
> +
> +	iov.iov_len = 1;
> +	ret = recvmsg(fd, &msg, 0);
> +	if (ret < 0)
> +		die_perror("recvmsg");
> +	assert(ret == 1);
> +
> +	get_tcp_inq(&msg, &tcp_inq);
> +
> +	/* tcp_inq should be 1 due to received fin. */
> +	assert(tcp_inq == 1);
> +
> +	iov.iov_len = 1;
> +	ret = recvmsg(fd, &msg, 0);
> +	if (ret < 0)
> +		die_perror("recvmsg");
> +
> +	/* expect EOF */
> +	assert(ret == 0);
> +	get_tcp_inq(&msg, &tcp_inq);
> +	assert(tcp_inq == 1);
> +
> +	close(fd);
> +}
> +
> +static int xaccept(int s)
> +{
> +	int fd = accept(s, NULL, 0);
> +
> +	if (fd < 0)
> +		die_perror("accept");
> +
> +	return fd;
> +}
> +
> +static int server(int unixfd)
> +{
> +	int fd = -1, r, on = 1;
> +
> +	switch (pf) {
> +	case AF_INET:
> +		fd = sock_listen_mptcp("127.0.0.1", "15432");
> +		break;
> +	case AF_INET6:
> +		fd = sock_listen_mptcp("::1", "15432");
> +		break;
> +	default:
> +		xerror("Unknown pf %d\n", pf);
> +		break;
> +	}
> +
> +	r = write(unixfd, "conn", 4);
> +	assert(r == 4);
> +
> +	alarm(15);
> +	r = xaccept(fd);
> +
> +	if (-1 == setsockopt(r, IPPROTO_TCP, TCP_INQ, &on, sizeof(on)))
> +		die_perror("setsockopt");
> +
> +	process_one_client(r, unixfd);
> +
> +	return 0;
> +}
> +
> +static int client(int unixfd)
> +{
> +	int fd = -1;
> +
> +	alarm(15);
> +
> +	switch (pf) {
> +	case AF_INET:
> +		fd = sock_connect_mptcp("127.0.0.1", "15432", proto_tx);
> +		break;
> +	case AF_INET6:
> +		fd = sock_connect_mptcp("::1", "15432", proto_tx);
> +		break;
> +	default:
> +		xerror("Unknown pf %d\n", pf);
> +	}
> +
> +	connect_one_server(fd, unixfd);
> +
> +	return 0;
> +}
> +
> +static void init_rng(void)
> +{
> +	int fd = open("/dev/urandom", O_RDONLY);
> +	unsigned int foo;
> +
> +	if (fd > 0) {
> +		int ret = read(fd, &foo, sizeof(foo));
> +
> +		if (ret < 0)
> +			srand(fd + foo);
> +		close(fd);
> +	}
> +
> +	srand(foo);
> +}
> +
> +static pid_t xfork(void)
> +{
> +	pid_t p = fork();
> +
> +	if (p < 0)
> +		die_perror("fork");
> +	else if (p == 0)
> +		init_rng();
> +
> +	return p;
> +}
> +
> +static int rcheck(int wstatus, const char *what)
> +{
> +	if (WIFEXITED(wstatus)) {
> +		if (WEXITSTATUS(wstatus) == 0)
> +			return 0;
> +		fprintf(stderr, "%s exited, status=%d\n", what, WEXITSTATUS(wstatus));
> +		return WEXITSTATUS(wstatus);
> +	} else if (WIFSIGNALED(wstatus)) {
> +		xerror("%s killed by signal %d\n", what, WTERMSIG(wstatus));
> +	} else if (WIFSTOPPED(wstatus)) {
> +		xerror("%s stopped by signal %d\n", what, WSTOPSIG(wstatus));
> +	}
> +
> +	return 111;
> +}
> +
> +int main(int argc, char *argv[])
> +{
> +	int e1, e2, wstatus;
> +	pid_t s, c, ret;
> +	int unixfds[2];
> +
> +	parse_opts(argc, argv);
> +
> +	e1 = socketpair(AF_UNIX, SOCK_DGRAM, 0, unixfds);
> +	if (e1 < 0)
> +		die_perror("pipe");
> +
> +	s = xfork();
> +	if (s == 0)
> +		return server(unixfds[1]);
> +
> +	close(unixfds[1]);
> +
> +	/* wait until server bound a socket */
> +	e1 = read(unixfds[0], &e1, 4);
> +	assert(e1 == 4);
> +
> +	c = xfork();
> +	if (c == 0)
> +		return client(unixfds[0]);
> +
> +	close(unixfds[0]);
> +
> +	ret = waitpid(s, &wstatus, 0);
> +	if (ret == -1)
> +		die_perror("waitpid");
> +	e1 = rcheck(wstatus, "server");
> +	ret = waitpid(c, &wstatus, 0);
> +	if (ret == -1)
> +		die_perror("waitpid");
> +	e2 = rcheck(wstatus, "client");
> +
> +	return e1 ? e1 : e2;
> +}
> diff --git a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
> index c8c364369599..0879da915014 100755
> --- a/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
> +++ b/tools/testing/selftests/net/mptcp/mptcp_sockopt.sh
> @@ -279,6 +279,45 @@ run_tests()
> 	fi
> }
>
> +do_tcpinq_test()
> +{
> +	ip netns exec "$ns1" ./mptcp_inq "$@"
> +	lret=$?
> +	if [ $lret -ne 0 ];then
> +		ret=$lret
> +		echo "FAIL: mptcp_inq $@" 1>&2
> +		return $lret
> +	fi
> +
> +	echo "PASS: TCP_INQ cmsg/ioctl $@"
> +	return $lret
> +}
> +
> +do_tcpinq_tests()
> +{
> +	local lret=0
> +
> +	ip netns exec "$ns1" iptables -F
> +	ip netns exec "$ns1" ip6tables -F
> +
> +	for args in "-t tcp" "-r tcp"; do
> +		do_tcpinq_test $args
> +		lret=$?
> +		if [ $lret -ne 0 ] ; then
> +			return $lret
> +		fi
> +		do_tcpinq_test -6 $args
> +		lret=$?
> +		if [ $lret -ne 0 ] ; then
> +			return $lret
> +		fi
> +	done
> +
> +	do_tcpinq_test -r tcp -t tcp
> +
> +	return $?
> +}
> +
> sin=$(mktemp)
> sout=$(mktemp)
> cin=$(mktemp)
> @@ -300,4 +339,5 @@ if [ $ret -eq 0 ];then
> 	echo "PASS: SOL_MPTCP getsockopt has expected information"
> fi
>
> +do_tcpinq_tests
> exit $ret
> -- 
> 2.32.0
>
>

--
Mat Martineau
Intel

Re: [PATCH v3 mptcp-next 0/4] TCP_INQ support

[Mat Martineau]

On Tue, 16 Nov 2021, Florian Westphal wrote:

> No major changes since v2, see individual patches for details.
>
> This adds TCP_INQ for mptcp and extends the selftest infra.
> Patch 3 adds ioctls to retrieve amount of in-sequence bytes
> ready to for read() (older version of TCP_INQ) and counters for
> unsent and unacked byte count.
>
> Last patch is new in v2, its a more specific TCP_INQ test case.
> See individual patches for changes since v1.
>
> Florian Westphal (4):
>  mptcp: add TCP_INQ cmsg support
>  selftests: mptcp: add TCP_INQ support
>  mptcp: add SIOCINQ, OUTQ and OUTQNSD ioctls
>  selftests: mptcp: add inq test case
>
> net/mptcp/protocol.c                          |  89 ++-
> net/mptcp/protocol.h                          |   1 +
> net/mptcp/sockopt.c                           |  37 ++
> tools/testing/selftests/net/mptcp/.gitignore  |   1 +
> tools/testing/selftests/net/mptcp/Makefile    |   2 +-
> .../selftests/net/mptcp/mptcp_connect.c       |  58 +-
> tools/testing/selftests/net/mptcp/mptcp_inq.c | 603 ++++++++++++++++++
> .../selftests/net/mptcp/mptcp_sockopt.sh      |  44 +-
> 8 files changed, 830 insertions(+), 5 deletions(-)
> create mode 100644 tools/testing/selftests/net/mptcp/mptcp_inq.c
>
> -- 
> 2.32.0
>

Thanks for the revisions Florian. Looks good to me (with the one comment 
about the unused variable warnings in patch 4 to address before 
upstreaming).

Reviewed-by: Mat Martineau <mathew.j.martineau@linux.intel.com>

--
Mat Martineau
Intel

Re: [PATCH mptcp-next v3 4/4] selftests: mptcp: add inq test case

[Florian Westphal]

Mat Martineau <mathew.j.martineau@linux.intel.com> wrote:
> > +static void connect_one_server(int fd, int unixfd)
> > +{
> > +	size_t len, i, total, sent;
> > +	char buf[4096], buf2[4096];
> > +	int queued, on = 1;
> 
> Still getting the 'unused variable' warnings on this line, but we can squash
> a fix for that (or wait to apply just this last patch?).

Sigh.  Mathieu, can you zap the line above (queued, on) after applying &
squash this?

Re: [PATCH mptcp-next v3 1/4] mptcp: add TCP_INQ cmsg support

[Mat Martineau]

On Tue, 16 Nov 2021, Florian Westphal wrote:

> Support the TCP_INQ setsockopt.
>
> This is a boolean that tells recvmsg path to include the remaining
> in-sequence bytes in the cmsg data.
>
> v2: do not use CB(skb)->offset, increment map_seq instead (Paolo Abeni)
> v3: adjust CB(skb)->map_seq when taking skb from ofo queue (Paolo Abeni)

To follow up from the meeting discussion today, I thought about it some 
more and suggest that we leave the v2/v3 annotations as-is when applying 
to the export branch.

In the unlikely scenario where a "netdev v2" would be needed, we could 
clarify MPTCP vs netdev or just skip to v4 :)

>
> Closes: https://github.com/multipath-tcp/mptcp_net-next/issues/224
> Signed-off-by: Florian Westphal <fw@strlen.de>
> ---
> net/mptcp/protocol.c | 36 +++++++++++++++++++++++++++++++++++-
> net/mptcp/protocol.h |  1 +
> net/mptcp/sockopt.c  | 37 +++++++++++++++++++++++++++++++++++++
> 3 files changed, 73 insertions(+), 1 deletion(-)

--
Mat Martineau
Intel

Re: [PATCH v3 mptcp-next 0/4] TCP_INQ support

[Matthieu Baerts]

Hi Florian, Mat,

On 16/11/2021 14:48, Florian Westphal wrote:
> No major changes since v2, see individual patches for details.
> 
> This adds TCP_INQ for mptcp and extends the selftest infra.
> Patch 3 adds ioctls to retrieve amount of in-sequence bytes
> ready to for read() (older version of TCP_INQ) and counters for
> unsent and unacked byte count.
> 
> Last patch is new in v2, its a more specific TCP_INQ test case.
> See individual patches for changes since v1.

Thank you for the patches, the reviews and the reply to my question
about the ChangeLog!

Now in our tree (features for next) with Mat's RvB tag (and without some
warnings from checkpatch in the .c files of the selftests, nothing
important)

- 2633e427d5c5: mptcp: add TCP_INQ cmsg support

- 18b1970dd676: selftests: mptcp: add TCP_INQ support

- b85f910f78d7: mptcp: add SIOCINQ, OUTQ and OUTQNSD ioctls

- 75e2fd074427: selftests: mptcp: add inq test case

- Results: 3bb02e3cfe5d..3ebf8b8d1426

Builds and tests are now in progress:

https://cirrus-ci.com/github/multipath-tcp/mptcp_net-next/export/20211119T160804
https://github.com/multipath-tcp/mptcp_net-next/actions/workflows/build-validation.yml?query=branch:export

Cheers,
Matt
-- 
Tessares | Belgium | Hybrid Access Solutions
www.tessares.net

Re: [PATCH v3 mptcp-next 0/4] TCP_INQ support

[Matthieu Baerts]

On 19/11/2021 17:09, Matthieu Baerts wrote:
> Hi Florian, Mat,
> 
> On 16/11/2021 14:48, Florian Westphal wrote:
>> No major changes since v2, see individual patches for details.
>>
>> This adds TCP_INQ for mptcp and extends the selftest infra.
>> Patch 3 adds ioctls to retrieve amount of in-sequence bytes
>> ready to for read() (older version of TCP_INQ) and counters for
>> unsent and unacked byte count.
>>
>> Last patch is new in v2, its a more specific TCP_INQ test case.
>> See individual patches for changes since v1.
> 
> Thank you for the patches, the reviews and the reply to my question
> about the ChangeLog!
> 
> Now in our tree (features for next) with Mat's RvB tag (and without some
> warnings from checkpatch in the .c files of the selftests, nothing
> important)

and without "int queued, on = 1;" line in the selftests.

Cheers,
Matt
-- 
Tessares | Belgium | Hybrid Access Solutions
www.tessares.net