path: root/udp_send.c

/* SPDX-License-Identifier: GPL-2.0 */

/** \file udp_send.c Para_server's udp sender. */

#include <netinet/in.h>
#include <sys/types.h>
#include <netinet/udp.h>
#include <net/if.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <lopsub.h>

#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,
};