fix ENOBUFS errors with backoff+retry
This commit is contained in:
parent
ee414035de
commit
52ee2fd391
2 changed files with 23 additions and 11 deletions
290
new.c
290
new.c
|
|
@ -1,290 +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.
|
||||
*
|
||||
* 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 == -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;
|
||||
}
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue