X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=478c5d9fcfa58c80097cb21f67a7748c964aae12;hp=140458e7ed9b1d4e3a03d05c2e26a6b36d66db54;hb=f099900540c4e4c54b10181254b895ccfe6ef410;hpb=6ba0a5055cfdec25d54d90a56f9345d5d4be09be diff --git a/udp_send.c b/udp_send.c index 140458e7..478c5d9f 100644 --- a/udp_send.c +++ b/udp_send.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2005-2009 Andre Noll + * Copyright (C) 2005-2010 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ @@ -7,8 +7,12 @@ /** \file udp_send.c Para_server's udp sender. */ +#include #include #include +#include +#include +#include #include "server.cmdline.h" #include "para.h" @@ -17,36 +21,37 @@ #include "afh.h" #include "afs.h" #include "server.h" -#include "vss.h" #include "list.h" #include "send.h" +#include "vss.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; static int sender_status; +static struct sender *self; static void udp_close_target(struct udp_target *ut) { @@ -61,9 +66,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); } @@ -147,15 +153,16 @@ err: /** The maximal size of the per-target chunk queue. */ #define UDP_CQ_BYTES 40000 -static int udp_init_session(struct udp_target *ut) +static int udp_open(void *client, struct fec_client_parms **fcp) { + struct udp_target *ut = client; int ret; char *iface = NULL; if (ut->fd >= 0) /* nothing to do */ return 0; - ret = makesock(AF_UNSPEC, IPPROTO_UDP, 0, TARGET_ADDR(ut), ut->port); + ret = para_connect_simple(IPPROTO_UDP, ut->host, ut->port); if (ret < 0) return ret; ut->fd = ret; @@ -176,121 +183,30 @@ 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); + ut->fcp.max_slice_bytes = 1472; /* FIXME */ + *fcp = &ut->fcp; + PARA_NOTICE_LOG("sending to udp %s#%d\n", ut->host, ut->port); 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) { + int ubuntu_glibc_headers_suck; if (ut->fd < 0) continue; - write(ut->fd, buf, UDP_AUDIO_HEADER_LEN); + if (!buf) + len = vss_get_fec_eof_packet(&buf); + /* ignore return value, we're closing the target anyway. */ + ubuntu_glibc_headers_suck = write(ut->fd, buf, len); /* STFU */ 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 +222,69 @@ 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; + + if (sender_status == SENDER_OFF) + return 0; + ret = send_queued_chunks(ut->fd, ut->cq, 0); + if (ret == -ERRNO_TO_PARA_ERROR(ECONNREFUSED)) + ret = 0; + if (ret < 0) + goto fail; + if (!len) + return 0; + if (!ret) { /* still data left in the queue */ + ret = cq_force_enqueue(ut->cq, buf, len); + assert(ret >= 0); + } + ret = write_nonblock(ut->fd, buf, len, 0); + if (ret == -ERRNO_TO_PARA_ERROR(ECONNREFUSED)) + ret = 0; + if (ret < 0) + goto fail; + if (ret != len) { + ret = cq_force_enqueue(ut->cq, buf + ret, len - ret); + assert(ret >= 0); + } + 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; + /* XXX RESOLVED LATER vss_add_fec_client(self, ut, &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,18 +294,24 @@ 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; } ret = make_message( "udp sender:\n" "\tstatus: %s\n" - "\tport: udp %d\n" + "\tport: %s\n" "\ttargets: %s\n", (sender_status == SENDER_ON)? "on" : "off", - conf.udp_default_port_arg, + stringify_port(conf.udp_default_port_arg, "udp"), tgts? tgts : "(none)" ); free(tgts); @@ -362,31 +320,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 +338,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 +363,9 @@ 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->send_fec = udp_send_fec; + s->open = udp_open; s->pre_select = NULL; s->post_select = NULL; s->shutdown_clients = udp_shutdown_targets; @@ -425,5 +379,6 @@ void udp_send_init(struct sender *s) udp_init_target_list(); if (!conf.udp_no_autostart_given) sender_status = SENDER_ON; + self = s; PARA_DEBUG_LOG("udp sender init complete\n"); }