GCOV - Varnish Cache


/*-
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006-2015 Varnish Software AS
 * All rights reserved.
 *
 * Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
 *
 * 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 "cache_int.h"
#include "cache_filter.h"
#include "cache_objhead.h"

void
VDP_Panic(struct vsb *vsb, const struct vdp_ctx *vdc)
{
        struct vdp_entry *vde;

        if (PAN_dump_struct(vsb, vdc, VDP_CTX_MAGIC, "vdc"))
                return;
        VSB_printf(vsb, "nxt = %p,\n", vdc->nxt);
        VSB_printf(vsb, "retval = %d,\n", vdc->retval);

        if (!VTAILQ_EMPTY(&vdc->vdp)) {
                VSB_cat(vsb, "filters = {\n");
                VSB_indent(vsb, 2);
                VTAILQ_FOREACH(vde, &vdc->vdp, list)
                        VSB_printf(vsb, "%s = %p { priv = %p }\n",
                            vde->vdp->name, vde, vde->priv);
                VSB_indent(vsb, -2);
                VSB_cat(vsb, "},\n");
        }

        VSB_indent(vsb, -2);
        VSB_cat(vsb, "},\n");
}

/*
 * Ensure that transports have called VDP_Close()
 * to avoid leaks in VDPs
 */
void
VDP_Fini(const struct vdp_ctx *vdc)
{
        assert(VTAILQ_EMPTY(&vdc->vdp));
}

void
VDP_Init(struct vdp_ctx *vdc, struct worker *wrk, struct vsl_log *vsl,
    const struct req *req, const struct busyobj *bo, intmax_t *clen)
{
        AN(vdc);
        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        AN(vsl);

        AN(clen);

        assert((req ? 1 : 0) ^ (bo ? 1 : 0));

        AN(clen);
        assert(*clen >= -1);

        INIT_OBJ(vdc, VDP_CTX_MAGIC);
        VTAILQ_INIT(&vdc->vdp);
        vdc->wrk = wrk;
        vdc->vsl = vsl;
        vdc->clen = clen;

        if (req != NULL) {
                CHECK_OBJ(req, REQ_MAGIC);
                vdc->oc = req->objcore;
                vdc->hp = req->resp;
        }
        else {
                CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
                vdc->oc = bo->bereq_body;
                vdc->hp = bo->bereq;
        }
}

/* VDP_bytes
 *
 * Pushes len bytes at ptr down the delivery processor list.
 *
 * This function picks and calls the next delivery processor from the
 * list. The return value is the return value of the delivery
 * processor. Upon seeing a non-zero return value, that lowest value
 * observed is latched in ->retval and all subsequent calls to
 * VDP_bytes will return that value directly without calling the next
 * processor.
 *
 * VDP_END marks the end of successful processing, it is issued by
 * VDP_DeliverObj() and may also be sent downstream by processors ending the
 * stream (for return value != 0)
 *
 * VDP_END must at most be received once per processor, so any VDP sending it
 * downstream must itself not forward it a second time.
 *
 * Valid return values (of VDP_bytes and any VDP function):
 * r < 0:  Error, breaks out early on an error condition
 * r == 0: Continue
 * r > 0:  Stop, breaks out early without error condition
 */

int
VDP_bytes(struct vdp_ctx *vdc, enum vdp_action act,
    const void *ptr, ssize_t len)
{
        int retval;
        struct vdp_entry *vdpe;

        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        if (vdc->retval)
                return (vdc->retval);
        vdpe = vdc->nxt;
        CHECK_OBJ_NOTNULL(vdpe, VDP_ENTRY_MAGIC);

        /* at most one VDP_END call */
        assert(vdpe->end == VDP_NULL);

        if (act == VDP_NULL)
                assert(len > 0);
        else if (act == VDP_END)
                vdpe->end = VDP_END;
        else
                assert(act == VDP_FLUSH);

        /* Call the present layer, while pointing to the next layer down */
        vdc->nxt = VTAILQ_NEXT(vdpe, list);
        vdpe->calls++;
        vdc->bytes_done = len;
        retval = vdpe->vdp->bytes(vdc, act, &vdpe->priv, ptr, len);
        vdpe->bytes_in += vdc->bytes_done;
        if (retval && (vdc->retval == 0 || retval < vdc->retval))
                vdc->retval = retval; /* Latch error value */
        vdc->nxt = vdpe;
        return (vdc->retval);
}

