write/alsa: Add btr support.

Not yet finished.

Supporting btr required some changes in the buffer tree code
as well.

diff --git a/alsa_write.c b/alsa_write.c
index 9d99c8e91af6c4c78e402f9efde4ce00e40cb785..030a5ec60e740ce7a3fe11ad82e72f50d9de4095 100644 (file)
--- a/alsa_write.c
+++ b/alsa_write.c
@@ -17,6 +17,7 @@
 #include <dirent.h>
 #include <alsa/asoundlib.h>
 #include <sys/time.h>
 #include <dirent.h>
 #include <alsa/asoundlib.h>
 #include <sys/time.h>

+#include <stdbool.h>

 
 #include "para.h"
 #include "fd.h"
 
 #include "para.h"
 #include "fd.h"


@@ -27,6 +28,7 @@
 #include "write.h"
 #include "alsa_write.cmdline.h"
 #include "error.h"
 #include "write.h"
 #include "alsa_write.cmdline.h"
 #include "error.h"

+#include "buffer_tree.h"

 
 /** always use 16 bit little endian */
 #define FORMAT SND_PCM_FORMAT_S16_LE
 
 /** always use 16 bit little endian */
 #define FORMAT SND_PCM_FORMAT_S16_LE


@@ -128,7 +130,7 @@ static int alsa_init(struct private_alsa_write_data *pad,
 }
 
 /* Open an instance of the alsa writer. */
 }
 
 /* Open an instance of the alsa writer. */

-static int alsa_open(struct writer_node *wn)
+static int alsa_open_nobtr(struct writer_node *wn)

 {
        struct alsa_write_args_info *conf = wn->conf;
        struct writer_node_group *wng = wn->wng;
 {
        struct alsa_write_args_info *conf = wn->conf;
        struct writer_node_group *wng = wn->wng;


@@ -147,8 +149,66 @@ static int alsa_open(struct writer_node *wn)
        return 1;
 }
 
        return 1;
 }
 

