X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=e336a9e7413d469c71a6221cab0e86b0aae24867;hp=121373dc2f4fb77bb62985d20909b8ccefc972ae;hb=a5ad3d4c598bafe28e90e5b623e0a754d6e973d5;hpb=0c9c3239eb328c8b802447b4f9d2c922085e74ee diff --git a/vss.c b/vss.c index 121373dc..e336a9e7 100644 --- a/vss.c +++ b/vss.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 1997-2010 Andre Noll + * Copyright (C) 1997-2011 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ @@ -12,7 +12,6 @@ */ #include -#include #include #include "para.h" @@ -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. */ @@ -133,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 */ @@ -167,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. */ @@ -211,10 +213,11 @@ 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_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 bool need_audio_header(struct fec_client *fc, struct vss_task *vsst) @@ -326,6 +329,7 @@ static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) fc->src_data = para_realloc(fc->src_data, k * sizeof(char *)); 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->state = FEC_STATE_READY_TO_RUN; fc->next_header_time.tv_sec = 0; @@ -334,6 +338,26 @@ static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) return 1; } +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) { @@ -348,7 +372,7 @@ 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++) { - const char *buf; + char *buf; size_t len; int chunk_num = g->first_chunk + i; @@ -356,7 +380,7 @@ static void compute_group_size(struct vss_task *vsst, struct fec_group *g, break; if (chunk_num >= mmd->afd.afhi.chunks_total) /* eof */ break; - afh_get_chunk(chunk_num, &mmd->afd.afhi, vsst->map, &buf, &len); + vss_get_chunk(chunk_num, vsst, &buf, &len); if (g->bytes + len > max_bytes) break; /* Include this chunk */ @@ -478,7 +502,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; - const char *buf; + char *buf; struct fec_group *g = &fc->group; if (fc->state == FEC_STATE_NONE) { @@ -515,22 +539,36 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) fc->current_slice_num = 0; if (g->num == 0) 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 += g->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 */ - afh_get_chunk(g->first_chunk, &mmd->afd.afhi, vsst->map, &buf, &len); + 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) { /* * Can not use the memory mapped audio file for this - * slice as it goes beyond the map. This slice will not - * be fully used. + * slice as it goes beyond the map. */ uint32_t payload_size = vsst->map + mmd->size - buf; memcpy(fc->extra_src_buf, buf, payload_size); @@ -626,6 +664,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); } @@ -650,26 +689,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; @@ -760,42 +779,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; - - 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; + struct timeval tv; + struct fec_client *fc; + 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) @@ -806,6 +821,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; @@ -862,7 +879,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()) { @@ -900,9 +916,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) @@ -939,6 +953,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; @@ -973,8 +991,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); @@ -984,8 +1002,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); @@ -1041,7 +1059,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) { @@ -1054,8 +1072,7 @@ 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; @@ -1064,6 +1081,23 @@ static void vss_send(struct vss_task *vsst) } mmd->chunks_sent++; mmd->current_chunk++; + /* + * 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. + */ + buf += len; + for (i = 0; i < 5 && buf < vsst->map + mmd->size; i++) { + __a_unused volatile char x = *buf; + buf += 4096; + } } } @@ -1146,5 +1180,6 @@ void init_vss_task(int afs_socket) tv_add(&vsst->autoplay_barrier, &vsst->announce_tv, &vsst->data_send_barrier); } + sprintf(vsst->task.status, "vss task"); register_task(&vsst->task); }