int
VDP_Push(VRT_CTX, struct vdp_ctx *vdc, struct ws *ws, const struct vdp *vdp,
    void *priv)
{
        struct vdp_entry *vdpe;

        CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        CHECK_OBJ_ORNULL(vdc->oc, OBJCORE_MAGIC);
        CHECK_OBJ_NOTNULL(vdc->hp, HTTP_MAGIC);
        AN(vdc->clen);
        assert(*vdc->clen >= -1);
        AN(ws);
        AN(vdp);
        AN(vdp->name);

        if (vdc->retval)
                return (vdc->retval);

        if (DO_DEBUG(DBG_PROCESSORS))
                VSLb(vdc->vsl, SLT_Debug, "VDP_push(%s)", vdp->name);

        vdpe = WS_Alloc(ws, sizeof *vdpe);
        if (vdpe == NULL) {
                AZ(vdc->retval);
                vdc->retval = -1;
                return (vdc->retval);
        }
        INIT_OBJ(vdpe, VDP_ENTRY_MAGIC);
        vdpe->vdp = vdp;
        vdpe->priv = priv;
        VTAILQ_INSERT_TAIL(&vdc->vdp, vdpe, list);
        vdc->nxt = VTAILQ_FIRST(&vdc->vdp);

        AZ(vdc->retval);
        if (vdpe->vdp->init != NULL)
                vdc->retval = vdpe->vdp->init(ctx, vdc, &vdpe->priv);
        vdc->oc = NULL;

        if (vdc->retval) {
                VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
                vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
                vdc->retval = vdc->retval > 0 ? 0 : vdc->retval;
        } else
                AN(vdp->bytes);
        return (vdc->retval);
}

uint64_t
VDP_Close(struct vdp_ctx *vdc, struct objcore *oc, struct boc *boc)
{
        struct vdp_entry *vdpe;
        uint64_t rv = 0;

        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
        CHECK_OBJ_ORNULL(oc, OBJCORE_MAGIC);
        CHECK_OBJ_ORNULL(boc, BOC_MAGIC);

        while ((vdpe = VTAILQ_FIRST(&vdc->vdp)) != NULL) {
                CHECK_OBJ(vdpe, VDP_ENTRY_MAGIC);
                rv = vdpe->bytes_in;
                VSLb(vdc->vsl, SLT_VdpAcct, "%s %ju %ju", vdpe->vdp->name,
                    (uintmax_t)vdpe->calls, (uintmax_t)rv);
                if (vdpe->vdp->fini != NULL)
                        AZ(vdpe->vdp->fini(vdc, &vdpe->priv));
                AZ(vdpe->priv);
                VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
                vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
#ifdef VDP_PEDANTIC_ARMED
                // enable when we are confident to get VDP_END right
                if (vdc->nxt == NULL && vdc->retval >= 0)
                        assert(vdpe->end == VDP_END);
#endif
        }
        if (oc != NULL)
                HSH_Cancel(vdc->wrk, oc, boc);
        return (rv);
}

/*--------------------------------------------------------------------*/

/*
 * Push a VDPIO vdp. This can only be used with only vdpio-enabled VDPs or
 * after a successful upgrade
 */
