Implement generic MAX2771 raw Ethernet recorder
This commit is contained in:
onur
@@ -0,0 +1,15 @@
|
||||
CC ?= gcc
|
||||
CFLAGS ?= -O3 -std=c11 -Wall -Wextra -Wpedantic
|
||||
TARGET := max2771_raw_rx
|
||||
|
||||
.PHONY: all clean test
|
||||
all: $(TARGET)
|
||||
|
||||
$(TARGET): max2771_raw_rx.c
|
||||
$(CC) $(CFLAGS) -o $@ $<
|
||||
|
||||
test: $(TARGET)
|
||||
./$(TARGET) --self-test
|
||||
|
||||
clean:
|
||||
$(RM) $(TARGET)
|
||||
@@ -0,0 +1,571 @@
|
||||
#define _GNU_SOURCE
|
||||
#include <arpa/inet.h>
|
||||
#include <errno.h>
|
||||
#include <getopt.h>
|
||||
#include <linux/if_packet.h>
|
||||
#include <linux/if_ether.h>
|
||||
#include <net/if.h>
|
||||
#include <poll.h>
|
||||
#include <signal.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdint.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/mman.h>
|
||||
#include <sys/socket.h>
|
||||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#define MX27_ETHERTYPE 0x88b5u
|
||||
#define MX27_HEADER_LEN 32u
|
||||
#define OUTPUT_BUFFER_SIZE (8u * 1024u * 1024u)
|
||||
#define RING_BLOCK_SIZE (2u * 1024u * 1024u)
|
||||
#define RING_BLOCK_COUNT 64u
|
||||
#define RING_FRAME_SIZE 2048u
|
||||
|
||||
enum gap_mode { GAP_STOP, GAP_SKIP, GAP_ZERO_FILL };
|
||||
|
||||
struct options {
|
||||
const char *interface_name;
|
||||
const char *output_path;
|
||||
const char *log_path;
|
||||
bool source_filter;
|
||||
uint8_t source_mac[6];
|
||||
enum gap_mode gap_mode;
|
||||
double duration;
|
||||
uint64_t max_bytes;
|
||||
bool quiet;
|
||||
bool verbose;
|
||||
bool self_test;
|
||||
};
|
||||
|
||||
struct sample_writer {
|
||||
FILE *file;
|
||||
bool high_pending;
|
||||
uint8_t pending_byte;
|
||||
uint64_t samples_written;
|
||||
uint64_t bytes_written;
|
||||
uint64_t sample_limit;
|
||||
};
|
||||
|
||||
struct stats {
|
||||
uint64_t received_frames;
|
||||
uint64_t invalid_frames;
|
||||
uint64_t missing_frames;
|
||||
uint64_t out_of_order_frames;
|
||||
uint64_t host_missing_samples;
|
||||
uint64_t fpga_dropped_samples;
|
||||
uint64_t first_sample_index;
|
||||
uint64_t last_sample_index;
|
||||
bool have_sample_index;
|
||||
uint32_t expected_sequence;
|
||||
bool have_sequence;
|
||||
uint64_t expected_sample_index;
|
||||
bool have_expected_sample;
|
||||
uint32_t last_drop_low;
|
||||
bool have_drop_count;
|
||||
uint8_t protocol_version;
|
||||
uint8_t observed_source[6];
|
||||
bool have_source;
|
||||
uint64_t socket_packets;
|
||||
uint64_t socket_drops;
|
||||
uint64_t freeze_count;
|
||||
};
|
||||
|
||||
struct receiver_context {
|
||||
struct options *opt;
|
||||
struct sample_writer writer;
|
||||
struct stats stats;
|
||||
};
|
||||
|
||||
static volatile sig_atomic_t stop_requested;
|
||||
|
||||
static void signal_handler(int signal_number) {
|
||||
(void)signal_number;
|
||||
stop_requested = 1;
|
||||
}
|
||||
|
||||
static uint16_t read_be16(const uint8_t *p) {
|
||||
return (uint16_t)(((uint16_t)p[0] << 8) | (uint16_t)p[1]);
|
||||
}
|
||||
|
||||
static uint32_t read_be32(const uint8_t *p) {
|
||||
return ((uint32_t)p[0] << 24) | ((uint32_t)p[1] << 16) |
|
||||
((uint32_t)p[2] << 8) | (uint32_t)p[3];
|
||||
}
|
||||
|
||||
static uint64_t read_be64(const uint8_t *p) {
|
||||
return ((uint64_t)read_be32(p) << 32) | (uint64_t)read_be32(p + 4);
|
||||
}
|
||||
|
||||
static double monotonic_seconds(void) {
|
||||
struct timespec ts;
|
||||
if (clock_gettime(CLOCK_MONOTONIC, &ts) != 0) {
|
||||
perror("clock_gettime");
|
||||
exit(EXIT_FAILURE);
|
||||
}
|
||||
return (double)ts.tv_sec + (double)ts.tv_nsec / 1000000000.0;
|
||||
}
|
||||
|
||||
static int parse_mac(const char *text, uint8_t mac[6]) {
|
||||
unsigned int v[6];
|
||||
char extra;
|
||||
if (sscanf(text, "%x:%x:%x:%x:%x:%x%c", &v[0], &v[1], &v[2],
|
||||
&v[3], &v[4], &v[5], &extra) != 6) return -1;
|
||||
for (size_t i = 0; i < 6; ++i) {
|
||||
if (v[i] > 255u) return -1;
|
||||
mac[i] = (uint8_t)v[i];
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void format_mac(char out[18], const uint8_t mac[6]) {
|
||||
(void)snprintf(out, 18, "%02x:%02x:%02x:%02x:%02x:%02x",
|
||||
mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]);
|
||||
}
|
||||
|
||||
static int write_bytes(struct sample_writer *w, const uint8_t *data, size_t length) {
|
||||
if (length == 0u) return 0;
|
||||
if (fwrite(data, 1u, length, w->file) != length) {
|
||||
fprintf(stderr, "output write failed: %s\n", strerror(errno));
|
||||
return -1;
|
||||
}
|
||||
w->bytes_written += (uint64_t)length;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static bool writer_at_limit(const struct sample_writer *w) {
|
||||
return w->sample_limit != 0u && w->samples_written >= w->sample_limit;
|
||||
}
|
||||
|
||||
static int write_nibble(struct sample_writer *w, uint8_t value) {
|
||||
if (writer_at_limit(w)) return 1;
|
||||
value &= 0x0fu;
|
||||
if (!w->high_pending) {
|
||||
w->pending_byte = (uint8_t)(value << 4);
|
||||
w->high_pending = true;
|
||||
} else {
|
||||
uint8_t byte = (uint8_t)(w->pending_byte | value);
|
||||
if (write_bytes(w, &byte, 1u) != 0) return -1;
|
||||
w->high_pending = false;
|
||||
}
|
||||
++w->samples_written;
|
||||
return writer_at_limit(w) ? 1 : 0;
|
||||
}
|
||||
|
||||
static int write_zero_samples(struct sample_writer *w, uint64_t count) {
|
||||
static const uint8_t zeros[65536] = {0};
|
||||
if (w->sample_limit != 0u && count > w->sample_limit - w->samples_written)
|
||||
count = w->sample_limit - w->samples_written;
|
||||
if (w->high_pending && count != 0u) {
|
||||
int rc = write_nibble(w, 0u);
|
||||
--count;
|
||||
if (rc != 0) return rc;
|
||||
}
|
||||
while (count >= 2u) {
|
||||
uint64_t byte_count = count / 2u;
|
||||
size_t chunk = byte_count > sizeof(zeros) ? sizeof(zeros) : (size_t)byte_count;
|
||||
if (write_bytes(w, zeros, chunk) != 0) return -1;
|
||||
w->samples_written += (uint64_t)chunk * 2u;
|
||||
count -= (uint64_t)chunk * 2u;
|
||||
}
|
||||
if (count != 0u) return write_nibble(w, 0u);
|
||||
return writer_at_limit(w) ? 1 : 0;
|
||||
}
|
||||
|
||||
static int write_packed_samples(struct sample_writer *w, const uint8_t *packed,
|
||||
uint32_t first_sample, uint32_t count) {
|
||||
if (w->sample_limit != 0u) {
|
||||
uint64_t remaining = w->sample_limit - w->samples_written;
|
||||
if ((uint64_t)count > remaining) count = (uint32_t)remaining;
|
||||
}
|
||||
if (!w->high_pending && (first_sample & 1u) == 0u && (count & 1u) == 0u) {
|
||||
if (write_bytes(w, packed + first_sample / 2u, count / 2u) != 0) return -1;
|
||||
w->samples_written += count;
|
||||
return writer_at_limit(w) ? 1 : 0;
|
||||
}
|
||||
for (uint32_t i = 0; i < count; ++i) {
|
||||
uint32_t sample = first_sample + i;
|
||||
uint8_t byte = packed[sample / 2u];
|
||||
uint8_t nibble = (sample & 1u) == 0u ? (uint8_t)(byte >> 4) : (uint8_t)(byte & 0x0fu);
|
||||
int rc = write_nibble(w, nibble);
|
||||
if (rc != 0) return rc;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int finish_writer(struct sample_writer *w) {
|
||||
if (w->high_pending) {
|
||||
if (write_bytes(w, &w->pending_byte, 1u) != 0) return -1;
|
||||
w->high_pending = false;
|
||||
}
|
||||
return fflush(w->file) == 0 ? 0 : -1;
|
||||
}
|
||||
|
||||
/* Return 0 to continue, 1 for requested clean stop, and -1 on fatal I/O error. */
|
||||
static int process_frame(struct receiver_context *ctx, const uint8_t *frame, size_t length) {
|
||||
struct options *opt = ctx->opt;
|
||||
struct stats *s = &ctx->stats;
|
||||
if (length < ETH_HLEN + MX27_HEADER_LEN) return 0;
|
||||
if (read_be16(frame + 12) != MX27_ETHERTYPE) return 0;
|
||||
const uint8_t *source = frame + 6;
|
||||
if (opt->source_filter && memcmp(source, opt->source_mac, 6) != 0) return 0;
|
||||
|
||||
const uint8_t *h = frame + ETH_HLEN;
|
||||
if (memcmp(h, "MX27", 4) != 0 || h[4] != 1u || read_be16(h + 6) != MX27_HEADER_LEN) {
|
||||
++s->invalid_frames;
|
||||
return 0;
|
||||
}
|
||||
uint16_t sample_count = read_be16(h + 20);
|
||||
size_t sample_bytes = ((size_t)sample_count + 1u) / 2u;
|
||||
if (h[22] != 4u || h[23] != 1u || sample_count == 0u ||
|
||||
MX27_HEADER_LEN + sample_bytes > 1500u ||
|
||||
length < ETH_HLEN + MX27_HEADER_LEN + sample_bytes) {
|
||||
++s->invalid_frames;
|
||||
return 0;
|
||||
}
|
||||
|
||||
uint8_t flags = h[5];
|
||||
uint32_t sequence = read_be32(h + 8);
|
||||
uint64_t first_index = read_be64(h + 12);
|
||||
uint32_t drop_low = read_be32(h + 24);
|
||||
bool discontinuity = (flags & 1u) != 0u;
|
||||
uint64_t positive_gap = 0u;
|
||||
uint32_t skip_leading = 0u;
|
||||
|
||||
if (!s->have_source) {
|
||||
memcpy(s->observed_source, source, 6);
|
||||
s->have_source = true;
|
||||
s->protocol_version = h[4];
|
||||
}
|
||||
|
||||
if (s->have_sequence && sequence != s->expected_sequence) {
|
||||
uint32_t delta = sequence - s->expected_sequence;
|
||||
discontinuity = true;
|
||||
if (delta < 0x80000000u) s->missing_frames += delta;
|
||||
else ++s->out_of_order_frames;
|
||||
if (!opt->quiet)
|
||||
fprintf(stderr, "packet sequence discontinuity: expected %u got %u\n",
|
||||
s->expected_sequence, sequence);
|
||||
}
|
||||
s->expected_sequence = sequence + 1u;
|
||||
s->have_sequence = true;
|
||||
|
||||
if (s->have_expected_sample && first_index != s->expected_sample_index) {
|
||||
discontinuity = true;
|
||||
if (first_index > s->expected_sample_index) {
|
||||
positive_gap = first_index - s->expected_sample_index;
|
||||
s->host_missing_samples += positive_gap;
|
||||
} else {
|
||||
uint64_t overlap = s->expected_sample_index - first_index;
|
||||
skip_leading = overlap >= sample_count ? sample_count : (uint32_t)overlap;
|
||||
}
|
||||
if (!opt->quiet)
|
||||
fprintf(stderr, "sample-index discontinuity: expected %llu got %llu\n",
|
||||
(unsigned long long)s->expected_sample_index,
|
||||
(unsigned long long)first_index);
|
||||
}
|
||||
|
||||
if (!s->have_drop_count) {
|
||||
s->fpga_dropped_samples = drop_low;
|
||||
s->have_drop_count = true;
|
||||
} else {
|
||||
uint32_t delta = drop_low - s->last_drop_low;
|
||||
if (delta != 0u) discontinuity = true;
|
||||
s->fpga_dropped_samples += delta;
|
||||
}
|
||||
s->last_drop_low = drop_low;
|
||||
|
||||
if (discontinuity && opt->gap_mode == GAP_STOP) {
|
||||
if (!opt->quiet) fprintf(stderr, "stopping on discontinuity (--gap-mode stop)\n");
|
||||
return 1;
|
||||
}
|
||||
if (positive_gap != 0u && opt->gap_mode == GAP_ZERO_FILL) {
|
||||
int rc = write_zero_samples(&ctx->writer, positive_gap);
|
||||
if (rc != 0) return rc;
|
||||
}
|
||||
|
||||
uint32_t samples_to_write = (uint32_t)sample_count - skip_leading;
|
||||
int rc = write_packed_samples(&ctx->writer, h + MX27_HEADER_LEN,
|
||||
skip_leading, samples_to_write);
|
||||
if (rc < 0) return rc;
|
||||
|
||||
++s->received_frames;
|
||||
if (!s->have_sample_index) {
|
||||
s->first_sample_index = first_index;
|
||||
s->have_sample_index = true;
|
||||
}
|
||||
s->last_sample_index = first_index + (uint64_t)sample_count - 1u;
|
||||
s->expected_sample_index = first_index + (uint64_t)sample_count;
|
||||
s->have_expected_sample = true;
|
||||
if (opt->verbose)
|
||||
fprintf(stderr, "frame=%u first=%llu samples=%u flags=0x%02x dropped=%u\n",
|
||||
sequence, (unsigned long long)first_index, sample_count, flags, drop_low);
|
||||
return rc;
|
||||
}
|
||||
|
||||
static void usage(FILE *stream, const char *program) {
|
||||
fprintf(stream,
|
||||
"Usage: %s --interface IFACE [options]\n"
|
||||
" --output FILE packed sample output (default max2771_capture.bin)\n"
|
||||
" --log FILE text log (default OUTPUT.log)\n"
|
||||
" --source-mac MAC accept only this source MAC\n"
|
||||
" --gap-mode MODE stop, skip, or zero-fill (default zero-fill)\n"
|
||||
" --duration SECONDS stop after elapsed wall time\n"
|
||||
" --max-bytes BYTES stop at this many output bytes\n"
|
||||
" --quiet | --verbose select diagnostic detail\n"
|
||||
" --help show this help\n", program);
|
||||
}
|
||||
|
||||
static int parse_options(int argc, char **argv, struct options *opt) {
|
||||
static const struct option long_options[] = {
|
||||
{"interface", required_argument, NULL, 'i'}, {"output", required_argument, NULL, 'o'},
|
||||
{"log", required_argument, NULL, 'l'}, {"source-mac", required_argument, NULL, 's'},
|
||||
{"gap-mode", required_argument, NULL, 'g'}, {"duration", required_argument, NULL, 'd'},
|
||||
{"max-bytes", required_argument, NULL, 'm'}, {"quiet", no_argument, NULL, 'q'},
|
||||
{"verbose", no_argument, NULL, 'v'}, {"self-test", no_argument, NULL, 't'},
|
||||
{"help", no_argument, NULL, 'h'}, {NULL, 0, NULL, 0}
|
||||
};
|
||||
memset(opt, 0, sizeof(*opt));
|
||||
opt->output_path = "max2771_capture.bin";
|
||||
opt->gap_mode = GAP_ZERO_FILL;
|
||||
int c;
|
||||
while ((c = getopt_long(argc, argv, "i:o:l:s:g:d:m:qvth", long_options, NULL)) != -1) {
|
||||
char *end = NULL;
|
||||
switch (c) {
|
||||
case 'i': opt->interface_name = optarg; break;
|
||||
case 'o': opt->output_path = optarg; break;
|
||||
case 'l': opt->log_path = optarg; break;
|
||||
case 's':
|
||||
if (parse_mac(optarg, opt->source_mac) != 0) { fprintf(stderr,"invalid MAC: %s\n",optarg); return -1; }
|
||||
opt->source_filter = true; break;
|
||||
case 'g':
|
||||
if (strcmp(optarg,"stop")==0) opt->gap_mode=GAP_STOP;
|
||||
else if (strcmp(optarg,"skip")==0) opt->gap_mode=GAP_SKIP;
|
||||
else if (strcmp(optarg,"zero-fill")==0) opt->gap_mode=GAP_ZERO_FILL;
|
||||
else { fprintf(stderr,"invalid gap mode: %s\n",optarg); return -1; }
|
||||
break;
|
||||
case 'd':
|
||||
errno=0; opt->duration=strtod(optarg,&end);
|
||||
if(errno!=0 || end==optarg || *end!='\0' || opt->duration<=0.0) return -1;
|
||||
break;
|
||||
case 'm':
|
||||
errno=0; opt->max_bytes=strtoull(optarg,&end,0);
|
||||
if(errno!=0 || end==optarg || *end!='\0' || opt->max_bytes==0u) return -1;
|
||||
break;
|
||||
case 'q': opt->quiet=true; opt->verbose=false; break;
|
||||
case 'v': opt->verbose=true; opt->quiet=false; break;
|
||||
case 't': opt->self_test=true; break;
|
||||
case 'h': usage(stdout,argv[0]); exit(EXIT_SUCCESS);
|
||||
default: return -1;
|
||||
}
|
||||
}
|
||||
if (!opt->self_test && opt->interface_name == NULL) return -1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int open_bound_socket(const char *interface_name) {
|
||||
unsigned int ifindex = if_nametoindex(interface_name);
|
||||
if (ifindex == 0u) { fprintf(stderr,"unknown interface %s: %s\n",interface_name,strerror(errno)); return -1; }
|
||||
int fd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
|
||||
if (fd < 0) { perror("socket(AF_PACKET)"); return -1; }
|
||||
int rcvbuf = 256 * 1024 * 1024;
|
||||
if (setsockopt(fd,SOL_SOCKET,SO_RCVBUF,&rcvbuf,sizeof(rcvbuf)) != 0)
|
||||
fprintf(stderr,"warning: SO_RCVBUF request failed: %s\n",strerror(errno));
|
||||
struct sockaddr_ll address;
|
||||
memset(&address,0,sizeof(address));
|
||||
address.sll_family=AF_PACKET;
|
||||
address.sll_protocol=htons(ETH_P_ALL);
|
||||
address.sll_ifindex=(int)ifindex;
|
||||
if (bind(fd,(struct sockaddr *)&address,sizeof(address)) != 0) {
|
||||
fprintf(stderr,"bind(%s): %s\n",interface_name,strerror(errno)); close(fd); return -1;
|
||||
}
|
||||
return fd;
|
||||
}
|
||||
|
||||
static int receive_mmap(int fd, struct receiver_context *ctx, double start_time) {
|
||||
int version=TPACKET_V3;
|
||||
if(setsockopt(fd,SOL_PACKET,PACKET_VERSION,&version,sizeof(version))!=0) return -2;
|
||||
struct tpacket_req3 req;
|
||||
memset(&req,0,sizeof(req));
|
||||
req.tp_block_size=RING_BLOCK_SIZE;
|
||||
req.tp_frame_size=RING_FRAME_SIZE;
|
||||
req.tp_block_nr=RING_BLOCK_COUNT;
|
||||
req.tp_frame_nr=(RING_BLOCK_SIZE/RING_FRAME_SIZE)*RING_BLOCK_COUNT;
|
||||
req.tp_retire_blk_tov=64u;
|
||||
if(setsockopt(fd,SOL_PACKET,PACKET_RX_RING,&req,sizeof(req))!=0) return -2;
|
||||
size_t ring_size=(size_t)req.tp_block_size*req.tp_block_nr;
|
||||
uint8_t *ring=mmap(NULL,ring_size,PROT_READ|PROT_WRITE,MAP_SHARED,fd,0);
|
||||
if(ring==MAP_FAILED) { perror("mmap PACKET_RX_RING"); return -1; }
|
||||
struct pollfd pfd={.fd=fd,.events=POLLIN,.revents=0};
|
||||
unsigned int block=0u;
|
||||
int result=0;
|
||||
while(!stop_requested) {
|
||||
if(ctx->opt->duration>0.0 && monotonic_seconds()-start_time>=ctx->opt->duration) break;
|
||||
struct tpacket_block_desc *bd=(struct tpacket_block_desc *)(ring+(size_t)block*req.tp_block_size);
|
||||
if((bd->hdr.bh1.block_status&TP_STATUS_USER)==0u) {
|
||||
int pr=poll(&pfd,1,250);
|
||||
if(pr<0 && errno!=EINTR) { perror("poll"); result=-1; break; }
|
||||
continue;
|
||||
}
|
||||
struct tpacket3_hdr *packet=(struct tpacket3_hdr *)((uint8_t *)bd+bd->hdr.bh1.offset_to_first_pkt);
|
||||
for(uint32_t n=0;n<bd->hdr.bh1.num_pkts;n++) {
|
||||
const uint8_t *data=(const uint8_t *)packet+packet->tp_mac;
|
||||
int rc=process_frame(ctx,data,packet->tp_snaplen);
|
||||
if(rc!=0) { result=rc<0?-1:0; stop_requested=1; break; }
|
||||
packet=(struct tpacket3_hdr *)((uint8_t *)packet+packet->tp_next_offset);
|
||||
}
|
||||
__sync_synchronize();
|
||||
bd->hdr.bh1.block_status=TP_STATUS_KERNEL;
|
||||
block=(block+1u)%req.tp_block_nr;
|
||||
}
|
||||
if(munmap(ring,ring_size)!=0) perror("munmap");
|
||||
struct tpacket_req3 disable;
|
||||
memset(&disable,0,sizeof(disable));
|
||||
if(setsockopt(fd,SOL_PACKET,PACKET_RX_RING,&disable,sizeof(disable))!=0)
|
||||
fprintf(stderr,"warning: disabling RX ring failed: %s\n",strerror(errno));
|
||||
return result;
|
||||
}
|
||||
|
||||
static int receive_recvfrom(int fd, struct receiver_context *ctx, double start_time) {
|
||||
uint8_t buffer[65536];
|
||||
struct pollfd pfd={.fd=fd,.events=POLLIN,.revents=0};
|
||||
while(!stop_requested) {
|
||||
if(ctx->opt->duration>0.0 && monotonic_seconds()-start_time>=ctx->opt->duration) break;
|
||||
int pr=poll(&pfd,1,250);
|
||||
if(pr<0) { if(errno==EINTR) continue; perror("poll"); return -1; }
|
||||
if(pr==0) continue;
|
||||
ssize_t length=recvfrom(fd,buffer,sizeof(buffer),0,NULL,NULL);
|
||||
if(length<0) { if(errno==EINTR) continue; perror("recvfrom"); return -1; }
|
||||
int rc=process_frame(ctx,buffer,(size_t)length);
|
||||
if(rc!=0) return rc<0?-1:0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void collect_socket_stats(int fd, struct stats *s) {
|
||||
struct tpacket_stats_v3 ps;
|
||||
socklen_t len=sizeof(ps);
|
||||
memset(&ps,0,sizeof(ps));
|
||||
if(getsockopt(fd,SOL_PACKET,PACKET_STATISTICS,&ps,&len)==0) {
|
||||
s->socket_packets=ps.tp_packets;
|
||||
s->socket_drops=ps.tp_drops;
|
||||
s->freeze_count=ps.tp_freeze_q_cnt;
|
||||
}
|
||||
}
|
||||
|
||||
static int write_log(const char *path, const struct receiver_context *ctx,
|
||||
const char *start_text, double elapsed) {
|
||||
FILE *log=fopen(path,"w");
|
||||
if(log==NULL) { fprintf(stderr,"open log %s: %s\n",path,strerror(errno)); return -1; }
|
||||
char mac[18]="unknown";
|
||||
if(ctx->stats.have_source) format_mac(mac,ctx->stats.observed_source);
|
||||
const char *mode=ctx->opt->gap_mode==GAP_STOP?"stop":ctx->opt->gap_mode==GAP_SKIP?"skip":"zero-fill";
|
||||
fprintf(log,"start_time: %s\ninterface: %s\nsource_mac: %s\nprotocol_version: %u\n",
|
||||
start_text,ctx->opt->interface_name?ctx->opt->interface_name:"self-test",mac,ctx->stats.protocol_version);
|
||||
fprintf(log,"gap_mode: %s\nfirst_sample_index: %llu\nlast_sample_index: %llu\n",
|
||||
mode,(unsigned long long)ctx->stats.first_sample_index,(unsigned long long)ctx->stats.last_sample_index);
|
||||
fprintf(log,"received_frames: %llu\nmissing_frames: %llu\nout_of_order_frames: %llu\n",
|
||||
(unsigned long long)ctx->stats.received_frames,(unsigned long long)ctx->stats.missing_frames,
|
||||
(unsigned long long)ctx->stats.out_of_order_frames);
|
||||
fprintf(log,"fpga_dropped_samples: %llu\nhost_observed_missing_samples: %llu\ninvalid_frames: %llu\n",
|
||||
(unsigned long long)ctx->stats.fpga_dropped_samples,
|
||||
(unsigned long long)ctx->stats.host_missing_samples,
|
||||
(unsigned long long)ctx->stats.invalid_frames);
|
||||
fprintf(log,"bytes_written: %llu\nsamples_written: %llu\nelapsed_seconds: %.6f\naverage_sample_rate: %.3f\n",
|
||||
(unsigned long long)ctx->writer.bytes_written,(unsigned long long)ctx->writer.samples_written,
|
||||
elapsed,elapsed>0.0?(double)ctx->writer.samples_written/elapsed:0.0);
|
||||
fprintf(log,"socket_packets: %llu\nsocket_drops: %llu\nsocket_freeze_count: %llu\n",
|
||||
(unsigned long long)ctx->stats.socket_packets,(unsigned long long)ctx->stats.socket_drops,
|
||||
(unsigned long long)ctx->stats.freeze_count);
|
||||
if(fclose(log)!=0) { perror("close log"); return -1; }
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void put_be32(uint8_t *p,uint32_t v) { p[0]=(uint8_t)(v>>24);p[1]=(uint8_t)(v>>16);p[2]=(uint8_t)(v>>8);p[3]=(uint8_t)v; }
|
||||
static void put_be64(uint8_t *p,uint64_t v) { put_be32(p,(uint32_t)(v>>32));put_be32(p+4,(uint32_t)v); }
|
||||
|
||||
static int parser_self_test(struct options *opt) {
|
||||
uint8_t frame[ETH_HLEN+MX27_HEADER_LEN+2u];
|
||||
memset(frame,0,sizeof(frame));
|
||||
memset(frame,0xff,6); frame[6]=0x02;frame[7]=0x27;frame[8]=0x71;frame[11]=0x01;
|
||||
frame[12]=0x88;frame[13]=0xb5;
|
||||
uint8_t *h=frame+ETH_HLEN;
|
||||
memcpy(h,"MX27",4);h[4]=1;h[6]=0;h[7]=32;h[20]=0;h[21]=4;h[22]=4;h[23]=1;
|
||||
h[32]=0x01;h[33]=0x23;
|
||||
struct receiver_context ctx;
|
||||
memset(&ctx,0,sizeof(ctx));ctx.opt=opt;ctx.writer.file=tmpfile();
|
||||
if(ctx.writer.file==NULL) { perror("tmpfile"); return -1; }
|
||||
opt->gap_mode=GAP_SKIP;opt->quiet=true;
|
||||
put_be32(h+8,10);put_be64(h+12,100);
|
||||
if(process_frame(&ctx,frame,sizeof(frame))!=0) return -1;
|
||||
put_be32(h+8,12);put_be64(h+12,108);h[5]=1u;put_be32(h+24,4u);
|
||||
if(process_frame(&ctx,frame,sizeof(frame))!=0) return -1;
|
||||
int ok=(ctx.stats.received_frames==2u && ctx.stats.missing_frames==1u &&
|
||||
ctx.stats.host_missing_samples==4u && ctx.stats.fpga_dropped_samples==4u &&
|
||||
ctx.writer.samples_written==8u);
|
||||
(void)fclose(ctx.writer.file);
|
||||
if(!ok) { fprintf(stderr,"parser self-test failed\n"); return -1; }
|
||||
puts("PASS host parser self-test: packet and sample gaps detected");
|
||||
return 0;
|
||||
}
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
struct options opt;
|
||||
if(parse_options(argc,argv,&opt)!=0) { usage(stderr,argv[0]); return EXIT_FAILURE; }
|
||||
if(opt.self_test) return parser_self_test(&opt)==0?EXIT_SUCCESS:EXIT_FAILURE;
|
||||
|
||||
char *default_log=NULL;
|
||||
if(opt.log_path==NULL) {
|
||||
size_t n=strlen(opt.output_path)+5u;
|
||||
default_log=malloc(n);
|
||||
if(default_log==NULL) { perror("malloc"); return EXIT_FAILURE; }
|
||||
(void)snprintf(default_log,n,"%s.log",opt.output_path);
|
||||
opt.log_path=default_log;
|
||||
}
|
||||
struct receiver_context ctx;
|
||||
memset(&ctx,0,sizeof(ctx));ctx.opt=&opt;
|
||||
ctx.writer.file=fopen(opt.output_path,"wb");
|
||||
if(ctx.writer.file==NULL) { fprintf(stderr,"open output %s: %s\n",opt.output_path,strerror(errno));free(default_log);return EXIT_FAILURE; }
|
||||
char *output_buffer=malloc(OUTPUT_BUFFER_SIZE);
|
||||
if(output_buffer==NULL || setvbuf(ctx.writer.file,output_buffer,_IOFBF,OUTPUT_BUFFER_SIZE)!=0) {
|
||||
fprintf(stderr,"failed to allocate/configure output buffer\n");
|
||||
free(output_buffer);fclose(ctx.writer.file);free(default_log);return EXIT_FAILURE;
|
||||
}
|
||||
if(opt.max_bytes!=0u) ctx.writer.sample_limit=opt.max_bytes>UINT64_MAX/2u?UINT64_MAX:opt.max_bytes*2u;
|
||||
int fd=open_bound_socket(opt.interface_name);
|
||||
if(fd<0) { fclose(ctx.writer.file);free(output_buffer);free(default_log);return EXIT_FAILURE; }
|
||||
|
||||
struct sigaction action;
|
||||
memset(&action,0,sizeof(action));action.sa_handler=signal_handler;sigemptyset(&action.sa_mask);
|
||||
if(sigaction(SIGINT,&action,NULL)!=0 || sigaction(SIGTERM,&action,NULL)!=0) {
|
||||
perror("sigaction");close(fd);fclose(ctx.writer.file);free(output_buffer);free(default_log);return EXIT_FAILURE;
|
||||
}
|
||||
time_t wall_start=time(NULL);
|
||||
struct tm tm_start;
|
||||
char start_text[64];
|
||||
if(localtime_r(&wall_start,&tm_start)==NULL) strcpy(start_text,"unknown");
|
||||
else (void)strftime(start_text,sizeof(start_text),"%Y-%m-%dT%H:%M:%S%z",&tm_start);
|
||||
double start=monotonic_seconds();
|
||||
if(!opt.quiet) fprintf(stderr,"capturing EtherType 0x88B5 on %s -> %s\n",opt.interface_name,opt.output_path);
|
||||
int rc=receive_mmap(fd,&ctx,start);
|
||||
if(rc==-2) {
|
||||
if(!opt.quiet) fprintf(stderr,"TPACKET_V3 unavailable; falling back to recvfrom()\n");
|
||||
rc=receive_recvfrom(fd,&ctx,start);
|
||||
}
|
||||
double elapsed=monotonic_seconds()-start;
|
||||
collect_socket_stats(fd,&ctx.stats);
|
||||
if(finish_writer(&ctx.writer)!=0) { perror("flush output");rc=-1; }
|
||||
if(fclose(ctx.writer.file)!=0) { perror("close output");rc=-1; }
|
||||
close(fd);
|
||||
if(write_log(opt.log_path,&ctx,start_text,elapsed)!=0) rc=-1;
|
||||
if(!opt.quiet) {
|
||||
fprintf(stderr,"frames=%llu missing_frames=%llu missing_samples=%llu fpga_drops=%llu bytes=%llu elapsed=%.3fs socket_drops=%llu\n",
|
||||
(unsigned long long)ctx.stats.received_frames,(unsigned long long)ctx.stats.missing_frames,
|
||||
(unsigned long long)ctx.stats.host_missing_samples,(unsigned long long)ctx.stats.fpga_dropped_samples,
|
||||
(unsigned long long)ctx.writer.bytes_written,elapsed,(unsigned long long)ctx.stats.socket_drops);
|
||||
}
|
||||
free(output_buffer);free(default_log);
|
||||
return rc==0?EXIT_SUCCESS:EXIT_FAILURE;
|
||||
}
|
||||
Reference in New Issue
Block a user