GCOV - Varnish Cache


/*-
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006-2015 Varnish Software AS
 * All rights reserved.
 *
 * Author: Martin Blix Grydeland <martin@varnish-software.com>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 */

#include "config.h"

#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "vdef.h"
#include "vas.h"
#include "miniobj.h"

#include "vqueue.h"
#include "vre.h"
#include "vtim.h"
#include "vtree.h"

#include "vapi/vsl.h"

#include "vsl_api.h"

#define VTX_CACHE 10
#define VTX_BUFSIZE_MIN 64
#define VTX_SHMCHUNKS 3

static const char * const vsl_t_names[VSL_t__MAX] = {
        [VSL_t_unknown] = "unknown",
        [VSL_t_sess]    = "sess",
        [VSL_t_req]     = "req",
        [VSL_t_bereq]   = "bereq",
        [VSL_t_raw]     = "raw",
};

static const char * const vsl_r_names[VSL_r__MAX] = {
        [VSL_r_unknown] = "unknown",
        [VSL_r_http_1]  = "HTTP/1",
        [VSL_r_rxreq]   = "rxreq",
        [VSL_r_esi]     = "esi",
        [VSL_r_restart] = "restart",
        [VSL_r_pass]    = "pass",
        [VSL_r_fetch]   = "fetch",
        [VSL_r_bgfetch] = "bgfetch",
        [VSL_r_pipe]    = "pipe",
};

struct vtx;
VTAILQ_HEAD(vtxhead, vtx);

struct vslc_raw {
        unsigned                magic;
#define VSLC_RAW_MAGIC          0x247EBD44

        struct VSL_cursor       cursor;

        const uint32_t          *ptr;
};

struct synth {
        unsigned                magic;
#define SYNTH_MAGIC             0xC654479F

        VTAILQ_ENTRY(synth)     list;
        size_t                  offset;
        uint32_t                data[VSL_OVERHEAD + VSL_WORDS(64)];
};
VTAILQ_HEAD(synthhead, synth);

enum chunk_t {
        chunk_t__unassigned,
        chunk_t_shm,
        chunk_t_buf,
};

struct chunk {
        unsigned                                magic;
#define CHUNK_MAGIC                             0x48DC0194
        enum chunk_t                            type;
        union {
                struct {
                        struct VSLC_ptr         start;
                        VTAILQ_ENTRY(chunk)     shmref;
                } shm;
                struct {
                        uint32_t                *data;
                        size_t                  space;
                } buf;
        };
        size_t                                  len;
        struct vtx                              *vtx;
        VTAILQ_ENTRY(chunk)                     list;
};
VTAILQ_HEAD(chunkhead, chunk);

struct vslc_vtx {
        unsigned                magic;
#define VSLC_VTX_MAGIC          0x74C6523F

        struct VSL_cursor       cursor;

        struct vtx              *vtx;
        struct synth            *synth;
        struct chunk            *chunk;
        size_t                  chunkstart;
        size_t                  offset;
};

struct vtx_key {
        uint64_t                vxid;
        VRBT_ENTRY(vtx_key)     entry;
};
VRBT_HEAD(vtx_tree, vtx_key);

struct vtx {
        struct vtx_key          key;
        unsigned                magic;
#define VTX_MAGIC               0xACC21D09
        VTAILQ_ENTRY(vtx)       list_child;
        VTAILQ_ENTRY(vtx)       list_vtx;

        double                  t_start;
        unsigned                flags;
#define VTX_F_BEGIN             0x1 /* Begin record processed */
#define VTX_F_END               0x2 /* End record processed */
#define VTX_F_COMPLETE          0x4 /* Marked complete. No new children
                                       should be appended */
#define VTX_F_READY             0x8 /* This vtx and all it's children are
                                       complete */

        enum VSL_transaction_e  type;
        enum VSL_reason_e       reason;

        struct vtx              *parent;
        struct vtxhead          child;
        unsigned                n_child;
        unsigned                n_childready;
        unsigned                n_descend;

        struct synthhead        synth;

        struct chunk            shmchunks[VTX_SHMCHUNKS];
        struct chunkhead        shmchunks_free;

        struct chunkhead        chunks;
        size_t                  len;

        struct vslc_vtx         c;
};

struct VSLQ {
        unsigned                magic;
#define VSLQ_MAGIC              0x23A8BE97

        struct VSL_data         *vsl;
        struct VSL_cursor       *c;
        struct vslq_query       *query;

        enum VSL_grouping_e     grouping;

        /* Structured mode */
        struct vtx_tree         tree;
        struct vtxhead          ready;
        struct vtxhead          incomplete;
        int                     n_outstanding;
        struct chunkhead        shmrefs;
        struct vtxhead          cache;
        unsigned                n_cache;

        /* Rate limiting */
        double                  credits;
        vtim_mono               last_use;

        /* Raw mode */
        struct {
                struct vslc_raw         c;
                struct VSL_transaction  trans;
                struct VSL_transaction  *ptrans[2];
                struct VSLC_ptr         start;
                ssize_t                 len;
                ssize_t                 offset;
        } raw;
};

static void vtx_synth_rec(struct vtx *vtx, unsigned tag, const char *fmt, ...);
/*lint -esym(534, vtx_diag) */
static int vtx_diag(struct vtx *vtx, const char *msg);
/*lint -esym(534, vtx_diag_tag) */
static int vtx_diag_tag(struct vtx *vtx, const uint32_t *ptr,
    const char *reason);

static inline int
vtx_keycmp(const struct vtx_key *a, const struct vtx_key *b)
{
        if (a->vxid < b->vxid)
                return (-1);
        if (a->vxid > b->vxid)
                return (1);
        return (0);
}

VRBT_GENERATE_REMOVE_COLOR(vtx_tree, vtx_key, entry, static)
VRBT_GENERATE_REMOVE(vtx_tree, vtx_key, entry, static)
VRBT_GENERATE_INSERT_COLOR(vtx_tree, vtx_key, entry, static)
VRBT_GENERATE_INSERT_FINISH(vtx_tree, vtx_key, entry, static)
VRBT_GENERATE_INSERT(vtx_tree, vtx_key, entry, vtx_keycmp, static)
VRBT_GENERATE_FIND(vtx_tree, vtx_key, entry, vtx_keycmp, static)

static enum vsl_status v_matchproto_(vslc_next_f)
vslc_raw_next(const struct VSL_cursor *cursor)
{
        struct vslc_raw *c;

        CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_RAW_MAGIC);
        assert(&c->cursor == cursor);

        AN(c->ptr);
        if (c->cursor.rec.ptr == NULL) {
                c->cursor.rec.ptr = c->ptr;
                return (vsl_more);
        } else {
                c->cursor.rec.ptr = NULL;
                return (vsl_end);
        }
}

static enum vsl_status v_matchproto_(vslc_reset_f)
vslc_raw_reset(const struct VSL_cursor *cursor)
{
        struct vslc_raw *c;

        CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_RAW_MAGIC);
        assert(&c->cursor == cursor);

        AN(c->ptr);
        c->cursor.rec.ptr = NULL;

        return (vsl_end);
}

static const struct vslc_tbl vslc_raw_tbl = {
        .magic  = VSLC_TBL_MAGIC,
        .delete = NULL,
        .next   = vslc_raw_next,
        .reset  = vslc_raw_reset,
        .check  = NULL,
};

static enum vsl_status v_matchproto_(vslc_next_f)
vslc_vtx_next(const struct VSL_cursor *cursor)
{
        struct vslc_vtx *c;
        const uint32_t *ptr;
        unsigned overrun;

        CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_VTX_MAGIC);
        assert(&c->cursor == cursor);
        CHECK_OBJ_NOTNULL(c->vtx, VTX_MAGIC);

        do {
                CHECK_OBJ_ORNULL(c->synth, SYNTH_MAGIC);
                if (c->synth != NULL && c->synth->offset == c->offset) {
                        /* We're at the offset of the next synth record,
                           point to it and advance the pointer */
                        c->cursor.rec.ptr = c->synth->data;
                        c->synth = VTAILQ_NEXT(c->synth, list);
                } else {
                        overrun = c->offset > c->vtx->len;
                        AZ(overrun);
                        if (c->offset == c->vtx->len)
                                return (vsl_end);

                        /* Advance chunk pointer */
                        if (c->chunk == NULL) {
                                c->chunk = VTAILQ_FIRST(&c->vtx->chunks);
                                c->chunkstart = 0;
                        }
                        CHECK_OBJ_NOTNULL(c->chunk, CHUNK_MAGIC);
                        while (c->offset >= c->chunkstart + c->chunk->len) {
                                c->chunkstart += c->chunk->len;
                                c->chunk = VTAILQ_NEXT(c->chunk, list);
                                CHECK_OBJ_NOTNULL(c->chunk, CHUNK_MAGIC);
                        }

                        /* Point to the next stored record */
                        if (c->chunk->type == chunk_t_shm)
                                ptr = c->chunk->shm.start.ptr;
                        else {
                                assert(c->chunk->type == chunk_t_buf);
                                ptr = c->chunk->buf.data;
                        }
                        c->cursor.rec.ptr = ptr + c->offset - c->chunkstart;
                        c->offset += VSL_NEXT(c->cursor.rec.ptr) -
                            c->cursor.rec.ptr;
                }
        } while (VSL_TAG(c->cursor.rec.ptr) == SLT__Batch);

        return (vsl_more);
}

