fix ENOBUFS errors with backoff+retry

.gitignore

@@ -15,6 +15,7 @@ venv/
 # C build artifacts
 *.o
 pixelflut-render
+pixelflut-bench
 render
 a.out
 

Makefile

@@ -1,18 +1,32 @@
-CC      ?= cc
-CFLAGS  ?= -std=c11 -O2 -pthread \
-           -Wall -Wextra -Wpedantic -Wshadow -Wformat=2 \
-           -Wcast-align -Wstrict-prototypes -Wmissing-prototypes \
-           -Wpointer-arith -Wwrite-strings
-LDFLAGS ?= -pthread
+# Portable across GNU make (Linux) and BSD make (FreeBSD).
+#
+# We do NOT read the make flavor's predefined CFLAGS: BSD make's sys.mk sets it
+# to "-O2 -pipe", which would fight our OPT setting. Optimization lives in OPT,
+# required language/warning flags in STD/WARN, and EXTRA is yours to append.
+#
+# OPT defaults to an aggressive, build-on-target profile. -march=native tunes
+# for THIS machine's CPU, so only build where you run (which is how this repo
+# is used). If you cross-build, override it, e.g.  make OPT="-O2".
+CC    ?= cc
+OPT   ?= -O3 -march=native
+EXTRA ?=
 
-TARGET  := pixelflut-render
-SRC     := new.c
+STD  = -std=c11
+WARN = -Wall -Wextra -Wpedantic -Wshadow -Wformat=2 \
+       -Wcast-align -Wstrict-prototypes -Wmissing-prototypes \
+       -Wpointer-arith -Wwrite-strings
 
-$(TARGET): $(SRC)
-	$(CC) $(CFLAGS) -o $@ $(SRC) $(LDFLAGS)
+TARGETS = pixelflut-render pixelflut-bench
+
+all: $(TARGETS)
+
+pixelflut-render: new.c pixelflut.h
+	$(CC) $(STD) $(OPT) -pthread $(WARN) $(EXTRA) -o $@ new.c
+
+pixelflut-bench: bench.c pixelflut.h
+	$(CC) $(STD) $(OPT) -pthread $(WARN) $(EXTRA) -o $@ bench.c
 
 clean:
-	rm -f $(TARGET)
-
-.PHONY: clean
+	rm -f $(TARGETS)
 
+.PHONY: all clean

bench.c

