X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=478c5d9fcfa58c80097cb21f67a7748c964aae12;hp=7585eaa962ecb4577ed16e41c88235ed6ccec60a;hb=96dd1284a1b7d565560979fdc7ec647aeceb6b52;hpb=4a96cfc150f797745c054bd5516e80079bdf52a8 diff --git a/udp_send.c b/udp_send.c index 7585eaa9..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,9 +7,12 @@ /** \file udp_send.c Para_server's udp sender. */ +#include #include #include +#include #include +#include #include "server.cmdline.h" #include "para.h" @@ -18,9 +21,9 @@ #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 "net.h" #include "fd.h" @@ -48,6 +51,7 @@ struct udp_target { static struct list_head targets; static int sender_status; +static struct sender *self; static void udp_close_target(struct udp_target *ut) { @@ -149,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, ut->host, ut->port); + ret = para_connect_simple(IPPROTO_UDP, ut->host, ut->port); if (ret < 0) return ret; ut->fd = ret; @@ -178,6 +183,8 @@ static int udp_init_session(struct udp_target *ut) } add_close_on_fork_list(ut->fd); ut->cq = cq_new(UDP_CQ_BYTES); + 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; } @@ -189,11 +196,13 @@ static void udp_shutdown_targets(void) size_t len = 0; /* STFU, gcc */ list_for_each_entry_safe(ut, tmp, &targets, node) { + int ubuntu_glibc_headers_suck; if (ut->fd < 0) continue; if (!buf) len = vss_get_fec_eof_packet(&buf); - write(ut->fd, buf, len); + /* ignore return value, we're closing the target anyway. */ + ubuntu_glibc_headers_suck = write(ut->fd, buf, len); /* STFU */ udp_close_target(ut); } } @@ -229,27 +238,29 @@ static int udp_com_delete(struct sender_command_data *scd) 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); + int ret; - if (ret < 0) - goto fail; + 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_enqueue(ut->cq, buf, len); - if (ret < 0) - goto fail; + 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_enqueue(ut->cq, buf + ret, len - ret); - if (ret < 0) - goto fail; + ret = cq_force_enqueue(ut->cq, buf + ret, len - ret); + assert(ret >= 0); } return 1; fail: @@ -257,27 +268,23 @@ fail: return ret; } -static void udp_add_target(const char *host, int port) +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 = 16; - ut->fcp.data_slices_per_group = 14; - ut->fcp.max_slice_bytes = 1400; - ut->fcp.send = udp_send_fec; - ut->fcp.private_data = ut; - vss_add_fec_client(&ut->fcp, &ut->fc); + 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) { - udp_add_target(scd->host, scd->port); + udp_add_target(scd); return 1; } @@ -288,19 +295,23 @@ 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; } 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); @@ -309,25 +320,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" @@ -339,7 +354,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. */ @@ -349,6 +364,8 @@ void udp_send_init(struct sender *s) s->info = udp_info; s->help = udp_help; 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; @@ -362,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"); }