static enum vsl_status v_matchproto_(vslc_reset_f)
vslc_vtx_reset(const struct VSL_cursor *cursor)
{
        struct vslc_vtx *c;

        CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_VTX_MAGIC);
        assert(&c->cursor == cursor);
        CHECK_OBJ_NOTNULL(c->vtx, VTX_MAGIC);
        c->synth = VTAILQ_FIRST(&c->vtx->synth);
        c->chunk = NULL;
        c->chunkstart = 0;
        c->offset = 0;
        c->cursor.rec.ptr = NULL;

        return (vsl_end);
}

static const struct vslc_tbl vslc_vtx_tbl = {
        .magic  = VSLC_TBL_MAGIC,
        .delete = NULL,
        .next   = vslc_vtx_next,
        .reset  = vslc_vtx_reset,
        .check  = NULL,
};

/* Create a buf chunk */
static struct chunk *
chunk_newbuf(struct vtx *vtx, const uint32_t *ptr, size_t len)
{
        struct chunk *chunk;

        ALLOC_OBJ(chunk, CHUNK_MAGIC);
        XXXAN(chunk);
        chunk->type = chunk_t_buf;
        chunk->vtx = vtx;
        chunk->buf.space = VTX_BUFSIZE_MIN;
        while (chunk->buf.space < len)
                chunk->buf.space *= 2;
        chunk->buf.data = malloc(sizeof (uint32_t) * chunk->buf.space);
        AN(chunk->buf.data);
        vmemcpy(chunk->buf.data, ptr, sizeof(uint32_t) * len);
        chunk->len = len;
        return (chunk);
}

/* Free a buf chunk */
static void
chunk_freebuf(struct chunk **pchunk)
{
        struct chunk *chunk;

        TAKE_OBJ_NOTNULL(chunk, pchunk, CHUNK_MAGIC);
        assert(chunk->type == chunk_t_buf);
        free(chunk->buf.data);
        FREE_OBJ(chunk);
}

/* Append a set of records to a chunk */
static void
chunk_appendbuf(struct chunk *chunk, const uint32_t *ptr, size_t len)
{

        CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
        assert(chunk->type == chunk_t_buf);
        if (chunk->buf.space < chunk->len + len) {
                while (chunk->buf.space < chunk->len + len)
                        chunk->buf.space *= 2;
                chunk->buf.data = realloc(chunk->buf.data,
                    sizeof (uint32_t) * chunk->buf.space);
        }
        vmemcpy(chunk->buf.data + chunk->len, ptr, sizeof(uint32_t) * len);
        chunk->len += len;
}

/* Transform a shm chunk to a buf chunk */
static void
chunk_shm_to_buf(struct VSLQ *vslq, struct chunk *chunk)
{
        struct vtx *vtx;
        struct chunk *buf;

        CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
        assert(chunk->type == chunk_t_shm);
        vtx = chunk->vtx;
        CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);

        buf = VTAILQ_PREV(chunk, chunkhead, list);
        if (buf != NULL && buf->type == chunk_t_buf)
                /* Previous is a buf chunk, append to it */
                chunk_appendbuf(buf, chunk->shm.start.ptr, chunk->len);
        else {
                /* Create a new buf chunk and insert it before this */
                buf = chunk_newbuf(vtx, chunk->shm.start.ptr, chunk->len);
                AN(buf);
                VTAILQ_INSERT_BEFORE(chunk, buf, list);
        }

        /* Reset cursor chunk pointer, vslc_vtx_next will set it correctly */
        vtx->c.chunk = NULL;

        /* Remove from the shmref list and vtx, and put chunk back
           on the free list */
        VTAILQ_REMOVE(&vslq->shmrefs, chunk, shm.shmref);
        VTAILQ_REMOVE(&vtx->chunks, chunk, list);
        VTAILQ_INSERT_HEAD(&vtx->shmchunks_free, chunk, list);
}

/* Append a set of records to a vtx structure */
static enum vsl_status
vtx_append(struct VSLQ *vslq, struct vtx *vtx, const struct VSLC_ptr *start,
    size_t len)
{
        struct chunk *chunk;
        enum vsl_check i;

        AN(vtx);
        AN(len);
        AN(start);

        i = VSL_Check(vslq->c, start);
        if (i == vsl_check_e_inval)
                return (vsl_e_overrun);

        if (i == vsl_check_valid && !VTAILQ_EMPTY(&vtx->shmchunks_free)) {
                /* Shmref it */
                chunk = VTAILQ_FIRST(&vtx->shmchunks_free);
                CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
                assert(chunk->type == chunk_t_shm);
                assert(chunk->vtx == vtx);
                VTAILQ_REMOVE(&vtx->shmchunks_free, chunk, list);
                chunk->shm.start = *start;
                chunk->len = len;
                VTAILQ_INSERT_TAIL(&vtx->chunks, chunk, list);

                /* Append to shmref list */
                VTAILQ_INSERT_TAIL(&vslq->shmrefs, chunk, shm.shmref);
        } else {
                /* Buffer it */
                chunk = VTAILQ_LAST(&vtx->chunks, chunkhead);
                CHECK_OBJ_ORNULL(chunk, CHUNK_MAGIC);
                if (chunk != NULL && chunk->type == chunk_t_buf) {
                        /* Tail is a buf chunk, append to that */
                        chunk_appendbuf(chunk, start->ptr, len);
                } else {
                        /* Append new buf chunk */
                        chunk = chunk_newbuf(vtx, start->ptr, len);
                        AN(chunk);
                        VTAILQ_INSERT_TAIL(&vtx->chunks, chunk, list);
                }
        }
        vtx->len += len;
        return (vsl_more);
}

/* Allocate a new vtx structure */
static struct vtx *
vtx_new(struct VSLQ *vslq)
{
        struct vtx *vtx;
        int i;

        AN(vslq);
        if (vslq->n_cache) {
                AZ(VTAILQ_EMPTY(&vslq->cache));
                vtx = VTAILQ_FIRST(&vslq->cache);
                VTAILQ_REMOVE(&vslq->cache, vtx, list_child);
                vslq->n_cache--;
        } else {
                ALLOC_OBJ(vtx, VTX_MAGIC);
                AN(vtx);

                VTAILQ_INIT(&vtx->child);
                VTAILQ_INIT(&vtx->shmchunks_free);
                for (i = 0; i < VTX_SHMCHUNKS; i++) {
                        vtx->shmchunks[i].magic = CHUNK_MAGIC;
                        vtx->shmchunks[i].type = chunk_t_shm;
                        vtx->shmchunks[i].vtx = vtx;
                        VTAILQ_INSERT_TAIL(&vtx->shmchunks_free,
                            &vtx->shmchunks[i], list);
                }
                VTAILQ_INIT(&vtx->chunks);
                VTAILQ_INIT(&vtx->synth);
                vtx->c.magic = VSLC_VTX_MAGIC;
                vtx->c.vtx = vtx;
                vtx->c.cursor.priv_tbl = &vslc_vtx_tbl;
                vtx->c.cursor.priv_data = &vtx->c;
        }

        CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
        vtx->key.vxid = 0;
        vtx->t_start = VTIM_mono();
        vtx->flags = 0;
        vtx->type = VSL_t_unknown;
        vtx->reason = VSL_r_unknown;
        vtx->parent = NULL;
        vtx->n_child = 0;
        vtx->n_childready = 0;
        vtx->n_descend = 0;
        vtx->len = 0;
        AN(vslc_vtx_reset(&vtx->c.cursor) == vsl_end);

        return (vtx);
}

/* Disuse a vtx and all it's children, freeing any resources held. Free or
   cache the vtx for later use */
