integrate writers into para_audiod

Seems to work, but might still be full of bugs.

diff --git a/audiod.c b/audiod.c
index dccc2e15ba92bbbeb68f5acdb683270f9e25b482..7f10378945e899656f036e5d8e16fa7da34c01d6 100644 (file)
--- a/audiod.c
+++ b/audiod.c
@@ -36,6 +36,8 @@
 #include "daemon.h"
 #include "string.h"
 #include "fd.h"
+#include "write.h"
+#include "write_common.h"
 
 /** define the array of error lists needed by para_audiod */
 INIT_AUDIOD_ERRLISTS;
@@ -53,48 +55,47 @@ enum {AUDIOD_OFF, AUDIOD_ON, AUDIOD_STANDBY};
 
 /** defines how to handle one supported audio format */
 struct audio_format_info {
-/** pointer to the receiver for this audio format */
+       /** pointer to the receiver for this audio format */
        struct receiver *receiver;
-/** the receiver configuration */
+       /** the receiver configuration */
        void *receiver_conf;
-/** the number of filters that should be activated for this audio format */
+       /** the number of filters that should be activated for this audio format */
        unsigned int num_filters;
-/** pointer to the array of filters to be activated */
+       /** pointer to the array of filters to be activated */
        struct filter **filters;
-/** pointer to the array of filter configurations */
+       /** pointer to the array of filter configurations */
        void **filter_conf;
-/** output of the last filter is written to stdin of this command */
-       char *write_cmd;
-/** do not start receiver/filters/writer before this time */
+       /** the number of filters that should be activated for this audio format */
+       unsigned int num_writers;
+       /** pointer to the array of writers to be activated */
+       struct writer **writers;
+       /** pointer to the array of writer configurations */
+       void **writer_conf;
+       /** do not start receiver/filters/writer before this time */
        struct timeval restart_barrier;
 };
 
 /**
  * describes one instance of a receiver-filter-writer chain
  *
- * \sa receier_node, receiver, filter, filter_node, filter_chain_info
+ * \sa receier_node, receiver, filter, filter_node, filter_chain, writer,
+ * writer_node, writer_node_group.
   */
 struct slot_info {
        /** number of the audio format in this slot */
        int format;
-       /** the file descriptor of the writer */
-       int write_fd;
-       /** the process id of the writer */
-       pid_t wpid;
        /** time of the last successful read from the receiver */
        struct timeval rtime;
        /** time the last write to the write fd happend */
        struct timeval wtime;
        /** writer start time */
        struct timeval wstime;
-       /** did we include \a write_fd in the fdset */
-       int  wcheck;
-       /** set to one if we have sent the TERM signal to \a wpid */
-       int wkilled;
        /** the receiver info associated with this slot */
        struct receiver_node *receiver_node;
        /** the active filter chain */
        struct filter_chain *fc;
+       /** the active writer node group */
+       struct writer_node_group *wng;
 };
 static struct slot_info slot[MAX_STREAM_SLOTS];
 
@@ -374,7 +375,7 @@ static struct timeval *wstime(void)
        struct timeval *max = NULL;
        FOR_EACH_SLOT(i) {
                struct slot_info *s = &slot[i];
-               if (s->wpid <= 0)
+               if (!s->wng)
                        continue;
                if (max && tv_diff(&s->wstime, max, NULL) <= 0)
                        continue;
@@ -392,7 +393,7 @@ __malloc static char *decoder_flags(void)
                char flag = '0';
                if (s->receiver_node)
                        flag += 1;
-               if (s->wpid > 0)
+               if (s->wng)
                        flag += 2;
                decoder_flags[i] = flag;
        }
@@ -465,18 +466,6 @@ static void clear_slot(int slot_num)
        s->format = -1;
 }
 
