X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=udp_send.c;h=8fca5bc6c2759c27b4e0402cba3e341b60b82ebc;hp=adf233979c31e2aff79ecdf94b5b16fd6f8765da;hb=b52960c3c1419f0c9198e29425d25eb69bdd3fa4;hpb=ace5a88d69e5d41ce4959b43d278cc1c8db93706 diff --git a/udp_send.c b/udp_send.c index adf23397..8fca5bc6 100644 --- a/udp_send.c +++ b/udp_send.c @@ -1,36 +1,35 @@ /* - * Copyright (C) 2005-2010 Andre Noll + * Copyright (C) 2005-2014 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ /** \file udp_send.c Para_server's udp sender. */ - -#include -#include -#include +#include #include +#include +#include #include #include -#include +#include +#include +#include #include "server.cmdline.h" #include "para.h" #include "error.h" #include "string.h" #include "afh.h" -#include "afs.h" #include "server.h" #include "list.h" #include "send.h" +#include "sched.h" #include "vss.h" #include "portable_io.h" #include "net.h" #include "fd.h" -#include "sched.h" #include "close_on_fork.h" -#include "chunk_queue.h" /** * Time window during which ICMP Destination/Port Unreachable messages are @@ -43,8 +42,6 @@ struct udp_target { /** Track time (seconds) of last ICMP Port Unreachable error */ time_t last_unreachable; - /** Common sender client data */ - struct sender_client *sc; /** The opaque structure returned by vss_add_fec_client(). */ struct fec_client *fc; /** The FEC parameters for this target. */ @@ -56,11 +53,16 @@ static int sender_status; static void udp_close_target(struct sender_client *sc) { - if (sc->cq != NULL) { - del_close_on_fork_list(sc->fd); - cq_destroy(sc->cq); - sc->cq = NULL; - } + const char *buf; + size_t len = vss_get_fec_eof_packet(&buf); + + /* + * Ignore the return value of write() since we are closing the target + * anyway. The sole purpose of the "do_nothing" statement is to silence + * gcc. + */ + if (write(sc->fd, buf, len)) + do_nothing; } static void udp_delete_target(struct sender_client *sc, const char *msg) @@ -69,6 +71,8 @@ static void udp_delete_target(struct sender_client *sc, const char *msg) PARA_NOTICE_LOG("deleting %s (%s) from list\n", sc->name, msg); udp_close_target(sc); + close(sc->fd); + del_close_on_fork_list(sc->fd); vss_del_fec_client(ut->fc); list_del(&sc->node); free(sc->name); @@ -152,30 +156,16 @@ err: return -ERRNO_TO_PARA_ERROR(errno); } -/** The maximal size of the per-target chunk queue. */ -#define UDP_CQ_BYTES 40000 - static void udp_init_session(struct sender_client *sc) { - if (sc->cq == NULL) { - sc->cq = cq_new(UDP_CQ_BYTES); - add_close_on_fork_list(sc->fd); - PARA_NOTICE_LOG("sending to udp %s\n", sc->name); - } + PARA_NOTICE_LOG("sending to udp %s\n", sc->name); } static void udp_shutdown_targets(void) { struct sender_client *sc, *tmp; - const char *buf; - size_t len = vss_get_fec_eof_packet(&buf); - list_for_each_entry_safe(sc, tmp, &targets, node) - if (sc->cq != NULL) { - /* ignore return value, closing the target anyway. */ - (void)write(sc->fd, buf, len); - udp_close_target(sc); - } + udp_close_target(sc); } static int udp_resolve_target(const char *url, struct sender_command_data *scd) @@ -288,26 +278,18 @@ static int udp_check_socket_state(struct sender_client *sc) return -ERRNO_TO_PARA_ERROR(ret); } -static int udp_send_fec(struct sender_client *sc, char *buf, size_t len) +static void udp_send_fec(struct sender_client *sc, char *buf, size_t len) { int ret; if (sender_status == SENDER_OFF) - return 0; - if (len == 0 && !cq_peek(sc->cq)) - return 0; + return; + if (len == 0) + return; ret = udp_check_socket_state(sc); if (ret < 0) goto fail; - ret = send_queued_chunks(sc->fd, sc->cq); - if (ret < 0) - goto fail; - if (!ret) { /* still data left in the queue */ - ret = cq_force_enqueue(sc->cq, buf, len); - assert(ret >= 0); - return 0; - } - ret = write_nonblock(sc->fd, buf, len); + ret = xwrite(sc->fd, buf, len); if (ret == -ERRNO_TO_PARA_ERROR(ECONNREFUSED)) { /* * Happens if meanwhile an ICMP Destination / Port Unreachable @@ -317,14 +299,9 @@ static int udp_send_fec(struct sender_client *sc, char *buf, size_t len) } if (ret < 0) goto fail; - if (ret != len) { - ret = cq_force_enqueue(sc->cq, buf + ret, len - ret); - assert(ret >= 0); - } - return 1; + return; fail: udp_delete_target(sc, para_strerror(-ret)); - return ret; } static int udp_com_add(struct sender_command_data *scd) @@ -339,12 +316,12 @@ static int udp_com_add(struct sender_command_data *scd) return -E_TARGET_EXISTS; } ut = para_calloc(sizeof(*ut)); - sc = ut->sc = para_calloc(sizeof(*sc)); + sc = para_calloc(sizeof(*sc)); 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.init_fec = udp_init_fec; ut->fcp.send_fec = udp_send_fec; + ut->fcp.need_periodic_header = true; sc->private_data = ut; sc->fd = -1; @@ -363,6 +340,7 @@ static int udp_com_add(struct sender_command_data *scd) PARA_INFO_LOG("adding to target list (%s)\n", sc->name); ut->fc = vss_add_fec_client(sc, &ut->fcp); para_list_add(&sc->node, &targets); + add_close_on_fork_list(sc->fd); return 1; err: if (sc->fd >= 0) @@ -374,16 +352,15 @@ err: return ret; } -static char *udp_info(void) +static char *udp_status(void) { struct sender_client *sc; char *ret, *tgts = NULL; list_for_each_entry(sc, &targets, node) { struct udp_target *ut = sc->private_data; - char *tmp = make_message("%s%s/%u:%u:%u ", + char *tmp = make_message("%s%s/%u:%u ", tgts ? : "", sc->name, - ut->fcp.max_slice_bytes, ut->fcp.data_slices_per_group, ut->fcp.slices_per_group ); @@ -391,10 +368,9 @@ static char *udp_info(void) tgts = tmp; } ret = make_message( - "udp sender:\n" - "\tstatus: %s\n" - "\tport: %s\n" - "\ttargets: %s\n", + "status: %s\n" + "port: %s\n" + "targets: %s\n", (sender_status == SENDER_ON)? "on" : "off", stringify_port(conf.udp_default_port_arg, "udp"), tgts? tgts : "(none)" @@ -447,7 +423,7 @@ static char *udp_help(void) void udp_send_init(struct sender *s) { INIT_LIST_HEAD(&targets); - s->info = udp_info; + s->status = udp_status; s->help = udp_help; s->send = NULL; s->pre_select = NULL;