TCP splice with splice() experimentation

common.h

/*  */
#define ERRORF(x...) fprintf(stderr, x)

#define FATAL(x...)                                                            \
	do {                                                                   \
		ERRORF("[-] PROGRAM ABORT : " x);                              \
		ERRORF("\n\tLocation : %s(), %s:%u\n\n", __FUNCTION__,         \
		       __FILE__, __LINE__);                                    \
		exit(EXIT_FAILURE);                                            \
	} while (0)

#define PFATAL(x...)                                                           \
	do {                                                                   \
		ERRORF("[-] SYSTEM ERROR : " x);                               \
		ERRORF("\n\tLocation : %s(), %s:%u\n", __FUNCTION__, __FILE__, \
		       __LINE__);                                              \
		perror("      OS message ");                                   \
		ERRORF("\n");                                                  \
		exit(EXIT_FAILURE);                                            \
	} while (0)

#define MIN(a, b) ((a) < (b) ? (a) : (b))

/* net.c */
int net_parse_sockaddr(struct sockaddr_storage *ss, const char *addr);
int net_connect_tcp_blocking(struct sockaddr_storage *sas);
int net_getpeername(int sd, struct sockaddr_storage *ss);
const char *net_ntop(struct sockaddr_storage *ss);
int net_bind_tcp(struct sockaddr_storage *ss);
int net_accept(int sd, struct sockaddr_storage *ss);

/* inlines */
#define TIMESPEC_NSEC(ts) ((ts)->tv_sec * 1000000000ULL + (ts)->tv_nsec)
inline static uint64_t realtime_now()
{
	struct timespec now_ts;
	clock_gettime(CLOCK_MONOTONIC, &now_ts);
	return TIMESPEC_NSEC(&now_ts);
}

Makefile

all:
	clang -g -O2 -Wall -Wextra net.c send.c -o send
	clang -g -O2 -Wall -Wextra net.c receive.c -o receive
	clang -g -O2 -Wall -Wextra net.c proxy-naive.c -o proxy-naive
	clang -g -O2 -Wall -Wextra net.c proxy-splice.c -o proxy-splice

proxy-naive.c

#include <arpa/inet.h>
#include <errno.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

int main(int argc, char **argv)
{
	if (argc < 3) {
		FATAL("Usage: %s <listen:port> <target:port>", argv[0]);
	}

	struct sockaddr_storage listen, target;
	net_parse_sockaddr(&listen, argv[1]);
	net_parse_sockaddr(&target, argv[2]);

	fprintf(stderr, "[+] Accepting on %s, ", net_ntop(&listen));
	fprintf(stderr, "transmitting to %s\n", net_ntop(&target));

	int sd = net_bind_tcp(&listen);
	if (sd < 0) {
		PFATAL("connect()");
	}

again_accept:;
	struct sockaddr_storage client;
	int cd = net_accept(sd, &client);
	shutdown(cd, SHUT_WR);
	uint64_t t0 = realtime_now();

	int td = net_connect_tcp_blocking(&target);
	if (td < 0) {
		PFATAL("connect()");
	}
	shutdown(td, SHUT_RD);

	char buf[512 * 1024];

	uint64_t sum = 0;
	while (1) {
		int n = recv(cd, buf, sizeof(buf), 0);
		if (n < 0) {
			if (errno == EINTR) {
				continue;
			}
			if (errno == ECONNRESET) {
				fprintf(stderr, "[!] ECONNRESET\n");
				break;
			}
			PFATAL("read()");
		}

		if (n == 0) {
			/* On TCP socket zero means EOF */
			fprintf(stderr, "[-] edge side EOF\n");
			break;
		}

		sum += n;

		int t = 0;
		while (t < n) {
			/* TODO: Is this needed? Won't send() block till all
			 * data is sent? */
			int m = send(td, &buf[t], n - t, MSG_NOSIGNAL);
			if (m < 0) {
				if (errno == EINTR) {
					continue;
				}
				if (errno == ECONNRESET) {
					fprintf(stderr,
						"[!] ECONNRESET on origin\n");
					break;
				}
				if (errno == EPIPE) {
					fprintf(stderr,
						"[!] EPIPE on origin\n");
					break;
				}
				PFATAL("send()");
			}
			if (m == 0) {
				int err;
				socklen_t err_len = sizeof(err);
				int r = getsockopt(td, SOL_SOCKET, SO_ERROR,
						   &err, &err_len);
				if (r < 0) {
					PFATAL("getsockopt()");
				}
				errno = err;
				PFATAL("send()");
			}
			t += m;
		}
	}
	shutdown(td, SHUT_WR);
	shutdown(cd, SHUT_RD);
	close(cd);
	close(td);
	uint64_t t1 = realtime_now();

	fprintf(stderr, "[+] Read %.1fMiB in %.1fms\n", sum / (1024 * 1024.),
		(t1 - t0) / 1000000.);
	goto again_accept;

	return 0;
}

