afh: Dynamic chunks.

[paraslash.git] / afh_recv.c

diff --git a/afh_recv.c b/afh_recv.c
index febda8a99a26acc5894ba21c255fc9d2a8a8958d..08f0d1e7c4c615ac96b8ddc27a39af41f5fa1625 100644 (file)
--- a/afh_recv.c
+++ b/afh_recv.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2011-2013 Andre Noll <maan@systemlinux.org>
+ * Copyright (C) 2011 Andre Noll <maan@tuebingen.mpg.de>
  *
  * Licensed under the GPL v2. For licencing details see COPYING.
  */
@@ -8,7 +8,6 @@
 
 #include <regex.h>
 #include <sys/types.h>
-#include <stdbool.h>
 
 #include "para.h"
 #include "error.h"
@@ -32,6 +31,7 @@ struct private_afh_recv_data {
        long unsigned last_chunk;
        struct timeval stream_start;
        uint32_t current_chunk;
+       void *afh_context;
 };
 
 static int afh_execute(struct btr_node *btrn, const char *cmd, char **result)
@@ -41,11 +41,11 @@ static int afh_execute(struct btr_node *btrn, const char *cmd, char **result)
 
        *result = NULL;
        if (!strcmp(cmd, "seconds_total")) {
-               *result = make_message("%lu", pard->afhi.seconds_total);
+               *result = make_message("%" PRIu32, pard->afhi.seconds_total);
                return 1;
        }
        if (!strcmp(cmd, "chunks_total")) {
-               *result = make_message("%lu", pard->afhi.chunks_total);
+               *result = make_message("%" PRIu32, pard->afhi.chunks_total);
                return 1;
        }
        if (!strcmp(cmd, "afhi")) {
@@ -59,8 +59,9 @@ static int afh_execute(struct btr_node *btrn, const char *cmd, char **result)
                        return ret;
                if (x >= pard->afhi.chunks_total)
                        return -ERRNO_TO_PARA_ERROR(EINVAL);
-               pard->first_chunk = pard->current_chunk = x;
-               rn->task.error = 0;
+               pard->first_chunk = afh_get_start_chunk(x, &pard->afhi,
+                       pard->audio_format_num);
+               pard->current_chunk = pard->first_chunk;
                return 1;
        }
        return -E_BTR_NAVAIL;
@@ -70,10 +71,8 @@ static void *afh_recv_parse_config(int argc, char **argv)
 {
        struct afh_recv_args_info *tmp = para_calloc(sizeof(*tmp));
 
-       if (!afh_recv_cmdline_parser(argc, argv, tmp))
-               return tmp;
-       free(tmp);
-       return NULL;
+       afh_recv_cmdline_parser(argc, argv, tmp);
+       return tmp;
 }
 
 static void afh_recv_free_config(void *conf)
@@ -112,9 +111,13 @@ static int afh_recv_open(struct receiver_node *rn)
        if (PARA_ABS(conf->begin_chunk_arg) >= afhi->chunks_total)
                goto out_clear_afhi;
        if (conf->begin_chunk_arg >= 0)
-               pard->first_chunk = conf->begin_chunk_arg;
+               pard->first_chunk = afh_get_start_chunk(
+                       conf->begin_chunk_arg, &pard->afhi,
+                       pard->audio_format_num);
        else
-               pard->first_chunk = afhi->chunks_total + conf->begin_chunk_arg;
+               pard->first_chunk = afh_get_start_chunk(
+                       afhi->chunks_total + conf->begin_chunk_arg,
+                       &pard->afhi, pard->audio_format_num);
        if (conf->end_chunk_given) {
                ret = -ERRNO_TO_PARA_ERROR(EINVAL);
                if (PARA_ABS(conf->end_chunk_arg) > afhi->chunks_total)
@@ -150,17 +153,18 @@ static void afh_recv_close(struct receiver_node *rn)
        clear_afhi(&pard->afhi);
        para_munmap(pard->map, pard->map_size);
        close(pard->fd);
+       afh_close(pard->afh_context, pard->audio_format_num);
        freep(&rn->private_data);
 }
 