@@ -0,0 +1,546 @@
+/*
+ * pixelflut-bench - UDP send-throughput benchmark for the Pixelflut renderer.
+ *
+ * It measures how fast we can actually push Pixelflut datagrams out of the
+ * kernel, which is the number we care about: "it is a waste to not be sending
+ * packets". Each sender thread owns its own socket and a pre-built ring of
+ * packets, then loops on sendmmsg() as hard as it can. A monitor prints the
+ * achieved packet rate and payload throughput once per second.
+ *
+ * By default it runs a self-contained loopback test: an in-process sink binds
+ * the target port and drains packets so the senders never block, letting us
+ * measure the pure generate+syscall ceiling of this machine. Point it at a
+ * real server with -H to benchmark an actual link.
+ *
+ *   ./pixelflut-bench                 # loopback self-test, 1 thread, 5 s
+ *   ./pixelflut-bench -t 8 -d 10      # 8 sender threads for 10 s
+ *   ./pixelflut-bench -H fcda:f100::d -p 5005 -t 8   # hammer a real display
+ */
+
+#define _GNU_SOURCE /* sendmmsg(), recvmmsg(), struct mmsghdr */
+
+#include <errno.h>
+#include <inttypes.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+
+#include <pthread.h>
+
+#include "pixelflut.h"
+
+/* Portable host -> little-endian 16-bit conversion (matches the wire format). */
+#if defined(__linux__)
+#  include <endian.h>
+#elif defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__)
+#  include <sys/endian.h>
+#elif defined(__APPLE__)
+#  include <libkern/OSByteOrder.h>
+#  define htole16(v) OSSwapHostToLittleInt16(v)
+#endif
+
+/* Frame geometry mirrored from the renderer, so -g reproduces its real
+ * per-pixel generation cost (xorshf96 + divisions + frame lookups). */
+#define FRAME_WIDTH     640
+#define FRAME_HEIGHT    360
+#define FRAME_CHANNELS  3
+#define FRAME_PIXELS    (FRAME_WIDTH * FRAME_HEIGHT)
+#define FRAME_BYTES     (FRAME_PIXELS * FRAME_CHANNELS)
+#define CANVAS_OFFSET_X 1200
+#define CANVAS_OFFSET_Y 40
+
+#define PAYLOAD_BYTES ((uint64_t)sizeof(struct pixelflut_packet))
+
+/* Rough per-packet wire overhead (IPv6 + UDP headers) for a link estimate. */
+#define WIRE_OVERHEAD_BYTES 48
+
+/* sendmmsg()/recvmmsg() accept at most this many messages per call. */
+#define MAX_BATCH 1024
+
+/* Per-thread counters, each on its own cache line to avoid false sharing. */
+typedef struct {
+	_Alignas(64) _Atomic uint64_t packets;
+	_Atomic uint64_t bytes;
+	_Atomic uint64_t calls;
+	_Atomic uint64_t errors;
+} counters_t;
+
+struct config {
+	struct sockaddr_storage addr;
+	socklen_t addrlen;
+	int family;
+	int batch;          /* messages per sendmmsg() call and ring size */
+	int sndbuf;         /* SO_SNDBUF, 0 = leave kernel default */
+	int rcvbuf;         /* SO_RCVBUF for the sink */
+	int port;
+	bool generate;      /* regenerate pixel payload each pass (like the renderer) */
+	bool drain;         /* sink actively reads; if false it only holds the port */
+	const uint8_t *frame; /* shared read-only source frame for -g */
+};
+
+struct sender_arg {
+	const struct config *cfg;
+	counters_t *ctr;
+	int index;
+};
+
+static _Atomic bool g_running = true;
+
+static double now_sec(void)
+{
+	struct timespec ts;
+	clock_gettime(CLOCK_MONOTONIC, &ts);
+	return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
+}
+
+static void set_bufsize(int fd, int optname, int size, const char *what)
+{
+	if (size > 0 && setsockopt(fd, SOL_SOCKET, optname, &size, sizeof size) == -1)
+		fprintf(stderr, "warning: setsockopt(%s, %d) failed: %s\n", what,
+		        size, strerror(errno));
+}
+
+/* Fill a ring of packets with a valid header and arbitrary pixel data. */
+static void build_packets(struct pixelflut_packet *pkts, int n)
+{
+	for (int i = 0; i < n; i++) {
+		pkts[i].cmd = PIXELFLUT_CMD_SET;
+		pkts[i].subcmd = PIXELFLUT_SUBCMD_RGB;
+		for (int j = 0; j < PIXELS_PER_PACKET; j++) {
+			struct pixelflut_pixel *px = &pkts[i].pixels[j];
+			px->x = (uint16_t)(i + j);
+			px->y = (uint16_t)(i - j);
+			px->r = (uint8_t)i;
+			px->g = (uint8_t)j;
+			px->b = 0x80;
+		}
+	}
+}
+
+/* xorshf96 PRNG, per-thread state (same generator the renderer uses). */
+static unsigned long xorshf96(unsigned long *x, unsigned long *y, unsigned long *z)
+{
+	*x ^= *x << 16;
+	*x ^= *x >> 5;
+	*x ^= *x << 1;
+	unsigned long t = *x;
+	*x = *y;
+	*y = *z;
+	*z = t ^ *x ^ *y;
+	return *z;
+}
+
+/* Repaint a ring of packets from random frame pixels, exactly like the
+ * renderer's hot loop, to measure generate+send throughput under -g. */
+static void generate_ring(struct pixelflut_packet *pkts, int n,
+                          const uint8_t *frame, unsigned long *x,
+                          unsigned long *y, unsigned long *z)
+{
+	for (int i = 0; i < n; i++) {
+		for (int j = 0; j < PIXELS_PER_PACKET; j++) {
+			uint64_t src = xorshf96(x, y, z) % FRAME_PIXELS;
+			uint16_t fx = (uint16_t)(src % FRAME_WIDTH);
+			uint16_t fy = (uint16_t)(src / FRAME_WIDTH);
+			const uint8_t *rgb = &frame[src * FRAME_CHANNELS];
+
+			struct pixelflut_pixel *px = &pkts[i].pixels[j];
+			px->x = htole16((uint16_t)(CANVAS_OFFSET_X + fx));
+			px->y = htole16((uint16_t)(CANVAS_OFFSET_Y + fy));
+			px->r = rgb[0];
+			px->g = rgb[1];
+			px->b = rgb[2];
+		}
+	}
+}
+
+static void *sender_main(void *arg)
+{
+	const struct sender_arg *sa = arg;
+	const struct config *cfg = sa->cfg;
+	counters_t *ctr = sa->ctr;
+	const int n = cfg->batch;
+
+	/* Distinct PRNG seeds per thread so they paint different pixels. */
+	unsigned long rng_x = 123456789UL ^ (2654435761UL * (unsigned long)(sa->index + 1));
+	unsigned long rng_y = 362436069UL + (unsigned long)(sa->index + 1);
+	unsigned long rng_z = 521288629UL;
+
+	int fd = socket(cfg->family, SOCK_DGRAM, 0);
+	if (fd == -1) {
+		perror("socket");
+		return NULL;
+	}
+	set_bufsize(fd, SO_SNDBUF, cfg->sndbuf, "SO_SNDBUF");
+	if (connect(fd, (const struct sockaddr *)&cfg->addr, cfg->addrlen) == -1) {
+		perror("connect");
+		close(fd);
+		return NULL;
+	}
+
+	struct pixelflut_packet *pkts = malloc((size_t)n * sizeof *pkts);
+	struct mmsghdr *msgs = calloc((size_t)n, sizeof *msgs);
+	struct iovec *iov = calloc((size_t)n, sizeof *iov);
+	if (pkts == NULL || msgs == NULL || iov == NULL) {
+		fprintf(stderr, "sender: out of memory\n");
+		goto out;
+	}
+	build_packets(pkts, n);
+	for (int i = 0; i < n; i++) {
+		iov[i].iov_base = &pkts[i];
+		iov[i].iov_len = sizeof pkts[i];
+		msgs[i].msg_hdr.msg_iov = &iov[i];
+		msgs[i].msg_hdr.msg_iovlen = 1;
+	}
+
+	uint64_t packets = 0, bytes = 0, calls = 0, errors = 0;
+	int next = 0;
+	while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
+		/* Repaint the whole ring before each pass, like the renderer. */
+		if (cfg->generate && next == 0)
+			generate_ring(pkts, n, cfg->frame, &rng_x, &rng_y, &rng_z);
+
+		int sent = sendmmsg(fd, &msgs[next], (unsigned int)(n - next), 0);
+		if (sent < 0) {
+			if (errno == EINTR)
+				continue;
+			errors++;
+			next = 0;
+			continue;
+		}
+		packets += (uint64_t)sent;
+		bytes += (uint64_t)sent * PAYLOAD_BYTES;
+		calls++;
+		next += sent;
+		if (next >= n)
+			next = 0;
+
+		/* Publish cumulative counts for the monitor (cheap, per call). */
+		atomic_store_explicit(&ctr->packets, packets, memory_order_relaxed);
+		atomic_store_explicit(&ctr->bytes, bytes, memory_order_relaxed);
+		atomic_store_explicit(&ctr->calls, calls, memory_order_relaxed);
+		atomic_store_explicit(&ctr->errors, errors, memory_order_relaxed);
+	}
+
+out:
+	free(pkts);
+	free(msgs);
+	free(iov);
+	close(fd);
+	return NULL;
+}
+
+/* Loopback sink: bind the target port and drain packets so senders never
+ * block on a full socket buffer. */
+static void *sink_main(void *arg)
+{
+	const struct sender_arg *sa = arg;
+	const struct config *cfg = sa->cfg;
+	counters_t *ctr = sa->ctr;
+	const int n = (cfg->batch < MAX_BATCH) ? cfg->batch : MAX_BATCH;
+
+	int fd = socket(cfg->family, SOCK_DGRAM, 0);
+	if (fd == -1) {
+		perror("sink socket");
+		return NULL;
+	}
+	int one = 1;
+	setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
+	set_bufsize(fd, SO_RCVBUF, cfg->rcvbuf, "SO_RCVBUF");
+	/* Wake up periodically so we can notice g_running going false. */
+	struct timeval tv = {.tv_sec = 0, .tv_usec = 200000};
+	setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
+	if (bind(fd, (const struct sockaddr *)&cfg->addr, cfg->addrlen) == -1) {
+		perror("sink bind");
+		close(fd);
+		return NULL;
+	}
+
+	if (!cfg->drain) {
+		/* Send-ceiling mode: keep the port bound (so senders get no ICMP
+		 * port-unreachable) but never read. The kernel drops datagrams at
+		 * the full receive buffer, and this thread stays idle so every CPU
+		 * is available to the senders. */
+		while (atomic_load_explicit(&g_running, memory_order_relaxed))
+			usleep(50000);
+		close(fd);
+		return NULL;
+	}
+
+	const size_t slot = 2048;
+	uint8_t *bufs = malloc((size_t)n * slot);
+	struct mmsghdr *msgs = calloc((size_t)n, sizeof *msgs);
+	struct iovec *iov = calloc((size_t)n, sizeof *iov);
+	if (bufs == NULL || msgs == NULL || iov == NULL) {
+		fprintf(stderr, "sink: out of memory\n");
+		goto out;
+	}
+	for (int i = 0; i < n; i++) {
+		iov[i].iov_base = bufs + (size_t)i * slot;
+		iov[i].iov_len = slot;
+		msgs[i].msg_hdr.msg_iov = &iov[i];
+		msgs[i].msg_hdr.msg_iovlen = 1;
+	}
+
+	uint64_t packets = 0, bytes = 0;
+	while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
+		int got = recvmmsg(fd, msgs, (unsigned int)n, 0, NULL);
+		if (got < 0) {
+			if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
+				continue;
+			perror("recvmmsg");
+			break;
+		}
+		packets += (uint64_t)got;
+		for (int i = 0; i < got; i++)
+			bytes += msgs[i].msg_len;
+		atomic_store_explicit(&ctr->packets, packets, memory_order_relaxed);
+		atomic_store_explicit(&ctr->bytes, bytes, memory_order_relaxed);
+	}
+
+out:
+	free(bufs);
+	free(msgs);
+	free(iov);
+	close(fd);
+	return NULL;
+}
+
+static uint64_t sum_packets(const counters_t *c, int n)
+{
+	uint64_t t = 0;
+	for (int i = 0; i < n; i++)
+		t += atomic_load_explicit(&c[i].packets, memory_order_relaxed);
+	return t;
+}
+
+static uint64_t sum_calls(const counters_t *c, int n)
+{
+	uint64_t t = 0;
+	for (int i = 0; i < n; i++)
+		t += atomic_load_explicit(&c[i].calls, memory_order_relaxed);
+	return t;
+}
+
+static uint64_t sum_errors(const counters_t *c, int n)
+{
+	uint64_t t = 0;
+	for (int i = 0; i < n; i++)
+		t += atomic_load_explicit(&c[i].errors, memory_order_relaxed);
+	return t;
+}
+
+static void print_rate(const char *label, uint64_t packets, double secs)
+{
+	double pps = (double)packets / secs;
+	double gbit = (double)packets * (double)PAYLOAD_BYTES * 8.0 / secs / 1e9;
+	double wire = (double)packets * (double)(PAYLOAD_BYTES + WIRE_OVERHEAD_BYTES) *
+	              8.0 / secs / 1e9;
+	printf("%-8s %8.3f Mpps  %8.3f Gbit/s payload  %8.3f Gbit/s wire\n", label,
+	       pps / 1e6, gbit, wire);
+}
+
+static int resolve(const char *host, int port, bool passive, struct config *cfg)
+{
+	char portstr[16];
+	snprintf(portstr, sizeof portstr, "%d", port);
+	struct addrinfo hints;
+	memset(&hints, 0, sizeof hints);
+	hints.ai_family = AF_UNSPEC;
+	hints.ai_socktype = SOCK_DGRAM;
+	hints.ai_flags = passive ? AI_PASSIVE : 0;
+
+	struct addrinfo *res = NULL;
+	int rc = getaddrinfo(host, portstr, &hints, &res);
+	if (rc != 0) {
+		fprintf(stderr, "getaddrinfo(%s): %s\n", host ? host : "*",
+		        gai_strerror(rc));
+		return -1;
+	}
+	memcpy(&cfg->addr, res->ai_addr, res->ai_addrlen);
+	cfg->addrlen = res->ai_addrlen;
+	cfg->family = res->ai_family;
+	freeaddrinfo(res);
+	return 0;
+}
+
+static void usage(const char *prog)
+{
+	fprintf(stderr,
+	        "usage: %s [-H host] [-p port] [-t threads] [-d secs] "
+	        "[-s batch] [-B sndbuf] [-R rcvbuf] [-g]\n"
+	        "  -H host    target host (default: loopback self-test with sink)\n"
+	        "  -p port    target UDP port (default 5005)\n"
+	        "  -t threads sender threads (default 1)\n"
+	        "  -d secs    measurement duration (default 5)\n"
+	        "  -s batch   messages per sendmmsg() call, 1..%d (default %d)\n"
+	        "  -B bytes   SO_SNDBUF per sender socket (default: kernel)\n"
+	        "  -R bytes   SO_RCVBUF for the loopback sink (default 4 MiB)\n"
+	        "  -g         regenerate pixel payload each pass (renderer workload)\n"
+	        "  -N         send-ceiling mode: sink holds the port but does not\n"
+	        "             drain, freeing a core (recv counts are meaningless)\n",
+	        prog, MAX_BATCH, MAX_BATCH);
+}
+
+int main(int argc, char *argv[])
+{
+	const char *host = NULL;
+	int port = 5005;
+	int threads = 1;
+	int duration = 5;
+	struct config cfg = {
+	    .batch = MAX_BATCH, .sndbuf = 0, .rcvbuf = 4 << 20, .drain = true};
+
+	int opt;
+	while ((opt = getopt(argc, argv, "H:p:t:d:s:B:R:gNh")) != -1) {
+		switch (opt) {
+		case 'H': host = optarg; break;
+		case 'p': port = atoi(optarg); break;
+		case 't': threads = atoi(optarg); break;
+		case 'd': duration = atoi(optarg); break;
+		case 's': cfg.batch = atoi(optarg); break;
+		case 'B': cfg.sndbuf = atoi(optarg); break;
+		case 'R': cfg.rcvbuf = atoi(optarg); break;
+		case 'g': cfg.generate = true; break;
+		case 'N': cfg.drain = false; break;
+		case 'h': usage(argv[0]); return EXIT_SUCCESS;
+		default: usage(argv[0]); return EXIT_FAILURE;
+		}
+	}
+	if (threads < 1 || duration < 1 || cfg.batch < 1 || cfg.batch > MAX_BATCH) {
+		usage(argv[0]);
+		return EXIT_FAILURE;
+	}
+	cfg.port = port;
+
+	bool loopback = (host == NULL);
+	const char *target = loopback ? "::1" : host;
+	if (resolve(target, port, false, &cfg) != 0)
+		return EXIT_FAILURE;
+
+	char addrbuf[INET6_ADDRSTRLEN] = "?";
+	void *sin_addr =
+	    cfg.family == AF_INET6
+	        ? (void *)&((struct sockaddr_in6 *)&cfg.addr)->sin6_addr
+	        : (void *)&((struct sockaddr_in *)&cfg.addr)->sin_addr;
+	inet_ntop(cfg.family, sin_addr, addrbuf, sizeof addrbuf);
+
+	printf("target      %s port %d (%s)\n", addrbuf, port,
+	       loopback ? "loopback self-test" : "external");
+	printf("threads     %d\n", threads);
+	printf("batch       %d msgs/sendmmsg\n", cfg.batch);
+	printf("packet      %" PRIu64 " bytes payload (%d pixels)\n", PAYLOAD_BYTES,
+	       PIXELS_PER_PACKET);
+	printf("workload    %s\n", cfg.generate ? "generate + send" : "send only");
+	printf("duration    %d s\n\n", duration);
+
+	/* Shared read-only source frame for -g (contents are irrelevant). */
+	uint8_t *frame = NULL;
+	if (cfg.generate) {
+		frame = malloc(FRAME_BYTES);
+		if (frame == NULL) {
+			fprintf(stderr, "out of memory\n");
+			return EXIT_FAILURE;
+		}
+		memset(frame, 0x7f, FRAME_BYTES);
+		cfg.frame = frame;
+	}
+
+	/* Optional loopback sink that drains what we send. */
+	counters_t *rctr = NULL;
+	pthread_t sink_thread = 0;
+	struct sender_arg sink_arg;
+	if (loopback) {
+		rctr = aligned_alloc(64, sizeof *rctr);
+		if (rctr == NULL) {
+			fprintf(stderr, "out of memory\n");
+			return EXIT_FAILURE;
+		}
+		memset(rctr, 0, sizeof *rctr);
+		sink_arg = (struct sender_arg){.cfg = &cfg, .ctr = rctr};
+		if (pthread_create(&sink_thread, NULL, sink_main, &sink_arg) != 0) {
+			perror("pthread_create sink");
+			return EXIT_FAILURE;
+		}
+		usleep(100000); /* let the sink bind before senders connect */
+	}
+
+	counters_t *sctr = aligned_alloc(64, (size_t)threads * sizeof *sctr);
+	struct sender_arg *sargs = calloc((size_t)threads, sizeof *sargs);
+	pthread_t *tids = calloc((size_t)threads, sizeof *tids);
+	if (sctr == NULL || sargs == NULL || tids == NULL) {
+		fprintf(stderr, "out of memory\n");
+		return EXIT_FAILURE;
+	}
+	memset(sctr, 0, (size_t)threads * sizeof *sctr);
+	for (int i = 0; i < threads; i++) {
+		sargs[i] = (struct sender_arg){.cfg = &cfg, .ctr = &sctr[i], .index = i};
+		if (pthread_create(&tids[i], NULL, sender_main, &sargs[i]) != 0) {
+			perror("pthread_create sender");
+			return EXIT_FAILURE;
+		}
+	}
+
+	/* Monitor: sample once per second and print the delta. */
+	double t0 = now_sec();
+	uint64_t prev_pkts = 0;
+	double prev_t = t0;
+	for (int s = 1; s <= duration; s++) {
+		struct timespec req = {.tv_sec = 1, .tv_nsec = 0};
+		nanosleep(&req, NULL);
+		double t = now_sec();
+		uint64_t pkts = sum_packets(sctr, threads);
+		char label[16];
+		snprintf(label, sizeof label, "[%2ds]", s);
+		print_rate(label, pkts - prev_pkts, t - prev_t);
+		prev_pkts = pkts;
+		prev_t = t;
+	}
+
+	atomic_store_explicit(&g_running, false, memory_order_relaxed);
+	for (int i = 0; i < threads; i++)
+		pthread_join(tids[i], NULL);
+
+	double elapsed = now_sec() - t0;
+	uint64_t tx_pkts = sum_packets(sctr, threads);
+	uint64_t tx_calls = sum_calls(sctr, threads);
+	uint64_t tx_errs = sum_errors(sctr, threads);
+
+	printf("\n=== totals over %.2f s ===\n", elapsed);
+	print_rate("send", tx_pkts, elapsed);
+	printf("send     %" PRIu64 " packets, %" PRIu64 " sendmmsg calls "
+	       "(avg %.1f msgs/call), %" PRIu64 " errors\n",
+	       tx_pkts, tx_calls, tx_calls ? (double)tx_pkts / (double)tx_calls : 0.0,
+	       tx_errs);
+
+	if (loopback) {
+		if (cfg.drain) {
+			uint64_t rx_pkts = atomic_load(&rctr->packets);
+			print_rate("recv", rx_pkts, elapsed);
+			if (tx_pkts > 0)
+				printf("recv     %" PRIu64 " packets (%.1f%% of sent reached "
+				       "the sink)\n",
+				       rx_pkts, 100.0 * (double)rx_pkts / (double)tx_pkts);
+		} else {
+			printf("recv     discard sink (send-ceiling mode)\n");
+		}
+		pthread_join(sink_thread, NULL);
+		free(rctr);
+	}
+
+	free(sctr);
+	free(sargs);
+	free(tids);
+	free(frame);
+	return EXIT_SUCCESS;
+}

