X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=blobdiff_plain;f=buffer_tree.c;h=8a3175133e69aa57eff7da5155341cbf05532d03;hp=c5417c2b7fd447e9fa118d4b67e7a91e30b6814e;hb=d6b25bf854c164021550dfa8ff9fb4cfb104582e;hpb=b62e2796b85c7d7f3138fe729f4637853e0fafe0 diff --git a/buffer_tree.c b/buffer_tree.c index c5417c2b..8a317513 100644 --- a/buffer_tree.c +++ b/buffer_tree.c @@ -1,12 +1,7 @@ -/* - * Copyright (C) 2009-2011 Andre Noll - * - * Licensed under the GPL v2. For licencing details see COPYING. - */ +/* Copyright (C) 2009 Andre Noll , see file COPYING. */ /** \file buffer_tree.c Buffer tree and buffer pool implementations. */ #include -#include #include "para.h" #include "list.h" @@ -31,6 +26,8 @@ struct btr_buffer { int refcount; /* NULL means no buffer pool but a malloced buffer. */ struct btr_pool *pool; + /* Only relevant if pool is NULL. */ + bool dont_free; }; struct btr_buffer_reference { @@ -257,6 +254,8 @@ static void btr_pool_deallocate(struct btr_pool *btrp, size_t size) * * \param bnd Specifies how to create the new node. * + * \return A pointer to the newly allocated node. + * * This function always succeeds (or calls exit()). The returned pointer must * be freed using btr_free_node() after the node has been removed from the * buffer tree via btr_remove_node(). @@ -292,9 +291,10 @@ struct btr_node *btr_new_node(struct btr_node_description *bnd) bnd->child->parent = btrn; goto out; } - PARA_EMERG_LOG("inserting internal nodes not yet supported.\n"); - exit(EXIT_FAILURE); - assert(bnd->child->parent == bnd->parent); + list_add_tail(&btrn->node, &bnd->parent->children); + list_move(&bnd->child->node, &btrn->children); + bnd->child->parent = btrn; + PARA_INFO_LOG("added %s as internal node\n", bnd->name); out: return btrn; } @@ -318,7 +318,7 @@ static void dealloc_buffer(struct btr_buffer *btrb) { if (btrb->pool) btr_pool_deallocate(btrb->pool, btrb->size); - else + else if (!btrb->dont_free) free(btrb->buf); } @@ -382,7 +382,8 @@ void btr_add_output(char *buf, size_t size, struct btr_node *btrn) { struct btr_buffer *btrb; - assert(size != 0); + if (size == 0) + return; if (list_empty(&btrn->children)) { free(buf); return; @@ -391,6 +392,35 @@ void btr_add_output(char *buf, size_t size, struct btr_node *btrn) add_btrb_to_children(btrb, btrn, 0); } +/** + * Insert a buffer into the buffer tree, non-freeing variant. + * + * \param buf See \ref btr_add_output(). + * \param size See \ref btr_add_output(). + * \param btrn See \ref btr_add_output(). + * + * This is similar to btr_add_output() but additionally sets the \p dont_free + * flag on \a buf. If the refcount for the buffer drops to zero, \a buf will + * not be deallocated if this flag is set. + * + * The \p dont_free bit also prevents the children of \a btrn from modifying + * the buffer contents inplace. Specifically, \ref btr_inplace_ok() returns + * false if there is any buffer in the input queue with the \p dont_free bit + * set. + */ +void btr_add_output_dont_free(const char *buf, size_t size, struct btr_node *btrn) +{ + struct btr_buffer *btrb; + + if (size == 0) + return; + if (list_empty(&btrn->children)) + return; + btrb = new_btrb((char *)buf, size); + btrb->dont_free = true; + add_btrb_to_children(btrb, btrn, 0); +} + /** * Feed data to child nodes of a buffer tree node. * @@ -410,7 +440,8 @@ void btr_add_output_pool(struct btr_pool *btrp, size_t size, char *buf; size_t avail; - assert(size != 0); + if (size == 0) + return; if (list_empty(&btrn->children)) return; avail = btr_pool_get_buffer(btrp, &buf); @@ -509,14 +540,16 @@ static bool btr_no_children(struct btr_node *btrn) } /** - * Find out whether a node is an orphan node. + * Find out whether a node is an orphan. * * \param btrn The buffer tree node. * * \return True if \a btrn has no parent. * - * This function will always return true for the root node. However in case - * nodes have been removed from the tree, other nodes may become orphans too. + * This function returns true for the root node and false for any other node. + * + * After a (non-leaf) node was removed removed from the tree, the function + * returns true for all child nodes. */ bool btr_no_parent(struct btr_node *btrn) { @@ -543,9 +576,15 @@ bool btr_no_parent(struct btr_node *btrn) */ bool btr_inplace_ok(struct btr_node *btrn) { - if (!btrn->parent) - return true; - return list_is_singular(&btrn->parent->children); + struct btr_buffer_reference *br; + FOR_EACH_BUFFER_REF(br, btrn) { + struct btr_buffer *btrb = br->btrb; + if (btrb->refcount > 1) + return false; + if (btrb->dont_free == true) + return false; + } + return true; } static inline size_t br_available_bytes(struct btr_buffer_reference *br) @@ -561,40 +600,95 @@ static size_t btr_get_buffer_by_reference(struct btr_buffer_reference *br, char } /** - * Obtain the next buffer of the input queue of a buffer tree node. + * Obtain the next buffer of the input queue, omitting data. * * \param btrn The node whose input queue is to be queried. - * \param bufp Result pointer. + * \param omit Number of bytes to be omitted. + * \param bufp Result pointer. It is OK to pass \p NULL here. + * + * If a buffer tree node needs more input data but can not consume the data it + * already has (because it might be needed again later) this function can be + * used instead of btr_next_buffer() to get a reference to the buffer obtained + * by skipping the given number of bytes. Skipped input bytes are not consumed. + * + * With a zero \a omit argument, this function is equivalent to \ref + * btr_next_buffer(). * - * \return The number of bytes that can be read from buf. Zero if the input - * buffer queue is empty. In this case the value of \a bufp is undefined. + * \return Number of bytes in \a bufp. If there are less than or equal to \a + * omit many bytes available in the input queue of the buffer tree node pointed + * to by \a btrn, the function returns zero and the value of \a bufp is + * undefined. */ -size_t btr_next_buffer(struct btr_node *btrn, char **bufp) +size_t btr_next_buffer_omit(struct btr_node *btrn, size_t omit, char **bufp) { struct btr_buffer_reference *br; + size_t wrap_count, sz, rv = 0; char *buf, *result = NULL; - size_t sz, rv = 0; - FOR_EACH_BUFFER_REF(br, btrn) { + br = get_first_input_br(btrn); + if (!br) + return 0; + wrap_count = br->wrap_count; + if (wrap_count > 0) { /* we have a wrap buffer */ sz = btr_get_buffer_by_reference(br, &buf); - if (!result) { - result = buf; - rv = sz; - if (!br->btrb->pool) - break; - continue; + if (sz > omit) { /* and it's big enough */ + result = buf + omit; + rv = sz - omit; + /* + * Wrap buffers are allocated by malloc(), so the next + * buffer ref will not align nicely, so we return the + * tail of the wrap buffer. + */ + goto out; } - if (!br->btrb->pool) - break; - if (result + rv != buf) - break; - rv += sz; + /* + * The next wrap_count bytes exist twice, in the wrap buffer + * and as a buffer reference in the buffer tree pool. + */ + omit += wrap_count; + } + /* + * For buffer tree pools, the buffers in the list align, i.e. the next + * buffer in the list starts directly at the end of its predecessor. In + * this case we merge adjacent buffers and return one larger buffer + * instead. + */ + FOR_EACH_BUFFER_REF(br, btrn) { + sz = btr_get_buffer_by_reference(br, &buf); + if (result) { + if (result + rv != buf) + goto out; + rv += sz; + } else if (sz > omit) { + result = buf + omit; + rv = sz - omit; + } else + omit -= sz; } + if (!result) + return 0; +out: if (bufp) *bufp = result; return rv; } +/** + * Obtain the next buffer of the input queue of a buffer tree node. + * + * \param btrn The node whose input queue is to be queried. + * \param bufp Result pointer. + * + * \return The number of bytes that can be read from buf. + * + * The call of this function is is equivalent to calling \ref + * btr_next_buffer_omit() with an \a omit value of zero. + */ +size_t btr_next_buffer(struct btr_node *btrn, char **bufp) +{ + return btr_next_buffer_omit(btrn, 0, bufp); +} + /** * Deallocate the given number of bytes from the input queue. * @@ -603,7 +697,7 @@ size_t btr_next_buffer(struct btr_node *btrn, char **bufp) * * This function must be used to get rid of existing buffer references in the * node's input queue. If no references to a buffer remain, the underlying - * buffers are either freed (in the non-buffer tree case) or the read head of + * buffers are either freed (in the non-buffer pool case) or the read head of * the buffer pool is being advanced. * * Note that \a numbytes may be smaller than the buffer size. In this case the @@ -660,6 +754,11 @@ void btr_consume(struct btr_node *btrn, size_t numbytes) return btr_consume(btrn, sz); } +/** + * Clear the input queue of a buffer tree node. + * + * \param btrn The node whose input queue should be cleared. + */ void btr_drain(struct btr_node *btrn) { struct btr_buffer_reference *br, *tmp; @@ -668,17 +767,8 @@ void btr_drain(struct btr_node *btrn) btr_drop_buffer_reference(br); } -/** - * Free all resources allocated by btr_new_node(). - * - * \param btrn Pointer to a btr node obtained by \ref btr_new_node(). - * - * Like free(3), it is OK to call this with a \p NULL pointer argument. - */ -void btr_free_node(struct btr_node *btrn) +static void btr_free_node(struct btr_node *btrn) { - if (!btrn) - return; free(btrn->name); free(btrn); } @@ -686,26 +776,33 @@ void btr_free_node(struct btr_node *btrn) /** * Remove a node from a buffer tree. * - * \param btrn The node to remove. + * \param btrnp Determines the node to remove. * - * This makes all child nodes of \a btrn orphans and removes \a btrn from the - * list of children of its parent. Moreover, the input queue of \a btrn is - * flushed if it is not empty. + * This orphans all children of the node given by \a btrnp and removes this + * node from the child list of its parent. Moreover, the input queue is flushed + * and the node pointer given by \a btrp is set to \p NULL. * * \sa \ref btr_splice_out_node. */ -void btr_remove_node(struct btr_node *btrn) +void btr_remove_node(struct btr_node **btrnp) { struct btr_node *ch; + struct btr_node *btrn; - if (!btrn) + if (!btrnp) return; - PARA_NOTICE_LOG("removing btr node %s from buffer tree\n", btrn->name); + btrn = *btrnp; + if (!btrn) + goto out; + PARA_INFO_LOG("removing btr node %s from buffer tree\n", btrn->name); FOR_EACH_CHILD(ch, btrn) ch->parent = NULL; btr_drain(btrn); if (btrn->parent) list_del(&btrn->node); + btr_free_node(btrn); +out: + *btrnp = NULL; } /** @@ -739,7 +836,7 @@ size_t btr_get_input_queue_size(struct btr_node *btrn) /** * Remove a node from the buffer tree, reconnecting parent and children. * - * \param btrn The node to splice out. + * \param btrnp The node to splice out. * * This function is used by buffer tree nodes that do not exist during the * whole lifetime of the buffer tree. Unlike btr_remove_node(), calling @@ -747,9 +844,9 @@ size_t btr_get_input_queue_size(struct btr_node *btrn) * but reconnects the buffer tree by making all child nodes of \a btrn children * of the parent of \a btrn. */ -void btr_splice_out_node(struct btr_node *btrn) +void btr_splice_out_node(struct btr_node **btrnp) { - struct btr_node *ch, *tmp; + struct btr_node *btrn = *btrnp, *ch, *tmp; assert(btrn); PARA_NOTICE_LOG("splicing out %s\n", btrn->name); @@ -762,8 +859,12 @@ void btr_splice_out_node(struct btr_node *btrn) ch->parent = btrn->parent; if (btrn->parent) list_move(&ch->node, &btrn->parent->children); + else + list_del(&ch->node); } assert(list_empty(&btrn->children)); + btr_free_node(btrn); + *btrnp = NULL; } /** @@ -771,8 +872,8 @@ void btr_splice_out_node(struct btr_node *btrn) * * \param btrn The node whose output queue size should be computed. * - * This function iterates over all children of the given node and returns the - * size of the largest input queue. + * \return This function iterates over all children of the given node and + * returns the size of the largest input queue. */ size_t btr_get_output_queue_size(struct btr_node *btrn) { @@ -787,37 +888,36 @@ size_t btr_get_output_queue_size(struct btr_node *btrn) } /** - * Execute a inter-node command on a parent node. + * Execute an inter-node command on the given node or on a parent node. * * \param btrn The node to start looking. * \param command The command to execute. * \param value_result Additional arguments and result value. * - * This function traverses the buffer tree upwards and looks for parent nodes - * of \a btrn that understands \a command. On the first such node the command - * is executed, and the result is stored in \a value_result. + * This function traverses the buffer tree from \a btrn upwards and looks for + * the first node that understands \a command. On this node \a command is + * executed, and the result is stored in \a value_result. * * \return \p -ENOTSUP if no parent node of \a btrn understands \a command. * Otherwise the return value of the command handler is returned. + * + * \sa \ref receiver::execute, \ref filter::execute, \ref writer::execute. */ 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; - if (!parent) - return -ERRNO_TO_PARA_ERROR(ENOTSUP); - if (!parent->execute) + if (!btrn->execute) continue; - PARA_INFO_LOG("parent: %s, cmd: %s\n", parent->name, command); - ret = parent->execute(parent, command, value_result); + PARA_INFO_LOG("executing %s on %s\n", command, btrn->name); + ret = btrn->execute(btrn, 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, + PARA_INFO_LOG("%s(%s): %s\n", command, btrn->name, *value_result); return 1; } @@ -827,10 +927,11 @@ int btr_exec_up(struct btr_node *btrn, const char *command, char **value_result) /** * Obtain the context of a buffer node tree. * - * The returned pointer equals the context pointer used at creation time of the - * node. + * \param btrn The node whose output queue size should be computed. + * + * \return A pointer to the \a context address specified at node creation time. * - * \sa btr_new_node(), struct \ref btr_node_description. + * \sa \ref btr_new_node(), struct \ref btr_node_description. */ void *btr_context(struct btr_node *btrn) { @@ -1104,21 +1205,20 @@ int btr_node_status(struct btr_node *btrn, size_t min_iqs, size_t iqs; assert(btrn); - if (type != BTR_NT_LEAF) { - if (btr_no_children(btrn)) - return -E_BTR_NO_CHILD; - if (btr_get_output_queue_size(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; - } + if (type != BTR_NT_LEAF && btr_no_children(btrn)) + return -E_BTR_NO_CHILD; + if (type != BTR_NT_ROOT && btr_eof(btrn)) + return -E_BTR_EOF; + + if (btr_get_output_queue_size(btrn) > BTRN_MAX_PENDING) + return 0; + if (type == BTR_NT_ROOT) + return 1; + 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; } @@ -1134,3 +1234,18 @@ void btr_get_node_start(struct btr_node *btrn, struct timeval *tv) { *tv = btrn->start; } + +/** + * Get the parent node of a buffer tree node. + * + * \param btrn The node whose parent should be returned. + * + * \a btrn must not be \p NULL. + * + * \return The parent of \a btrn, or \p NULL if \a btrn is the + * root node of the buffer tree. + */ +struct btr_node *btr_parent(struct btr_node *btrn) +{ + return btrn->parent; +}