-static void kill_stream_writer(int slot_num)
-{
-       struct slot_info *s = &slot[slot_num];
-
-       if (s->format < 0 || s->wkilled || s->wpid <= 0)
-               return;
-       PARA_DEBUG_LOG("kill -TERM %d (%s stream writer in slot %d)\n",
-               s->wpid, audio_formats[s->format], slot_num);
-       kill(s->wpid, SIGTERM);
-       s->wkilled = 1;
-}
-
 static void set_restart_barrier(int format, struct timeval *now)
 {
        struct timeval tmp;
@@ -505,22 +494,31 @@ static void close_receiver(int slot_num)
        s->receiver_node = NULL;
        set_restart_barrier(s->format, NULL);
 }
+static void close_writer(int slot_num)
+{
+       struct slot_info *s = &slot[slot_num];
+       if (s->wng) {
+               PARA_INFO_LOG("slot %d: closing writer node group\n",
+                       slot_num);
+               wng_close(s->wng);
+               s->wng = NULL;
+       }
+       if (s->fc)
+               *s->fc->output_eof = 1; /* FIXME */
+}
+
 
 static void kill_all_decoders(void)
 {
        int i;
 
        FOR_EACH_SLOT(i)
-               if (slot[i].format >= 0) {
-                       PARA_INFO_LOG("stopping decoder in slot %d\n", i);
-                       kill_stream_writer(i);
-               }
+               close_writer(i);
 }
 
 static void check_sigchld(void)
 {
        pid_t pid;
-       int i;
        struct timeval now;
        gettimeofday(&now, NULL);
 
@@ -528,21 +526,6 @@ reap_next_child:
        pid = para_reap_child();
        if (pid <= 0)
                return;
-       FOR_EACH_SLOT(i) {
-               struct slot_info *s = &slot[i];
-               long lifetime;
-               if (s->format < 0)
-                       continue;
-               if (pid == s->wpid) {
-                       s->wpid = -1;
-                       lifetime = now.tv_sec - s->wstime.tv_sec;
-                       PARA_INFO_LOG("%s stream writer in slot %d died "
-                               "after %li secs\n",
-                               audio_formats[s->format], i, lifetime);
-                       set_restart_barrier(s->format, &now);
-                       goto reap_next_child;
-               }
-       }
        PARA_CRIT_LOG("para_client died (pid %d)\n", pid);
        goto reap_next_child;
 }
@@ -558,7 +541,7 @@ static int get_empty_slot(void)
                        clear_slot(i);
                        return i;
                }
-               if (s->write_fd > 0 || s->wpid > 0)
+               if (s->wng)
                        continue;
                if (s->receiver_node)
                        continue;
@@ -579,7 +562,7 @@ static int decoder_running(int format)
                s = &slot[i];
                if (s->format == format && s->receiver_node)
                        ret |= 1;
-               if (s->format == format && s->wpid > 0)
+               if (s->format == format && s->wng)
                        ret |= 2;
        }
        return ret;
@@ -621,6 +604,7 @@ static void __noreturn clean_exit(int status, const char *msg)
        exit(status);
 }
 
+#if 0
 __malloc static char *glob_cmd(char *cmd)
 {
        char *ret, *replacement;
@@ -645,6 +629,7 @@ __malloc static char *glob_cmd(char *cmd)
 out:
        return ret;
 }
+#endif
 
 /** get the number of filters for the given audio format */
 int num_filters(int audio_format_num)
@@ -664,17 +649,15 @@ static void open_filters(int slot_num)
        struct audio_format_info *a = &afi[s->format];
        int nf = a->num_filters;
        int i;
-       static int output_eof; /* FIXME */
 
-       s->fc = para_calloc(sizeof(struct filter_chain));
-       INIT_LIST_HEAD(&s->fc->filters);
+       s->fc = NULL;
        if (!nf)
                return;
