X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=499af66e2addfcce8991fb21f5a0e58de4de4d6f;hp=ca4d8cf291c60f38f64ed488716019c1635a75c7;hb=b6100d88b5a03e824991b4a5bfd90b0e951e8d2b;hpb=91baab1e9bbf957d139d64741db02be6490da45a;ds=inline diff --git a/udp_send.c b/udp_send.c index ca4d8cf2..499af66e 100644 --- a/udp_send.c +++ b/udp_send.c @@ -9,6 +9,9 @@ #include #include +#include +#include +#include #include "server.cmdline.h" #include "para.h" @@ -21,7 +24,6 @@ #include "list.h" #include "send.h" #include "portable_io.h" -#include "udp_header.h" #include "net.h" #include "fd.h" #include "sched.h" @@ -40,6 +42,10 @@ struct udp_target { int fd; /** The list of queued chunks for this fd. */ struct chunk_queue *cq; + /** The opaque structure returned by vss_add_fec_client(). */ + struct fec_client *fc; + /** The FEC parameters for this target. */ + struct fec_client_parms fcp; }; static struct list_head targets; @@ -61,6 +67,7 @@ static void udp_delete_target(struct udp_target *ut, const char *msg) PARA_NOTICE_LOG("deleting %s#%d (%s) from list\n", ut->host, ut->port, msg); udp_close_target(ut); + vss_del_fec_client(ut->fc); list_del(&ut->node); free(ut); } @@ -173,121 +180,28 @@ static int udp_init_session(struct udp_target *ut) } add_close_on_fork_list(ut->fd); ut->cq = cq_new(UDP_CQ_BYTES); - PARA_NOTICE_LOG("sending to udp %s#%d\n", ut->host, ut->port); + PARA_NOTICE_LOG("sending to udp %s#%d using fec parms %d:%d:%d\n", + ut->host, ut->port , ut->fcp.max_slice_bytes, + ut->fcp.data_slices_per_group, ut->fcp.slices_per_group); return 1; } -static void udp_send_buf(char *buf, size_t len) -{ - struct udp_target *ut, *tmp; - int ret; - - list_for_each_entry_safe(ut, tmp, &targets, node) { - ret = udp_init_session(ut); - if (ret < 0) { - udp_delete_target(ut, para_strerror(-ret)); - continue; - } - ret = send_queued_chunks(ut->fd, ut->cq, 0); - if (ret < 0) { - udp_delete_target(ut, para_strerror(-ret)); - continue; - } - if (!len) - continue; - if (!ret) { /* still data left in the queue */ - ret = cq_enqueue(ut->cq, buf, len); - if (ret < 0) { - udp_delete_target(ut, para_strerror(-ret)); - continue; - } - } - ret = write_nonblock(ut->fd, buf, len, 0); - if (ret < 0) { - udp_delete_target(ut, para_strerror(-ret)); - continue; - } - if (ret != len) { - ret = cq_enqueue(ut->cq, buf + ret, len - ret); - if (ret < 0) { - udp_delete_target(ut, para_strerror(-ret)); - continue; - } - } - } -} - static void udp_shutdown_targets(void) { - char buf[UDP_AUDIO_HEADER_LEN]; struct udp_target *ut, *tmp; - struct udp_audio_header uah = { - .stream_type = UDP_UNKNOWN_STREAM, - .packet_type = UDP_EOF_PACKET, - }; + const char *buf = NULL; + size_t len = 0; /* STFU, gcc */ - write_udp_audio_header(buf, &uah); list_for_each_entry_safe(ut, tmp, &targets, node) { if (ut->fd < 0) continue; - write(ut->fd, buf, UDP_AUDIO_HEADER_LEN); + if (!buf) + len = vss_get_fec_eof_packet(&buf); + write(ut->fd, buf, len); udp_close_target(ut); } } -static int need_extra_header(long unsigned current_chunk) -{ - static struct timeval last_header; - struct timeval diff; - - if (!current_chunk) - return 0; - tv_diff(now, &last_header, &diff); - if (tv2ms(&diff) < conf.udp_header_interval_arg) - return 0; - last_header = *now; - return 1; -} - -static void udp_send(long unsigned current_chunk, __a_unused long unsigned chunks_sent, - const char *buf, size_t len, const char *header_buf, - size_t header_len) -{ - char *sendbuf; - size_t sendbuf_len; - struct timeval *chunk_tv; - struct udp_audio_header uah; - -// PARA_NOTICE_LOG("len: %zd, header_len: %zd\n", len, header_len); - if (sender_status != SENDER_ON) - return; - - /* we might not yet know the chunk time */ - chunk_tv = vss_chunk_time(); - if (!chunk_tv) - return; - if (list_empty(&targets)) - return; - uah.stream_type = header_len? UDP_HEADER_STREAM : UDP_PLAIN_STREAM; - uah.header_len = need_extra_header(current_chunk)? header_len : 0; - if (!current_chunk) - uah.packet_type = UDP_BOF_PACKET; - else if (uah.header_len) - uah.packet_type = UDP_HEADER_PACKET; - else - uah.packet_type = UDP_DATA_PACKET; - uah.payload_len = uah.header_len + len; - sendbuf_len = UDP_AUDIO_HEADER_LEN + uah.payload_len; - sendbuf = para_malloc(sendbuf_len); - write_udp_audio_header(sendbuf, &uah); - if (uah.header_len) - memcpy(sendbuf + UDP_AUDIO_HEADER_LEN, header_buf, - uah.header_len); - memcpy(sendbuf + UDP_AUDIO_HEADER_LEN + uah.header_len, buf, len); - udp_send_buf(sendbuf, sendbuf_len); - free(sendbuf); -} - static int udp_com_on(__a_unused struct sender_command_data *scd) { sender_status = SENDER_ON; @@ -316,21 +230,57 @@ static int udp_com_delete(struct sender_command_data *scd) return 1; } -static void udp_add_target(const char *host, int port) +static int udp_send_fec(char *buf, size_t len, void *private_data) +{ + struct udp_target *ut = private_data; + int ret = udp_init_session(ut); + + if (ret < 0) + goto fail; + ret = send_queued_chunks(ut->fd, ut->cq, 0); + if (ret < 0) + goto fail; + if (!len) + return 0; + if (!ret) { /* still data left in the queue */ + ret = cq_enqueue(ut->cq, buf, len); + if (ret < 0) + goto fail; + } + ret = write_nonblock(ut->fd, buf, len, 0); + if (ret < 0) + goto fail; + if (ret != len) { + ret = cq_enqueue(ut->cq, buf + ret, len - ret); + if (ret < 0) + goto fail; + } + return 1; +fail: + udp_delete_target(ut, para_strerror(-ret)); + return ret; +} + +static void udp_add_target(struct sender_command_data *scd) { struct udp_target *ut = para_calloc(sizeof(struct udp_target)); - strncpy(ut->host, host, sizeof(ut->host)); - ut->port = port > 0 ? port : conf.udp_default_port_arg; + strncpy(ut->host, scd->host, sizeof(ut->host)); + ut->port = scd->port > 0 ? scd->port : conf.udp_default_port_arg; ut->fd = -1; /* not yet connected */ - PARA_INFO_LOG("adding to target list (%s#%d)\n", - ut->host, ut->port); + PARA_INFO_LOG("adding to target list (%s#%d)\n", ut->host, ut->port); para_list_add(&ut->node, &targets); + ut->fcp.slices_per_group = scd->slices_per_group; + ut->fcp.data_slices_per_group = scd->data_slices_per_group; + ut->fcp.max_slice_bytes = scd->max_slice_bytes; + ut->fcp.send = udp_send_fec; + ut->fcp.private_data = ut; + vss_add_fec_client(&ut->fcp, &ut->fc); } static int udp_com_add(struct sender_command_data *scd) { - udp_add_target(scd->host, scd->port); + udp_add_target(scd); return 1; } @@ -341,9 +291,13 @@ static char *udp_info(void) list_for_each_entry(ut, &targets, node) { bool is_v6 = strchr(ut->host, ':') != NULL; - char *tmp = make_message("%s%s%s%s:%d ", tgts ? : "", - is_v6 ? "[" : "", ut->host, - is_v6 ? "]" : "", ut->port); + char *tmp = make_message("%s%s%s%s:%d/%u:%u:%u ", tgts ? : "", + is_v6 ? "[" : "", ut->host, + is_v6 ? "]" : "", ut->port, + ut->fcp.max_slice_bytes, + ut->fcp.data_slices_per_group, + ut->fcp.slices_per_group + ); free(tgts); tgts = tmp; } @@ -362,25 +316,29 @@ static char *udp_info(void) static void udp_init_target_list(void) { - char host[MAX_HOSTLEN]; - int port, i; + struct sender_command_data scd; + int i; INIT_LIST_HEAD(&targets); - for (i = 0; i < conf.udp_target_given; i++) - if (parse_url(conf.udp_target_arg[i], host, - sizeof(host), &port) == NULL) + for (i = 0; i < conf.udp_target_given; i++) { + if (parse_fec_url(conf.udp_target_arg[i], &scd) < 0) { PARA_CRIT_LOG("syntax error for udp target option " - "#%d, ignoring\n", i); - else - udp_add_target(host, port); + "#%d, ignoring\n", i); + continue; + } + udp_add_target(&scd); + } } static char *udp_help(void) { return make_message( "usage: {on|off}\n" - "usage: {add|delete} host[:port]\n" + "usage: {add|delete} host[:port][/packet_size:k:n]\n" "examples: add 224.0.1.38:1500 (IPv4 multicast)\n" + " add 224.0.1.38:1500/1400:14:16\n" + " (likewise, using 1400 byte packets, with 14\n" + " data slices per 16 slice FEC group)\n" " add 10.10.1.42 (using default port)\n" " add [FF05::42]:1500 (IPv6 multicast)\n" " add [::1] (IPv6 localhost/default port)\n" @@ -392,7 +350,7 @@ static char *udp_help(void) /** * The init function of para_server's udp sender. * - * \param s Pointer to the http sender struct. + * \param s Pointer to the udp sender struct. * * It initializes all function pointers of \a s and the list of udp targets. */ @@ -401,7 +359,7 @@ void udp_send_init(struct sender *s) INIT_LIST_HEAD(&targets); s->info = udp_info; s->help = udp_help; - s->send = udp_send; + s->send = NULL; s->pre_select = NULL; s->post_select = NULL; s->shutdown_clients = udp_shutdown_targets;