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