/* SPDX-License-Identifier: GPL-2.0 */ /** \file udp_send.c Para_server's udp sender. */ #include #include #include #include #include #include #include #include "server.lsg.h" #include "para.h" #include "error.h" #include "string.h" #include "afh.h" #include "net.h" #include "list.h" #include "server.h" #include "sched.h" #include "send.h" #include "vss.h" #include "portable_io.h" #include "fd.h" #include "fec.h" /* * Time window during which ICMP Destination/Port Unreachable messages are * ignored, covering transient receiver problems such as restarting the * client, rebooting, reconfiguration, or handover. */ #define UDP_MAX_UNREACHABLE_TIME 30 /* Describes one entry in the list of targets for the udp sender. */ struct udp_target { /* Track time (seconds) of last ICMP Port Unreachable error */ time_t last_unreachable; /* The opaque structure returned by vss_add_fec_client(). */ struct fec_client *fc; /* The FEC parameters for this target. */ struct fec_client_parms fcp; /* Whether we already sent the FEC eof packet to this target. */ bool sent_fec_eof; }; static INITIALIZED_LIST_HEAD(targets); static int sender_status; static void udp_close_target(struct sender_client *sc) { struct udp_target *ut = sc->private_data; if (process_is_command_handler()) return; if (ut->sent_fec_eof) return; PARA_INFO_LOG("sending FEC EOF\n"); /* Ignore write() errors since we are closing the target anyway. */ if (write(sc->fd, FEC_EOF_PACKET, FEC_HEADER_SIZE)) do_nothing; /* avoid "ignoring return value" warning */ ut->sent_fec_eof = true; } static void udp_delete_target(struct sender_client *sc, const char *msg) { struct udp_target *ut = sc->private_data; if (!process_is_command_handler()) PARA_NOTICE_LOG("deleting %s (%s) from list\n", sc->name, msg); udp_close_target(sc); if (!process_is_command_handler()) { close(sc->fd); del_close_on_fork_list(sc->fd); } vss_del_fec_client(ut->fc); list_del(&sc->node); free(sc->name); free(sc); free(ut); } /* Perform AF-independent multicast sender setup. */ static int setup_multicast(struct sender_client *sc) { struct sockaddr_storage ss; socklen_t sslen = sizeof(ss); int ttl = OPT_INT32_VAL(UDP_TTL), id = 0; struct in_addr *in4 = &((struct sockaddr_in *)&ss)->sin_addr; const int on = 1; if (OPT_GIVEN(UDP_MCAST_IFACE)) { const char *iface = OPT_STRING_VAL(UDP_MCAST_IFACE); id = if_nametoindex(iface); if (id == 0) PARA_WARNING_LOG("could not resolve interface '%s', " "using default", iface); } if (getpeername(sc->fd, (struct sockaddr *)&ss, &sslen) < 0) return -ERRNO_TO_PARA_ERROR(errno); if (ss.ss_family == AF_INET6) { struct in6_addr *in6 = &((struct sockaddr_in6 *)&ss)->sin6_addr; if (!IN6_IS_ADDR_MULTICAST(in6)) return 0; if (id != 0 && setsockopt(sc->fd, IPPROTO_IPV6, IPV6_MULTICAST_IF, &id, sizeof(id)) < 0) return -ERRNO_TO_PARA_ERROR(errno); if (setsockopt(sc->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, &on, sizeof(on)) < 0) return -ERRNO_TO_PARA_ERROR(errno); if (ttl < 0 || ttl > 255) ttl = -1; /* use kernel default (see RFC 3493, 5.2) */ if (setsockopt(sc->fd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &ttl, sizeof(ttl)) < 0) return -ERRNO_TO_PARA_ERROR(errno); return 1; } /* AF_INET */ if (!IN_MULTICAST(htonl(in4->s_addr))) return 0; if (id != 0) { #ifdef HAVE_IP_MREQN struct ip_mreqn mn = {.imr_ifindex = id}; if (setsockopt(sc->fd, IPPROTO_IP, IP_MULTICAST_IF, &mn, sizeof(mn)) < 0) return -ERRNO_TO_PARA_ERROR(errno); #endif } /* * Enable receiving multicast messages generated on the local host * At least on Linux, this is enabled by default. */ if (setsockopt(sc->fd, IPPROTO_IP, IP_MULTICAST_LOOP, &on, sizeof(on)) < 0) return -ERRNO_TO_PARA_ERROR(errno); if (ttl < 0 || ttl > 255) ttl = 1; /* use local subnet (do not flood out into the WAN) */ if (setsockopt(sc->fd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0) return -ERRNO_TO_PARA_ERROR(errno); return 1; } static void udp_shutdown_targets(void) { struct sender_client *sc, *tmp; list_for_each_entry_safe(sc, tmp, &targets, node) udp_close_target(sc); } static void udp_shutdown(void) { struct sender_client *sc, *tmp; list_for_each_entry_safe(sc, tmp, &targets, node) udp_delete_target(sc, "shutdown"); } static int udp_resolve_target(const char *url, struct sender_command_data *scd) { const char *result; int ret, port; ret = parse_fec_url(url, scd); if (ret) return ret; port = scd->port > 0 ? scd->port : OPT_UINT32_VAL(UDP_DEFAULT_PORT); ret = para_connect(IPPROTO_UDP, scd->host, port); if (ret < 0) return ret; result = remote_name(ret); close(ret); if (!parse_url(result, scd->host, sizeof(scd->host), &scd->port)) return -E_ADDRESS_LOOKUP; return 1; } static int udp_com_on(__a_unused struct sender_command_data *scd) { sender_status = SENDER_on; return 1; } static int udp_com_off(__a_unused struct sender_command_data *scd) { udp_shutdown_targets(); sender_status = SENDER_off; return 1; } static struct sender_client *udp_lookup_target(struct sender_command_data *scd) { struct sender_client *sc, *tmp; char host[MAX_HOSTLEN]; int32_t port; list_for_each_entry_safe(sc, tmp, &targets, node) { parse_url(sc->name, host, sizeof(host), &port); /* Unspecified port means wildcard port match */ if (scd->port > 0 && scd->port != port) continue; if (strcmp(host, scd->host)) continue; return sc; } return NULL; } static int udp_com_delete(struct sender_command_data *scd) { struct sender_client *sc = udp_lookup_target(scd); if (sc) { udp_delete_target(sc, "com_delete"); return 1; } PARA_NOTICE_LOG("not deleting non-existing target '%s'\n", scd->host); return -E_TARGET_NOT_FOUND; } /* * Initialize UDP session and set maximum payload size. * * The socket must be connected. See RFC 1122, 3.3.3. If the protocol family * could not be determined, AF_INET is assumed. * * - RFC 1122, 3.3.3 defines EMTU_S ("Effective MTU for sending") and recommends * to use an EMTU_S size of of 576 bytes if the IPv4 path MTU is unknown; * - RFC 2460, 5. requires a minimum IPv6 MTU of 1280 bytes; * - RFC 5405, 3.2 recommends that if path MTU discovery is not done, * UDP senders should use the respective minimum values of EMTU_S. * * Returns: The maximum message size of the address family type. */ static int udp_init_fec(struct sender_client *sc) { struct udp_target *ut = sc->private_data; struct sockaddr_storage ss = {.ss_family = 0}; socklen_t sslen = sizeof(ss); int af_type = AF_INET; PARA_INFO_LOG("sending to udp %s\n", sc->name); ut->sent_fec_eof = false; if (getpeername(sc->fd, (struct sockaddr *)&ss, &sslen) >= 0) { if (!ss_is_addr_v4mapped(&ss)) af_type = ss.ss_family; } /* Subtract approximation of IP header overhead, neglecting options. */ return af_type == AF_INET6? 1280 - 40 : 576 - 20; } static void udp_send_fec(struct sender_client *sc, char *buf, size_t len) { struct udp_target *ut = sc->private_data; int ret, err; socklen_t errlen = sizeof(err); if (sender_status == SENDER_off || len == 0) return; /* Check and clear socket error if any. */ if (getsockopt(sc->fd, SOL_SOCKET, SO_ERROR, &err, &errlen) < 0) return udp_delete_target(sc, strerror(errno)); /* * ECONNREFUSED happens if an ICMP Destination / Port Unreachable * has arrived. Ignore, as persistent errors will be caught below. */ if (err != 0 && err != ECONNREFUSED) return udp_delete_target(sc, strerror(err)); ret = xwrite(sc->fd, buf, len); if (ret < 0 && ret != -ERRNO_TO_PARA_ERROR(ECONNREFUSED)) return udp_delete_target(sc, para_strerror(-ret)); if (ret >= 0) /* success */ return; if (now->tv_sec <= ut->last_unreachable + UDP_MAX_UNREACHABLE_TIME) return; if (now->tv_sec > ut->last_unreachable + 2 * UDP_MAX_UNREACHABLE_TIME) { ut->last_unreachable = now->tv_sec; return; } /* No errors are allowed during this time window. */ udp_delete_target(sc, strerror(ECONNREFUSED)); } static int udp_com_add(struct sender_command_data *scd) { int ret; struct udp_target *ut; struct sender_client *sc = udp_lookup_target(scd); if (sc) { PARA_NOTICE_LOG("target %s already exists - not adding it\n", sc->name); return -E_TARGET_EXISTS; } ut = zalloc(sizeof(*ut)); sc = zalloc(sizeof(*sc)); ut->fcp.slices_per_group = scd->slices_per_group; ut->fcp.data_slices_per_group = scd->data_slices_per_group; 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; ret = para_connect(IPPROTO_UDP, scd->host, scd->port); if (ret < 0) goto err; sc->fd = ret; ret = setup_multicast(sc); if (ret < 0) goto err; ret = mark_fd_nonblocking(sc->fd); if (ret < 0) goto err; sc->name = para_strdup(remote_name(sc->fd)); 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) close(sc->fd); PARA_NOTICE_LOG("failed to set up %s:%d (%s)- not adding it\n", scd->host, scd->port, para_strerror(-ret)); free(sc); free(ut); return ret; } 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 ", tgts ? : "", sc->name, ut->fcp.data_slices_per_group, ut->fcp.slices_per_group ); free(tgts); tgts = tmp; } ret = make_message( "status: %s\n" "port: %s\n" "targets: %s\n", (sender_status == SENDER_on)? "on" : "off", stringify_port(OPT_UINT32_VAL(UDP_DEFAULT_PORT), "udp"), tgts? tgts : "(none)" ); free(tgts); return ret; } static char *udp_help(void) { return make_message( "usage: {on|off}\n" "usage: {add|delete} ip_address[:port][/[packet_size:]k:n]\n" " - k is the number of data slices per FEC group\n" " - n is the total number of slices in a FEC group\n" " - packet_size reduces the slice size below path MTU\n\n" "examples: add 224.0.1.38:1500 (IPv4 multicast)\n" " add 224.0.1.38:8080/14:16 (explicit FEC)\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 10.1.2.3:456 (IPv4 with explicit port)\n" " delete [badc0de::1] (IPv6 with port wildcard)\n" ); } /* Initialize the list of udp targets. */ static void udp_send_init(void) { struct sender_command_data scd; sender_status = OPT_GIVEN(UDP_NO_AUTOSTART)? SENDER_off : SENDER_on; for (int i = 0; i < OPT_GIVEN(UDP_TARGET); i++) { const char *arg = lls_string_val(i, OPT_RESULT(UDP_TARGET)); if (udp_resolve_target(arg, &scd) < 0) PARA_ERROR_LOG("not adding requested target '%s'\n", arg); else udp_com_add(&scd); } } /** * The UDP sender. * * In contrast to the HTTP sender, the UDP sender is active in the sense * that it initiates the network connection according to its list of targets * rather than passively waiting for clients to connect. UDP streams are * always sent FEC-encoded. Another difference to the HTTP sender is that * the UDP sender supports IP multicasting. */ const struct sender udp_sender = { .name = "udp", .init = udp_send_init, .shutdown = udp_shutdown, .shutdown_clients = udp_shutdown_targets, .resolve_target = udp_resolve_target, .client_cmds = { [SENDER_on] = udp_com_on, [SENDER_off] = udp_com_off, [SENDER_add] = udp_com_add, [SENDER_delete] = udp_com_delete, }, .help = udp_help, .status = udp_status, };