new.c

@@ -1,239 +0,0 @@
-/*
- * Pixelflut UDP renderer.
- *
- * Reads raw RGB24 frames that main.py continuously writes to INPUT_FILE
- * (via mmap) and floods them at a Pixelflut server using the binary UDP
- * protocol. Pixels are picked in a random order so the picture "dissolves"
- * onto the wall instead of being drawn scanline by scanline.
- *
- * Binary Pixelflut packet layout (little-endian on the wire):
- *
- *   +------+---------+  pixelflut_packet
- *   | cmd  | subcmd  |  2-byte header
- *   +------+---------+---------+----+----+----+  pixelflut_pixel[0]
- *   |   x (u16)      | y (u16) | r  | g  | b  |  7 bytes
- *   +----------------+---------+----+----+----+
- *   |                   ...                   |  PIXELS_PER_PACKET pixels
- *   +----------------------------------------+
- */
-
-#define _GNU_SOURCE /* sendmmsg(), struct mmsghdr, SO_BINDTODEVICE */
-
-#include <stdint.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <unistd.h>
-
-#include <arpa/inet.h>
-#include <net/if.h>
-#include <netinet/in.h>
-#include <sys/socket.h>
-#include <sys/types.h>
-
-#include <fcntl.h>
-#include <sys/mman.h>
-#include <sys/stat.h>
-
-#include <pthread.h>
-
-/* Portable host -> little-endian 16-bit conversion (matches the wire format). */
-#if defined(__linux__)
-#  include <endian.h>
-#elif defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__)
-#  include <sys/endian.h>
-#elif defined(__APPLE__)
-#  include <libkern/OSByteOrder.h>
-#  define htole16(v) OSSwapHostToLittleInt16(v)
-#endif
-
-/* ---- Frame geometry (must match the producer, see main.py). ------------- */
-#define FRAME_WIDTH    640
-#define FRAME_HEIGHT   360
-#define FRAME_CHANNELS 3 /* RGB, one byte per channel */
-#define FRAME_PIXELS   (FRAME_WIDTH * FRAME_HEIGHT)
-#define FRAME_BYTES    (FRAME_PIXELS * FRAME_CHANNELS)
-
-/* Top-left corner where the frame is drawn on the shared wall. */
-#define CANVAS_OFFSET_X 1200
-#define CANVAS_OFFSET_Y 40
-
-/* ---- Pixelflut binary protocol. ---------------------------------------- */
-#define PIXELFLUT_CMD_SET    0x00 /* "set pixels" command ... */
-#define PIXELFLUT_SUBCMD_RGB 0x01 /* ... with 3-byte RGB colours */
-#define PIXELS_PER_PACKET    120  /* pixels batched into one UDP datagram */
-
-/* Number of datagrams needed to cover a whole frame once. */
-#define PACKET_COUNT (FRAME_PIXELS / PIXELS_PER_PACKET)
-
-/* ---- Target / input. --------------------------------------------------- */
-/* #define DISPLAY_HOST "100.65.0.2" */
-#define DISPLAY_HOST "fcda:f100::d"
-#define DISPLAY_PORT 5005
-#define INPUT_FILE   "/tmp/video"
-
-/* A single pixel as it appears on the wire. */
-#pragma pack(push, 1)
-struct pixelflut_pixel {
-	uint16_t x; /* little-endian */
-	uint16_t y; /* little-endian */
-	uint8_t r;
-	uint8_t g;
-	uint8_t b;
-};
-
-/* A full datagram: header followed by a batch of pixels. */
-struct pixelflut_packet {
-	uint8_t cmd;
-	uint8_t subcmd;
-	struct pixelflut_pixel pixels[PIXELS_PER_PACKET];
-};
-#pragma pack(pop)
-
-_Static_assert(sizeof(struct pixelflut_pixel) == 7,
-               "pixel must be exactly 7 bytes on the wire");
-_Static_assert(FRAME_PIXELS % PIXELS_PER_PACKET == 0,
-               "frame pixels must divide evenly into packets");
-
-/* Pre-built datagrams and the scatter/gather metadata sendmmsg() needs. */
-static struct pixelflut_packet packets[PACKET_COUNT];
-static struct mmsghdr messages[PACKET_COUNT];
-static struct iovec iovecs[PACKET_COUNT];
-
-/* xorshf96 PRNG state (Marsaglia), period 2^96-1. */
-static unsigned long rng_x = 123456789;
-static unsigned long rng_y = 362436069;
-static unsigned long rng_z = 521288629;
-
-static unsigned long xorshf96(void)
-{
-	rng_x ^= rng_x << 16;
-	rng_x ^= rng_x >> 5;
-	rng_x ^= rng_x << 1;
-
-	unsigned long t = rng_x;
-	rng_x = rng_y;
-	rng_y = rng_z;
-	rng_z = t ^ rng_x ^ rng_y;
-
-	return rng_z;
-}
-
-/* Continuously blast the pre-built packets at the display. */
-static void *send_thread(void *arg)
-{
-	const char *ifname = arg;
-
-	int fd = socket(AF_INET6, SOCK_DGRAM, 0);
-	if (fd == -1) {
-		perror("socket");
-		exit(EXIT_FAILURE);
-	}
-
-#ifdef SO_BINDTODEVICE
-	/* Best effort: pin traffic to the given interface (needs privileges). */
-	if (ifname != NULL && ifname[0] != '\0') {
-		struct ifreq ifr;
-		memset(&ifr, 0, sizeof ifr);
-		snprintf(ifr.ifr_name, sizeof ifr.ifr_name, "%s", ifname);
-		if (setsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, &ifr,
-		               sizeof ifr) == -1) {
-			perror("setsockopt(SO_BINDTODEVICE)");
-		}
-	}
-#else
-	(void)ifname;
-#endif
-
-	struct sockaddr_in6 addr;
-	memset(&addr, 0, sizeof addr);
-	addr.sin6_family = AF_INET6;
-	addr.sin6_port = htons(DISPLAY_PORT);
-	if (inet_pton(AF_INET6, DISPLAY_HOST, &addr.sin6_addr) != 1) {
-		fprintf(stderr, "invalid display host: %s\n", DISPLAY_HOST);
-		exit(EXIT_FAILURE);
-	}
-
-	if (connect(fd, (struct sockaddr *)&addr, sizeof addr) == -1) {
-		perror("connect");
-		exit(EXIT_FAILURE);
-	}
-
-	sleep(1); /* give the producer time to fill the first frame */
-
-	int next = 0;
-	for (;;) {
-		int sent = sendmmsg(fd, &messages[next], PACKET_COUNT - next, 0);
-		if (sent == -1) {
-			perror("sendmmsg");
-			continue;
-		}
-		next += sent; /* resume after a partial send, wrap at the end */
-		if (next >= PACKET_COUNT)
-			next = 0;
-	}
-
-	return NULL; /* unreachable */
-}
-
-int main(int argc, char *argv[])
-{
-	if (argc != 2) {
-		fprintf(stderr, "usage: %s <interface>\n", argv[0]);
-		return EXIT_FAILURE;
-	}
-
-	rng_x = (unsigned long)rand();
-	rng_y = (unsigned long)rand();
-	rng_z = (unsigned long)rand();
-
-	/* Pre-fill the constant parts of every datagram once. */
-	for (int i = 0; i < PACKET_COUNT; i++) {
-		packets[i].cmd = PIXELFLUT_CMD_SET;
-		packets[i].subcmd = PIXELFLUT_SUBCMD_RGB;
-
-		iovecs[i].iov_base = &packets[i];
-		iovecs[i].iov_len = sizeof packets[i];
-		messages[i].msg_hdr.msg_iov = &iovecs[i];
-		messages[i].msg_hdr.msg_iovlen = 1;
-	}
-
-	int frame_fd = open(INPUT_FILE, O_RDONLY);
-	if (frame_fd == -1) {
-		perror("open " INPUT_FILE);
-		return EXIT_FAILURE;
-	}
-
-	const uint8_t *frame =
-	    mmap(NULL, FRAME_BYTES, PROT_READ, MAP_SHARED, frame_fd, 0);
-	if (frame == MAP_FAILED) {
-		perror("mmap");
-		return EXIT_FAILURE;
-	}
-
-	pthread_t sender;
-	if (pthread_create(&sender, NULL, send_thread, argv[1]) != 0) {
-		perror("pthread_create");
-		return EXIT_FAILURE;
-	}
-
-	/* Repeatedly repaint the frame in a random pixel order. */
-	for (;;) {
-		for (int i = 0; i < PACKET_COUNT; i++) {
-			for (int j = 0; j < PIXELS_PER_PACKET; j++) {
-				uint64_t src = xorshf96() % FRAME_PIXELS;
-				uint16_t fx = (uint16_t)(src % FRAME_WIDTH);
-				uint16_t fy = (uint16_t)(src / FRAME_WIDTH);
-				const uint8_t *rgb = &frame[src * FRAME_CHANNELS];
-
-				struct pixelflut_pixel *px = &packets[i].pixels[j];
-				px->x = htole16((uint16_t)(CANVAS_OFFSET_X + fx));
-				px->y = htole16((uint16_t)(CANVAS_OFFSET_Y + fy));
-				px->r = rgb[0];
-				px->g = rgb[1];
-				px->b = rgb[2];
-			}
-		}
-	}
-}
-

