path: root/sync_filter.c

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

/** \file sync_filter.c Playback synchronization filter. */

#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <netdb.h>
#include <lopsub.h>

#include "filter_cmd.lsg.h"
#include "para.h"
#include "list.h"
#include "net.h"
#include "sched.h"
#include "buffer_tree.h"
#include "filter.h"
#include "string.h"
#include "fd.h"
#include "error.h"

struct sync_buddy_info {
	const char *url;
	char *host;
	int port;
	struct addrinfo *ai;
	bool disabled;
};

/* One active buddy */
struct sync_buddy {
	int fd;
	struct sync_buddy_info *sbi;
	bool ping_received;
	struct list_head node;
};

/* Allocated in ->open(), stored in fn->private_data */
struct sync_filter_context {
	int listen_fd;
	struct list_head buddies;
	struct timeval timeout;
	bool ping_sent;
};

#define FOR_EACH_BUDDY(_buddy, _list) \
	list_for_each_entry(_buddy, _list, node)
#define FOR_EACH_BUDDY_SAFE(_buddy, _tmp_buddy, _list) \
	list_for_each_entry_safe(_buddy, _tmp_buddy, _list, node)

static void sync_close_buddy(struct sync_buddy *buddy)
{
	PARA_DEBUG_LOG("closing %s, fd %d\n", buddy->sbi->url, buddy->fd);
	close(buddy->fd);
	list_del(&buddy->node);
	free(buddy);
}

static void sync_close_buddies(struct sync_filter_context *ctx)
{
	struct sync_buddy *buddy, *tmp;

	FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies)
		sync_close_buddy(buddy);
}

static void sync_close(struct filter_node *fn)
{
	struct sync_filter_context *ctx = fn->private_data;

	sync_close_buddies(ctx);
	if (ctx->listen_fd >= 0) {
		close(ctx->listen_fd);
		ctx->listen_fd = -1;
	}
	free(ctx);
	fn->private_data = NULL;
}

static void sync_teardown(const struct lls_parse_result *lpr, void *conf)
{
	struct sync_buddy_info *sbi = conf;
	int i, num_buddies = FILTER_CMD_OPT_GIVEN(SYNC, BUDDY, lpr);

	for (i = 0; i < num_buddies; i++) {
		free(sbi[i].host);
		freeaddrinfo(sbi[i].ai);
	}
	free(sbi);
}

static void sync_open(struct filter_node *fn)
{
	int i, ret;
	struct sync_buddy *buddy;
	struct sync_filter_context *ctx;
	struct sync_buddy_info *sbi = fn->conf;
	uint32_t port = FILTER_CMD_OPT_UINT32_VAL(SYNC, PORT, fn->lpr);
	unsigned buddy_given;
	const struct lls_opt_result *r_b;

	ctx = fn->private_data = zalloc(sizeof(*ctx));
	init_list_head(&ctx->buddies);

	/* create socket to listen for incoming packets */
	ret = makesock(
		IPPROTO_UDP,
		true /* passive */,
		NULL /* no host required */,
		port
	);
	if (ret < 0) {
		PARA_ERROR_LOG("could not create UDP listening socket %u\n",
			port);
		return;
	}
	ctx->listen_fd = ret;
	PARA_INFO_LOG("listening on fd %d\n", ctx->listen_fd);

	r_b = FILTER_CMD_OPT_RESULT(SYNC, BUDDY, fn->lpr);
	buddy_given = lls_opt_given(r_b);
	for (i = 0; i < buddy_given; i++) {
		int fd;
		const char *url = lls_string_val(i, r_b);

		/* make buddy udp socket from address info */
		assert(sbi->ai);
		ret = makesock_addrinfo(
			IPPROTO_UDP,
			false /* not passive */,
			sbi[i].ai
		);
		if (ret < 0) {
			PARA_WARNING_LOG("could not make socket for %s\n",
				url);
			goto fail;
		}
		fd = ret;
		ret = mark_fd_nonblocking(fd);
		if (ret < 0) {
			PARA_ERROR_LOG("unable to set nonblock mode for %s\n",
				url);
			close(fd);
			goto fail;
		}
		buddy = alloc(sizeof(*buddy));
		buddy->fd = fd;
		buddy->sbi = sbi + i;
		buddy->ping_received = false;
		para_list_add(&buddy->node, &ctx->buddies);

		PARA_INFO_LOG("opened buddy %s on fd %d\n", url, fd);
		continue;
fail:
		PARA_WARNING_LOG("%s\n", para_strerror(-ret));
	}
}