+static int alsa_open_btr(struct writer_node *wn)
+{
+       struct alsa_write_args_info *conf = wn->conf;
+       struct private_alsa_write_data *pad = para_calloc(sizeof(*pad));
+       int ret;
+       char *buf = NULL;
+
+       sprintf(wn->task.status, "alsa writer");
+       wn->private_data = pad;
+
+       /* defaults */
+       pad->samplerate = conf->samplerate_arg;
+       pad->channels = conf->channels_arg;
+
+       if (!conf->samplerate_given) { /* config option trumps btr_exec */
+               /* ask parent btr nodes */
+               ret = btr_exec_up(wn->btrn, "samplerate", &buf);
+               if (ret >= 0) {
+                       int32_t rate;
+                       ret = para_atoi32(buf, &rate);
+                       if (ret < 0) /* should not happen */
+                               goto out;
+                       pad->samplerate = rate;
+               }
+               freep(&buf);
+       }
+
+       if (!conf->channels_given) {
+               ret = btr_exec_up(wn->btrn, "channels", &buf);
+               if (ret >= 0) {
+                       int32_t ch;
+                       ret = para_atoi32(buf, &ch);
+                       if (ret < 0)
+                               goto out;
+                       pad->channels = ch;
+               }
+               freep(&buf);
+       }
+       PARA_INFO_LOG("%d channel(s), %dHz\n", pad->channels, pad->samplerate);
+       ret = 1;
+out:
+       freep(&buf);
+       if (ret < 0)
+               free(pad);
+       return ret;
+}
+static int alsa_open(struct writer_node *wn)
+{
+       struct alsa_write_args_info *conf = wn->conf;
+
+       if (conf->buffer_tree_given)
+               return alsa_open_btr(wn);
+       else
+               return alsa_open_nobtr(wn);
+
+}
+

 static int alsa_write_pre_select(struct sched *s, struct writer_node *wn)
 {
 static int alsa_write_pre_select(struct sched *s, struct writer_node *wn)
 {

+       struct alsa_write_args_info *conf = wn->conf;

        struct private_alsa_write_data *pad = wn->private_data;
        struct writer_node_group *wng = wn->wng;
        struct timeval tv;
        struct private_alsa_write_data *pad = wn->private_data;
        struct writer_node_group *wng = wn->wng;
        struct timeval tv;


@@ -156,8 +216,15 @@ static int alsa_write_pre_select(struct sched *s, struct writer_node *wn)
 
        if (!pad->handle)
                return 1;
 
        if (!pad->handle)
                return 1;

-       if (*wng->loaded - wn->written < pad->bytes_per_frame)
-               return 1;
+       if (conf->buffer_tree_given) {
+               size_t sz = btr_get_input_queue_size(wn->btrn);
+               //PARA_CRIT_LOG("input queue: %zu\n", sz);
+               //if (sz < pad->bytes_per_frame)
+               //      return 1;
+       } else {
+               if (*wng->loaded - wn->written < pad->bytes_per_frame)
+                       return 1;
+       }

        /*
         * Data is available to be written to the alsa handle.  Compute number
         * of milliseconds until next buffer underrun would occur.
        /*
         * Data is available to be written to the alsa handle.  Compute number
         * of milliseconds until next buffer underrun would occur.


@@ -179,9 +246,16 @@ static int alsa_write_pre_select(struct sched *s, struct writer_node *wn)
        ms2tv(underrun, &tv);
        if (tv_diff(&s->timeout, &tv, NULL) > 0)
                s->timeout = tv;
        ms2tv(underrun, &tv);
        if (tv_diff(&s->timeout, &tv, NULL) > 0)
                s->timeout = tv;

+       //PARA_CRIT_LOG("timout: %lu\n", tv2ms(&s->timeout));

        return 1;
 }
 
        return 1;
 }
 

+static void alsa_write_pre_select_btr(struct sched *s, struct task *t)
+{
+       struct writer_node *wn = container_of(t, struct writer_node, task);
+       t->error = alsa_write_pre_select(s, wn);
+}
+

 static void xrun(snd_pcm_t *handle)
 {
        snd_pcm_status_t *status;
 static void xrun(snd_pcm_t *handle)
 {
        snd_pcm_status_t *status;


@@ -242,6 +316,68 @@ static int alsa_write_post_select(__a_unused struct sched *s,
        return -E_ALSA_WRITE;
 }
 
        return -E_ALSA_WRITE;
 }
 

+static void alsa_write_post_select_btr(__a_unused struct sched *s,
+               struct task *t)
+{
+       struct writer_node *wn = container_of(t, struct writer_node, task);
+       struct private_alsa_write_data *pad = wn->private_data;
+       char *data;
+       size_t bytes;
+       snd_pcm_sframes_t frames, avail;
+       int ret;
+
+again:
+       bytes = btr_next_buffer(wn->btrn, &data);
+       //PARA_CRIT_LOG("have: %zu\n", bytes);
+       t->error = 0;
+       ret = -E_ALSA_ORPHAN;
+       if (btr_no_parent(wn->btrn) && (!pad->handle || bytes < pad->bytes_per_frame))
+               goto err;
+       if (!pad->handle) {
+               if (bytes == 0) /* no data available */
+                       return;
+               PARA_CRIT_LOG("alsa init\n");
+               ret = alsa_init(pad, wn->conf);
+               if (ret < 0)
+                       goto err;
+       }
+       for (;;) {
+               if (bytes == 0)
+                       return;
+               if (bytes >= pad->bytes_per_frame)
+                       break;
+               /* should not be possible to reach this */
+               PARA_CRIT_LOG("dropping %zu byte buffer\n", bytes);
+               btr_consume(wn->btrn, bytes);
+               bytes = btr_next_buffer(wn->btrn, &data);
+       }
+       frames = bytes / pad->bytes_per_frame;
+       avail = snd_pcm_avail_update(pad->handle);
+       if (avail <= 0)
+               return;
+       frames = PARA_MIN(frames, avail);
+       //PARA_CRIT_LOG("writing %ld frames\n", frames);
+       frames = snd_pcm_writei(pad->handle, data, frames);
+       if (frames >= 0) {
+               btr_consume(wn->btrn, frames * pad->bytes_per_frame);
+               goto again;
+       }
+       if (frames == -EPIPE) {
+               xrun(pad->handle);
+               snd_pcm_prepare(pad->handle);
+               return;
+       }
+       PARA_WARNING_LOG("%s\n", snd_strerror(-frames));
+       if (frames == -EAGAIN)
+               return;
+       ret = -E_ALSA_WRITE;
+err:
+       assert(ret < 0);
+       btr_del_node(wn->btrn);
+       wn->btrn = NULL;
+       t->error = ret;
+}
+

 static void alsa_close(struct writer_node *wn)
 {
        struct private_alsa_write_data *pad = wn->private_data;
 static void alsa_close(struct writer_node *wn)
 {
        struct private_alsa_write_data *pad = wn->private_data;


@@ -287,7 +423,9 @@ void alsa_write_init(struct writer *w)
        w->open = alsa_open;
        w->close = alsa_close;
        w->pre_select = alsa_write_pre_select;
        w->open = alsa_open;
        w->close = alsa_close;
        w->pre_select = alsa_write_pre_select;

+       w->pre_select_btr = alsa_write_pre_select_btr;

        w->post_select = alsa_write_post_select;
        w->post_select = alsa_write_post_select;

+       w->post_select_btr = alsa_write_post_select_btr;

        w->parse_config = alsa_parse_config;
        w->shutdown = NULL; /* nothing to do */
        w->help = (struct ggo_help) {
        w->parse_config = alsa_parse_config;
        w->shutdown = NULL; /* nothing to do */
        w->help = (struct ggo_help) {

diff --git a/buffer_tree.c b/buffer_tree.c
index 1ca4a1536cdd452116303bbf43f7f24de4ef9c42..2a52f097a325641c5695ae841eaa4608263d3d99 100644 (file)
--- a/buffer_tree.c
+++ b/buffer_tree.c
@@ -5,6 +5,7 @@
 #include "list.h"
 #include "string.h"
 #include "buffer_tree.h"
 #include "list.h"
 #include "string.h"
 #include "buffer_tree.h"

+#include "error.h"

 
 
 struct btr_buffer {
 
 
 struct btr_buffer {


@@ -34,6 +35,7 @@ struct btr_node {
         * used by this btr node.
         */
        struct list_head input_queue;
         * used by this btr node.
         */
        struct list_head input_queue;

+       btr_command_handler execute;

 };
 
 #define FOR_EACH_CHILD(_tn, _btrn) list_for_each_entry((_tn), \
 };
 
 #define FOR_EACH_CHILD(_tn, _btrn) list_for_each_entry((_tn), \


@@ -44,12 +46,14 @@ struct btr_node {
 #define FOR_EACH_BUFFER_REF_SAFE(_br, _tmp, _btrn) \
        list_for_each_entry_safe((_br), (_tmp), &(_btrn)->input_queue, node)
 
 #define FOR_EACH_BUFFER_REF_SAFE(_br, _tmp, _btrn) \
        list_for_each_entry_safe((_br), (_tmp), &(_btrn)->input_queue, node)
 

-struct btr_node *btr_new_node(char *name, struct btr_node *parent)
+struct btr_node *btr_new_node(char *name, struct btr_node *parent,
+               btr_command_handler handler)

 {
        struct btr_node *btrn = para_malloc(sizeof(*btrn));
 
        btrn->name = para_strdup(name);
        btrn->parent = parent;
 {
        struct btr_node *btrn = para_malloc(sizeof(*btrn));
 
        btrn->name = para_strdup(name);
        btrn->parent = parent;

+       btrn->execute = handler;

        if (parent)
                list_add_tail(&btrn->node, &parent->children);
        INIT_LIST_HEAD(&btrn->children);
        if (parent)
                list_add_tail(&btrn->node, &parent->children);
        INIT_LIST_HEAD(&btrn->children);


@@ -79,6 +83,7 @@ static void btr_drop_buffer_reference(struct btr_buffer_reference *br)
 {
        struct btr_buffer *btrb = br->btrb;
 
 {
        struct btr_buffer *btrb = br->btrb;
 

+       //PARA_CRIT_LOG("dropping buffer reference %p\n", br);

        list_del(&br->node);
        free(br);
        btrb->refcount--;
        list_del(&br->node);
        free(br);
        btrb->refcount--;


@@ -204,8 +209,10 @@ size_t btr_get_input_queue_size(struct btr_node *btrn)
        struct btr_buffer_reference *br;
        size_t size = 0;
 
        struct btr_buffer_reference *br;
        size_t size = 0;
 

-       FOR_EACH_BUFFER_REF(br, btrn)
+       FOR_EACH_BUFFER_REF(br, btrn) {
+               //PARA_CRIT_LOG("size: %zu\n", size);

                size += br_available_bytes(br);
                size += br_available_bytes(br);

+       }

        return size;
 }
 
        return size;
 }
 


@@ -225,3 +232,32 @@ size_t btr_bytes_pending(struct btr_node *btrn)
        }
        return max_size;
 }
        }
        return max_size;
 }

+
+int btr_exec(struct btr_node *btrn, const char *command, char **value_result)
+{
+       if (!btrn)
+               return -ERRNO_TO_PARA_ERROR(EINVAL);
+       if (!btrn->execute)
+               return -ERRNO_TO_PARA_ERROR(ENOTSUP);
+       return btrn->execute(command, value_result);
+}
+
+int btr_exec_up(struct btr_node *btrn, const char *command, char **value_result)
+{
+       int ret;
+
+       for (; btrn; btrn = btrn->parent) {
+               struct btr_node *parent = btrn->parent;
+               PARA_CRIT_LOG("parent: %p\n", parent);
+               if (!parent)
+                       return -ERRNO_TO_PARA_ERROR(ENOTSUP);
+               if (!parent->execute)
+                       continue;
+               ret = parent->execute(command, value_result);
+               if (ret == -ERRNO_TO_PARA_ERROR(ENOTSUP))
+                       continue;
+               if (ret < 0)
+                       return ret;
+       }
+       return -ERRNO_TO_PARA_ERROR(ENOTSUP);
+}

diff --git a/buffer_tree.h b/buffer_tree.h
index 04217801f70b934eb25987aeb4237e919d13bc4c..c84a93058e575ed6a8fdbc79c49160b2d0271985 100644 (file)
--- a/buffer_tree.h
+++ b/buffer_tree.h
@@ -1,7 +1,10 @@
 
 struct btr_node;
 
 
 struct btr_node;
 

-struct btr_node *btr_new_node(char *name, struct btr_node *parent);
+typedef int (*btr_command_handler)(const char *command, char **result);
+
+struct btr_node *btr_new_node(char *name, struct btr_node *parent,
+               btr_command_handler handler);

 void btr_del_node(struct btr_node *btrn);
 void btr_add_output(char *buf, size_t size, struct btr_node *btrn);
 bool btr_no_children(struct btr_node *btrn);
 void btr_del_node(struct btr_node *btrn);
 void btr_add_output(char *buf, size_t size, struct btr_node *btrn);
 bool btr_no_children(struct btr_node *btrn);


@@ -10,3 +13,5 @@ size_t btr_get_input_queue_size(struct btr_node *btrn);
 bool btr_no_parent(struct btr_node *btrn);
 size_t btr_next_buffer(struct btr_node *btrn, char **bufp);
 void btr_consume(struct btr_node *btrn, size_t numbytes);
 bool btr_no_parent(struct btr_node *btrn);
 size_t btr_next_buffer(struct btr_node *btrn, char **bufp);
 void btr_consume(struct btr_node *btrn, size_t numbytes);

+int btr_exec(struct btr_node *btrn, const char *command, char **value_result);
+int btr_exec_up(struct btr_node *btrn, const char *command, char **value_result);

diff --git a/error.h b/error.h
index 9ef417ff7ee05ad6bd42cad23e1c85e8c1e45c33..f5fdea64ccc6f172f91061f605304bb2b5d75b11 100644 (file)
--- a/error.h
+++ b/error.h
@@ -404,6 +404,7 @@ extern const char **para_errlist[];
        PARA_ERROR(SET_RATE, "snd_pcm_hw_params_set_rate_near failed"), \
        PARA_ERROR(START_THRESHOLD, "snd_pcm_sw_params_set_start_threshold() failed"), \
        PARA_ERROR(STOP_THRESHOLD, "snd_pcm_sw_params_set_stop_threshold() failed"), \
        PARA_ERROR(SET_RATE, "snd_pcm_hw_params_set_rate_near failed"), \
        PARA_ERROR(START_THRESHOLD, "snd_pcm_sw_params_set_start_threshold() failed"), \
        PARA_ERROR(STOP_THRESHOLD, "snd_pcm_sw_params_set_stop_threshold() failed"), \

+       PARA_ERROR(ALSA_ORPHAN, "alsa btr node has no parent"), \

 
 
 #define FILE_WRITE_ERRORS \
 
 
 #define FILE_WRITE_ERRORS \

diff --git a/ggo/buffer_tree.m4 b/ggo/buffer_tree.m4
new file mode 100644 (file)
index 0000000..7cf6fbe
--- /dev/null
+++ b/ggo/buffer_tree.m4
@@ -0,0 +1,9 @@
+<qu>
+option "buffer_tree" B
+#~~~~~~~~~~~~~~~~~~~~~
+"use the new buffer tree API"
+flag off
+details = "
+       Experimental. Don't use.
+"
+</qu>

diff --git a/http_recv.c b/http_recv.c
index a6d792b7c6a8818759febf796287daf0680f6956..59f00d831c6bf3f03747a808599b929a66e7edc3 100644 (file)
--- a/http_recv.c
+++ b/http_recv.c
@@ -215,7 +215,7 @@ static int http_recv_open(struct receiver_node *rn)
        phd->fd = fd;
        phd->status = HTTP_CONNECTED;
        if (conf->buffer_tree_given)
        phd->fd = fd;
        phd->status = HTTP_CONNECTED;
        if (conf->buffer_tree_given)

-               rn->btrn = btr_new_node("receiver", NULL);
+               rn->btrn = btr_new_node("receiver", NULL, NULL);

        return 1;
 }
 
        return 1;
 }
 

diff --git a/recv.c b/recv.c
index 297fbe92be868d477fc32ed8c8dc480342c1d325..57931943b4a02bdcaa501d0a149388c7d1874a0c 100644 (file)
--- a/recv.c
+++ b/recv.c
@@ -98,7 +98,7 @@ int main(int argc, char *argv[])
        r_opened = 1;
 
        if (conf.buffer_tree_given)
        r_opened = 1;
 
        if (conf.buffer_tree_given)

-               sot.btrn = btr_new_node("stdout", rn.btrn);
+               sot.btrn = btr_new_node("stdout", rn.btrn, NULL);

 
        stdout_set_defaults(&sot);
        sot.bufp = &rn.buf;
 
        stdout_set_defaults(&sot);
        sot.bufp = &rn.buf;

diff --git a/stdin.c b/stdin.c
index d080e2165b169ec3fd03bf0c669ac6ec05188985..836369992f01b7f7a62c1b33c5674a13ba6ee89c 100644 (file)
--- a/stdin.c
+++ b/stdin.c
@@ -46,7 +46,7 @@ static void stdin_pre_select(struct sched *s, struct task *t)
        para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno);
 }
 
        para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno);
 }
 