static void
vtx_retire(struct VSLQ *vslq, struct vtx **pvtx)
{
        struct vtx *vtx;
        struct vtx *child;
        struct synth *synth;
        struct chunk *chunk;

        AN(vslq);
        TAKE_OBJ_NOTNULL(vtx, pvtx, VTX_MAGIC);

        assert((vtx->flags & (VTX_F_COMPLETE | VTX_F_READY)) == (VTX_F_COMPLETE | VTX_F_READY));
        AZ(vtx->parent);

        while (!VTAILQ_EMPTY(&vtx->child)) {
                child = VTAILQ_FIRST(&vtx->child);
                assert(child->parent == vtx);
                AN(vtx->n_child);
                assert(vtx->n_descend >= child->n_descend + 1);
                VTAILQ_REMOVE(&vtx->child, child, list_child);
                child->parent = NULL;
                vtx->n_child--;
                vtx->n_descend -= child->n_descend + 1;
                vtx_retire(vslq, &child);
                AZ(child);
        }
        AZ(vtx->n_child);
        AZ(vtx->n_descend);
        vtx->n_childready = 0;
        // remove rval is no way to check if element was present
        (void)VRBT_REMOVE(vtx_tree, &vslq->tree, &vtx->key);
        vtx->key.vxid = 0;
        vtx->flags = 0;

        while (!VTAILQ_EMPTY(&vtx->synth)) {
                synth = VTAILQ_FIRST(&vtx->synth);
                CHECK_OBJ_NOTNULL(synth, SYNTH_MAGIC);
                VTAILQ_REMOVE(&vtx->synth, synth, list);
                FREE_OBJ(synth);
        }

        while (!VTAILQ_EMPTY(&vtx->chunks)) {
                chunk = VTAILQ_FIRST(&vtx->chunks);
                CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
                VTAILQ_REMOVE(&vtx->chunks, chunk, list);
                if (chunk->type == chunk_t_shm) {
                        VTAILQ_REMOVE(&vslq->shmrefs, chunk, shm.shmref);
                        VTAILQ_INSERT_HEAD(&vtx->shmchunks_free, chunk, list);
                } else {
                        assert(chunk->type == chunk_t_buf);
                        chunk_freebuf(&chunk);
                        AZ(chunk);
                }
        }
        vtx->len = 0;
        AN(vslq->n_outstanding);
        vslq->n_outstanding--;

        if (vslq->n_cache < VTX_CACHE) {
                VTAILQ_INSERT_HEAD(&vslq->cache, vtx, list_child);
                vslq->n_cache++;
        } else
                FREE_OBJ(vtx);

}

/* Lookup a vtx by vxid from the managed list */
static struct vtx *
vtx_lookup(const struct VSLQ *vslq, uint64_t vxid)
{
        struct vtx_key lkey, *key;
        struct vtx *vtx;

        AN(vslq);
        lkey.vxid = vxid;
        key = VRBT_FIND(vtx_tree, &vslq->tree, &lkey);
        if (key == NULL)
                return (NULL);
        CAST_OBJ_NOTNULL(vtx, (void *)key, VTX_MAGIC);
        return (vtx);
}

/* Insert a new vtx into the managed list */
static struct vtx *
vtx_add(struct VSLQ *vslq, uint64_t vxid)
{
        struct vtx *vtx;

        AN(vslq);
        vtx = vtx_new(vslq);
        AN(vtx);
        vtx->key.vxid = vxid;
        AZ(VRBT_INSERT(vtx_tree, &vslq->tree, &vtx->key));
        VTAILQ_INSERT_TAIL(&vslq->incomplete, vtx, list_vtx);
        vslq->n_outstanding++;
        return (vtx);
}

/* Mark a vtx complete, update child counters and if possible push it or
   it's top parent to the ready state */
static void
vtx_mark_complete(struct VSLQ *vslq, struct vtx *vtx)
{

        AN(vslq);
        AN(vtx->flags & VTX_F_END);
        AZ(vtx->flags & VTX_F_COMPLETE);

        if (vtx->type == VSL_t_unknown)
                vtx_diag(vtx, "vtx of unknown type marked complete");

        vtx->flags |= VTX_F_COMPLETE;
        VTAILQ_REMOVE(&vslq->incomplete, vtx, list_vtx);

        while (1) {
                AZ(vtx->flags & VTX_F_READY);
                if (vtx->flags & VTX_F_COMPLETE &&
                    vtx->n_child == vtx->n_childready)
                        vtx->flags |= VTX_F_READY;
                else
                        return;
                if (vtx->parent == NULL) {
                        /* Top level vtx ready */
                        VTAILQ_INSERT_TAIL(&vslq->ready, vtx, list_vtx);
                        return;
                }
                vtx = vtx->parent;
                vtx->n_childready++;
                assert(vtx->n_child >= vtx->n_childready);
        }
}

/* Add a child to a parent, and update child counters */
static void
vtx_set_parent(struct vtx *parent, struct vtx *child)
{

        CHECK_OBJ_NOTNULL(parent, VTX_MAGIC);
        CHECK_OBJ_NOTNULL(child, VTX_MAGIC);
        assert(parent != child);
        AZ(parent->flags & VTX_F_COMPLETE);
        AZ(child->flags & VTX_F_COMPLETE);
        AZ(child->parent);
        child->parent = parent;
        VTAILQ_INSERT_TAIL(&parent->child, child, list_child);
        parent->n_child++;
        do
                parent->n_descend += 1 + child->n_descend;
        while ((parent = parent->parent) != NULL);
}

/* Parse a begin or link record. Returns the number of elements that was
   successfully parsed. */
static int
vtx_parse_link(const char *str, enum VSL_transaction_e *ptype,
    uint64_t *pvxid, enum VSL_reason_e *preason, uint64_t *psub)
{
        char type[16], reason[16];
        uintmax_t vxid, sub;
        int i;
        enum VSL_transaction_e et;
        enum VSL_reason_e er;

        AN(str);
        AN(ptype);
        AN(pvxid);
        AN(preason);

        i = sscanf(str, "%15s %ju %15s %ju", type, &vxid, reason, &sub);
        if (i < 1)
                return (0);

        /* transaction type */
        for (et = VSL_t_unknown; et < VSL_t__MAX; et++)
                if (!vstrcmp(type, vsl_t_names[et]))
                        break;
        if (et >= VSL_t__MAX)
                et = VSL_t_unknown;
        *ptype = et;
        if (i == 1)
                return (1);

        /* vxid */
        assert((vxid & ~VSL_IDENTMASK) == 0);
        *pvxid = vxid;
        if (i == 2)
                return (2);

        /* transaction reason */
        for (er = VSL_r_unknown; er < VSL_r__MAX; er++)
                if (!vstrcmp(reason, vsl_r_names[er]))
                        break;
        if (er >= VSL_r__MAX)
                er = VSL_r_unknown;
        *preason = er;
        if (i == 3)
                return (3);

        /* request sub-level */
        if (psub != NULL)
                *psub = sub;
        return (4);
}

/* Parse and process a begin record */
static int
vtx_scan_begin(struct VSLQ *vslq, struct vtx *vtx, const uint32_t *ptr)
{
        int i;
        enum VSL_transaction_e type;
        enum VSL_reason_e reason;
        uint64_t p_vxid;
        struct vtx *p_vtx;

        assert(VSL_TAG(ptr) == SLT_Begin);

        AZ(vtx->flags & VTX_F_READY);

        i = vtx_parse_link(VSL_CDATA(ptr), &type, &p_vxid, &reason, NULL);
        if (i < 3)
                return (vtx_diag_tag(vtx, ptr, "parse error"));
        if (type == VSL_t_unknown)
                (void)vtx_diag_tag(vtx, ptr, "unknown vxid type");

        /* Check/set vtx type */
        if (vtx->type != VSL_t_unknown && vtx->type != type)
                /* Type not matching the one previously set by a link
                   record */
                (void)vtx_diag_tag(vtx, ptr, "type mismatch");
        vtx->type = type;
        vtx->reason = reason;

        if (p_vxid == 0)
                /* Zero means no parent */
                return (0);
        if (p_vxid == vtx->key.vxid)
                return (vtx_diag_tag(vtx, ptr, "link to self"));

        if (vslq->grouping == VSL_g_vxid)
                return (0);     /* No links */
        if (vslq->grouping == VSL_g_request && vtx->type == VSL_t_req &&
            vtx->reason == VSL_r_rxreq)
                return (0);     /* No links */

        if (vtx->parent != NULL) {
                if (vtx->parent->key.vxid != p_vxid) {
                        /* This vtx already belongs to a different
                           parent */
                        return (vtx_diag_tag(vtx, ptr, "link mismatch"));
                } else
                        /* Link already exists */
                        return (0);
        }

        p_vtx = vtx_lookup(vslq, p_vxid);
        if (p_vtx == NULL) {
                /* Not seen parent yet. Create it. */
                p_vtx = vtx_add(vslq, p_vxid);
                AN(p_vtx);
        } else {
                CHECK_OBJ_NOTNULL(p_vtx, VTX_MAGIC);
                if (p_vtx->flags & VTX_F_COMPLETE)
                        return (vtx_diag_tag(vtx, ptr, "link too late"));
        }

        /* Create link */
        vtx_set_parent(p_vtx, vtx);

        return (0);
}