/*
 * Build an array of struct sync_buddy_info with one entry for each buddy given
 * in the arguments. This array is not affected by sync_close(), so information
 * stored there can be used for multiple instances (para_audiod). We store the
 * resolved url and the ->disabled bit in this array.
 */
static void *sync_setup(const struct lls_parse_result *lpr)
{
	int i, ret;
	unsigned n;
	struct sync_buddy_info *sbi;
	const struct lls_opt_result *r_b;

	r_b = FILTER_CMD_OPT_RESULT(SYNC, BUDDY, lpr);
	n = lls_opt_given(r_b);
	sbi = arr_alloc(n, sizeof(*sbi));
	PARA_INFO_LOG("initializing buddy info array of length %u\n", n);
	for (i = 0; i < n; i++) {
		const char *url = lls_string_val(i, r_b);
		size_t len = strlen(url);
		char *host = alloc(len + 1);
		int port;
		struct addrinfo *ai;

		if (!parse_url(url, host, len, &port)) {
			PARA_ERROR_LOG("could not parse url %s\n", url);
			exit(EXIT_FAILURE);
		}
		if (port < 0)
			port = FILTER_CMD_OPT_UINT32_VAL(SYNC, PORT, lpr);
		ret = lookup_address(IPPROTO_UDP, false /* not passive */,
			host, port, &ai);
		if (ret < 0) {
			PARA_ERROR_LOG("host lookup failure for %s: %s\n",
				url, para_strerror(-ret));
			exit(EXIT_FAILURE);
		}
		sbi[i].url = url;
		sbi[i].host = host;
		sbi[i].port = port;
		sbi[i].ai = ai;
		sbi[i].disabled = false;
		PARA_DEBUG_LOG("buddy #%d: %s\n", i, url);
	}
	return sbi;
}

/*
 * True if we sent a packet to all buddies and received a packet from each
 * enabled buddy.
 */
static bool sync_complete(struct sync_filter_context *ctx)
{
	struct sync_buddy *buddy;

	if (!ctx->ping_sent)
		return false;
	FOR_EACH_BUDDY(buddy, &ctx->buddies)
		if (!buddy->sbi->disabled && !buddy->ping_received)
			return false;
	return true;
}

static void sync_disable_active_buddies(struct sync_filter_context *ctx)
{
	struct sync_buddy *buddy;

	FOR_EACH_BUDDY(buddy, &ctx->buddies) {
		if (buddy->sbi->disabled)
			continue;
		if (buddy->ping_received == true)
			continue;
		PARA_NOTICE_LOG("disabling %s\n", buddy->sbi->url);
		buddy->sbi->disabled = true;
	}
}

static void sync_set_timeout(struct sync_filter_context *ctx,
		struct lls_parse_result *lpr)
{
	uint32_t ms = FILTER_CMD_OPT_UINT32_VAL(SYNC, TIMEOUT, lpr);
	struct timeval to;

	ms2tv(ms, &to);
	tv_add(now, &to, &ctx->timeout);
}

static void sync_pre_monitor(struct sched *s, void *context)
{
	int ret;
	struct filter_node *fn = context;
	struct sync_filter_context *ctx = fn->private_data;

	if (list_empty(&ctx->buddies))
		return sched_min_delay(s);
	if (ctx->listen_fd < 0)
		return sched_min_delay(s);
	ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
	if (ret < 0)
		return sched_min_delay(s);
	sched_monitor_readfd(ctx->listen_fd, s);
	if (ret == 0)
		return;
	if (ctx->timeout.tv_sec == 0) { /* must ping buddies */
		sync_set_timeout(ctx, fn->lpr);
		return sched_min_delay(s);
	}
	if (sync_complete(ctx)) /* push down what we have */
		return sched_min_delay(s);
	sched_request_barrier_or_min_delay(&ctx->timeout, s);
}

