Switch audiod over to the buffer tree API.
authorAndre Noll <maan@systemlinux.org>
Wed, 6 Jan 2010 23:48:47 +0000 (00:48 +0100)
committerAndre Noll <maan@systemlinux.org>
Wed, 6 Jan 2010 23:48:47 +0000 (00:48 +0100)
Still a bit rough and there are too many btr merges.

24 files changed:
alsa_write.c
amp_filter.c
audiod.c
audiod.h
audiod_command.c
buffer_tree.c
buffer_tree.h
compress_filter.c
dccp_recv.c
file_write.c
filter.c
filter.h
http_recv.c
mp3dec_filter.c
oggdec_filter.c
oss_write.c
prebuffer_filter.c
recv.h
udp_recv.c
wmadec_filter.c
write.c
write.h
write_common.c
write_common.h

index 7762197..8e3688f 100644 (file)
@@ -66,6 +66,7 @@ static int alsa_init(struct private_alsa_write_data *pad,
        snd_pcm_uframes_t period_size;
        int err;
 
+       PARA_INFO_LOG("opening %s\n", conf->device_arg);
        err = snd_pcm_open(&pad->handle, conf->device_arg,
                SND_PCM_STREAM_PLAYBACK, SND_PCM_NONBLOCK);
        if (err < 0)
@@ -290,7 +291,6 @@ static void alsa_close(struct writer_node *wn)
                snd_pcm_close(pad->handle);
                snd_config_update_free_global();
        }
-       alsa_cmdline_parser_free(wn->conf);
        free(pad);
 }
 
@@ -312,7 +312,7 @@ again:
                return;
        btr_merge(btrn, wn->min_iqs);
        bytes = btr_next_buffer(btrn, &data);