-#define STDIN_MAX_PENDING (1024 * 1024)
+#define STDIN_MAX_PENDING (100 * 1024)

 
 static void stdin_pre_select_btr(struct sched *s, struct task *t)
 {
 
 static void stdin_pre_select_btr(struct sched *s, struct task *t)
 {


@@ -120,6 +120,7 @@ static void stdin_post_select_btr(struct sched *s, struct task *t)
 
        buf = para_malloc(STDIN_INPUT_BUFFER_SIZE);
        ret = read(STDIN_FILENO, buf, STDIN_INPUT_BUFFER_SIZE);
 
        buf = para_malloc(STDIN_INPUT_BUFFER_SIZE);
        ret = read(STDIN_FILENO, buf, STDIN_INPUT_BUFFER_SIZE);

+       //PARA_CRIT_LOG("read ret: %d\n", ret);

        if (ret < 0)
                t->error = -ERRNO_TO_PARA_ERROR(errno);
        if (ret == 0)
        if (ret < 0)
                t->error = -ERRNO_TO_PARA_ERROR(errno);
        if (ret == 0)

diff --git a/write.c b/write.c
index 34cea48c6a81ab2d7d6743e28706181415362824..21a0dad9253e66877ada6da5814062764d1269fb 100644 (file)
--- a/write.c
+++ b/write.c
@@ -9,6 +9,7 @@
 #include <regex.h>
 #include <sys/types.h>
 #include <dirent.h>
 #include <regex.h>
 #include <sys/types.h>
 #include <dirent.h>

+#include <stdbool.h>

 
 #include "para.h"
 #include "string.h"
 
 #include "para.h"
 #include "string.h"


@@ -21,6 +22,7 @@
 #include "write_common.h"
 #include "fd.h"
 #include "error.h"
 #include "write_common.h"
 #include "fd.h"
 #include "error.h"

+#include "buffer_tree.h"

 
 INIT_WRITE_ERRLISTS;
 
 
 INIT_WRITE_ERRLISTS;
 


@@ -179,6 +181,36 @@ __noreturn static void print_help_and_die(void)
        exit(0);
 }
 
        exit(0);
 }
 