+       s->fc = para_calloc(sizeof(struct filter_chain));
+       INIT_LIST_HEAD(&s->fc->filters);
        s->fc->inbuf = s->receiver_node->buf;
        s->fc->in_loaded = &s->receiver_node->loaded;
        s->fc->input_eof = &s->receiver_node->eof;
-       s->fc->output_eof = &output_eof;
-       output_eof = 0;
 
        s->fc->task.pre_select = filter_pre_select;
        s->fc->task.event_handler = filter_event_handler;
@@ -726,30 +709,53 @@ static struct filter_node *find_filter_node(int slot_num, int format, int filter
        return NULL;
 }
 
-static void start_stream_writer(int slot_num)
+static void wng_event_handler(struct task *t)
 {
-       int ret, fds[3] = {1, -1, -1};
+       struct writer_node_group *g = t->private_data;
+       int i;
+
+       PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret));
+       unregister_task(t);
+       FOR_EACH_SLOT(i) {
+               if (slot[i].wng != g)
+                       continue;
+               wng_close(g);
+               wng_destroy(g);
+               slot[i].wng = NULL;
+       }
+}
+
+static void start_stream_writer(int slot_num, struct timeval *now)
+{
+       int ret, i;
        struct slot_info *s = &slot[slot_num];
        struct audio_format_info *a = &afi[s->format];
-       char *glob = NULL;
 
-       if (a->write_cmd)
-               glob = glob_cmd(a->write_cmd);
-       if (!glob)
-               glob = para_strdup("para_write -w alsa");
-       PARA_INFO_LOG("starting stream writer: %s\n", glob);
+       PARA_INFO_LOG("opening %s filters\n", audio_formats[s->format]);
        open_filters(slot_num);
-       ret = para_exec_cmdline_pid(&s->wpid, glob, fds);
-       free(glob);
-       if (ret < 0) {
-               PARA_ERROR_LOG("exec failed (%d)\n", ret);
-               return;
+       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->buf = s->fc->outbuf;
+               s->wng->loaded = s->fc->out_loaded;
+               s->wng->input_eof = &s->fc->eof;
+               s->fc->output_eof = &s->wng->eof;
+       } else {
+               s->wng->buf = s->receiver_node->buf;
+               s->wng->loaded = &s->receiver_node->loaded;
+               s->wng->input_eof = &s->receiver_node->eof;
        }
-       s->write_fd = fds[0];
-       add_close_on_fork_list(s->write_fd);
-       /* we write to this fd in do_select, so we need non-blocking */
-       mark_fd_nonblock(s->write_fd);
-       gettimeofday(&s->wstime, NULL);
+       s->wng->task.event_handler = wng_event_handler;
+       for (i = 0; i < a->num_writers; i++) {
+               s->wng->writer_nodes[i].conf = a->writer_conf[i];
+               s->wng->writer_nodes[i].writer = a->writers[i];
+               sprintf(s->wng->writer_nodes[i].task.status, "writer_ node");
+       }
+       ret = wng_open(s->wng);
+       s->wstime = *now;
        current_decoder = slot_num;
        activate_inactive_grab_clients(slot_num, s->format, &s->fc->filters);
 }
@@ -936,71 +942,23 @@ static void check_timeouts(void)
                /* check read time */
                if (s->receiver_node &&
                        now.tv_sec > s->rtime.tv_sec + timeout) {
-                       PARA_INFO_LOG("%s input buffer (slot %d) not ready\n",
+                       PARA_INFO_LOG("%s stream (slot %d) not ready\n",
                                audio_formats[s->format], slot_num);
                        s->receiver_node->eof = 1;
                }
-               /* check write time */
-               if (s->wpid > 0 && !s->wkilled &&
-                       now.tv_sec > s->wtime.tv_sec + timeout) {
-                       PARA_INFO_LOG("%s output buffer (slot %d) not ready\n",
-                               audio_formats[s->format], slot_num);
-                       kill_stream_writer(slot_num);
-               }
        }
 }
 
