-/*
- * Copyright (C) 1997-2013 Andre Noll <maan@systemlinux.org>
- *
- * Licensed under the GPL v2. For licencing details see COPYING.
- */
+/* Copyright (C) 1997 Andre Noll <maan@tuebingen.mpg.de>, see file COPYING. */
/** \file vss.c The virtual streaming system.
*
#include <arpa/inet.h>
#include <sys/un.h>
#include <netdb.h>
+#include <lopsub.h>
+#include "server.lsg.h"
#include "para.h"
#include "error.h"
#include "portable_io.h"
#include "string.h"
#include "afh.h"
#include "afs.h"
-#include "server.h"
#include "net.h"
-#include "server.cmdline.h"
+#include "server.h"
#include "list.h"
#include "send.h"
#include "sched.h"
#include "fd.h"
extern struct misc_meta_data *mmd;
-
-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 {
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;
};
/**
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
+ * transport protocol allows determination of this value. The user
* can specify a slice size up to this value.
*/
ret = fcp->init_fec(fc->sc);
return 1;
}
-static void vss_get_chunk(int chunk_num, struct vss_task *vsst,
+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
* 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;
+ return 0;
}
- 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) {
+ *buf = NULL;
+ *sz = 0;
+ }
+ return ret;
}
-static void compute_group_size(struct vss_task *vsst, struct fec_group *g,
+static int compute_group_size(struct vss_task *vsst, struct fec_group *g,
int max_bytes)
{
char *buf;
size_t len;
- int i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time()));
+ int ret, i, max_chunks = PARA_MAX(1LU, 150 / tv2ms(vss_chunk_time()));
if (g->first_chunk == 0) {
g->num_chunks = 1;
- vss_get_chunk(0, vsst, &buf, &len);
+ ret = vss_get_chunk(0, vsst, &buf, &len);
+ if (ret < 0)
+ return ret;
g->bytes = len;
- return;
+ return 0;
}
g->num_chunks = 0;
break;
if (chunk_num >= mmd->afd.afhi.chunks_total) /* eof */
break;
- vss_get_chunk(chunk_num, vsst, &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 */
g->num_chunks++;
}
assert(g->num_chunks);
+ return 1;
}
/*
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;
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);
* Setup data slices. Note that for ogg streams chunk 0 points to a
* buffer on the heap rather than to the mapped audio file.
*/
- vss_get_chunk(g->first_chunk, vsst, &buf, &len);
+ 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) {
/*
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
);
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;
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++;
}
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);
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);
}
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;
}
#ifndef MAP_POPULATE
+/** As of 2018, neither FreeBSD-11.2 nor NetBSD-8.0 have MAP_POPULATE. */
#define MAP_POPULATE 0
#endif
if (ret < 0)
goto err;
vsst->afsss = AFS_SOCKET_READY;
- PARA_DEBUG_LOG("fd: %d, code: %u, shmid: %u\n", passed_fd, afs_code,
- afs_data);
ret = -E_NOFD;
- if (afs_code != NEXT_AUDIO_FILE)
+ if (afs_code != NEXT_AUDIO_FILE) {
+ PARA_ERROR_LOG("afs code: %u, expected: %d\n", afs_code,
+ NEXT_AUDIO_FILE);
goto err;
+ }
if (passed_fd < 0)
goto err;
shmid = afs_data;
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,
- 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;
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);
*/
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)
}
compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv,
&mmd->stream_start, &due);
- if (tv_diff(&due, now, NULL) <= 0) {
- 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.
*/
- vss_get_chunk(mmd->current_chunk, vsst, &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);
}
- /*
- * 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.
- */
- 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++) {
- __a_unused volatile char x = *buf;
- buf += 4096;
- }
- }
- mmd->chunks_sent++;
- mmd->current_chunk++;
}
+ mmd->chunks_sent++;
+ mmd->current_chunk++;
}
-static int 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) {
+ afh_free_header(vsst->header_buf, mmd->afd.audio_format_id);
+ 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));
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()))
* This also initializes all supported senders and starts streaming
* if the --autoplay command line flag was given.
*/
-void init_vss_task(int afs_socket, struct sched *s)
+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(s, &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();
+ }
}