+
+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)
+{
+ 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;
+}
+
+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;
+}