X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=buffer_tree.c;h=1cb295b4de02e3ca3003f83043cf34d536f2ffc5;hp=bbd055b970c09d22beb13b03d446294752426d71;hb=d29a1515fbb74f7853fa771698e90646c24bde33;hpb=5bd897365d6e849b01b34cef28ea542d9cfbdfa9 diff --git a/buffer_tree.c b/buffer_tree.c index bbd055b9..1cb295b4 100644 --- a/buffer_tree.c +++ b/buffer_tree.c @@ -4,7 +4,8 @@ #include "para.h" #include "list.h" #include "string.h" -//#include "buffer_tree.h" +#include "buffer_tree.h" +#include "error.h" struct btr_buffer { @@ -34,31 +35,45 @@ struct btr_node { * used by this btr node. */ struct list_head input_queue; + btr_command_handler execute; + void *context; }; -#define FOR_EACH_TARGET_NODE(_tn, _btrn) list_for_each_entry((_tn), \ +#define FOR_EACH_CHILD(_tn, _btrn) list_for_each_entry((_tn), \ &((_btrn)->children), node) +#define FOR_EACH_CHILD_SAFE(_tn, _tmp, _btrn) \ + list_for_each_entry_safe((_tn), (_tmp), &((_btrn)->children), node) #define FOR_EACH_BUFFER_REF(_br, _btrn) \ list_for_each_entry((_br), &(_btrn)->input_queue, node) #define FOR_EACH_BUFFER_REF_SAFE(_br, _tmp, _btrn) \ list_for_each_entry_safe((_br), (_tmp), &(_btrn)->input_queue, node) -INIT_STDERR_LOGGING(0); - -struct btr_node *btr_new_node(char *name, struct btr_node *parent) +struct btr_node *btr_new_node(const char *name, struct btr_node *parent, + btr_command_handler handler, void *context) { struct btr_node *btrn = para_malloc(sizeof(*btrn)); btrn->name = para_strdup(name); btrn->parent = parent; + btrn->execute = handler; + btrn->context = context; if (parent) list_add_tail(&btrn->node, &parent->children); INIT_LIST_HEAD(&btrn->children); INIT_LIST_HEAD(&btrn->input_queue); + if (parent) + PARA_INFO_LOG("added %s as child of %s\n", name, parent->name); + else + PARA_INFO_LOG("added %s as btr root\n", name); return btrn; } +/* + * Allocate a new btr buffer. + * + * The freshly allocated buffer will have a zero refcount. + */ static struct btr_buffer *new_btrb(char *buf, size_t size) { struct btr_buffer *btrb = para_malloc(sizeof(*btrb)); @@ -69,10 +84,14 @@ static struct btr_buffer *new_btrb(char *buf, size_t size) return btrb; } -void btr_drop_buffer_reference(struct btr_buffer_reference *br) +/* + * Deallocate the reference, release the resources if refcount drops to zero. + */ +static void btr_drop_buffer_reference(struct btr_buffer_reference *br) { struct btr_buffer *btrb = br->btrb; + //PARA_CRIT_LOG("dropping buffer reference %p\n", br); list_del(&br->node); free(br); btrb->refcount--; @@ -82,15 +101,16 @@ void btr_drop_buffer_reference(struct btr_buffer_reference *br) } } -static void add_btrb_to_targets(struct btr_buffer *btrb, struct btr_node *btrn) +static void add_btrb_to_children(struct btr_buffer *btrb, + struct btr_node *btrn, size_t consumed) { - struct btr_node *tn; /* target node */ + struct btr_node *ch; - FOR_EACH_TARGET_NODE(tn, btrn) { + FOR_EACH_CHILD(ch, btrn) { struct btr_buffer_reference *br = para_malloc(sizeof(*br)); br->btrb = btrb; - br->consumed = 0; - list_add_tail(&br->node, &tn->input_queue); + br->consumed = consumed; + list_add_tail(&br->node, &ch->input_queue); btrb->refcount++; } } @@ -100,12 +120,12 @@ void btr_add_output(char *buf, size_t size, struct btr_node *btrn) struct btr_buffer *btrb; btrb = new_btrb(buf, size); - add_btrb_to_targets(btrb, btrn); + add_btrb_to_children(btrb, btrn, 0); } -void btr_pushdown_br(struct btr_buffer_reference *br, struct btr_node *btrn) +static void btr_pushdown_br(struct btr_buffer_reference *br, struct btr_node *btrn) { - add_btrb_to_targets(br->btrb, btrn); + add_btrb_to_children(br->btrb, btrn, br->consumed); btr_drop_buffer_reference(br); } @@ -117,6 +137,7 @@ void btr_pushdown(struct btr_node *btrn) btr_pushdown_br(br, btrn); } +/* Return true if this node has no children. */ bool btr_no_children(struct btr_node *btrn) { return list_empty(&btrn->children); @@ -134,13 +155,6 @@ bool btr_inplace_ok(struct btr_node *btrn) return list_is_singular(&btrn->parent->children); } -struct btr_buffer_reference *btr_next_br(struct btr_node *btrn) -{ - if (list_empty(&btrn->input_queue)) - return NULL; - return list_first_entry(&btrn->input_queue, struct btr_buffer_reference, node); -} - static inline size_t br_available_bytes(struct btr_buffer_reference *br) { return br->btrb->size - br->consumed; @@ -152,9 +166,29 @@ size_t btr_get_buffer_by_reference(struct btr_buffer_reference *br, char **buf) return br_available_bytes(br); } -void btr_increase_used_bytes(struct btr_buffer_reference *br, size_t consumed) +/** + * \return zero if the input buffer queue is empty. + */ +size_t btr_next_buffer(struct btr_node *btrn, char **bufp) { - br->consumed += consumed; + struct btr_buffer_reference *br; + + if (list_empty(&btrn->input_queue)) { + *bufp = NULL; + return 0; + } + br = list_first_entry(&btrn->input_queue, struct btr_buffer_reference, node); + return btr_get_buffer_by_reference(br, bufp); +} + +void btr_consume(struct btr_node *btrn, size_t numbytes) +{ + struct btr_buffer_reference *br; + + assert(!list_empty(&btrn->input_queue)); + br = list_first_entry(&btrn->input_queue, struct btr_buffer_reference, node); + assert(br->consumed + numbytes <= br->btrb->size); + br->consumed += numbytes; if (br->consumed == br->btrb->size) btr_drop_buffer_reference(br); } @@ -168,10 +202,13 @@ static void flush_input_queue(struct btr_node *btrn) void btr_del_node(struct btr_node *btrn) { - struct btr_node *tn; + struct btr_node *ch; - FOR_EACH_TARGET_NODE(tn, btrn) - tn->parent = NULL; + if (!btrn) + return; + PARA_NOTICE_LOG("deleting %s\n", btrn->name); + FOR_EACH_CHILD(ch, btrn) + ch->parent = NULL; flush_input_queue(btrn); if (btrn->parent) list_del(&btrn->node); @@ -184,12 +221,133 @@ size_t btr_get_input_queue_size(struct btr_node *btrn) 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); + } return size; } -int main(void) +void btr_splice_out_node(struct btr_node *btrn) +{ + struct btr_node *ch, *tmp; + + assert(btrn); + PARA_NOTICE_LOG("splicing out %s\n", btrn->name); + btr_pushdown(btrn); + if (btrn->parent) + list_del(&btrn->node); + FOR_EACH_CHILD_SAFE(ch, tmp, btrn) { + PARA_INFO_LOG("parent(%s): %s\n", ch->name, + btrn->parent? btrn->parent->name : "NULL"); + ch->parent = btrn->parent; + if (btrn->parent) + list_move(&ch->node, &btrn->parent->children); + } + assert(list_empty(&btrn->children)); + free(btrn->name); + free(btrn); +} + +/** + * Return the size of the largest input queue. + * + * Iterates over all children of the given node. + */ +size_t btr_bytes_pending(struct btr_node *btrn) +{ + size_t max_size = 0; + struct btr_node *ch; + + FOR_EACH_CHILD(ch, btrn) { + size_t size = btr_get_input_queue_size(ch); + max_size = PARA_MAX(max_size, 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(btrn, command, value_result); +} + +int btr_exec_up(struct btr_node *btrn, const char *command, char **value_result) { - return 1; + int ret; + + for (; btrn; btrn = btrn->parent) { + struct btr_node *parent = btrn->parent; + if (!parent) + return -ERRNO_TO_PARA_ERROR(ENOTSUP); + if (!parent->execute) + continue; + PARA_INFO_LOG("parent: %s, cmd: %s\n", parent->name, command); + ret = parent->execute(parent, command, value_result); + if (ret == -ERRNO_TO_PARA_ERROR(ENOTSUP)) + continue; + if (ret < 0) + return ret; + if (value_result && *value_result) + PARA_NOTICE_LOG("%s(%s): %s\n", command, parent->name, + *value_result); + return 1; + } + return -ERRNO_TO_PARA_ERROR(ENOTSUP); +} + +void *btr_context(struct btr_node *btrn) +{ + return btrn->context; +} + +/** + * Merge the first two input buffers into one. + * + * This is a quite expensive operation. + * + * \return The number of buffers that have been merged (zero, one or two). + */ +int btr_merge(struct btr_node *btrn) +{ + struct btr_buffer_reference *brs[2], *br; + char *bufs[2], *buf; + size_t szs[2], sz; + int i; + + if (list_empty(&btrn->input_queue)) + return 0; + if (list_is_singular(&btrn->input_queue)) + return 1; + i = 0; + /* get references to the first two buffers */ + FOR_EACH_BUFFER_REF(br, btrn) { + brs[i] = br; + szs[i] = btr_get_buffer_by_reference(brs[i], bufs + i); + i++; + if (i == 2) + break; + } + /* 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); + buf = para_malloc(sz); + /* TODO: Avoid this memcopy by introducing btr buffer pool. */ + memcpy(buf, bufs[0], szs[0]); + memcpy(buf + szs[0], bufs[1], szs[1]); + + br = para_malloc(sizeof(*br)); + br->btrb = new_btrb(buf, sz); + br->btrb->refcount = 1; + br->consumed = 0; + + /* replace the first two refs by the new one */ + btr_drop_buffer_reference(brs[0]); + btr_drop_buffer_reference(brs[1]); + para_list_add(&br->node, &btrn->input_queue); + return 2; }