X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=fc3bb2f5c5f657a45112cf2db3856407c8c0c611;hp=a21f478d78dd3c3c56f14196264ce8424b107254;hb=d656b93201ccb745f267a847b07e9dfa3b33f9f5;hpb=1673dedb71f81a93e282bb09d6ae7911b82d7503 diff --git a/udp_send.c b/udp_send.c index a21f478d..fc3bb2f5 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" @@ -157,7 +160,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, ut->host, ut->port); + ret = para_connect_simple(IPPROTO_UDP, ut->host, ut->port); if (ret < 0) return ret; ut->fd = ret; @@ -178,9 +181,7 @@ 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 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); + PARA_NOTICE_LOG("sending to udp %s#%d\n", ut->host, ut->port); return 1; } @@ -231,27 +232,32 @@ 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 (sender_status == SENDER_OFF) + return 0; + ret = udp_init_session(ut); if (ret < 0) goto fail; 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: @@ -289,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); @@ -344,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. */