static struct sync_buddy *sync_find_buddy(struct sockaddr *addr,
		struct list_head *list)
{
	struct sync_buddy *buddy;

	FOR_EACH_BUDDY(buddy, list)
		if (sockaddr_equal(buddy->sbi->ai->ai_addr, addr))
			return buddy;
	return NULL;
}

static int sync_post_monitor(__a_unused struct sched *s, void *context)
{
	int ret;
	struct filter_node *fn = context;
	struct sync_filter_context *ctx = fn->private_data;
	struct sync_buddy *buddy, *tmp;

	if (list_empty(&ctx->buddies))
		goto success;
	ret = -E_SYNC_LISTEN_FD;
	if (ctx->listen_fd < 0)
		goto fail;
	ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
	if (ret < 0)
		goto fail;
	if (ret == 0)
		return 0;
	if (ctx->timeout.tv_sec == 0)
		sync_set_timeout(ctx, fn->lpr);
	else {
		if (tv_diff(&ctx->timeout, now, NULL) < 0) {
			sync_disable_active_buddies(ctx);
			goto success;
		}
	}
	if (!ctx->ping_sent) {
		FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies) {
			char c = '\0';
			PARA_INFO_LOG("pinging %s (%s)\n",
				buddy->sbi->url, buddy->sbi->disabled?
				"disabled" : "enabled");
			ret = xwrite(buddy->fd, &c, 1);
			if (ret < 0) {
				PARA_WARNING_LOG("failed to write to %s: %s\n",
					buddy->sbi->url, para_strerror(-ret));
				sync_close_buddy(buddy);
			}
		}
		ctx->ping_sent = true;
	}
	if (sched_read_ok(ctx->listen_fd, s)) {
		char c;
		for (;;) {
			struct sockaddr src_addr;
			socklen_t len = sizeof(src_addr);
			ret = recvfrom(ctx->listen_fd, &c, 1, MSG_DONTWAIT,
				&src_addr, &len);
			if (ret < 0) {
				if (errno == EAGAIN || errno == EWOULDBLOCK)
					break;
				ret = -ERRNO_TO_PARA_ERROR(errno);
				goto fail;
			}
			buddy = sync_find_buddy(&src_addr, &ctx->buddies);
			if (!buddy) {
				PARA_NOTICE_LOG("pinged by unknown\n");
				continue;
			}
			PARA_DEBUG_LOG("pinged by %s\n", buddy->sbi->url);
			if (buddy->sbi->disabled) {
				PARA_NOTICE_LOG("enabling %s\n",
					buddy->sbi->url);
				buddy->sbi->disabled = false;
			}
			buddy->ping_received = true;
		}
	}
	if (!sync_complete(ctx))
		return 1;
	/*
	 * Although all enabled buddies are in sync we do not splice out
	 * ourselves immediately. We rather wait until the timeout expires,
	 * or the buddy list has become empty. This opens a time window
	 * for disabled buddies to become enabled by sending us a packet.
	 */
	btr_pushdown(fn->btrn);
	return 1;
success:
	ret = -E_SYNC_COMPLETE; /* success */
	goto out;
fail:
	if (ret != -E_EOF)
		PARA_WARNING_LOG("%s\n", para_strerror(-ret));
out:
	sync_close_buddies(ctx);
	btr_splice_out_node(&fn->btrn);
	assert(ret < 0);
	return ret;
}

/** \cond doxygen_ignore */
const struct filter lsg_filter_cmd_com_sync_user_data = {
	.setup = sync_setup,
	.open = sync_open,
	.pre_monitor = sync_pre_monitor,
	.post_monitor = sync_post_monitor,
	.close = sync_close,
	.teardown = sync_teardown
};
/** \endcond */