/* Parse and process a link record */
static int
vtx_scan_link(struct VSLQ *vslq, struct vtx *vtx, const uint32_t *ptr)
{
        int i;
        enum VSL_transaction_e c_type;
        enum VSL_reason_e c_reason;
        uint64_t c_vxid;
        struct vtx *c_vtx;

        assert(VSL_TAG(ptr) == SLT_Link);

        AZ(vtx->flags & VTX_F_READY);

        i = vtx_parse_link(VSL_CDATA(ptr), &c_type, &c_vxid, &c_reason, NULL);
        if (i < 3)
                return (vtx_diag_tag(vtx, ptr, "parse error"));
        if (c_type == VSL_t_unknown)
                (void)vtx_diag_tag(vtx, ptr, "unknown vxid type");

        if (vslq->grouping == VSL_g_vxid)
                return (0);     /* No links */
        if (vslq->grouping == VSL_g_request && vtx->type == VSL_t_sess)
                return (0);     /* No links */

        if (c_vxid == 0)
                return (vtx_diag_tag(vtx, ptr, "illegal link vxid"));
        if (c_vxid == vtx->key.vxid)
                return (vtx_diag_tag(vtx, ptr, "link to self"));

        /* Lookup and check child vtx */
        c_vtx = vtx_lookup(vslq, c_vxid);
        if (c_vtx == NULL) {
                /* Child not seen before. Insert it and create link */
                c_vtx = vtx_add(vslq, c_vxid);
                AN(c_vtx);
                AZ(c_vtx->parent);
                c_vtx->type = c_type;
                c_vtx->reason = c_reason;
                vtx_set_parent(vtx, c_vtx);
                return (0);
        }

        CHECK_OBJ(c_vtx, VTX_MAGIC);
        if (c_vtx->parent == vtx)
                /* Link already exists */
                return (0);
        if (c_vtx->parent != NULL && c_vtx->parent != vtx)
                return (vtx_diag_tag(vtx, ptr, "duplicate link"));
        if (c_vtx->flags & VTX_F_COMPLETE)
                return (vtx_diag_tag(vtx, ptr, "link too late"));
        if (c_vtx->type != VSL_t_unknown && c_vtx->type != c_type)
                (void)vtx_diag_tag(vtx, ptr, "type mismatch");

        c_vtx->type = c_type;
        c_vtx->reason = c_reason;
        vtx_set_parent(vtx, c_vtx);
        return (0);
}

/* Scan the records of a vtx, performing processing actions on specific
   records */
static void
vtx_scan(struct VSLQ *vslq, struct vtx *vtx)
{
        const uint32_t *ptr;
        enum VSL_tag_e tag;

        while (!(vtx->flags & VTX_F_COMPLETE) &&
            vslc_vtx_next(&vtx->c.cursor) == 1) {
                ptr = vtx->c.cursor.rec.ptr;
                if (VSL_ID(ptr) != vtx->key.vxid) {
                        (void)vtx_diag_tag(vtx, ptr, "vxid mismatch");
                        continue;
                }

                tag = VSL_TAG(ptr);
                assert(tag != SLT__Batch);

                switch (tag) {
                case SLT_Begin:
                        if (vtx->flags & VTX_F_BEGIN)
                                (void)vtx_diag_tag(vtx, ptr, "duplicate begin");
                        else {
                                (void)vtx_scan_begin(vslq, vtx, ptr);
                                vtx->flags |= VTX_F_BEGIN;
                        }
                        break;

                case SLT_Link:
                        (void)vtx_scan_link(vslq, vtx, ptr);
                        break;

                case SLT_End:
                        AZ(vtx->flags & VTX_F_END);
                        vtx->flags |= VTX_F_END;
                        vtx_mark_complete(vslq, vtx);
                        break;

                default:
                        break;
                }
        }
}

/* Force a vtx into complete status by synthing the necessary outstanding
   records */
static void
vtx_force(struct VSLQ *vslq, struct vtx *vtx, const char *reason)
{

        AZ(vtx->flags & (VTX_F_COMPLETE | VTX_F_READY));
        vtx_scan(vslq, vtx);
        if (!(vtx->flags & VTX_F_BEGIN))
                vtx_synth_rec(vtx, SLT_Begin, "%s %u synth",
                    vsl_t_names[vtx->type], 0);
        vtx_diag(vtx, reason);
        if (!(vtx->flags & VTX_F_END))
                vtx_synth_rec(vtx, SLT_End, "synth");
        vtx_scan(vslq, vtx);
        AN(vtx->flags & VTX_F_COMPLETE);
}

static int
vslq_ratelimit(struct VSLQ *vslq)
{
        vtim_mono now;
        vtim_dur delta;

        CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);
        CHECK_OBJ_NOTNULL(vslq->vsl, VSL_MAGIC);

        now = VTIM_mono();
        delta = now - vslq->last_use;
        vslq->credits += (delta / vslq->vsl->R_opt_p) * vslq->vsl->R_opt_l;
        vslq->credits = vmin_t(double, vslq->credits, vslq->vsl->R_opt_l);
        vslq->last_use = now;

        if (vslq->credits < 1.0)
                return (0);

        vslq->credits -= 1.0;
        return (1);
}

/* Build transaction array, do the query and callback. Returns 0 or the
   return value from func */
static int
vslq_callback(struct VSLQ *vslq, struct vtx *vtx, VSLQ_dispatch_f *func,
    void *priv)
{
        unsigned n = vtx->n_descend + 1;
        struct vtx *vtxs[n];
        struct VSL_transaction trans[n];
        struct VSL_transaction *ptrans[n + 1];
        unsigned i, j;

        AN(vslq);
        CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
        AN(vtx->flags & VTX_F_READY);
        AN(func);

        if (vslq->grouping == VSL_g_session &&
            vtx->type != VSL_t_sess)
                return (0);
        if (vslq->grouping == VSL_g_request &&
            vtx->type != VSL_t_req)
                return (0);

        /* Build transaction array */
        AN(vslc_vtx_reset(&vtx->c.cursor) == vsl_end);
        vtxs[0] = vtx;
        trans[0].level = 1;
        trans[0].vxid = vtx->key.vxid;
        trans[0].vxid_parent = 0;
        trans[0].type = vtx->type;
        trans[0].reason = vtx->reason;
        trans[0].c = &vtx->c.cursor;
        i = 1;
        j = 0;
        while (j < i) {
                VTAILQ_FOREACH(vtx, &vtxs[j]->child, list_child) {
                        assert(i < n);
                        AN(vslc_vtx_reset(&vtx->c.cursor) == vsl_end);
                        vtxs[i] = vtx;
                        if (vtx->reason == VSL_r_restart)
                                /* Restarts stay at the same level as parent */
                                trans[i].level = trans[j].level;
                        else
                                trans[i].level = trans[j].level + 1;
                        trans[i].vxid = vtx->key.vxid;
                        trans[i].vxid_parent = trans[j].vxid;
                        trans[i].type = vtx->type;
                        trans[i].reason = vtx->reason;
                        trans[i].c = &vtx->c.cursor;
                        i++;
                }
                j++;
        }
        assert(i == n);

        /* Build pointer array */
        for (i = 0; i < n; i++)
                ptrans[i] = &trans[i];
        ptrans[i] = NULL;

        /* Query test goes here */
        if (vslq->query != NULL && !vslq_runquery(vslq->query, ptrans))
                return (0);

        if (vslq->vsl->R_opt_l != 0 && !vslq_ratelimit(vslq))
                return (0);

        /* Callback */
        return ((func)(vslq->vsl, ptrans, priv));
}

/* Create a synthetic log record. The record will be inserted at the
   current cursor offset */
