GCOV - Varnish Cache


/*-
 * Copyright (c) 2007-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 <stdlib.h>

#include "cache/cache_varnishd.h"

#include "cache/cache_obj.h"
#include "cache/cache_objhead.h"

#include "storage/storage.h"
#include "storage/storage_simple.h"

#include "vtim.h"

/* Flags for allocating memory in sml_stv_alloc */
#define LESS_MEM_ALLOCED_IS_OK  1

// marker pointer for sml_trimstore
static void *trim_once = &trim_once;

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

static struct storage *
objallocwithnuke(struct worker *, const struct stevedore *, ssize_t size,
    int flags);

static struct storage *
sml_stv_alloc(const struct stevedore *stv, ssize_t size, int flags)
{
        struct storage *st;

        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        AN(stv->sml_alloc);

        if (!(flags & LESS_MEM_ALLOCED_IS_OK)) {
                if (size > cache_param->fetch_maxchunksize)
                        return (NULL);
                else
                        return (stv->sml_alloc(stv, size));
        }

        if (size > cache_param->fetch_maxchunksize)
                size = cache_param->fetch_maxchunksize;

        assert(size <= UINT_MAX);       /* field limit in struct storage */

        for (;;) {
                /* try to allocate from it */
                assert(size > 0);
                st = stv->sml_alloc(stv, size);
                if (st != NULL)
                        break;

                if (size <= cache_param->fetch_chunksize)
                        break;

                size /= 2;
        }
        CHECK_OBJ_ORNULL(st, STORAGE_MAGIC);
        return (st);
}

static void
sml_stv_free(const struct stevedore *stv, struct storage *st)
{

        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
        if (stv->sml_free != NULL)
                stv->sml_free(st);
}

/*--------------------------------------------------------------------
 * This function is called by stevedores ->allocobj() method, which
 * very often will be SML_allocobj() below, to convert a slab
 * of storage into object which the stevedore can then register in its
 * internal state, before returning it to STV_NewObject().
 * As you probably guessed: All this for persistence.
 */

struct object *
SML_MkObject(const struct stevedore *stv, struct objcore *oc, void *ptr)
{
        struct object *o;

        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        AN(stv->methods);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);

        assert(PAOK(ptr));

        o = ptr;
        INIT_OBJ(o, OBJECT_MAGIC);

        VTAILQ_INIT(&o->list);

        oc->stobj->stevedore = stv;
        oc->stobj->priv = o;
        oc->stobj->priv2 = 0;
        return (o);
}

/*--------------------------------------------------------------------
 * This is the default ->allocobj() which all stevedores who do not
 * implement persistent storage can rely on.
 */

int v_matchproto_(storage_allocobj_f)
SML_allocobj(struct worker *wrk, const struct stevedore *stv,
    struct objcore *oc, unsigned wsl)
{
        struct object *o;
        struct storage *st = NULL;
        unsigned ltot;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);

        AN(stv->sml_alloc);

        ltot = sizeof(*o) + PRNDUP(wsl);

        do {
                st = stv->sml_alloc(stv, ltot);
                if (st != NULL && st->space < ltot) {
                        stv->sml_free(st);
                        st = NULL;
                }
        } while (st == NULL && LRU_NukeOne(wrk, stv->lru));
        if (st == NULL)
                return (0);

        CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
        o = SML_MkObject(stv, oc, st->ptr);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
        st->len = sizeof(*o);
        o->objstore = st;
        return (1);
}

void * v_matchproto_(storage_allocbuf_t)
SML_AllocBuf(struct worker *wrk, const struct stevedore *stv, size_t size,
    uintptr_t *ppriv)
{
        struct storage *st;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        AN(ppriv);

        if (size > UINT_MAX)
                return (NULL);
        st = objallocwithnuke(wrk, stv, size, 0);
        if (st == NULL)
                return (NULL);
        assert(st->space >= size);
        st->flags = STORAGE_F_BUFFER;
        st->len = size;
        *ppriv = (uintptr_t)st;
        return (st->ptr);
}

void v_matchproto_(storage_freebuf_t)
SML_FreeBuf(struct worker *wrk, const struct stevedore *stv, uintptr_t priv)
{
        struct storage *st;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);

        CAST_OBJ_NOTNULL(st, (void *)priv, STORAGE_MAGIC);
        assert(st->flags == STORAGE_F_BUFFER);
        sml_stv_free(stv, st);
}

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

static struct object *
sml_getobj(struct worker *wrk, struct objcore *oc)
{
        const struct stevedore *stv;
        struct object *o;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
        stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        if (stv->sml_getobj != NULL)
                return (stv->sml_getobj(wrk, oc));
        if (oc->stobj->priv == NULL)
                return (NULL);
        CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);
        return (o);
}

static void v_matchproto_(objslim_f)
sml_slim(struct worker *wrk, struct objcore *oc)
{
        const struct stevedore *stv;
        struct object *o;
        struct storage *st, *stn;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);

        stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);

#define OBJ_AUXATTR(U, l)                                       \
        do {                                                    \
                if (o->aa_##l != NULL) {                        \
                        sml_stv_free(stv, o->aa_##l);           \
                        o->aa_##l = NULL;                       \
                }                                               \
        } while (0);
#include "tbl/obj_attr.h"

        VTAILQ_FOREACH_SAFE(st, &o->list, list, stn) {
                CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
                VTAILQ_REMOVE(&o->list, st, list);
                sml_stv_free(stv, st);
        }
}

static void
sml_bocfini(const struct stevedore *stv, struct boc *boc)
{
        struct storage *st;

        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        CHECK_OBJ_NOTNULL(boc, BOC_MAGIC);

        if (boc->stevedore_priv == NULL ||
            boc->stevedore_priv == trim_once)
                return;

        /* Free any leftovers from Trim */
        TAKE_OBJ_NOTNULL(st, &boc->stevedore_priv, STORAGE_MAGIC);
        sml_stv_free(stv, st);
}

/*
 * called in two cases:
 * - oc->boc == NULL: cache object on LRU freed
 * - oc->boc != NULL: cache object replaced for backend error
 */
static void v_matchproto_(objfree_f)
sml_objfree(struct worker *wrk, struct objcore *oc)
{
        const struct stevedore *stv;
        struct storage *st;
        struct object *o;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
        stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
        CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);

        sml_slim(wrk, oc);
        st = o->objstore;
        CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
        FINI_OBJ(o);

        if (oc->boc != NULL)
                sml_bocfini(stv, oc->boc);
        else if (stv->lru != NULL)
                LRU_Remove(oc);

        sml_stv_free(stv, st);

        memset(oc->stobj, 0, sizeof oc->stobj);

        wrk->stats->n_object--;
}

