From 8d8040f5a653c75ee4e8e9f7dbb06f71ae087160 Mon Sep 17 00:00:00 2001 From: Mark Hills Date: Fri, 4 Sep 2009 23:04:44 +0100 Subject: [PATCH] Initial import --- .gitignore | 6 ++ Makefile | 24 +++++ defaults.h | 14 +++ device.c | 78 +++++++++++++++ device.h | 10 ++ jitter.c | 130 +++++++++++++++++++++++++ jitter.h | 34 +++++++ rx.c | 263 ++++++++++++++++++++++++++++++++++++++++++++++++++ sched.c | 32 ++++++ sched.h | 6 ++ test-jitter.c | 34 +++++++ tx.c | 235 ++++++++++++++++++++++++++++++++++++++++++++ 12 files changed, 866 insertions(+) create mode 100644 .gitignore create mode 100644 Makefile create mode 100644 defaults.h create mode 100644 device.c create mode 100644 device.h create mode 100644 jitter.c create mode 100644 jitter.h create mode 100644 rx.c create mode 100644 sched.c create mode 100644 sched.h create mode 100644 test-jitter.c create mode 100644 tx.c diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d111759 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.config +*.o +*.d +tx +rx +test-jitter diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..8447902 --- /dev/null +++ b/Makefile @@ -0,0 +1,24 @@ +-include .config + +CFLAGS+=-MMD -Wall + +.PHONY: all clean + +all: rx tx + +rx: rx.o device.o jitter.o sched.o +rx: LDLIBS+=-lcelt -lasound + +tx: tx.o device.o sched.o +tx: LDLIBS+=-lcelt -lasound + +test-jitter: test-jitter.o jitter.o + +clean: + rm -f *.o + rm -f *.d + rm -f tx + rm -f rx + rm -f test-jitter + +-include *.d diff --git a/defaults.h b/defaults.h new file mode 100644 index 0000000..9c30215 --- /dev/null +++ b/defaults.h @@ -0,0 +1,14 @@ +#ifndef DEFAULTS_H +#define DEFAULTS_H + +#define DEFAULT_DEVICE "default" +#define DEFAULT_BUFFER 16 + +#define DEFAULT_SERVICE "1350" +#define DEFAULT_FRAME 96 + +#define DEFAULT_RATE 44100 +#define DEFAULT_CHANNELS 2 +#define DEFAULT_BITRATE 128 + +#endif diff --git a/device.c b/device.c new file mode 100644 index 0000000..4dc8ad4 --- /dev/null +++ b/device.c @@ -0,0 +1,78 @@ +#include +#include + +#define CHK(call, r) { \ + if (r < 0) { \ + aerror(call, r); \ + return -1; \ + } \ +} + +void aerror(const char *msg, int r) +{ + fputs(msg, stderr); + fputs(": ", stderr); + fputs(snd_strerror(r), stderr); + fputc('\n', stderr); +} + +int set_alsa_hw(snd_pcm_t *pcm, + unsigned int rate, unsigned int channels, + unsigned int buffer) +{ + int r, dir; + snd_pcm_hw_params_t *hw; + + snd_pcm_hw_params_alloca(&hw); + + r = snd_pcm_hw_params_any(pcm, hw); + CHK("snd_pcm_hw_params_any", r); + + r = snd_pcm_hw_params_set_rate_resample(pcm, hw, 1); + CHK("snd_pcm_hw_params_set_rate_resample", r); + + r = snd_pcm_hw_params_set_access(pcm, hw, SND_PCM_ACCESS_RW_INTERLEAVED); + CHK("snd_pcm_hw_params_set_access", r); + + r = snd_pcm_hw_params_set_format(pcm, hw, SND_PCM_FORMAT_FLOAT); + CHK("snd_pcm_hw_params_set_format", r); + + r = snd_pcm_hw_params_set_rate(pcm, hw, rate, 0); + CHK("snd_pcm_hw_params_set_rate", r); + + r = snd_pcm_hw_params_set_channels(pcm, hw, channels); + CHK("snd_pcm_hw_params_set_channels", r); + + dir = 0; + r = snd_pcm_hw_params_set_buffer_time_near(pcm, hw, &buffer, &dir); + CHK("snd_pcm_hw_params_set_buffer_time_near", r); + + r = snd_pcm_hw_params(pcm, hw); + CHK("hw_params", r); + + return 0; +} + +int set_alsa_sw(snd_pcm_t *pcm) +{ + int r; + snd_pcm_sw_params_t *sw; + snd_pcm_uframes_t boundary; + + snd_pcm_sw_params_alloca(&sw); + + r = snd_pcm_sw_params_current(pcm, sw); + CHK("snd_pcm_sw_params_current", r); + + r = snd_pcm_sw_params_get_boundary(sw, &boundary); + CHK("snd_pcm_sw_params_get_boundary", r); + + r = snd_pcm_sw_params_set_stop_threshold(pcm, sw, boundary); + CHK("snd_pcm_sw_params_set_stop_threshold", r); + + r = snd_pcm_sw_params(pcm, sw); + CHK("snd_pcm_sw_params", r); + + return 0; + +} diff --git a/device.h b/device.h new file mode 100644 index 0000000..d6846c5 --- /dev/null +++ b/device.h @@ -0,0 +1,10 @@ +#ifndef DEVICE_H +#define DEVICE_H + +void aerror(const char *msg, int r); +int set_alsa_hw(snd_pcm_t *pcm, + unsigned int rate, unsigned int channels, + unsigned int buffer); +int set_alsa_sw(snd_pcm_t *pcm); + +#endif diff --git a/jitter.c b/jitter.c new file mode 100644 index 0000000..e9dea1d --- /dev/null +++ b/jitter.c @@ -0,0 +1,130 @@ +#include +#include + +#include "jitter.h" + +void jitbuf_init(struct jitbuf_t *jb) +{ + size_t n; + + for (n = 0; n < JITBUF_MAX; n++) + jb->buf[n] = NULL; + + jb->base = 0; + jb->out = 0; + jb->entries = 0; +} + +void jitbuf_clear(struct jitbuf_t *jb) +{ +} + +/* + * Push an entry into the buffer, with the given sequence number + * + * Pre: data is not NULL + */ + +int jitbuf_push(struct jitbuf_t *jb, seq_t t, void *data) +{ + void **entry; + + assert(data != NULL); + + if (jb->entries == 0 && t != jb->base) { + fprintf(stderr, "Restarting (%d)\n", t); + jb->base = t; + } else { + if (t < jb->base) { + fprintf(stderr, "Late (%d)\n", t); + return -1; + } + if (t >= jb->base + JITBUF_MAX) { + fprintf(stderr, "Early (%d)\n", t); + return -1; + } + } + + entry = &jb->buf[(t - jb->base + jb->out) % JITBUF_MAX]; + + if (*entry != NULL) { + fprintf(stderr, "Dup (%d)\n", t); + return -1; + } + + *entry = data; + jb->entries++; + + return 0; +} + +/* + * Get the entry at the front of the buffer, in sequence number order + * + * Return: NULL if no entry, otherwise entry next in sequence + */ + +void* jitbuf_front(const struct jitbuf_t *jb) +{ + return jb->buf[jb->out]; +} + +/* + * Pop the next entry in the buffer + */ + +void jitbuf_pop(struct jitbuf_t *jb) +{ + void **entry; + + entry = &jb->buf[jb->out]; + if (*entry != NULL) { + jb->entries--; + *entry = NULL; + } + jb->base++; + jb->out++; + if (jb->out >= JITBUF_MAX) + jb->out = 0; +} + +/* + * Return true if this buffer has enough data flow to be declared ready + */ + +bool jitbuf_ready(const struct jitbuf_t *jb, seq_t count) +{ + size_t n; + + assert(count <= JITBUF_MAX); + + for (n = 0; n < count; n++) { + if (jb->buf[(jb->out + n) % JITBUF_MAX] == NULL) + return false; + } + return true; +} + +bool jitbuf_empty(const struct jitbuf_t *jb) +{ + if (jb->entries > 0) + return false; + else + return true; +} + +void jitbuf_debug(struct jitbuf_t *jb, FILE *fd) +{ + seq_t n; + + fprintf(fd, "% 3d ", jb->entries); + + for (n = 0; n < JITBUF_MAX; n++) { + if (jb->buf[(jb->out + n) % JITBUF_MAX] != NULL) { + fputc('#', fd); + } else { + fputc(' ', fd); + } + } + fputc('\n', fd); +} diff --git a/jitter.h b/jitter.h new file mode 100644 index 0000000..a42cd1a --- /dev/null +++ b/jitter.h @@ -0,0 +1,34 @@ +#ifndef JITTER_H +#define JITTER_H + +#include +#include +#include + +#define JITBUF_MAX 64 + +typedef unsigned int seq_t; + +/* + * De-jitter buffer + */ + +struct jitbuf_t { + seq_t base; + size_t out, entries, contiguous; + void *buf[JITBUF_MAX]; +}; + +void jitbuf_init(struct jitbuf_t *jb); +void jitbuf_clear(struct jitbuf_t *jb); + +int jitbuf_push(struct jitbuf_t *jb, seq_t t, void *data); +void* jitbuf_front(const struct jitbuf_t *jb); +void jitbuf_pop(struct jitbuf_t *jb); + +bool jitbuf_ready(const struct jitbuf_t *jb, seq_t count); +bool jitbuf_empty(const struct jitbuf_t *jb); + +void jitbuf_debug(struct jitbuf_t *jb, FILE *fd); + +#endif diff --git a/rx.c b/rx.c new file mode 100644 index 0000000..0d9c6d0 --- /dev/null +++ b/rx.c @@ -0,0 +1,263 @@ +#include +#include +#include +#include +#include +#include + +#include "defaults.h" +#include "device.h" +#include "jitter.h" +#include "sched.h" + +#define ENCODED_BYTES 128 + +/* + * Bind to a network socket for receiving packets on the given port + */ + +static int listen_on_network(const char *service) +{ + int r, sd; + struct addrinfo hints, *servinfo, *p; + + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_DGRAM; + hints.ai_flags = AI_PASSIVE; /* use my IP */ + + r = getaddrinfo(NULL, service, &hints, &servinfo); + if (r != 0) { + fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(r)); + return -1; + } + + /* Bind to the first one we can */ + + for (p = servinfo; p != NULL; p = p->ai_next) { + sd = socket(p->ai_family, p->ai_socktype | SOCK_NONBLOCK, + p->ai_protocol); + if (sd == -1) { + perror("socket"); + continue; + } + + if (bind(sd, p->ai_addr, p->ai_addrlen) == -1) { + if (close(sd) == -1) + abort(); + perror("bind"); + continue; + } + + break; + } + + freeaddrinfo(servinfo); + + if (p == NULL) { + fputs("Failed to bind socket\n", stderr); + return -1; + } else { + return sd; + } +} + +/* + * Playback audio from the buffer to the audio device + */ + +static int play_audio(struct jitbuf_t *jb, CELTDecoder *decoder, + snd_pcm_t *snd) +{ + int r; + void *data; + float *pcm; + snd_pcm_sframes_t f; + + pcm = alloca(sizeof(float) * DEFAULT_FRAME * DEFAULT_CHANNELS); + + data = jitbuf_front(jb); + if (data == NULL) { + r = celt_decode_float(decoder, NULL, 0, pcm); + } else { + r = celt_decode_float(decoder, data, ENCODED_BYTES, pcm); + } + if (r != 0) { + fputs("Error in celt_decode\n", stderr); + return -1; + } + + f = snd_pcm_writei(snd, pcm, DEFAULT_FRAME); + if (f < 0) { + aerror("snd_pcm_writei", f); + return -1; + } + if (f < DEFAULT_FRAME) + fprintf(stderr, "Short write %ld\n", f); + + jitbuf_pop(jb); + free(data); + + return 0; +} + +/* + * Read from the network into the buffer + */ + +static int read_from_network(int sd, struct jitbuf_t *jb) +{ + size_t z; + char buf[32768]; + void *data; + seq_t seq; + + z = recvfrom(sd, buf, sizeof(buf), 0, NULL, NULL); + if (z == -1) { + if (errno == EAGAIN) + return -EAGAIN; + perror("recvfrom"); + return -1; + } + if (z != sizeof(uint32_t) + ENCODED_BYTES) { + fputs("small packet", stderr); + return -1; + } + + /* First 32-bits is sequence number, rest is data */ + seq = ntohl(*(uint32_t*)buf); + z -= sizeof(uint32_t); + assert(z >= 0); + data = malloc(z); + memcpy(data, buf + sizeof(uint32_t), z); + + if (jitbuf_push(jb, seq, data) == -1) + free(data); + + return 0; +} + +/* + * The main loop of receiving audio and playing it back + */ + +static int run_rx(int sd, CELTDecoder *decoder, snd_pcm_t *snd) +{ + int r; + struct jitbuf_t jb; + + jitbuf_init(&jb); + + for (;;) { + for (;;) { + r = read_from_network(sd, &jb); + if (r == -EAGAIN) + break; + if (r != 0) + return -1; + } + + jitbuf_debug(&jb, stderr); + + if (play_audio(&jb, decoder, snd) == -1) + return -1; + } + + jitbuf_clear(&jb); +} + +static void usage(FILE *fd) +{ + fprintf(fd, "Usage: rx []\n"); + + fprintf(fd, "\nAudio device (ALSA) parameters:\n"); + fprintf(fd, " -d Device name (default '%s')\n", + DEFAULT_DEVICE); + fprintf(fd, " -m Buffer time (milliseconds, default %d)\n", + DEFAULT_BUFFER); + + fprintf(fd, "\nNetwork parameters:\n"); + fprintf(fd, " -p UDP port number or name (default %s)\n", + DEFAULT_SERVICE); +} + +int main(int argc, char *argv[]) +{ + int sd, r; + snd_pcm_t *snd; + CELTMode *mode; + CELTDecoder *decoder; + + /* command-line options */ + const char *device = DEFAULT_DEVICE, + *service = DEFAULT_SERVICE; + unsigned int buffer = DEFAULT_BUFFER; + + for (;;) { + int c; + + c = getopt(argc, argv, "d:m:p:"); + if (c == -1) + break; + + switch (c) { + case 'd': + device = optarg; + break; + case 'm': + buffer = atoi(optarg); + break; + case 'p': + service = optarg; + break; + default: + usage(stderr); + return -1; + } + } + + mode = celt_mode_create(DEFAULT_RATE, DEFAULT_CHANNELS, DEFAULT_FRAME, + NULL); + if (mode == NULL) { + fputs("celt_mode_create failed\n", stderr); + return -1; + } + decoder = celt_decoder_create(mode); + if (decoder == NULL) { + fputs("celt_decoder_create failed\n", stderr); + return -1; + } + + sd = listen_on_network(service); + if (sd == -1) + return -1; + + r = snd_pcm_open(&snd, device, SND_PCM_STREAM_PLAYBACK, 0); + if (r < 0) { + aerror("snd_pcm_open", r); + return -1; + } + if (set_alsa_hw(snd, DEFAULT_RATE, DEFAULT_CHANNELS, + buffer * 1000) == -1) + { + return -1; + } + if (set_alsa_sw(snd) == -1) + return -1; + + if (go_realtime() != 0) + return -1; + + r = run_rx(sd, decoder, snd); + + if (snd_pcm_close(snd) < 0) + abort(); + + celt_decoder_destroy(decoder); + celt_mode_destroy(mode); + + if (close(sd) == -1) + abort(); + + return r; +} diff --git a/sched.c b/sched.c new file mode 100644 index 0000000..3e4fdc3 --- /dev/null +++ b/sched.c @@ -0,0 +1,32 @@ +#include +#include + +#include "sched.h" + +#define REALTIME_PRIORITY 80 + +int go_realtime(void) +{ + int max_pri; + struct sched_param sp; + + if (sched_getparam(0, &sp)) { + perror("sched_getparam"); + return -1; + } + + max_pri = sched_get_priority_max(SCHED_FIFO); + sp.sched_priority = REALTIME_PRIORITY; + + if (sp.sched_priority > max_pri) { + fprintf(stderr, "Invalid priority (maximum %d)\n", max_pri); + return -1; + } + + if (sched_setscheduler(0, SCHED_FIFO, &sp)) { + perror("sched_setscheduler"); + return -1; + } + + return 0; +} diff --git a/sched.h b/sched.h new file mode 100644 index 0000000..82d309b --- /dev/null +++ b/sched.h @@ -0,0 +1,6 @@ +#ifndef MISC_H +#define MISC_H + +int go_realtime(void); + +#endif diff --git a/test-jitter.c b/test-jitter.c new file mode 100644 index 0000000..ab4adc4 --- /dev/null +++ b/test-jitter.c @@ -0,0 +1,34 @@ +#include +#include + +#include "jitter.h" + +int main(int argc, char *argv[]) +{ + int n; + struct jitbuf_t jb; + const char *test = "abcdefghijklmnopqrstuvwxyz1234567890"; + + jitbuf_init(&jb); + + for (n = 0; n < strlen(test); n++) { + int m; + char *out; + size_t pos; + + out = jitbuf_front(&jb); + if (out != NULL) + fputc(*out, stdout); + else + fputc('#', stdout); + jitbuf_pop(&jb); + + for (m = 0; m < 8; m++) { + pos = n + rand() % (strlen(test) - n); + jitbuf_push(&jb, pos, (void*)&test[pos]); + } + } + fputc('\n', stdout); + + jitbuf_clear(&jb); +} diff --git a/tx.c b/tx.c new file mode 100644 index 0000000..0c62592 --- /dev/null +++ b/tx.c @@ -0,0 +1,235 @@ +#include +#include +#include +#include +#include +#include + +#include "defaults.h" +#include "device.h" +#include "sched.h" + +#define ENCODED_BYTES 128 + +typedef int seq_t; + +/* + * Bind to network socket, for sending to the given host and port + */ + +static int bind_to_network(const char *host, const char *service, + struct addrinfo **servinfo, struct addrinfo **theirs) +{ + int r, sd; + struct addrinfo hints, *p; + + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_DGRAM; + + r = getaddrinfo(host, service, &hints, servinfo); + if (r != 0) { + fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(r)); + return -1; + } + + for (p = *servinfo; p != NULL; p = p->ai_next) { + sd = socket(p->ai_family, p->ai_socktype, p->ai_protocol); + if (sd == -1) { + perror("socket"); + continue; + } + break; + } + + if (p == NULL) { + fputs("Failed to bind socket\n", stderr); + return -1; + } else { + *theirs = p; + return sd; + } +} + +/* + * Encode a frame of audio and send it over the network + */ + +static int send_audio(snd_pcm_t *snd, CELTEncoder *encoder, + int sd, struct addrinfo *theirs, seq_t *seq) +{ + float *pcm; + void *packet, *encoded; + ssize_t z; + snd_pcm_sframes_t f; + + pcm = alloca(sizeof(float) * DEFAULT_FRAME * DEFAULT_CHANNELS); + packet = alloca(ENCODED_BYTES + sizeof(uint32_t)); + encoded = packet + sizeof(uint32_t); + + f = snd_pcm_readi(snd, pcm, DEFAULT_FRAME); + if (f < 0) { + aerror("snd_pcm_readi", f); + return -1; + } + if (f < DEFAULT_FRAME) + fprintf(stderr, "Short read, %ld\n", f); + + z = celt_encode_float(encoder, pcm, NULL, encoded, ENCODED_BYTES); + if (z < 0) { + fputs("celt_encode_float failed\n", stderr); + return -1; + } + + *(uint32_t*)packet = htonl(*seq); + (*seq)++; + + z = sendto(sd, packet, sizeof(uint32_t) + ENCODED_BYTES, 0, + theirs->ai_addr, theirs->ai_addrlen); + if (z == -1) { + perror("sendto"); + return -1; + } + fputc('>', stderr); + + return 0; +} + +static int run_tx(snd_pcm_t *snd, CELTEncoder *encoder, + int sd, struct addrinfo *theirs) +{ + seq_t seq = 0; + + for (;;) { + int r; + + r = send_audio(snd, encoder, sd, theirs, &seq); + if (r == -1) + return -1; + } +} + +static void usage(FILE *fd) +{ + fprintf(fd, "Usage: tx [] \n"); + + fprintf(fd, "\nAudio device (ALSA) parameters:\n"); + fprintf(fd, " -d Device name (default '%s')\n", + DEFAULT_DEVICE); + fprintf(fd, " -m Buffer time (milliseconds, default %d)\n", + DEFAULT_BUFFER); + + fprintf(fd, "\nNetwork parameters:\n"); + fprintf(fd, " -p UDP port number or name (default %s)\n", + DEFAULT_SERVICE); + fprintf(fd, " -f Frame size (default %d)\n", + DEFAULT_FRAME); + + fprintf(fd, "\nEncoding parameters:\n"); + fprintf(fd, " -r Sample rate (default %d)\n", + DEFAULT_RATE); + fprintf(fd, " -c Number of channels (default %d)\n", + DEFAULT_CHANNELS); + fprintf(fd, " -b Bitrate (kbps, approx., default %d)\n", + DEFAULT_BITRATE); +} + +int main(int argc, char *argv[]) +{ + int sd, r; + struct addrinfo *servinfo, *theirs; + snd_pcm_t *snd; + CELTMode *mode; + CELTEncoder *encoder; + + /* command-line options */ + const char *device = DEFAULT_DEVICE, + *service = DEFAULT_SERVICE, + *host; + unsigned int buffer = DEFAULT_BUFFER, + rate = DEFAULT_RATE, + channels = DEFAULT_CHANNELS; + size_t frame = DEFAULT_FRAME; + + for (;;) { + int c; + + c = getopt(argc, argv, "c:d:f:m:p:r:"); + if (c == -1) + break; + + switch (c) { + case 'c': + channels = atoi(optarg); + break; + case 'd': + device = optarg; + break; + case 'f': + frame = atol(optarg); + break; + case 'm': + buffer = atoi(optarg); + break; + case 'p': + service = optarg; + break; + case 'r': + rate = atoi(optarg); + break; + default: + usage(stderr); + return -1; + } + } + + if (argv[optind] == NULL) { + usage(stderr); + return -1; + } + + host = argv[optind]; + + mode = celt_mode_create(rate, channels, frame, NULL); + if (mode == NULL) { + fputs("celt_mode_create failed\n", stderr); + return -1; + } + encoder = celt_encoder_create(mode); + if (encoder == NULL) { + fputs("celt_encoder_create failed\n", stderr); + return -1; + } + + sd = bind_to_network(host, service, &servinfo, &theirs); + if (sd == -1) + return -1; + + r = snd_pcm_open(&snd, device, SND_PCM_STREAM_CAPTURE, 0); + if (r < 0) { + aerror("snd_pcm_open", r); + return -1; + } + if (set_alsa_hw(snd, rate, channels, buffer * 1000) == -1) + return -1; + if (set_alsa_sw(snd) == -1) + return -1; + + if (go_realtime() != 0) + return -1; + + r = run_tx(snd, encoder, sd, theirs); + + freeaddrinfo(servinfo); + + if (snd_pcm_close(snd) < 0) + abort(); + + celt_encoder_destroy(encoder); + celt_mode_destroy(mode); + + if (close(sd) == -1) + abort(); + + return r; +}