static void
vtx_synth_rec(struct vtx *vtx, unsigned tag, const char *fmt, ...)
{
        struct synth *synth, *it;
        va_list ap;
        char *buf;
        int l, buflen;
        uint64_t vxid;

        ALLOC_OBJ(synth, SYNTH_MAGIC);
        AN(synth);

        buf = VSL_DATA(synth->data);
        buflen = sizeof(synth->data) - VSL_BYTES(VSL_OVERHEAD);
        va_start(ap, fmt);
        l = vsnprintf(buf, buflen, fmt, ap);
        assert(l >= 0);
        va_end(ap);
        if (l > buflen - 1)
                l = buflen - 1;
        buf[l++] = '\0';        /* NUL-terminated */
        vxid = vtx->key.vxid;
        switch (vtx->type) {
        case VSL_t_req:
                vxid |= VSL_CLIENTMARKER;
                break;
        case VSL_t_bereq:
                vxid |= VSL_BACKENDMARKER;
                break;
        default:
                break;
        }
        synth->data[2] = vxid >> 32;
        synth->data[1] = vxid;
        synth->data[0] = (((tag & VSL_IDMASK) << VSL_IDSHIFT) |
            (VSL_VERSION_3 << VSL_VERSHIFT) | l);
        synth->offset = vtx->c.offset;

        VTAILQ_FOREACH_REVERSE(it, &vtx->synth, synthhead, list) {
                /* Make sure the synth list is sorted on offset */
                CHECK_OBJ_NOTNULL(it, SYNTH_MAGIC);
                if (synth->offset >= it->offset)
                        break;
        }
        if (it != NULL)
                VTAILQ_INSERT_AFTER(&vtx->synth, it, synth, list);
        else
                VTAILQ_INSERT_HEAD(&vtx->synth, synth, list);

        /* Update cursor */
        CHECK_OBJ_ORNULL(vtx->c.synth, SYNTH_MAGIC);
        if (vtx->c.synth == NULL || vtx->c.synth->offset > synth->offset)
                vtx->c.synth = synth;
}

/* Add a diagnostic SLT_VSL synth record to the vtx. */
static int
vtx_diag(struct vtx *vtx, const char *msg)
{

        vtx_synth_rec(vtx, SLT_VSL, msg);
        return (-1);
}

/* Add a SLT_VSL diag synth record to the vtx. Takes an offending record
   that will be included in the log record */
static int
vtx_diag_tag(struct vtx *vtx, const uint32_t *ptr, const char *reason)
{

        vtx_synth_rec(vtx, SLT_VSL, "%s (%ju:%s \"%.*s\")", reason, VSL_ID(ptr),
            VSL_tags[VSL_TAG(ptr)], (int)VSL_LEN(ptr), VSL_CDATA(ptr));
        return (-1);
}

struct VSLQ *
VSLQ_New(struct VSL_data *vsl, struct VSL_cursor **cp,
    enum VSL_grouping_e grouping, const char *querystring)
{
        struct vslq_query *query;
        struct VSLQ *vslq;

        CHECK_OBJ_NOTNULL(vsl, VSL_MAGIC);
        if (grouping >= VSL_g__MAX) {
                (void)vsl_diag(vsl, "Illegal query grouping");
                return (NULL);
        }
        if (querystring != NULL) {
                query = vslq_newquery(vsl, grouping, querystring);
                if (query == NULL)
                        return (NULL);
        } else
                query = NULL;

        ALLOC_OBJ(vslq, VSLQ_MAGIC);
        AN(vslq);
        vslq->vsl = vsl;
        if (cp != NULL) {
                vslq->c = *cp;
                *cp = NULL;
        }
        vslq->grouping = grouping;
        vslq->query = query;
        if (vslq->vsl->R_opt_l != 0) {
                vslq->last_use = VTIM_mono();
                vslq->credits = 1;
        }

        /* Setup normal mode */
        VRBT_INIT(&vslq->tree);
        VTAILQ_INIT(&vslq->ready);
        VTAILQ_INIT(&vslq->incomplete);
        VTAILQ_INIT(&vslq->shmrefs);
        VTAILQ_INIT(&vslq->cache);

        /* Setup raw mode */
        vslq->raw.c.magic = VSLC_RAW_MAGIC;
        vslq->raw.c.cursor.priv_tbl = &vslc_raw_tbl;
        vslq->raw.c.cursor.priv_data = &vslq->raw.c;
        vslq->raw.trans.level = 0;
        vslq->raw.trans.type = VSL_t_raw;
        vslq->raw.trans.reason = VSL_r_unknown;
        vslq->raw.trans.c = &vslq->raw.c.cursor;
        vslq->raw.ptrans[0] = &vslq->raw.trans;
        vslq->raw.ptrans[1] = NULL;

        return (vslq);
}

void
VSLQ_Delete(struct VSLQ **pvslq)
{
        struct VSLQ *vslq;
        struct vtx *vtx;

        TAKE_OBJ_NOTNULL(vslq, pvslq, VSLQ_MAGIC);

        (void)VSLQ_Flush(vslq, NULL, NULL);
        AZ(vslq->n_outstanding);

        if (vslq->c != NULL) {
                VSL_DeleteCursor(vslq->c);
                vslq->c = NULL;
        }

        if (vslq->query != NULL)
                vslq_deletequery(&vslq->query);
        AZ(vslq->query);

        while (!VTAILQ_EMPTY(&vslq->cache)) {
                AN(vslq->n_cache);
                vtx = VTAILQ_FIRST(&vslq->cache);
                CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
                VTAILQ_REMOVE(&vslq->cache, vtx, list_child);
                vslq->n_cache--;
                FREE_OBJ(vtx);
        }

        FREE_OBJ(vslq);
}

void
VSLQ_SetCursor(struct VSLQ *vslq, struct VSL_cursor **cp)
{

        CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);

        if (vslq->c != NULL) {
                (void)VSLQ_Flush(vslq, NULL, NULL);
                AZ(vslq->n_outstanding);
                VSL_DeleteCursor(vslq->c);
                vslq->c = NULL;
        }

        if (cp != NULL) {
                AN(*cp);
                vslq->c = *cp;
                *cp = NULL;
        }
}

/* Regard each log line as a single transaction, feed it through the query
   and do the callback */
static int
vslq_raw(struct VSLQ *vslq, VSLQ_dispatch_f *func, void *priv)
{
        enum vsl_status r = vsl_more;
        int i;

        assert(vslq->grouping == VSL_g_raw);

        assert(vslq->raw.offset <= vslq->raw.len);
        do {
                if (vslq->raw.offset == vslq->raw.len) {
                        r = VSL_Next(vslq->c);
                        if (r != vsl_more)
                                return (r);
                        AN(vslq->c->rec.ptr);
                        vslq->raw.start = vslq->c->rec;
                        if (VSL_TAG(vslq->c->rec.ptr) == SLT__Batch)
                                vslq->raw.len = VSL_END(vslq->c->rec.ptr,
                                    VSL_BATCHLEN(vslq->c->rec.ptr)) -
                                    vslq->c->rec.ptr;
                        else
                                vslq->raw.len = VSL_NEXT(vslq->raw.start.ptr) -
                                    vslq->raw.start.ptr;
                        assert(vslq->raw.len > 0);
                        vslq->raw.offset = 0;
                }

                vslq->raw.c.ptr = vslq->raw.start.ptr + vslq->raw.offset;
                vslq->raw.c.cursor.rec.ptr = NULL;
                vslq->raw.trans.vxid = VSL_ID(vslq->raw.c.ptr);
                vslq->raw.offset += VSL_NEXT(vslq->raw.c.ptr) - vslq->raw.c.ptr;
        } while (VSL_TAG(vslq->raw.c.ptr) == SLT__Batch);

        assert (r == vsl_more);

        if (func == NULL)
                return (r);

        if (vslq->query != NULL &&
            !vslq_runquery(vslq->query, vslq->raw.ptrans))
                return (r);

        if (vslq->vsl->R_opt_l != 0 && !vslq_ratelimit(vslq))
                return (r);

        i = (func)(vslq->vsl, vslq->raw.ptrans, priv);
        if (i)
                return (i);

        return (r);
}

/* Check the beginning of the shmref list, and buffer refs that are at
 * warning level.
 */
static enum vsl_status
vslq_shmref_check(struct VSLQ *vslq)
{
        struct chunk *chunk;
        enum vsl_check i;

        while ((chunk = VTAILQ_FIRST(&vslq->shmrefs)) != NULL) {
                CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
                assert(chunk->type == chunk_t_shm);
                i = VSL_Check(vslq->c, &chunk->shm.start);
                switch (i) {
                case vsl_check_valid:
                        /* First on list is OK, refs behind it must also
                           be OK */
                        return (vsl_more);
                case vsl_check_warn:
                        /* Buffer this chunk */
                        chunk_shm_to_buf(vslq, chunk);
                        break;
                default:
                        /* Too late to buffer */
                        return (vsl_e_overrun);
                }
        }

        return (vsl_more);
}

