/* SPDX-License-Identifier: GPL-2.0 */ /** \file vss.c The virtual streaming system. * * This contains the audio streaming code of para_server which is independent * of the current audio format, audio file selector and of the activated * senders. */ #include #include #include #include #include #include #include #include #include "server.lsg.h" #include "para.h" #include "error.h" #include "portable_io.h" #include "fec.h" #include "string.h" #include "afh.h" #include "afs.h" #include "net.h" #include "list.h" #include "server.h" #include "sched.h" #include "send.h" #include "vss.h" #include "ipc.h" #include "fd.h" extern struct misc_meta_data *mmd; extern const struct sender udp_sender, http_sender; const struct sender * const senders[] = { &http_sender, &udp_sender, NULL}; /* The possible states of the afs socket. */ enum afs_socket_status { /* Socket is inactive. */ AFS_SOCKET_READY, /* Socket fd was monitored for writing. */ AFS_SOCKET_CHECK_FOR_WRITE, /* vss wrote a request to the socket and waits for reply from afs. */ AFS_SOCKET_AFD_PENDING }; /* The task structure for the virtual streaming system. */ struct vss_task { /* End of the announcing interval. */ struct timeval data_send_barrier; /* End of the EOF interval. */ struct timeval eof_barrier; /* Only used if --autoplay_delay was given. */ struct timeval autoplay_barrier; /* Used for afs-server communication. */ int afs_socket; /* The current state of afs_socket. */ enum afs_socket_status afsss; /* The memory mapped audio file. */ void *map; /* The size of the memory mapping. */ size_t mapsize; /* Used by the scheduler. */ 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; }; /* * The list of currently connected fec clients. * * Senders may use \ref vss_add_fec_client() to add entries to the list. */ static struct list_head fec_client_list; /* * Data associated with one FEC group. * * A FEC group consists of a fixed number of slices and this number is given * by the slices_per_group parameter of struct \ref fec_client_parms. Each * FEC group contains a number of chunks of the current audio file. * * FEC slices directly correspond to the data packages sent by the paraslash * senders that use FEC. Each slice is identified by its group number and its * number within the group. All slices have the same size, but the last slice * of the group may not be filled entirely. */ struct fec_group { /* The number of the FEC group. */ uint32_t num; /* Number of bytes in this group. */ uint32_t bytes; /* The first chunk of the current audio file belonging to the group. */ uint32_t first_chunk; /* The number of chunks contained in this group. */ uint32_t num_chunks; /* When the first chunk was sent. */ struct timeval start; /* The duration of the full group. */ struct timeval duration; /* The group duration divided by the number of slices. */ 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; }; /* 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 */ FEC_STATE_READY_TO_RUN /**< initialized and enabled */ }; struct fec_client { /* 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. */ struct fec_parms *parms; /* The position of this client in the fec client list. */ struct list_head node; /* When the first slice for this client was sent. */ struct timeval stream_start; /* The first chunk sent to this FEC client. */ int first_stream_chunk; /* Describes the current group. */ struct fec_group group; /* The current slice. */ uint8_t current_slice_num; /* The data to be FEC-encoded. */ unsigned char **src_data; /* Last time an audio header was sent. */ struct timeval next_header_time; /* Extra slices needed to store largest chunk + header. */ int num_extra_slices; /* Contains the FEC-encoded data. */ unsigned char *enc_buf; /* Maximal packet size. */ int mps; }; /* Write a fec header to the encoding buffer of a fec client. */ static void write_fec_header(struct fec_client *fc, struct vss_task *vsst) { char *buf = (char *)fc->enc_buf; struct fec_group *g = &fc->group; struct fec_client_parms *p = fc->fcp; write_u32(buf, FEC_MAGIC); write_u8(buf + 4, p->slices_per_group + fc->num_extra_slices); write_u8(buf + 5, p->data_slices_per_group + fc->num_extra_slices); write_u32(buf + 6, g->num_header_slices? vsst->header_len : 0); write_u32(buf + 10, g->num); 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, 8); } 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 false; } if (!vsst->header_buf) 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 true; } static bool need_data_slices(struct fec_client *fc, struct vss_task *vsst) { 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; } static int fc_num_data_slices(const struct fec_client *fc) { return fc->fcp->data_slices_per_group + fc->num_extra_slices; } static int fc_num_slices(const struct fec_client *fc) { return fc->fcp->slices_per_group + fc->num_extra_slices; } 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; return ret; } /* set group start and group duration */ static void set_group_timing(struct fec_client *fc, struct vss_task *vsst) { struct fec_group *g = &fc->group; struct timeval *chunk_tv = &mmd->afd.afhi.chunk_tv; 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 initialize_fec_client(struct fec_client *fc, struct vss_task *vsst) { int i, k, n, ret; int hs, ds, rs; /* header/data/redundant slices */ struct fec_client_parms *fcp = fc->fcp; /* * Set the maximum slice size to the Maximum Packet Size if the * transport protocol allows determination of this value. The user * can specify a slice size up to this value. */ ret = fcp->init_fec(fc->sc); if (ret < 0) return ret; fc->mps = ret; if (fc->mps <= FEC_HEADER_SIZE) return -ERRNO_TO_PARA_ERROR(EINVAL); /* free previous buffers, if any */ if (fc->src_data) { k = fc_num_data_slices(fc); for (i = 0; i < k; i++) free(fc->src_data[i]); free(fc->src_data); fc->src_data = NULL; } free(fc->enc_buf); 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.afhi.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 = fc_num_slices(fc); PARA_INFO_LOG("mps: %d, k: %d, n: %d, extra slices: %d\n", fc->mps, k, n, fc->num_extra_slices); fec_free(fc->parms); ret = fec_new(k, n, &fc->parms); if (ret < 0) return ret; fc->src_data = arr_alloc(k, sizeof(char *)); for (i = 0; i < k; i++) fc->src_data[i] = alloc(fc->mps); fc->enc_buf = alloc(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 int vss_get_chunk(int chunk_num, struct vss_task *vsst, char **buf, uint32_t *len) { 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 */ *len = 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, len, &vsst->afh_context); if (ret < 0) { *buf = NULL; *len = 0; } return ret; } static int compute_group_size(struct vss_task *vsst, struct fec_group *g, int max_bytes) { char *buf; uint32_t len; int ret, i, max_chunks; 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; /* * 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. */ max_chunks = PARA_MAX(1LU, 150 / tv2ms(&mmd->afd.afhi.chunk_tv)); for (i = 0;; i++) { 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; 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->bytes += len; g->num_chunks++; } if (g->num_chunks == 0) return -E_EOF; PARA_DEBUG_LOG("group #%u: %u chunks, %u bytes total\n", g->num, g->num_chunks, g->bytes); return 1; } /* * 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_num_data_slices(fc); int n = fc_num_slices(fc); 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; 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; 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; 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); ret = num_slices(vsst->header_len, g->slice_bytes, n - k); if (ret < 0) return ret; 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, c; size_t copy, src_copied, slice_copied; struct fec_group *g = &fc->group; if (fc->state == FEC_STATE_NONE) { ret = initialize_fec_client(fc, vsst); if (ret < 0) return ret; g->first_chunk = mmd->current_chunk; g->num = 0; g->start = *now; } else { struct timeval tmp; if (g->first_chunk + g->num_chunks >= mmd->afd.afhi.chunks_total) return 0; /* * Start and duration of this group depend only on the previous * group. Compute the new group start as g->start += g->duration. */ tmp = g->start; tv_add(&tmp, &g->duration, &g->start); set_group_timing(fc, vsst); g->first_chunk += g->num_chunks; g->num++; } ret = compute_slice_size(fc, vsst); if (ret < 0) return ret; assert(g->slice_bytes > 0); fc->current_slice_num = 0; if (g->num == 0) set_group_timing(fc, vsst); /* setup header slices */ for (i = 0, src_copied = 0; i < g->num_header_slices; i++) { copy = PARA_MIN((size_t)g->slice_bytes, vsst->header_len - src_copied); if (copy == 0) break; memcpy(fc->src_data[i], vsst->header_buf + src_copied, copy); if (copy < g->slice_bytes) memset(fc->src_data[i] + copy, 0, g->slice_bytes - copy); src_copied += copy; } /* * There might be more than one header slice to fill although only the * first one will be used. Zero out any remaining header slices. */ while (i < g->num_header_slices) memset(fc->src_data[i++], 0, g->slice_bytes); slice_copied = 0; for (c = g->first_chunk; c < g->first_chunk + g->num_chunks; c++) { char *buf; uint32_t src_len; ret = vss_get_chunk(c, vsst, &buf, &src_len); if (ret < 0) return ret; if (src_len == 0) continue; src_copied = 0; while (src_copied < src_len) { copy = PARA_MIN((size_t)g->slice_bytes - slice_copied, src_len - src_copied); memcpy(fc->src_data[i] + slice_copied, buf + src_copied, copy); src_copied += copy; slice_copied += copy; if (slice_copied == g->slice_bytes) { i++; slice_copied = 0; } } } if (i < fc_num_data_slices(fc) && slice_copied < g->slice_bytes) memset(fc->src_data[i] + slice_copied, 0, g->slice_bytes - slice_copied); /* zero out remaining slices, if any */ while (++i < fc_num_data_slices(fc)) memset(fc->src_data[i], 0, g->slice_bytes); 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 ); PARA_DEBUG_LOG("slice_bytes: %d, %d header slices, %d data slices\n", g->slice_bytes, g->num_header_slices, fc_num_data_slices(fc) ); return 1; } static int compute_next_fec_slice(struct fec_client *fc, struct vss_task *vsst) { 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) return 0; if (ret < 0) { PARA_ERROR_LOG("%s\n", para_strerror(-ret)); PARA_ERROR_LOG("FEC client temporarily disabled\n"); fc->state = FEC_STATE_DISABLED; return ret; } } write_fec_header(fc, vsst); fec_encode(fc->parms, (const unsigned char * const*)fc->src_data, fc->enc_buf + FEC_HEADER_SIZE, fc->current_slice_num, fc->group.slice_bytes); return 1; } /** * Add one entry to the list of active fec clients. * * \param sc Generic sender_client data of the transport layer. * \param fcp FEC parameters as supplied by the transport layer. * * \return Newly allocated fec_client struct. */ struct fec_client *vss_add_fec_client(struct sender_client *sc, struct fec_client_parms *fcp) { struct fec_client *fc = zalloc(sizeof(*fc)); fc->sc = sc; fc->fcp = fcp; para_list_add(&fc->node, &fec_client_list); return fc; } /** * Remove one entry from the list of active fec clients. * * \param fc The client to be removed. */ void vss_del_fec_client(struct fec_client *fc) { int i; list_del(&fc->node); free(fc->enc_buf); if (fc->src_data) { for (i = 0; i < fc_num_data_slices(fc); i++) free(fc->src_data[i]); free(fc->src_data); } fec_free(fc->parms); free(fc); } /* * Compute if/when next slice is due. If it isn't due yet and diff is not * Null, compute the time difference next - now, where * * next = stream_start + (first_group_chunk - first_stream_chunk) * * chunk_time + slice_num * slice_time */ static bool next_slice_is_due(struct fec_client *fc, struct timeval *diff) { struct timeval tmp, next; if (fc->state == FEC_STATE_NONE) return true; tv_scale(fc->current_slice_num, &fc->group.slice_duration, &tmp); tv_add(&tmp, &fc->group.start, &next); return tv_diff(&next, now, diff) < 0; } static void set_eof_barrier(struct vss_task *vsst) { struct fec_client *fc; struct timeval timeout = {1, 0}; if (!vsst->map) goto out; list_for_each_entry(fc, &fec_client_list, node) { struct timeval group_duration; if (fc->state != FEC_STATE_READY_TO_RUN) continue; tv_scale(fc->group.num_chunks, &mmd->afd.afhi.chunk_tv, &group_duration); if (tv_diff(&timeout, &group_duration, NULL) < 0) timeout = group_duration; } out: tv_add(now, &timeout, &vsst->eof_barrier); } /** * Check if the "P" (playing) vss status flag is set. * * \return True if playing, false otherwise. */ bool vss_playing(void) { return mmd->new_vss_status_flags & VSS_PLAYING; } /* Check whether the N (next) status flag is set. */ static bool vss_next(void) { return mmd->new_vss_status_flags & VSS_NEXT; } /* Check whether a reposition request is pending. */ static bool vss_repos(void) { return mmd->new_vss_status_flags & VSS_REPOS; } /** * Check if the virtual streaming system is currently paused. * * \return True if paused, false otherwise. */ bool vss_paused(void) { return !(mmd->new_vss_status_flags & VSS_NEXT) && !(mmd->new_vss_status_flags & VSS_PLAYING); } /** * Check if the virtual streaming system is currently stopped. * * \return True iff stopped. */ bool vss_stopped(void) { return (mmd->new_vss_status_flags & VSS_NEXT) && !(mmd->new_vss_status_flags & VSS_PLAYING); } static bool barrier_has_passed(const char *bname, const struct timeval *barrier) { long unsigned ms; struct timeval diff; if (tv_diff(now, barrier, &diff) > 0) return true; ms = tv2ms(&diff); if (ms > 0) PARA_INFO_LOG("%s barrier: %lums left\n", bname, ms); return false; } static void vss_eof(struct vss_task *vsst) { if (mmd->new_vss_status_flags & VSS_NOMORE) mmd->new_vss_status_flags = VSS_NEXT; 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->afd.afhi.seconds_total = 0; mmd->afd.afhi.chunks_total = 0; mmd->afd.afhi.chunk_tv.tv_sec = 0; mmd->afd.afhi.chunk_tv.tv_usec = 0; free(mmd->afd.afhi.chunk_table); mmd->afd.afhi.chunk_table = NULL; vsst->mapsize = 0; afh_close(vsst->afh_context, mmd->afd.audio_format_id); vsst->afh_context = NULL; mmd->events++; } static bool need_to_request_new_audio_file(struct vss_task *vsst) { if (vsst->map) /* have audio file */ return false; if (!vss_playing()) /* don't need one */ return false; if (mmd->new_vss_status_flags & VSS_NOMORE) return false; if (vsst->afsss != AFS_SOCKET_READY) /* already requested one */ return false; if (!barrier_has_passed("autoplay_delay", &vsst->autoplay_barrier)) return false; return true; } static void vss_pre_monitor(struct sched *s, void *context) { struct vss_task *vsst = context; int i; struct timeval tv; struct fec_client *fc; if (vsst->afsss == AFS_SOCKET_CHECK_FOR_WRITE) { sched_monitor_writefd(vsst->afs_socket, s); } else if (vsst->afsss == AFS_SOCKET_AFD_PENDING) sched_monitor_readfd(vsst->afs_socket, s); FOR_EACH_SENDER(i) { if (!senders[i]->pre_monitor) continue; senders[i]->pre_monitor(s); } if (!vss_playing() || !vsst->map) return; if (vss_next() && vsst->map) return sched_min_delay(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 I/O 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, &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); } } static int recv_afd(int afs_socket, int *fd, uint32_t *code, uint32_t *data) { char control[255] __a_aligned(8), buf[8]; struct iovec iov = {.iov_base = buf, .iov_len = sizeof(buf)}; struct msghdr msg = { .msg_iov = &iov, .msg_iovlen = 1, .msg_control = control, .msg_controllen = sizeof(control), }; struct cmsghdr *cmsg; *fd = -1; if (recvmsg(afs_socket, &msg, 0) < 0) return -ERRNO_TO_PARA_ERROR(errno); if (iov.iov_len != sizeof(buf)) return -E_AFS_SHORT_READ; *code = *(uint32_t*)buf; *data = *(uint32_t*)(buf + 4); for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) { if (cmsg->cmsg_level != SOL_SOCKET || cmsg->cmsg_type != SCM_RIGHTS) continue; if ((cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int) != 1) continue; *fd = *(int *)CMSG_DATA(cmsg); } return 1; } /* As of 2025, neither FreeBSD-13.5 nor NetBSD-10.0 have MAP_POPULATE. */ #ifndef MAP_POPULATE #define MAP_POPULATE 0 #endif static void recv_afs_result(struct vss_task *vsst, const struct sched *s) { int ret, passed_fd, shmid; uint32_t afs_code = 0, afs_data = 0; struct stat statbuf; if (!sched_read_ok(vsst->afs_socket, s)) return; ret = recv_afd(vsst->afs_socket, &passed_fd, &afs_code, &afs_data); if (ret == -ERRNO_TO_PARA_ERROR(EAGAIN)) return; vsst->afsss = AFS_SOCKET_READY; if (ret < 0) goto err; if (afs_code == NO_ADMISSIBLE_FILES) { PARA_NOTICE_LOG("no admissible files\n"); ret = 0; goto err; } ret = -E_NOFD; 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 = load_afd(shmid, &mmd->afd); if (ret < 0) goto err; shm_destroy(shmid); ret = fstat(passed_fd, &statbuf); if (ret < 0) { PARA_ERROR_LOG("fstat error:\n"); ret = -ERRNO_TO_PARA_ERROR(errno); goto err; } 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->offset = 0; mmd->events++; mmd->num_played++; mmd->new_vss_status_flags &= (~VSS_NEXT); 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); mmd->afd.afhi.chunk_table = NULL; if (passed_fd >= 0) close(passed_fd); if (ret < 0) PARA_ERROR_LOG("%s\n", para_strerror(-ret)); mmd->new_vss_status_flags = VSS_NEXT; } /* * If the next chunk needs to be sent, pass a pointer to the chunk data to all * registered fec clients and to each sender's ->send() method. */ static void vss_send(struct vss_task *vsst) { int i, ret; bool fec_active = false; struct timeval due; struct fec_client *fc, *tmp_fc; char *buf; uint32_t len; if (!vsst->map || !vss_playing()) return; if (!barrier_has_passed("eof", &vsst->eof_barrier)) return; if (!barrier_has_passed("data send", &vsst->data_send_barrier)) 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 = true; continue; } if (compute_next_fec_slice(fc, vsst) <= 0) continue; 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); fec_active = true; } if (mmd->current_chunk >= mmd->afd.afhi.chunks_total) { /* eof */ if (!fec_active) mmd->new_vss_status_flags |= VSS_NEXT; return; } compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv, &mmd->stream_start, &due); if (tv_diff(&due, now, NULL) > 0) return; if (!mmd->chunks_sent) { mmd->stream_start = *now; mmd->events++; } 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 ->send() even if len is zero because senders might * have data queued which can be sent now. */ FOR_EACH_SENDER(i) { if (!senders[i]->send) continue; senders[i]->send(mmd->current_chunk, buf, len, vsst->header_buf, vsst->header_len); } } mmd->chunks_sent++; mmd->current_chunk++; } static int vss_post_monitor(struct sched *s, void *context) { int ret, i; struct vss_task *vsst = context; bool about2pause, next_flag_flipped; ret = task_get_notification(vsst->task); if (ret < 0) { afh_free_header(vsst->header_buf, mmd->afd.audio_format_id); afh_close(vsst->afh_context, mmd->afd.audio_format_id); return ret; } /* 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] (&mmd->sender_cmd_data); if (ret < 0) PARA_ERROR_LOG("%s\n", para_strerror(-ret)); } mmd->sender_cmd_data.cmd_num = -1; } vss_send(vsst); /* might change vss status flags */ about2pause = vss_paused() && (mmd->vss_status_flags & VSS_PLAYING); next_flag_flipped = vss_next() && !(mmd->vss_status_flags & VSS_NEXT); if (next_flag_flipped || about2pause || vss_repos()) { /* eos */ struct fec_client *fc, *tmp; /* shut down senders and fec clients */ 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; mmd->chunks_sent = 0; set_eof_barrier(vsst); } if (vss_repos()) /* set current chunk as requested */ mmd->current_chunk = afh_get_start_chunk( mmd->repos_request, &mmd->afd.afhi, mmd->afd.audio_format_id); if (about2pause || vss_repos()) { /* set offset for the next stream */ struct timeval offset; tv_scale(mmd->current_chunk, &mmd->afd.afhi.chunk_tv, &offset); mmd->offset = tv2ms(&offset); mmd->new_vss_status_flags &= ~VSS_REPOS; } if (next_flag_flipped) vss_eof(vsst); if (need_to_request_new_audio_file(vsst)) { PARA_DEBUG_LOG("ready and playing, but no audio file\n"); vsst->afsss = AFS_SOCKET_CHECK_FOR_WRITE; } else if (vsst->afsss == AFS_SOCKET_CHECK_FOR_WRITE) { if (sched_write_ok(vsst->afs_socket, s)) { PARA_INFO_LOG("requesting new fd from afs\n"); ret = write_buffer(vsst->afs_socket, "new"); if (ret < 0) return ret; vsst->afsss = AFS_SOCKET_AFD_PENDING; } } else if (vsst->afsss == AFS_SOCKET_AFD_PENDING) recv_afs_result(vsst, s); FOR_EACH_SENDER(i) if (senders[i]->post_monitor) senders[i]->post_monitor(); if ((vss_playing() && !(mmd->vss_status_flags & VSS_PLAYING)) || (vss_next() && vss_playing())) tv_add(now, &announce_tv, &vsst->data_send_barrier); return 0; } /** * Initialize the virtual streaming system. * * This initializes all supported senders and registers the vss task to the * scheduler. If the autoplay command line option was given, the VSS_PLAYING * status flag is set to start streaming. * * \param afs_socket Used for communication with the audio file selector. * \param s The scheduler instance to register the vss task. */ void vss_init(int afs_socket, struct sched *s) { static struct vss_task vss_task_struct, *vsst = &vss_task_struct; int i; vsst->header_interval.tv_sec = 5; /* should this be configurable? */ vsst->afs_socket = afs_socket; init_list_head(&fec_client_list); FOR_EACH_SENDER(i) { PARA_INFO_LOG("initializing %s sender\n", senders[i]->name); senders[i]->init(); } mmd->sender_cmd_data.cmd_num = -1; if (OPT_GIVEN(AUTOPLAY)) { struct timeval tmp; mmd->vss_status_flags |= VSS_PLAYING; mmd->new_vss_status_flags |= VSS_PLAYING; ms2tv(OPT_UINT32_VAL(AUTOPLAY_DELAY), &tmp); tv_add(clock_get_realtime(NULL), &tmp, &vsst->autoplay_barrier); tv_add(&vsst->autoplay_barrier, &announce_tv, &vsst->data_send_barrier); } vsst->task = task_register(&(struct task_info) { .name = "vss", .pre_monitor = vss_pre_monitor, .post_monitor = vss_post_monitor, .context = vsst, }, s); } /** * Turn off the virtual streaming system. * * This is only executed on exit. It calls the ->shutdown method of all senders. */ void vss_shutdown(void) { int i; bool is_command_handler = process_is_command_handler(); FOR_EACH_SENDER(i) { if (!senders[i]->shutdown) continue; if (!is_command_handler) PARA_NOTICE_LOG("shutting down %s sender\n", senders[i]->name); senders[i]->shutdown(); } }