efc10116f788a79bee5e2e75dfdfc0e036d06f1a
[paraslash.git] / sync_filter.c
1 /*
2 * Copyright (C) 2013 Andre Noll <maan@tuebingen.mpg.de>
3 *
4 * Licensed under the GPL v2. For licencing details see COPYING.
5 */
6
7 /** \file sync_filter.c Playback synchronization filter. */
8
9 #include <netinet/in.h>
10 #include <sys/socket.h>
11 #include <regex.h>
12 #include <sys/types.h>
13 #include <arpa/inet.h>
14 #include <sys/un.h>
15 #include <netdb.h>
16 #include <lopsub.h>
17
18 #include "filter_cmd.lsg.h"
19 #include "para.h"
20 #include "list.h"
21 #include "net.h"
22 #include "sched.h"
23 #include "buffer_tree.h"
24 #include "filter.h"
25 #include "string.h"
26 #include "fd.h"
27 #include "error.h"
28
29 struct sync_buddy_info {
30 const char *url;
31 char *host;
32 int port;
33 struct addrinfo *ai;
34 bool disabled;
35 };
36
37 /* One active buddy */
38 struct sync_buddy {
39 int fd;
40 struct sync_buddy_info *sbi;
41 bool ping_received;
42 struct list_head node;
43 };
44
45 /* Allocated in ->open(), stored in fn->private_data */
46 struct sync_filter_context {
47 int listen_fd;
48 struct list_head buddies;
49 struct timeval timeout;
50 bool ping_sent;
51 };
52
53 #define FOR_EACH_BUDDY(_buddy, _list) \
54 list_for_each_entry(_buddy, _list, node)
55 #define FOR_EACH_BUDDY_SAFE(_buddy, _tmp_buddy, _list) \
56 list_for_each_entry_safe(_buddy, _tmp_buddy, _list, node)
57
58 static void sync_close_buddy(struct sync_buddy *buddy)
59 {
60 PARA_DEBUG_LOG("closing %s, fd %d\n", buddy->sbi->url, buddy->fd);
61 close(buddy->fd);
62 list_del(&buddy->node);
63 free(buddy);
64 }
65
66 static void sync_close_buddies(struct sync_filter_context *ctx)
67 {
68 struct sync_buddy *buddy, *tmp;
69
70 FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies)
71 sync_close_buddy(buddy);
72 }
73
74 static void sync_close(struct filter_node *fn)
75 {
76 struct sync_filter_context *ctx = fn->private_data;
77
78 sync_close_buddies(ctx);
79 if (ctx->listen_fd >= 0) {
80 close(ctx->listen_fd);
81 ctx->listen_fd = -1;
82 }
83 free(ctx);
84 fn->private_data = NULL;
85 }
86
87 static void sync_teardown(const struct lls_parse_result *lpr, void *conf)
88 {
89 struct sync_buddy_info *sbi = conf;
90 int i, num_buddies = FILTER_CMD_OPT_GIVEN(SYNC, BUDDY, lpr);
91
92 for (i = 0; i < num_buddies; i++) {
93 free(sbi[i].host);
94 freeaddrinfo(sbi[i].ai);
95 }
96 free(sbi);
97 }
98
99 static void sync_open(struct filter_node *fn)
100 {
101 int i, ret;
102 struct sync_buddy *buddy;
103 struct sync_filter_context *ctx;
104 struct sync_buddy_info *sbi = fn->conf;
105 uint32_t port = FILTER_CMD_OPT_UINT32_VAL(SYNC, PORT, fn->lpr);
106 unsigned buddy_given;
107 const struct lls_opt_result *r_b;
108
109 ctx = fn->private_data = para_calloc(sizeof(*ctx));
110 INIT_LIST_HEAD(&ctx->buddies);
111
112 /* create socket to listen for incoming packets */
113 ret = makesock(
114 IPPROTO_UDP,
115 true /* passive */,
116 NULL /* no host required */,
117 port,
118 NULL /* no flowopts */
119 );
120 if (ret < 0) {
121 PARA_ERROR_LOG("could not create UDP listening socket %u\n",
122 port);
123 return;
124 }
125 ctx->listen_fd = ret;
126 PARA_INFO_LOG("listening on fd %d\n", ctx->listen_fd);
127
128 r_b = FILTER_CMD_OPT_RESULT(SYNC, BUDDY, fn->lpr);
129 buddy_given = lls_opt_given(r_b);
130 for (i = 0; i < buddy_given; i++) {
131 int fd;
132 const char *url = lls_string_val(i, r_b);
133
134 /* make buddy udp socket from address info */
135 assert(sbi->ai);
136 ret = makesock_addrinfo(
137 IPPROTO_UDP,
138 false /* not passive */,
139 sbi[i].ai,
140 NULL /* no flowopts */
141 );
142 if (ret < 0) {
143 PARA_WARNING_LOG("could not make socket for %s\n",
144 url);
145 goto fail;
146 }
147 fd = ret;
148 ret = mark_fd_nonblocking(fd);
149 if (ret < 0) {
150 PARA_ERROR_LOG("unable to set nonblock mode for %s\n",
151 url);
152 close(fd);
153 goto fail;
154 }
155 buddy = para_malloc(sizeof(*buddy));
156 buddy->fd = fd;
157 buddy->sbi = sbi + i;
158 buddy->ping_received = false;
159 para_list_add(&buddy->node, &ctx->buddies);
160
161 PARA_INFO_LOG("opened buddy %s on fd %d\n", url, fd);
162 continue;
163 fail:
164 PARA_WARNING_LOG("%s\n", para_strerror(-ret));
165 }
166 }
167
168 /*
169 * Build an array of struct sync_buddy_info with one entry for each buddy given
170 * in the arguments. This array is not affected by sync_close(), so information
171 * stored there can be used for multiple instances (para_audiod). We store the
172 * resolved url and the ->disabled bit in this array.
173 */
174 static void *sync_setup(const struct lls_parse_result *lpr)
175 {
176 int i, ret;
177 unsigned n;
178 struct sync_buddy_info *sbi;
179 const struct lls_opt_result *r_b;
180
181 r_b = FILTER_CMD_OPT_RESULT(SYNC, BUDDY, lpr);
182 n = lls_opt_given(r_b);
183 sbi = para_malloc(n * sizeof(*sbi));
184 PARA_INFO_LOG("initializing buddy info array of length %u\n", n);
185 for (i = 0; i < n; i++) {
186 const char *url = lls_string_val(i, r_b);
187 size_t len = strlen(url);
188 char *host = para_malloc(len + 1);
189 int port;
190 struct addrinfo *ai;
191
192 if (!parse_url(url, host, len, &port)) {
193 PARA_ERROR_LOG("could not parse url %s\n", url);
194 exit(EXIT_FAILURE);
195 }
196 if (port < 0)
197 port = FILTER_CMD_OPT_UINT32_VAL(SYNC, PORT, lpr);
198 ret = lookup_address(IPPROTO_UDP, false /* not passive */,
199 host, port, &ai);
200 if (ret < 0) {
201 PARA_ERROR_LOG("host lookup failure for %s: %s\n",
202 url, para_strerror(-ret));
203 exit(EXIT_FAILURE);
204 }
205 sbi[i].url = url;
206 sbi[i].host = host;
207 sbi[i].port = port;
208 sbi[i].ai = ai;
209 sbi[i].disabled = false;
210 PARA_DEBUG_LOG("buddy #%d: %s\n", i, url);
211 }
212 return sbi;
213 }
214
215 /*
216 * True if we sent a packet to all buddies and received a packet from each
217 * enabled buddy.
218 */
219 static bool sync_complete(struct sync_filter_context *ctx)
220 {
221 struct sync_buddy *buddy;
222
223 if (!ctx->ping_sent)
224 return false;
225 FOR_EACH_BUDDY(buddy, &ctx->buddies)
226 if (!buddy->sbi->disabled && !buddy->ping_received)
227 return false;
228 return true;
229 }
230
231 static void sync_disable_active_buddies(struct sync_filter_context *ctx)
232 {
233 struct sync_buddy *buddy;
234
235 FOR_EACH_BUDDY(buddy, &ctx->buddies) {
236 if (buddy->sbi->disabled)
237 continue;
238 if (buddy->ping_received == true)
239 continue;
240 PARA_NOTICE_LOG("disabling %s\n", buddy->sbi->url);
241 buddy->sbi->disabled = true;
242 }
243 }
244
245 static void sync_set_timeout(struct sync_filter_context *ctx,
246 struct lls_parse_result *lpr)
247 {
248 uint32_t ms = FILTER_CMD_OPT_UINT32_VAL(SYNC, TIMEOUT, lpr);
249 struct timeval to;
250
251 ms2tv(ms, &to);
252 tv_add(now, &to, &ctx->timeout);
253 }
254
255 static void sync_pre_select(struct sched *s, void *context)
256 {
257 int ret;
258 struct filter_node *fn = context;
259 struct sync_filter_context *ctx = fn->private_data;
260
261 if (list_empty(&ctx->buddies))
262 return sched_min_delay(s);
263 if (ctx->listen_fd < 0)
264 return sched_min_delay(s);
265 ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
266 if (ret < 0)
267 return sched_min_delay(s);
268 para_fd_set(ctx->listen_fd, &s->rfds, &s->max_fileno);
269 if (ret == 0)
270 return;
271 if (ctx->timeout.tv_sec == 0) { /* must ping buddies */
272 sync_set_timeout(ctx, fn->lpr);
273 return sched_min_delay(s);
274 }
275 if (sync_complete(ctx)) /* push down what we have */
276 return sched_min_delay(s);
277 sched_request_barrier_or_min_delay(&ctx->timeout, s);
278 }
279
280 static struct sync_buddy *sync_find_buddy(struct sockaddr *addr,
281 struct list_head *list)
282 {
283 struct sync_buddy *buddy;
284
285 FOR_EACH_BUDDY(buddy, list)
286 if (sockaddr_equal(buddy->sbi->ai->ai_addr, addr))
287 return buddy;
288 return NULL;
289 }
290
291 static int sync_post_select(__a_unused struct sched *s, void *context)
292 {
293 int ret;
294 struct filter_node *fn = context;
295 struct sync_filter_context *ctx = fn->private_data;
296 struct sync_buddy *buddy, *tmp;
297
298 if (list_empty(&ctx->buddies))
299 goto success;
300 ret = -E_SYNC_LISTEN_FD;
301 if (ctx->listen_fd < 0)
302 goto fail;
303 ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
304 if (ret < 0)
305 goto fail;
306 if (ret == 0)
307 return 0;
308 if (ctx->timeout.tv_sec == 0)
309 sync_set_timeout(ctx, fn->lpr);
310 else {
311 if (tv_diff(&ctx->timeout, now, NULL) < 0) {
312 sync_disable_active_buddies(ctx);
313 goto success;
314 }
315 }
316 if (!ctx->ping_sent) {
317 FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies) {
318 char c = '\0';
319 PARA_INFO_LOG("pinging %s (%s)\n",
320 buddy->sbi->url, buddy->sbi->disabled?
321 "disabled" : "enabled");
322 ret = xwrite(buddy->fd, &c, 1);
323 if (ret < 0) {
324 PARA_WARNING_LOG("failed to write to %s: %s\n",
325 buddy->sbi->url, para_strerror(-ret));
326 sync_close_buddy(buddy);
327 }
328 }
329 ctx->ping_sent = true;
330 }
331 if (FD_ISSET(ctx->listen_fd, &s->rfds)) {
332 char c;
333 for (;;) {
334 struct sockaddr src_addr;
335 socklen_t len = sizeof(src_addr);
336 ret = recvfrom(ctx->listen_fd, &c, 1, MSG_DONTWAIT,
337 &src_addr, &len);
338 if (ret < 0) {
339 if (errno == EAGAIN || errno == EWOULDBLOCK)
340 break;
341 ret = -ERRNO_TO_PARA_ERROR(errno);
342 goto fail;
343 }
344 buddy = sync_find_buddy(&src_addr, &ctx->buddies);
345 if (!buddy) {
346 PARA_NOTICE_LOG("pinged by unknown\n");
347 continue;
348 }
349 PARA_DEBUG_LOG("pinged by %s\n", buddy->sbi->url);
350 if (buddy->sbi->disabled) {
351 PARA_NOTICE_LOG("enabling %s\n",
352 buddy->sbi->url);
353 buddy->sbi->disabled = false;
354 }
355 buddy->ping_received = true;
356 }
357 }
358 if (!sync_complete(ctx))
359 return 1;
360 /*
361 * Although all enabled buddies are in sync we do not splice out
362 * ourselves immediately. We rather wait until the timeout expires,
363 * or the buddy list has become empty. This opens a time window
364 * for disabled buddies to become enabled by sending us a packet.
365 */
366 btr_pushdown(fn->btrn);
367 return 1;
368 success:
369 ret = -E_SYNC_COMPLETE; /* success */
370 goto out;
371 fail:
372 PARA_WARNING_LOG("%s\n", para_strerror(-ret));
373 out:
374 sync_close_buddies(ctx);
375 btr_splice_out_node(&fn->btrn);
376 assert(ret < 0);
377 return ret;
378 }
379
380 const struct filter lsg_filter_cmd_com_sync_user_data = {
381 .setup = sync_setup,
382 .open = sync_open,
383 .pre_select = sync_pre_select,
384 .post_select = sync_post_select,
385 .close = sync_close,
386 .teardown = sync_teardown
387 };