static unsigned
vslq_candidate(struct VSLQ *vslq, const uint32_t *ptr)
{
        enum VSL_transaction_e type;
        enum VSL_reason_e reason;
        struct VSL_data *vsl;
        enum VSL_tag_e tag;
        uint64_t p_vxid, sub;
        int i;

        CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);
        AN(ptr);

        assert(vslq->grouping != VSL_g_raw);
        if (vslq->grouping == VSL_g_session)
                return (1); /* All are needed */

        vsl = vslq->vsl;
        CHECK_OBJ_NOTNULL(vsl, VSL_MAGIC);
        if (vslq->grouping == VSL_g_vxid) {
                if (!vsl->c_opt && !vsl->b_opt)
                        AZ(vsl->E_opt);
                else if (!vsl->b_opt && !VSL_CLIENT(ptr))
                        return (0);
                else if (!vsl->c_opt && !VSL_BACKEND(ptr))
                        return (0);
                /* Need to parse the Begin tag - fallthrough to below */
        }

        tag = VSL_TAG(ptr);
        assert(tag == SLT_Begin);
        i = vtx_parse_link(VSL_CDATA(ptr), &type, &p_vxid, &reason, &sub);
        if (i < 3 || type == VSL_t_unknown)
                return (0);

        if (vslq->grouping == VSL_g_request && type == VSL_t_sess)
                return (0);

        if (vslq->grouping == VSL_g_vxid && i > 3 && sub > 0 && !vsl->E_opt)
                return (0);

        return (1);
}

/* Process next input record */
static enum vsl_status
vslq_next(struct VSLQ *vslq)
{
        const uint32_t *ptr;
        struct VSL_cursor *c;
        enum vsl_status r;
        enum VSL_tag_e tag;
        ssize_t len;
        uint64_t vxid;
        unsigned keep;
        struct vtx *vtx;

        c = vslq->c;
        r = VSL_Next(c);
        if (r != vsl_more)
                return (r);

        assert (r == vsl_more);

        tag = (enum VSL_tag_e)VSL_TAG(c->rec.ptr);
        if (tag == SLT__Batch) {
                vxid = VSL_BATCHID(c->rec.ptr);
                len = VSL_END(c->rec.ptr, VSL_BATCHLEN(c->rec.ptr)) -
                    c->rec.ptr;
                if (len == 0)
                        return (r);
                ptr = VSL_NEXT(c->rec.ptr);
                tag = (enum VSL_tag_e)VSL_TAG(ptr);
        } else {
                vxid = VSL_ID(c->rec.ptr);
                len = VSL_NEXT(c->rec.ptr) - c->rec.ptr;
                ptr = c->rec.ptr;
        }
        assert(len > 0);
        if (vxid == 0)
                /* Skip non-transactional records */
                return (r);

        vtx = vtx_lookup(vslq, vxid);
        keep = tag != SLT_Begin || vslq_candidate(vslq, ptr);
        if (vtx == NULL && tag == SLT_Begin && keep) {
                vtx = vtx_add(vslq, vxid);
                AN(vtx);
        }
        if (vtx != NULL) {
                AN(keep);
                r = vtx_append(vslq, vtx, &c->rec, len);
                if (r == vsl_more)
                        vtx_scan(vslq, vtx);
        }

        return (r);
}

/* Test query and report any ready transactions */
static int
vslq_process_ready(struct VSLQ *vslq, VSLQ_dispatch_f *func, void *priv)
{
        struct vtx *vtx;
        int i = 0;

        AN(vslq);

        while (!VTAILQ_EMPTY(&vslq->ready)) {
                vtx = VTAILQ_FIRST(&vslq->ready);
                CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
                VTAILQ_REMOVE(&vslq->ready, vtx, list_vtx);
                AN(vtx->flags & VTX_F_READY);
                if (func != NULL)
                        i = vslq_callback(vslq, vtx, func, priv);
                vtx_retire(vslq, &vtx);
                AZ(vtx);
                if (i)
                        return (i);
        }

        return (0);
}

/* Process the input cursor, calling the callback function on matching
   transaction sets */
int
VSLQ_Dispatch(struct VSLQ *vslq, VSLQ_dispatch_f *func, void *priv)
{
        enum vsl_status r;
        int i;
        double now;
        struct vtx *vtx;

        CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);

        /* Check that we have a cursor */
        if (vslq->c == NULL)
                return (vsl_e_abandon);

        if (vslq->grouping == VSL_g_raw)
                return (vslq_raw(vslq, func, priv));

        /* Process next cursor input */
        r = vslq_next(vslq);
        if (r != vsl_more)
                /* At end of log or cursor reports error condition */
                return (r);

        /* Check shmref list and buffer if necessary */
        r = vslq_shmref_check(vslq);
        if (r != vsl_more)
                /* Buffering of shm ref failed */
                return (r);

        assert (r == vsl_more);

        /* Check vtx timeout */
        now = VTIM_mono();
        while (!VTAILQ_EMPTY(&vslq->incomplete)) {
                vtx = VTAILQ_FIRST(&vslq->incomplete);
                CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
                if (now - vtx->t_start < vslq->vsl->T_opt)
                        break;
                vtx_force(vslq, vtx, "timeout");
                AN(vtx->flags & VTX_F_COMPLETE);
        }

        /* Check store limit */
        while (vslq->n_outstanding > vslq->vsl->L_opt &&
            !(VTAILQ_EMPTY(&vslq->incomplete))) {
                vtx = VTAILQ_FIRST(&vslq->incomplete);
                CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
                vtx_force(vslq, vtx, "store overflow");
                AN(vtx->flags & VTX_F_COMPLETE);
                i = vslq_process_ready(vslq, func, priv);
                if (i)
                        /* User return code */
                        return (i);
        }

        /* Check ready list */
        if (!VTAILQ_EMPTY(&vslq->ready)) {
                i = vslq_process_ready(vslq, func, priv);
                if (i)
                        /* User return code */
                        return (i);
        }

        return (vsl_more);
}

/* Flush any incomplete vtx held on to. Do callbacks if func != NULL */
int
VSLQ_Flush(struct VSLQ *vslq, VSLQ_dispatch_f *func, void *priv)
{
        struct vtx *vtx;

        CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);

        while (!VTAILQ_EMPTY(&vslq->incomplete)) {
                vtx = VTAILQ_FIRST(&vslq->incomplete);
                CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
                AZ(vtx->flags & VTX_F_COMPLETE);
                vtx_force(vslq, vtx, "flush");
        }

        return (vslq_process_ready(vslq, func, priv));
}

		vinyl-cache/lib/libvinylapi/vsl_dispatch.c
