X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=8f177bab56b90a221f4ac84b73e4c54971df4a0c;hp=b9afc8ed637374c82918f3448695a1a2fdaa94a2;hb=d0d27cfa56d05eaa0e5567c75adfe606eab2957e;hpb=0bd00f72312a79900d211e735b135c220ddd0d68 diff --git a/vss.c b/vss.c index b9afc8ed..8f177bab 100644 --- a/vss.c +++ b/vss.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 1997-2011 Andre Noll + * Copyright (C) 1997-2012 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ @@ -26,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; @@ -132,6 +132,7 @@ struct fec_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 */ @@ -360,8 +361,17 @@ static void vss_get_chunk(int chunk_num, struct vss_task *vsst, 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; /* @@ -371,8 +381,6 @@ static void compute_group_size(struct vss_task *vsst, struct fec_group *g, * of exactly one chunk for these audio formats. */ for (i = 0;; i++) { - char *buf; - size_t len; int chunk_num = g->first_chunk + i; if (g->bytes > 0 && i >= max_chunks) /* duration limit */ @@ -501,7 +509,7 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) { int ret, i, k, n, data_slices; size_t len; - char *buf; + char *buf, *p; struct fec_group *g = &fc->group; if (fc->state == FEC_STATE_NONE) { @@ -561,16 +569,20 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) assert(i == g->num_header_slices - 1); } - /* setup data slices */ + /* + * 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 (; i < g->num_header_slices + data_slices; i++) { - if (buf + g->slice_bytes > vsst->map + mmd->size) { + 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. + * We must make a copy for this slice since using p + * directly would exceed the buffer. */ - uint32_t payload_size = vsst->map + mmd->size - buf; - memcpy(fc->extra_src_buf, buf, payload_size); + 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); @@ -578,8 +590,8 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) i++; break; } - fc->src_data[i] = (const unsigned char *)buf; - buf += g->slice_bytes; + fc->src_data[i] = (const unsigned char *)p; + p += g->slice_bytes; } if (i < k) { /* use arbitrary data for all remaining slices */ @@ -952,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; @@ -986,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); @@ -1074,6 +1090,25 @@ static void vss_send(struct vss_task *vsst) 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++; } @@ -1101,7 +1136,7 @@ static void vss_post_select(struct sched *s, struct task *t) recv_afs_result(vsst, &s->rfds); else if (FD_ISSET(vsst->afs_socket, &s->wfds)) { PARA_NOTICE_LOG("requesting new fd from afs\n"); - ret = send_buffer(vsst->afs_socket, "new"); + ret = write_buffer(vsst->afs_socket, "new"); if (ret < 0) PARA_CRIT_LOG("%s\n", para_strerror(-ret)); else @@ -1122,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; @@ -1159,5 +1195,5 @@ void init_vss_task(int afs_socket) &vsst->data_send_barrier); } sprintf(vsst->task.status, "vss task"); - register_task(&vsst->task); + register_task(s, &vsst->task); }