X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_recv.c;h=3d680814c163816cd8dd8cdbe9f579a955c5b4fc;hp=9ea35d8d53ed2db68a54bfd1bc5527866d4f3061;hb=5f303cc5a96bfeaa66d2d6e899bf56d1d03ed085;hpb=6ba0a5055cfdec25d54d90a56f9345d5d4be09be diff --git a/udp_recv.c b/udp_recv.c index 9ea35d8d..3d680814 100644 --- a/udp_recv.c +++ b/udp_recv.c @@ -5,55 +5,48 @@ */ /** \file udp_recv.c Paraslash's udp receiver */ +#include #include +#include +#include #include "para.h" #include "error.h" #include "portable_io.h" -#include "udp_header.h" #include "list.h" #include "sched.h" #include "ggo.h" #include "recv.h" #include "udp_recv.cmdline.h" -#include "audiod.h" #include "string.h" #include "net.h" #include "fd.h" +#include "buffer_tree.h" /** The size of the receiver node buffer. */ #define UDP_RECV_CHUNK_SIZE (128 * 1024) - /** * Data specific to the udp receiver. * * \sa \ref receiver, \ref receiver_node. */ struct private_udp_recv_data { - /** - * Whether a header was received. - * - * A flag indicating whether this receiver already received a packet - * which contains the audio file header. - * - * This flag has no effect if the audio stream indicates that no extra - * headers will be sent (mp3, aac). Otherwise, all data packets are - * dropped until the header is received. - */ - int have_header; /** The socket file descriptor. */ int fd; - /** Non-zero on short reads. */ - uint16_t need_more; - /** Copied from the first audio header received. */ - uint16_t stream_type; + struct btr_pool *btrp; }; static void udp_recv_pre_select(struct sched *s, struct task *t) { struct receiver_node *rn = container_of(t, struct receiver_node, task); struct private_udp_recv_data *purd = rn->private_data; + int ret; + if (rn->btrn) { + ret = generic_recv_pre_select(s, t); + if (ret <= 0) + return; + } para_fd_set(purd->fd, &s->rfds, &s->max_fileno); } @@ -62,53 +55,6 @@ static int enough_space(size_t nbytes, size_t loaded) return nbytes + loaded < UDP_RECV_CHUNK_SIZE; } -/* - * Perform some sanity checks on an udp audio file header. - * - * return: negative on error, 0: discard data, 1: use data - */ -static int examine_audio_header(struct private_udp_recv_data *purd, - struct udp_audio_header *uah, size_t packet_size) -{ - /* payload_len includes header */ - if (uah->payload_len < uah->header_len) - return -E_UDP_BAD_HEADER; - switch (uah->packet_type) { - case UDP_EOF_PACKET: - return -E_RECV_EOF; - case UDP_BOF_PACKET: - purd->have_header = 1; - /* fall through */ - case UDP_DATA_PACKET: - if (uah->header_len) /* header in no-header packet */ - return -E_UDP_BAD_HEADER; - break; - case UDP_HEADER_PACKET: - if (!uah->header_len) /** no header in header packet */ - return -E_UDP_BAD_HEADER; - break; - default: /* bad packet type */ - return -E_UDP_BAD_HEADER; - } - /* check stream type */ - if (uah->stream_type != UDP_PLAIN_STREAM && - uah->stream_type != UDP_HEADER_STREAM) - return -E_UDP_BAD_STREAM_TYPE; - if (purd->stream_type == UDP_UNKNOWN_STREAM) - purd->stream_type = uah->stream_type; - /* stream type must not change */ - if (uah->stream_type != purd->stream_type) - return -E_UDP_BAD_STREAM_TYPE; - if (!purd->have_header && uah->stream_type == UDP_HEADER_STREAM) - /* can't use the data, wait for header packet */ - return 0; - if (packet_size < uah->payload_len + UDP_AUDIO_HEADER_LEN) - /* we read only a part of the package */ - purd->need_more = uah->payload_len - + UDP_AUDIO_HEADER_LEN - packet_size; - return 1; -} - static int add_rn_output(struct receiver_node *rn, char *buf, size_t len) { if (!len) @@ -120,16 +66,13 @@ static int add_rn_output(struct receiver_node *rn, char *buf, size_t len) return 1; } -static void udp_recv_post_select(__a_unused struct sched *s, struct task *t) +static void udp_recv_post_select_nobtr(__a_unused struct sched *s, struct task *t) { struct receiver_node *rn = container_of(t, struct receiver_node, task); struct private_udp_recv_data *purd = rn->private_data; int ret; char tmpbuf[UDP_RECV_CHUNK_SIZE]; - uint16_t data_len; - char *data_buf; size_t packet_size; - struct udp_audio_header uah; if (rn->output_error && *rn->output_error < 0) { t->error = *rn->output_error; @@ -148,45 +91,65 @@ static void udp_recv_post_select(__a_unused struct sched *s, struct task *t) if (!ret) return; packet_size = ret; - for (;;) { - uint16_t num; - - if (!purd->need_more) { - ret = read_udp_audio_header(tmpbuf, packet_size, &uah); - if (ret >= 0) - break; - goto success; /* drop data */ - } - num = PARA_MIN(purd->need_more, (uint16_t)packet_size); - assert(num > 0); - t->error = add_rn_output(rn, tmpbuf, num); - if (t->error < 0) + if (packet_size >= FEC_EOF_PACKET_LEN) + if (!memcmp(tmpbuf, FEC_EOF_PACKET, FEC_EOF_PACKET_LEN)) return; - purd->need_more -= num; - if (packet_size <= num) - goto success; - packet_size -= num; - memmove(tmpbuf, tmpbuf + num, packet_size); - } - assert(!purd->need_more); - t->error = examine_audio_header(purd, &uah, packet_size); - if (t->error <= 0) + t->error = add_rn_output(rn, tmpbuf, packet_size); + if (t->error < 0) + return; +success: + t->error = 1; +} + +static void udp_recv_post_select_btr(__a_unused struct sched *s, struct task *t) +{ + struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct private_udp_recv_data *purd = rn->private_data; + struct btr_node *btrn = rn->btrn; + int ret; + char *buf = NULL; + size_t packet_size, bufsize; + + t->error = 0; + ret = btr_node_status(btrn, 0, BTR_NT_ROOT); + if (ret < 0) + goto err; + if (ret == 0) + return; + if (!FD_ISSET(purd->fd, &s->rfds)) return; - data_len = uah.payload_len; - data_buf = tmpbuf + UDP_AUDIO_HEADER_LEN; - if (uah.packet_type == UDP_HEADER_PACKET) { - if (purd->have_header) { /* skip header */ - data_buf += uah.header_len; - data_len -= uah.header_len; - } else { /* only use the header */ - purd->have_header = 1; - data_len = uah.header_len; + bufsize = btr_pool_get_buffer(purd->btrp, &buf); + ret = -E_UDP_OVERRUN; + if (bufsize == 0) + goto err; + ret = recv_bin_buffer(purd->fd, buf, bufsize); + if (ret == 0) + ret = -E_RECV_EOF; + if (ret < 0) + goto err; + packet_size = ret; + if (packet_size >= FEC_EOF_PACKET_LEN) { + if (!memcmp(buf, FEC_EOF_PACKET, FEC_EOF_PACKET_LEN)) { + PARA_INFO_LOG("received eof packet\n"); + ret = -E_RECV_EOF; + goto err; } } - t->error = add_rn_output(rn, data_buf, data_len); + btr_add_output_pool(purd->btrp, packet_size, btrn); return; -success: - t->error = 1; +err: + btr_remove_node(btrn); + t->error = ret; + close(purd->fd); + purd->fd = -1; +} + +static void udp_recv_post_select(__a_unused struct sched *s, struct task *t) +{ + struct receiver_node *rn = container_of(t, struct receiver_node, task); + if (rn->btrn) + return udp_recv_post_select_btr(s, t); + udp_recv_post_select_nobtr(s, t); } static void udp_shutdown(void) @@ -200,6 +163,8 @@ static void udp_recv_close(struct receiver_node *rn) if (purd->fd >= 0) close(purd->fd); + PARA_CRIT_LOG("%p: close\n", rn); + btr_pool_free(purd->btrp); free(rn->private_data); free(rn->buf); } @@ -301,11 +266,13 @@ static int udp_recv_open(struct receiver_node *rn) } ret = mark_fd_nonblocking(purd->fd); - if (ret < 0) + if (ret < 0) { + close(purd->fd); goto err; - purd->stream_type = UDP_UNKNOWN_STREAM; - PARA_NOTICE_LOG("receiving from %s:%d, fd=%d\n", c->host_arg, + } + PARA_CRIT_LOG("rn %p: receiving from %s:%d, fd=%d\n", rn, c->host_arg, c->port_arg, purd->fd); + purd->btrp = btr_pool_new("udp_recv", 320 * 1024); return purd->fd; err: free(rn->private_data); @@ -313,6 +280,11 @@ err: return ret; } +static void udp_recv_free_config(void *conf) +{ + udp_recv_cmdline_parser_free(conf); +} + /** * The init function of the udp receiver. * @@ -331,8 +303,10 @@ void udp_recv_init(struct receiver *r) r->pre_select = udp_recv_pre_select; r->post_select = udp_recv_post_select; r->parse_config = udp_recv_parse_config; + r->free_config = udp_recv_free_config; r->help = (struct ggo_help) { .short_help = udp_recv_args_info_help, .detailed_help = udp_recv_args_info_detailed_help }; + udp_recv_cmdline_parser_free(&dummy); }