0		/*-
1		* Copyright (c) 2006 Verdens Gang AS
2		* Copyright (c) 2006-2015 Varnish Software AS
3		* All rights reserved.
4		*
5		* Author: Martin Blix Grydeland <martin@varnish-software.com>
6		*
7		* SPDX-License-Identifier: BSD-2-Clause
8		*
9		* Redistribution and use in source and binary forms, with or without
10		* modification, are permitted provided that the following conditions
11		* are met:
12		* 1. Redistributions of source code must retain the above copyright
13		* notice, this list of conditions and the following disclaimer.
14		* 2. Redistributions in binary form must reproduce the above copyright
15		* notice, this list of conditions and the following disclaimer in the
16		* documentation and/or other materials provided with the distribution.
17		*
18		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
19		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
22		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
24		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
27		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
28		* SUCH DAMAGE.
29		*
30		*/
31
32		#include "config.h"
33
34		#include <stdarg.h>
35		#include <stdint.h>
36		#include <stdio.h>
37		#include <stdlib.h>
38		#include <string.h>
39
40		#include "vdef.h"
41		#include "vas.h"
42		#include "miniobj.h"
43
44		#include "vqueue.h"
45		#include "vre.h"
46		#include "vtim.h"
47		#include "vtree.h"
48
49		#include "vapi/vsl.h"
50
51		#include "vsl_api.h"
52
53		#define VTX_CACHE 10
54		#define VTX_BUFSIZE_MIN 64
55		#define VTX_SHMCHUNKS 3
56
57		static const char * const vsl_t_names[VSL_t__MAX] = {
58		[VSL_t_unknown] = "unknown",
59		[VSL_t_sess] = "sess",
60		[VSL_t_req] = "req",
61		[VSL_t_bereq] = "bereq",
62		[VSL_t_raw] = "raw",
63		};
64
65		static const char * const vsl_r_names[VSL_r__MAX] = {
66		[VSL_r_unknown] = "unknown",
67		[VSL_r_http_1] = "HTTP/1",
68		[VSL_r_rxreq] = "rxreq",
69		[VSL_r_esi] = "esi",
70		[VSL_r_restart] = "restart",
71		[VSL_r_pass] = "pass",
72		[VSL_r_fetch] = "fetch",
73		[VSL_r_bgfetch] = "bgfetch",
74		[VSL_r_pipe] = "pipe",
75		};
76
77		struct vtx;
78		VTAILQ_HEAD(vtxhead, vtx);
79
80		struct vslc_raw {
81		unsigned magic;
82		#define VSLC_RAW_MAGIC 0x247EBD44
83
84		struct VSL_cursor cursor;
85
86		const uint32_t *ptr;
87		};
88
89		struct synth {
90		unsigned magic;
91		#define SYNTH_MAGIC 0xC654479F
92
93		VTAILQ_ENTRY(synth) list;
94		size_t offset;
95		uint32_t data[VSL_OVERHEAD + VSL_WORDS(64)];
96		};
97		VTAILQ_HEAD(synthhead, synth);
98
99		enum chunk_t {
100		chunk_t__unassigned,
101		chunk_t_shm,
102		chunk_t_buf,
103		};
104
105		struct chunk {
106		unsigned magic;
107		#define CHUNK_MAGIC 0x48DC0194
108		enum chunk_t type;
109		union {
110		struct {
111		struct VSLC_ptr start;
112		VTAILQ_ENTRY(chunk) shmref;
113		} shm;
114		struct {
115		uint32_t *data;
116		size_t space;
117		} buf;
118		};
119		size_t len;
120		struct vtx *vtx;
121		VTAILQ_ENTRY(chunk) list;
122		};
123		VTAILQ_HEAD(chunkhead, chunk);
124
125		struct vslc_vtx {
126		unsigned magic;
127		#define VSLC_VTX_MAGIC 0x74C6523F
128
129		struct VSL_cursor cursor;
130
131		struct vtx *vtx;
132		struct synth *synth;
133		struct chunk *chunk;
134		size_t chunkstart;
135		size_t offset;
136		};
137
138		struct vtx_key {
139		uint64_t vxid;
140		VRBT_ENTRY(vtx_key) entry;
141		};
142		VRBT_HEAD(vtx_tree, vtx_key);
143
144		struct vtx {
145		struct vtx_key key;
146		unsigned magic;
147		#define VTX_MAGIC 0xACC21D09
148		VTAILQ_ENTRY(vtx) list_child;
149		VTAILQ_ENTRY(vtx) list_vtx;
150
151		double t_start;
152		unsigned flags;
153		#define VTX_F_BEGIN 0x1 /* Begin record processed */
154		#define VTX_F_END 0x2 /* End record processed */
155		#define VTX_F_COMPLETE 0x4 /* Marked complete. No new children
156		should be appended */
157		#define VTX_F_READY 0x8 /* This vtx and all it's children are
158		complete */
159
160		enum VSL_transaction_e type;
161		enum VSL_reason_e reason;
162
163		struct vtx *parent;
164		struct vtxhead child;
165		unsigned n_child;
166		unsigned n_childready;
167		unsigned n_descend;
168
169		struct synthhead synth;
170
171		struct chunk shmchunks[VTX_SHMCHUNKS];
172		struct chunkhead shmchunks_free;
173
174		struct chunkhead chunks;
175		size_t len;
176
177		struct vslc_vtx c;
178		};
179
180		struct VSLQ {
181		unsigned magic;
182		#define VSLQ_MAGIC 0x23A8BE97
183
184		struct VSL_data *vsl;
185		struct VSL_cursor *c;
186		struct vslq_query *query;
187
188		enum VSL_grouping_e grouping;
189
190		/* Structured mode */
191		struct vtx_tree tree;
192		struct vtxhead ready;
193		struct vtxhead incomplete;
194		int n_outstanding;
195		struct chunkhead shmrefs;
196		struct vtxhead cache;
197		unsigned n_cache;
198
199		/* Rate limiting */
200		double credits;
201		vtim_mono last_use;
202
203		/* Raw mode */
204		struct {
205		struct vslc_raw c;
206		struct VSL_transaction trans;
207		struct VSL_transaction *ptrans[2];
208		struct VSLC_ptr start;
209		ssize_t len;
210		ssize_t offset;
211		} raw;
212		};
213
214		static void vtx_synth_rec(struct vtx vtx, unsigned tag, const char fmt, ...);
215		/lint -esym(534, vtx_diag) /
216		static int vtx_diag(struct vtx vtx, const char msg);
217		/lint -esym(534, vtx_diag_tag) /
218		static int vtx_diag_tag(struct vtx vtx, const uint32_t ptr,
219		const char *reason);
220
221		static inline int
222	55403	vtx_keycmp(const struct vtx_key a, const struct vtx_key b)
223		{
224	55403	if (a->vxid < b->vxid)
225	1588	return (-1);
226	53815	if (a->vxid > b->vxid)
227	22100	return (1);
228	31715	return (0);
229	55403	}
230
231	3481	VRBT_GENERATE_REMOVE_COLOR(vtx_tree, vtx_key, entry, static)
232	8628	VRBT_GENERATE_REMOVE(vtx_tree, vtx_key, entry, static)
233	4322	VRBT_GENERATE_INSERT_COLOR(vtx_tree, vtx_key, entry, static)
234	5320	VRBT_GENERATE_INSERT_FINISH(vtx_tree, vtx_key, entry, static)
235	9784	VRBT_GENERATE_INSERT(vtx_tree, vtx_key, entry, vtx_keycmp, static)
236	59239	VRBT_GENERATE_FIND(vtx_tree, vtx_key, entry, vtx_keycmp, static)
237
238		static enum vsl_status v_matchproto_(vslc_next_f)
239	115976	vslc_raw_next(const struct VSL_cursor *cursor)
240		{
241		struct vslc_raw *c;
242
243	115976	CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_RAW_MAGIC);
244	115976	assert(&c->cursor == cursor);
245
246	115976	AN(c->ptr);
247	115976	if (c->cursor.rec.ptr == NULL) {
248	58390	c->cursor.rec.ptr = c->ptr;
249	58390	return (vsl_more);
250		} else {
251	57586	c->cursor.rec.ptr = NULL;
252	57586	return (vsl_end);
253		}
254	115976	}
255
256		static enum vsl_status v_matchproto_(vslc_reset_f)
257	15444	vslc_raw_reset(const struct VSL_cursor *cursor)
258		{
259		struct vslc_raw *c;
260
261	15444	CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_RAW_MAGIC);
262	15444	assert(&c->cursor == cursor);
263
264	15444	AN(c->ptr);
265	15444	c->cursor.rec.ptr = NULL;
266
267	15444	return (vsl_end);
268		}
269
270		static const struct vslc_tbl vslc_raw_tbl = {
271		.magic = VSLC_TBL_MAGIC,
272		.delete = NULL,
273		.next = vslc_raw_next,
274		.reset = vslc_raw_reset,
275		.check = NULL,
276		};
277
278		static enum vsl_status v_matchproto_(vslc_next_f)
279	334218	vslc_vtx_next(const struct VSL_cursor *cursor)
280		{
281		struct vslc_vtx *c;
282		const uint32_t *ptr;
283		unsigned overrun;
284
285	334218	CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_VTX_MAGIC);
286	334218	assert(&c->cursor == cursor);
287	334218	CHECK_OBJ_NOTNULL(c->vtx, VTX_MAGIC);
288
289	334218	do {
290	362476	CHECK_OBJ_ORNULL(c->synth, SYNTH_MAGIC);
291	362476	if (c->synth != NULL && c->synth->offset == c->offset) {
292		/* We're at the offset of the next synth record,
293		point to it and advance the pointer */
294	1699	c->cursor.rec.ptr = c->synth->data;
295	1699	c->synth = VTAILQ_NEXT(c->synth, list);
296	1699	} else {
297	360777	overrun = c->offset > c->vtx->len;
298	360777	AZ(overrun);
299	360777	if (c->offset == c->vtx->len)
300	35587	return (vsl_end);
301
302		/* Advance chunk pointer */
303	325190	if (c->chunk == NULL) {
304	10036	c->chunk = VTAILQ_FIRST(&c->vtx->chunks);
305	10036	c->chunkstart = 0;
306	10036	}
307	325190	CHECK_OBJ_NOTNULL(c->chunk, CHUNK_MAGIC);
308	334267	while (c->offset >= c->chunkstart + c->chunk->len) {
309	9077	c->chunkstart += c->chunk->len;
310	9077	c->chunk = VTAILQ_NEXT(c->chunk, list);
311	9077	CHECK_OBJ_NOTNULL(c->chunk, CHUNK_MAGIC);
312		}
313
314		/* Point to the next stored record */
315	325190	if (c->chunk->type == chunk_t_shm)
316	262237	ptr = c->chunk->shm.start.ptr;
317		else {
318	62953	assert(c->chunk->type == chunk_t_buf);
319	62953	ptr = c->chunk->buf.data;
320		}
321	325190	c->cursor.rec.ptr = ptr + c->offset - c->chunkstart;
322	650380	c->offset += VSL_NEXT(c->cursor.rec.ptr) -
323	325190	c->cursor.rec.ptr;
324		}
325	326889	} while (VSL_TAG(c->cursor.rec.ptr) == SLT__Batch);
326
327	298631	return (vsl_more);
328	334218	}
329
330		static enum vsl_status v_matchproto_(vslc_reset_f)
331	13702	vslc_vtx_reset(const struct VSL_cursor *cursor)
332		{
333		struct vslc_vtx *c;
334
335	13702	CAST_OBJ_NOTNULL(c, cursor->priv_data, VSLC_VTX_MAGIC);
336	13702	assert(&c->cursor == cursor);
337	13702	CHECK_OBJ_NOTNULL(c->vtx, VTX_MAGIC);
338	13702	c->synth = VTAILQ_FIRST(&c->vtx->synth);
339	13702	c->chunk = NULL;
340	13702	c->chunkstart = 0;
341	13702	c->offset = 0;
342	13702	c->cursor.rec.ptr = NULL;
343
344	13702	return (vsl_end);
345		}
346
347		static const struct vslc_tbl vslc_vtx_tbl = {
348		.magic = VSLC_TBL_MAGIC,
349		.delete = NULL,
350		.next = vslc_vtx_next,
351		.reset = vslc_vtx_reset,
352		.check = NULL,
353		};
354
355		/* Create a buf chunk */
356		static struct chunk *
357	1874	chunk_newbuf(struct vtx vtx, const uint32_t ptr, size_t len)
358		{
359		struct chunk *chunk;
360
361	1874	ALLOC_OBJ(chunk, CHUNK_MAGIC);
362	1874	XXXAN(chunk);
363	1874	chunk->type = chunk_t_buf;
364	1874	chunk->vtx = vtx;
365	1874	chunk->buf.space = VTX_BUFSIZE_MIN;
366	1874	while (chunk->buf.space < len)
367	0	chunk->buf.space *= 2;
368	1874	chunk->buf.data = malloc(sizeof (uint32_t) * chunk->buf.space);
369	1874	AN(chunk->buf.data);
370	1874	vmemcpy(chunk->buf.data, ptr, sizeof(uint32_t) * len);
371	1874	chunk->len = len;
372	1874	return (chunk);
373		}
374
375		/* Free a buf chunk */
376		static void
377	1873	chunk_freebuf(struct chunk **pchunk)
378		{
379		struct chunk *chunk;
380
381	1873	TAKE_OBJ_NOTNULL(chunk, pchunk, CHUNK_MAGIC);
382	1873	assert(chunk->type == chunk_t_buf);
383	1873	free(chunk->buf.data);
384	1873	FREE_OBJ(chunk);
385	1873	}
386
387		/* Append a set of records to a chunk */
388		static void
389	25057	chunk_appendbuf(struct chunk chunk, const uint32_t ptr, size_t len)
390		{
391
392	25057	CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
393	25057	assert(chunk->type == chunk_t_buf);
394	25057	if (chunk->buf.space < chunk->len + len) {
395	7938	while (chunk->buf.space < chunk->len + len)
396	3969	chunk->buf.space *= 2;
397	7938	chunk->buf.data = realloc(chunk->buf.data,
398	3969	sizeof (uint32_t) * chunk->buf.space);
399	3969	}
400	25057	vmemcpy(chunk->buf.data + chunk->len, ptr, sizeof(uint32_t) * len);
401	25057	chunk->len += len;
402	25057	}
403
404		/* Transform a shm chunk to a buf chunk */
405		static void
406	0	chunk_shm_to_buf(struct VSLQ vslq, struct chunk chunk)
407		{
408		struct vtx *vtx;
409		struct chunk *buf;
410
411	0	CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
412	0	assert(chunk->type == chunk_t_shm);
413	0	vtx = chunk->vtx;
414	0	CHECK_OBJ_NOTNULL(vtx, VTX_MAGIC);
415
416	0	buf = VTAILQ_PREV(chunk, chunkhead, list);
417	0	if (buf != NULL && buf->type == chunk_t_buf)
418		/* Previous is a buf chunk, append to it */
419	0	chunk_appendbuf(buf, chunk->shm.start.ptr, chunk->len);
420		else {
421		/* Create a new buf chunk and insert it before this */
422	0	buf = chunk_newbuf(vtx, chunk->shm.start.ptr, chunk->len);
423	0	AN(buf);
424	0	VTAILQ_INSERT_BEFORE(chunk, buf, list);
425		}
426
427		/* Reset cursor chunk pointer, vslc_vtx_next will set it correctly */
428	0	vtx->c.chunk = NULL;
429
430		/* Remove from the shmref list and vtx, and put chunk back
431		on the free list */
432	0	VTAILQ_REMOVE(&vslq->shmrefs, chunk, shm.shmref);
433	0	VTAILQ_REMOVE(&vtx->chunks, chunk, list);
434	0	VTAILQ_INSERT_HEAD(&vtx->shmchunks_free, chunk, list);
435	0	}
436
437		/* Append a set of records to a vtx structure */
438		static enum vsl_status
439	36167	vtx_append(struct VSLQ vslq, struct vtx vtx, const struct VSLC_ptr *start,
440		size_t len)
441		{
442		struct chunk *chunk;
443		enum vsl_check i;
444
445	36167	AN(vtx);
446	36167	AN(len);
447	36167	AN(start);
448
449	36167	i = VSL_Check(vslq->c, start);
450	36167	if (i == vsl_check_e_inval)
451	0	return (vsl_e_overrun);
452
453	36167	if (i == vsl_check_valid && !VTAILQ_EMPTY(&vtx->shmchunks_free)) {
454		/* Shmref it */
455	9238	chunk = VTAILQ_FIRST(&vtx->shmchunks_free);
456	9238	CHECK_OBJ_NOTNULL(chunk, CHUNK_MAGIC);
457	9238	assert(chunk->type == chunk_t_shm);
458	9238	assert(chunk->vtx == vtx);
459	9238	VTAILQ_REMOVE(&vtx->shmchunks_free, chunk, list);
460	9238	chunk->shm.start = *start;
461	9238	chunk->len = len;
462	9238	VTAILQ_INSERT_TAIL(&vtx->chunks, chunk, list);
463
464		/* Append to shmref list */
465	9238	VTAILQ_INSERT_TAIL(&vslq->shmrefs, chunk, shm.shmref);
466	9238	} else {
467		/* Buffer it */
468	26929	chunk = VTAILQ_LAST(&vtx->chunks, chunkhead);
469	26929	CHECK_OBJ_ORNULL(chunk, CHUNK_MAGIC);
470	26929	if (chunk != NULL && chunk->type == chunk_t_buf) {
471		/* Tail is a buf chunk, append to that */
472	25055	chunk_appendbuf(chunk, start->ptr, len);
473	25055	} else {
474		/* Append new buf chunk */
475	1874	chunk = chunk_newbuf(vtx, start->ptr, len);
476	1874	AN(chunk);
477	1874	VTAILQ_INSERT_TAIL(&vtx->chunks, chunk, list);
478		}
479		}
480	36167	vtx->len += len;
481	36167	return (vsl_more);
482	36167	}
483
484		/* Allocate a new vtx structure */
485		static struct vtx *
486	5320	vtx_new(struct VSLQ *vslq)
487		{
488		struct vtx *vtx;
489		int i;
490
491	5320	AN(vslq);
492	5320	if (vslq->n_cache) {
493	2411	AZ(VTAILQ_EMPTY(&vslq->cache));
494	2411	vtx = VTAILQ_FIRST(&vslq->cache);
495	2411	VTAILQ_REMOVE(&vslq->cache, vtx, list_child);
496	2411	vslq->n_cache--;
497	2411	} else {
498	2909	ALLOC_OBJ(vtx, VTX_MAGIC);
499	2909	AN(vtx);