int
VDPIO_Push(VRT_CTX, struct vdp_ctx *vdc, struct ws *ws, const struct vdp *vdp,
    void *priv)
{
        struct vdp_entry *vdpe;
        int r;

        CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        CHECK_OBJ_ORNULL(vdc->oc, OBJCORE_MAGIC);
        CHECK_OBJ_NOTNULL(vdc->hp, HTTP_MAGIC);
        AN(vdc->clen);
        assert(*vdc->clen >= -1);
        AN(ws);
        AN(vdp);
        AN(vdp->name);

        if (vdc->retval < 0)
                return (vdc->retval);

        AN(vdp->io_init);

        // the first VDP (which leases from storage) only gets the minimum
        // capacity requirement of 1
        if (vdc->retval == 0) {
                assert(VTAILQ_EMPTY(&vdc->vdp));
                vdc->retval = 1;
        }

        if (DO_DEBUG(DBG_PROCESSORS))
                VSLb(vdc->vsl, SLT_Debug, "VDPIO_push(%s)", vdp->name);

        vdpe = WS_Alloc(ws, sizeof *vdpe);
        if (vdpe == NULL) {
                vdc->retval = -ENOMEM;
                return (vdc->retval);
        }
        INIT_OBJ(vdpe, VDP_ENTRY_MAGIC);
        vdpe->vdp = vdp;
        vdpe->priv = priv;
        VTAILQ_INSERT_TAIL(&vdc->vdp, vdpe, list);
        vdc->nxt = VTAILQ_FIRST(&vdc->vdp);

        assert(vdc->retval > 0);
        if (vdpe->vdp->io_init != NULL) {
                r = vdpe->vdp->io_init(ctx, vdc, &vdpe->priv, vdc->retval);
                if (r <= 0) {
                        VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
                        vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
                }
                else
                        AN(vdp->io_lease);
                if (r != 0)
                        vdc->retval = r;
        }
        vdc->oc = NULL;
        return (vdc->retval);
}

/*
 * upgrade an already initialized VDP filter chain to VDPIO, if possible
 * returns:
 * > 0 cap
 * -ENOTSUP io_upgrade missing for at least one filter
 * vdc->retval if < 0
 */
int
VDPIO_Upgrade(VRT_CTX, struct vdp_ctx *vdc)
{
        struct vdp_entry *vdpe;
        int cap, r;

        CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);

        VTAILQ_FOREACH(vdpe, &vdc->vdp, list)
                if (vdpe->vdp->io_upgrade == NULL)
                        return (-ENOTSUP);

        if (vdc->retval < 0)
                return (vdc->retval);

        // minimum capacity requirement for the first filter (after storage)
        r = cap = 1;
        VTAILQ_FOREACH(vdpe, &vdc->vdp, list) {
                r = vdpe->vdp->io_upgrade(ctx, vdc, &vdpe->priv, cap);
                if (DO_DEBUG(DBG_PROCESSORS)) {
                        VSLb(vdc->vsl, SLT_Debug, "VDPIO_Upgrade "
                            "%d = %s(cap = %d)",
                            r, vdpe->vdp->name, cap);
                }
                if (r < 0)
                        return ((vdc->retval = r));
                // XXX remove if filter does not want to be pushed?
                assert(r != 0);
                cap = r;
        }
        return ((vdc->retval = r));
}

uint64_t
VDPIO_Close1(struct vdp_ctx *vdc, struct vdp_entry *vdpe)
{
        uint64_t rv;

        CHECK_OBJ_NOTNULL(vdpe, VDP_ENTRY_MAGIC);
        rv = vdpe->bytes_in;
        VSLb(vdc->vsl, SLT_VdpAcct, "%s %ju %ju", vdpe->vdp->name,
            (uintmax_t)vdpe->calls, (uintmax_t)rv);
        if (vdpe->vdp->io_fini != NULL)
                vdpe->vdp->io_fini(vdc, &vdpe->priv);
        AZ(vdpe->priv);
        VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
        vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
        return (rv);
}

uint64_t
VDPIO_Close(struct vdp_ctx *vdc, struct objcore *oc, struct boc *boc)
{
        struct vdp_entry *vdpe;
        uint64_t rv = 0;

        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
        CHECK_OBJ_ORNULL(oc, OBJCORE_MAGIC);
        CHECK_OBJ_ORNULL(boc, BOC_MAGIC);

        while ((vdpe = VTAILQ_FIRST(&vdc->vdp)) != NULL)
                rv = VDPIO_Close1(vdc, vdpe);

        if (oc != NULL)
                HSH_Cancel(vdc->wrk, oc, boc);
        return (rv);
}

/*
 * ============================================================
 * VDPIO helpers: VAI management
 *
 * Transports should not need to talk to the VAI Object interface directly,
 * because its state is kept in the vdp_ctx
 *
 * So we wrap init, return and fini
 */