// kept for reviewers - XXX remove later
#undef VAI_DBG

struct sml_hdl {
        struct vai_hdl_preamble preamble;
#define SML_HDL_MAGIC           0x37dfd996
        struct vai_qe           qe;
        struct pool_task        task;   // unfortunate
        struct ws               *ws;    // NULL is malloc()
        struct objcore          *oc;
        struct object           *obj;
        const struct stevedore  *stv;
        struct boc              *boc;

        struct storage          *st;    // updated by _lease()

        // only for _lease_boc()
        uint64_t                st_off; // already returned fragment of current st
        uint64_t                avail, returned;
        struct storage          *last;  // to resume, held back by _return()
};

static inline uint64_t
st2lease(const struct storage *st)
{
        uint64_t r = (uintptr_t)st;

        if (sizeof(void *) < 8) //lint !e506 !e774
                r <<= 1;

        return (r);
}

static inline struct storage *
lease2st(uint64_t l)
{

        if (sizeof(void *) < 8) //lint !e506 !e774
                l >>= 1;

        return ((void *)(uintptr_t)l);
}

static inline void
sml_ai_viov_fill(struct viov *viov, struct storage *st)
{
        viov->iov.iov_base = TRUST_ME(st->ptr);
        viov->iov.iov_len = st->len;
        viov->lease = st2lease(st);
        VAI_ASSERT_LEASE(viov->lease);
}

// sml has no mechanism to notify "I got free space again now"
// (we could add that, but because storage.h is used in mgt, a first attempt
//  looks at least like this would cause some include spill for vai_q_head or
//  something similar)
//
// So anyway, to get ahead we just implement a pretty stupid "call the notify
// some time later" on a thread
static void
sml_ai_later_task(struct worker *wrk, void *priv)
{
        struct sml_hdl *hdl;
        const vtim_dur dur = 0.0042;

        (void)wrk;
        VTIM_sleep(dur);
        CAST_VAI_HDL_NOTNULL(hdl, priv, SML_HDL_MAGIC);
        memset(&hdl->task, 0, sizeof hdl->task);
        hdl->qe.cb(hdl, hdl->qe.priv);
}
static void
sml_ai_later(struct worker *wrk, struct sml_hdl *hdl)
{
        AZ(hdl->task.func);
        AZ(hdl->task.priv);
        hdl->task.func = sml_ai_later_task;
        hdl->task.priv = hdl;
        AZ(Pool_Task(wrk->pool, &hdl->task, TASK_QUEUE_BG));
}


static int
sml_ai_buffer(struct worker *wrk, vai_hdl vhdl, struct vscarab *scarab)
{
        const struct stevedore *stv;
        struct sml_hdl *hdl;
        struct storage *st;
        struct viov *vio;
        int r = 0;

        (void) wrk;
        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
        stv = hdl->stv;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);

        VSCARAB_FOREACH(vio, scarab)
                if (vio->iov.iov_len > UINT_MAX)
                        return (-EINVAL);

        VSCARAB_FOREACH(vio, scarab) {
                st = objallocwithnuke(wrk, stv, vio->iov.iov_len, 0);
                if (st == NULL)
                        break;
                assert(st->space >= vio->iov.iov_len);
                st->flags = STORAGE_F_BUFFER;
                st->len = st->space;

                sml_ai_viov_fill(vio, st);
                r++;
        }
        if (r == 0) {
                sml_ai_later(wrk, hdl);
                r = -EAGAIN;
        }
        return (r);
}

static int
sml_ai_lease_simple(struct worker *wrk, vai_hdl vhdl, struct vscarab *scarab)
{
        struct storage *st;
        struct sml_hdl *hdl;
        struct viov *viov;
        int r = 0;

        (void) wrk;
        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
        VSCARAB_CHECK_NOTNULL(scarab);

        AZ(hdl->st_off);
        st = hdl->st;
        while (st != NULL && (viov = VSCARAB_GET(scarab)) != NULL) {
                CHECK_OBJ(st, STORAGE_MAGIC);
                sml_ai_viov_fill(viov, st);
                r++;
                st = VTAILQ_PREV(st, storagehead, list);
        }
        hdl->st = st;
        if (st == NULL)
                scarab->flags |= VSCARAB_F_END;
        return (r);
}

/*
 * on leases while streaming (with a boc):
 *
 * SML uses the lease return facility to implement the "free behind" for
 * OC_F_TRANSIENT objects. When streaming, we also return leases on
 * fragments of sts, but we must only "free behind" when we are done with the
 * last fragment.
 *
 * So we use a magic lease to signal "this is only a fragment", which we ignore
 * on returns
 */