-       if (bytes < pad->bytes_per_frame) { /* eof */
+       if (ret < 0 || bytes < pad->bytes_per_frame) { /* eof */
                assert(btr_no_parent(btrn));
                ret = -E_ALSA_EOF;
                if (!pad->handle)
@@ -321,13 +321,13 @@ again:
                avail = snd_pcm_avail_update(pad->handle);
                if (avail + 1000 > pad->buffer_frames)
                        goto err;
+               PARA_DEBUG_LOG("waiting for device to drain\n");
                return;
        }
        if (!pad->handle) {
                struct alsa_write_args_info *conf = wn->conf;
                if (bytes == 0) /* no data available */
                        return;
-               PARA_CRIT_LOG("alsa init\n");
                /* defaults */
                pad->samplerate = conf->samplerate_arg;
                pad->channels = conf->channels_arg;
@@ -394,12 +394,17 @@ err_out:
        return NULL;
 }
 
+static void alsa_free_config(void *conf)
+{
+       alsa_cmdline_parser_free(conf);
+}
+
 /**
- * the init function of the alsa writer
+ * The init function of the alsa writer.
  *
- * \param w pointer to the writer to initialize
+ * \param w Pointer to the writer to initialize.
  *
- * \sa struct writer
+ * \sa \ref struct writer.
  */
 void alsa_write_init(struct writer *w)
 {
@@ -414,6 +419,7 @@ void alsa_write_init(struct writer *w)
        w->post_select_btr = alsa_write_post_select_btr;
        w->parse_config = alsa_parse_config;
        w->shutdown = NULL; /* nothing to do */
+       w->free_config = alsa_free_config;
        w->help = (struct ggo_help) {
                .short_help = alsa_write_args_info_help,
                .detailed_help = alsa_write_args_info_detailed_help
index d0a893f..e91791b 100644 (file)
@@ -156,6 +156,11 @@ err:
        btr_remove_node(btrn);
 }
 
+static void amp_free_config(void *conf)
+{
+       amp_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the amplify filter.
  *
@@ -172,6 +177,7 @@ void amp_filter_init(struct filter *f)
        f->pre_select = generic_filter_pre_select;
        f->post_select = amp_post_select;
        f->parse_config = amp_parse_config;
+       f->free_config = amp_free_config;
        f->help = (struct ggo_help) {
                .short_help = amp_filter_args_info_help,
                .detailed_help = amp_filter_args_info_detailed_help
index d946093..db7ae8a 100644 (file)
--- a/audiod.c
+++ b/audiod.c
@@ -180,9 +180,12 @@ char *get_time_string(int slot_num)
 {
        int ret, seconds = 0, length;
        struct timeval *tmp, sum, sss, /* server stream start */
+               rstime, /* receiver start time */
+               wstime, /* writer start time */
                wtime, /* now - writer start */
                rskip; /* receiver start - sss */
        struct slot_info *s = slot_num < 0? NULL : &slot[slot_num];
+       char *msg;
 
        if (audiod_status == AUDIOD_OFF)
                goto empty;
@@ -194,7 +197,7 @@ char *get_time_string(int slot_num)
        if (audiod_status == AUDIOD_ON && !s)
                goto empty;
        /* valid status items and playing */
-       if (s) { /* writer active in this slot */
+       if (s && s->wns) { /* writer active in this slot */
                length = s->seconds_total;
                tmp = &s->server_stream_start;
        } else { /* standby mode, rely on status items */
@@ -205,15 +208,18 @@ char *get_time_string(int slot_num)
                tv_diff(tmp, &stat_task->sa_time_diff, &sss);
        else
                tv_add(tmp, &stat_task->sa_time_diff, &sss);
-       if (!s) {
+       if (!s || !s->wns) {
                struct timeval diff;
                tv_diff(now, &sss, &diff);
                seconds = diff.tv_sec + stat_task->offset_seconds;
                goto out;
        }
-       tv_diff(now, &s->wstime, &wtime);
+       btr_get_node_start(s->wns[0].btrn, &wstime);
+       tv_diff(now, &wstime, &wtime);
+       //PARA_CRIT_LOG("offset %d\n", s->offset_seconds);
        seconds = s->offset_seconds;
-       ret = tv_diff(&s->rstime, &sss, &rskip);
+       btr_get_node_start(s->receiver_node->btrn, &rstime);
+       ret = tv_diff(&rstime, &sss, &rskip);
        if (ret > 0) { /* audiod was started in the middle of the stream */
                tv_add(&wtime, &rskip, &sum);
                seconds += sum.tv_sec;
@@ -222,7 +228,7 @@ char *get_time_string(int slot_num)
 out:
        seconds = PARA_MIN(seconds, length);
        seconds = PARA_MAX(seconds, 0);
-       return make_message(
+       msg = make_message(
                "%s%d:%02d [%d:%02d] (%d%%/%d:%02d)",
                s? "" : "~",
                seconds / 60,
@@ -233,6 +239,8 @@ out:
                length / 60,
                length % 60
        );
+       PARA_DEBUG_LOG("slot %d: %s\n", slot_num, msg);
+       return msg;
 empty:
        return para_strdup(NULL);
 }
@@ -312,30 +320,73 @@ static void close_receiver(int slot_num)
        PARA_NOTICE_LOG("closing %s receiver in slot %d\n",
                audio_formats[s->format], slot_num);
        a->receiver->close(s->receiver_node);
+       btr_free_node(s->receiver_node->btrn);
        free(s->receiver_node);
        s->receiver_node = NULL;
        stat_task->current_audio_format_num = -1;
 }
 
-static void kill_all_decoders(int error)
+static void writer_cleanup(struct writer_node *wn)
+{
+       struct writer *w = writers + wn->writer_num;
+
+       w->close(wn);
+       btr_free_node(wn->btrn);
+}
+
+static void close_writers(struct slot_info *s)
+{
+       struct audio_format_info *a;
+       int i;
+
+       if (s->format < 0)
+               return;
+       a = afi + s->format;
+       if (a->num_writers == 0)
+               writer_cleanup(s->wns);
+       else {
+               for (i = 0; i < a->num_writers; i++)
+                       writer_cleanup(s->wns + i);
+       }
+       free(s->wns);
+       s->wns = NULL;
+}
+
+static void _close_filters(struct slot_info *s)
 {
        int i;
+       struct audio_format_info *a = afi + s->format;
+       if (a->num_filters == 0)
+               return;
+       for (i = 0; i < a->num_filters; i++) {
+               struct filter_node *fn = s->fns + i;
+               struct filter *f = filters + fn->filter_num;
+
+               f->close(fn);
+               btr_free_node(fn->btrn);
+       }
+       free(s->fns);
+       s->fns = NULL;
+}
+
+static void kill_all_decoders(int error)
+{
+       int i, j;
 
        FOR_EACH_SLOT(i) {
                struct slot_info *s = &slot[i];
-               if (s->wng && s->wng->task.error >= 0) {
-                       PARA_INFO_LOG("deactivating wng in slot %d\n",
-                               i);
-                       s->wng->task.error = error;
-               }
-               if (s->fc && s->fc->task.error >= 0) {
-                       PARA_INFO_LOG("deactivatimg filter chain in slot %d\n", i);
-                       s->fc->task.error = error;
-               }
-               if (s->receiver_node && s->receiver_node->task.error >= 0) {
-                       PARA_INFO_LOG("deactivating receiver_node in slot %d\n", i);
+               struct audio_format_info *a;
+               if (s->format < 0)
+                       continue;
+               a = afi + s->format;
+               if (s->wns)
+                       for (j = 0; j < a->num_writers; j++)
+                               s->wns[j].task.error = error;
+               if (s->fns)
+                       for (j = 0; j < a->num_writers; j++)
+                               s->fns[j].task.error = error;
+               if (s->receiver_node)
                        s->receiver_node->task.error = error;
-               }
        }
 }
 
@@ -350,7 +401,7 @@ static int get_empty_slot(void)
                        clear_slot(i);
                        return i;
                }
-               if (s->wng || s->receiver_node || s->fc)
+               if (s->wns || s->receiver_node || s->fns)
                        continue;
                clear_slot(i);
                return i;
@@ -372,90 +423,71 @@ int num_filters(int audio_format_num)
        return afi[audio_format_num].num_filters;
 }
 
-static void open_filters(int slot_num)
+static void open_filters(struct slot_info *s)
 {
-       struct slot_info *s = &slot[slot_num];
-       struct audio_format_info *a = &afi[s->format];
+       struct audio_format_info *a = afi + s->format;
        struct filter_node *fn;
        int nf = a->num_filters;
+       struct btr_node *parent;
        int i;
 
-       s->fc = NULL;
-       if (!nf)
+       if (nf == 0)
                return;
        PARA_INFO_LOG("opening %s filters\n", audio_formats[s->format]);
-       s->fc = para_calloc(sizeof(struct filter_chain));
-       s->fc->filter_nodes = para_malloc(nf * sizeof(struct filter_node));
-       s->fc->inbufp = &s->receiver_node->buf;
-       s->fc->in_loaded = &s->receiver_node->loaded;
-       s->fc->input_error = &s->receiver_node->task.error;
-       s->fc->task.pre_select = NULL;
-       s->fc->task.post_select = filter_post_select;
-       s->fc->task.error = 0;
-       s->fc->num_filters = nf;
-
-       s->receiver_node->output_error = &s->fc->task.error;
-       sprintf(s->fc->task.status, "filter chain");
-       FOR_EACH_FILTER_NODE(fn, s->fc, i) {
+       assert(s->fns == NULL);
+       s->fns = para_calloc(nf * sizeof(struct filter_node));
+       parent = s->receiver_node->btrn;
+       for (i = 0; i < nf; i++) {
                struct filter *f = filters + a->filter_nums[i];
+               fn = s->fns + i;
                fn->filter_num = a->filter_nums[i];
                fn->conf = a->filter_conf[i];
-               fn->fc = s->fc;
-               fn->loaded = 0;
-               INIT_LIST_HEAD(&fn->callbacks);
+               fn->task.pre_select = f->pre_select;
+               fn->task.post_select = f->post_select;
+               fn->btrn = btr_new_node(f->name, parent, f->execute, fn);
                f->open(fn);
+               register_task(&fn->task);
+               parent = fn->btrn;
                PARA_NOTICE_LOG("%s filter %d/%d (%s) started in slot %d\n",
-                       audio_formats[s->format], i,  nf, f->name, slot_num);
-               s->fc->outbufp = &fn->buf;
-               s->fc->out_loaded = &fn->loaded;
+                       audio_formats[s->format], i,  nf, f->name, s - slot);
+               sprintf(fn->task.status, "%s (slot %d)", f->name, s - slot);
        }
-       register_task(&s->fc->task);
 }
 
-static void open_writers(int slot_num)
+static void open_writers(struct slot_info *s)
 {
-       int ret, i;
-       struct slot_info *s = &slot[slot_num];
-       struct audio_format_info *a = &afi[s->format];
-
-       PARA_INFO_LOG("opening %s writers\n", audio_formats[s->format]);
-       if (!a->num_writers)
-               s->wng = setup_default_wng();
-       else
-               s->wng = wng_new(a->num_writers);
-       if (s->fc) {
-               s->wng->bufp = s->fc->outbufp;
-               s->wng->loaded = s->fc->out_loaded;
-               s->wng->input_error = &s->fc->task.error;
-               s->wng->channels = &s->fc->channels;
-               s->wng->samplerate = &s->fc->samplerate;
-               s->fc->output_error = &s->wng->task.error;
-               PARA_INFO_LOG("samplerate: %d\n", *s->wng->samplerate);
-       } else {
-               s->wng->bufp = &s->receiver_node->buf;
-               s->wng->loaded = &s->receiver_node->loaded;
-               s->wng->input_error = &s->receiver_node->task.error;
-       }
-       for (i = 0; i < a->num_writers; i++) {
-               s->wng->writer_nodes[i].conf = a->writer_conf[i];
-               s->wng->writer_nodes[i].writer_num = a->writer_nums[i];
+       int i;
+       struct audio_format_info *a = afi + s->format;
+       struct writer_node *wn;
+       struct btr_node *parent = s->fns[a->num_filters - 1].btrn;
+
+       assert(s->wns == NULL);
+       s->wns = para_calloc(PARA_MAX(1U, a->num_writers)
+               * sizeof(struct writer_node));
+       if (a->num_writers == 0)
+               setup_writer_node(NULL, parent, s->wns);
+       else {
+               PARA_INFO_LOG("opening %s writers\n", audio_formats[s->format]);
+               for (i = 0; i < a->num_writers; i++) {
+                       wn = s->wns + i;
+                       wn->conf = a->writer_conf[i];
+                       wn->writer_num = a->writer_nums[i];
+                       register_writer_node(wn, parent);
+               }
        }
-       ret = wng_open(s->wng);
-       if (ret < 0)
-               return;
-       s->wstime = *now;
        s->server_stream_start = stat_task->server_stream_start.tv_sec?
                stat_task->server_stream_start : *now;
        s->offset_seconds = stat_task->offset_seconds;
        s->seconds_total = stat_task->length_seconds;
-       activate_inactive_grab_clients(s->format, s->fc);
 }
 
+/* returns slot num on success */
 static int open_receiver(int format)
 {
        struct audio_format_info *a = &afi[format];
        struct slot_info *s;
        int ret, slot_num;
+       struct receiver *r = a->receiver;
        struct receiver_node *rn;
        const struct timeval restart_delay = {2, 0};
 
@@ -467,22 +499,23 @@ static int open_receiver(int format)
        s->format = format;
        s->receiver_node = para_calloc(sizeof(struct receiver_node));
        rn = s->receiver_node;
-       rn->receiver = a->receiver;
+       rn->receiver = r;
        rn->conf = a->receiver_conf;
-       ret = a->receiver->open(s->receiver_node);
+       rn->btrn = btr_new_node(r->name, NULL, NULL, rn);
+       ret = r->open(rn);
        if (ret < 0) {
-               free(s->receiver_node);
+               btr_free_node(rn->btrn);
+               free(rn);
                s->receiver_node = NULL;
                goto err;
        }
        PARA_NOTICE_LOG("started %s: %s receiver in slot %d\n",
-               audio_formats[s->format], a->receiver->name, slot_num);
-       rn->task.pre_select = a->receiver->pre_select;
-       rn->task.post_select = a->receiver->post_select;
-       s->rstime = *now;
-       sprintf(rn->task.status, "%s receiver node", rn->receiver->name);
+               audio_formats[s->format], r->name, slot_num);
+       rn->task.pre_select = r->pre_select;
+       rn->task.post_select = r->post_select;
+       sprintf(rn->task.status, "%s receiver node", r->name);
        register_task(&rn->task);
-       ret = 1;
+       ret = slot_num;
 err:
        if (ret < 0)
                PARA_ERROR_LOG("%s\n", para_strerror(-ret));
@@ -508,19 +541,20 @@ static int receiver_running(int format)
        return ret;
 }
 
-static void open_current_receiver(struct sched *s)
+/* returns slot num on success. */
+static int open_current_receiver(struct sched *s)
 {
        struct timeval diff;
        int ret, cafn = stat_task->current_audio_format_num;
 
        if (cafn < 0 || !stat_task->ct)
-               return;
+               return -1;
        /* Do nothing if the 'N' flag is set or the 'P' flag is unset */
        if (stat_task->vss_status != VSS_STATUS_FLAG_PLAYING)
-               return;
+               return -1;
        ret = receiver_running(cafn);
        if (ret > 0) /* already running and not eof */
-               return;
+               return -1;
        if (ret < 0) { /* eof */
                /*
                 * para_server uses a zero start time during the announcement
@@ -528,16 +562,19 @@ static void open_current_receiver(struct sched *s)
                 * this period begins to avoid restarting the receiver that
                 * belongs to the file just completed.
                 */
-               if (stat_task->server_stream_start.tv_sec)
-                       return;
+               PARA_CRIT_LOG("->>>>>>>>>>>>>>>>>> no delay\n");
+               if (stat_task->server_stream_start.tv_sec != 0) {
+                       PARA_CRIT_LOG("->>>>>>>>>>>>>>>>>><delay\n");
+                       return -1;
+               }
        }
        if (tv_diff(now, &afi[cafn].restart_barrier, &diff) < 0) {
                /* avoid busy loop */
                s->timeout = diff;
-               return;
+               return -1;
        }
        /* start a new receiver */
-       open_receiver(cafn);
+       return open_receiver(cafn);
 }
 
 static unsigned compute_time_diff(const struct timeval *status_time)
@@ -906,9 +943,9 @@ static void command_post_select(struct sched *s, struct task *t)
        int ret;
        struct command_task *ct = container_of(t, struct command_task, task);
        static struct timeval last_status_dump;
-       struct timeval tmp;
+       struct timeval tmp, delay = {0, 500 * 1000};
 
-       tv_add(&last_status_dump, &(struct timeval){0, 500 * 1000}, &tmp);
+       tv_add(&last_status_dump, &delay, &tmp);
        if (tv_diff(&tmp, now, NULL) < 0) {
                audiod_status_dump();
                last_status_dump = *now;
@@ -975,19 +1012,22 @@ static void set_stat_task_restart_barrier(unsigned seconds)
 static void try_to_close_slot(int slot_num)
 {
        struct slot_info *s = &slot[slot_num];
+       struct audio_format_info *a = afi + s->format;
+       int i;
 
        if (s->format < 0)
                return;
        if (s->receiver_node && s->receiver_node->task.error != -E_TASK_UNREGISTERED)
                return;
-       if (s->fc && s->fc->task.error != -E_TASK_UNREGISTERED)
-               return;
-       if (s->wng && s->wng->task.error != -E_TASK_UNREGISTERED)
-               return;
+       for (i = 0; i < a->num_filters; i++)
+               if (s->fns && s->fns[i].task.error != -E_TASK_UNREGISTERED)
+                       return;
+       for (i = 0; i < a->num_writers; i++)
+               if (s->wns && s->wns[i].task.error != -E_TASK_UNREGISTERED)
+                       return;
        PARA_INFO_LOG("closing slot %d\n", slot_num);
-       wng_close(s->wng);
-       close_filters(s->fc);
-       free(s->fc);
+       close_writers(s);
+       _close_filters(s);
        close_receiver(slot_num);
        clear_slot(slot_num);
 }
@@ -998,50 +1038,28 @@ static void try_to_close_slot(int slot_num)
  */
 static void start_stop_decoders(struct sched *s)
 {
-       int i;
+       int i, ret;
+       struct slot_info *sl;
+       struct audio_format_info *a;
 
        FOR_EACH_SLOT(i)
                try_to_close_slot(i);
-       if (audiod_status != AUDIOD_ON ||
-                       !(stat_task->vss_status & VSS_STATUS_FLAG_PLAYING))
-               return kill_all_decoders(-E_NOT_PLAYING);
-       open_current_receiver(s);
-       FOR_EACH_SLOT(i) {
-               struct slot_info *sl = &slot[i];
-               struct audio_format_info *a;
-               struct timeval diff;
-
-               if (sl->format < 0)
-                       continue;
-               a = &afi[sl->format];
-               if (!sl->receiver_node)
-                       continue;
-               if ((!a->num_filters || sl->fc) && sl->wng)
-                       continue; /* everything already started */
-               if (!a->num_filters) {
-                       if (sl->receiver_node->loaded && !sl->wng) {
-                               open_writers(i);
-                       }
-                       continue;
-               }
-               if (sl->receiver_node->loaded && !sl->fc) {
-                       open_filters(i);
-                       continue;
-               }
-               if (sl->wng || !sl->fc || !*sl->fc->out_loaded)
-                       continue;
-               if (tv_diff(now, &initial_delay_barrier, &diff) > 0) {
-                       open_writers(i);
-                       continue;
-               }
-               PARA_INFO_LOG("initial delay: %lu ms left\n", tv2ms(&diff));
-               if (tv_diff(&s->timeout, &diff, NULL) > 0) {
-                       s->timeout = diff;
-               }
-       }
+//     if (audiod_status != AUDIOD_ON ||
+//                     !(stat_task->vss_status & VSS_STATUS_FLAG_PLAYING))
+//             return kill_all_decoders(-E_NOT_PLAYING);
+       ret = open_current_receiver(s);
+       if (ret < 0)
+               return;
+       sl = slot + ret;
+       a = afi + sl->format;
+       if (a->num_filters)
+               open_filters(sl);
+       open_writers(sl);
+       btr_log_tree(sl->receiver_node->btrn, LL_NOTICE);
+       s->timeout.tv_sec = 0;
+       s->timeout.tv_usec = 1;
 }
 
-
 /* restart the client task if necessary */
 static void status_pre_select(struct sched *s, struct task *t)
 {
@@ -1240,7 +1258,7 @@ int main(int argc, char *argv[])
        register_task(&sig_task->task);
        register_task(&cmd_task->task);
        register_task(&stat_task->task);
-       s.default_timeout.tv_sec = 0;
+       s.default_timeout.tv_sec = 2;
        s.default_timeout.tv_usec = 999 * 1000;
        ret = schedule(&s);
 
index 2d8fec1..90ca22c 100644 (file)
--- a/audiod.h
+++ b/audiod.h
@@ -51,10 +51,6 @@ struct audiod_command {
 struct slot_info {
        /** Number of the audio format in this slot. */
        int format;
-       /** Receiver start time. */
-       struct timeval rstime;
-       /** Writer start time. */
-       struct timeval wstime;
        /** The stream_start status item announced by para_server.  */
        struct timeval server_stream_start;
        /** The offset status item announced by para_server. */
@@ -65,8 +61,10 @@ struct slot_info {
        struct receiver_node *receiver_node;
        /** The active filter chain. */
        struct filter_chain *fc;
-       /** The active writer node group. */
-       struct writer_node_group *wng;
+       /** The array of filter nodes. */
+       struct filter_node *fns;
+       /** The array of writers attached to the last filter. */
+       struct writer_node *wns;
 };
 
 extern struct slot_info slot[MAX_STREAM_SLOTS];
index b790fc1..a5f1bde 100644 (file)
@@ -24,6 +24,7 @@
 #include "net.h"
 #include "daemon.h"
 #include "string.h"
+#include "write.h"
 #include "fd.h"
 #include "audiod_command_list.h"
 
@@ -192,18 +193,22 @@ __malloc static char *audiod_status_string(void)
 
 static int get_play_time_slot_num(void)
 {
-       int i, oldest = -1;
+       int i, oldest_slot = -1;
+       struct timeval oldest_wstime = {0, 0};
 
        FOR_EACH_SLOT(i) {
                struct slot_info *s = &slot[i];
-               if (!s->wng)
+               struct timeval wstime;
+               if (!s->wns)
                        continue;
-               if (oldest >= 0 && tv_diff(&s->wstime, &slot[oldest].wstime,
-                               NULL) > 0)
+               btr_get_node_start(s->wns[0].btrn, &wstime);
+               if (oldest_slot >= 0 && tv_diff(&wstime, &oldest_wstime, NULL) > 0)
                        continue;
-               oldest = i;
+               oldest_wstime = wstime;
+               oldest_slot = i;
        }
-       return oldest;
+       //PARA_CRIT_LOG("oldest slot: %d\n", oldest_slot);
+       return oldest_slot;
 }
 
 __malloc static char *decoder_flags(void)
@@ -216,9 +221,9 @@ __malloc static char *decoder_flags(void)
                char flag = '0';
                if (s->receiver_node)
                        flag += 1;
-               if (s->fc)
+               if (s->fns)
                        flag += 2;
-               if (s->wng)
+               if (s->wns)
                        flag += 4;
                flags[i] = flag;
        }
index 23599b8..1be2037 100644 (file)
@@ -6,6 +6,7 @@
 #include "string.h"
 #include "buffer_tree.h"
 #include "error.h"
+#include "sched.h"
 
 
 struct btr_buffer {
@@ -29,6 +30,8 @@ struct btr_node {
        struct list_head node;
        /* The children nodes of this btr node are linked together in a list. */
        struct list_head children;
+       /* Time of first data transfer. */
+       struct timeval start;
        /**
         * The input queue is a list of references to btr buffers. Each item on
         * the list represents an input buffer which has not been completely
@@ -58,6 +61,8 @@ struct btr_node *btr_new_node(const char *name, struct btr_node *parent,
        btrn->parent = parent;
        btrn->execute = handler;
        btrn->context = context;
+       btrn->start.tv_sec = 0;
+       btrn->start.tv_usec = 0;
        if (parent)
                list_add_tail(&btrn->node, &parent->children);
        INIT_LIST_HEAD(&btrn->children);
@@ -106,12 +111,16 @@ static void add_btrb_to_children(struct btr_buffer *btrb,
 {
        struct btr_node *ch;
 
+       if (btrn->start.tv_sec == 0)
+               btrn->start = *now;
        FOR_EACH_CHILD(ch, btrn) {
                struct btr_buffer_reference *br = para_malloc(sizeof(*br));
                br->btrb = btrb;
                br->consumed = consumed;
                list_add_tail(&br->node, &ch->input_queue);
                btrb->refcount++;
+               if (ch->start.tv_sec == 0)
+                       ch->start = *now;
        }
 }
 
@@ -353,8 +362,8 @@ static int merge_input(struct btr_node *btrn)
        }
        /* make a new btrb that combines the two buffers and a br to it. */
        sz = szs[0] + szs[1];
-       //PARA_CRIT_LOG("merging input buffers: (%zu, %zu) -> %zu\n",
-       //      szs[0], szs[1], sz);
+       PARA_DEBUG_LOG("merging input buffers: (%zu, %zu) -> %zu\n",
+               szs[0], szs[1], sz);
        buf = para_malloc(sz);
        /* TODO: Avoid this memcopy by introducing btr buffer pool. */
        memcpy(buf, bufs[0], szs[0]);
@@ -379,6 +388,7 @@ void btr_merge(struct btr_node *btrn, size_t dest_size)
                size_t len = btr_next_buffer(btrn, &buf);
                if (len >= dest_size)
                        return;
+               PARA_DEBUG_LOG("input size = %zu < %zu = dest\n", len, dest_size);
                if (merge_input(btrn) < 2)
                        return;
        }
@@ -434,3 +444,8 @@ int btr_node_status(struct btr_node *btrn, size_t min_iqs,
        }
        return 1;
 }
+
+void btr_get_node_start(struct btr_node *btrn, struct timeval *tv)
+{
+       *tv = btrn->start;
+}
index 9039968..9de9926 100644 (file)
@@ -33,3 +33,4 @@ int btr_pushdown_one(struct btr_node *btrn);
 bool btr_inplace_ok(struct btr_node *btrn);
 int btr_node_status(struct btr_node *btrn, size_t min_iqs,
                enum btr_node_type type);
+void btr_get_node_start(struct btr_node *btrn, struct timeval *tv);
index a0b53bd..3905561 100644 (file)
@@ -188,6 +188,11 @@ static void open_compress(struct filter_node *fn)
        pcd->max_gain = 1 << (pcd->conf->inertia_arg + pcd->conf->aggressiveness_arg);
 }
 
+static void compress_free_config(void *conf)
+{
+       compress_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the compress filter.
  *
@@ -204,6 +209,7 @@ void compress_filter_init(struct filter *f)
        f->pre_select = generic_filter_pre_select;
        f->post_select = compress_post_select;
        f->parse_config = compress_parse_config;
+       f->free_config = compress_free_config;
        f->help = (struct ggo_help) {
                .short_help = compress_filter_args_info_help,
                .detailed_help = compress_filter_args_info_detailed_help
index 58c969a..4e1860e 100644 (file)
@@ -136,6 +136,11 @@ static void dccp_recv_post_select(struct sched *s, struct task *t)
                t->error = -E_RECV_EOF;
 }
 
+static void dccp_recv_free_config(void *conf)
+{
+       dccp_recv_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the dccp receiver.
  *
@@ -154,6 +159,7 @@ void dccp_recv_init(struct receiver *r)
        r->pre_select = dccp_recv_pre_select;
        r->post_select = dccp_recv_post_select;
        r->parse_config = dccp_recv_parse_config;
+       r->free_config = dccp_recv_free_config;
        r->help = (struct ggo_help) {
                .short_help = dccp_recv_args_info_help,
                .detailed_help = dccp_recv_args_info_detailed_help
index 4c04386..7a24b3a 100644 (file)
@@ -182,6 +182,11 @@ __malloc static void *file_write_parse_config(const char *options)
        return NULL;
 }
 
+static void file_write_free_config(void *conf)
+{
+       file_cmdline_parser_free(conf);
+}
+
 /** the init function of the file writer */
 void file_write_init(struct writer *w)
 {
@@ -194,6 +199,7 @@ void file_write_init(struct writer *w)
        w->post_select = file_write_post_select;
        w->post_select_btr = file_write_post_select_btr;
        w->parse_config = file_write_parse_config;
+       w->free_config = file_write_free_config;
        w->close = file_write_close;
        w->shutdown = NULL; /* nothing to do */
        w->help = (struct ggo_help) {
index e7fc2cd..599aaad 100644 (file)
--- a/filter.c
+++ b/filter.c
@@ -154,7 +154,7 @@ static int parse_config(int argc, char *argv[])
        return 1;
 }
 
-/* TODO: support more than one filter, actually parse options */
+/* TODO: support buffer tree, not just a chain. */
 static int __noreturn main_btr(void)
 {
        static struct sched s;
index ae0a11e..4a17c10 100644 (file)
--- a/filter.h
+++ b/filter.h
@@ -191,6 +191,7 @@ struct filter {
         * argv. On failure, a negative paraslash error code must be returned.
         */
        int (*parse_config)(int argc, char **argv, void **config);
+       void (*free_config)(void *conf);
 
        /** The help texts for this filter. */
        struct ggo_help help;
index b3b6210..ed362bf 100644 (file)
@@ -103,7 +103,7 @@ static void http_recv_pre_select(struct sched *s, struct task *t)
                para_fd_set(phd->fd, &s->rfds, &s->max_fileno);
 }
 
-#define HTTP_RECV_READ_BUF_SIZE 4096
+#define HTTP_RECV_READ_BUF_SIZE 16384
 
 static void http_recv_post_select(struct sched *s, struct task *t)
 {
@@ -186,7 +186,6 @@ static void http_recv_close(struct receiver_node *rn)
        close(phd->fd);
        free(rn->buf);
        free(rn->private_data);
-       http_recv_cmdline_parser_free(rn->conf);
 }
 
 static void *http_recv_parse_config(int argc, char **argv)
@@ -221,12 +220,17 @@ static int http_recv_open(struct receiver_node *rn)
        return 1;
 }
 
+static void http_recv_free_config(void *conf)
+{
+       http_recv_cmdline_parser_free(conf);
+}
+
 /**
- * the init function of the http receiver
+ * The init function of the http receiver.
  *
- * \param r pointer to the receiver struct to initialize
+ * \param r Pointer to the receiver struct to initialize.
  *
- * Just initialize all function pointers of \a r.
+ * This initializes all function pointers of \a r.
  */
 void http_recv_init(struct receiver *r)
 {
@@ -239,6 +243,7 @@ void http_recv_init(struct receiver *r)
        r->post_select = http_recv_post_select;
        r->shutdown = http_shutdown;
        r->parse_config = http_recv_parse_config;
+       r->free_config = http_recv_free_config;
        r->help = (struct ggo_help) {
                .short_help = http_recv_args_info_help,
                .detailed_help = http_recv_args_info_detailed_help
index 397a242..2472b2c 100644 (file)
@@ -179,7 +179,6 @@ static void mp3dec_close(struct filter_node *fn)
 
        free(fn->buf);
        fn->buf = NULL;
-       mp3dec_cmdline_parser_free(fn->conf);
        free(pmd);
        fn->private_data = NULL;
 }
@@ -318,6 +317,11 @@ static int mp3dec_execute(struct btr_node *btrn, const char *cmd, char **result)
        }
        return -ERRNO_TO_PARA_ERROR(ENOTSUP);
 }
+
+static void mp3dec_free_config(void *conf)
+{
+       mp3dec_cmdline_parser_free(conf);
+}
 /**
  * The init function of the mp3dec filter.
  *
@@ -334,6 +338,7 @@ void mp3dec_filter_init(struct filter *f)
        f->convert = mp3dec;
        f->close = mp3dec_close;
        f->parse_config = mp3dec_parse_config;
+       f->free_config = mp3dec_free_config;
        f->pre_select = generic_filter_pre_select;
        f->post_select = mp3dec_post_select;
        f->execute = mp3dec_execute;
index 2c9d25f..511e640 100644 (file)
@@ -384,6 +384,11 @@ err:
        return ret;
 }
 
+static void oggdec_free_config(void *conf)
+{
+       oggdec_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the ogg vorbis decoder.
  *
@@ -400,6 +405,7 @@ void oggdec_filter_init(struct filter *f)
        f->pre_select = generic_filter_pre_select;
        f->post_select = ogg_post_select;
        f->parse_config = oggdec_parse_config;
+       f->free_config = oggdec_free_config;
        f->execute = oggdec_execute;
        f->help = (struct ggo_help) {
                .short_help = oggdec_filter_args_info_help,
index 20b0500..4e5f9b5 100644 (file)
@@ -282,6 +282,11 @@ err_out:
        return NULL;
 }
 
+static void oss_free_config(void *conf)
+{
+       oss_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the oss writer.
  *
@@ -301,6 +306,7 @@ void oss_write_init(struct writer *w)
        w->post_select = oss_post_select;
        w->post_select_btr = oss_post_select_btr;
        w->parse_config = oss_parse_config;
+       w->free_config = oss_free_config;
        w->shutdown = NULL;
        w->help = (struct ggo_help) {
                .short_help = oss_write_args_info_help,
index adfe9de..207d332 100644 (file)
@@ -162,6 +162,11 @@ static void prebuffer_open(struct filter_node *fn)
        fn->buf = para_malloc(fn->bufsize);
 }
 
+static void prebuffer_free_config(void *conf)
+{
+       prebuffer_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the prebuffer filter.
  *
@@ -176,6 +181,7 @@ void prebuffer_filter_init(struct filter *f)
        f->close = prebuffer_close;
        f->convert = prebuffer_convert;
        f->parse_config = prebuffer_parse_config;
+       f->free_config = prebuffer_free_config;
        f->pre_select = prebuffer_pre_select;
        f->post_select = prebuffer_post_select;
        f->help = (struct ggo_help) {
diff --git a/recv.h b/recv.h
index eb6175c..3f13f10 100644 (file)
--- a/recv.h
+++ b/recv.h
@@ -56,6 +56,7 @@ struct receiver {
         * \a argc and \a argv.
         */
        void *(*parse_config)(int argc, char **argv);
+       void (*free_config)(void *conf);
        /**
         * Open one instance of the receiver.
         *
index bb166c5..f68a710 100644 (file)
@@ -18,7 +18,6 @@
 #include "ggo.h"
 #include "recv.h"
 #include "udp_recv.cmdline.h"
-#include "audiod.h"
 #include "string.h"
 #include "net.h"
 #include "fd.h"
@@ -273,6 +272,11 @@ err:
        return ret;
 }
 
+static void udp_recv_free_config(void *conf)
+{
+       udp_recv_cmdline_parser_free(conf);
+}
+
 /**
  * The init function of the udp receiver.
  *
@@ -291,6 +295,7 @@ void udp_recv_init(struct receiver *r)
        r->pre_select = udp_recv_pre_select;
        r->post_select = udp_recv_post_select;
        r->parse_config = udp_recv_parse_config;
+       r->free_config = udp_recv_free_config;
        r->help = (struct ggo_help) {
                .short_help = udp_recv_args_info_help,
                .detailed_help = udp_recv_args_info_detailed_help
index 00bdd6e..49f69b5 100644 (file)
@@ -1265,7 +1265,7 @@ next_buffer:
        for (;;) {
                char *out;
                int out_size = WMA_OUTPUT_BUFFER_SIZE;
-               if (converted + WMA_FRAME_SKIP + pwd->ahi.block_align > len)
+               if (converted + fn->min_iqs > len)
                        break;
                out = para_malloc(WMA_OUTPUT_BUFFER_SIZE);
                ret = wma_decode_superframe(pwd, out,
diff --git a/write.c b/write.c
index 4d06e80..c64ecb6 100644 (file)
--- a/write.c
+++ b/write.c
@@ -282,7 +282,7 @@ static int main_btr(struct sched *s)
 {
        int i, ret;
        struct check_wav_task_btr _cwt, *cwt = &_cwt;
-       struct writer_node **wns;
+       struct writer_node *wns;
 
        loglevel = get_loglevel_by_name(conf.loglevel_arg);
        sit.btrn = btr_new_node("stdin", NULL /* stdin has no parent */, NULL, NULL);
@@ -300,17 +300,17 @@ static int main_btr(struct sched *s)
        ret = -E_WRITE_SYNTAX;
        if (!conf.writer_given) {
                i = 0;
-               wns = para_malloc(sizeof(*wns));
-               wns[0] = setup_writer_node(NULL, cwt->btrn);
-               if (!wns[0])
+               wns = para_calloc(sizeof(*wns));
+               ret = setup_writer_node(NULL, cwt->btrn, wns);
+               if (ret < 0)
                        goto out;
                i = 1;
        } else {
-               wns = para_malloc(conf.writer_given * sizeof(*wns));
+               wns = para_calloc(conf.writer_given * sizeof(*wns));
                for (i = 0; i < conf.writer_given; i++) {
-                       wns[i] = setup_writer_node(conf.writer_arg[i],
-                               cwt->btrn);
-                       if (!wns[i])
+                       ret = setup_writer_node(conf.writer_arg[i],
+                               cwt->btrn, wns + i);
+                       if (ret < 0)
                                goto out;
                }
        }
@@ -320,7 +320,7 @@ static int main_btr(struct sched *s)
        ret = schedule(s);
 out:
        for (i--; i >= 0; i--) {
-               struct writer_node *wn = wns[i];
+               struct writer_node *wn = wns + i;
                struct writer *w = writers + wn->writer_num;
 
                w->close(wn);
diff --git a/write.h b/write.h
index 49e56c3..8e24451 100644 (file)
--- a/write.h
+++ b/write.h
@@ -46,6 +46,7 @@ struct writer {
         * more than once with different values of \a options.
         */
        void *(*parse_config)(const char *options);
+       void (*free_config)(void *conf);
        /**
         * Open one instance of this writer.
         *
index 9af8d68..8c33b39 100644 (file)
@@ -237,6 +237,21 @@ struct writer_node_group *setup_default_wng(void)
        return wng;
 }
 
+void register_writer_node(struct writer_node *wn, struct btr_node *parent)
+{
+       struct writer *w = writers + wn->writer_num;
+       char *name = make_message("%s writer", writer_names[wn->writer_num]);
+       int ret;
+
+       wn->btrn = btr_new_node(name, parent, w->execute, wn);
+       strcpy(wn->task.status, name);
+       free(name);
+       ret = w->open(wn);
+       wn->task.post_select = w->post_select_btr;
+       wn->task.pre_select = w->pre_select_btr;
+       register_task(&wn->task);
+}
+
 /**
  * Setup a writer node with the default writer.
  *
@@ -253,32 +268,19 @@ struct writer_node_group *setup_default_wng(void)
  *
  * \return A pointer to the allocated writer node group.
  */
-struct writer_node *setup_writer_node(const char *arg, struct btr_node *parent)
+int setup_writer_node(const char *arg, struct btr_node *parent,
+               struct writer_node *wn)
 {
-       struct writer_node *wn = para_calloc(sizeof(*wn));
-       struct writer *w;
-       char *name;
-
        if (arg)
                wn->conf = check_writer_arg(arg, &wn->writer_num);
        else {
                wn->writer_num = DEFAULT_WRITER;
                wn->conf = writers[DEFAULT_WRITER].parse_config("");
        }
-       if (!wn->conf) {
-               free(wn);
-               return NULL;
-       }
-       w = writers + wn->writer_num;
-       name = make_message("%s writer", writer_names[wn->writer_num]);
-       wn->btrn = btr_new_node(name, parent, w->execute, wn);
-       strcpy(wn->task.status, name);
-       free(name);
-       w->open(wn);
-       wn->task.post_select = w->post_select_btr;
-       wn->task.pre_select = w->pre_select_btr;
-       register_task(&wn->task);
-       return wn;
+       if (!wn->conf)
+               return -E_WRITE_COMMON_SYNTAX;
+       register_writer_node(wn, parent);
+       return 1;
 }
 
 
index ef5a7c0..8c33a51 100644 (file)
@@ -13,6 +13,8 @@ void writer_init(void);
 void *check_writer_arg(const char *wa, int *writer_num);
 struct writer_node_group *setup_default_wng(void);
 void print_writer_helps(int detailed);
-struct writer_node *setup_writer_node(const char *arg, struct btr_node *parent);
+void register_writer_node(struct writer_node *wn, struct btr_node *parent);
+int setup_writer_node(const char *arg, struct btr_node *parent,
+               struct writer_node *wn);
 int get_btr_samplerate(struct btr_node *btrn, int32_t *result);
 int get_btr_channels(struct btr_node *btrn, int32_t *result);