X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=bfea37a191f8b90db60db720916eb37ebf900223;hp=25082834eb0160de013dd11639f3354160ee261b;hb=e5fbc490c2c16ecfa7bce58a18e11a0f7d382b91;hpb=e56d454fbd865f34cb5e043ff43585754631b079 diff --git a/vss.c b/vss.c index 25082834..bfea37a1 100644 --- a/vss.c +++ b/vss.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 1997-2010 Andre Noll + * Copyright (C) 1997-2012 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ @@ -12,7 +12,6 @@ */ #include -#include #include #include "para.h" @@ -27,10 +26,10 @@ #include "server.cmdline.h" #include "list.h" #include "send.h" +#include "sched.h" #include "vss.h" #include "ipc.h" #include "fd.h" -#include "sched.h" extern struct misc_meta_data *mmd; @@ -86,7 +85,7 @@ struct vss_task { /** Used by the scheduler. */ struct task task; /** Pointer to the header of the mapped audio file. */ - const char *header_buf; + char *header_buf; /** Length of the audio file header. */ size_t header_len; /** Time between audio file headers are sent. */ @@ -129,8 +128,11 @@ struct fec_group { struct timeval slice_duration; /** Group contains the audio file header that occupies that many slices. */ uint8_t num_header_slices; + /** Number of bytes per slice for this group. */ + uint16_t slice_bytes; }; +/** A FEC client is always in one of these states. */ enum fec_client_state { FEC_STATE_NONE = 0, /**< not initialized and not enabled */ FEC_STATE_DISABLED, /**< temporarily disabled */ @@ -165,6 +167,8 @@ struct fec_client { struct timeval next_header_time; /** Used for the last source pointer of an audio file. */ unsigned char *extra_src_buf; + /** Needed for the last slice of the audio file header. */ + unsigned char *extra_header_buf; /** Extra slices needed to store largest chunk + header. */ int num_extra_slices; /** Contains the FEC-encoded data. */ @@ -209,47 +213,68 @@ static void write_fec_header(struct fec_client *fc, struct vss_task *vsst) write_u32(buf + 14, g->bytes); write_u8(buf + 18, fc->current_slice_num); - write_u16(buf + 20, fc->mps - FEC_HEADER_SIZE); + write_u8(buf + 19, 0); /* unused */ + write_u16(buf + 20, g->slice_bytes); write_u8(buf + 22, g->first_chunk? 0 : 1); write_u8(buf + 23, vsst->header_len? 1 : 0); - memset(buf + 24, 0, 7); + memset(buf + 24, 0, 8); } -static int need_audio_header(struct fec_client *fc, struct vss_task *vsst) +static bool need_audio_header(struct fec_client *fc, struct vss_task *vsst) { if (!mmd->current_chunk) { tv_add(now, &vsst->header_interval, &fc->next_header_time); - return 0; + return false; } if (!vsst->header_buf) - return 0; - if (!vsst->header_len) - return 0; - if (fc->group.num && tv_diff(&fc->next_header_time, now, NULL) > 0) - return 0; + return false; + if (vsst->header_len == 0) + return false; + if (fc->group.num > 0) { + if (!fc->fcp->need_periodic_header) + return false; + if (tv_diff(&fc->next_header_time, now, NULL) > 0) + return false; + } tv_add(now, &vsst->header_interval, &fc->next_header_time); - return 1; + return true; } -static int num_slices(long unsigned bytes, int mps, int rs) +static bool need_data_slices(struct fec_client *fc, struct vss_task *vsst) +{ + if (fc->group.num > 0) + return true; + if (!vsst->header_buf) + return true; + if (vsst->header_len == 0) + return true; + if (fc->fcp->need_periodic_header) + return true; + return false; +} + +static int num_slices(long unsigned bytes, int max_payload, int rs) { - int m = mps - FEC_HEADER_SIZE; int ret; - assert(m > 0); + assert(max_payload > 0); assert(rs > 0); - ret = DIV_ROUND_UP(bytes, m); + ret = DIV_ROUND_UP(bytes, max_payload); if (ret + rs > 255) return -E_BAD_CT; return ret; } /* set group start and group duration */ -static void set_group_timing(struct fec_client *fc, struct fec_group *g) +static void set_group_timing(struct fec_client *fc, struct vss_task *vsst) { + struct fec_group *g = &fc->group; struct timeval *chunk_tv = vss_chunk_time(); - tv_scale(g->num_chunks, chunk_tv, &g->duration); + if (!need_data_slices(fc, vsst)) + ms2tv(200, &g->duration); + else + tv_scale(g->num_chunks, chunk_tv, &g->duration); tv_divide(fc->fcp->slices_per_group + fc->num_extra_slices, &g->duration, &g->slice_duration); PARA_DEBUG_LOG("durations (group/chunk/slice): %lu/%lu/%lu\n", @@ -258,7 +283,7 @@ static void set_group_timing(struct fec_client *fc, struct fec_group *g) static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) { - int k, n, ret, mps; + int k, n, ret; int hs, ds, rs; /* header/data/redundant slices */ struct fec_client_parms *fcp = fc->fcp; @@ -272,59 +297,220 @@ static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) ret = fcp->init_fec(fc->sc); if (ret < 0) return ret; - mps = ret; + fc->mps = ret; } else - mps = generic_max_transport_msg_size(fc->sc->fd); - if (mps <= FEC_HEADER_SIZE) + fc->mps = generic_max_transport_msg_size(fc->sc->fd); + if (fc->mps <= FEC_HEADER_SIZE) return -ERRNO_TO_PARA_ERROR(EINVAL); rs = fc->fcp->slices_per_group - fc->fcp->data_slices_per_group; - ret = num_slices(vsst->header_len, mps, rs); + ret = num_slices(vsst->header_len, fc->mps - FEC_HEADER_SIZE, rs); if (ret < 0) - goto err; + return ret; hs = ret; - ret = num_slices(afh_get_largest_chunk_size(&mmd->afd.afhi), - mps, rs); + ret = num_slices(mmd->afd.max_chunk_size, fc->mps - FEC_HEADER_SIZE, rs); if (ret < 0) - goto err; + return ret; ds = ret; - k = ret + ds; + if (fc->fcp->need_periodic_header) + k = hs + ds; + else + k = PARA_MAX(hs, ds); if (k < fc->fcp->data_slices_per_group) k = fc->fcp->data_slices_per_group; + fc->num_extra_slices = k - fc->fcp->data_slices_per_group; n = k + rs; - PARA_CRIT_LOG("hs: %d, ds: %d, rs: %d, k: %d, n: %d\n", hs, ds, rs, k, n); fec_free(fc->parms); ret = fec_new(k, n, &fc->parms); if (ret < 0) return ret; - fc->num_extra_slices = k - fc->fcp->data_slices_per_group; - PARA_NOTICE_LOG("fec parms %d:%d:%d (%d extra slices)\n", - mps, k, n, fc->num_extra_slices); + PARA_INFO_LOG("mps: %d, k: %d, n: %d, extra slices: %d\n", + fc->mps, k, n, fc->num_extra_slices); fc->src_data = para_realloc(fc->src_data, k * sizeof(char *)); - fc->enc_buf = para_realloc(fc->enc_buf, mps); - memset(fc->enc_buf, 0, mps); - fc->extra_src_buf = para_realloc(fc->extra_src_buf, mps); - memset(fc->extra_src_buf, 0, mps); + fc->enc_buf = para_realloc(fc->enc_buf, fc->mps); + fc->extra_src_buf = para_realloc(fc->extra_src_buf, fc->mps); + fc->extra_header_buf = para_realloc(fc->extra_header_buf, fc->mps); - fc->mps = mps; fc->state = FEC_STATE_READY_TO_RUN; fc->next_header_time.tv_sec = 0; fc->stream_start = *now; fc->first_stream_chunk = mmd->current_chunk; return 1; -err: - fec_free(fc->parms); - return ret; +} + +static void vss_get_chunk(int chunk_num, struct vss_task *vsst, + char **buf, size_t *sz) +{ + /* + * Chunk zero is special for header streams: It is the first portion of + * the audio file which consists of the audio file header. It may be + * arbitrary large due to embedded meta data. Audio format handlers may + * replace the header by a stripped one with meta data omitted which is + * of bounded size. We always use the stripped header for streaming + * rather than the unmodified header (chunk zero). + */ + if (chunk_num == 0 && vsst->header_len > 0) { + *buf = vsst->header_buf; /* stripped header */ + *sz = vsst->header_len; + return; + } + afh_get_chunk(chunk_num, &mmd->afd.afhi, vsst->map, (const char **)buf, + sz); +} + +static void compute_group_size(struct vss_task *vsst, struct fec_group *g, + int max_bytes) +{ + char *buf; + size_t len; + int i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time())); + + if (g->first_chunk == 0) { + g->num_chunks = 1; + vss_get_chunk(0, vsst, &buf, &len); + g->bytes = len; + return; + } + + g->num_chunks = 0; + g->bytes = 0; + /* + * Include chunks into the group until the group duration is at least + * 150ms. For ogg and wma, a single chunk's duration (ogg page/wma + * super frame) is already larger than 150ms, so a FEC group consists + * of exactly one chunk for these audio formats. + */ + for (i = 0;; i++) { + int chunk_num = g->first_chunk + i; + + if (g->bytes > 0 && i >= max_chunks) /* duration limit */ + break; + if (chunk_num >= mmd->afd.afhi.chunks_total) /* eof */ + break; + vss_get_chunk(chunk_num, vsst, &buf, &len); + if (g->bytes + len > max_bytes) + break; + /* Include this chunk */ + g->bytes += len; + g->num_chunks++; + } + assert(g->num_chunks); +} + +/* + * Compute the slice size of the next group. + * + * The FEC parameters n and k are fixed but the slice size varies per + * FEC group. We'd like to choose slices as small as possible to avoid + * unnecessary FEC calculations but large enough to guarantee that the + * k data slices suffice to encode the header (if needed) and the data + * chunk(s). + * + * Once we know the payload of the next group, we define the number s + * of bytes per slice for this group by + * + * s = ceil(payload / k) + * + * However, for header streams, computing s is more complicated since no + * overlapping of header and data slices is possible. Hence we have k >= + * 2 and s must satisfy + * + * (*) ceil(h / s) + ceil(d / s) <= k + * + * where h and d are payload of the header and the data chunk(s) + * respectively. In general there is no value for s such that (*) + * becomes an equality, for example if h = 4000, d = 5000 and k = 10. + * + * We use the following approach for computing a suitable value for s: + * + * Let + * k1 := ceil(k * min(h, d) / (h + d)), + * k2 := k - k1. + * + * Note that k >= 2 implies k1 > 0 and k2 > 0, so + * + * s := max(ceil(min(h, d) / k1), ceil(max(h, d) / k2)) + * + * is well-defined. Inequality (*) holds for this value of s since k1 + * slices suffice to store min(h, d) while k2 slices suffice to store + * max(h, d), i.e. the first addent of (*) is bounded by k1 and the + * second by k2. + * + * For the above example we obtain + * + * k1 = ceil(10 * 4000 / 9000) = 5, k2 = 5, + * s = max(4000 / 5, 5000 / 5) = 1000, + * + * which is optimal since a slice size of 999 bytes would already require + * 11 slices. + */ +static int compute_slice_size(struct fec_client *fc, struct vss_task *vsst) +{ + struct fec_group *g = &fc->group; + int k = fc->fcp->data_slices_per_group + fc->num_extra_slices; + int n = fc->fcp->slices_per_group + fc->num_extra_slices; + int ret, k1, k2, h, d, min, max, sum; + int max_slice_bytes = fc->mps - FEC_HEADER_SIZE; + int max_group_bytes; + + if (!need_audio_header(fc, vsst)) { + max_group_bytes = k * max_slice_bytes; + g->num_header_slices = 0; + compute_group_size(vsst, g, max_group_bytes); + g->slice_bytes = DIV_ROUND_UP(g->bytes, k); + if (g->slice_bytes == 0) + g->slice_bytes = 1; + return 1; + } + if (!need_data_slices(fc, vsst)) { + g->bytes = 0; + g->num_chunks = 0; + g->slice_bytes = DIV_ROUND_UP(vsst->header_len, k); + g->num_header_slices = k; + return 1; + } + h = vsst->header_len; + max_group_bytes = (k - num_slices(h, max_slice_bytes, n - k)) + * max_slice_bytes; + compute_group_size(vsst, g, max_group_bytes); + d = g->bytes; + if (d == 0) { + g->slice_bytes = DIV_ROUND_UP(h, k); + ret = num_slices(vsst->header_len, g->slice_bytes, n - k); + if (ret < 0) + return ret; + g->num_header_slices = ret; + return 1; + } + min = PARA_MIN(h, d); + max = PARA_MAX(h, d); + sum = h + d; + k1 = DIV_ROUND_UP(k * min, sum); + k2 = k - k1; + assert(k1 > 0); + assert(k2 > 0); + + g->slice_bytes = PARA_MAX(DIV_ROUND_UP(min, k1), DIV_ROUND_UP(max, k2)); + /* + * This value of s := g->slice_bytes satisfies inequality (*) above, + * but it might be larger than max_slice_bytes. However, we know that + * max_slice_bytes are sufficient to store header and data, so: + */ + g->slice_bytes = PARA_MIN((int)g->slice_bytes, max_slice_bytes); + + ret = num_slices(vsst->header_len, g->slice_bytes, n - k); + if (ret < 0) + return ret; + g->num_header_slices = ret; + return 1; } static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) { int ret, i, k, n, data_slices; size_t len; - const char *buf, *start_buf; + char *buf, *p; struct fec_group *g = &fc->group; - unsigned slice_bytes; - uint32_t max_data_size; if (fc->state == FEC_STATE_NONE) { ret = initialize_fec_client(fc, vsst); @@ -333,7 +519,6 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) g->first_chunk = mmd->current_chunk; g->num = 0; g->start = *now; - } else { struct timeval tmp; if (g->first_chunk + g->num_chunks >= mmd->afd.afhi.chunks_total) @@ -344,75 +529,82 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) */ tmp = g->start; tv_add(&tmp, &g->duration, &g->start); - set_group_timing(fc, g); + set_group_timing(fc, vsst); g->first_chunk += g->num_chunks; g->num++; } - slice_bytes = fc->mps - FEC_HEADER_SIZE; - PARA_CRIT_LOG("slice_bytes: %d\n", slice_bytes); k = fc->fcp->data_slices_per_group + fc->num_extra_slices; n = fc->fcp->slices_per_group + fc->num_extra_slices; - PARA_CRIT_LOG("k: %d, n: %d\n", k, n); - if (need_audio_header(fc, vsst)) { - ret = num_slices(vsst->header_len, slice_bytes, n - k); - if (ret < 0) - return ret; - g->num_header_slices = ret; - } else - g->num_header_slices = 0; - afh_get_chunk(g->first_chunk, &mmd->afd.afhi, vsst->map, &start_buf, - &len); - data_slices = k - g->num_header_slices; - assert(data_slices); - max_data_size = slice_bytes * data_slices; - g->bytes = 0; - for (i = g->first_chunk; i < mmd->afd.afhi.chunks_total; i++) { - afh_get_chunk(i, &mmd->afd.afhi, vsst->map, &buf, &len); - if (g->bytes + len > max_data_size) - break; - g->bytes += len; - } - g->num_chunks = i - g->first_chunk; - assert(g->num_chunks); + + compute_slice_size(fc, vsst); + assert(g->slice_bytes > 0); + ret = num_slices(g->bytes, g->slice_bytes, n - k); + if (ret < 0) + return ret; + data_slices = ret; + assert(g->num_header_slices + data_slices <= k); fc->current_slice_num = 0; if (g->num == 0) - set_group_timing(fc, g); - + set_group_timing(fc, vsst); /* setup header slices */ buf = vsst->header_buf; for (i = 0; i < g->num_header_slices; i++) { - fc->src_data[i] = (const unsigned char *)buf; - buf += slice_bytes; + uint32_t payload_size; + if (buf + g->slice_bytes <= vsst->header_buf + vsst->header_len) { + fc->src_data[i] = (const unsigned char *)buf; + buf += g->slice_bytes; + continue; + } + /* + * Can not use vss->header_buf for this slice as it + * goes beyond the buffer. This slice will not be fully + * used. + */ + payload_size = vsst->header_buf + vsst->header_len - buf; + memcpy(fc->extra_header_buf, buf, payload_size); + if (payload_size < g->slice_bytes) + memset(fc->extra_header_buf + payload_size, 0, + g->slice_bytes - payload_size); + fc->src_data[i] = fc->extra_header_buf; + assert(i == g->num_header_slices - 1); } - /* setup data slices */ - buf = start_buf; - for (i = g->num_header_slices; i < k; i++) { - if (buf + slice_bytes > vsst->map + mmd->size) + /* + * Setup data slices. Note that for ogg streams chunk 0 points to a + * buffer on the heap rather than to the mapped audio file. + */ + vss_get_chunk(g->first_chunk, vsst, &buf, &len); + for (p = buf; i < g->num_header_slices + data_slices; i++) { + if (p + g->slice_bytes > buf + g->bytes) { /* - * Can not use the memory mapped audio file for this - * slice as it goes beyond the map. This slice will not - * be fully used. + * We must make a copy for this slice since using p + * directly would exceed the buffer. */ + uint32_t payload_size = buf + g->bytes - p; + assert(payload_size + FEC_HEADER_SIZE <= fc->mps); + memcpy(fc->extra_src_buf, p, payload_size); + if (payload_size < g->slice_bytes) + memset(fc->extra_src_buf + payload_size, 0, + g->slice_bytes - payload_size); + fc->src_data[i] = fc->extra_src_buf; + i++; break; - fc->src_data[i] = (const unsigned char *)buf; - buf += slice_bytes; + } + fc->src_data[i] = (const unsigned char *)p; + p += g->slice_bytes; } if (i < k) { - uint32_t payload_size = vsst->map + mmd->size - buf; - memcpy(fc->extra_src_buf, buf, payload_size); - fc->src_data[i] = fc->extra_src_buf; - i++; /* use arbitrary data for all remaining slices */ buf = vsst->map; for (; i < k; i++) fc->src_data[i] = (const unsigned char *)buf; } - PARA_DEBUG_LOG("FEC group %d: %d chunks (%d - %d), " - "%d header slices, %d data slices\n", + PARA_DEBUG_LOG("FEC group %d: %d chunks (%d - %d), %d bytes\n", g->num, g->num_chunks, g->first_chunk, - g->first_chunk + g->num_chunks - 1, - g->num_header_slices, data_slices + g->first_chunk + g->num_chunks - 1, g->bytes + ); + PARA_DEBUG_LOG("slice_bytes: %d, %d header slices, %d data slices\n", + g->slice_bytes, g->num_header_slices, data_slices ); return 1; } @@ -433,7 +625,7 @@ static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst) } write_fec_header(fc, vsst); fec_encode(fc->parms, fc->src_data, fc->enc_buf + FEC_HEADER_SIZE, - fc->current_slice_num, fc->mps - FEC_HEADER_SIZE); + fc->current_slice_num, fc->group.slice_bytes); return 1; } @@ -483,6 +675,7 @@ void vss_del_fec_client(struct fec_client *fc) free(fc->src_data); free(fc->enc_buf); free(fc->extra_src_buf); + free(fc->extra_header_buf); fec_free(fc->parms); free(fc); } @@ -507,26 +700,6 @@ static int next_slice_is_due(struct fec_client *fc, struct timeval *diff) return ret < 0? 1 : 0; } -static void compute_slice_timeout(struct timeval *timeout) -{ - struct fec_client *fc; - - list_for_each_entry(fc, &fec_client_list, node) { - struct timeval diff; - - if (fc->state != FEC_STATE_READY_TO_RUN) - continue; - if (next_slice_is_due(fc, &diff)) { - timeout->tv_sec = 0; - timeout->tv_usec = 0; - return; - } - /* timeout = min(timeout, diff) */ - if (tv_diff(&diff, timeout, NULL) < 0) - *timeout = diff; - } -} - static void set_eof_barrier(struct vss_task *vsst) { struct fec_client *fc; @@ -617,42 +790,38 @@ static int chk_barrier(const char *bname, const struct timeval *barrier, return -1; } -/* - * != NULL: timeout for next chunk - * NULL: nothing to do - */ -static struct timeval *vss_compute_timeout(struct vss_task *vsst) +static void vss_compute_timeout(struct sched *s, struct vss_task *vsst) { - static struct timeval the_timeout; - struct timeval next_chunk; + struct timeval tv; + struct fec_client *fc; - if (vss_next() && vsst->map) { - /* only sleep a bit, nec*/ - the_timeout.tv_sec = 0; - the_timeout.tv_usec = 100; - return &the_timeout; - } - if (chk_barrier("autoplay_delay", &vsst->autoplay_barrier, - &the_timeout, 1) < 0) - return &the_timeout; - if (chk_barrier("eof", &vsst->eof_barrier, &the_timeout, 1) < 0) - return &the_timeout; - if (chk_barrier("data send", &vsst->data_send_barrier, - &the_timeout, 1) < 0) - return &the_timeout; if (!vss_playing() || !vsst->map) - return NULL; + return; + if (vss_next() && vsst->map) /* only sleep a bit, nec*/ + return sched_request_timeout_ms(100, s); + + /* Each of these barriers must have passed until we may proceed */ + if (sched_request_barrier(&vsst->autoplay_barrier, s) == 1) + return; + if (sched_request_barrier(&vsst->eof_barrier, s) == 1) + return; + if (sched_request_barrier(&vsst->data_send_barrier, s) == 1) + return; + /* + * Compute the select timeout as the minimal time until the next + * chunk/slice is due for any client. + */ compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv, - &mmd->stream_start, &next_chunk); - if (chk_barrier("chunk", &next_chunk, &the_timeout, 0) >= 0) { - /* chunk is due or bof */ - the_timeout.tv_sec = 0; - the_timeout.tv_usec = 0; - return &the_timeout; + &mmd->stream_start, &tv); + if (sched_request_barrier_or_min_delay(&tv, s) == 0) + return; + list_for_each_entry(fc, &fec_client_list, node) { + if (fc->state != FEC_STATE_READY_TO_RUN) + continue; + if (next_slice_is_due(fc, &tv)) + return sched_min_delay(s); + sched_request_timeout(&tv, s); } - /* compute min of current timeout and next slice time */ - compute_slice_timeout(&the_timeout); - return &the_timeout; } static void vss_eof(struct vss_task *vsst) @@ -663,6 +832,8 @@ static void vss_eof(struct vss_task *vsst) if (mmd->new_vss_status_flags & VSS_NOMORE) mmd->new_vss_status_flags = VSS_NEXT; set_eof_barrier(vsst); + afh_free_header(vsst->header_buf, mmd->afd.audio_format_id); + vsst->header_buf = NULL; para_munmap(vsst->map, mmd->size); vsst->map = NULL; mmd->chunks_sent = 0; @@ -719,7 +890,6 @@ static void set_mmd_offset(void) static void vss_pre_select(struct sched *s, struct task *t) { int i; - struct timeval *tv; struct vss_task *vsst = container_of(t, struct vss_task, task); if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { @@ -757,9 +927,7 @@ static void vss_pre_select(struct sched *s, struct task *t) continue; senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds); } - tv = vss_compute_timeout(vsst); - if (tv) - sched_request_timeout(tv, s); + vss_compute_timeout(s, vsst); } static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data) @@ -796,6 +964,10 @@ static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data) return 1; } +#ifndef MAP_POPULATE +#define MAP_POPULATE 0 +#endif + static void recv_afs_result(struct vss_task *vsst, fd_set *rfds) { int ret, passed_fd, shmid; @@ -830,8 +1002,8 @@ static void recv_afs_result(struct vss_task *vsst, fd_set *rfds) } mmd->size = statbuf.st_size; mmd->mtime = statbuf.st_mtime; - ret = para_mmap(mmd->size, PROT_READ, MAP_PRIVATE, passed_fd, - 0, &vsst->map); + ret = para_mmap(mmd->size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, + passed_fd, 0, &vsst->map); if (ret < 0) goto err; close(passed_fd); @@ -841,8 +1013,8 @@ static void recv_afs_result(struct vss_task *vsst, fd_set *rfds) mmd->events++; mmd->num_played++; mmd->new_vss_status_flags &= (~VSS_NEXT); - afh_get_header(&mmd->afd.afhi, vsst->map, &vsst->header_buf, - &vsst->header_len); + afh_get_header(&mmd->afd.afhi, mmd->afd.audio_format_id, + vsst->map, mmd->size, &vsst->header_buf, &vsst->header_len); return; err: free(mmd->afd.afhi.chunk_table); @@ -884,8 +1056,9 @@ static void vss_send(struct vss_task *vsst) if (compute_next_fec_slice(fc, vsst) <= 0) continue; PARA_DEBUG_LOG("sending %d:%d (%u bytes)\n", fc->group.num, - fc->current_slice_num, fc->mps); - fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf, fc->mps); + fc->current_slice_num, fc->group.slice_bytes); + fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf, + fc->group.slice_bytes + FEC_HEADER_SIZE); fc->current_slice_num++; fec_active = 1; } @@ -897,7 +1070,7 @@ static void vss_send(struct vss_task *vsst) compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv, &mmd->stream_start, &due); if (tv_diff(&due, now, NULL) <= 0) { - const char *buf; + char *buf; size_t len; if (!mmd->chunks_sent) { @@ -910,14 +1083,32 @@ static void vss_send(struct vss_task *vsst) * they might have still some data queued which can be sent in * this case. */ - afh_get_chunk(mmd->current_chunk, &mmd->afd.afhi, vsst->map, - &buf, &len); + vss_get_chunk(mmd->current_chunk, vsst, &buf, &len); for (i = 0; senders[i].name; i++) { if (!senders[i].send) continue; senders[i].send(mmd->current_chunk, mmd->chunks_sent, buf, len, vsst->header_buf, vsst->header_len); } + /* + * Prefault next chunk(s) + * + * If the backing device of the memory-mapped audio file is + * slow and read-ahead is turned off or prevented for some + * reason, e.g. due to memory pressure, it may take much longer + * than the chunk interval to get the next chunk on the wire, + * causing buffer underruns on the client side. Mapping the + * file with MAP_POPULATE seems to help a bit, but it does not + * eliminate the delays completely. Moreover, it is supported + * only on Linux. So we do our own read-ahead here. + */ + if (mmd->current_chunk > 0) { /* chunk 0 might be on the heap */ + buf += len; + for (i = 0; i < 5 && buf < vsst->map + mmd->size; i++) { + __a_unused volatile char x = *buf; + buf += 4096; + } + } mmd->chunks_sent++; mmd->current_chunk++; } @@ -966,11 +1157,12 @@ static void vss_post_select(struct sched *s, struct task *t) * Initialize the virtual streaming system task. * * \param afs_socket The fd for communication with afs. + * \param s The scheduler to register the vss task to. * * This also initializes all supported senders and starts streaming * if the --autoplay command line flag was given. */ -void init_vss_task(int afs_socket) +void init_vss_task(int afs_socket, struct sched *s) { static struct vss_task vss_task_struct, *vsst = &vss_task_struct; int i; @@ -1002,5 +1194,6 @@ void init_vss_task(int afs_socket) tv_add(&vsst->autoplay_barrier, &vsst->announce_tv, &vsst->data_send_barrier); } - register_task(&vsst->task); + sprintf(vsst->task.status, "vss task"); + register_task(s, &vsst->task); }