pixelflut-freebsd/pixelflut.c

/*
 * 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 80
#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;
}