X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=e26e8c8303c2c228a59adfaf64e6b7c5fd7b0fdf;hp=afde0dcb829b42422310eaf33361b990bb98c2ac;hb=d5538ff0dd9f6531a1a319b49c32bd72597fb2c3;hpb=28f8405e062fcff1f0ce90eb01ffeaca299cffa7 diff --git a/vss.c b/vss.c index afde0dcb..e26e8c83 100644 --- a/vss.c +++ b/vss.c @@ -131,14 +131,20 @@ struct fec_group { uint8_t num_header_slices; }; +enum fec_client_state { + FEC_STATE_NONE = 0, /**< not initialized and not enabled */ + FEC_STATE_DISABLED, /**< temporarily disabled */ + FEC_STATE_READY_TO_RUN /**< initialized and enabled */ +}; + /** * Describes one connected FEC client. */ struct fec_client { - /** If negative, this client is temporarily disabled. */ - int error; - /** UDP or DCCP. */ - struct sender *sender; + /** Current state of the client */ + enum fec_client_state state; + /** The connected sender client (transport layer). */ + struct sender_client *sc; /** Parameters requested by the client. */ struct fec_client_parms *fcp; /** Used by the core FEC code. */ @@ -163,8 +169,6 @@ struct fec_client { int num_extra_slices; /** Contains the FEC-encoded data. */ unsigned char *enc_buf; - /** Pointer obtained from sender when the client is added. */ - void *private_data; }; /** @@ -373,7 +377,6 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst) { - assert(fc->error >= 0); if (fc->first_stream_chunk < 0 || fc->current_slice_num == fc->fcp->slices_per_group + fc->num_extra_slices) { int ret = setup_next_fec_group(fc, vsst); @@ -382,8 +385,8 @@ static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst) if (ret < 0) { PARA_ERROR_LOG("%s\n", para_strerror(-ret)); PARA_ERROR_LOG("FEC client temporarily disabled\n"); - fc->error = ret; - return fc->error; + fc->state = FEC_STATE_DISABLED; + return ret; } } write_fec_header(fc, vsst); @@ -412,21 +415,20 @@ size_t vss_get_fec_eof_packet(const char **buf) /** * Add one entry to the list of active fec clients. * - * \param fcp Describes the fec parameters to be used for this client. - * \param result An opaque pointer that must be used by remove the client later. + * \param sc Generic sender_client data of the transport layer. + * \param fcp FEC parameters as supplied by the transport layer. * - * \return Standard. + * \return Newly allocated fec_client struct. */ -int vss_add_fec_client(struct sender *sender, void *private_data, - struct fec_client **result) +struct fec_client *vss_add_fec_client(struct sender_client *sc, + struct fec_client_parms *fcp) { struct fec_client *fc = para_calloc(sizeof(*fc)); - fc->private_data = private_data; - fc->sender = sender; + fc->sc = sc; + fc->fcp = fcp; para_list_add(&fc->node, &fec_client_list); - *result = fc; - return 1; + return fc; } /** @@ -468,11 +470,10 @@ static void compute_slice_timeout(struct timeval *timeout) { struct fec_client *fc; - assert(vss_playing()); list_for_each_entry(fc, &fec_client_list, node) { struct timeval diff; - if (fc->error < 0) + if (fc->state != FEC_STATE_READY_TO_RUN) continue; if (next_slice_is_due(fc, &diff)) { timeout->tv_sec = 0; @@ -495,7 +496,7 @@ static void set_eof_barrier(struct vss_task *vsst) list_for_each_entry(fc, &fec_client_list, node) { struct timeval group_duration; - if (fc->error < 0) + if (fc->state != FEC_STATE_READY_TO_RUN) continue; tv_scale(fc->group.num_chunks, chunk_tv, &group_duration); if (tv_diff(&timeout, &group_duration, NULL) < 0) @@ -635,18 +636,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; @@ -699,8 +688,7 @@ static void vss_pre_select(struct sched *s, struct task *t) senders[i].shutdown_clients(); list_for_each_entry_safe(fc, tmp, &fec_client_list, node) { fc->first_stream_chunk = -1; - fc->error = 0; - fc->fcp = NULL; + fc->state = FEC_STATE_NONE; } mmd->stream_start.tv_sec = 0; mmd->stream_start.tv_usec = 0; @@ -825,17 +813,23 @@ err: mmd->new_vss_status_flags = VSS_NEXT; } -static int open_fec_client(struct fec_client *fc) +static int initialize_fec_client(struct fec_client *fc) { int ret; - struct fec_client_parms *fcp; + struct fec_client_parms *fcp = fc->fcp; - ret = fc->sender->open(fc->private_data, &fc->fcp); - if (ret < 0) { - fc->fcp = NULL; - return ret; + 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; } - fcp = fc->fcp; 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, @@ -848,6 +842,7 @@ static int open_fec_client(struct fec_client *fc) 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); @@ -877,14 +872,18 @@ static void vss_send(struct vss_task *vsst) &due, 1) < 0) return; list_for_each_entry_safe(fc, tmp_fc, &fec_client_list, node) { - if (fc->error < 0) + switch (fc->state) { + case FEC_STATE_DISABLED: continue; - if (!fc->fcp) { - ret = open_fec_client(fc); + 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; @@ -894,9 +893,8 @@ static void vss_send(struct vss_task *vsst) continue; PARA_DEBUG_LOG("sending %d:%d (%u bytes)\n", fc->group.num, fc->current_slice_num, fc->fcp->max_slice_bytes); - fc->sender->send_fec((char *)fc->enc_buf, - fc->fcp->max_slice_bytes, - fc->private_data); + fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf, + fc->fcp->max_slice_bytes); fc->current_slice_num++; fec_active = 1; } @@ -944,8 +942,12 @@ static void vss_post_select(struct sched *s, struct task *t) int num = mmd->sender_cmd_data.cmd_num, sender_num = mmd->sender_cmd_data.sender_num; - if (senders[sender_num].client_cmds[num]) - senders[sender_num].client_cmds[num](&mmd->sender_cmd_data); + if (senders[sender_num].client_cmds[num]) { + ret = senders[sender_num].client_cmds[num] + (&mmd->sender_cmd_data); + if (ret < 0) + PARA_ERROR_LOG("%s\n", para_strerror(-ret)); + } mmd->sender_cmd_data.cmd_num = -1; } if (vsst->afsss != AFS_SOCKET_CHECK_FOR_WRITE)