proxy-splice.c

#define _GNU_SOURCE /* splice */

#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

int main(int argc, char **argv)
{
	if (argc < 3) {
		FATAL("Usage: %s <listen:port> <target:port>", argv[0]);
	}

	struct sockaddr_storage listen, target;
	net_parse_sockaddr(&listen, argv[1]);
	net_parse_sockaddr(&target, argv[2]);

	fprintf(stderr, "[+] Accepting on %s, ", net_ntop(&listen));
	fprintf(stderr, "transmitting to %s\n", net_ntop(&target));

	int sd = net_bind_tcp(&listen);
	if (sd < 0) {
		PFATAL("connect()");
	}

again_accept:;
	struct sockaddr_storage client;
	int cd = net_accept(sd, &client);
	shutdown(cd, SHUT_WR);
	uint64_t t0 = realtime_now();

	int td = net_connect_tcp_blocking(&target);
	if (td < 0) {
		PFATAL("connect()");
	}
	shutdown(td, SHUT_RD);

	int pfd[2];
	int r = pipe(pfd);
	if (r < 0) {
		PFATAL("pipe()");
	}

	r = fcntl(pfd[0], F_SETPIPE_SZ, 1 * 1024 * 1024);
	if (r < 0) {
		PFATAL("fcntl()");
	}

#define SPLICE_MAX (1024 * 1024)
	uint64_t sum = 0;
	while (1) {
		int n = splice(cd, NULL, pfd[1], NULL, SPLICE_MAX,
			       SPLICE_F_MOVE);
		if (n < 0) {
			if (errno == EINTR) {
				continue;
			}
			if (errno == ECONNRESET) {
				fprintf(stderr, "[!] ECONNRESET\n");
				break;
			}
			if (errno == EAGAIN) {
				break;
			}
			PFATAL("splice()");
		}
		if (n == 0) {
			/* On TCP socket zero means EOF */
			fprintf(stderr, "[-] edge side EOF\n");
			break;
		}

		sum += n;

		int m = splice(pfd[0], NULL, td, NULL, n, SPLICE_F_MOVE);
		if (m < 0) {
			if (errno == EINTR) {
				continue;
			}
			if (errno == ECONNRESET) {
				fprintf(stderr, "[!] ECONNRESET on origin\n");
				break;
			}
			if (errno == EPIPE) {
				fprintf(stderr, "[!] EPIPE on origin\n");
				break;
			}
			PFATAL("send()");
		}
		if (m == 0) {
			int err;
			socklen_t err_len = sizeof(err);
			int r = getsockopt(td, SOL_SOCKET, SO_ERROR, &err,
					   &err_len);
			if (r < 0) {
				PFATAL("getsockopt()");
			}
			errno = err;
			PFATAL("send()");
		}
		if (m != n) {
			FATAL("");
		}
	}
	shutdown(td, SHUT_WR);
	shutdown(cd, SHUT_RD);
	close(cd);
	close(td);
	uint64_t t1 = realtime_now();

	fprintf(stderr, "[+] Read %.1fMiB in %.1fms\n", sum / (1024 * 1024.),
		(t1 - t0) / 1000000.);
	goto again_accept;

	return 0;
}

receive.c

#include <arpa/inet.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

