X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=vss.c;h=aec357ab7e5bdd936e438412a3c94216c72bbff2;hp=78f8bab8a1edf53bbc9777e85d4e4f10fcce4023;hb=ac4e01ba1a2f302bf31dd511a7aff8da78660b15;hpb=532e1d01498090d8f17f39e4495790715d826457 diff --git a/vss.c b/vss.c index 78f8bab8..55eaf2eb 100644 --- a/vss.c +++ b/vss.c @@ -1,8 +1,4 @@ -/* - * Copyright (C) 1997-2011 Andre Noll - * - * Licensed under the GPL v2. For licencing details see COPYING. - */ +/* Copyright (C) 1997 Andre Noll , see file COPYING. */ /** \file vss.c The virtual streaming system. * @@ -11,10 +7,17 @@ * senders. */ +#include +#include #include -#include #include +#include +#include +#include +#include +#include +#include "server.lsg.h" #include "para.h" #include "error.h" #include "portable_io.h" @@ -24,13 +27,12 @@ #include "afs.h" #include "server.h" #include "net.h" -#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; @@ -38,24 +40,9 @@ extern void dccp_send_init(struct sender *); extern void http_send_init(struct sender *); extern void udp_send_init(struct sender *); -/** The list of supported senders. */ -struct sender senders[] = { - { - .name = "http", - .init = http_send_init, - }, - { - .name = "dccp", - .init = dccp_send_init, - }, - { - .name = "udp", - .init = udp_send_init, - }, - { - .name = NULL, - } -}; +extern const struct sender udp_sender, dccp_sender, http_sender; +const struct sender * const senders[] = { + &http_sender, &dccp_sender, &udp_sender, NULL}; /** The possible states of the afs socket. */ enum afs_socket_status { @@ -83,14 +70,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. */ - const char *header_buf; + 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; }; /** @@ -133,6 +124,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 +159,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 +205,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 +321,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,10 +330,50 @@ static int initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) return 1; } -static void compute_group_size(struct vss_task *vsst, struct fec_group *g, +static int 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 + * 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) { + assert(vsst->header_buf); + *buf = vsst->header_buf; /* stripped header */ + *sz = vsst->header_len; + return 0; + } + 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) { + *buf = NULL; + *sz = 0; + } + return ret; +} + +static int 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())); + char *buf; + size_t len; + int ret, i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time())); + + if (g->first_chunk == 0) { + g->num_chunks = 1; + ret = vss_get_chunk(0, vsst, &buf, &len); + if (ret < 0) + return ret; + g->bytes = len; + return 0; + } g->num_chunks = 0; g->bytes = 0; @@ -348,15 +384,15 @@ 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; - 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); + ret = vss_get_chunk(chunk_num, vsst, &buf, &len); + if (ret < 0) + return ret; if (g->bytes + len > max_bytes) break; /* Include this chunk */ @@ -364,6 +400,7 @@ static void compute_group_size(struct vss_task *vsst, struct fec_group *g, g->num_chunks++; } assert(g->num_chunks); + return 1; } /* @@ -425,7 +462,9 @@ static int compute_slice_size(struct fec_client *fc, struct vss_task *vsst) 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); + ret = compute_group_size(vsst, g, max_group_bytes); + if (ret < 0) + return ret; g->slice_bytes = DIV_ROUND_UP(g->bytes, k); if (g->slice_bytes == 0) g->slice_bytes = 1; @@ -441,7 +480,9 @@ static int compute_slice_size(struct fec_client *fc, struct vss_task *vsst) 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); + ret = compute_group_size(vsst, g, max_group_bytes); + if (ret < 0) + return ret; d = g->bytes; if (d == 0) { g->slice_bytes = DIV_ROUND_UP(h, k); @@ -478,7 +519,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, *p; struct fec_group *g = &fc->group; if (fc->state == FEC_STATE_NONE) { @@ -515,25 +556,51 @@ 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); + /* + * 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. */ } - /* setup data slices */ - 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) { + /* + * Setup data slices. Note that for ogg streams chunk 0 points to a + * buffer on the heap rather than to the mapped audio file. + */ + ret = vss_get_chunk(g->first_chunk, vsst, &buf, &len); + if (ret < 0) + return ret; + 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. This slice will not - * be fully used. + * 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); @@ -541,8 +608,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 */ @@ -550,7 +617,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 ); @@ -626,6 +693,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 +718,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 +808,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,7 +850,9 @@ 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); - para_munmap(vsst->map, mmd->size); + afh_free_header(vsst->header_buf, mmd->afd.audio_format_id); + vsst->header_buf = NULL; + para_munmap(vsst->map, vsst->mapsize); vsst->map = NULL; mmd->chunks_sent = 0; //mmd->offset = 0; @@ -815,8 +861,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++; } @@ -845,69 +892,28 @@ 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 timeval *tv; - 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); vsst->afsss = AFS_SOCKET_CHECK_FOR_WRITE; } else para_fd_set(vsst->afs_socket, &s->rfds, &s->max_fileno); - for (i = 0; senders[i].name; i++) { - if (!senders[i].pre_select) + FOR_EACH_SENDER(i) { + if (!senders[i]->pre_select) continue; - senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds); + 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) { - 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; @@ -939,6 +945,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; @@ -971,12 +981,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, passed_fd, - 0, &vsst->map); + ret = para_mmap(statbuf.st_size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, + passed_fd, &vsst->map); if (ret < 0) goto err; + vsst->mapsize = statbuf.st_size; close(passed_fd); mmd->chunks_sent = 0; mmd->current_chunk = 0; @@ -984,8 +993,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, vsst->mapsize, &vsst->header_buf, &vsst->header_len); return; err: free(mmd->afd.afhi.chunk_table); @@ -1006,32 +1015,34 @@ err: */ static void vss_send(struct vss_task *vsst) { - int i, fec_active = 0; + int i, ret; + bool fec_active = false; struct timeval due; struct fec_client *fc, *tmp_fc; + char *buf; + size_t len; if (!vsst->map || !vss_playing()) return; if (chk_barrier("eof", &vsst->eof_barrier, &due, 1) < 0) return; - if (chk_barrier("data send", &vsst->data_send_barrier, - &due, 1) < 0) + if (chk_barrier("data send", &vsst->data_send_barrier, &due, 1) < 0) return; list_for_each_entry_safe(fc, tmp_fc, &fec_client_list, node) { if (fc->state == FEC_STATE_DISABLED) continue; if (!next_slice_is_due(fc, NULL)) { - fec_active = 1; + fec_active = true; continue; } 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->current_slice_num++; fc->fcp->send_fec(fc->sc, (char *)fc->enc_buf, fc->group.slice_bytes + FEC_HEADER_SIZE); - fc->current_slice_num++; - fec_active = 1; + fec_active = true; } if (mmd->current_chunk >= mmd->afd.afhi.chunks_total) { /* eof */ if (!fec_active) @@ -1040,45 +1051,74 @@ 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; - size_t len; - - if (!mmd->chunks_sent) { - mmd->stream_start = *now; - mmd->events++; - set_mmd_offset(); - } + if (tv_diff(&due, now, NULL) > 0) + return; + if (!mmd->chunks_sent) { + mmd->stream_start = *now; + mmd->events++; + set_mmd_offset(); + } + ret = vss_get_chunk(mmd->current_chunk, vsst, &buf, &len); + if (ret < 0) { + PARA_ERROR_LOG("could not get chunk %lu: %s\n", + mmd->current_chunk, para_strerror(-ret)); + } else { /* - * We call the send function also in case of empty chunks as - * they might have still some data queued which can be sent in - * this case. + * We call ->send() even if len is zero because senders might + * have data queued which can be sent now. */ - afh_get_chunk(mmd->current_chunk, &mmd->afd.afhi, vsst->map, - &buf, &len); - for (i = 0; senders[i].name; i++) { - if (!senders[i].send) + FOR_EACH_SENDER(i) { + if (!senders[i]->send) continue; - senders[i].send(mmd->current_chunk, mmd->chunks_sent, + senders[i]->send(mmd->current_chunk, mmd->chunks_sent, buf, len, vsst->header_buf, vsst->header_len); } - mmd->chunks_sent++; - mmd->current_chunk++; } + mmd->chunks_sent++; + mmd->current_chunk++; } -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; + ret = task_get_notification(vsst->task); + if (ret < 0) + return ret; + if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { + /* shut down senders and fec clients */ + struct fec_client *fc, *tmp; + FOR_EACH_SENDER(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; - if (senders[sender_num].client_cmds[num]) { - ret = senders[sender_num].client_cmds[num] + 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)); @@ -1089,63 +1129,79 @@ 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 vsst->afsss = AFS_SOCKET_AFD_PENDING; } - for (i = 0; senders[i].name; i++) { - if (!senders[i].post_select) + FOR_EACH_SENDER(i) { + if (!senders[i]->post_select) continue; - senders[i].post_select(&s->rfds, &s->wfds); + senders[i]->post_select(&s->rfds, &s->wfds); } if ((vss_playing() && !(mmd->vss_status_flags & VSS_PLAYING)) || (vss_next() && vss_playing())) tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); vss_send(vsst); + return 0; } /** * 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 vss_init(int afs_socket, struct sched *s) { static struct vss_task vss_task_struct, *vsst = &vss_task_struct; int i; - char *hn = para_hostname(), *home = para_homedir(); - long unsigned announce_time = conf.announce_time_arg > 0? - conf.announce_time_arg : 300, - autoplay_delay = conf.autoplay_delay_arg > 0? - conf.autoplay_delay_arg : 0; + long unsigned announce_time = OPT_UINT32_VAL(ANNOUNCE_TIME), + autoplay_delay = OPT_UINT32_VAL(AUTOPLAY_DELAY); 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); - for (i = 0; senders[i].name; i++) { - PARA_NOTICE_LOG("initializing %s sender\n", senders[i].name); - senders[i].init(&senders[i]); + FOR_EACH_SENDER(i) { + PARA_NOTICE_LOG("initializing %s sender\n", senders[i]->name); + senders[i]->init(); } - free(hn); - free(home); mmd->sender_cmd_data.cmd_num = -1; - if (conf.autoplay_given) { + if (OPT_GIVEN(AUTOPLAY)) { struct timeval tmp; 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(&vsst->task); + vsst->task = task_register(&(struct task_info) { + .name = "vss", + .pre_select = vss_pre_select, + .post_select = vss_post_select, + .context = vsst, + }, s); +} + +/** + * Turn off the virtual streaming system. + * + * This is only executed on exit. It calls the ->shutdowwn method of all senders. + */ +void vss_shutdown(void) +{ + int i; + + FOR_EACH_SENDER(i) { + if (!senders[i]->shutdown) + continue; + PARA_NOTICE_LOG("shutting down %s sender\n", senders[i]->name); + senders[i]->shutdown(); + } }