static int
sml_ai_lease_boc(struct worker *wrk, vai_hdl vhdl, struct vscarab *scarab)
{
        enum boc_state_e state = BOS_INVALID;
        struct storage *next;
        struct sml_hdl *hdl;
        struct viov *viov;
        int r = 0;

        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
        VSCARAB_CHECK_NOTNULL(scarab);

        if (hdl->avail == hdl->returned) {
                hdl->avail = ObjVAIGetExtend(wrk, hdl->oc, hdl->returned,
                    &state, &hdl->qe);
                if (state == BOS_FAILED) {
                        hdl->last = NULL;
                        return (-EPIPE);
                }
                else if (state == BOS_FINISHED)
                        (void)0;
                else if (hdl->avail == hdl->returned) {
                        // ObjVAIGetExtend() has scheduled a notification
                        if (hdl->boc->transit_buffer > 0)
                                return (-ENOBUFS);
                        else
                                return (-EAGAIN);
                }
                else
                        assert(state < BOS_FINISHED);
        }
        Lck_Lock(&hdl->boc->mtx);
        if (hdl->st == NULL && hdl->last != NULL) {
                /* when the "last" st completed, we did not yet have a next, so
                 * resume from there. Because "last" might have been returned and
                 * deleted, we can not just use the pointer, but rather need to
                 * iterate the st list.
                 * if we can not find "last", it also has been returned and
                 * deleted, and the current write head (VTAILQ_LAST) is our next
                 * st, which can also be null if we are done.
                 */
                VTAILQ_FOREACH_REVERSE(next, &hdl->obj->list, storagehead, list) {
                        if (next == hdl->last) {
                                hdl->st = VTAILQ_PREV(next, storagehead, list);
                                break;
                        }
                }
        }
        hdl->last = NULL;
        if (hdl->st == NULL) {
                assert(hdl->returned == 0 || hdl->avail == hdl->returned);
                hdl->st = VTAILQ_LAST(&hdl->obj->list, storagehead);
        }
        if (hdl->st == NULL)
                assert(hdl->avail == hdl->returned);

        while (hdl->avail > hdl->returned && (viov = VSCARAB_GET(scarab)) != NULL) {
                CHECK_OBJ_NOTNULL(hdl->st, STORAGE_MAGIC); // ObjVAIGetExtend ensures
                assert(hdl->st_off <= hdl->st->space);
                size_t av = hdl->avail - hdl->returned;
                size_t l = hdl->st->space - hdl->st_off;
                AN(l);
                if (l > av)
                        l = av;
                viov->iov.iov_base = TRUST_ME(hdl->st->ptr + hdl->st_off);
                viov->iov.iov_len = l;
                if (hdl->st_off + l == hdl->st->space) {
                        next = VTAILQ_PREV(hdl->st, storagehead, list);
                        AZ(hdl->last);
                        if (next == NULL)
                                hdl->last = hdl->st;
                        else
                                CHECK_OBJ(next, STORAGE_MAGIC);
#ifdef VAI_DBG
                        VSLb(wrk->vsl, SLT_Debug, "off %zu + l %zu == space st %p next st %p stvprv %p",
                            hdl->st_off, l, hdl->st, next, hdl->boc->stevedore_priv);
#endif
                        viov->lease = st2lease(hdl->st);
                        hdl->st_off = 0;
                        hdl->st = next;
                }
                else {
                        viov->lease = VAI_LEASE_NORET;
                        hdl->st_off += l;
                }
                hdl->returned += l;
                VAI_ASSERT_LEASE(viov->lease);
                r++;
        }

        Lck_Unlock(&hdl->boc->mtx);
        if (state != BOS_FINISHED && hdl->avail == hdl->returned) {
                hdl->avail = ObjVAIGetExtend(wrk, hdl->oc, hdl->returned,
                    &state, &hdl->qe);
        }
        if (state == BOS_FINISHED && hdl->avail == hdl->returned)
                scarab->flags |= VSCARAB_F_END;
        return (r);
}

// return only buffers, used if object is not streaming
static void v_matchproto_(vai_return_f)
sml_ai_return_buffers(struct worker *wrk, vai_hdl vhdl, struct vscaret *scaret)
{
        struct storage *st;
        struct sml_hdl *hdl;
        uint64_t *p;

        (void) wrk;
        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);

        VSCARET_FOREACH(p, scaret) {
                if (*p == VAI_LEASE_NORET)
                        continue;
                CAST_OBJ_NOTNULL(st, lease2st(*p), STORAGE_MAGIC);
                if ((st->flags & STORAGE_F_BUFFER) == 0)
                        continue;
                sml_stv_free(hdl->stv, st);
        }
        VSCARET_INIT(scaret, scaret->capacity);
}

// generic return for buffers and object leases, used when streaming
static void v_matchproto_(vai_return_f)
sml_ai_return(struct worker *wrk, vai_hdl vhdl, struct vscaret *scaret)
{
        struct storage *st;
        struct sml_hdl *hdl;
        uint64_t *p;

        (void) wrk;
        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
        VSCARET_CHECK_NOTNULL(scaret);
        if (scaret->used == 0)
                return;

        // callback is only registered if needed
        assert(hdl->boc != NULL && (hdl->oc->flags & OC_F_TRANSIENT) != 0);

        // filter noret and last
        VSCARET_LOCAL(todo, scaret->used);
        VSCARET_FOREACH(p, scaret) {
                if (*p == VAI_LEASE_NORET)
                        continue;
                CAST_OBJ_NOTNULL(st, lease2st(*p), STORAGE_MAGIC);
                if (st == hdl->last)
                        continue;
                VSCARET_ADD(todo, *p);
        }
        VSCARET_INIT(scaret, scaret->capacity);

        Lck_Lock(&hdl->boc->mtx);
        VSCARET_FOREACH(p, todo) {
                CAST_OBJ_NOTNULL(st, lease2st(*p), STORAGE_MAGIC);
                if ((st->flags & STORAGE_F_BUFFER) != 0)
                        continue;
                VTAILQ_REMOVE(&hdl->obj->list, st, list);
                if (st == hdl->boc->stevedore_priv)
                        hdl->boc->stevedore_priv = trim_once;
        }
        Lck_Unlock(&hdl->boc->mtx);

        VSCARET_FOREACH(p, todo) {
                CAST_OBJ_NOTNULL(st, lease2st(*p), STORAGE_MAGIC);
                sml_stv_free(hdl->stv, st);
        }
}

static void v_matchproto_(vai_fini_f)
sml_ai_fini(struct worker *wrk, vai_hdl *vai_hdlp)
{
        struct sml_hdl *hdl;

        AN(vai_hdlp);
        CAST_VAI_HDL_NOTNULL(hdl, *vai_hdlp, SML_HDL_MAGIC);
        *vai_hdlp = NULL;

        if (hdl->boc != NULL) {
                ObjVAICancel(wrk, hdl->boc, &hdl->qe);
                HSH_DerefBoc(wrk, hdl->oc);
                hdl->boc = NULL;
        }

        if (hdl->ws != NULL)
                WS_Release(hdl->ws, 0);
        else
                free(hdl);
}

static vai_hdl v_matchproto_(vai_init_f)
sml_ai_init(struct worker *wrk, struct objcore *oc, struct ws *ws,
    vai_notify_cb *notify, void *notify_priv)
{
        struct sml_hdl *hdl;
        const size_t sz = sizeof *hdl;

        if (ws != NULL && WS_ReserveSize(ws, (unsigned)sz))
                hdl = WS_Reservation(ws);
        else {
                hdl = malloc(sz);
                ws = NULL;
        }

        AN(hdl);
        INIT_VAI_HDL(hdl, SML_HDL_MAGIC);
        hdl->preamble.vai_lease = sml_ai_lease_simple;
        hdl->preamble.vai_buffer = sml_ai_buffer;
        hdl->preamble.vai_return = sml_ai_return_buffers;
        hdl->preamble.vai_fini = sml_ai_fini;
        hdl->ws = ws;

        hdl->oc = oc;
        hdl->obj = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(hdl->obj, OBJECT_MAGIC);
        hdl->stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(hdl->stv, STEVEDORE_MAGIC);

        hdl->st = VTAILQ_LAST(&hdl->obj->list, storagehead);
        CHECK_OBJ_ORNULL(hdl->st, STORAGE_MAGIC);

        hdl->qe.magic = VAI_Q_MAGIC;
        hdl->qe.cb = notify;
        hdl->qe.hdl = hdl;
        hdl->qe.priv = notify_priv;

        hdl->boc = HSH_RefBoc(oc);
        if (hdl->boc == NULL)
                return (hdl);
        /* we only initialize notifications if we have a boc, so
         * any wrong attempt triggers magic checks.
         */
        hdl->preamble.vai_lease = sml_ai_lease_boc;
        if ((hdl->oc->flags & OC_F_TRANSIENT) != 0)
                hdl->preamble.vai_return = sml_ai_return;
        return (hdl);
}

