/*
 * DF-5848 -- deterministic trigger for the kernel-nfsd m_dup_data
 * MT_DATA assert (sys/kern/uipc_mbuf.c:1806 via nfs_realign).
 *
 * Mechanism being exercised:
 *  - tcp_output() builds each segment in an m_gethdr(..., MT_HEADER) mbuf
 *    (sys/netinet/tcp_output.c:898-942); when the segment payload fits
 *    inline (len <= MHLEN - hdrlen - max_linkhdr) the whole segment is
 *    ONE MT_HEADER mbuf.
 *  - over the loopback interface the SAME mbuf (header-stripped by m_adj)
 *    is appended to the server socket's so_rcv (if_simloop does not copy),
 *    so the nfsd's soreceive returns MT_HEADER data mbufs
 *    (uipc_socket.c:1473/1779 explicitly bless MT_HEADER for stream data).
 *  - a SunRPC record whose on-wire size is NOT a multiple of 4 makes the
 *    first mbuf of the completed record fail nfs_realign's alignment test
 *    (m_len & 3), and nfs_realign() -> m_dup_data() then trips
 *    KKASSERT(m->m_type == MT_DATA) on the MT_HEADER mbuf.
 *
 * So: open TCP to the kernel nfsd (2049), set TCP_NODELAY so each small
 * record is sent as its own all-in-one MT_HEADER segment, and stream
 * LASTFRAG records whose total size % 4 != 0.  Pipelined bursts are mixed
 * in to also exercise the multi-record-per-segment split path in
 * nfsrv_getstream().
 *
 * The RPC payload itself is irrelevant XDR garbage -- nfs_realign runs in
 * nfsrv_getstream() BEFORE any RPC parsing, so even GARBAGE records hit it.
 * Replies are drained opportunistically to keep the connection healthy.
 *
 * build:  cc -O2 -o df5848_pipe df5848_pipe.c
 * run:    ./df5848_pipe <seconds> [host]   (root: binds reserved src port
 *                                          because vfs.nfs.reserved_port_only)
 */
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>

static double
now(void)
{
	struct timespec ts;
	clock_gettime(CLOCK_MONOTONIC, &ts);
	return (ts.tv_sec + ts.tv_nsec * 1e-9);
}

int
main(int argc, char **argv)
{
	double dur = argc > 1 ? atof(argv[1]) : 60.0;
	const char *host = argc > 2 ? argv[2] : "127.0.0.1";
	struct sockaddr_in dst, src;
	unsigned char buf[4096];
	unsigned long nrec = 0, nburst = 0, ndrain = 0;
	double t0, last;
	int s, one = 1, i, burst;

	s = socket(AF_INET, SOCK_STREAM, 0);
	if (s < 0) {
		perror("socket");
		return 1;
	}

	/*
	 * Kernel nfsd with vfs.nfs.reserved_port_only only services
	 * clients from reserved ports; bind one (needs root, which the
	 * run script has).
	 */
	memset(&src, 0, sizeof(src));
	src.sin_family = AF_INET;
	src.sin_port = htons(791);
	src.sin_addr.s_addr = htonl(INADDR_ANY);
	(void)bind(s, (struct sockaddr *)&src, sizeof(src));

	memset(&dst, 0, sizeof(dst));
	dst.sin_family = AF_INET;
	dst.sin_port = htons(2049);
	if (inet_pton(AF_INET, host, &dst.sin_addr) != 1) {
		fprintf(stderr, "inet_pton %s\n", host);
		return 1;
	}
	if (connect(s, (struct sockaddr *)&dst, sizeof(dst)) < 0) {
		perror("connect 2049 (is the kernel nfsd listening?)");
		return 1;
	}
	setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
	fcntl(s, F_SETFL, O_NONBLOCK);

	printf("df5848_pipe: connected to %s:2049 (src port %d), "
	    "streaming misaligned LASTFRAG records for %.0fs\n",
	    host, ntohs(src.sin_port), dur);
	fflush(stdout);

	t0 = now();
	last = t0;
	i = 0;
	while (now() - t0 < dur) {
		/*
		 * Record sizes 4+138..4+145: totals 142..149 bytes, so the
		 * segment payload (after IP+TCP strip) has m_len % 4 != 0
		 * for most picks; all fit inline in one MT_HEADER mbuf.
		 */
		int paylen = 138 + (i % 8);
		int reclen = 4 + paylen;

		*(uint32_t *)buf = htonl(0x80000000u | paylen);
		memset(buf + 4, 0, paylen);

		if (write(s, buf, reclen) < 0 && errno != EAGAIN) {
			perror("write");
			break;
		}
		nrec++;

		/*
		 * Every 8th record: pipelined burst of 16 records in one
		 * write (multi-record segment -> nfsrv_getstream split path).
		 */
		if ((++i % 8) == 0) {
			int j, k;
			burst = 0;
			for (j = 0; j < 16; j++) {
				int pl = 130 + ((i + j) % 11);
				*(uint32_t *)(buf + burst) =
				    htonl(0x80000000u | pl);
				memset(buf + burst + 4, 0, pl);
				burst += 4 + pl;
			}
			if (write(s, buf, burst) < 0 && errno != EAGAIN) {
				perror("write burst");
				break;
			}
			nburst += 16;
		}

		/* drain replies so the connection keeps flowing */
		for (;;) {
			ssize_t r = read(s, buf, sizeof(buf));
			if (r > 0) {
				ndrain++;
				continue;
			}
			break;
		}

		if (now() - last >= 5.0) {
			printf("df5848_pipe: %lus elapsed, %lu single records, "
			    "%lu burst records, %lu reply reads\n",
			    (unsigned long)(now() - t0), nrec, nburst, ndrain);
			fflush(stdout);
			last = now();
		}
		usleep(2000);	/* let each small record go out alone */
	}
	printf("df5848_pipe: DONE single=%lu burst=%lu drained=%lu\n",
	    nrec, nburst, ndrain);
	return 0;
}