int main(int argc, char **argv)
{
	if (argc < 2) {
		FATAL("Usage: %s <listen:port>", argv[0]);
	}

	struct sockaddr_storage listen;
	net_parse_sockaddr(&listen, argv[1]);

	uint64_t transmission_sz = 1 * 1024 * 1024; // 1MiB
	if (argc > 2) {
		char *endptr;
		long ts = strtol(argv[2], &endptr, 10);
		if (ts < 0 || *endptr != '\0') {
			FATAL("Can't parse number %s", argv[2]);
		}
		transmission_sz = ts * 1024 * 1024ULL; // In MiB
	}

	fprintf(stderr, "[+] Accepting on %s\n", net_ntop(&listen));

	int sd = net_bind_tcp(&listen);
	if (sd < 0) {
		PFATAL("connect()");
	}

again_accept:;
	struct sockaddr_storage client;
	int cd = net_accept(sd, &client);
	shutdown(cd, SHUT_WR);
	uint64_t t0 = realtime_now();

	uint64_t sum = 0;
	while (1) {
		/* We don't actually need to copy the data to
		 * userspace. Let's just discard it with MSG_TRUNC. */
		int n = recv(cd, NULL, -1, MSG_TRUNC);
		if (n < 0 && errno == EINTR) {
			continue;
		}

		if (n < 0) {
			if (errno == ECONNRESET) {
				fprintf(stderr, "[!] ECONNRESET\n");
				break;
			}
			PFATAL("read()");
		}

		if (n == 0) {
			/* On TCP socket zero means EOF */
			break;
		}
		sum += n;
	}
	shutdown(cd, SHUT_RD);
	close(cd);
	uint64_t t1 = realtime_now();

	fprintf(stderr, "[+] Read %.1fMiB in %.1fms\n", sum / (1024 * 1024.),
		(t1 - t0) / 1000000.);
	goto again_accept;

	return 0;
}

send.c

#include <arpa/inet.h>
#include <errno.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

int main(int argc, char **argv)
{
	if (argc < 2) {
		FATAL("Usage: %s <target:port> <amount in MiB>", argv[0]);
	}

	struct sockaddr_storage target;
	net_parse_sockaddr(&target, argv[1]);

	uint64_t transmission_sz = 1 * 1024 * 1024; // 1MiB
	if (argc > 2) {
		char *endptr;
		long ts = strtol(argv[2], &endptr, 10);
		if (ts < 0 || *endptr != '\0') {
			FATAL("Can't parse number %s", argv[2]);
		}
		transmission_sz = ts * 1024 * 1024ULL; // In MiB
	}

	fprintf(stderr, "[+] Sending %.1fMiB to %s\n",
		transmission_sz / (1024 * 1024.), net_ntop(&target));

	int fd = net_connect_tcp_blocking(&target);
	if (fd < 0) {
		PFATAL("connect()");
	}

	shutdown(fd, SHUT_RD);

	char buf[128 * 1024];
	memset(buf, -1, sizeof(buf));

	uint64_t t0 = realtime_now();
	uint64_t sum = 0;
	while (sum < transmission_sz) {
		int n = send(fd, buf, MIN(sizeof(buf), transmission_sz - sum),
			     MSG_NOSIGNAL);
		if (n < 0) {
			if (errno == EINTR) {
				continue;
			}
			if (errno == ECONNRESET) {
				fprintf(stderr, "[!] ECONNRESET\n");
				break;
			}
			if (errno == EPIPE) {
				fprintf(stderr, "[!] EPIPE\n");
				break;
			}
			PFATAL("send()");
		}

		if (n == 0) {
			int err;
			socklen_t err_len = sizeof(err);
			int r = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err,
					   &err_len);
			if (r < 0) {
				PFATAL("getsockopt()");
			}
			errno = err;
			PFATAL("send()");
		}
		sum += n;
	}
	shutdown(fd, SHUT_WR);
	close(fd);
	uint64_t t1 = realtime_now();

	fprintf(stderr, "[+] Wrote %.1fMiB in %.1fms\n", sum / (1024 * 1024.),
		(t1 - t0) / 1000000.);
	return 0;
}

It seems that net.c was absent.

mhm, maybe this can help https://github.com/cloudflare/cloudflare-blog/tree/master/2019-02-tcp-splice

mhm, maybe this can help https://github.com/cloudflare/cloudflare-blog/tree/master/2019-02-tcp-splice

Thanks for the informative reply. It is very useful to me.