-static size_t get_loaded_bytes(int slot_num)
-{
-       size_t loaded = 0;
-       struct slot_info *s = &slot[slot_num];
-       struct receiver_node *rn = s->receiver_node;
-
-       if (s->format < 0)
-               goto out;
-
-       if (afi[s->format].num_filters) {
-               if (s->fc)
-                       loaded = *s->fc->out_loaded;
-       } else {
-               if (rn)
-                       loaded = rn->loaded;
-       }
-out:
-       return loaded;
-}
-
-static void close_writer(int slot_num)
-{
-       struct slot_info *s = &slot[slot_num];
-       if (s->write_fd > 0) {
-               PARA_INFO_LOG("slot %d: closing write fd %d\n", slot_num,
-                       s->write_fd);
-               close(s->write_fd);
-               del_close_on_fork_list(s->write_fd);
-               s->write_fd = -1;
-       }
-       if (s->fc)
-               *s->fc->output_eof = 1; /* FIXME */
-}
-
-
 static void close_decoder_if_idle(int slot_num)
 {
        struct slot_info *s = &slot[slot_num];
-       struct receiver_node *rn = s->receiver_node;
 
        if (s->format < 0)
                return;
        if (!s->fc)
                return;
-       if (!rn->eof && !s->fc->eof && s->wpid > 0)
+       if (s->wng)
                return;
-       if (!s->fc->eof && s->wpid > 0 && get_loaded_bytes(slot_num))
-               return;
-       close_writer(slot_num);
-       if (s->wpid > 0)
-               return; /* wait until writer dies before closing filters */
        PARA_INFO_LOG("closing all filters in slot %d (filter_chain %p)\n",
                slot_num, s->fc);
        close_filters(s->fc);
@@ -1009,7 +967,7 @@ static void close_decoder_if_idle(int slot_num)
        clear_slot(slot_num);
 }
 
