vss.c

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