vss.c

   1 /*
   2  * Copyright (C) 1997-2012 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                 fc->src_data[i] = fc->extra_header_buf;
 569                 assert(i == g->num_header_slices - 1);
 570         }
 571
 572         /*
 573          * Setup data slices. Note that for ogg streams chunk 0 points to a
 574          * buffer on the heap rather than to the mapped audio file.
 575          */
 576         vss_get_chunk(g->first_chunk, vsst, &buf, &len);
 577         for (p = buf; i < g->num_header_slices + data_slices; i++) {
 578                 if (p + g->slice_bytes > buf + g->bytes) {
 579                         /*
 580                          * We must make a copy for this slice since using p
 581                          * directly would exceed the buffer.
 582                          */
 583                         uint32_t payload_size = buf + g->bytes - p;
 584                         assert(payload_size + FEC_HEADER_SIZE <= fc->mps);
 585                         memcpy(fc->extra_src_buf, p, payload_size);
 586                         if (payload_size < g->slice_bytes)
 587                                 memset(fc->extra_src_buf + payload_size, 0,
 588                                         g->slice_bytes - payload_size);
 589                         fc->src_data[i] = fc->extra_src_buf;
 590                         i++;
 591                         break;
 592                 }
 593                 fc->src_data[i] = (const unsigned char *)p;
 594                 p += g->slice_bytes;
 595         }
 596         if (i < k) {
 597                 /* use arbitrary data for all remaining slices */
 598                 buf = vsst->map;
 599                 for (; i < k; i++)
 600                         fc->src_data[i] = (const unsigned char *)buf;
 601         }
 602         PARA_DEBUG_LOG("FEC group %d: %d chunks (%d - %d), %d bytes\n",
 603                 g->num, g->num_chunks, g->first_chunk,
 604                 g->first_chunk + g->num_chunks - 1, g->bytes
 605         );
 606         PARA_DEBUG_LOG("slice_bytes: %d, %d header slices, %d data slices\n",
 607                 g->slice_bytes, g->num_header_slices, data_slices
 608         );
 609         return 1;
 610 }
 611
 612 static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst)
 613 {
 614         if (fc->state == FEC_STATE_NONE || fc->current_slice_num
 615                         == fc->fcp->slices_per_group + fc->num_extra_slices) {
 616                 int ret = setup_next_fec_group(fc, vsst);
 617                 if (ret == 0)
 618                         return 0;
 619                 if (ret < 0) {
 620                         PARA_ERROR_LOG("%s\n", para_strerror(-ret));
 621                         PARA_ERROR_LOG("FEC client temporarily disabled\n");
 622                         fc->state = FEC_STATE_DISABLED;
 623                         return ret;
 624                 }
 625         }
 626         write_fec_header(fc, vsst);
 627         fec_encode(fc->parms, fc->src_data, fc->enc_buf + FEC_HEADER_SIZE,
 628                 fc->current_slice_num, fc->group.slice_bytes);
 629         return 1;
 630 }
 631
 632 /**
 633  * Return a buffer that marks the end of the stream.
 634  *
 635  * \param buf Result pointer.
 636  * \return The length of the eof buffer.
 637  *
 638  * This is used for (multicast) udp streaming where closing the socket on the
 639  * sender might not give rise to an eof condition at the peer.
 640  */
 641 size_t vss_get_fec_eof_packet(const char **buf)
 642 {
 643         static const char fec_eof_packet[FEC_HEADER_SIZE] = FEC_EOF_PACKET;
 644         *buf = fec_eof_packet;
 645         return FEC_HEADER_SIZE;
 646 }
 647
 648 /**
 649  * Add one entry to the list of active fec clients.
 650  *
 651  * \param sc  Generic sender_client data of the transport layer.
 652  * \param fcp FEC parameters as supplied by the transport layer.
 653  *
 654  * \return Newly allocated fec_client struct.
 655  */
 656 struct fec_client *vss_add_fec_client(struct sender_client *sc,
 657                                       struct fec_client_parms *fcp)
 658 {
 659         struct fec_client *fc = para_calloc(sizeof(*fc));
 660
 661         fc->sc  = sc;
 662         fc->fcp = fcp;
 663         para_list_add(&fc->node, &fec_client_list);
 664         return fc;
 665 }
 666
 667 /**
 668  * Remove one entry from the list of active fec clients.
 669  *
 670  * \param fc The client to be removed.
 671  */
 672 void vss_del_fec_client(struct fec_client *fc)
 673 {
 674         list_del(&fc->node);
 675         free(fc->src_data);
 676         free(fc->enc_buf);
 677         free(fc->extra_src_buf);
 678         free(fc->extra_header_buf);
 679         fec_free(fc->parms);
 680         free(fc);
 681 }
 682
 683 /*
 684  * Compute if/when next slice is due. If it isn't due yet and \a diff is
 685  * not \p Null, compute the time difference next - now, where
 686  *
 687  *      next = stream_start + (first_group_chunk - first_stream_chunk)
 688  *              * chunk_time + slice_num * slice_time
 689  */
 690 static int next_slice_is_due(struct fec_client *fc, struct timeval *diff)
 691 {
 692         struct timeval tmp, next;
 693         int ret;
 694
 695         if (fc->state == FEC_STATE_NONE)
 696                 return 1;
 697         tv_scale(fc->current_slice_num, &fc->group.slice_duration, &tmp);
 698         tv_add(&tmp, &fc->group.start, &next);
 699         ret = tv_diff(&next, now, diff);
 700         return ret < 0? 1 : 0;
 701 }
 702
 703 static void set_eof_barrier(struct vss_task *vsst)
 704 {
 705         struct fec_client *fc;
 706         struct timeval timeout = {1, 0}, *chunk_tv = vss_chunk_time();
 707
 708         if (!chunk_tv)
 709                 goto out;
 710         list_for_each_entry(fc, &fec_client_list, node) {
 711                 struct timeval group_duration;
 712
 713                 if (fc->state != FEC_STATE_READY_TO_RUN)
 714                         continue;
 715                 tv_scale(fc->group.num_chunks, chunk_tv, &group_duration);
 716                 if (tv_diff(&timeout, &group_duration, NULL) < 0)
 717                         timeout = group_duration;
 718         }
 719 out:
 720         tv_add(now, &timeout, &vsst->eof_barrier);
 721 }
 722
 723 /**
 724  * Check if vss status flag \a P (playing) is set.
 725  *
 726  * \return Greater than zero if playing, zero otherwise.
 727  *
 728  */
 729 unsigned int vss_playing(void)
 730 {
 731         return mmd->new_vss_status_flags & VSS_PLAYING;
 732 }
 733
 734 /**
 735  * Check if the \a N (next) status flag is set.
 736  *
 737  * \return Greater than zero if set, zero if not.
 738  *
 739  */
 740 unsigned int vss_next(void)
 741 {
 742         return mmd->new_vss_status_flags & VSS_NEXT;
 743 }
 744
 745 /**
 746  * Check if a reposition request is pending.
 747  *
 748  * \return Greater than zero if true, zero otherwise.
 749  *
 750  */
 751 unsigned int vss_repos(void)
 752 {
 753         return mmd->new_vss_status_flags & VSS_REPOS;
 754 }
 755
 756 /**
 757  * Check if the vss is currently paused.
 758  *
 759  * \return Greater than zero if paused, zero otherwise.
 760  *
 761  */
 762 unsigned int vss_paused(void)
 763 {
 764         return !(mmd->new_vss_status_flags & VSS_NEXT)
 765                 && !(mmd->new_vss_status_flags & VSS_PLAYING);
 766 }
 767
 768 /**
 769  * Check if the vss is currently stopped.
 770  *
 771  * \return Greater than zero if paused, zero otherwise.
 772  *
 773  */
 774 unsigned int vss_stopped(void)
 775 {
 776         return (mmd->new_vss_status_flags & VSS_NEXT)
 777                 && !(mmd->new_vss_status_flags & VSS_PLAYING);
 778 }
 779
 780 static int chk_barrier(const char *bname, const struct timeval *barrier,
 781                 struct timeval *diff, int print_log)
 782 {
 783         long ms;
 784
 785         if (tv_diff(now, barrier, diff) > 0)
 786                 return 1;
 787         ms = tv2ms(diff);
 788         if (print_log && ms)
 789                 PARA_DEBUG_LOG("%s barrier: %lims left\n", bname, ms);
 790         return -1;
 791 }
 792
 793 static void vss_compute_timeout(struct sched *s, struct vss_task *vsst)
 794 {
 795         struct timeval tv;
 796         struct fec_client *fc;
 797
 798         if (!vss_playing() || !vsst->map)
 799                 return;
 800         if (vss_next() && vsst->map) /* only sleep a bit, nec*/
 801                 return sched_request_timeout_ms(100, s);
 802
 803         /* Each of these barriers must have passed until we may proceed */
 804         if (sched_request_barrier(&vsst->autoplay_barrier, s) == 1)
 805                 return;
 806         if (sched_request_barrier(&vsst->eof_barrier, s) == 1)
 807                 return;
 808         if (sched_request_barrier(&vsst->data_send_barrier, s) == 1)
 809                 return;
 810         /*
 811          * Compute the select timeout as the minimal time until the next
 812          * chunk/slice is due for any client.
 813          */
 814         compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv,
 815                 &mmd->stream_start, &tv);
 816         if (sched_request_barrier_or_min_delay(&tv, s) == 0)
 817                 return;
 818         list_for_each_entry(fc, &fec_client_list, node) {
 819                 if (fc->state != FEC_STATE_READY_TO_RUN)
 820                         continue;
 821                 if (next_slice_is_due(fc, &tv))
 822                         return sched_min_delay(s);
 823                 sched_request_timeout(&tv, s);
 824         }
 825 }
 826
 827 static void vss_eof(struct vss_task *vsst)
 828 {
 829
 830         if (!vsst->map)
 831                 return;
 832         if (mmd->new_vss_status_flags & VSS_NOMORE)
 833                 mmd->new_vss_status_flags = VSS_NEXT;
 834         set_eof_barrier(vsst);
 835         afh_free_header(vsst->header_buf, mmd->afd.audio_format_id);
 836         vsst->header_buf = NULL;
 837         para_munmap(vsst->map, mmd->size);
 838         vsst->map = NULL;
 839         mmd->chunks_sent = 0;
 840         //mmd->offset = 0;
 841         mmd->afd.afhi.seconds_total = 0;
 842         mmd->afd.afhi.chunk_tv.tv_sec = 0;
 843         mmd->afd.afhi.chunk_tv.tv_usec = 0;
 844         free(mmd->afd.afhi.chunk_table);
 845         mmd->afd.afhi.chunk_table = NULL;
 846         mmd->mtime = 0;
 847         mmd->size = 0;
 848         mmd->events++;
 849 }
 850
 851 static int need_to_request_new_audio_file(struct vss_task *vsst)
 852 {
 853         struct timeval diff;
 854
 855         if (vsst->map) /* have audio file */
 856                 return 0;
 857         if (!vss_playing()) /* don't need one */
 858                 return 0;
 859         if (mmd->new_vss_status_flags & VSS_NOMORE)
 860                 return 0;
 861         if (vsst->afsss == AFS_SOCKET_AFD_PENDING) /* already requested one */
 862                 return 0;
 863         if (chk_barrier("autoplay_delay", &vsst->autoplay_barrier,
 864                         &diff, 1) < 0)
 865                 return 0;
 866         return 1;
 867 }
 868
 869 static void set_mmd_offset(void)
 870 {
 871         struct timeval offset;
 872         tv_scale(mmd->current_chunk, &mmd->afd.afhi.chunk_tv, &offset);
 873         mmd->offset = tv2ms(&offset);
 874 }
 875
 876 /**
 877  * Compute the timeout for the main select-loop of the scheduler.
 878  *
 879  * \param s Pointer to the server scheduler.
 880  * \param t Pointer to the vss task structure.
 881  *
 882  * Before the timeout is computed, the current vss status flags are evaluated
 883  * and acted upon by calling appropriate functions from the lower layers.
 884  * Possible actions include
 885  *
 886  *      - request a new audio file from afs,
 887  *      - shutdown of all senders (stop/pause command),
 888  *      - reposition the stream (ff/jmp command).
 889  */
 890 static void vss_pre_select(struct sched *s, struct task *t)
 891 {
 892         int i;
 893         struct vss_task *vsst = container_of(t, struct vss_task, task);
 894
 895         if (!vsst->map || vss_next() || vss_paused() || vss_repos()) {
 896                 struct fec_client *fc, *tmp;
 897                 for (i = 0; senders[i].name; i++)
 898                         if (senders[i].shutdown_clients)
 899                                 senders[i].shutdown_clients();
 900                 list_for_each_entry_safe(fc, tmp, &fec_client_list, node)
 901                         fc->state = FEC_STATE_NONE;
 902                 mmd->stream_start.tv_sec = 0;
 903                 mmd->stream_start.tv_usec = 0;
 904         }
 905         if (vss_next())
 906                 vss_eof(vsst);
 907         else if (vss_paused()) {
 908                 if (mmd->chunks_sent)
 909                         set_eof_barrier(vsst);
 910                 mmd->chunks_sent = 0;
 911         } else if (vss_repos()) {
 912                 tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier);
 913                 set_eof_barrier(vsst);
 914                 mmd->chunks_sent = 0;
 915                 mmd->current_chunk = mmd->repos_request;
 916                 mmd->new_vss_status_flags &= ~VSS_REPOS;
 917                 set_mmd_offset();
 918         }
 919         if (need_to_request_new_audio_file(vsst)) {
 920                 PARA_DEBUG_LOG("ready and playing, but no audio file\n");
 921                 para_fd_set(vsst->afs_socket, &s->wfds, &s->max_fileno);
 922                 vsst->afsss = AFS_SOCKET_CHECK_FOR_WRITE;
 923         } else
 924                 para_fd_set(vsst->afs_socket, &s->rfds, &s->max_fileno);
 925         for (i = 0; senders[i].name; i++) {
 926                 if (!senders[i].pre_select)
 927                         continue;
 928                 senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds);
 929         }
 930         vss_compute_timeout(s, vsst);
 931 }
 932
 933 static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data)
 934 {
 935         char control[255], buf[8];
 936         struct msghdr msg = {.msg_iov = NULL};
 937         struct cmsghdr *cmsg;
 938         struct iovec iov;
 939         int ret = 0;
 940
 941         *fd = -1;
 942         iov.iov_base = buf;
 943         iov.iov_len = sizeof(buf);
 944         msg.msg_iov = &iov;
 945         msg.msg_iovlen = 1;
 946         msg.msg_control = control;
 947         msg.msg_controllen = sizeof(control);
 948         memset(buf, 0, sizeof(buf));
 949         ret = recvmsg(afs_socket, &msg, 0);
 950         if (ret < 0)
 951                 return -ERRNO_TO_PARA_ERROR(errno);
 952         if (iov.iov_len != sizeof(buf))
 953                 return -E_AFS_SHORT_READ;
 954         *code = *(uint32_t*)buf;
 955         *data =  *(uint32_t*)(buf + 4);
 956         for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
 957                 if (cmsg->cmsg_level != SOL_SOCKET
 958                         || cmsg->cmsg_type != SCM_RIGHTS)
 959                         continue;
 960                 if ((cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int) != 1)
 961                         continue;
 962                 *fd = *(int *)CMSG_DATA(cmsg);
 963         }
 964         return 1;
 965 }
 966
 967 #ifndef MAP_POPULATE
 968 #define MAP_POPULATE 0
 969 #endif
 970
 971 static void recv_afs_result(struct vss_task *vsst, fd_set *rfds)
 972 {
 973         int ret, passed_fd, shmid;
 974         uint32_t afs_code = 0, afs_data = 0;
 975         struct stat statbuf;
 976
 977         if (!FD_ISSET(vsst->afs_socket, rfds))
 978                 return;
 979         ret = recv_afs_msg(vsst->afs_socket, &passed_fd, &afs_code, &afs_data);
 980         if (ret == -ERRNO_TO_PARA_ERROR(EAGAIN))
 981                 return;
 982         if (ret < 0)
 983                 goto err;
 984         vsst->afsss = AFS_SOCKET_READY;
 985         PARA_DEBUG_LOG("fd: %d, code: %u, shmid: %u\n", passed_fd, afs_code,
 986                 afs_data);
 987         ret = -E_NOFD;
 988         if (afs_code != NEXT_AUDIO_FILE)
 989                 goto err;
 990         if (passed_fd < 0)
 991                 goto err;
 992         shmid = afs_data;
 993         ret = load_afd(shmid, &mmd->afd);
 994         if (ret < 0)
 995                 goto err;
 996         shm_destroy(shmid);
 997         ret = fstat(passed_fd, &statbuf);
 998         if (ret < 0) {
 999                 PARA_ERROR_LOG("fstat error:\n");
1000                 ret = -ERRNO_TO_PARA_ERROR(errno);
1001                 goto err;
1002         }
1003         mmd->size = statbuf.st_size;
1004         mmd->mtime = statbuf.st_mtime;
1005         ret = para_mmap(mmd->size, PROT_READ, MAP_PRIVATE | MAP_POPULATE,
1006                 passed_fd, 0, &vsst->map);
1007         if (ret < 0)
1008                 goto err;
1009         close(passed_fd);
1010         mmd->chunks_sent = 0;
1011         mmd->current_chunk = 0;
1012         mmd->offset = 0;
1013         mmd->events++;
1014         mmd->num_played++;
1015         mmd->new_vss_status_flags &= (~VSS_NEXT);
1016         afh_get_header(&mmd->afd.afhi, mmd->afd.audio_format_id,
1017                 vsst->map, mmd->size, &vsst->header_buf, &vsst->header_len);
1018         return;
1019 err:
1020         free(mmd->afd.afhi.chunk_table);
1021         if (passed_fd >= 0)
1022                 close(passed_fd);
1023         PARA_ERROR_LOG("%s\n", para_strerror(-ret));
1024         mmd->new_vss_status_flags = VSS_NEXT;
1025 }
1026
1027 /**
1028  * Main sending function.
1029  *
1030  * This function gets called from vss_post_select(). It checks whether the next
1031  * chunk of data should be pushed out. It obtains a pointer to the data to be
1032  * sent out as well as its length from mmd->afd.afhi. This information is then
1033  * passed to each supported sender's send() function as well as to the send()
1034  * functions of each registered fec client.
1035  */
1036 static void vss_send(struct vss_task *vsst)
1037 {
1038         int i, fec_active = 0;
1039         struct timeval due;
1040         struct fec_client *fc, *tmp_fc;
1041
1042         if (!vsst->map || !vss_playing())
1043                 return;
1044         if (chk_barrier("eof", &vsst->eof_barrier, &due, 1) < 0)
1045                 return;
1046         if (chk_barrier("data send", &vsst->data_send_barrier,
1047                         &due, 1) < 0)
1048                 return;
1049         list_for_each_entry_safe(fc, tmp_fc, &fec_client_list, node) {
1050                 if (fc->state == FEC_STATE_DISABLED)
1051                         continue;
1052                 if (!next_slice_is_due(fc, NULL)) {
1053                         fec_active = 1;
1054                         continue;
1055                 }
1056                 if (compute_next_fec_slice(fc, vsst) <= 0)
1057                         continue;
1058                 PARA_DEBUG_LOG("sending %d:%d (%u bytes)\n", fc->group.num,
1059                         fc->current_slice_num, fc->group.slice_bytes);
1060                 fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf,
1061                         fc->group.slice_bytes + FEC_HEADER_SIZE);
1062                 fc->current_slice_num++;
1063                 fec_active = 1;
1064         }
1065         if (mmd->current_chunk >= mmd->afd.afhi.chunks_total) { /* eof */
1066                 if (!fec_active)
1067                         mmd->new_vss_status_flags |= VSS_NEXT;
1068                 return;
1069         }
1070         compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv,
1071                 &mmd->stream_start, &due);
1072         if (tv_diff(&due, now, NULL) <= 0) {
1073                 char *buf;
1074                 size_t len;
1075
1076                 if (!mmd->chunks_sent) {
1077                         mmd->stream_start = *now;
1078                         mmd->events++;
1079                         set_mmd_offset();
1080                 }
1081                 /*
1082                  * We call the send function also in case of empty chunks as
1083                  * they might have still some data queued which can be sent in
1084                  * this case.
1085                  */
1086                 vss_get_chunk(mmd->current_chunk, vsst, &buf, &len);
1087                 for (i = 0; senders[i].name; i++) {
1088                         if (!senders[i].send)
1089                                 continue;
1090                         senders[i].send(mmd->current_chunk, mmd->chunks_sent,
1091                                 buf, len, vsst->header_buf, vsst->header_len);
1092                 }
1093                 /*
1094                  * Prefault next chunk(s)
1095                  *
1096                  * If the backing device of the memory-mapped audio file is
1097                  * slow and read-ahead is turned off or prevented for some
1098                  * reason, e.g. due to memory pressure, it may take much longer
1099                  * than the chunk interval to get the next chunk on the wire,
1100                  * causing buffer underruns on the client side. Mapping the
1101                  * file with MAP_POPULATE seems to help a bit, but it does not
1102                  * eliminate the delays completely. Moreover, it is supported
1103                  * only on Linux. So we do our own read-ahead here.
1104                  */
1105                 if (mmd->current_chunk > 0) { /* chunk 0 might be on the heap */
1106                         buf += len;
1107                         for (i = 0; i < 5 && buf < vsst->map + mmd->size; i++) {
1108                                 __a_unused volatile char x = *buf;
1109                                 buf += 4096;
1110                         }
1111                 }
1112                 mmd->chunks_sent++;
1113                 mmd->current_chunk++;
1114         }
1115 }
1116
1117 static void vss_post_select(struct sched *s, struct task *t)
1118 {
1119         int ret, i;
1120         struct vss_task *vsst = container_of(t, struct vss_task, task);
1121
1122
1123         if (mmd->sender_cmd_data.cmd_num >= 0) {
1124                 int num = mmd->sender_cmd_data.cmd_num,
1125                         sender_num = mmd->sender_cmd_data.sender_num;
1126
1127                 if (senders[sender_num].client_cmds[num]) {
1128                         ret = senders[sender_num].client_cmds[num]
1129                                 (&mmd->sender_cmd_data);
1130                         if (ret < 0)
1131                                 PARA_ERROR_LOG("%s\n", para_strerror(-ret));
1132                 }
1133                 mmd->sender_cmd_data.cmd_num = -1;
1134         }
1135         if (vsst->afsss != AFS_SOCKET_CHECK_FOR_WRITE)
1136                 recv_afs_result(vsst, &s->rfds);
1137         else if (FD_ISSET(vsst->afs_socket, &s->wfds)) {
1138                 PARA_NOTICE_LOG("requesting new fd from afs\n");
1139                 ret = write_buffer(vsst->afs_socket, "new");
1140                 if (ret < 0)
1141                         PARA_CRIT_LOG("%s\n", para_strerror(-ret));
1142                 else
1143                         vsst->afsss = AFS_SOCKET_AFD_PENDING;
1144         }
1145         for (i = 0; senders[i].name; i++) {
1146                 if (!senders[i].post_select)
1147                         continue;
1148                 senders[i].post_select(&s->rfds, &s->wfds);
1149         }
1150         if ((vss_playing() && !(mmd->vss_status_flags & VSS_PLAYING)) ||
1151                         (vss_next() && vss_playing()))
1152                 tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier);
1153         vss_send(vsst);
1154 }
1155
1156 /**
1157  * Initialize the virtual streaming system task.
1158  *
1159  * \param afs_socket The fd for communication with afs.
1160  * \param s The scheduler to register the vss task to.
1161  *
1162  * This also initializes all supported senders and starts streaming
1163  * if the --autoplay command line flag was given.
1164  */
1165 void init_vss_task(int afs_socket, struct sched *s)
1166 {
1167         static struct vss_task vss_task_struct, *vsst = &vss_task_struct;
1168         int i;
1169         char *hn = para_hostname(), *home = para_homedir();
1170         long unsigned announce_time = conf.announce_time_arg > 0?
1171                         conf.announce_time_arg : 300,
1172                 autoplay_delay = conf.autoplay_delay_arg > 0?
1173                         conf.autoplay_delay_arg : 0;
1174         vsst->header_interval.tv_sec = 5; /* should this be configurable? */
1175         vsst->afs_socket = afs_socket;
1176         vsst->task.pre_select = vss_pre_select;
1177         vsst->task.post_select = vss_post_select;
1178         ms2tv(announce_time, &vsst->announce_tv);
1179         PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst->announce_tv));
1180         INIT_LIST_HEAD(&fec_client_list);
1181         for (i = 0; senders[i].name; i++) {
1182                 PARA_NOTICE_LOG("initializing %s sender\n", senders[i].name);
1183                 senders[i].init(&senders[i]);
1184         }
1185         free(hn);
1186         free(home);
1187         mmd->sender_cmd_data.cmd_num = -1;
1188         if (conf.autoplay_given) {
1189                 struct timeval tmp;
1190                 mmd->vss_status_flags |= VSS_PLAYING;
1191                 mmd->new_vss_status_flags |= VSS_PLAYING;
1192                 ms2tv(autoplay_delay, &tmp);
1193                 tv_add(now, &tmp, &vsst->autoplay_barrier);
1194                 tv_add(&vsst->autoplay_barrier, &vsst->announce_tv,
1195                         &vsst->data_send_barrier);
1196         }
1197         sprintf(vsst->task.status, "vss task");
1198         register_task(s, &vsst->task);
1199 }