-static void audiod_pre_select(struct sched *s, struct task *t)
+static void audiod_pre_select(__a_unused struct sched *s, __a_unused struct task *t)
 {
        int i;
 
@@ -1019,28 +977,20 @@ static void audiod_pre_select(struct sched *s, struct task *t)
                start_current_receiver();
        check_timeouts();
        FOR_EACH_SLOT(i) {
-               struct audio_format_info *a;
                struct receiver_node *rn;
 
                close_decoder_if_idle(i);
-               slot[i].wcheck = 0;
                if (slot[i].format < 0)
                        continue;
-               a = &afi[slot[i].format];
                rn = slot[i].receiver_node;
-               if (rn && rn->loaded && !slot[i].wpid) {
+               if (rn && rn->loaded && !slot[i].wng) {
                        PARA_INFO_LOG("no writer in slot %d\n", i);
-                       start_stream_writer(i);
+                       start_stream_writer(i, &s->now);
                }
-               if (slot[i].write_fd <= 0)
-                       continue;
-               if (!get_loaded_bytes(i))
-                       continue;
-               para_fd_set(slot[i].write_fd, &s->wfds, &s->max_fileno);
-               slot[i].wcheck = 1;
        }
 }
 
+#if 0
 static int write_audio_data(int slot_num)
 {
        struct slot_info *s = &slot[slot_num];
@@ -1077,20 +1027,22 @@ static int write_audio_data(int slot_num)
                gettimeofday(&s->wtime, NULL);
        return ret;
 }
+#endif
 
-static void audiod_post_select(struct sched *s, struct task *t)
+
+static void audiod_post_select(struct sched *s, __a_unused struct task *t)
 {
-       int ret, i;
+       int i;
 
        FOR_EACH_SLOT(i) {
                struct receiver_node *rn = slot[i].receiver_node;
 
                if (rn && rn->loaded)
                        slot[i].rtime = s->now;
-               if (slot[i].write_fd <= 0 || !slot[i].wcheck
-                       || !FD_ISSET(slot[i].write_fd, &s->wfds))
-                       continue;
-               ret = write_audio_data(i);
+//             if (slot[i].write_fd <= 0 || !slot[i].wcheck
+//                     || !FD_ISSET(slot[i].write_fd, &s->wfds))
+//                     continue;
+//             ret = write_audio_data(i);
        }
 }
 
@@ -1172,15 +1124,37 @@ out:
 
 static int init_stream_io(void)
 {
-       int i, ret, receiver_num, nf;
+       int i, ret, receiver_num, nf, nw;
        char *cmd;
+       struct audio_format_info *a;
 
-       for (i = 0; i < conf.stream_write_cmd_given; i++) {
-               ret = parse_stream_command(conf.stream_write_cmd_arg[i], &cmd);
+       init_supported_writers();
+       nw = PARA_MAX(1, conf.writer_given);
+       PARA_INFO_LOG("allocating space for %d writers\n", nw);
+       FOR_EACH_AUDIO_FORMAT(i) {
+               a = &afi[i];
+               a->writer_conf = para_malloc(nw * sizeof(void *));
+               a->writers = para_malloc(nw * sizeof(struct writer *));
+               a->num_writers = 0;
+       }
+       for (i = 0; i < conf.writer_given; i++) {
+               void *wconf;
+               int writer_num;
+               ret = parse_stream_command(conf.writer_arg[i], &cmd);
                if (ret < 0)
                        goto out;
-               afi[ret].write_cmd = para_strdup(cmd);
-               PARA_INFO_LOG("%s write command: %s\n", audio_formats[ret], afi[ret].write_cmd);
+               a = &afi[ret];
+               nw = a->num_writers;
+               wconf = check_writer_arg(cmd, &writer_num);
+               if (!wconf) {
+                       ret = writer_num;
+                       goto out;
+               }
+               a->writers[nw] = &writers[ret];
+               a->writer_conf[nw] = wconf;
+               PARA_INFO_LOG("%s writer #%d: %s\n", audio_formats[ret],
+                       nw, writer_names[writer_num]);
+               a->num_writers++;
        }
        for (i = 0; receivers[i].name; i++) {
                PARA_INFO_LOG("initializing %s receiver\n", receivers[i].name);
@@ -1210,7 +1184,7 @@ static int init_stream_io(void)
         */
        cmd = para_strdup(receivers[0].name);
        FOR_EACH_AUDIO_FORMAT(i) {
-               struct audio_format_info *a = &afi[i];
+               a = &afi[i];
                if (a->receiver_conf)
                        continue;
                a->receiver_conf = check_receiver_arg(cmd, &receiver_num);
@@ -1224,7 +1198,7 @@ static int init_stream_io(void)
        nf = PARA_MAX(2,  conf.filter_given) + 1;
        PARA_INFO_LOG("allocating space for %d filters\n", nf);
        FOR_EACH_AUDIO_FORMAT(i) {
-               afi[i].filter_conf = para_malloc(nf * sizeof(char *));
+               afi[i].filter_conf = para_malloc(nf * sizeof(void *));
                afi[i].filters = para_malloc(nf * sizeof(struct filter *));
        }
        if (!conf.no_default_filters_given)
@@ -1825,6 +1799,7 @@ int main(int argc, char *argv[])
                PARA_EMERG_LOG("init stream io error: %s\n", PARA_STRERROR(-i));
                exit(EXIT_FAILURE);
        }
+//     return 0;
        server_uptime(UPTIME_SET);
        set_initial_status();
        FOR_EACH_SLOT(i)

diff --git a/audiod.ggo b/audiod.ggo
index dd385a16c09d0ecc070d3ef9af396256fe70c963..89cad59c69e1045fcbf21d682ecd787053805159 100644 (file)
--- a/audiod.ggo
+++ b/audiod.ggo
@@ -143,21 +143,26 @@ see --no_default_filters."
 string typestr="filter_spec" optional multiple
 
 
-option "stream_write_cmd" w
+option "writer" w
 #~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 "Specify stream writer.
 
-May be given multiple times, once for each
-supported audio format. Default value is
-'para_write -w alsa' for both mp3 and ogg.
-You can use the START_TIME() macro for these
-commands.  Each occurence of START_TIME()
-gets replaced at runtime by the stream start
-time announced by para_server, plus any
-offsets."
+May be given multiple times, even multiple
+times for the same audio format.  Default
+value is 'alsa' for all supported audio
+formats.  You can use the START_TIME() macro
+for these commands.  Each occurence of
+START_TIME() gets replaced at runtime by the
+stream start time announced by para_server,
+plus any offsets.
 
-       string typestr="format:command"
+Example:
+
+       -w 'aac:osx'
+
+"
+       string typestr="writer_spec"
        optional
        multiple
 

diff --git a/stdout.h b/stdout.h
index 5e45ace8dc20fc88e30d2305ee73b438fe65471e..4dafafdec596d3ff60fd70dbc2291482165f71d0 100644 (file)
--- a/stdout.h
+++ b/stdout.h
@@ -1,3 +1,4 @@
+/** \file stdout.h common code for uitlities that write to stdout */
 struct stdout_task {
        char *buf;
        size_t *bufsize;

diff --git a/write.c b/write.c
index b692ac5080af8dfee4efd79a95048e6c8ad6999d..2efe07ce730264be610fc35adc2bc1cdd9fc0721 100644 (file)
--- a/write.c
+++ b/write.c
@@ -161,6 +161,17 @@ out:
        return NULL;
 }
 
+static void wng_event_handler(struct task *t)
+{
+       struct writer_node_group *g = t->private_data;
+
+       PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret));
+       unregister_task(t);
+       wng_close(g);
+       wng_destroy(g);
+}
+
+
 static void idt_event_handler(struct task *t)
 {
        int ret;
@@ -170,6 +181,7 @@ static void idt_event_handler(struct task *t)
        wng->buf = sit.buf;
        wng->loaded = &sit.loaded;
        wng->input_eof = &sit.eof;
+       wng->task.event_handler = wng_event_handler;
        ret = wng_open(wng);
        if (ret < 0) {
                PARA_ERROR_LOG("%s\n", PARA_STRERROR(-ret));

diff --git a/write_common.c b/write_common.c
index de0502f52c5ef3dd2cbb614857eb03c8c87e3470..7fb8d12337b6fdc28a1af741e5a0cd6a74f21ea4 100644 (file)
--- a/write_common.c
+++ b/write_common.c
@@ -99,16 +99,6 @@ void wng_close(struct writer_node_group *g)
        }
 }
 
-static void wng_event_handler(struct task *t)
-{
-       struct writer_node_group *g = t->private_data;
-
-       PARA_INFO_LOG("%s\n", PARA_STRERROR(-t->ret));
-       unregister_task(t);
-       wng_close(g);
-       wng_destroy(g);
-}
-
 struct writer_node_group *wng_new(unsigned num_writers)
 {
        struct writer_node_group *g = para_calloc(sizeof(struct writer_node_group));
@@ -118,7 +108,6 @@ struct writer_node_group *wng_new(unsigned num_writers)
        g->written = para_calloc(num_writers * sizeof(size_t));
        g->task.private_data = g;
        g->task.post_select = wng_post_select;
-       g->task.event_handler = wng_event_handler;
        g->task.flags = POST_ADD_TAIL;
        return g;
 }
@@ -135,6 +124,7 @@ void *check_writer_arg(char *wa, int *writer_num)
 {
        int i;
 
+       *writer_num = -E_WRITE_COMMON_SYNTAX;
        PARA_INFO_LOG("checking  %s\n", wa);
        FOR_EACH_WRITER(i) {
                const char *name = writer_names[i];