X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=24dfc6bb51af4edf67c645c0d3ccbda2ad8f1868;hp=cfa64a143cb1f5d665990dc6aeb23ef94d064d4c;hb=b6b571e6c6fb52207b11fc7833b272ec8cfa28bf;hpb=9c5fbc5dd8b53604d7e73fb6714ee2b5e3458866 diff --git a/vss.c b/vss.c index cfa64a14..24dfc6bb 100644 --- a/vss.c +++ b/vss.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 1997-2013 Andre Noll + * Copyright (C) 1997 Andre Noll * * Licensed under the GPL v2. For licencing details see COPYING. */ @@ -11,8 +11,14 @@ * senders. */ +#include +#include #include #include +#include +#include +#include +#include #include "para.h" #include "error.h" @@ -82,14 +88,18 @@ struct vss_task { enum afs_socket_status afsss; /** The memory mapped audio file. */ char *map; + /** The size of the memory mapping. */ + size_t mapsize; /** Used by the scheduler. */ - struct task task; + struct task *task; /** Pointer to the header of the mapped audio file. */ char *header_buf; /** Length of the audio file header. */ size_t header_len; /** Time between audio file headers are sent. */ struct timeval header_interval; + /* Only used if afh supports dynamic chunks. */ + void *afh_context; }; /** @@ -341,6 +351,8 @@ static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) static void vss_get_chunk(int chunk_num, struct vss_task *vsst, char **buf, size_t *sz) { + int ret; + /* * 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 @@ -354,8 +366,15 @@ static void vss_get_chunk(int chunk_num, struct vss_task *vsst, *sz = vsst->header_len; return; } - afh_get_chunk(chunk_num, &mmd->afd.afhi, vsst->map, (const char **)buf, - sz); + ret = afh_get_chunk(chunk_num, &mmd->afd.afhi, + mmd->afd.audio_format_id, vsst->map, vsst->mapsize, + (const char **)buf, sz, &vsst->afh_context); + if (ret < 0) { + PARA_WARNING_LOG("could not get chunk %d: %s\n", + chunk_num, para_strerror(-ret)); + *buf = NULL; + *sz = 0; + } } static void compute_group_size(struct vss_task *vsst, struct fec_group *g, @@ -565,8 +584,14 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) 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); + /* + * There might be more than one header slice to fill although + * only the first one will be used. Set all header slices to + * our extra buffer. + */ + while (i < g->num_header_slices) + fc->src_data[i++] = fc->extra_header_buf; + break; /* we don't want i to be increased. */ } /* @@ -599,7 +624,7 @@ static int setup_next_fec_group(struct fec_client *fc, struct vss_task *vsst) for (; i < k; i++) fc->src_data[i] = (const unsigned char *)buf; } - PARA_DEBUG_LOG("FEC group %d: %d chunks (%d - %d), %d bytes\n", + PARA_DEBUG_LOG("FEC group %u: %u chunks (%u - %u), %u bytes\n", g->num, g->num_chunks, g->first_chunk, g->first_chunk + g->num_chunks - 1, g->bytes ); @@ -834,7 +859,7 @@ static void vss_eof(struct vss_task *vsst) 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); + para_munmap(vsst->map, vsst->mapsize); vsst->map = NULL; mmd->chunks_sent = 0; //mmd->offset = 0; @@ -843,8 +868,9 @@ static void vss_eof(struct vss_task *vsst) mmd->afd.afhi.chunk_tv.tv_usec = 0; free(mmd->afd.afhi.chunk_table); mmd->afd.afhi.chunk_table = NULL; - mmd->mtime = 0; - mmd->size = 0; + vsst->mapsize = 0; + afh_close(vsst->afh_context, mmd->afd.audio_format_id); + vsst->afh_context = NULL; mmd->events++; } @@ -873,49 +899,11 @@ static void set_mmd_offset(void) mmd->offset = tv2ms(&offset); } -/** - * Compute the timeout for the main select-loop of the scheduler. - * - * \param s Pointer to the server scheduler. - * \param t Pointer to the vss task structure. - * - * Before the timeout is computed, the current vss status flags are evaluated - * and acted upon by calling appropriate functions from the lower layers. - * Possible actions include - * - * - request a new audio file from afs, - * - shutdown of all senders (stop/pause command), - * - reposition the stream (ff/jmp command). - */ -static void vss_pre_select(struct sched *s, struct task *t) +static void vss_pre_select(struct sched *s, void *context) { int i; - struct vss_task *vsst = container_of(t, struct vss_task, task); + struct vss_task *vsst = context; - if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { - struct fec_client *fc, *tmp; - 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->state = FEC_STATE_NONE; - mmd->stream_start.tv_sec = 0; - mmd->stream_start.tv_usec = 0; - } - if (vss_next()) - vss_eof(vsst); - else if (vss_paused()) { - if (mmd->chunks_sent) - set_eof_barrier(vsst); - mmd->chunks_sent = 0; - } else if (vss_repos()) { - tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); - set_eof_barrier(vsst); - mmd->chunks_sent = 0; - mmd->current_chunk = mmd->repos_request; - mmd->new_vss_status_flags &= ~VSS_REPOS; - set_mmd_offset(); - } if (need_to_request_new_audio_file(vsst)) { PARA_DEBUG_LOG("ready and playing, but no audio file\n"); para_fd_set(vsst->afs_socket, &s->wfds, &s->max_fileno); @@ -932,7 +920,7 @@ static void vss_pre_select(struct sched *s, struct task *t) static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data) { - char control[255], buf[8]; + char control[255] __a_aligned(8), buf[8]; struct msghdr msg = {.msg_iov = NULL}; struct cmsghdr *cmsg; struct iovec iov; @@ -1000,12 +988,11 @@ static void recv_afs_result(struct vss_task *vsst, fd_set *rfds) ret = -ERRNO_TO_PARA_ERROR(errno); goto err; } - mmd->size = statbuf.st_size; - mmd->mtime = statbuf.st_mtime; - ret = para_mmap(mmd->size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, + ret = para_mmap(statbuf.st_size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, passed_fd, 0, &vsst->map); if (ret < 0) goto err; + vsst->mapsize = statbuf.st_size; close(passed_fd); mmd->chunks_sent = 0; mmd->current_chunk = 0; @@ -1014,7 +1001,7 @@ static void recv_afs_result(struct vss_task *vsst, fd_set *rfds) mmd->num_played++; mmd->new_vss_status_flags &= (~VSS_NEXT); afh_get_header(&mmd->afd.afhi, mmd->afd.audio_format_id, - vsst->map, mmd->size, &vsst->header_buf, &vsst->header_len); + vsst->map, vsst->mapsize, &vsst->header_buf, &vsst->header_len); return; err: free(mmd->afd.afhi.chunk_table); @@ -1055,7 +1042,7 @@ 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, + PARA_DEBUG_LOG("sending %u:%u (%u bytes)\n", fc->group.num, 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); @@ -1104,7 +1091,7 @@ static void vss_send(struct vss_task *vsst) */ 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++) { + for (i = 0; i < 5 && buf < vsst->map + vsst->mapsize; i++) { __a_unused volatile char x = *buf; buf += 4096; } @@ -1114,12 +1101,38 @@ static void vss_send(struct vss_task *vsst) } } -static void vss_post_select(struct sched *s, struct task *t) +static int vss_post_select(struct sched *s, void *context) { int ret, i; - struct vss_task *vsst = container_of(t, struct vss_task, task); - + struct vss_task *vsst = context; + if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { + /* shut down senders and fec clients */ + struct fec_client *fc, *tmp; + 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->state = FEC_STATE_NONE; + mmd->stream_start.tv_sec = 0; + mmd->stream_start.tv_usec = 0; + } + if (vss_next()) + vss_eof(vsst); + else if (vss_paused()) { + if (mmd->chunks_sent) + set_eof_barrier(vsst); + mmd->chunks_sent = 0; + } else if (vss_repos()) { /* repositioning due to ff/jmp command */ + tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); + set_eof_barrier(vsst); + mmd->chunks_sent = 0; + mmd->current_chunk = afh_get_start_chunk(mmd->repos_request, + &mmd->afd.afhi, mmd->afd.audio_format_id); + mmd->new_vss_status_flags &= ~VSS_REPOS; + set_mmd_offset(); + } + /* If a sender command is pending, run it. */ if (mmd->sender_cmd_data.cmd_num >= 0) { int num = mmd->sender_cmd_data.cmd_num, sender_num = mmd->sender_cmd_data.sender_num; @@ -1151,6 +1164,7 @@ static void vss_post_select(struct sched *s, struct task *t) (vss_next() && vss_playing())) tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); vss_send(vsst); + return 0; } /** @@ -1173,8 +1187,6 @@ void init_vss_task(int afs_socket, struct sched *s) conf.autoplay_delay_arg : 0; vsst->header_interval.tv_sec = 5; /* should this be configurable? */ vsst->afs_socket = afs_socket; - vsst->task.pre_select = vss_pre_select; - vsst->task.post_select = vss_post_select; ms2tv(announce_time, &vsst->announce_tv); PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst->announce_tv)); INIT_LIST_HEAD(&fec_client_list); @@ -1190,10 +1202,14 @@ void init_vss_task(int afs_socket, struct sched *s) mmd->vss_status_flags |= VSS_PLAYING; mmd->new_vss_status_flags |= VSS_PLAYING; ms2tv(autoplay_delay, &tmp); - tv_add(now, &tmp, &vsst->autoplay_barrier); + tv_add(clock_get_realtime(NULL), &tmp, &vsst->autoplay_barrier); tv_add(&vsst->autoplay_barrier, &vsst->announce_tv, &vsst->data_send_barrier); } - sprintf(vsst->task.status, "vss task"); - register_task(s, &vsst->task); + vsst->task = task_register(&(struct task_info) { + .name = "vss task", + .pre_select = vss_pre_select, + .post_select = vss_post_select, + .context = vsst, + }, s); }