vss.c

   1 /*
   2  * Copyright (C) 1997 Andre Noll <maan@tuebingen.mpg.de>
   3  *
   4  * Licensed under the GPL v2. For licencing details see COPYING.
   5  */
   6
   7 /** \file vss.c The virtual streaming system.
   8  *
   9  * This contains the audio streaming code of para_server which is independent
  10  * of the current audio format, audio file selector and of the activated
  11  * senders.
  12  */
  13
  14 #include <sys/socket.h>
  15 #include <netinet/in.h>
  16 #include <regex.h>
  17 #include <osl.h>
  18 #include <sys/types.h>
  19 #include <arpa/inet.h>
  20 #include <sys/un.h>
  21 #include <netdb.h>
  22
  23 #include "para.h"
  24 #include "error.h"
  25 #include "portable_io.h"
  26 #include "fec.h"
  27 #include "string.h"
  28 #include "afh.h"
  29 #include "afs.h"
  30 #include "server.h"
  31 #include "net.h"
  32 #include "server.cmdline.h"
  33 #include "list.h"
  34 #include "send.h"
  35 #include "sched.h"
  36 #include "vss.h"
  37 #include "ipc.h"
  38 #include "fd.h"
  39
  40 extern struct misc_meta_data *mmd;
  41
  42 extern void dccp_send_init(struct sender *);
  43 extern void http_send_init(struct sender *);
  44 extern void udp_send_init(struct sender *);
  45
  46 /** The list of supported senders. */
  47 struct sender senders[] = {
  48         {
  49                 .name = "http",
  50                 .init = http_send_init,
  51         },
  52         {
  53                 .name = "dccp",
  54                 .init = dccp_send_init,
  55         },
  56         {
  57                 .name = "udp",
  58                 .init = udp_send_init,
  59         },
  60         {
  61                 .name = NULL,
  62         }
  63 };
  64
  65 /** The possible states of the afs socket. */
  66 enum afs_socket_status {
  67         /** Socket is inactive. */
  68         AFS_SOCKET_READY,
  69         /** Socket fd was included in the write fd set for select(). */
  70         AFS_SOCKET_CHECK_FOR_WRITE,
  71         /** vss wrote a request to the socket and waits for reply from afs. */
  72         AFS_SOCKET_AFD_PENDING
  73 };
  74
  75 /** The task structure for the virtual streaming system. */
  76 struct vss_task {
  77         /** Copied from the -announce_time command line option. */
  78         struct timeval announce_tv;
  79         /** End of the announcing interval. */
  80         struct timeval data_send_barrier;
  81         /** End of the EOF interval. */
  82         struct timeval eof_barrier;
  83         /** Only used if --autoplay_delay was given. */
  84         struct timeval autoplay_barrier;
  85         /** Used for afs-server communication. */
  86         int afs_socket;
  87         /** The current state of \a afs_socket. */
  88         enum afs_socket_status afsss;
  89         /** The memory mapped audio file. */
  90         char *map;
  91         /** The size of the memory mapping. */
  92         size_t mapsize;
  93         /** Used by the scheduler. */
  94         struct task *task;
  95         /** Pointer to the header of the mapped audio file. */
  96         char *header_buf;
  97         /** Length of the audio file header. */
  98         size_t header_len;
  99         /** Time between audio file headers are sent. */
 100         struct timeval header_interval;
 101         /* Only used if afh supports dynamic chunks. */
 102         void *afh_context;
 103 };
 104
 105 /**
 106  * The list of currently connected fec clients.
 107  *
 108  * Senders may use \ref vss_add_fec_client() to add entries to the list.
 109  */
 110 static struct list_head fec_client_list;
 111
 112 /**
 113  * Data associated with one FEC group.
 114  *
 115  * A FEC group consists of a fixed number of slices and this number is given by
 116  * the \a slices_per_group parameter of struct \ref fec_client_parms. Each FEC
 117  * group contains a number of chunks of the current audio file.
 118  *
 119  * FEC slices directly correspond to the data packages sent by the paraslash
 120  * senders that use FEC. Each slice is identified by its group number and its
 121  * number within the group. All slices have the same size, but the last slice
 122  * of the group may not be filled entirely.
 123  */
 124 struct fec_group {
 125         /** The number of the FEC group. */
 126         uint32_t num;
 127         /** Number of bytes in this group. */
 128         uint32_t bytes;
 129         /** The first chunk of the current audio file belonging to the group. */
 130         uint32_t first_chunk;
 131         /** The number of chunks contained in this group. */
 132         uint32_t num_chunks;
 133         /** When the first chunk was sent. */
 134         struct timeval start;
 135         /** The duration of the full group. */
 136         struct timeval duration;
 137         /** The group duration divided by the number of slices. */
 138         struct timeval slice_duration;
 139         /** Group contains the audio file header that occupies that many slices. */
 140         uint8_t num_header_slices;
 141         /** Number of bytes per slice for this group. */
 142         uint16_t slice_bytes;
 143 };
 144
 145 /** A FEC client is always in one of these states. */
 146 enum fec_client_state {
 147         FEC_STATE_NONE = 0,     /**< not initialized and not enabled */
 148         FEC_STATE_DISABLED,     /**< temporarily disabled */
 149         FEC_STATE_READY_TO_RUN  /**< initialized and enabled */
 150 };
 151
 152 /**
 153  * Describes one connected FEC client.
 154  */
 155 struct fec_client {
 156         /** Current state of the client */
 157         enum fec_client_state state;
 158         /** The connected sender client (transport layer). */
 159         struct sender_client *sc;
 160         /** Parameters requested by the client. */
 161         struct fec_client_parms *fcp;
 162         /** Used by the core FEC code. */
 163         struct fec_parms *parms;
 164         /** The position of this client in the fec client list. */
 165         struct list_head node;
 166         /** When the first slice for this client was sent. */
 167         struct timeval stream_start;
 168         /** The first chunk sent to this FEC client. */
 169         int first_stream_chunk;
 170         /** Describes the current group. */
 171         struct fec_group group;
 172         /** The current slice. */
 173         uint8_t current_slice_num;
 174         /** The data to be FEC-encoded (point to a region within the mapped audio file). */
 175         const unsigned char **src_data;
 176         /** Last time an audio  header was sent. */
 177         struct timeval next_header_time;
 178         /** Used for the last source pointer of an audio file. */
 179         unsigned char *extra_src_buf;
 180         /** Needed for the last slice of the audio file header. */
 181         unsigned char *extra_header_buf;
 182         /** Extra slices needed to store largest chunk + header. */
 183         int num_extra_slices;
 184         /** Contains the FEC-encoded data. */
 185         unsigned char *enc_buf;
 186         /** Maximal packet size. */
 187         int mps;
 188 };
 189
 190 /**
 191  * Get the chunk time of the current audio file.
 192  *
 193  * \return A pointer to a struct containing the chunk time, or NULL,
 194  * if currently no audio file is selected.
 195  */
 196 struct timeval *vss_chunk_time(void)
 197 {
 198         if (mmd->afd.afhi.chunk_tv.tv_sec == 0 &&
 199                         mmd->afd.afhi.chunk_tv.tv_usec == 0)
 200                 return NULL;
 201         return &mmd->afd.afhi.chunk_tv;
 202 }
 203
 204 /**
 205  * Write a fec header to a buffer.
 206  *
 207  * \param buf The buffer to write to.
 208  * \param h The fec header to write.
 209  */
 210 static void write_fec_header(struct fec_client *fc, struct vss_task *vsst)
 211 {
 212         char *buf = (char *)fc->enc_buf;
 213         struct fec_group *g = &fc->group;
 214         struct fec_client_parms *p = fc->fcp;
 215
 216         write_u32(buf, FEC_MAGIC);
 217
 218         write_u8(buf + 4, p->slices_per_group + fc->num_extra_slices);
 219         write_u8(buf + 5, p->data_slices_per_group + fc->num_extra_slices);
 220         write_u32(buf + 6, g->num_header_slices? vsst->header_len : 0);
 221
 222         write_u32(buf + 10, g->num);
 223         write_u32(buf + 14, g->bytes);
 224
 225         write_u8(buf + 18, fc->current_slice_num);
 226         write_u8(buf + 19, 0); /* unused */
 227         write_u16(buf + 20, g->slice_bytes);
 228         write_u8(buf + 22, g->first_chunk? 0 : 1);
 229         write_u8(buf + 23, vsst->header_len? 1 : 0);
 230         memset(buf + 24, 0, 8);
 231 }
 232
 233 static bool need_audio_header(struct fec_client *fc, struct vss_task *vsst)
 234 {
 235         if (!mmd->current_chunk) {
 236                 tv_add(now, &vsst->header_interval, &fc->next_header_time);
 237                 return false;
 238         }
 239         if (!vsst->header_buf)
 240                 return false;
 241         if (vsst->header_len == 0)
 242                 return false;
 243         if (fc->group.num > 0) {
 244                 if (!fc->fcp->need_periodic_header)
 245                         return false;
 246                 if (tv_diff(&fc->next_header_time, now, NULL) > 0)
 247                         return false;
 248         }
 249         tv_add(now, &vsst->header_interval, &fc->next_header_time);
 250         return true;
 251 }
 252
 253 static bool need_data_slices(struct fec_client *fc, struct vss_task *vsst)
 254 {
 255         if (fc->group.num > 0)
 256                 return true;
 257         if (!vsst->header_buf)
 258                 return true;
 259         if (vsst->header_len == 0)
 260                 return true;
 261         if (fc->fcp->need_periodic_header)
 262                 return true;
 263         return false;
 264 }
 265
 266 static int num_slices(long unsigned bytes, int max_payload, int rs)
 267 {
 268         int ret;
 269
 270         assert(max_payload > 0);
 271         assert(rs > 0);
 272         ret = DIV_ROUND_UP(bytes, max_payload);
 273         if (ret + rs > 255)
 274                 return -E_BAD_CT;
 275         return ret;
 276 }
 277
 278 /* set group start and group duration */
 279 static void set_group_timing(struct fec_client *fc, struct vss_task *vsst)
 280 {
 281         struct fec_group *g = &fc->group;
 282         struct timeval *chunk_tv = vss_chunk_time();
 283
 284         if (!need_data_slices(fc, vsst))
 285                 ms2tv(200, &g->duration);
 286         else
 287                 tv_scale(g->num_chunks, chunk_tv, &g->duration);
 288         tv_divide(fc->fcp->slices_per_group + fc->num_extra_slices,
 289                 &g->duration, &g->slice_duration);
 290         PARA_DEBUG_LOG("durations (group/chunk/slice): %lu/%lu/%lu\n",
 291                 tv2ms(&g->duration), tv2ms(chunk_tv), tv2ms(&g->slice_duration));
 292 }
 293
 294 static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst)
 295 {
 296         int k, n, ret;
 297         int hs, ds, rs; /* header/data/redundant slices */
 298         struct fec_client_parms *fcp = fc->fcp;
 299
 300         /* set mps */
 301         if (fcp->init_fec) {
 302                 /*
 303                  * Set the maximum slice size to the Maximum Packet Size if the
 304                  * transport protocol allows to determine this value. The user
 305                  * can specify a slice size up to this value.
 306                  */
 307                 ret = fcp->init_fec(fc->sc);
 308                 if (ret < 0)
 309                         return ret;
 310                 fc->mps = ret;
 311         } else
 312                 fc->mps = generic_max_transport_msg_size(fc->sc->fd);
 313         if (fc->mps <= FEC_HEADER_SIZE)
 314                 return -ERRNO_TO_PARA_ERROR(EINVAL);
 315
 316         rs = fc->fcp->slices_per_group - fc->fcp->data_slices_per_group;
 317         ret = num_slices(vsst->header_len, fc->mps - FEC_HEADER_SIZE, rs);
 318         if (ret < 0)
 319                 return ret;
 320         hs = ret;
 321         ret = num_slices(mmd->afd.max_chunk_size, fc->mps - FEC_HEADER_SIZE, rs);
 322         if (ret < 0)
 323                 return ret;
 324         ds = ret;
 325         if (fc->fcp->need_periodic_header)
 326                 k = hs + ds;
 327         else
 328                 k = PARA_MAX(hs, ds);
 329         if (k < fc->fcp->data_slices_per_group)
 330                 k = fc->fcp->data_slices_per_group;
 331         fc->num_extra_slices = k - fc->fcp->data_slices_per_group;
 332         n = k + rs;
 333         fec_free(fc->parms);
 334         ret = fec_new(k, n, &fc->parms);
 335         if (ret < 0)
 336                 return ret;
 337         PARA_INFO_LOG("mps: %d, k: %d, n: %d, extra slices: %d\n",
 338                 fc->mps, k, n, fc->num_extra_slices);
 339         fc->src_data = para_realloc(fc->src_data, k * sizeof(char *));
 340         fc->enc_buf = para_realloc(fc->enc_buf, fc->mps);
 341         fc->extra_src_buf = para_realloc(fc->extra_src_buf, fc->mps);
 342         fc->extra_header_buf = para_realloc(fc->extra_header_buf, fc->mps);
 343
 344         fc->state = FEC_STATE_READY_TO_RUN;
 345         fc->next_header_time.tv_sec = 0;
 346         fc->stream_start = *now;
 347         fc->first_stream_chunk = mmd->current_chunk;
 348         return 1;
 349 }
 350
 351 static void vss_get_chunk(int chunk_num, struct vss_task *vsst,
 352                 char **buf, size_t *sz)
 353 {
 354         int ret;
 355
 356         /*
 357          * Chunk zero is special for header streams: It is the first portion of
 358          * the audio file which consists of the audio file header. It may be
 359          * arbitrary large due to embedded meta data. Audio format handlers may
 360          * replace the header by a stripped one with meta data omitted which is
 361          * of bounded size. We always use the stripped header for streaming
 362          * rather than the unmodified header (chunk zero).
 363          */
 364         if (chunk_num == 0 && vsst->header_len > 0) {
 365                 assert(vsst->header_buf);
 366                 *buf = vsst->header_buf; /* stripped header */
 367                 *sz = vsst->header_len;
 368                 return;
 369         }
 370         ret = afh_get_chunk(chunk_num, &mmd->afd.afhi,
 371                 mmd->afd.audio_format_id, vsst->map, vsst->mapsize,
 372                 (const char **)buf, sz, &vsst->afh_context);
 373         if (ret < 0) {
 374                 PARA_WARNING_LOG("could not get chunk %d: %s\n",
 375                         chunk_num, para_strerror(-ret));
 376                 *buf = NULL;
 377                 *sz = 0;
 378         }
 379 }
 380
 381 static void compute_group_size(struct vss_task *vsst, struct fec_group *g,
 382                 int max_bytes)
 383 {
 384         char *buf;
 385         size_t len;
 386         int i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time()));
 387
 388         if (g->first_chunk == 0) {
 389                 g->num_chunks = 1;
 390                 vss_get_chunk(0, vsst, &buf, &len);
 391                 g->bytes = len;
 392                 return;
 393         }
 394
 395         g->num_chunks = 0;
 396         g->bytes = 0;
 397         /*
 398          * Include chunks into the group until the group duration is at least
 399          * 150ms.  For ogg and wma, a single chunk's duration (ogg page/wma
 400          * super frame) is already larger than 150ms, so a FEC group consists
 401          * of exactly one chunk for these audio formats.
 402          */
 403         for (i = 0;; i++) {
 404                 int chunk_num = g->first_chunk + i;
 405
 406                 if (g->bytes > 0 && i >= max_chunks) /* duration limit */
 407                         break;
 408                 if (chunk_num >= mmd->afd.afhi.chunks_total) /* eof */
 409                         break;
 410                 vss_get_chunk(chunk_num, vsst, &buf, &len);
 411                 if (g->bytes + len > max_bytes)
 412                         break;
 413                 /* Include this chunk */
 414                 g->bytes += len;
 415                 g->num_chunks++;
 416         }
 417         assert(g->num_chunks);
 418 }
 419
 420 /*
 421  * Compute the slice size of the next group.
 422  *
 423  * The FEC parameters n and k are fixed but the slice size varies per
 424  * FEC group.  We'd like to choose slices as small as possible to avoid
 425  * unnecessary FEC calculations but large enough to guarantee that the
 426  * k data slices suffice to encode the header (if needed) and the data
 427  * chunk(s).
 428  *
 429  * Once we know the payload of the next group, we define the number s
 430  * of bytes per slice for this group by
 431  *
 432  *      s = ceil(payload / k)
 433  *
 434  * However, for header streams, computing s is more complicated since no
 435  * overlapping of header and data slices is possible. Hence we have k >=
 436  * 2 and s must satisfy
 437  *
 438  * (*)  ceil(h / s) + ceil(d / s) <= k
 439  *
 440  * where h and d are payload of the header and the data chunk(s)
 441  * respectively. In general there is no value for s such that (*)
 442  * becomes an equality, for example if h = 4000, d = 5000 and k = 10.
 443  *
 444  * We use the following approach for computing a suitable value for s:
 445  *
 446  * Let
 447  *      k1 := ceil(k * min(h, d) / (h + d)),
 448  *      k2 := k - k1.
 449  *
 450  * Note that k >= 2 implies k1 > 0 and k2 > 0, so
 451  *
 452  *      s := max(ceil(min(h, d) / k1), ceil(max(h, d) / k2))
 453  *
 454  * is well-defined. Inequality (*) holds for this value of s since k1
 455  * slices suffice to store min(h, d) while k2 slices suffice to store
 456  * max(h, d), i.e. the first addent of (*) is bounded by k1 and the
 457  * second by k2.
 458  *
 459  * For the above example we obtain
 460  *
 461  *      k1 = ceil(10 * 4000 / 9000) = 5, k2 = 5,
 462  *      s = max(4000 / 5, 5000 / 5) = 1000,
 463  *
 464  * which is optimal since a slice size of 999 bytes would already require
 465  * 11 slices.
 466  */
 467 static int compute_slice_size(struct fec_client *fc, struct vss_task *vsst)
 468 {
 469         struct fec_group *g = &fc->group;
 470         int k = fc->fcp->data_slices_per_group + fc->num_extra_slices;
 471         int n = fc->fcp->slices_per_group + fc->num_extra_slices;
 472         int ret, k1, k2, h, d, min, max, sum;
 473         int max_slice_bytes = fc->mps - FEC_HEADER_SIZE;
 474         int max_group_bytes;
 475
 476         if (!need_audio_header(fc, vsst)) {
 477                 max_group_bytes = k * max_slice_bytes;
 478                 g->num_header_slices = 0;
 479                 compute_group_size(vsst, g, max_group_bytes);
 480                 g->slice_bytes = DIV_ROUND_UP(g->bytes, k);
 481                 if (g->slice_bytes == 0)
 482                         g->slice_bytes = 1;
 483                 return 1;
 484         }
 485         if (!need_data_slices(fc, vsst)) {
 486                 g->bytes = 0;
 487                 g->num_chunks = 0;
 488                 g->slice_bytes = DIV_ROUND_UP(vsst->header_len, k);
 489                 g->num_header_slices = k;
 490                 return 1;
 491         }
 492         h = vsst->header_len;
 493         max_group_bytes = (k - num_slices(h, max_slice_bytes, n - k))
 494                 * max_slice_bytes;
 495         compute_group_size(vsst, g, max_group_bytes);
 496         d = g->bytes;
 497         if (d == 0) {
 498                 g->slice_bytes = DIV_ROUND_UP(h, k);
 499                 ret = num_slices(vsst->header_len, g->slice_bytes, n - k);
 500                 if (ret < 0)
 501                         return ret;
 502                 g->num_header_slices = ret;
 503                 return 1;
 504         }
 505         min = PARA_MIN(h, d);
 506         max = PARA_MAX(h, d);
 507         sum = h + d;
 508         k1 = DIV_ROUND_UP(k * min, sum);
 509         k2 = k - k1;
 510         assert(k1 > 0);
 511         assert(k2 > 0);
 512
 513         g->slice_bytes = PARA_MAX(DIV_ROUND_UP(min, k1), DIV_ROUND_UP(max, k2));
 514         /*
 515          * This value of s := g->slice_bytes satisfies inequality (*) above,
 516          * but it might be larger than max_slice_bytes. However, we know that
 517          * max_slice_bytes are sufficient to store header and data, so:
 518          */
 519         g->slice_bytes = PARA_MIN((int)g->slice_bytes, max_slice_bytes);
 520
 521         ret = num_slices(vsst->header_len, g->slice_bytes, n - k);
 522         if (ret < 0)
 523                 return ret;
 524         g->num_header_slices = ret;
 525         return 1;
 526 }
 527
 528 static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst)
 529 {
 530         int ret, i, k, n, data_slices;
 531         size_t len;
 532         char *buf, *p;
 533         struct fec_group *g = &fc->group;
 534
 535         if (fc->state == FEC_STATE_NONE) {
 536                 ret = initialize_fec_client(fc, vsst);
 537                 if (ret < 0)
 538                         return ret;
 539                 g->first_chunk = mmd->current_chunk;
 540                 g->num = 0;
 541                 g->start = *now;
 542         } else {
 543                 struct timeval tmp;
 544                 if (g->first_chunk + g->num_chunks >= mmd->afd.afhi.chunks_total)
 545                         return 0;
 546                 /*
 547                  * Start and duration of this group depend only on the previous
 548                  * group. Compute the new group start as g->start += g->duration.
 549                  */
 550                 tmp = g->start;
 551                 tv_add(&tmp, &g->duration, &g->start);
 552                 set_group_timing(fc, vsst);
 553                 g->first_chunk += g->num_chunks;
 554                 g->num++;
 555         }
 556         k = fc->fcp->data_slices_per_group + fc->num_extra_slices;
 557         n = fc->fcp->slices_per_group + fc->num_extra_slices;
 558
 559         compute_slice_size(fc, vsst);
 560         assert(g->slice_bytes > 0);
 561         ret = num_slices(g->bytes, g->slice_bytes, n - k);
 562         if (ret < 0)
 563                 return ret;
 564         data_slices = ret;
 565         assert(g->num_header_slices + data_slices <= k);
 566         fc->current_slice_num = 0;
 567         if (g->num == 0)
 568                 set_group_timing(fc, vsst);
 569         /* setup header slices */
 570         buf = vsst->header_buf;
 571         for (i = 0; i < g->num_header_slices; i++) {
 572                 uint32_t payload_size;
 573                 if (buf + g->slice_bytes <= vsst->header_buf + vsst->header_len) {
 574                         fc->src_data[i] = (const unsigned char *)buf;
 575                         buf += g->slice_bytes;
 576                         continue;
 577                 }
 578                 /*
 579                  * Can not use vss->header_buf for this slice as it
 580                  * goes beyond the buffer. This slice will not be fully
 581                  * used.
 582                  */
 583                 payload_size = vsst->header_buf + vsst->header_len - buf;
 584                 memcpy(fc->extra_header_buf, buf, payload_size);
 585                 if (payload_size < g->slice_bytes)
 586                         memset(fc->extra_header_buf + payload_size, 0,
 587                                 g->slice_bytes - payload_size);
 588                 /*
 589                  * There might be more than one header slice to fill although
 590                  * only the first one will be used. Set all header slices to
 591                  * our extra buffer.
 592                  */
 593                 while (i < g->num_header_slices)
 594                         fc->src_data[i++] = fc->extra_header_buf;
 595                 break; /* we don't want i to be increased. */
 596         }
 597
 598         /*
 599          * Setup data slices. Note that for ogg streams chunk 0 points to a
 600          * buffer on the heap rather than to the mapped audio file.
 601          */
 602         vss_get_chunk(g->first_chunk, vsst, &buf, &len);
 603         for (p = buf; i < g->num_header_slices + data_slices; i++) {
 604                 if (p + g->slice_bytes > buf + g->bytes) {
 605                         /*
 606                          * We must make a copy for this slice since using p
 607                          * directly would exceed the buffer.
 608                          */
 609                         uint32_t payload_size = buf + g->bytes - p;
 610                         assert(payload_size + FEC_HEADER_SIZE <= fc->mps);
 611                         memcpy(fc->extra_src_buf, p, payload_size);
 612                         if (payload_size < g->slice_bytes)
 613                                 memset(fc->extra_src_buf + payload_size, 0,
 614                                         g->slice_bytes - payload_size);
 615                         fc->src_data[i] = fc->extra_src_buf;
 616                         i++;
 617                         break;
 618                 }
 619                 fc->src_data[i] = (const unsigned char *)p;
 620                 p += g->slice_bytes;
 621         }
 622         if (i < k) {
 623                 /* use arbitrary data for all remaining slices */
 624                 buf = vsst->map;
 625                 for (; i < k; i++)
 626                         fc->src_data[i] = (const unsigned char *)buf;
 627         }
 628         PARA_DEBUG_LOG("FEC group %u: %u chunks (%u - %u), %u bytes\n",
 629                 g->num, g->num_chunks, g->first_chunk,
 630                 g->first_chunk + g->num_chunks - 1, g->bytes
 631         );
 632         PARA_DEBUG_LOG("slice_bytes: %d, %d header slices, %d data slices\n",
 633                 g->slice_bytes, g->num_header_slices, data_slices
 634         );
 635         return 1;
 636 }
 637
 638 static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst)
 639 {
 640         if (fc->state == FEC_STATE_NONE || fc->current_slice_num
 641                         == fc->fcp->slices_per_group + fc->num_extra_slices) {
 642                 int ret = setup_next_fec_group(fc, vsst);
 643                 if (ret == 0)
 644                         return 0;
 645                 if (ret < 0) {
 646                         PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 647                         PARA_ERROR_LOG("FEC client temporarily disabled\n");
 648                         fc->state = FEC_STATE_DISABLED;
 649                         return ret;
 650                 }
 651         }
 652         write_fec_header(fc, vsst);
 653         fec_encode(fc->parms, fc->src_data, fc->enc_buf + FEC_HEADER_SIZE,
 654                 fc->current_slice_num, fc->group.slice_bytes);
 655         return 1;
 656 }
 657
 658 /**
 659  * Return a buffer that marks the end of the stream.
 660  *
 661  * \param buf Result pointer.
 662  * \return The length of the eof buffer.
 663  *
 664  * This is used for (multicast) udp streaming where closing the socket on the
 665  * sender might not give rise to an eof condition at the peer.
 666  */
 667 size_t vss_get_fec_eof_packet(const char **buf)
 668 {
 669         static const char fec_eof_packet[FEC_HEADER_SIZE] = FEC_EOF_PACKET;
 670         *buf = fec_eof_packet;
 671         return FEC_HEADER_SIZE;
 672 }
 673
 674 /**
 675  * Add one entry to the list of active fec clients.
 676  *
 677  * \param sc  Generic sender_client data of the transport layer.
 678  * \param fcp FEC parameters as supplied by the transport layer.
 679  *
 680  * \return Newly allocated fec_client struct.
 681  */
 682 struct fec_client *vss_add_fec_client(struct sender_client *sc,
 683                                       struct fec_client_parms *fcp)
 684 {
 685         struct fec_client *fc = para_calloc(sizeof(*fc));
 686
 687         fc->sc  = sc;
 688         fc->fcp = fcp;
 689         para_list_add(&fc->node, &fec_client_list);
 690         return fc;
 691 }
 692
 693 /**
 694  * Remove one entry from the list of active fec clients.
 695  *
 696  * \param fc The client to be removed.
 697  */
 698 void vss_del_fec_client(struct fec_client *fc)
 699 {
 700         list_del(&fc->node);
 701         free(fc->src_data);
 702         free(fc->enc_buf);
 703         free(fc->extra_src_buf);
 704         free(fc->extra_header_buf);
 705         fec_free(fc->parms);
 706         free(fc);
 707 }
 708
 709 /*
 710  * Compute if/when next slice is due. If it isn't due yet and \a diff is
 711  * not \p Null, compute the time difference next - now, where
 712  *
 713  *      next = stream_start + (first_group_chunk - first_stream_chunk)
 714  *              * chunk_time + slice_num * slice_time
 715  */
 716 static int next_slice_is_due(struct fec_client *fc, struct timeval *diff)
 717 {
 718         struct timeval tmp, next;
 719         int ret;
 720
 721         if (fc->state == FEC_STATE_NONE)
 722                 return 1;
 723         tv_scale(fc->current_slice_num, &fc->group.slice_duration, &tmp);
 724         tv_add(&tmp, &fc->group.start, &next);
 725         ret = tv_diff(&next, now, diff);
 726         return ret < 0? 1 : 0;
 727 }
 728
 729 static void set_eof_barrier(struct vss_task *vsst)
 730 {
 731         struct fec_client *fc;
 732         struct timeval timeout = {1, 0}, *chunk_tv = vss_chunk_time();
 733
 734         if (!chunk_tv)
 735                 goto out;
 736         list_for_each_entry(fc, &fec_client_list, node) {
 737                 struct timeval group_duration;
 738
 739                 if (fc->state != FEC_STATE_READY_TO_RUN)
 740                         continue;
 741                 tv_scale(fc->group.num_chunks, chunk_tv, &group_duration);
 742                 if (tv_diff(&timeout, &group_duration, NULL) < 0)
 743                         timeout = group_duration;
 744         }
 745 out:
 746         tv_add(now, &timeout, &vsst->eof_barrier);
 747 }
 748
 749 /**
 750  * Check if vss status flag \a P (playing) is set.
 751  *
 752  * \return Greater than zero if playing, zero otherwise.
 753  *
 754  */
 755 unsigned int vss_playing(void)
 756 {
 757         return mmd->new_vss_status_flags & VSS_PLAYING;
 758 }
 759
 760 /**
 761  * Check if the \a N (next) status flag is set.
 762  *
 763  * \return Greater than zero if set, zero if not.
 764  *
 765  */
 766 unsigned int vss_next(void)
 767 {
 768         return mmd->new_vss_status_flags & VSS_NEXT;
 769 }
 770
 771 /**
 772  * Check if a reposition request is pending.
 773  *
 774  * \return Greater than zero if true, zero otherwise.
 775  *
 776  */
 777 unsigned int vss_repos(void)
 778 {
 779         return mmd->new_vss_status_flags & VSS_REPOS;
 780 }
 781
 782 /**
 783  * Check if the vss is currently paused.
 784  *
 785  * \return Greater than zero if paused, zero otherwise.
 786  *
 787  */
 788 unsigned int vss_paused(void)
 789 {
 790         return !(mmd->new_vss_status_flags & VSS_NEXT)
 791                 && !(mmd->new_vss_status_flags & VSS_PLAYING);
 792 }
 793
 794 /**
 795  * Check if the vss is currently stopped.
 796  *
 797  * \return Greater than zero if paused, zero otherwise.
 798  *
 799  */
 800 unsigned int vss_stopped(void)
 801 {
 802         return (mmd->new_vss_status_flags & VSS_NEXT)
 803                 && !(mmd->new_vss_status_flags & VSS_PLAYING);
 804 }
 805
 806 static int chk_barrier(const char *bname, const struct timeval *barrier,
 807                 struct timeval *diff, int print_log)
 808 {
 809         long ms;
 810
 811         if (tv_diff(now, barrier, diff) > 0)
 812                 return 1;
 813         ms = tv2ms(diff);
 814         if (print_log && ms)
 815                 PARA_DEBUG_LOG("%s barrier: %lims left\n", bname, ms);
 816         return -1;
 817 }
 818
 819 static void vss_compute_timeout(struct sched *s, struct vss_task *vsst)
 820 {
 821         struct timeval tv;
 822         struct fec_client *fc;
 823
 824         if (!vss_playing() || !vsst->map)
 825                 return;
 826         if (vss_next() && vsst->map) /* only sleep a bit, nec*/
 827                 return sched_request_timeout_ms(100, s);
 828
 829         /* Each of these barriers must have passed until we may proceed */
 830         if (sched_request_barrier(&vsst->autoplay_barrier, s) == 1)
 831                 return;
 832         if (sched_request_barrier(&vsst->eof_barrier, s) == 1)
 833                 return;
 834         if (sched_request_barrier(&vsst->data_send_barrier, s) == 1)
 835                 return;
 836         /*
 837          * Compute the select timeout as the minimal time until the next
 838          * chunk/slice is due for any client.
 839          */
 840         compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv,
 841                 &mmd->stream_start, &tv);
 842         if (sched_request_barrier_or_min_delay(&tv, s) == 0)
 843                 return;
 844         list_for_each_entry(fc, &fec_client_list, node) {
 845                 if (fc->state != FEC_STATE_READY_TO_RUN)
 846                         continue;
 847                 if (next_slice_is_due(fc, &tv))
 848                         return sched_min_delay(s);
 849                 sched_request_timeout(&tv, s);
 850         }
 851 }
 852
 853 static void vss_eof(struct vss_task *vsst)
 854 {
 855
 856         if (!vsst->map)
 857                 return;
 858         if (mmd->new_vss_status_flags & VSS_NOMORE)
 859                 mmd->new_vss_status_flags = VSS_NEXT;
 860         set_eof_barrier(vsst);
 861         afh_free_header(vsst->header_buf, mmd->afd.audio_format_id);
 862         vsst->header_buf = NULL;
 863         para_munmap(vsst->map, vsst->mapsize);
 864         vsst->map = NULL;
 865         mmd->chunks_sent = 0;
 866         //mmd->offset = 0;
 867         mmd->afd.afhi.seconds_total = 0;
 868         mmd->afd.afhi.chunk_tv.tv_sec = 0;
 869         mmd->afd.afhi.chunk_tv.tv_usec = 0;
 870         free(mmd->afd.afhi.chunk_table);
 871         mmd->afd.afhi.chunk_table = NULL;
 872         vsst->mapsize = 0;
 873         afh_close(vsst->afh_context, mmd->afd.audio_format_id);
 874         vsst->afh_context = NULL;
 875         mmd->events++;
 876 }
 877
 878 static int need_to_request_new_audio_file(struct vss_task *vsst)
 879 {
 880         struct timeval diff;
 881
 882         if (vsst->map) /* have audio file */
 883                 return 0;
 884         if (!vss_playing()) /* don't need one */
 885                 return 0;
 886         if (mmd->new_vss_status_flags & VSS_NOMORE)
 887                 return 0;
 888         if (vsst->afsss == AFS_SOCKET_AFD_PENDING) /* already requested one */
 889                 return 0;
 890         if (chk_barrier("autoplay_delay", &vsst->autoplay_barrier,
 891                         &diff, 1) < 0)
 892                 return 0;
 893         return 1;
 894 }
 895
 896 static void set_mmd_offset(void)
 897 {
 898         struct timeval offset;
 899         tv_scale(mmd->current_chunk, &mmd->afd.afhi.chunk_tv, &offset);
 900         mmd->offset = tv2ms(&offset);
 901 }
 902
 903 static void vss_pre_select(struct sched *s, void *context)
 904 {
 905         int i;
 906         struct vss_task *vsst = context;
 907
 908         if (need_to_request_new_audio_file(vsst)) {
 909                 PARA_DEBUG_LOG("ready and playing, but no audio file\n");
 910                 para_fd_set(vsst->afs_socket, &s->wfds, &s->max_fileno);
 911                 vsst->afsss = AFS_SOCKET_CHECK_FOR_WRITE;
 912         } else
 913                 para_fd_set(vsst->afs_socket, &s->rfds, &s->max_fileno);
 914         for (i = 0; senders[i].name; i++) {
 915                 if (!senders[i].pre_select)
 916                         continue;
 917                 senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds);
 918         }
 919         vss_compute_timeout(s, vsst);
 920 }
 921
 922 static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data)
 923 {
 924         char control[255] __a_aligned(8), buf[8];
 925         struct msghdr msg = {.msg_iov = NULL};
 926         struct cmsghdr *cmsg;
 927         struct iovec iov;
 928         int ret = 0;
 929
 930         *fd = -1;
 931         iov.iov_base = buf;
 932         iov.iov_len = sizeof(buf);
 933         msg.msg_iov = &iov;
 934         msg.msg_iovlen = 1;
 935         msg.msg_control = control;
 936         msg.msg_controllen = sizeof(control);
 937         memset(buf, 0, sizeof(buf));
 938         ret = recvmsg(afs_socket, &msg, 0);
 939         if (ret < 0)
 940                 return -ERRNO_TO_PARA_ERROR(errno);
 941         if (iov.iov_len != sizeof(buf))
 942                 return -E_AFS_SHORT_READ;
 943         *code = *(uint32_t*)buf;
 944         *data =  *(uint32_t*)(buf + 4);
 945         for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
 946                 if (cmsg->cmsg_level != SOL_SOCKET
 947                         || cmsg->cmsg_type != SCM_RIGHTS)
 948                         continue;
 949                 if ((cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int) != 1)
 950                         continue;
 951                 *fd = *(int *)CMSG_DATA(cmsg);
 952         }
 953         return 1;
 954 }
 955
 956 #ifndef MAP_POPULATE
 957 #define MAP_POPULATE 0
 958 #endif
 959
 960 static void recv_afs_result(struct vss_task *vsst, fd_set *rfds)
 961 {
 962         int ret, passed_fd, shmid;
 963         uint32_t afs_code = 0, afs_data = 0;
 964         struct stat statbuf;
 965
 966         if (!FD_ISSET(vsst->afs_socket, rfds))
 967                 return;
 968         ret = recv_afs_msg(vsst->afs_socket, &passed_fd, &afs_code, &afs_data);
 969         if (ret == -ERRNO_TO_PARA_ERROR(EAGAIN))
 970                 return;
 971         if (ret < 0)
 972                 goto err;
 973         vsst->afsss = AFS_SOCKET_READY;
 974         PARA_DEBUG_LOG("fd: %d, code: %u, shmid: %u\n", passed_fd, afs_code,
 975                 afs_data);
 976         ret = -E_NOFD;
 977         if (afs_code != NEXT_AUDIO_FILE)
 978                 goto err;
 979         if (passed_fd < 0)
 980                 goto err;
 981         shmid = afs_data;
 982         ret = load_afd(shmid, &mmd->afd);
 983         if (ret < 0)
 984                 goto err;
 985         shm_destroy(shmid);
 986         ret = fstat(passed_fd, &statbuf);
 987         if (ret < 0) {
 988                 PARA_ERROR_LOG("fstat error:\n");
 989                 ret = -ERRNO_TO_PARA_ERROR(errno);
 990                 goto err;
 991         }
 992         ret = para_mmap(statbuf.st_size, PROT_READ, MAP_PRIVATE | MAP_POPULATE,
 993                 passed_fd, 0, &vsst->map);
 994         if (ret < 0)
 995                 goto err;
 996         vsst->mapsize = statbuf.st_size;
 997         close(passed_fd);
 998         mmd->chunks_sent = 0;
 999         mmd->current_chunk = 0;
1000         mmd->offset = 0;
1001         mmd->events++;
1002         mmd->num_played++;
1003         mmd->new_vss_status_flags &= (~VSS_NEXT);
1004         afh_get_header(&mmd->afd.afhi, mmd->afd.audio_format_id,
1005                 vsst->map, vsst->mapsize, &vsst->header_buf, &vsst->header_len);
1006         return;
1007 err:
1008         free(mmd->afd.afhi.chunk_table);
1009         if (passed_fd >= 0)
1010                 close(passed_fd);
1011         PARA_ERROR_LOG("%s\n", para_strerror(-ret));
1012         mmd->new_vss_status_flags = VSS_NEXT;
1013 }
1014
1015 /**
1016  * Main sending function.
1017  *
1018  * This function gets called from vss_post_select(). It checks whether the next
1019  * chunk of data should be pushed out. It obtains a pointer to the data to be
1020  * sent out as well as its length from mmd->afd.afhi. This information is then
1021  * passed to each supported sender's send() function as well as to the send()
1022  * functions of each registered fec client.
1023  */
1024 static void vss_send(struct vss_task *vsst)
1025 {
1026         int i, fec_active = 0;
1027         struct timeval due;
1028         struct fec_client *fc, *tmp_fc;
1029
1030         if (!vsst->map || !vss_playing())
1031                 return;
1032         if (chk_barrier("eof", &vsst->eof_barrier, &due, 1) < 0)
1033                 return;
1034         if (chk_barrier("data send", &vsst->data_send_barrier,
1035                         &due, 1) < 0)
1036                 return;
1037         list_for_each_entry_safe(fc, tmp_fc, &fec_client_list, node) {
1038                 if (fc->state == FEC_STATE_DISABLED)
1039                         continue;
1040                 if (!next_slice_is_due(fc, NULL)) {
1041                         fec_active = 1;
1042                         continue;
1043                 }
1044                 if (compute_next_fec_slice(fc, vsst) <= 0)
1045                         continue;
1046                 PARA_DEBUG_LOG("sending %u:%u (%u bytes)\n", fc->group.num,
1047                         fc->current_slice_num, fc->group.slice_bytes);
1048                 fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf,
1049                         fc->group.slice_bytes + FEC_HEADER_SIZE);
1050                 fc->current_slice_num++;
1051                 fec_active = 1;
1052         }
1053         if (mmd->current_chunk >= mmd->afd.afhi.chunks_total) { /* eof */
1054                 if (!fec_active)
1055                         mmd->new_vss_status_flags |= VSS_NEXT;
1056                 return;
1057         }
1058         compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv,
1059                 &mmd->stream_start, &due);
1060         if (tv_diff(&due, now, NULL) <= 0) {
1061                 char *buf;
1062                 size_t len;
1063
1064                 if (!mmd->chunks_sent) {
1065                         mmd->stream_start = *now;
1066                         mmd->events++;
1067                         set_mmd_offset();
1068                 }
1069                 /*
1070                  * We call the send function also in case of empty chunks as
1071                  * they might have still some data queued which can be sent in
1072                  * this case.
1073                  */
1074                 vss_get_chunk(mmd->current_chunk, vsst, &buf, &len);
1075                 for (i = 0; senders[i].name; i++) {
1076                         if (!senders[i].send)
1077                                 continue;
1078                         senders[i].send(mmd->current_chunk, mmd->chunks_sent,
1079                                 buf, len, vsst->header_buf, vsst->header_len);
1080                 }
1081                 /*
1082                  * Prefault next chunk(s)
1083                  *
1084                  * If the backing device of the memory-mapped audio file is
1085                  * slow and read-ahead is turned off or prevented for some
1086                  * reason, e.g. due to memory pressure, it may take much longer
1087                  * than the chunk interval to get the next chunk on the wire,
1088                  * causing buffer underruns on the client side. Mapping the
1089                  * file with MAP_POPULATE seems to help a bit, but it does not
1090                  * eliminate the delays completely. Moreover, it is supported
1091                  * only on Linux. So we do our own read-ahead here.
1092                  */
1093                 if (mmd->current_chunk > 0) { /* chunk 0 might be on the heap */
1094                         buf += len;
1095                         for (i = 0; i < 5 && buf < vsst->map + vsst->mapsize; i++) {
1096                                 __a_unused volatile char x = *buf;
1097                                 buf += 4096;
1098                         }
1099                 }
1100                 mmd->chunks_sent++;
1101                 mmd->current_chunk++;
1102         }
1103 }
1104
1105 static int vss_post_select(struct sched *s, void *context)
1106 {
1107         int ret, i;
1108         struct vss_task *vsst = context;
1109
1110         if (!vsst->map || vss_next() || vss_paused() || vss_repos()) {
1111                 /* shut down senders and fec clients */
1112                 struct fec_client *fc, *tmp;
1113                 for (i = 0; senders[i].name; i++)
1114                         if (senders[i].shutdown_clients)
1115                                 senders[i].shutdown_clients();
1116                 list_for_each_entry_safe(fc, tmp, &fec_client_list, node)
1117                         fc->state = FEC_STATE_NONE;
1118                 mmd->stream_start.tv_sec = 0;
1119                 mmd->stream_start.tv_usec = 0;
1120         }
1121         if (vss_next())
1122                 vss_eof(vsst);
1123         else if (vss_paused()) {
1124                 if (mmd->chunks_sent)
1125                         set_eof_barrier(vsst);
1126                 mmd->chunks_sent = 0;
1127         } else if (vss_repos()) { /* repositioning due to ff/jmp command */
1128                 tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier);
1129                 set_eof_barrier(vsst);
1130                 mmd->chunks_sent = 0;
1131                 mmd->current_chunk = afh_get_start_chunk(mmd->repos_request,
1132                         &mmd->afd.afhi, mmd->afd.audio_format_id);
1133                 mmd->new_vss_status_flags &= ~VSS_REPOS;
1134                 set_mmd_offset();
1135         }
1136         /* If a sender command is pending, run it. */
1137         if (mmd->sender_cmd_data.cmd_num >= 0) {
1138                 int num = mmd->sender_cmd_data.cmd_num,
1139                         sender_num = mmd->sender_cmd_data.sender_num;
1140
1141                 if (senders[sender_num].client_cmds[num]) {
1142                         ret = senders[sender_num].client_cmds[num]
1143                                 (&mmd->sender_cmd_data);
1144                         if (ret < 0)
1145                                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
1146                 }
1147                 mmd->sender_cmd_data.cmd_num = -1;
1148         }
1149         if (vsst->afsss != AFS_SOCKET_CHECK_FOR_WRITE)
1150                 recv_afs_result(vsst, &s->rfds);
1151         else if (FD_ISSET(vsst->afs_socket, &s->wfds)) {
1152                 PARA_NOTICE_LOG("requesting new fd from afs\n");
1153                 ret = write_buffer(vsst->afs_socket, "new");
1154                 if (ret < 0)
1155                         PARA_CRIT_LOG("%s\n", para_strerror(-ret));
1156                 else
1157                         vsst->afsss = AFS_SOCKET_AFD_PENDING;
1158         }
1159         for (i = 0; senders[i].name; i++) {
1160                 if (!senders[i].post_select)
1161                         continue;
1162                 senders[i].post_select(&s->rfds, &s->wfds);
1163         }
1164         if ((vss_playing() && !(mmd->vss_status_flags & VSS_PLAYING)) ||
1165                         (vss_next() && vss_playing()))
1166                 tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier);
1167         vss_send(vsst);
1168         return 0;
1169 }
1170
1171 /**
1172  * Initialize the virtual streaming system task.
1173  *
1174  * \param afs_socket The fd for communication with afs.
1175  * \param s The scheduler to register the vss task to.
1176  *
1177  * This also initializes all supported senders and starts streaming
1178  * if the --autoplay command line flag was given.
1179  */
1180 void init_vss_task(int afs_socket, struct sched *s)
1181 {
1182         static struct vss_task vss_task_struct, *vsst = &vss_task_struct;
1183         int i;
1184         char *hn = para_hostname(), *home = para_homedir();
1185         long unsigned announce_time = conf.announce_time_arg > 0?
1186                         conf.announce_time_arg : 300,
1187                 autoplay_delay = conf.autoplay_delay_arg > 0?
1188                         conf.autoplay_delay_arg : 0;
1189         vsst->header_interval.tv_sec = 5; /* should this be configurable? */
1190         vsst->afs_socket = afs_socket;
1191         ms2tv(announce_time, &vsst->announce_tv);
1192         PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst->announce_tv));
1193         INIT_LIST_HEAD(&fec_client_list);
1194         for (i = 0; senders[i].name; i++) {
1195                 PARA_NOTICE_LOG("initializing %s sender\n", senders[i].name);
1196                 senders[i].init(&senders[i]);
1197         }
1198         free(hn);
1199         free(home);
1200         mmd->sender_cmd_data.cmd_num = -1;
1201         if (conf.autoplay_given) {
1202                 struct timeval tmp;
1203                 mmd->vss_status_flags |= VSS_PLAYING;
1204                 mmd->new_vss_status_flags |= VSS_PLAYING;
1205                 ms2tv(autoplay_delay, &tmp);
1206                 tv_add(clock_get_realtime(NULL), &tmp, &vsst->autoplay_barrier);
1207                 tv_add(&vsst->autoplay_barrier, &vsst->announce_tv,
1208                         &vsst->data_send_barrier);
1209         }
1210         vsst->task = task_register(&(struct task_info) {
1211                 .name = "vss task",
1212                 .pre_select = vss_pre_select,
1213                 .post_select = vss_post_select,
1214                 .context = vsst,
1215         }, s);
1216 }