/*
 * trivial notification to allow the iterator to simply block
 */
struct sml_notify {
        unsigned                magic;
#define SML_NOTIFY_MAGIC        0x4589af31
        unsigned                hasmore;
        pthread_mutex_t         mtx;
        pthread_cond_t          cond;
};

static void
sml_notify_init(struct sml_notify *sn)
{

        INIT_OBJ(sn, SML_NOTIFY_MAGIC);
        AZ(pthread_mutex_init(&sn->mtx, NULL));
        AZ(pthread_cond_init(&sn->cond, NULL));
}

static void
sml_notify_fini(struct sml_notify *sn)
{

        CHECK_OBJ_NOTNULL(sn, SML_NOTIFY_MAGIC);
        AZ(pthread_mutex_destroy(&sn->mtx));
        AZ(pthread_cond_destroy(&sn->cond));
}

static void v_matchproto_(vai_notify_cb)
sml_notify(vai_hdl hdl, void *priv)
{
        struct sml_notify *sn;

        (void) hdl;
        CAST_OBJ_NOTNULL(sn, priv, SML_NOTIFY_MAGIC);
        AZ(pthread_mutex_lock(&sn->mtx));
        sn->hasmore = 1;
        AZ(pthread_cond_signal(&sn->cond));
        AZ(pthread_mutex_unlock(&sn->mtx));

}

static void
sml_notify_wait(struct sml_notify *sn)
{

        CHECK_OBJ_NOTNULL(sn, SML_NOTIFY_MAGIC);
        AZ(pthread_mutex_lock(&sn->mtx));
        while (sn->hasmore == 0)
                AZ(pthread_cond_wait(&sn->cond, &sn->mtx));
        AN(sn->hasmore);
        sn->hasmore = 0;
        AZ(pthread_mutex_unlock(&sn->mtx));
}

static int v_matchproto_(objiterator_f)
sml_iterator(struct worker *wrk, struct objcore *oc,
    void *priv, objiterate_f *func, int final)
{
        struct sml_notify sn;
        struct viov *vio, *last;
        unsigned u, uu;
        vai_hdl hdl;
        int nn, r, r2, islast;

        VSCARAB_LOCAL(scarab, 16);
        VSCARET_LOCAL(scaret, 16);

        (void) final; // phase out?
        sml_notify_init(&sn);
        hdl = ObjVAIinit(wrk, oc, NULL, sml_notify, &sn);
        AN(hdl);

        r = u = 0;

        do {
                do {
                        nn = ObjVAIlease(wrk, hdl, scarab);
                        if (nn <= 0 || scarab->flags & VSCARAB_F_END)
                                break;
                } while (scarab->used < scarab->capacity);

                /*
                 * nn is the wait/return action or 0
                 * nn tells us if to flush
                 */
                uu = u;
                last = VSCARAB_LAST(scarab);
                VSCARAB_FOREACH(vio, scarab) {
                        islast = vio == last;
                        AZ(u & OBJ_ITER_END);
                        if (islast && scarab->flags & VSCARAB_F_END)
                                u |= OBJ_ITER_END;

                        // flush if it is the scarab's last IOV and we will block next
                        // or if we need space in the return leases array
                        uu = u;
                        if ((islast && nn < 0) || scaret->used == scaret->capacity - 1)
                                uu |= OBJ_ITER_FLUSH;
                        r = func(priv, uu, vio->iov.iov_base, vio->iov.iov_len);
                        if (r != 0)
                                break;

                        // sufficient space ensured by capacity check above
                        VSCARET_ADD(scaret, vio->lease);

                        // whenever we have flushed, return leases
                        if ((uu & OBJ_ITER_FLUSH) && scaret->used > 0)
                                ObjVAIreturn(wrk, hdl, scaret);
                }

                // return leases which we did not use if error (break)
                VSCARAB_FOREACH_RESUME(vio, scarab) {
                        if (scaret->used == scaret->capacity)
                                ObjVAIreturn(wrk, hdl, scaret);
                        VSCARET_ADD(scaret, vio->lease);
                }

                // we have now completed the scarab
                VSCARAB_INIT(scarab, scarab->capacity);

                // flush before blocking if we did not already
                if (r == 0 && (nn == -ENOBUFS || nn == -EAGAIN) &&
                    (uu & OBJ_ITER_FLUSH) == 0) {
                        r = func(priv, OBJ_ITER_FLUSH, NULL, 0);
                        if (scaret->used > 0)
                                ObjVAIreturn(wrk, hdl, scaret);
                }

                if (r == 0 && (nn == -ENOBUFS || nn == -EAGAIN)) {
                        assert(scaret->used <= 1);
                        sml_notify_wait(&sn);
                }
                else if (r == 0 && nn < 0)
                        r = -1;
        } while (nn != 0 && r == 0);

        if ((u & OBJ_ITER_END) == 0) {
                r2 = func(priv, OBJ_ITER_END, NULL, 0);
                if (r == 0)
                        r = r2;
        }

        if (scaret->used > 0)
                ObjVAIreturn(wrk, hdl, scaret);

        ObjVAIfini(wrk, &hdl);
        sml_notify_fini(&sn);

        return (r);
}

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

static struct storage *
objallocwithnuke(struct worker *wrk, const struct stevedore *stv, ssize_t size,
    int flags)
{
        struct storage *st = NULL;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);

        if (size > cache_param->fetch_maxchunksize) {
                if (!(flags & LESS_MEM_ALLOCED_IS_OK))
                        return (NULL);
                size = cache_param->fetch_maxchunksize;
        }

        assert(size <= UINT_MAX);       /* field limit in struct storage */

        do {
                /* try to allocate from it */
                st = sml_stv_alloc(stv, size, flags);
                if (st != NULL)
                        break;

                /* no luck; try to free some space and keep trying */
                if (stv->lru == NULL)
                        break;
        } while (LRU_NukeOne(wrk, stv->lru));

        CHECK_OBJ_ORNULL(st, STORAGE_MAGIC);
        return (st);
}