// return true if error
int
VDPIO_Init(struct vdp_ctx *vdc, struct objcore *oc, struct ws *ws,
    vai_notify_cb *notify_cb, void *notify_priv, struct vscaret *scaret)
{
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        VSCARET_CHECK_NOTNULL(scaret);
        AN(scaret->capacity);
        AZ(scaret->used);

        AZ(vdc->vai_hdl);
        vdc->vai_hdl = ObjVAIinit(vdc->wrk, oc, ws, notify_cb, notify_priv);
        if (vdc->vai_hdl == NULL)
                return (1);
        vdc->scaret = scaret;
        return (0);
}

// return leases stashed in scaret
void
VDPIO_Return(const struct vdp_ctx *vdc)
{
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);

        ObjVAIreturn(vdc->wrk, vdc->vai_hdl, vdc->scaret);
}

// Notify after a VDP's io_lease has return -EAGAIN
// usually not called from the io thread, so it has a separate wrk
void
VDPIO_Notify(struct worker *wrk, const struct vdp_ctx *vdc)
{
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);

        ObjVAInotify(wrk, vdc->vai_hdl);
}

void
VDPIO_Fini(struct vdp_ctx *vdc)
{
        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);

        VDPIO_Return(vdc);
        ObjVAIfini(vdc->wrk, &vdc->vai_hdl);
}

/*--------------------------------------------------------------------*/
int v_matchproto_(objiterate_f)
VDP_ObjIterate(void *priv, unsigned flush, const void *ptr, ssize_t len)
{
        enum vdp_action act;

        if (flush == 0)
                act = VDP_NULL;
        else if ((flush & OBJ_ITER_END) != 0)
                act = VDP_END;
        else
                act = VDP_FLUSH;

        return (VDP_bytes(priv, act, ptr, len));
}


int
VDP_DeliverObj(struct vdp_ctx *vdc, struct objcore *oc)
{
        int r, final;

        CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
        CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
        AN(vdc->vsl);
        AZ(vdc->oc);
        vdc->hp = NULL;
        vdc->clen = NULL;
        final = oc->flags & OC_F_TRANSIENT ? 1 : 0;
        r = ObjIterate(vdc->wrk, oc, vdc, VDP_ObjIterate, final);
        if (r < 0)
                return (r);
        return (0);
}

		vinyl-cache/bin/vinyld/cache/cache_deliver_proc.c
