From ee414035de0265e51505f4a8ddde7ea1c0ecc255 Mon Sep 17 00:00:00 2001 From: Alexander Jacobsen Date: Fri, 17 Jul 2026 23:01:12 +0200 Subject: [PATCH 1/2] more faster --- .gitignore | 1 + Makefile | 40 ++-- bench.c | 546 ++++++++++++++++++++++++++++++++++++++++++++++++++++ new.c | 249 ++++++++++++++---------- pixelflut.h | 55 ++++++ 5 files changed, 779 insertions(+), 112 deletions(-) create mode 100644 bench.c create mode 100644 pixelflut.h diff --git a/.gitignore b/.gitignore index d59b712..abad749 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ venv/ # C build artifacts *.o pixelflut-render +pixelflut-bench render a.out diff --git a/Makefile b/Makefile index 003df8c..7a6c18a 100644 --- a/Makefile +++ b/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 diff --git a/bench.c b/bench.c new file mode 100644 index 0000000..a50fa81 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +#include "pixelflut.h" + +/* Portable host -> little-endian 16-bit conversion (matches the wire format). */ +#if defined(__linux__) +# include +#elif defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__) +# include +#elif defined(__APPLE__) +# include +# 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; +} diff --git a/new.c b/new.c index 8a18c1d..d692747 100644 --- a/new.c +++ b/new.c @@ -6,6 +6,13 @@ * 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 @@ -19,6 +26,7 @@ #define _GNU_SOURCE /* sendmmsg(), struct mmsghdr, SO_BINDTODEVICE */ +#include #include #include #include @@ -37,6 +45,8 @@ #include +#include "pixelflut.h" + /* Portable host -> little-endian 16-bit conversion (matches the wire format). */ #if defined(__linux__) # include @@ -58,76 +68,59 @@ #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. --------------------------------------------------- */ +/* ---- Target / input (override at build time with -D...). --------------- */ /* #define DISPLAY_HOST "100.65.0.2" */ -#define DISPLAY_HOST "fcda:f100::d" -#define DISPLAY_PORT 5005 -#define INPUT_FILE "/tmp/video" +#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 -/* 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; -}; +#define MAX_WORKERS 256 -/* 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 (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; +}; -/* 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) +static unsigned long xorshf96(struct rng *r) { - rng_x ^= rng_x << 16; - rng_x ^= rng_x >> 5; - rng_x ^= rng_x << 1; + r->x ^= r->x << 16; + r->x ^= r->x >> 5; + r->x ^= r->x << 1; - unsigned long t = rng_x; - rng_x = rng_y; - rng_y = rng_z; - rng_z = t ^ rng_x ^ rng_y; + unsigned long t = r->x; + r->x = r->y; + r->y = r->z; + r->z = t ^ r->x ^ r->y; - return rng_z; + return r->z; } -/* Continuously blast the pre-built packets at the display. */ -static void *send_thread(void *arg) -{ - const char *ifname = arg; +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"); - exit(EXIT_FAILURE); + return -1; } #ifdef SO_BINDTODEVICE @@ -137,9 +130,8 @@ static void *send_thread(void *arg) 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) { + sizeof ifr) == -1) perror("setsockopt(SO_BINDTODEVICE)"); - } } #else (void)ifname; @@ -151,26 +143,92 @@ static void *send_thread(void *arg) 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); + 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); } - sleep(1); /* give the producer time to fill the first frame */ + 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 next = 0; for (;;) { - int sent = sendmmsg(fd, &messages[next], PACKET_COUNT - next, 0); - if (sent == -1) { - perror("sendmmsg"); - continue; + 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 == -1) { + if (errno == EINTR) + continue; + /* ECONNREFUSED is stale ICMP feedback on a connected + * UDP socket; EAGAIN means the send buffer is full. + * Both are transient - repaint and keep sending. */ + if (errno != ECONNREFUSED && errno != EAGAIN && + errno != EWOULDBLOCK) + perror("sendmmsg"); + break; + } + next += sent; } - next += sent; /* resume after a partial send, wrap at the end */ - if (next >= PACKET_COUNT) - next = 0; } return NULL; /* unreachable */ @@ -178,25 +236,19 @@ static void *send_thread(void *arg) int main(int argc, char *argv[]) { - if (argc != 2) { - fprintf(stderr, "usage: %s \n", argv[0]); + if (argc < 2 || argc > 3) { + fprintf(stderr, "usage: %s [threads]\n", argv[0]); return EXIT_FAILURE; } + const char *ifname = argv[1]; - 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 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) { @@ -211,29 +263,28 @@ int main(int argc, char *argv[]) return EXIT_FAILURE; } - pthread_t sender; - if (pthread_create(&sender, NULL, send_thread, argv[1]) != 0) { - perror("pthread_create"); + 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; } - /* 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]; - } + 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; +} diff --git a/pixelflut.h b/pixelflut.h new file mode 100644 index 0000000..99ef224 --- /dev/null +++ b/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 + +/* 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 120 +#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 */ From 52ee2fd3918d81803fcc9220592af2957e7df0f5 Mon Sep 17 00:00:00 2001 From: Alexander Jacobsen Date: Sat, 18 Jul 2026 01:21:41 +0200 Subject: [PATCH 2/2] fix ENOBUFS errors with backoff+retry --- new.c => pixelflut.c | 32 ++++++++++++++++++++++---------- pixelflut.h | 2 +- 2 files changed, 23 insertions(+), 11 deletions(-) rename new.c => pixelflut.c (92%) diff --git a/new.c b/pixelflut.c similarity index 92% rename from new.c rename to pixelflut.c index d692747..1a8b2e6 100644 --- a/new.c +++ b/pixelflut.c @@ -216,18 +216,30 @@ static void *worker_main(void *arg) while (next < PACKET_COUNT) { int sent = sendmmsg(fd, &messages[next], (unsigned int)(PACKET_COUNT - next), 0); - if (sent == -1) { - if (errno == EINTR) - continue; - /* ECONNREFUSED is stale ICMP feedback on a connected - * UDP socket; EAGAIN means the send buffer is full. - * Both are transient - repaint and keep sending. */ - if (errno != ECONNREFUSED && errno != EAGAIN && - errno != EWOULDBLOCK) - perror("sendmmsg"); + 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; } - next += sent; + break; /* ECONNREFUSED / unexpected: stop, repaint, retry */ } } diff --git a/pixelflut.h b/pixelflut.h index 99ef224..638a0ba 100644 --- a/pixelflut.h +++ b/pixelflut.h @@ -26,7 +26,7 @@ * 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 120 +# define PIXELS_PER_PACKET 200 #endif #pragma pack(push, 1)