static int v_matchproto_(objgetspace_f)
sml_getspace(struct worker *wrk, struct objcore *oc, ssize_t *sz,
    uint8_t **ptr)
{
        struct object *o;
        struct storage *st;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
        CHECK_OBJ_NOTNULL(oc->boc, BOC_MAGIC);
        AN(sz);
        AN(ptr);
        if (*sz == 0)
                *sz = cache_param->fetch_chunksize;
        assert(*sz > 0);
        if (oc->boc->transit_buffer > 0)
                *sz = vmin_t(ssize_t, *sz, oc->boc->transit_buffer);

        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
        CHECK_OBJ_NOTNULL(oc->boc, BOC_MAGIC);

        st = VTAILQ_FIRST(&o->list);
        if (st != NULL && st->len < st->space) {
                *sz = st->space - st->len;
                *ptr = st->ptr + st->len;
                assert (*sz > 0);
                return (1);
        }

        st = objallocwithnuke(wrk, oc->stobj->stevedore, *sz,
            LESS_MEM_ALLOCED_IS_OK);
        if (st == NULL)
                return (0);

        CHECK_OBJ_NOTNULL(oc->boc, BOC_MAGIC);
        Lck_Lock(&oc->boc->mtx);
        VTAILQ_INSERT_HEAD(&o->list, st, list);
        Lck_Unlock(&oc->boc->mtx);

        *sz = st->space - st->len;
        assert (*sz > 0);
        *ptr = st->ptr + st->len;
        return (1);
}

static void v_matchproto_(objextend_f)
sml_extend(struct worker *wrk, struct objcore *oc, ssize_t l)
{
        struct object *o;
        struct storage *st;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        assert(l > 0);

        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
        st = VTAILQ_FIRST(&o->list);
        CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
        assert(st->len + l <= st->space);
        st->len += l;
}

static void v_matchproto_(objtrimstore_f)
sml_trimstore(struct worker *wrk, struct objcore *oc)
{
        const struct stevedore *stv;
        struct storage *st, *st1;
        struct object *o;

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

        stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);

        if (oc->boc->stevedore_priv != NULL)
                WRONG("sml_trimstore already called");
        oc->boc->stevedore_priv = trim_once;

        if (stv->sml_free == NULL)
                return;

        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
        st = VTAILQ_FIRST(&o->list);

        if (st == NULL)
                return;

        if (st->len == 0) {
                Lck_Lock(&oc->boc->mtx);
                VTAILQ_REMOVE(&o->list, st, list);
                Lck_Unlock(&oc->boc->mtx);
                /* sml_bocdone frees this */
                oc->boc->stevedore_priv = st;
                return;
        }

        if (st->space - st->len < 512)
                return;

        st1 = sml_stv_alloc(stv, st->len, 0);
        if (st1 == NULL)
                return;
        assert(st1->space >= st->len);

        memcpy(st1->ptr, st->ptr, st->len);
        st1->len = st->len;
        Lck_Lock(&oc->boc->mtx);
        VTAILQ_REMOVE(&o->list, st, list);
        VTAILQ_INSERT_HEAD(&o->list, st1, list);
        Lck_Unlock(&oc->boc->mtx);
        /* sml_bocdone frees this */
        oc->boc->stevedore_priv = st;
}

static void v_matchproto_(objbocdone_f)
sml_bocdone(struct worker *wrk, struct objcore *oc, struct boc *boc)
{
        const struct stevedore *stv;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
        CHECK_OBJ_NOTNULL(boc, BOC_MAGIC);
        stv = oc->stobj->stevedore;
        CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);

        sml_bocfini(stv, boc);

        if (stv->lru != NULL) {
                if (isnan(wrk->lastused))
                        wrk->lastused = VTIM_real();
                LRU_Add(oc, wrk->lastused);     // approx timestamp is OK
        }
}