-static void afh_recv_pre_select(struct sched *s, struct task *t)
+static void afh_recv_pre_select(struct sched *s, void *context)
 {
-       struct receiver_node *rn = container_of(t, struct receiver_node, task);
+       struct receiver_node *rn = context;
        struct private_afh_recv_data *pard = rn->private_data;
        struct afh_info *afhi = &pard->afhi;
        struct afh_recv_args_info *conf = rn->conf;
        struct timeval chunk_time;
-       int state = generic_recv_pre_select(s, t);
+       int state = generic_recv_pre_select(s, rn);
 
        if (state <= 0)
                return;
@@ -173,16 +177,16 @@ static void afh_recv_pre_select(struct sched *s, struct task *t)
        sched_request_barrier_or_min_delay(&chunk_time, s);
 }
 
-static void afh_recv_post_select(__a_unused struct sched *s, struct task *t)
+static int afh_recv_post_select(__a_unused struct sched *s, void *context)
 {
-       struct receiver_node *rn = container_of(t, struct receiver_node, task);
+       struct receiver_node *rn = context;
        struct afh_recv_args_info *conf = rn->conf;
        struct private_afh_recv_data *pard = rn->private_data;
        struct btr_node *btrn = rn->btrn;
        struct afh_info *afhi = &pard->afhi;
        int ret;
        char *buf;
-       const char *start, *end;
+       const char *start;
        size_t size;
        struct timeval chunk_time;
 
@@ -202,11 +206,16 @@ static void afh_recv_post_select(__a_unused struct sched *s, struct task *t)
                }
        }
        if (!conf->just_in_time_given) {
-               afh_get_chunk(pard->first_chunk, afhi, pard->map, &start, &size);
-               afh_get_chunk(pard->last_chunk, afhi, pard->map, &end, &size);
-               end += size;
-               PARA_INFO_LOG("adding %zu bytes\n", end - start);
-               btr_add_output_dont_free(start, end - start, btrn);
+               long unsigned n;
+               for (n = pard->first_chunk; n < pard->last_chunk; n++) {
+                       ret = afh_get_chunk(n, afhi, pard->audio_format_num,
+                               pard->map, pard->map_size, &start, &size,
+                               &pard->afh_context);
+                       if (ret < 0)
+                               goto out;
+                       PARA_INFO_LOG("adding %zu bytes\n", size);
+                       btr_add_output_dont_free(start, size, btrn);
+               }
                ret = -E_RECV_EOF;
                goto out;
        }
@@ -219,7 +228,10 @@ static void afh_recv_post_select(__a_unused struct sched *s, struct task *t)
                if (ret > 0)
                        goto out;
        }
-       afh_get_chunk(pard->current_chunk, afhi, pard->map, &start, &size);
+       ret = afh_get_chunk(pard->current_chunk, afhi,
+               pard->audio_format_num, pard->map,
+               pard->map_size, &start, &size,
+               &pard->afh_context);
        PARA_DEBUG_LOG("adding chunk %u\n", pard->current_chunk);
        btr_add_output_dont_free(start, size, btrn);
        if (pard->current_chunk >= pard->last_chunk) {
@@ -233,7 +245,7 @@ out:
                btr_remove_node(&rn->btrn);
                pard->current_chunk = pard->first_chunk;
        }
-       t->error = ret;
+       return ret;
 }
 
 /**
@@ -256,9 +268,6 @@ void afh_recv_init(struct receiver *r)
        r->parse_config = afh_recv_parse_config;
        r->free_config = afh_recv_free_config;
        r->execute = afh_execute;
-       r->help = (struct ggo_help) {
-               .short_help = afh_recv_args_info_help,
-               .detailed_help = afh_recv_args_info_detailed_help
-       };
+       r->help = (struct ggo_help)DEFINE_GGO_HELP(afh_recv);
        afh_recv_cmdline_parser_free(&dummy);
 }