X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=21ac37470e4748ea407fd7bacb96dfb42f3ee55a;hp=adc0cb63cea1204d3bfef811c492b807b74001c2;hb=e386c19376fa959b91710145923149d2d5dfbb29;hpb=fd652c08d4b5b26032646c449e1979c48018b70a diff --git a/vss.c b/vss.c index adc0cb63..21ac3747 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. */ @@ -129,6 +129,8 @@ 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; }; enum fec_client_state { @@ -169,6 +171,8 @@ struct fec_client { int num_extra_slices; /** Contains the FEC-encoded data. */ unsigned char *enc_buf; + /** Maximal packet size. */ + int mps; }; /** @@ -207,93 +211,281 @@ 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, p->max_slice_bytes - 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, struct fec_client *fc, uint8_t *result) +static bool need_data_slices(struct fec_client *fc, struct vss_task *vsst) { - unsigned long m = fc->fcp->max_slice_bytes - FEC_HEADER_SIZE; - unsigned rv, redundant_slices = fc->fcp->slices_per_group - - fc->fcp->data_slices_per_group; + 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; +} - if (!m) - return -E_BAD_CT; - rv = (bytes + m - 1) / m; - if (rv + redundant_slices > 255) +static int num_slices(long unsigned bytes, int max_payload, int rs) +{ + int ret; + + assert(max_payload > 0); + assert(rs > 0); + ret = DIV_ROUND_UP(bytes, max_payload); + if (ret + rs > 255) return -E_BAD_CT; - *result = rv; - return 1; + 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", tv2ms(&g->duration), tv2ms(chunk_tv), tv2ms(&g->slice_duration)); } -static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) +static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) { - int ret, i, k, data_slices; - size_t len; - const char *buf, *start_buf; - struct fec_group *g = &fc->group; - unsigned slice_bytes = fc->fcp->max_slice_bytes - FEC_HEADER_SIZE; - uint32_t max_data_size; - - if (fc->first_stream_chunk < 0) { - uint8_t hs, ds; /* needed header/data slices */ - uint8_t rs = fc->fcp->slices_per_group - - fc->fcp->data_slices_per_group; /* redundant slices */ - int n; + int k, n, ret; + int hs, ds, rs; /* header/data/redundant slices */ + struct fec_client_parms *fcp = fc->fcp; - ret = num_slices(vsst->header_len, fc, &hs); + /* set mps */ + if (fcp->init_fec) { + /* + * Set the maximum slice size to the Maximum Packet Size if the + * transport protocol allows to determine this value. The user + * can specify a slice size up to this value. + */ + ret = fcp->init_fec(fc->sc); if (ret < 0) return ret; - ret = num_slices(afh_get_largest_chunk_size(&mmd->afd.afhi), - fc, &ds); + fc->mps = ret; + } else + 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, fc->mps - FEC_HEADER_SIZE, rs); + if (ret < 0) + return ret; + hs = ret; + ret = num_slices(mmd->afd.max_chunk_size, fc->mps - FEC_HEADER_SIZE, rs); + if (ret < 0) + return ret; + ds = ret; + 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; + fec_free(fc->parms); + ret = fec_new(k, n, &fc->parms); + if (ret < 0) + return ret; + 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, fc->mps); + fc->extra_src_buf = para_realloc(fc->extra_src_buf, fc->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; +} + +static void compute_group_size(struct vss_task *vsst, struct fec_group *g, + int max_bytes) +{ + int i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time())); + + 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++) { + const char *buf; + size_t len; + 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; + afh_get_chunk(chunk_num, &mmd->afd.afhi, vsst->map, &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; - k = (int)hs + ds; - if (k > 255) - return -E_BAD_CT; - if (k < fc->fcp->data_slices_per_group) - k = fc->fcp->data_slices_per_group; - n = k + rs; - fc->num_extra_slices = k - fc->fcp->data_slices_per_group; - PARA_NOTICE_LOG("fec parms %d:%d:%d (%d extra slices)\n", - slice_bytes, k, n, fc->num_extra_slices); - fec_free(fc->parms); - fc->src_data = para_realloc(fc->src_data, k * sizeof(char *)); - ret = fec_new(k, n, &fc->parms); + 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; + struct fec_group *g = &fc->group; + + if (fc->state == FEC_STATE_NONE) { + ret = initialize_fec_client(fc, vsst); if (ret < 0) return ret; - fc->stream_start = *now; - fc->first_stream_chunk = mmd->current_chunk; g->first_chunk = mmd->current_chunk; g->num = 0; g->start = *now; @@ -307,77 +499,71 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) */ tmp = g->start; tv_add(&tmp, &g->duration, &g->start); - k = fc->fcp->data_slices_per_group + fc->num_extra_slices; - set_group_timing(fc, g); + set_group_timing(fc, vsst); g->first_chunk += g->num_chunks; g->num++; } - if (need_audio_header(fc, vsst)) { - ret = num_slices(vsst->header_len, fc, &g->num_header_slices); - if (ret < 0) - return 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); + k = fc->fcp->data_slices_per_group + fc->num_extra_slices; + n = fc->fcp->slices_per_group + fc->num_extra_slices; + + 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; + buf += g->slice_bytes; } /* setup data slices */ - buf = start_buf; - for (i = g->num_header_slices; i < k; i++) { - if (buf + slice_bytes > vsst->map + mmd->size) + afh_get_chunk(g->first_chunk, &mmd->afd.afhi, vsst->map, &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. */ + uint32_t payload_size = vsst->map + mmd->size - buf; + memcpy(fc->extra_src_buf, buf, 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; + buf += 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; } static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst) { - if (fc->first_stream_chunk < 0 || fc->current_slice_num + if (fc->state == FEC_STATE_NONE || fc->current_slice_num == fc->fcp->slices_per_group + fc->num_extra_slices) { int ret = setup_next_fec_group(fc, vsst); if (ret == 0) @@ -391,8 +577,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->fcp->max_slice_bytes - FEC_HEADER_SIZE); + fc->current_slice_num, fc->group.slice_bytes); return 1; } @@ -458,7 +643,7 @@ static int next_slice_is_due(struct fec_client *fc, struct timeval *diff) struct timeval tmp, next; int ret; - if (fc->first_stream_chunk < 0) + if (fc->state == FEC_STATE_NONE) return 1; tv_scale(fc->current_slice_num, &fc->group.slice_duration, &tmp); tv_add(&tmp, &fc->group.start, &next); @@ -636,18 +821,6 @@ static void vss_eof(struct vss_task *vsst) mmd->events++; } -/** - * Get the list of all supported audio formats. - * - * \return Aa space separated list of all supported audio formats - * It is not allocated at runtime, i.e. there is no need to free - * the returned string in the caller. - */ -const char *supported_audio_formats(void) -{ - return SUPPORTED_AUDIO_FORMATS; -} - static int need_to_request_new_audio_file(struct vss_task *vsst) { struct timeval diff; @@ -690,7 +863,7 @@ static void set_mmd_offset(void) static void vss_pre_select(struct sched *s, struct task *t) { int i; - struct timeval *tv, diff; + struct timeval *tv; struct vss_task *vsst = container_of(t, struct vss_task, task); if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { @@ -698,10 +871,8 @@ static void vss_pre_select(struct sched *s, struct task *t) for (i = 0; senders[i].name; i++) if (senders[i].shutdown_clients) senders[i].shutdown_clients(); - list_for_each_entry_safe(fc, tmp, &fec_client_list, node) { - fc->first_stream_chunk = -1; + list_for_each_entry_safe(fc, tmp, &fec_client_list, node) fc->state = FEC_STATE_NONE; - } mmd->stream_start.tv_sec = 0; mmd->stream_start.tv_usec = 0; } @@ -731,8 +902,8 @@ static void vss_pre_select(struct sched *s, struct task *t) senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds); } tv = vss_compute_timeout(vsst); - if (tv && tv_diff(tv, &s->timeout, &diff) < 0) - s->timeout = *tv; + if (tv) + sched_request_timeout(tv, s); } static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data) @@ -825,42 +996,6 @@ err: mmd->new_vss_status_flags = VSS_NEXT; } -static int initialize_fec_client(struct fec_client *fc) -{ - int ret; - struct fec_client_parms *fcp = fc->fcp; - - if (fcp->init_fec) { - /* - * Set the maximum slice size to the Maximum Packet Size if the - * transport protocol allows to determine this value. The user - * can specify a slice size up to this value. - */ - ret = fcp->init_fec(fc->sc); - if (ret < 0) - return ret; - if (!fcp->max_slice_bytes || fcp->max_slice_bytes > ret) - fcp->max_slice_bytes = ret; - } - if (fcp->max_slice_bytes < FEC_HEADER_SIZE + fcp->data_slices_per_group) - return -ERRNO_TO_PARA_ERROR(EINVAL); - ret = fec_new(fcp->data_slices_per_group, fcp->slices_per_group, - &fc->parms); - if (ret < 0) - goto err; - fc->first_stream_chunk = -1; /* stream not yet started */ - fc->src_data = para_malloc(fc->fcp->slices_per_group * sizeof(char *)); - fc->enc_buf = para_calloc(fc->fcp->max_slice_bytes); - fc->num_extra_slices = 0; - fc->extra_src_buf = para_calloc(fc->fcp->max_slice_bytes); - fc->next_header_time.tv_sec = 0; - fc->state = FEC_STATE_READY_TO_RUN; - return 1; -err: - fec_free(fc->parms); - return ret; -} - /** * Main sending function. * @@ -872,7 +1007,7 @@ err: */ static void vss_send(struct vss_task *vsst) { - int ret, i, fec_active = 0; + int i, fec_active = 0; struct timeval due; struct fec_client *fc, *tmp_fc; @@ -884,19 +1019,8 @@ static void vss_send(struct vss_task *vsst) &due, 1) < 0) return; list_for_each_entry_safe(fc, tmp_fc, &fec_client_list, node) { - switch (fc->state) { - case FEC_STATE_DISABLED: + if (fc->state == FEC_STATE_DISABLED) continue; - case FEC_STATE_NONE: - ret = initialize_fec_client(fc); - if (ret < 0) { - PARA_ERROR_LOG("%s\n", para_strerror(-ret)); - continue; - } - /* fall through */ - case FEC_STATE_READY_TO_RUN: - break; - } if (!next_slice_is_due(fc, NULL)) { fec_active = 1; continue; @@ -904,9 +1028,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->fcp->max_slice_bytes); + fc->current_slice_num, fc->group.slice_bytes); fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf, - fc->fcp->max_slice_bytes); + fc->group.slice_bytes + FEC_HEADER_SIZE); fc->current_slice_num++; fec_active = 1; }