afh: Dynamic chunks.
[paraslash.git] / vss.c
diff --git a/vss.c b/vss.c
index 3ace49e9e90e8fb27144964a0f61f1fae267e843..24dfc6bb51af4edf67c645c0d3ccbda2ad8f1868 100644 (file)
--- a/vss.c
+++ b/vss.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 1997-2014 Andre Noll <maan@systemlinux.org>
+ * Copyright (C) 1997 Andre Noll <maan@tuebingen.mpg.de>
  *
  * Licensed under the GPL v2. For licencing details see COPYING.
  */
@@ -88,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;
 };
 
 /**
@@ -347,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
@@ -360,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,
@@ -611,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
        );
@@ -846,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;
@@ -855,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++;
 }
 
@@ -885,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);
@@ -944,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;
@@ -1012,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;
@@ -1026,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);
@@ -1067,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);
@@ -1116,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;
                        }
@@ -1126,11 +1101,38 @@ static void vss_send(struct vss_task *vsst)
        }
 }
 
-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;
 
+       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;
@@ -1185,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);
@@ -1202,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);
 }