+/*
+ TODO: check wav, initial delay, multiple writers, non-default writers
+ */
+static int main_btr(struct sched *s)
+{
+       struct writer_node *wn = para_malloc(sizeof(*wn));
+       struct writer *w = writers + DEFAULT_WRITER;
+       int ret;
+
+       wn->writer_num = DEFAULT_WRITER;
+       wn->conf = writers[DEFAULT_WRITER].parse_config("-B");
+       sit.btrn = btr_new_node("stdin", NULL /* stdin has no parent */, NULL);
+       stdin_set_defaults(&sit);
+       register_task(&sit.task);
+
+       wn->btrn = btr_new_node("writer", sit.btrn, NULL);
+
+       sprintf(wn->task.status, "some writer");
+       w->open(wn);
+       wn->task.post_select = w->post_select_btr;
+       wn->task.pre_select = w->pre_select_btr;
+       register_task(&wn->task);
+
+       s->default_timeout.tv_sec = 10;
+       s->default_timeout.tv_usec = 50000;
+       ret = schedule(s);
+       w->close(wn);
+       return ret;
+}
+

 /**
  * Para_write's main function.
  *
 /**
  * Para_write's main function.
  *


@@ -203,6 +235,10 @@ int main(int argc, char *argv[])
        if (conf.help_given || conf.detailed_help_given)
                print_help_and_die();
 
        if (conf.help_given || conf.detailed_help_given)
                print_help_and_die();
 

+       if (conf.buffer_tree_given) {
+               ret = main_btr(&s);
+               goto out;
+       }

        wng = check_args();
        if (!wng)
                goto out;
        wng = check_args();
        if (!wng)
                goto out;

diff --git a/write.h b/write.h
index 8816be7314bcd14516afc4e91226f3f3124df6c1..b47784af8785f2df6481cd1d3c23f517d6697adc 100644 (file)
--- a/write.h
+++ b/write.h
@@ -23,6 +23,8 @@ struct writer_node {
        void *conf;
        /** How much of the wng's buffer is already written. */
        size_t written;
        void *conf;
        /** How much of the wng's buffer is already written. */
        size_t written;

+       struct btr_node *btrn;
+       struct task task; /* move this to btr as btr nodes do not make sense w/o task */

 };
 
 /** Describes one supported writer. */
 };
 
 /** Describes one supported writer. */


@@ -60,6 +62,7 @@ struct writer {
         * on errors.
         */
        int (*pre_select)(struct sched *s, struct writer_node *wn);
         * on errors.
         */
        int (*pre_select)(struct sched *s, struct writer_node *wn);

+       void (*pre_select_btr)(struct sched *s, struct task *t);

        /**
         * Write audio data.
         *
        /**
         * Write audio data.
         *


@@ -70,6 +73,7 @@ struct writer {
         * negative on errors.
         */
        int (*post_select)(struct sched *s, struct writer_node *wn);
         * negative on errors.
         */
        int (*post_select)(struct sched *s, struct writer_node *wn);

+       void (*post_select_btr)(struct sched *s, struct task *t);

        /**
         * Close one instance of the writer.
         *
        /**
         * Close one instance of the writer.
         *