0		/*-
1		* Copyright (c) 2006 Verdens Gang AS
2		* Copyright (c) 2006-2015 Varnish Software AS
3		* All rights reserved.
4		*
5		* Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
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		#include "config.h"
32
33		#include "cache_int.h"
34		#include "cache_filter.h"
35		#include "cache_objhead.h"
36
37		void
38	126	VDP_Panic(struct vsb vsb, const struct vdp_ctx vdc)
39		{
40		struct vdp_entry *vde;
41
42	126	if (PAN_dump_struct(vsb, vdc, VDP_CTX_MAGIC, "vdc"))
43	84	return;
44	42	VSB_printf(vsb, "nxt = %p,\n", vdc->nxt);
45	42	VSB_printf(vsb, "retval = %d,\n", vdc->retval);
46
47	42	if (!VTAILQ_EMPTY(&vdc->vdp)) {
48	21	VSB_cat(vsb, "filters = {\n");
49	21	VSB_indent(vsb, 2);
50	105	VTAILQ_FOREACH(vde, &vdc->vdp, list)
51	168	VSB_printf(vsb, "%s = %p { priv = %p }\n",
52	84	vde->vdp->name, vde, vde->priv);
53	21	VSB_indent(vsb, -2);
54	21	VSB_cat(vsb, "},\n");
55	21	}
56
57	42	VSB_indent(vsb, -2);
58	42	VSB_cat(vsb, "},\n");
59	126	}
60
61		/*
62		* Ensure that transports have called VDP_Close()
63		* to avoid leaks in VDPs
64		*/
65		void
66	94590	VDP_Fini(const struct vdp_ctx *vdc)
67		{
68	94590	assert(VTAILQ_EMPTY(&vdc->vdp));
69	94590	}
70
71		void
72	131990	VDP_Init(struct vdp_ctx vdc, struct worker wrk, struct vsl_log *vsl,
73		const struct req req, const struct busyobj bo, intmax_t *clen)
74		{
75	131990	AN(vdc);
76	131990	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
77	131990	AN(vsl);
78
79	131990	AN(clen);
80
81	131990	assert((req ? 1 : 0) ^ (bo ? 1 : 0));
82
83	131990	AN(clen);
84	131990	assert(*clen >= -1);
85
86	131990	INIT_OBJ(vdc, VDP_CTX_MAGIC);
87	131990	VTAILQ_INIT(&vdc->vdp);
88	131990	vdc->wrk = wrk;
89	131990	vdc->vsl = vsl;
90	131990	vdc->clen = clen;
91
92	131990	if (req != NULL) {
93	83249	CHECK_OBJ(req, REQ_MAGIC);
94	83249	vdc->oc = req->objcore;
95	83249	vdc->hp = req->resp;
96	83249	}
97		else {
98	48741	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
99	48741	vdc->oc = bo->bereq_body;
100	48741	vdc->hp = bo->bereq;
101		}
102	131990	}
103
104		/* VDP_bytes
105		*
106		* Pushes len bytes at ptr down the delivery processor list.
107		*
108		* This function picks and calls the next delivery processor from the
109		* list. The return value is the return value of the delivery
110		* processor. Upon seeing a non-zero return value, that lowest value
111		* observed is latched in ->retval and all subsequent calls to
112		* VDP_bytes will return that value directly without calling the next
113		* processor.
114		*
115		* VDP_END marks the end of successful processing, it is issued by
116		* VDP_DeliverObj() and may also be sent downstream by processors ending the
117		* stream (for return value != 0)
118		*
119		* VDP_END must at most be received once per processor, so any VDP sending it
120		* downstream must itself not forward it a second time.
121		*
122		* Valid return values (of VDP_bytes and any VDP function):
123		* r < 0: Error, breaks out early on an error condition
124		* r == 0: Continue
125		* r > 0: Stop, breaks out early without error condition
126		*/
127
128		int
129	577526	VDP_bytes(struct vdp_ctx *vdc, enum vdp_action act,
130		const void *ptr, ssize_t len)
131		{
132		int retval;
133		struct vdp_entry *vdpe;
134
135	577526	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
136	577526	if (vdc->retval)
137	601	return (vdc->retval);
138	576925	vdpe = vdc->nxt;
139	576925	CHECK_OBJ_NOTNULL(vdpe, VDP_ENTRY_MAGIC);
140
141		/* at most one VDP_END call */
142	576925	assert(vdpe->end == VDP_NULL);
143
144	576925	if (act == VDP_NULL)
145	124769	assert(len > 0);
146	452156	else if (act == VDP_END)
147	63772	vdpe->end = VDP_END;
148		else
149	388384	assert(act == VDP_FLUSH);
150
151		/* Call the present layer, while pointing to the next layer down */
152	576925	vdc->nxt = VTAILQ_NEXT(vdpe, list);
153	576925	vdpe->calls++;
154	576925	vdc->bytes_done = len;
155	576925	retval = vdpe->vdp->bytes(vdc, act, &vdpe->priv, ptr, len);
156	576925	vdpe->bytes_in += vdc->bytes_done;
157	576925	if (retval && (vdc->retval == 0 \|\| retval < vdc->retval))
158	1322	vdc->retval = retval; /* Latch error value */
159	576925	vdc->nxt = vdpe;
160	576925	return (vdc->retval);
161	577526	}
162
163		int
164	115988	VDP_Push(VRT_CTX, struct vdp_ctx vdc, struct ws ws, const struct vdp *vdp,
165		void *priv)
166		{
167		struct vdp_entry *vdpe;
168
169	115988	CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
170	115988	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
171	115988	CHECK_OBJ_ORNULL(vdc->oc, OBJCORE_MAGIC);
172	115988	CHECK_OBJ_NOTNULL(vdc->hp, HTTP_MAGIC);
173	115988	AN(vdc->clen);
174	115988	assert(*vdc->clen >= -1);
175	115988	AN(ws);
176	115988	AN(vdp);
177	115988	AN(vdp->name);
178
179	115988	if (vdc->retval)
180	0	return (vdc->retval);
181
182	115988	if (DO_DEBUG(DBG_PROCESSORS))
183	1029	VSLb(vdc->vsl, SLT_Debug, "VDP_push(%s)", vdp->name);
184
185	115988	vdpe = WS_Alloc(ws, sizeof *vdpe);
186	115988	if (vdpe == NULL) {
187	2688	AZ(vdc->retval);
188	2688	vdc->retval = -1;
189	2688	return (vdc->retval);
190		}
191	113300	INIT_OBJ(vdpe, VDP_ENTRY_MAGIC);
192	113300	vdpe->vdp = vdp;
193	113300	vdpe->priv = priv;
194	113300	VTAILQ_INSERT_TAIL(&vdc->vdp, vdpe, list);
195	113300	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
196
197	113300	AZ(vdc->retval);
198	113300	if (vdpe->vdp->init != NULL)
199	106726	vdc->retval = vdpe->vdp->init(ctx, vdc, &vdpe->priv);
200	113300	vdc->oc = NULL;
201
202	113300	if (vdc->retval) {
203	1344	VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
204	1344	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
205	1344	vdc->retval = vdc->retval > 0 ? 0 : vdc->retval;
206	1344	} else
207	111956	AN(vdp->bytes);
208	113300	return (vdc->retval);
209	115988	}
210
211		uint64_t
212	132010	VDP_Close(struct vdp_ctx vdc, struct objcore oc, struct boc *boc)
213		{
214		struct vdp_entry *vdpe;
215	132010	uint64_t rv = 0;
216
217	132010	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
218	132010	CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
219	132010	CHECK_OBJ_ORNULL(oc, OBJCORE_MAGIC);
220	132010	CHECK_OBJ_ORNULL(boc, BOC_MAGIC);
221
222	243559	while ((vdpe = VTAILQ_FIRST(&vdc->vdp)) != NULL) {
223	111549	CHECK_OBJ(vdpe, VDP_ENTRY_MAGIC);
224	111549	rv = vdpe->bytes_in;
225	223098	VSLb(vdc->vsl, SLT_VdpAcct, "%s %ju %ju", vdpe->vdp->name,
226	111549	(uintmax_t)vdpe->calls, (uintmax_t)rv);
227	111549	if (vdpe->vdp->fini != NULL)
228	20311	AZ(vdpe->vdp->fini(vdc, &vdpe->priv));
229	111549	AZ(vdpe->priv);
230	111549	VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
231	111549	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
232		#ifdef VDP_PEDANTIC_ARMED
233		// enable when we are confident to get VDP_END right
234		if (vdc->nxt == NULL && vdc->retval >= 0)
235		assert(vdpe->end == VDP_END);
236		#endif
237		}
238	132010	if (oc != NULL)
239	83246	HSH_Cancel(vdc->wrk, oc, boc);
240	132010	return (rv);
241		}
242
243		/--------------------------------------------------------------------/
244
245		/*
246		* Push a VDPIO vdp. This can only be used with only vdpio-enabled VDPs or
247		* after a successful upgrade
248		*/
249		int
250	2689	VDPIO_Push(VRT_CTX, struct vdp_ctx vdc, struct ws ws, const struct vdp *vdp,
251		void *priv)
252		{
253		struct vdp_entry *vdpe;
254		int r;
255
256	2689	CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
257	2689	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
258	2689	CHECK_OBJ_ORNULL(vdc->oc, OBJCORE_MAGIC);
259	2689	CHECK_OBJ_NOTNULL(vdc->hp, HTTP_MAGIC);
260	2689	AN(vdc->clen);
261	2689	assert(*vdc->clen >= -1);
262	2689	AN(ws);
263	2689	AN(vdp);
264	2689	AN(vdp->name);
265
266	2689	if (vdc->retval < 0)
267	0	return (vdc->retval);
268
269	2689	AN(vdp->io_init);
270
271		// the first VDP (which leases from storage) only gets the minimum
272		// capacity requirement of 1
273	2689	if (vdc->retval == 0) {
274	0	assert(VTAILQ_EMPTY(&vdc->vdp));
275	0	vdc->retval = 1;
276	0	}
277
278	2689	if (DO_DEBUG(DBG_PROCESSORS))
279	2687	VSLb(vdc->vsl, SLT_Debug, "VDPIO_push(%s)", vdp->name);
280
281	2689	vdpe = WS_Alloc(ws, sizeof *vdpe);
282	2689	if (vdpe == NULL) {
283	0	vdc->retval = -ENOMEM;
284	0	return (vdc->retval);
285		}
286	2687	INIT_OBJ(vdpe, VDP_ENTRY_MAGIC);
287	2687	vdpe->vdp = vdp;
288	2687	vdpe->priv = priv;
289	2687	VTAILQ_INSERT_TAIL(&vdc->vdp, vdpe, list);
290	2687	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
291
292	2687	assert(vdc->retval > 0);
293	2687	if (vdpe->vdp->io_init != NULL) {
294	2687	r = vdpe->vdp->io_init(ctx, vdc, &vdpe->priv, vdc->retval);
295	2687	if (r <= 0) {
296	0	VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
297	0	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
298	0	}
299		else
300	2687	AN(vdp->io_lease);
301	2687	if (r != 0)
302	2687	vdc->retval = r;
303	2687	}
304	2687	vdc->oc = NULL;
305	2687	return (vdc->retval);
306	2687	}
307
308		/*
309		* upgrade an already initialized VDP filter chain to VDPIO, if possible
310		* returns:
311		* > 0 cap
312		* -ENOTSUP io_upgrade missing for at least one filter
313		* vdc->retval if < 0
314		*/
315		int
316	840	VDPIO_Upgrade(VRT_CTX, struct vdp_ctx *vdc)
317		{
318		struct vdp_entry *vdpe;
319		int cap, r;
320
321	840	CHECK_OBJ_NOTNULL(ctx, VRT_CTX_MAGIC);
322	840	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
323
324	1344	VTAILQ_FOREACH(vdpe, &vdc->vdp, list)
325	672	if (vdpe->vdp->io_upgrade == NULL)
326	168	return (-ENOTSUP);
327
328	672	if (vdc->retval < 0)
329	0	return (vdc->retval);
330
331		// minimum capacity requirement for the first filter (after storage)
332	672	r = cap = 1;
333	1008	VTAILQ_FOREACH(vdpe, &vdc->vdp, list) {
334	336	r = vdpe->vdp->io_upgrade(ctx, vdc, &vdpe->priv, cap);
335	336	if (DO_DEBUG(DBG_PROCESSORS)) {
336	672	VSLb(vdc->vsl, SLT_Debug, "VDPIO_Upgrade "
337		"%d = %s(cap = %d)",
338	336	r, vdpe->vdp->name, cap);
339	336	}
340	336	if (r < 0)
341	0	return ((vdc->retval = r));
342		// XXX remove if filter does not want to be pushed?
343	336	assert(r != 0);
344	336	cap = r;
345	336	}
346	672	return ((vdc->retval = r));
347	840	}
348
349		uint64_t
350	3023	VDPIO_Close1(struct vdp_ctx vdc, struct vdp_entry vdpe)
351		{
352		uint64_t rv;
353
354	3023	CHECK_OBJ_NOTNULL(vdpe, VDP_ENTRY_MAGIC);
355	3023	rv = vdpe->bytes_in;
356	6046	VSLb(vdc->vsl, SLT_VdpAcct, "%s %ju %ju", vdpe->vdp->name,
357	3023	(uintmax_t)vdpe->calls, (uintmax_t)rv);
358	3023	if (vdpe->vdp->io_fini != NULL)
359	1680	vdpe->vdp->io_fini(vdc, &vdpe->priv);
360	3023	AZ(vdpe->priv);
361	3023	VTAILQ_REMOVE(&vdc->vdp, vdpe, list);
362	3023	vdc->nxt = VTAILQ_FIRST(&vdc->vdp);
363	3023	return (rv);
364		}
365
366		uint64_t
367	672	VDPIO_Close(struct vdp_ctx vdc, struct objcore oc, struct boc *boc)
368		{
369		struct vdp_entry *vdpe;
370	672	uint64_t rv = 0;
371
372	672	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
373	672	CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
374	672	CHECK_OBJ_ORNULL(oc, OBJCORE_MAGIC);
375	672	CHECK_OBJ_ORNULL(boc, BOC_MAGIC);
376
377	3023	while ((vdpe = VTAILQ_FIRST(&vdc->vdp)) != NULL)
378	2351	rv = VDPIO_Close1(vdc, vdpe);
379
380	672	if (oc != NULL)
381	672	HSH_Cancel(vdc->wrk, oc, boc);
382	672	return (rv);
383		}
384
385		/*
386		* ============================================================
387		* VDPIO helpers: VAI management
388		*
389		* Transports should not need to talk to the VAI Object interface directly,
390		* because its state is kept in the vdp_ctx
391		*
392		* So we wrap init, return and fini
393		*/
394
395		// return true if error
396		int
397	672	VDPIO_Init(struct vdp_ctx vdc, struct objcore oc, struct ws *ws,
398		vai_notify_cb notify_cb, void notify_priv, struct vscaret *scaret)
399		{
400	672	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
401	672	VSCARET_CHECK_NOTNULL(scaret);
402	672	AN(scaret->capacity);
403	672	AZ(scaret->used);
404
405	672	AZ(vdc->vai_hdl);
406	672	vdc->vai_hdl = ObjVAIinit(vdc->wrk, oc, ws, notify_cb, notify_priv);
407	672	if (vdc->vai_hdl == NULL)
408	0	return (1);
409	672	vdc->scaret = scaret;
410	672	return (0);
411	672	}
412
413		// return leases stashed in scaret
414		void
415	3359	VDPIO_Return(const struct vdp_ctx *vdc)
416		{
417	3359	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
418
419	3359	ObjVAIreturn(vdc->wrk, vdc->vai_hdl, vdc->scaret);
420	3359	}
421
422		// Notify after a VDP's io_lease has return -EAGAIN
423		// usually not called from the io thread, so it has a separate wrk
424		void
425	2687	VDPIO_Notify(struct worker wrk, const struct vdp_ctx vdc)
426		{
427	2687	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
428
429	2687	ObjVAInotify(wrk, vdc->vai_hdl);
430	2687	}
431
432		void
433	672	VDPIO_Fini(struct vdp_ctx *vdc)
434		{
435	672	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
436
437	672	VDPIO_Return(vdc);
438	672	ObjVAIfini(vdc->wrk, &vdc->vai_hdl);
439	672	}
440
441		/--------------------------------------------------------------------/
442		int v_matchproto_(objiterate_f)
443	217039	VDP_ObjIterate(void priv, unsigned flush, const void ptr, ssize_t len)
444		{
445		enum vdp_action act;
446
447	217039	if (flush == 0)
448	118894	act = VDP_NULL;
449	98145	else if ((flush & OBJ_ITER_END) != 0)
450	55025	act = VDP_END;
451		else
452	43120	act = VDP_FLUSH;
453
454	217039	return (VDP_bytes(priv, act, ptr, len));
455		}
456
457
458		int
459	50881	VDP_DeliverObj(struct vdp_ctx vdc, struct objcore oc)
460		{
461		int r, final;
462
463	50881	CHECK_OBJ_NOTNULL(vdc, VDP_CTX_MAGIC);
464	50881	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
465	50881	CHECK_OBJ_NOTNULL(vdc->wrk, WORKER_MAGIC);
466	50881	AN(vdc->vsl);
467	50881	AZ(vdc->oc);
468	50881	vdc->hp = NULL;
469	50881	vdc->clen = NULL;
470	50881	final = oc->flags & OC_F_TRANSIENT ? 1 : 0;
471	50881	r = ObjIterate(vdc->wrk, oc, vdc, VDP_ObjIterate, final);
472	50881	if (r < 0)
473	921	return (r);
474	49960	return (0);
475	50881	}