sync_filter.c

   1 /*
   2  * Copyright (C) 2013 Andre Noll <maan@systemlinux.org>
   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
  17 #include "para.h"
  18 #include "sync_filter.cmdline.h"
  19 #include "list.h"
  20 #include "net.h"
  21 #include "sched.h"
  22 #include "ggo.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 /* per open/close data */
  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 struct sync_filter_context {
  46         int listen_fd;
  47         struct list_head buddies;
  48         struct timeval timeout;
  49         bool ping_sent;
  50 };
  51
  52 struct sync_filter_config {
  53         struct sync_filter_args_info *conf;
  54         struct sync_buddy_info *buddy_info;
  55 };
  56
  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)
  61
  62 static void sync_close_buddy(struct sync_buddy *buddy)
  63 {
  64         if (buddy->fd < 0)
  65                 return;
  66         PARA_DEBUG_LOG("closing %s\n", buddy->sbi->url);
  67         close(buddy->fd);
  68         buddy->fd = -1;
  69 }
  70
  71 static void sync_close_buddies(struct sync_filter_context *ctx)
  72 {
  73         struct sync_buddy *buddy;
  74
  75         FOR_EACH_BUDDY(buddy, &ctx->buddies)
  76                 sync_close_buddy(buddy);
  77 }
  78
  79 static void sync_close(struct filter_node *fn)
  80 {
  81         struct sync_filter_context *ctx = fn->private_data;
  82
  83         sync_close_buddies(ctx);
  84         if (ctx->listen_fd >= 0) {
  85                 close(ctx->listen_fd);
  86                 ctx->listen_fd = -1;
  87         }
  88         free(ctx);
  89         fn->private_data = NULL;
  90 }
  91
  92 static void sync_free_config(void *conf)
  93 {
  94         struct sync_filter_config *sfc = conf;
  95         int i;
  96
  97         for (i = 0; i < sfc->conf->buddy_given; i++) {
  98                 free(sfc->buddy_info[i].host);
  99                 freeaddrinfo(sfc->buddy_info[i].ai);
 100         }
 101         sync_filter_cmdline_parser_free(sfc->conf);
 102         free(sfc);
 103 }
 104
 105 static void sync_open(struct filter_node *fn)
 106 {
 107         int i, ret;
 108         struct sync_filter_config *sfc = fn->conf;
 109         struct sync_buddy *buddy;
 110         struct sync_filter_context *ctx;
 111
 112         assert(sfc);
 113
 114         ctx = fn->private_data = para_calloc(sizeof(*ctx));
 115         INIT_LIST_HEAD(&ctx->buddies);
 116         ctx->listen_fd = -1;
 117
 118         /* create socket to listen for incoming packets */
 119         ret = makesock(
 120                 IPPROTO_UDP,
 121                 true /* passive */,
 122                 NULL /* no host required */,
 123                 sfc->conf->port_arg,
 124                 NULL /* no flowopts */
 125         );
 126         if (ret < 0) {
 127                 PARA_ERROR_LOG("could not create UDP listening socket %d\n",
 128                         sfc->conf->port_arg);
 129                 return;
 130         }
 131         ctx->listen_fd = ret;
 132         PARA_INFO_LOG("listening on fd %d\n", ctx->listen_fd);
 133
 134         for (i = 0; i < sfc->conf->buddy_given; i++) {
 135                 struct sync_buddy_info *sbi = sfc->buddy_info + i;
 136                 const char *url = sfc->conf->buddy_arg[i];
 137                 int fd;
 138
 139                 /* make buddy udp socket from address info */
 140                 assert(sbi->ai);
 141                 ret = makesock_addrinfo(
 142                         IPPROTO_UDP,
 143                         false /* not passive */,
 144                         sbi->ai,
 145                         NULL /* no flowopts */
 146                 );
 147                 if (ret < 0) {
 148                         PARA_WARNING_LOG("could not make socket for %s\n",
 149                                 url);
 150                         goto fail;
 151                 }
 152                 fd = ret;
 153                 ret = mark_fd_nonblocking(fd);
 154                 if (ret < 0) {
 155                         PARA_ERROR_LOG("unable to set nonblock mode for %s\n",
 156                                 url);
 157                         close(fd);
 158                         goto fail;
 159                 }
 160                 buddy = para_malloc(sizeof(*buddy));
 161                 buddy->fd = fd;
 162                 buddy->sbi = sbi;
 163                 buddy->ping_received = false;
 164                 para_list_add(&buddy->node, &ctx->buddies);
 165
 166                 PARA_INFO_LOG("opened buddy %s on fd %d\n", url, fd);
 167                 continue;
 168 fail:
 169                 PARA_WARNING_LOG("%s\n", para_strerror(-ret));
 170         }
 171 }
 172
 173 static int sync_parse_config(int argc, char **argv, void **result)
 174 {
 175         int i, ret, n;
 176         struct sync_filter_config *sfc;
 177         struct sync_filter_args_info *conf = para_malloc(sizeof(*conf));
 178
 179         sync_filter_cmdline_parser(argc, argv, conf); /* exits on error */
 180         sfc = para_calloc(sizeof(*sfc));
 181         sfc->conf = conf;
 182         n = conf->buddy_given;
 183         sfc->buddy_info = para_malloc((n + 1) * sizeof(*sfc->buddy_info));
 184         PARA_INFO_LOG("initializing buddy info array of length %d\n", n);
 185         for (i = 0; i < n; i++) {
 186                 const char *url = conf->buddy_arg[i];
 187                 size_t len = strlen(url);
 188                 char *host = para_malloc(len + 1);
 189                 int port;
 190                 struct addrinfo *ai;
 191                 struct sync_buddy_info *sbi = sfc->buddy_info + i;
 192
 193                 if (!parse_url(url, host, len, &port)) {
 194                         free(host);
 195                         PARA_ERROR_LOG("could not parse url %s\n", url);
 196                         ret = -ERRNO_TO_PARA_ERROR(EINVAL);
 197                         goto fail;
 198                 }
 199                 if (port < 0)
 200                         port = conf->port_arg;
 201                 ret = lookup_address(IPPROTO_UDP, false /* not passive */,
 202                         host, port, &ai);
 203                 if (ret < 0) {
 204                         PARA_ERROR_LOG("host lookup failure for %s\n", url);
 205                         free(host);
 206                         goto fail;
 207                 }
 208                 sbi->url = url;
 209                 sbi->host = host;
 210                 sbi->port = port;
 211                 sbi->ai = ai;
 212                 sbi->disabled = false;
 213                 PARA_DEBUG_LOG("buddy #%d: %s\n", i, url);
 214         }
 215         *result = sfc;
 216         return 1;
 217 fail:
 218         assert(ret < 0);
 219         PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 220         sync_free_config(sfc);
 221         return ret;
 222 }
 223
 224 /*
 225  * True if we sent a packet to all budies and received a packet from each
 226  * enabled buddy.
 227  */
 228 static bool sync_complete(struct sync_filter_context *ctx)
 229 {
 230         struct sync_buddy *buddy;
 231
 232         if (!ctx->ping_sent)
 233                 return false;
 234         FOR_EACH_BUDDY(buddy, &ctx->buddies)
 235                 if (!buddy->sbi->disabled && !buddy->ping_received)
 236                         return false;
 237         return true;
 238 }
 239
 240 static void sync_disable_active_buddies(struct sync_filter_context *ctx)
 241 {
 242         struct sync_buddy *buddy;
 243
 244         FOR_EACH_BUDDY(buddy, &ctx->buddies) {
 245                 if (buddy->sbi->disabled)
 246                         continue;
 247                 PARA_NOTICE_LOG("disabling %s\n", buddy->sbi->url);
 248                 buddy->sbi->disabled = true;
 249         }
 250 }
 251
 252 static void sync_set_timeout(struct sync_filter_context *ctx,
 253                 struct sync_filter_config *sfc)
 254 {
 255         struct timeval to;
 256
 257         ms2tv(sfc->conf->timeout_arg, &to);
 258         tv_add(now, &to, &ctx->timeout);
 259 }
 260
 261 static void sync_pre_select(struct sched *s, struct task *t)
 262 {
 263         int ret;
 264         struct filter_node *fn = container_of(t, struct filter_node, task);
 265         struct sync_filter_context *ctx = fn->private_data;
 266         struct sync_filter_config *sfc = fn->conf;
 267
 268         if (list_empty(&ctx->buddies))
 269                 return sched_min_delay(s);
 270         if (ctx->listen_fd < 0)
 271                 return sched_min_delay(s);
 272         ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
 273         if (ret < 0)
 274                 return sched_min_delay(s);
 275         para_fd_set(ctx->listen_fd, &s->rfds, &s->max_fileno);
 276         if (ret == 0)
 277                 return;
 278         if (ctx->timeout.tv_sec == 0) { /* must ping buddies */
 279                 sync_set_timeout(ctx, sfc);
 280                 return sched_min_delay(s);
 281         }
 282         if (sync_complete(ctx)) /* push down what we have */
 283                 return sched_min_delay(s);
 284         sched_request_barrier_or_min_delay(&ctx->timeout, s);
 285 }
 286
 287 static struct sync_buddy *sync_find_buddy(struct sockaddr *addr,
 288                 struct list_head *list)
 289 {
 290         struct sync_buddy *buddy;
 291
 292         FOR_EACH_BUDDY(buddy, list)
 293                 if (sockaddr_equal(buddy->sbi->ai->ai_addr, addr))
 294                         return buddy;
 295         return NULL;
 296 }
 297
 298 static int sync_post_select(__a_unused struct sched *s, struct task *t)
 299 {
 300         int ret;
 301         struct filter_node *fn = container_of(t, struct filter_node, task);
 302         struct sync_filter_context *ctx = fn->private_data;
 303         struct sync_filter_config *sfc = fn->conf;
 304         struct sync_buddy *buddy, *tmp;
 305
 306         if (list_empty(&ctx->buddies))
 307                 goto success;
 308         ret = -E_SYNC_LISTEN_FD;
 309         if (ctx->listen_fd < 0)
 310                 goto fail;
 311         ret = btr_node_status(fn->btrn, 0, BTR_NT_INTERNAL);
 312         if (ret < 0)
 313                 goto fail;
 314         if (ret == 0)
 315                 return 0;
 316         if (ctx->timeout.tv_sec == 0)
 317                 sync_set_timeout(ctx, sfc);
 318         else {
 319                 if (tv_diff(&ctx->timeout, now, NULL) < 0) {
 320                         sync_disable_active_buddies(ctx);
 321                         goto success;
 322                 }
 323         }
 324         if (!ctx->ping_sent) {
 325                 FOR_EACH_BUDDY_SAFE(buddy, tmp, &ctx->buddies) {
 326                         char c = '\0';
 327                         PARA_INFO_LOG("pinging %s (%s)\n",
 328                                 buddy->sbi->url, buddy->sbi->disabled?
 329                                 "disabled" : "enabled");
 330                         ret = xwrite(buddy->fd, &c, 1);
 331                         sync_close_buddy(buddy);
 332                         if (ret < 0) {
 333                                 PARA_WARNING_LOG("failed to write to %s: %s\n",
 334                                         buddy->sbi->url, para_strerror(-ret));
 335                                 list_del(&buddy->node);
 336                         }
 337                 }
 338                 ctx->ping_sent = true;
 339         }
 340         if (FD_ISSET(ctx->listen_fd, &s->rfds)) {
 341                 char c;
 342                 for (;;) {
 343                         struct sockaddr src_addr;
 344                         socklen_t len = sizeof(src_addr);
 345                         ret = recvfrom(ctx->listen_fd, &c, 1, MSG_DONTWAIT,
 346                                 &src_addr, &len);
 347                         if (ret < 0) {
 348                                 if (errno == EAGAIN || errno == EWOULDBLOCK)
 349                                         break;
 350                                 ret = -ERRNO_TO_PARA_ERROR(errno);
 351                                 goto fail;
 352                         }
 353                         buddy = sync_find_buddy(&src_addr, &ctx->buddies);
 354                         if (!buddy) {
 355                                 PARA_NOTICE_LOG("pinged by unknown\n");
 356                                 continue;
 357                         }
 358                         PARA_DEBUG_LOG("pinged by %s\n", buddy->sbi->url);
 359                         if (buddy->sbi->disabled) {
 360                                 PARA_NOTICE_LOG("enabling %s\n",
 361                                         buddy->sbi->url);
 362                                 buddy->sbi->disabled = false;
 363                         }
 364                         list_del(&buddy->node);
 365                 }
 366         }
 367         if (!sync_complete(ctx))
 368                 return 1;
 369         /*
 370          * Although all enabled buddies are in sync we do not splice out
 371          * ourselves immediately. We rather wait until the timout 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.
 374          */
 375         btr_pushdown(fn->btrn);
 376         return 1;
 377 success:
 378         ret = -E_SYNC_COMPLETE; /* success */
 379         goto out;
 380 fail:
 381         PARA_WARNING_LOG("%s\n", para_strerror(-ret));
 382 out:
 383         sync_close_buddies(ctx);
 384         btr_splice_out_node(&fn->btrn);
 385         assert(ret < 0);
 386         return ret;
 387 }
 388
 389 /**
 390  * The synchronization filter.
 391  *
 392  * \param f Pointer to the struct to initialize.
 393  */
 394 void sync_filter_init(struct filter *f)
 395 {
 396         struct sync_filter_args_info dummy;
 397
 398         sync_filter_cmdline_parser_init(&dummy);
 399         f->open = sync_open;
 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);
 406 }