X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=7217c960c6597123701338aa2dfedcbcc31115b3;hp=140458e7ed9b1d4e3a03d05c2e26a6b36d66db54;hb=165a3886ab506f4c13eac4f178a58493432a00ec;hpb=1e6a73e98fecaba8226d5a4aa3cdc84e410f8029 diff --git a/udp_send.c b/udp_send.c index 140458e7..7217c960 100644 --- a/udp_send.c +++ b/udp_send.c @@ -9,6 +9,8 @@ #include #include +#include +#include #include "server.cmdline.h" #include "para.h" @@ -21,28 +23,28 @@ #include "list.h" #include "send.h" #include "portable_io.h" -#include "udp_header.h" #include "net.h" #include "fd.h" #include "sched.h" #include "close_on_fork.h" #include "chunk_queue.h" -/** Convert in_addr to ascii. */ -#define TARGET_ADDR(oc) inet_ntoa((oc)->addr) - /** Describes one entry in the list of targets for the udp sender. */ struct udp_target { - /** Address info. */ - struct in_addr addr; /** The position of this target in the list of targets. */ struct list_head node; + /** The hostname (DNS name or IPv4/v6 address string). */ + char host[MAX_HOSTLEN]; /** The UDP port. */ int port; /** The socket fd. */ 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,9 +63,10 @@ static void udp_close_target(struct udp_target *ut) static void udp_delete_target(struct udp_target *ut, const char *msg) { - PARA_NOTICE_LOG("deleting %s:%d (%s) from list\n", TARGET_ADDR(ut), + 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); } @@ -155,7 +158,7 @@ static int udp_init_session(struct udp_target *ut) if (ut->fd >= 0) /* nothing to do */ return 0; - ret = makesock(AF_UNSPEC, IPPROTO_UDP, 0, TARGET_ADDR(ut), ut->port); + ret = makesock(AF_UNSPEC, IPPROTO_UDP, 0, ut->host, ut->port); if (ret < 0) return ret; ut->fd = ret; @@ -176,121 +179,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", TARGET_ADDR(ut), 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; @@ -306,33 +216,70 @@ static int udp_com_off(__a_unused struct sender_command_data *scd) static int udp_com_delete(struct sender_command_data *scd) { - char *a = para_strdup(inet_ntoa(scd->addr)); struct udp_target *ut, *tmp; + list_for_each_entry_safe(ut, tmp, &targets, node) { - if (scd->port != ut->port) + /* Unspecified port means wildcard port match */ + if (scd->port > 0 && scd->port != ut->port) continue; - if (strcmp(TARGET_ADDR(ut), a)) + if (strcmp(ut->host, scd->host)) continue; udp_delete_target(ut, "com_delete"); } return 1; } -static void udp_add_target(int port, struct in_addr *addr) +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)); - ut->port = port; - ut->addr = *addr; + + 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", - TARGET_ADDR(ut), 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) { - int port = (scd->port > 0)? scd->port : conf.udp_default_port_arg; - udp_add_target(port, &scd->addr); + udp_add_target(scd); return 1; } @@ -342,8 +289,14 @@ static char *udp_info(void) char *ret, *tgts = NULL; list_for_each_entry(ut, &targets, node) { - char *tmp = make_message("%s%s:%d ", tgts? tgts : "", - TARGET_ADDR(ut), ut->port); + bool is_v6 = strchr(ut->host, ':') != NULL; + 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,31 +315,17 @@ static char *udp_info(void) static void udp_init_target_list(void) { + struct sender_command_data scd; int i; INIT_LIST_HEAD(&targets); for (i = 0; i < conf.udp_target_given; i++) { - char *arg = para_strdup(conf.udp_target_arg[i]); - char *p = strchr(arg, ':'); - int port; - struct in_addr addr; - - if (!p) - goto err; - *p = '\0'; - if (!inet_pton(AF_INET, arg, &addr)) - goto err; - port = atoi(++p); - if (port < 0 || port > 65535) - port = conf.udp_default_port_arg; - udp_add_target(port, &addr); - goto success; -err: - PARA_CRIT_LOG("syntax error for udp target option " - "#%d, ignoring\n", i); -success: - free(arg); - continue; + if (parse_fec_url(conf.udp_target_arg[i], &scd) < 0) { + PARA_CRIT_LOG("syntax error for udp target option " + "#%d, ignoring\n", i); + continue; + } + udp_add_target(&scd); } } @@ -394,15 +333,23 @@ static char *udp_help(void) { return make_message( "usage: {on|off}\n" - "usage: {add|delete} IP port\n" - "example: add 224.0.1.38 8000 (multicast streaming)\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" + " delete myhost.net (host with port wildcard)\n" + " delete [badc0de::1] (IPv6 with port wildcard)\n" ); } /** * 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. */ @@ -411,7 +358,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;