2 * Copyright (C) 2013 Andre Noll <maan@tuebingen.mpg.de>
4 * Licensed under the GPL v2. For licencing details see COPYING.
7 /** \file sync_filter.c Playback synchronization filter. */
9 #include <netinet/in.h>
10 #include <sys/socket.h>
12 #include <sys/types.h>
13 #include <arpa/inet.h>
18 #include "sync_filter.cmdline.h"
23 #include "buffer_tree.h"
29 struct sync_buddy_info {
37 /* per open/close data */
40 struct sync_buddy_info *sbi;
42 struct list_head node;
45 struct sync_filter_context {
47 struct list_head buddies;
48 struct timeval timeout;
52 struct sync_filter_config {
53 struct sync_filter_args_info *conf;
54 struct sync_buddy_info *buddy_info;
57 #define FOR_EACH_BUDDY(_buddy, _list) \
58 list_for_each_entry(_buddy, _list, node)
59 #define FOR_EACH_BUDDY_SAFE(_buddy, _tmp_buddy, _list) \
60 list_for_each_entry_safe(_buddy, _tmp_buddy, _list, node)
62 static void sync_close_buddy(struct sync_buddy *buddy)
64 PARA_DEBUG_LOG("closing %s, fd %d\n", buddy->sbi->url, buddy->fd);
66 list_del(&buddy->node);
70 static void sync_close_buddies(struct sync_filter_context *ctx)
72 struct sync_buddy *buddy, *tmp;
74 FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies)
75 sync_close_buddy(buddy);
78 static void sync_close(struct filter_node *fn)
80 struct sync_filter_context *ctx = fn->private_data;
82 sync_close_buddies(ctx);
83 if (ctx->listen_fd >= 0) {
84 close(ctx->listen_fd);
88 fn->private_data = NULL;
91 static void sync_free_config(void *conf)
93 struct sync_filter_config *sfc = conf;
96 for (i = 0; i < sfc->conf->buddy_given; i++) {
97 free(sfc->buddy_info[i].host);
98 freeaddrinfo(sfc->buddy_info[i].ai);
100 sync_filter_cmdline_parser_free(sfc->conf);
104 static void sync_open(struct filter_node *fn)
107 struct sync_filter_config *sfc = fn->conf;
108 struct sync_buddy *buddy;
109 struct sync_filter_context *ctx;
113 ctx = fn->private_data = para_calloc(sizeof(*ctx));
114 INIT_LIST_HEAD(&ctx->buddies);
117 /* create socket to listen for incoming packets */
121 NULL /* no host required */,
123 NULL /* no flowopts */
126 PARA_ERROR_LOG("could not create UDP listening socket %d\n",
127 sfc->conf->port_arg);
130 ctx->listen_fd = ret;
131 PARA_INFO_LOG("listening on fd %d\n", ctx->listen_fd);
133 for (i = 0; i < sfc->conf->buddy_given; i++) {
134 struct sync_buddy_info *sbi = sfc->buddy_info + i;
135 const char *url = sfc->conf->buddy_arg[i];
138 /* make buddy udp socket from address info */
140 ret = makesock_addrinfo(
142 false /* not passive */,
144 NULL /* no flowopts */
147 PARA_WARNING_LOG("could not make socket for %s\n",
152 ret = mark_fd_nonblocking(fd);
154 PARA_ERROR_LOG("unable to set nonblock mode for %s\n",
159 buddy = para_malloc(sizeof(*buddy));
162 buddy->ping_received = false;
163 para_list_add(&buddy->node, &ctx->buddies);
165 PARA_INFO_LOG("opened buddy %s on fd %d\n", url, fd);
168 PARA_WARNING_LOG("%s\n", para_strerror(-ret));
172 static int sync_parse_config(int argc, char **argv, void **result)
175 struct sync_filter_config *sfc;
176 struct sync_filter_args_info *conf = para_malloc(sizeof(*conf));
178 sync_filter_cmdline_parser(argc, argv, conf); /* exits on error */
179 sfc = para_calloc(sizeof(*sfc));
181 n = conf->buddy_given;
182 sfc->buddy_info = para_malloc((n + 1) * sizeof(*sfc->buddy_info));
183 PARA_INFO_LOG("initializing buddy info array of length %d\n", n);
184 for (i = 0; i < n; i++) {
185 const char *url = conf->buddy_arg[i];
186 size_t len = strlen(url);
187 char *host = para_malloc(len + 1);
190 struct sync_buddy_info *sbi = sfc->buddy_info + i;
192 if (!parse_url(url, host, len, &port)) {
194 PARA_ERROR_LOG("could not parse url %s\n", url);
195 ret = -ERRNO_TO_PARA_ERROR(EINVAL);
199 port = conf->port_arg;
200 ret = lookup_address(IPPROTO_UDP, false /* not passive */,
203 PARA_ERROR_LOG("host lookup failure for %s\n", url);
211 sbi->disabled = false;
212 PARA_DEBUG_LOG("buddy #%d: %s\n", i, url);
218 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
219 sync_free_config(sfc);
224 * True if we sent a packet to all buddies and received a packet from each
227 static bool sync_complete(struct sync_filter_context *ctx)
229 struct sync_buddy *buddy;
233 FOR_EACH_BUDDY(buddy, &ctx->buddies)
234 if (!buddy->sbi->disabled && !buddy->ping_received)
239 static void sync_disable_active_buddies(struct sync_filter_context *ctx)
241 struct sync_buddy *buddy;
243 FOR_EACH_BUDDY(buddy, &ctx->buddies) {
244 if (buddy->sbi->disabled)
246 if (buddy->ping_received == true)
248 PARA_NOTICE_LOG("disabling %s\n", buddy->sbi->url);
249 buddy->sbi->disabled = true;
253 static void sync_set_timeout(struct sync_filter_context *ctx,
254 struct sync_filter_config *sfc)
258 ms2tv(sfc->conf->timeout_arg, &to);
259 tv_add(now, &to, &ctx->timeout);
262 static void sync_pre_select(struct sched *s, void *context)
265 struct filter_node *fn = context;
266 struct sync_filter_context *ctx = fn->private_data;
267 struct sync_filter_config *sfc = fn->conf;
269 if (list_empty(&ctx->buddies))
270 return sched_min_delay(s);
271 if (ctx->listen_fd < 0)
272 return sched_min_delay(s);
273 ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
275 return sched_min_delay(s);
276 para_fd_set(ctx->listen_fd, &s->rfds, &s->max_fileno);
279 if (ctx->timeout.tv_sec == 0) { /* must ping buddies */
280 sync_set_timeout(ctx, sfc);
281 return sched_min_delay(s);
283 if (sync_complete(ctx)) /* push down what we have */
284 return sched_min_delay(s);
285 sched_request_barrier_or_min_delay(&ctx->timeout, s);
288 static struct sync_buddy *sync_find_buddy(struct sockaddr *addr,
289 struct list_head *list)
291 struct sync_buddy *buddy;
293 FOR_EACH_BUDDY(buddy, list)
294 if (sockaddr_equal(buddy->sbi->ai->ai_addr, addr))
299 static int sync_post_select(__a_unused struct sched *s, void *context)
302 struct filter_node *fn = context;
303 struct sync_filter_context *ctx = fn->private_data;
304 struct sync_filter_config *sfc = fn->conf;
305 struct sync_buddy *buddy, *tmp;
307 if (list_empty(&ctx->buddies))
309 ret = -E_SYNC_LISTEN_FD;
310 if (ctx->listen_fd < 0)
312 ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
317 if (ctx->timeout.tv_sec == 0)
318 sync_set_timeout(ctx, sfc);
320 if (tv_diff(&ctx->timeout, now, NULL) < 0) {
321 sync_disable_active_buddies(ctx);
325 if (!ctx->ping_sent) {
326 FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies) {
328 PARA_INFO_LOG("pinging %s (%s)\n",
329 buddy->sbi->url, buddy->sbi->disabled?
330 "disabled" : "enabled");
331 ret = xwrite(buddy->fd, &c, 1);
333 PARA_WARNING_LOG("failed to write to %s: %s\n",
334 buddy->sbi->url, para_strerror(-ret));
335 sync_close_buddy(buddy);
338 ctx->ping_sent = true;
340 if (FD_ISSET(ctx->listen_fd, &s->rfds)) {
343 struct sockaddr src_addr;
344 socklen_t len = sizeof(src_addr);
345 ret = recvfrom(ctx->listen_fd, &c, 1, MSG_DONTWAIT,
348 if (errno == EAGAIN || errno == EWOULDBLOCK)
350 ret = -ERRNO_TO_PARA_ERROR(errno);
353 buddy = sync_find_buddy(&src_addr, &ctx->buddies);
355 PARA_NOTICE_LOG("pinged by unknown\n");
358 PARA_DEBUG_LOG("pinged by %s\n", buddy->sbi->url);
359 if (buddy->sbi->disabled) {
360 PARA_NOTICE_LOG("enabling %s\n",
362 buddy->sbi->disabled = false;
364 buddy->ping_received = true;
367 if (!sync_complete(ctx))
370 * Although all enabled buddies are in sync we do not splice out
371 * ourselves immediately. We rather wait until the timeout expires,
372 * or the buddy list has become empty. This opens a time window
373 * for disabled buddies to become enabled by sending us a packet.
375 btr_pushdown(fn->btrn);
378 ret = -E_SYNC_COMPLETE; /* success */
381 PARA_WARNING_LOG("%s\n", para_strerror(-ret));
383 sync_close_buddies(ctx);
384 btr_splice_out_node(&fn->btrn);
390 * The synchronization filter.
392 * \param f Pointer to the struct to initialize.
394 void sync_filter_init(struct filter *f)
396 struct sync_filter_args_info dummy;
398 sync_filter_cmdline_parser_init(&dummy);
400 f->close = sync_close;
401 f->pre_select = sync_pre_select;
402 f->post_select = sync_post_select;
403 f->parse_config = sync_parse_config;
404 f->free_config = sync_free_config;
405 f->help = (struct ggo_help)DEFINE_GGO_HELP(sync_filter);