static const void * v_matchproto_(objgetattr_f)
sml_getattr(struct worker *wrk, struct objcore *oc, enum obj_attr attr,
   ssize_t *len)
{
        struct object *o;
        ssize_t dummy;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);

        if (len == NULL)
                len = &dummy;
        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);

        switch (attr) {
                /* Fixed size attributes */
#define OBJ_FIXATTR(U, l, s)                                            \
        case OA_##U:                                                    \
                *len = sizeof o->fa_##l;                                \
                return (o->fa_##l);
#include "tbl/obj_attr.h"

                /* Variable size attributes */
#define OBJ_VARATTR(U, l)                                               \
        case OA_##U:                                                    \
                if (o->va_##l == NULL)                                  \
                        return (NULL);                                  \
                *len = o->va_##l##_len;                                 \
                return (o->va_##l);
#include "tbl/obj_attr.h"

                /* Auxiliary attributes */
#define OBJ_AUXATTR(U, l)                                               \
        case OA_##U:                                                    \
                if (o->aa_##l == NULL)                                  \
                        return (NULL);                                  \
                CHECK_OBJ_NOTNULL(o->aa_##l, STORAGE_MAGIC);            \
                *len = o->aa_##l->len;                                  \
                return (o->aa_##l->ptr);
#include "tbl/obj_attr.h"

        default:
                break;
        }
        WRONG("Unsupported OBJ_ATTR");
}

static void * v_matchproto_(objsetattr_f)
sml_setattr(struct worker *wrk, struct objcore *oc, enum obj_attr attr,
    ssize_t len, const void *ptr)
{
        struct object *o;
        void *retval = NULL;
        struct storage *st;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);

        o = sml_getobj(wrk, oc);
        CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
        st = o->objstore;

        switch (attr) {
                /* Fixed size attributes */
#define OBJ_FIXATTR(U, l, s)                                            \
        case OA_##U:                                                    \
                assert(len == sizeof o->fa_##l);                        \
                retval = o->fa_##l;                                     \
                break;
#include "tbl/obj_attr.h"

                /* Variable size attributes */
#define OBJ_VARATTR(U, l)                                               \
        case OA_##U:                                                    \
                if (o->va_##l##_len > 0) {                              \
                        AN(o->va_##l);                                  \
                        assert(len == o->va_##l##_len);                 \
                        retval = o->va_##l;                             \
                } else if (len > 0) {                                   \
                        assert(len <= UINT_MAX);                        \
                        assert(st->len + len <= st->space);             \
                        o->va_##l = st->ptr + st->len;                  \
                        st->len += len;                                 \
                        o->va_##l##_len = len;                          \
                        retval = o->va_##l;                             \
                }                                                       \
                break;
#include "tbl/obj_attr.h"

                /* Auxiliary attributes */
#define OBJ_AUXATTR(U, l)                                               \
        case OA_##U:                                                    \
                if (o->aa_##l != NULL) {                                \
                        CHECK_OBJ_NOTNULL(o->aa_##l, STORAGE_MAGIC);    \
                        assert(len == o->aa_##l->len);                  \
                        retval = o->aa_##l->ptr;                        \
                        break;                                          \
                }                                                       \
                if (len == 0)                                           \
                        break;                                          \
                o->aa_##l = objallocwithnuke(wrk, oc->stobj->stevedore, \
                    len, 0);                                            \
                if (o->aa_##l == NULL)                                  \
                        break;                                          \
                CHECK_OBJ_NOTNULL(o->aa_##l, STORAGE_MAGIC);            \
                assert(len <= o->aa_##l->space);                        \
                o->aa_##l->len = len;                                   \
                retval = o->aa_##l->ptr;                                \
                break;
#include "tbl/obj_attr.h"

        default:
                WRONG("Unsupported OBJ_ATTR");
                break;
        }

        if (retval != NULL && ptr != NULL)
                memcpy(retval, ptr, len);
        return (retval);
}

const struct obj_methods SML_methods = {
        .objfree        = sml_objfree,
        .objiterator    = sml_iterator,
        .objgetspace    = sml_getspace,
        .objextend      = sml_extend,
        .objtrimstore   = sml_trimstore,
        .objbocdone     = sml_bocdone,
        .objslim        = sml_slim,
        .objgetattr     = sml_getattr,
        .objsetattr     = sml_setattr,
        .objtouch       = LRU_Touch,
        .vai_init       = sml_ai_init
};

static void
sml_panic_st(struct vsb *vsb, const char *hd, const struct storage *st)
{
        VSB_printf(vsb, "%s = %p {priv=%p, ptr=%p, len=%u, space=%u},\n",
            hd, st, st->priv, st->ptr, st->len, st->space);
}

void
SML_panic(struct vsb *vsb, const struct objcore *oc)
{
        struct object *o;
        struct storage *st;

        VSB_printf(vsb, "Simple = %p,\n", oc->stobj->priv);
        if (oc->stobj->priv == NULL)
                return;
        o = oc->stobj->priv;
        PAN_CheckMagic(vsb, o, OBJECT_MAGIC);
        sml_panic_st(vsb, "Obj", o->objstore);

#define OBJ_FIXATTR(U, l, sz) \
        VSB_printf(vsb, "%s = ", #U); \
        VSB_quote(vsb, (const void*)o->fa_##l, sz, VSB_QUOTE_HEX); \
        VSB_printf(vsb, ",\n");

#define OBJ_VARATTR(U, l) \
        VSB_printf(vsb, "%s = {len=%u, ptr=%p},\n", \
            #U, o->va_##l##_len, o->va_##l);

#define OBJ_AUXATTR(U, l)                                               \
        do {                                                            \
                if (o->aa_##l != NULL) sml_panic_st(vsb, #U, o->aa_##l);\
        } while(0);

#include "tbl/obj_attr.h"

        VTAILQ_FOREACH(st, &o->list, list) {
                sml_panic_st(vsb, "Body", st);
        }
}

		varnish-cache/bin/varnishd/storage/storage_simple.c
0		/*-
1		* Copyright (c) 2007-2015 Varnish Software AS
2		* All rights reserved.
3		*
4		* Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
5		*
6		* SPDX-License-Identifier: BSD-2-Clause
7		*
8		* Redistribution and use in source and binary forms, with or without
9		* modification, are permitted provided that the following conditions
10		* are met:
11		* 1. Redistributions of source code must retain the above copyright
12		* notice, this list of conditions and the following disclaimer.
13		* 2. Redistributions in binary form must reproduce the above copyright
14		* notice, this list of conditions and the following disclaimer in the
15		* documentation and/or other materials provided with the distribution.
16		*
17		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
18		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
21		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
22		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
23		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
24		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
26		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
27		* SUCH DAMAGE.
28		*
29		*/
30
31		#include "config.h"
32
33		#include <stdlib.h>
34
35	2718	#include "cache/cache_varnishd.h"
36	2216
37	515	#include "cache/cache_obj.h"
38	357	#include "cache/cache_objhead.h"
39	2216
40		#include "storage/storage.h"
41		#include "storage/storage_simple.h"
42
43		#include "vtim.h"
44
45	3050	/* Flags for allocating memory in sml_stv_alloc */
46	5770	#define LESS_MEM_ALLOCED_IS_OK 1
47
48		// marker pointer for sml_trimstore
49		static void *trim_once = &trim_once;
50
51		/-------------------------------------------------------------------/
52	3970
53		static struct storage *
54		objallocwithnuke(struct worker , const struct stevedore , ssize_t size,
55		int flags);
56
57		static struct storage *
58	3425	sml_stv_alloc(const struct stevedore *stv, ssize_t size, int flags)
59		{
60		struct storage *st;
61
62	3425	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
63	3425	AN(stv->sml_alloc);
64
65	3425	if (!(flags & LESS_MEM_ALLOCED_IS_OK)) {
66	615	if (size > cache_param->fetch_maxchunksize)
67	0	return (NULL);
68		else
69	615	return (stv->sml_alloc(stv, size));
70		}
71
72	2810	if (size > cache_param->fetch_maxchunksize)
73	0	size = cache_param->fetch_maxchunksize;
74
75	2810	assert(size <= UINT_MAX); /* field limit in struct storage */
76
77	2945	for (;;) {
78		/* try to allocate from it */
79	2945	assert(size > 0);
80	2945	st = stv->sml_alloc(stv, size);
81	2945	if (st != NULL)
82	2799	break;
83
84	146	if (size <= cache_param->fetch_chunksize)
85	11	break;
86
87	135	size /= 2;
88		}
89	2810	CHECK_OBJ_ORNULL(st, STORAGE_MAGIC);
90	2810	return (st);
91	3425	}
92
93		static void
94	4408	sml_stv_free(const struct stevedore stv, struct storage st)
95		{
96
97	4408	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
98	4408	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
99	4408	if (stv->sml_free != NULL)
100	4408	stv->sml_free(st);
101	4408	}
102
103		/*--------------------------------------------------------------------
104		* This function is called by stevedores ->allocobj() method, which
105		* very often will be SML_allocobj() below, to convert a slab
106		* of storage into object which the stevedore can then register in its
107		* internal state, before returning it to STV_NewObject().
108		* As you probably guessed: All this for persistence.
109		*/
110
111		struct object *
112	2839	SML_MkObject(const struct stevedore stv, struct objcore oc, void *ptr)
113		{
114		struct object *o;
115
116	2839	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
117	2839	AN(stv->methods);
118	2839	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
119
120	2839	assert(PAOK(ptr));
121
122	2839	o = ptr;
123	2839	INIT_OBJ(o, OBJECT_MAGIC);
124
125	2839	VTAILQ_INIT(&o->list);
126
127	2839	oc->stobj->stevedore = stv;
128	2839	oc->stobj->priv = o;
129	2839	oc->stobj->priv2 = 0;
130	2839	return (o);
131		}
132
133		/*--------------------------------------------------------------------
134		* This is the default ->allocobj() which all stevedores who do not
135		* implement persistent storage can rely on.
136		*/
137
138		int v_matchproto_(storage_allocobj_f)
139	2828	SML_allocobj(struct worker wrk, const struct stevedore stv,
140		struct objcore *oc, unsigned wsl)
141		{
142		struct object *o;
143	2828	struct storage *st = NULL;
144		unsigned ltot;
145
146	2828	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
147	2828	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
148	2828	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
149
150	2828	AN(stv->sml_alloc);
151
152	2828	ltot = sizeof(*o) + PRNDUP(wsl);
153
154	2828	do {
155	2831	st = stv->sml_alloc(stv, ltot);
156	2831	if (st != NULL && st->space < ltot) {
157	0	stv->sml_free(st);
158	0	st = NULL;
159	0	}
160	2831	} while (st == NULL && LRU_NukeOne(wrk, stv->lru));
161	2828	if (st == NULL)
162	11	return (0);
163
164	2817	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
165	2817	o = SML_MkObject(stv, oc, st->ptr);
166	2817	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
167	2817	st->len = sizeof(*o);
168	2817	o->objstore = st;
169	2817	return (1);
170	2828	}
171
172		void * v_matchproto_(storage_allocbuf_t)
173	286	SML_AllocBuf(struct worker wrk, const struct stevedore stv, size_t size,
174		uintptr_t *ppriv)
175		{
176		struct storage *st;
177
178	286	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
179	286	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
180	286	AN(ppriv);
181
182	286	if (size > UINT_MAX)
183	0	return (NULL);
184	286	st = objallocwithnuke(wrk, stv, size, 0);
185	286	if (st == NULL)
186	0	return (NULL);
187	286	assert(st->space >= size);
188	286	st->flags = STORAGE_F_BUFFER;
189	286	st->len = size;
190	286	*ppriv = (uintptr_t)st;
191	286	return (st->ptr);
192	286	}
193
194		void v_matchproto_(storage_freebuf_t)
195	286	SML_FreeBuf(struct worker wrk, const struct stevedore stv, uintptr_t priv)
196		{
197		struct storage *st;
198
199	286	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
200	286	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
201
202	286	CAST_OBJ_NOTNULL(st, (void *)priv, STORAGE_MAGIC);
203	286	assert(st->flags == STORAGE_F_BUFFER);
204	286	sml_stv_free(stv, st);
205	286	}
206
207		/*---------------------------------------------------------------------
208		*/
209
210		static struct object *
211	155171	sml_getobj(struct worker wrk, struct objcore oc)
212		{
213		const struct stevedore *stv;
214		struct object *o;
215
216	155171	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
217	155171	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
218	155171	stv = oc->stobj->stevedore;
219	155171	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
220	155171	if (stv->sml_getobj != NULL)
221	290	return (stv->sml_getobj(wrk, oc));
222	154881	if (oc->stobj->priv == NULL)
223	0	return (NULL);
224	154881	CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);
225	154881	return (o);
226	155171	}
227
228		static void v_matchproto_(objslim_f)
229	3193	sml_slim(struct worker wrk, struct objcore oc)
230		{
231		const struct stevedore *stv;
232		struct object *o;
233		struct storage st, stn;
234
235	3193	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
236
237	3193	stv = oc->stobj->stevedore;
238	3193	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
239	3193	o = sml_getobj(wrk, oc);
240	3193	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
241
242		#define OBJ_AUXATTR(U, l) \
243		do { \
244		if (o->aa_##l != NULL) { \
245		sml_stv_free(stv, o->aa_##l); \
246		o->aa_##l = NULL; \
247		} \
248		} while (0);
249		#include "tbl/obj_attr.h"
250
251	4950	VTAILQ_FOREACH_SAFE(st, &o->list, list, stn) {
252	1757	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
253	1757	VTAILQ_REMOVE(&o->list, st, list);
254	1757	sml_stv_free(stv, st);
255	1757	}
256		}
257
258		static void
259	2856	sml_bocfini(const struct stevedore stv, struct boc boc)
260		{
261		struct storage *st;
262
263	2856	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
264	2856	CHECK_OBJ_NOTNULL(boc, BOC_MAGIC);
265
266	2856	if (boc->stevedore_priv == NULL \|\|
267	1879	boc->stevedore_priv == trim_once)
268	2654	return;
269
270		/* Free any leftovers from Trim */
271	202	TAKE_OBJ_NOTNULL(st, &boc->stevedore_priv, STORAGE_MAGIC);
272	202	sml_stv_free(stv, st);
273	2856	}
274
275		/*
276		* called in two cases:
277		* - oc->boc == NULL: cache object on LRU freed
278		* - oc->boc != NULL: cache object replaced for backend error
279		*/
280		static void v_matchproto_(objfree_f)
281	1788	sml_objfree(struct worker wrk, struct objcore oc)
282		{
283		const struct stevedore *stv;
284		struct storage *st;
285		struct object *o;
286
287	1788	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
288	1788	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
289	1788	stv = oc->stobj->stevedore;
290	1788	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
291	1788	CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);
292
293	1788	sml_slim(wrk, oc);
294	1788	st = o->objstore;
295	1788	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
296	1788	FINI_OBJ(o);
297
298	1788	if (oc->boc != NULL)
299	21	sml_bocfini(stv, oc->boc);
300	1767	else if (stv->lru != NULL)
301	1767	LRU_Remove(oc);
302
303	1788	sml_stv_free(stv, st);
304
305	1788	memset(oc->stobj, 0, sizeof oc->stobj);
306
307	1788	wrk->stats->n_object--;
308	1788	}
309
310		// kept for reviewers - XXX remove later
311		#undef VAI_DBG
312
313		struct sml_hdl {
314		struct vai_hdl_preamble preamble;
315		#define SML_HDL_MAGIC 0x37dfd996
316		struct vai_qe qe;
317		struct pool_task task; // unfortunate
318		struct ws *ws; // NULL is malloc()
319		struct objcore *oc;
320		struct object *obj;
321		const struct stevedore *stv;
322		struct boc *boc;
323
324		struct storage *st; // updated by _lease()
325
326		// only for _lease_boc()
327		uint64_t st_off; // already returned fragment of current st
328		uint64_t avail, returned;
329		struct storage *last; // to resume, held back by _return()
330		};
331
332		static inline uint64_t
333	3132	st2lease(const struct storage *st)
334		{
335	3132	uint64_t r = (uintptr_t)st;
336
337		if (sizeof(void *) < 8) //lint !e506 !e774
338		r <<= 1;
339
340	3132	return (r);
341		}
342
343		static inline struct storage *
344	3828	lease2st(uint64_t l)
345		{
346
347		if (sizeof(void *) < 8) //lint !e506 !e774
348		l >>= 1;
349
350	3828	return ((void *)(uintptr_t)l);
351		}
352
353		static inline void
354	1887	sml_ai_viov_fill(struct viov viov, struct storage st)
355		{
356	1887	viov->iov.iov_base = TRUST_ME(st->ptr);
357	1887	viov->iov.iov_len = st->len;
358	1887	viov->lease = st2lease(st);
359	1887	VAI_ASSERT_LEASE(viov->lease);
360	1887	}
361
362		// sml has no mechanism to notify "I got free space again now"
363		// (we could add that, but because storage.h is used in mgt, a first attempt
364		// looks at least like this would cause some include spill for vai_q_head or
365		// something similar)
366		//
367		// So anyway, to get ahead we just implement a pretty stupid "call the notify
368		// some time later" on a thread
369		static void
370	0	sml_ai_later_task(struct worker wrk, void priv)
371		{
372		struct sml_hdl *hdl;
373	0	const vtim_dur dur = 0.0042;
374
375	0	(void)wrk;
376	0	VTIM_sleep(dur);
377	0	CAST_VAI_HDL_NOTNULL(hdl, priv, SML_HDL_MAGIC);
378	0	memset(&hdl->task, 0, sizeof hdl->task);
379	0	hdl->qe.cb(hdl, hdl->qe.priv);
380	0	}
381		static void
382	0	sml_ai_later(struct worker wrk, struct sml_hdl hdl)
383		{
384	0	AZ(hdl->task.func);
385	0	AZ(hdl->task.priv);
386	0	hdl->task.func = sml_ai_later_task;
387	0	hdl->task.priv = hdl;
388	0	AZ(Pool_Task(wrk->pool, &hdl->task, TASK_QUEUE_BG));
389	0	}
390
391
392		static int
393	16	sml_ai_buffer(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
394		{
395		const struct stevedore *stv;
396		struct sml_hdl *hdl;
397		struct storage *st;
398		struct viov *vio;
399	16	int r = 0;
400
401	16	(void) wrk;
402	16	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
403	16	stv = hdl->stv;
404	16	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
405
406	32	VSCARAB_FOREACH(vio, scarab)
407	16	if (vio->iov.iov_len > UINT_MAX)
408	0	return (-EINVAL);
409
410	32	VSCARAB_FOREACH(vio, scarab) {
411	16	st = objallocwithnuke(wrk, stv, vio->iov.iov_len, 0);
412	16	if (st == NULL)
413	0	break;
414	16	assert(st->space >= vio->iov.iov_len);
415	16	st->flags = STORAGE_F_BUFFER;
416	16	st->len = st->space;
417
418	16	sml_ai_viov_fill(vio, st);
419	16	r++;
420	16	}
421	16	if (r == 0) {
422	0	sml_ai_later(wrk, hdl);
423	0	r = -EAGAIN;
424	0	}
425	16	return (r);
426	16	}
427
428		static int
429	3344	sml_ai_lease_simple(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
430		{
431		struct storage *st;
432		struct sml_hdl *hdl;
433		struct viov *viov;
434	3344	int r = 0;
435
436	3344	(void) wrk;
437	3344	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
438	3344	VSCARAB_CHECK_NOTNULL(scarab);
439
440	3344	AZ(hdl->st_off);
441	3344	st = hdl->st;
442	5215	while (st != NULL && (viov = VSCARAB_GET(scarab)) != NULL) {
443	1871	CHECK_OBJ(st, STORAGE_MAGIC);
444	1871	sml_ai_viov_fill(viov, st);
445	1871	r++;
446	1871	st = VTAILQ_PREV(st, storagehead, list);
447		}
448	3344	hdl->st = st;
449	3344	if (st == NULL)
450	3312	scarab->flags \|= VSCARAB_F_END;
451	3344	return (r);
452		}
453
454		/*
455		* on leases while streaming (with a boc):
456		*
457		* SML uses the lease return facility to implement the "free behind" for
458		* OC_F_TRANSIENT objects. When streaming, we also return leases on
459		* fragments of sts, but we must only "free behind" when we are done with the
460		* last fragment.
461		*
462		* So we use a magic lease to signal "this is only a fragment", which we ignore
463		* on returns
464		*/
465
466		static int
467	3389	sml_ai_lease_boc(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
468		{
469	3389	enum boc_state_e state = BOS_INVALID;
470		struct storage *next;
471		struct sml_hdl *hdl;
472		struct viov *viov;
473	3389	int r = 0;
474
475	3389	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
476	3389	VSCARAB_CHECK_NOTNULL(scarab);
477
478	3389	if (hdl->avail == hdl->returned) {
479	6712	hdl->avail = ObjVAIGetExtend(wrk, hdl->oc, hdl->returned,
480	3356	&state, &hdl->qe);
481	3356	if (state == BOS_FAILED) {
482	15	hdl->last = NULL;
483	15	return (-EPIPE);
484		}
485	3341	else if (state == BOS_FINISHED)
486	730	(void)0;
487	2611	else if (hdl->avail == hdl->returned) {
488		// ObjVAIGetExtend() has scheduled a notification
489	1337	if (hdl->boc->transit_buffer > 0)
490	329	return (-ENOBUFS);
491		else
492	1008	return (-EAGAIN);
493		}
494		else
495	1274	assert(state < BOS_FINISHED);
496	2004	}
497	2037	Lck_Lock(&hdl->boc->mtx);
498	2037	if (hdl->st == NULL && hdl->last != NULL) {
499		/* when the "last" st completed, we did not yet have a next, so