X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=buffer_tree.c;h=dfecf10a24906f2e165ed9368f69c23e99f6e291;hp=7b3355a28b98aa601431029c83063aac016d2d9d;hb=e3acf84bf26f2730575624b1a1d3979d84272427;hpb=e4713d70469e7ca1afd8e8b9a3e67c3764cf3933 diff --git a/buffer_tree.c b/buffer_tree.c index 7b3355a2..dfecf10a 100644 --- a/buffer_tree.c +++ b/buffer_tree.c @@ -6,13 +6,27 @@ #include "string.h" #include "buffer_tree.h" #include "error.h" +#include "sched.h" +struct btr_pool { + char *name; + char *area_start; + char *area_end; + char *rhead; + char *whead; +}; + +enum btr_buffer_flags { + /* changes the way the buffer is deallocated */ + BTR_BF_BTR_POOL = 1, +}; struct btr_buffer { char *buf; size_t size; /** The number of references to this buffer. */ int refcount; + struct btr_pool *pool; }; struct btr_buffer_reference { @@ -20,6 +34,7 @@ struct btr_buffer_reference { size_t consumed; /* Each buffer reference belongs to the buffer queue list of some buffer tree node. */ struct list_head node; + size_t wrap_count; }; struct btr_node { @@ -29,6 +44,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 @@ -39,15 +56,119 @@ struct btr_node { void *context; }; +struct btr_pool *btr_pool_new(const char *name, size_t area_size) +{ + struct btr_pool *btrp; + + PARA_INFO_LOG("%s, %zu bytes\n", name, area_size); + btrp = para_malloc(sizeof(*btrp)); + btrp->area_start = para_malloc(area_size); + btrp->area_end = btrp->area_start + area_size; + btrp->rhead = btrp->area_start; + btrp->whead = btrp->area_start; + btrp->name = para_strdup(name); + return btrp; +} + +/* whead = NULL means area full */ + +void btr_pool_free(struct btr_pool *btrp) +{ + if (!btrp) + return; + free(btrp->area_start); + free(btrp->name); + free(btrp); +} + +size_t btr_pool_size(struct btr_pool *btrp) +{ + return btrp->area_end - btrp->area_start; +} + +size_t btr_pool_filled(struct btr_pool *btrp) +{ + if (!btrp->whead) + return btr_pool_size(btrp); + if (btrp->rhead <= btrp->whead) + return btrp->whead - btrp->rhead; + return btr_pool_size(btrp) - (btrp->rhead - btrp->whead); +} + +size_t btr_pool_unused(struct btr_pool *btrp) +{ + return btr_pool_size(btrp) - btr_pool_filled(btrp); +} + +size_t btr_pool_available(struct btr_pool *btrp) +{ + if (!btrp->whead) + return 0; + if (btrp->rhead <= btrp->whead) + return btrp->area_end - btrp->whead; + return btrp->rhead - btrp->whead; +} + +size_t btr_pool_get_buffer(struct btr_pool *btrp, char **result) +{ + if (result) + *result = btrp->whead; + return btr_pool_available(btrp); +} + +void btr_pool_allocate(struct btr_pool *btrp, size_t size) +{ + char *end; + + if (size == 0) + return; + //PARA_CRIT_LOG("filled: %zu, alloc %zu\n", btr_pool_filled(btrp), size); + assert(size <= btr_pool_available(btrp)); + end = btrp->whead + size; + assert(end <= btrp->area_end); + + if (end == btrp->area_end) { + PARA_DEBUG_LOG("%s: end of pool area reached\n", btrp->name); + end = btrp->area_start; + } + if (end == btrp->rhead) { + PARA_DEBUG_LOG("btrp buffer full\n"); + end = NULL; /* buffer full */ + } + btrp->whead = end; + //PARA_CRIT_LOG("filled: %zu\n", btr_pool_filled(btrp)); +} + +static void btr_pool_deallocate(struct btr_pool *btrp, size_t size) +{ + char *end = btrp->rhead + size; + + //PARA_CRIT_LOG("filled: %zu, dealloc %zu\n", btr_pool_filled(btrp), size); + if (size == 0) + return; + assert(end <= btrp->area_end); + assert(size <= btr_pool_filled(btrp)); + if (end == btrp->area_end) + end = btrp->area_start; + if (!btrp->whead) + btrp->whead = btrp->rhead; + btrp->rhead = end; + if (btrp->rhead == btrp->whead) + btrp->rhead = btrp->whead = btrp->area_start; + //PARA_CRIT_LOG("filled: %zu\n", btr_pool_filled(btrp)); +} + #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) -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)); @@ -56,28 +177,50 @@ struct btr_node *btr_new_node(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); 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. + * The freshly allocated buffer will have a zero refcount and will + * not be associated with a btr pool. */ static struct btr_buffer *new_btrb(char *buf, size_t size) { - struct btr_buffer *btrb = para_malloc(sizeof(*btrb)); + struct btr_buffer *btrb = para_calloc(sizeof(*btrb)); btrb->buf = buf; btrb->size = size; - btrb->refcount = 0; return btrb; } +static void dealloc_buffer(struct btr_buffer *btrb) +{ + if (btrb->pool) + btr_pool_deallocate(btrb->pool, btrb->size); + else + free(btrb->buf); +} + +static struct btr_buffer_reference *get_first_input_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); +} + /* * Deallocate the reference, release the resources if refcount drops to zero. */ @@ -90,7 +233,7 @@ static void btr_drop_buffer_reference(struct btr_buffer_reference *br) free(br); btrb->refcount--; if (btrb->refcount == 0) { - free(btrb->buf); + dealloc_buffer(btrb); free(btrb); } } @@ -100,12 +243,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)); + struct btr_buffer_reference *br = para_calloc(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; } } @@ -113,10 +260,54 @@ void btr_add_output(char *buf, size_t size, struct btr_node *btrn) { struct btr_buffer *btrb; + assert(size != 0); + if (list_empty(&btrn->children)) { + free(buf); + return; + } btrb = new_btrb(buf, size); add_btrb_to_children(btrb, btrn, 0); } +void btr_add_output_pool(struct btr_pool *btrp, size_t size, + struct btr_node *btrn) +{ + struct btr_buffer *btrb; + char *buf; + size_t avail; + + assert(size != 0); + if (list_empty(&btrn->children)) + return; + avail = btr_pool_get_buffer(btrp, &buf); + assert(avail >= size); + btr_pool_allocate(btrp, size); + btrb = new_btrb(buf, size); + btrb->pool = btrp; + add_btrb_to_children(btrb, btrn, 0); +} + +void btr_copy(const void *src, size_t n, struct btr_pool *btrp, + struct btr_node *btrn) +{ + char *buf; + size_t sz, copy; + + if (n == 0) + return; + assert(n <= btr_pool_unused(btrp)); + sz = btr_pool_get_buffer(btrp, &buf); + copy = PARA_MIN(sz, n); + memcpy(buf, src, copy); + btr_add_output_pool(btrp, copy, btrn); + if (copy == n) + return; + sz = btr_pool_get_buffer(btrp, &buf); + assert(sz >= n - copy); + memcpy(buf, src + copy, n - copy); + btr_add_output_pool(btrp, n - copy, btrn); +} + static void btr_pushdown_br(struct btr_buffer_reference *br, struct btr_node *btrn) { add_btrb_to_children(br->btrb, btrn, br->consumed); @@ -131,6 +322,17 @@ void btr_pushdown(struct btr_node *btrn) btr_pushdown_br(br, btrn); } +int btr_pushdown_one(struct btr_node *btrn) +{ + struct btr_buffer_reference *br; + + if (list_empty(&btrn->input_queue)) + return 0; + br = list_first_entry(&btrn->input_queue, struct btr_buffer_reference, node); + btr_pushdown_br(br, btrn); + return 1; +} + /* Return true if this node has no children. */ bool btr_no_children(struct btr_node *btrn) { @@ -156,32 +358,98 @@ static inline size_t br_available_bytes(struct btr_buffer_reference *br) size_t btr_get_buffer_by_reference(struct btr_buffer_reference *br, char **buf) { - *buf = br->btrb->buf + br->consumed; + if (buf) + *buf = br->btrb->buf + br->consumed; return br_available_bytes(br); } +/** + * \return zero if the input buffer queue is empty. + */ size_t btr_next_buffer(struct btr_node *btrn, char **bufp) { struct btr_buffer_reference *br; + char *buf, *result = NULL; + size_t sz, rv = 0; - if (list_empty(&btrn->input_queue)) { - *bufp = NULL; - return 0; + FOR_EACH_BUFFER_REF(br, btrn) { + sz = btr_get_buffer_by_reference(br, &buf); + if (!result) { + result = buf; + rv = sz; + if (!br->btrb->pool) + break; + continue; + } + if (!br->btrb->pool) + break; + if (result + rv != buf) { + PARA_DEBUG_LOG("%s: pool merge impossible: %p != %p\n", + btrn->name, result + rv, buf); + break; + } +// PARA_CRIT_LOG("%s: inplace merge (%zu, %zu)->%zu\n", btrn->name, +// rv, sz, rv + sz); +// PARA_CRIT_LOG("%s: inplace merge %p (%zu)\n", btrn->name, +// result, sz); + rv += sz; } - br = list_first_entry(&btrn->input_queue, struct btr_buffer_reference, node); - return btr_get_buffer_by_reference(br, bufp); + if (bufp) + *bufp = result; + return rv; } void btr_consume(struct btr_node *btrn, size_t numbytes) { - struct btr_buffer_reference *br; + struct btr_buffer_reference *br, *tmp; + size_t sz; - 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); + if (numbytes == 0) + return; + br = get_first_input_br(btrn); + assert(br); + + //PARA_CRIT_LOG("wrap count: %zu\n", br->wrap_count); + if (br->wrap_count == 0) { + /* + * No wrap buffer. Drop buffer references whose buffer + * has been fully used. */ + FOR_EACH_BUFFER_REF_SAFE(br, tmp, btrn) { + if (br->consumed + numbytes <= br->btrb->size) { + br->consumed += numbytes; + if (br->consumed == br->btrb->size) + btr_drop_buffer_reference(br); + return; + } + numbytes -= br->btrb->size - br->consumed; + btr_drop_buffer_reference(br); + } + assert(true); + } + /* + + We have a wrap buffer, consume from it. If in total, + i.e. including previous calls to brt_consume(), less than + wrap_count has been consumed, there's nothing more we can do. + + Otherwise we drop the wrap buffer and consume from subsequent + buffers of the input queue the correct amount of bytes. This + is the total number of bytes that have been consumed from the + wrap buffer. +*/ + PARA_DEBUG_LOG("consuming %zu/%zu bytes from wrap buffer\n", numbytes, + br_available_bytes(br)); + + assert(numbytes <= br_available_bytes(br)); + if (br->consumed + numbytes < br->wrap_count) { + br->consumed += numbytes; + return; + } + PARA_DEBUG_LOG("dropping wrap buffer (%zu bytes)\n", br->btrb->size); + /* get rid of the wrap buffer */ + sz = br->consumed + numbytes; + btr_drop_buffer_reference(br); + return btr_consume(btrn, sz); } static void flush_input_queue(struct btr_node *btrn) @@ -191,20 +459,26 @@ static void flush_input_queue(struct btr_node *btrn) btr_drop_buffer_reference(br); } -void btr_del_node(struct btr_node *btrn) +void btr_free_node(struct btr_node *btrn) +{ + if (!btrn) + return; + free(btrn->name); + free(btrn); +} + +void btr_remove_node(struct btr_node *btrn) { struct btr_node *ch; if (!btrn) return; - PARA_NOTICE_LOG("deleting %s\n", btrn->name); + PARA_NOTICE_LOG("removing btr node %s from buffer tree\n", btrn->name); FOR_EACH_CHILD(ch, btrn) ch->parent = NULL; flush_input_queue(btrn); if (btrn->parent) list_del(&btrn->node); - free(btrn->name); - free(btrn); } size_t btr_get_input_queue_size(struct btr_node *btrn) @@ -219,22 +493,23 @@ size_t btr_get_input_queue_size(struct btr_node *btrn) return size; } -int btr_splice_out_node(struct btr_node *btrn) +void btr_splice_out_node(struct btr_node *btrn) { - struct btr_node *ch; + struct btr_node *ch, *tmp; - if (!btrn) - return -ERRNO_TO_PARA_ERROR(EINVAL); - if (btr_get_input_queue_size(btrn) != 0) - return -ERRNO_TO_PARA_ERROR(EINVAL); + assert(btrn); PARA_NOTICE_LOG("splicing out %s\n", btrn->name); + btr_pushdown(btrn); if (btrn->parent) list_del(&btrn->node); - FOR_EACH_CHILD(ch, btrn) + 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; - free(btrn->name); - free(btrn); - return 1; + if (btrn->parent) + list_move(&ch->node, &btrn->parent->children); + } + assert(list_empty(&btrn->children)); } /** @@ -269,16 +544,20 @@ int btr_exec_up(struct btr_node *btrn, const char *command, char **value_result) 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; + 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); } @@ -287,3 +566,238 @@ void *btr_context(struct btr_node *btrn) { return btrn->context; } + +static bool need_buffer_pool_merge(struct btr_node *btrn) +{ + struct btr_buffer_reference *br = get_first_input_br(btrn); + + if (!br) + return false; + if (br->wrap_count != 0) + return true; + if (br->btrb->pool) + return true; + return false; +} + +static void merge_input_pool(struct btr_node *btrn, size_t dest_size) +{ + struct btr_buffer_reference *br, *wbr; + int num_refs; /* including wrap buffer */ + char *buf, *buf1, *buf2 = NULL; + size_t sz, sz1, sz2 = 0, wsz; + + if (list_empty(&btrn->input_queue)) + return; + + num_refs = 0; + FOR_EACH_BUFFER_REF(br, btrn) { + num_refs++; + sz = btr_get_buffer_by_reference(br, &buf); + if (br->wrap_count != 0) { + assert(!wbr); + assert(num_refs == 1); + wbr = br; + if (sz >= dest_size) + return; + continue; + } + if (!buf1) { + buf1 = buf; + sz1 = sz; + goto next; + } + if (buf1 + sz1 == buf) { + sz1 += sz; + goto next; + } + if (!buf2) { + buf2 = buf; + sz2 = sz; + goto next; + } + assert(buf2 + sz2 == buf); + sz2 += sz; +next: + if (sz1 + sz2 >= dest_size) + break; + } + /* + * If the second buffer is large, we only take the first part of it to + * avoid having to memcpy() huge buffers. + */ + sz2 = PARA_MIN(sz2, (size_t)(64 * 1024)); + if (!wbr) { + assert(buf1); + if (!buf2) /* nothing to do */ + return; + /* Make a new wrap buffer combining buf1 and buf2. */ + sz = sz1 + sz2; + buf = para_malloc(sz); + PARA_DEBUG_LOG("merging input buffers: (%p:%zu, %p:%zu) -> %p:%zu\n", + buf1, sz1, buf2, sz2, buf, sz); + memcpy(buf, buf1, sz1); + memcpy(buf + sz1, buf2, sz2); + br = para_calloc(sizeof(*br)); + br->btrb = new_btrb(buf, sz); + br->btrb->refcount = 1; + br->consumed = 0; + /* This is a wrap buffer */ + br->wrap_count = sz1; + para_list_add(&br->node, &btrn->input_queue); + return; + } + /* + * We already have a wrap buffer, but it is too small. It might be + * partially used. + */ + wsz = br_available_bytes(wbr); + if (wbr->wrap_count == sz1 && wbr->btrb->size >= sz1 + sz2) /* nothing we can do about it */ + return; + assert(buf1 && buf2); + sz = sz1 + sz2 - wbr->btrb->size; /* amount of new data */ + wbr->btrb->size += sz; + PARA_DEBUG_LOG("increasing wrap buffer to %zu\n", wbr->btrb->size); + wbr->btrb->buf = para_realloc(wbr->btrb->buf, wbr->btrb->size); + /* copy the new data to the end of the reallocated buffer */ + assert(sz2 >= sz); + memcpy(wbr->btrb->buf + wbr->btrb->size - sz, buf2 + sz2 - sz, sz); +} + +/** + * Merge the first two input buffers into one. + * + * This is a quite expensive operation. + * + * \return The number of buffers that have been available (zero, one or two). + */ +static int merge_input(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]; + buf = para_malloc(sz); + PARA_DEBUG_LOG("%s: memory merging input buffers: (%zu, %zu) -> %zu\n", + btrn->name, szs[0], szs[1], sz); + memcpy(buf, bufs[0], szs[0]); + memcpy(buf + szs[0], bufs[1], szs[1]); + + br = para_calloc(sizeof(*br)); + br->btrb = new_btrb(buf, sz); + br->btrb->refcount = 1; + + /* 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; +} + +void btr_merge(struct btr_node *btrn, size_t dest_size) +{ + if (need_buffer_pool_merge(btrn)) + return merge_input_pool(btrn, dest_size); + for (;;) { + char *buf; + 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; + } +} + +bool btr_eof(struct btr_node *btrn) +{ + char *buf; + size_t len = btr_next_buffer(btrn, &buf); + + return (len == 0 && btr_no_parent(btrn)); +} + +void log_tree_recursively(struct btr_node *btrn, int loglevel, int depth) +{ + struct btr_node *ch; + const char spaces[] = " ", *space = spaces + 16 - depth; + + if (depth > 16) + return; + para_log(loglevel, "%s%s\n", space, btrn->name); + FOR_EACH_CHILD(ch, btrn) + log_tree_recursively(ch, loglevel, depth + 1); +} + +void btr_log_tree(struct btr_node *btrn, int loglevel) +{ + return log_tree_recursively(btrn, loglevel, 0); +} + +/* + * \return \a root if \a name is \p NULL. + */ +struct btr_node *btr_search_node(const char *name, struct btr_node *root) +{ + struct btr_node *ch; + + if (!name) + return root; + if (!strcmp(root->name, name)) + return root; + FOR_EACH_CHILD(ch, root) { + struct btr_node *result = btr_search_node(name, ch); + if (result) + return result; + } + return NULL; +} + +/** 640K ought to be enough for everybody ;) */ +#define BTRN_MAX_PENDING (640 * 1024) + +int btr_node_status(struct btr_node *btrn, size_t min_iqs, + enum btr_node_type type) +{ + size_t iqs; + + if (!btrn) + return 0; + if (type != BTR_NT_LEAF) { + if (btr_no_children(btrn)) + return -E_BTR_NO_CHILD; + if (btr_bytes_pending(btrn) > BTRN_MAX_PENDING) + return 0; + } + if (type != BTR_NT_ROOT) { + if (btr_eof(btrn)) + return -E_BTR_EOF; + iqs = btr_get_input_queue_size(btrn); + if (iqs == 0) /* we have a parent, because not eof */ + return 0; + if (iqs < min_iqs && !btr_no_parent(btrn)) + return 0; + } + return 1; +} + +void btr_get_node_start(struct btr_node *btrn, struct timeval *tv) +{ + *tv = btrn->start; +}