pixelflut.c

@@ -0,0 +1,302 @@
+/*
+ * Pixelflut UDP renderer.
+ *
+ * Reads raw RGB24 frames that main.py continuously writes to INPUT_FILE
+ * (via mmap) and floods them at a Pixelflut server using the binary UDP
+ * protocol. Pixels are picked in a random order so the picture "dissolves"
+ * onto the wall instead of being drawn scanline by scanline.
+ *
+ * Throughput comes from parallelism: one sender core tops out around
+ * ~0.65 Mpps on this hardware, so we run one self-contained worker thread per
+ * CPU. Each worker owns its own socket, packet ring and PRNG and simply loops
+ * "repaint the ring, blast it with sendmmsg()" as fast as it can. There is no
+ * shared mutable state between workers, so it scales close to linearly. See
+ * bench.c / pixelflut-bench for the measurements behind this design.
+ *
+ * Binary Pixelflut packet layout (little-endian on the wire):
+ *
+ *   +------+---------+  pixelflut_packet
+ *   | cmd  | subcmd  |  2-byte header
+ *   +------+---------+---------+----+----+----+  pixelflut_pixel[0]
+ *   |   x (u16)      | y (u16) | r  | g  | b  |  7 bytes
+ *   +----------------+---------+----+----+----+
+ *   |                   ...                   |  PIXELS_PER_PACKET pixels
+ *   +----------------------------------------+
+ */
+
+#define _GNU_SOURCE /* sendmmsg(), struct mmsghdr, SO_BINDTODEVICE */
+
+#include <errno.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <arpa/inet.h>
+#include <net/if.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+
+#include <fcntl.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+
+#include <pthread.h>
+
+#include "pixelflut.h"
+
+/* Portable host -> little-endian 16-bit conversion (matches the wire format). */
+#if defined(__linux__)
+#  include <endian.h>
+#elif defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__)
+#  include <sys/endian.h>
+#elif defined(__APPLE__)
+#  include <libkern/OSByteOrder.h>
+#  define htole16(v) OSSwapHostToLittleInt16(v)
+#endif
+
+/* ---- Frame geometry (must match the producer, see main.py). ------------- */
+#define FRAME_WIDTH    640
+#define FRAME_HEIGHT   360
+#define FRAME_CHANNELS 3 /* RGB, one byte per channel */
+#define FRAME_PIXELS   (FRAME_WIDTH * FRAME_HEIGHT)
+#define FRAME_BYTES    (FRAME_PIXELS * FRAME_CHANNELS)
+
+/* Top-left corner where the frame is drawn on the shared wall. */
+#define CANVAS_OFFSET_X 1200
+#define CANVAS_OFFSET_Y 40
+
+/* Number of datagrams needed to cover a whole frame once. */
+#define PACKET_COUNT (FRAME_PIXELS / PIXELS_PER_PACKET)
+
+/* ---- Target / input (override at build time with -D...). --------------- */
+/* #define DISPLAY_HOST "100.65.0.2" */
+#ifndef DISPLAY_HOST
+#  define DISPLAY_HOST "fcda:f100::d"
+#endif
+#ifndef DISPLAY_PORT
+#  define DISPLAY_PORT 5005
+#endif
+#ifndef INPUT_FILE
+#  define INPUT_FILE "/tmp/video"
+#endif
+
+#define MAX_WORKERS 256
+
+_Static_assert(FRAME_PIXELS % PIXELS_PER_PACKET == 0,
+               "frame pixels must divide evenly into packets");
+
+/* xorshf96 PRNG (Marsaglia), period 2^96-1. Per-worker so there is no
+ * contention and each worker paints a different random pixel stream. */
+struct rng {
+	unsigned long x, y, z;
+};
+
+static unsigned long xorshf96(struct rng *r)
+{
+	r->x ^= r->x << 16;
+	r->x ^= r->x >> 5;
+	r->x ^= r->x << 1;
+
+	unsigned long t = r->x;
+	r->x = r->y;
+	r->y = r->z;
+	r->z = t ^ r->x ^ r->y;
+
+	return r->z;
+}
+
+struct worker {
+	const char *ifname;
+	const uint8_t *frame;
+	int index;
+};
+
+/* Create a UDP socket connected to the display, optionally pinned to a NIC. */
+static int open_display_socket(const char *ifname)
+{
+	int fd = socket(AF_INET6, SOCK_DGRAM, 0);
+	if (fd == -1) {
+		perror("socket");
+		return -1;
+	}
+
+#ifdef SO_BINDTODEVICE
+	/* Best effort: pin traffic to the given interface (needs privileges). */
+	if (ifname != NULL && ifname[0] != '\0') {
+		struct ifreq ifr;
+		memset(&ifr, 0, sizeof ifr);
+		snprintf(ifr.ifr_name, sizeof ifr.ifr_name, "%s", ifname);
+		if (setsockopt(fd, SOL_SOCKET, SO_BINDTODEVICE, &ifr,
+		               sizeof ifr) == -1)
+			perror("setsockopt(SO_BINDTODEVICE)");
+	}
+#else
+	(void)ifname;
+#endif
+
+	struct sockaddr_in6 addr;
+	memset(&addr, 0, sizeof addr);
+	addr.sin6_family = AF_INET6;
+	addr.sin6_port = htons(DISPLAY_PORT);
+	if (inet_pton(AF_INET6, DISPLAY_HOST, &addr.sin6_addr) != 1) {
+		fprintf(stderr, "invalid display host: %s\n", DISPLAY_HOST);
+		close(fd);
+		return -1;
+	}
+
+	if (connect(fd, (struct sockaddr *)&addr, sizeof addr) == -1) {
+		perror("connect");
+		close(fd);
+		return -1;
+	}
+
+	return fd;
+}
+
+/* Repaint the whole ring by sampling random pixels from the current frame. */
+static void repaint(struct pixelflut_packet *packets, const uint8_t *frame,
+                    struct rng *rng)
+{
+	for (int i = 0; i < PACKET_COUNT; i++) {
+		for (int j = 0; j < PIXELS_PER_PACKET; j++) {
+			uint64_t src = xorshf96(rng) % FRAME_PIXELS;
+			uint16_t fx = (uint16_t)(src % FRAME_WIDTH);
+			uint16_t fy = (uint16_t)(src / FRAME_WIDTH);
+			const uint8_t *rgb = &frame[src * FRAME_CHANNELS];
+
+			struct pixelflut_pixel *px = &packets[i].pixels[j];
+			px->x = htole16((uint16_t)(CANVAS_OFFSET_X + fx));
+			px->y = htole16((uint16_t)(CANVAS_OFFSET_Y + fy));
+			px->r = rgb[0];
+			px->g = rgb[1];
+			px->b = rgb[2];
+		}
+	}
+}
+
+/* One worker: repaint and blast its own ring over its own socket, forever. */
+static void *worker_main(void *arg)
+{
+	const struct worker *w = arg;
+
+	struct rng rng = {
+	    .x = 123456789UL ^ (2654435761UL * (unsigned long)(w->index + 1)),
+	    .y = 362436069UL + (unsigned long)(w->index + 1),
+	    .z = 521288629UL,
+	};
+
+	int fd = open_display_socket(w->ifname);
+	if (fd == -1)
+		exit(EXIT_FAILURE);
+
+	struct pixelflut_packet *packets = malloc(PACKET_COUNT * sizeof *packets);
+	struct mmsghdr *messages = calloc(PACKET_COUNT, sizeof *messages);
+	struct iovec *iovecs = calloc(PACKET_COUNT, sizeof *iovecs);
+	if (packets == NULL || messages == NULL || iovecs == NULL) {
+		fprintf(stderr, "worker %d: out of memory\n", w->index);
+		exit(EXIT_FAILURE);
+	}
+
+	for (int i = 0; i < PACKET_COUNT; i++) {
+		packets[i].cmd = PIXELFLUT_CMD_SET;
+		packets[i].subcmd = PIXELFLUT_SUBCMD_RGB;
+		iovecs[i].iov_base = &packets[i];
+		iovecs[i].iov_len = sizeof packets[i];
+		messages[i].msg_hdr.msg_iov = &iovecs[i];
+		messages[i].msg_hdr.msg_iovlen = 1;
+	}
+
+	for (;;) {
+		repaint(packets, w->frame, &rng);
+
+		int next = 0;
+		while (next < PACKET_COUNT) {
+			int sent = sendmmsg(fd, &messages[next],
+			                    (unsigned int)(PACKET_COUNT - next), 0);
+			if (sent >= 0) {
+				next += sent;
+				continue;
+			}
+			switch (errno) {
+			case EINTR:
+				continue; /* interrupted before sending - retry now */
+			case ENOBUFS: /* interface send queue full: we outrun the link */
+			case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+			case EWOULDBLOCK:
+#endif
+				/* Transient back-pressure. Back off briefly so the queue
+				 * drains (rather than spinning on failing syscalls), then
+				 * resend the same batch - no need to repaint. */
+				usleep(200);
+				continue;
+			case ECONNREFUSED:
+				break; /* stale ICMP on connected UDP; repaint and retry */
+			default:
+				perror("sendmmsg");
+				break;
+			}
+			break; /* ECONNREFUSED / unexpected: stop, repaint, retry */
+		}
+	}
+
+	return NULL; /* unreachable */
+}
+
+int main(int argc, char *argv[])
+{
+	if (argc < 2 || argc > 3) {
+		fprintf(stderr, "usage: %s <interface> [threads]\n", argv[0]);
+		return EXIT_FAILURE;
+	}
+	const char *ifname = argv[1];
+
+	int workers = (int)sysconf(_SC_NPROCESSORS_ONLN);
+	if (argc == 3)
+		workers = atoi(argv[2]);
+	if (workers < 1)
+		workers = 1;
+	if (workers > MAX_WORKERS)
+		workers = MAX_WORKERS;
+
+	int frame_fd = open(INPUT_FILE, O_RDONLY);
+	if (frame_fd == -1) {
+		perror("open " INPUT_FILE);
+		return EXIT_FAILURE;
+	}
+
+	const uint8_t *frame =
+	    mmap(NULL, FRAME_BYTES, PROT_READ, MAP_SHARED, frame_fd, 0);
+	if (frame == MAP_FAILED) {
+		perror("mmap");
+		return EXIT_FAILURE;
+	}
+
+	printf("rendering %s to [%s]:%d via %s using %d worker(s)\n", INPUT_FILE,
+	       DISPLAY_HOST, DISPLAY_PORT, ifname, workers);
+
+	struct worker *wctx = calloc((size_t)workers, sizeof *wctx);
+	pthread_t *tids = calloc((size_t)workers, sizeof *tids);
+	if (wctx == NULL || tids == NULL) {
+		fprintf(stderr, "out of memory\n");
+		return EXIT_FAILURE;
+	}
+
+	for (int i = 0; i < workers; i++) {
+		wctx[i] = (struct worker){.ifname = ifname, .frame = frame, .index = i};
+		if (pthread_create(&tids[i], NULL, worker_main, &wctx[i]) != 0) {
+			perror("pthread_create");
+			return EXIT_FAILURE;
+		}
+	}
+
+	for (int i = 0; i < workers; i++)
+		pthread_join(tids[i], NULL);
+
+	free(wctx);
+	free(tids);
+	return EXIT_SUCCESS;
+}

pixelflut.h

@@ -0,0 +1,55 @@
+/*
+ * Pixelflut binary UDP protocol, shared by the renderer and the benchmark.
+ *
+ * Packet layout (little-endian on the wire):
+ *
+ *   +------+---------+  pixelflut_packet
+ *   | cmd  | subcmd  |  2-byte header
+ *   +------+---------+---------+----+----+----+  pixelflut_pixel[0]
+ *   |   x (u16)      | y (u16) | r  | g  | b  |  7 bytes
+ *   +----------------+---------+----+----+----+
+ *   |                   ...                   |  PIXELS_PER_PACKET pixels
+ *   +----------------------------------------+
+ */
+
+#ifndef PIXELFLUT_H
+#define PIXELFLUT_H
+
+#include <stdint.h>
+
+/* Header bytes. */
+#define PIXELFLUT_CMD_SET    0x00 /* "set pixels" command ... */
+#define PIXELFLUT_SUBCMD_RGB 0x01 /* ... with 3-byte RGB colours */
+
+/* Pixels batched into a single UDP datagram. Bigger datagrams amortize the
+ * per-message send cost over more pixels; keep the packet within the path MTU
+ * to avoid IP fragmentation (e.g. 207 pixels ~= 1451 B fits a 1500 B link).
+ * Override at build time with -DPIXELS_PER_PACKET=N. */
+#ifndef PIXELS_PER_PACKET
+#  define PIXELS_PER_PACKET 200
+#endif
+
+#pragma pack(push, 1)
+/* A single pixel as it appears on the wire. */
+struct pixelflut_pixel {
+	uint16_t x; /* little-endian */
+	uint16_t y; /* little-endian */
+	uint8_t r;
+	uint8_t g;
+	uint8_t b;
+};
+
+/* A full datagram: header followed by a batch of pixels. */
+struct pixelflut_packet {
+	uint8_t cmd;
+	uint8_t subcmd;
+	struct pixelflut_pixel pixels[PIXELS_PER_PACKET];
+};
+#pragma pack(pop)
+
+_Static_assert(sizeof(struct pixelflut_pixel) == 7,
+               "pixel must be exactly 7 bytes on the wire");
+_Static_assert(sizeof(struct pixelflut_packet) == 2 + 7 * PIXELS_PER_PACKET,
+               "packet must be header + PIXELS_PER_PACKET pixels, no padding");
+
+#endif /* PIXELFLUT_H */