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_int.h"

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

#include "storage/storage.h"
#include "storage/storage_simple.h"
#include "storage/storage_vai.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(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 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 = ptr2lease(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);
        assert(hdl->boc == hdl->oc->boc);

        if (hdl->avail == hdl->returned) {
                hdl->avail = ObjVAIGetExtend(wrk, hdl->oc, hdl->returned,
                    &state, &hdl->qe);
                assert(state >= BOS_STREAM);
                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)
                hdl->st = VTAILQ_PREV(hdl->last, storagehead, list);
        if (hdl->last != NULL && state < BOS_FINISHED) {
                viov = VSCARAB_GET(scarab);
                AN(viov);
                viov->iov = IOV_NIL;
                viov->lease = ptr2lease(hdl->last);
                r++;
        }
        if (hdl->last != NULL)
                hdl->last = NULL;
        if (hdl->st == NULL && hdl->returned == 0)
                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->boc == hdl->oc->boc);
                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;
                                viov->lease = VAI_LEASE_NORET;
                        }
                        else {
                                CHECK_OBJ(next, STORAGE_MAGIC);
                                viov->lease = ptr2lease(hdl->st);
                        }
#ifdef VAI_DBG
                        if (wrk->vsl)
                                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
                        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;
        viov_lease_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, lease2ptr(*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;
        viov_lease_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, lease2ptr(*p), STORAGE_MAGIC);
                VSCARET_ADD(todo, *p);
        }
        VSCARET_INIT(scaret, scaret->capacity);

        Lck_Lock(&hdl->boc->mtx);
        VSCARET_FOREACH(p, todo) {
                CAST_OBJ_NOTNULL(st, lease2ptr(*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, lease2ptr(*p), STORAGE_MAGIC);
#ifdef VAI_DBG
                if (wrk->vsl != NULL)
                        VSLb(wrk->vsl, SLT_Debug, "ret %p", st);
#endif
                sml_stv_free(hdl->stv, st);
        }
}

// call the notify_cb to allow VDPs to use -EAGAIN also
static void v_matchproto_(vai_return_f)
sml_ai_notify(struct worker *wrk, vai_hdl vhdl)
{
        struct sml_hdl *hdl;

        (void) wrk;
        CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
        CHECK_OBJ(&hdl->qe, VAI_Q_MAGIC);
        hdl->qe.cb(vhdl, hdl->qe.priv);
}

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_notify = sml_ai_notify;
        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->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) {
                hdl->st = VTAILQ_LAST(&hdl->obj->list, storagehead);
                CHECK_OBJ_ORNULL(hdl->st, STORAGE_MAGIC);
                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);
        PTOK(pthread_mutex_init(&sn->mtx, NULL));
        PTOK(pthread_cond_init(&sn->cond, NULL));
}

static void
sml_notify_fini(struct sml_notify *sn)
{

        CHECK_OBJ_NOTNULL(sn, SML_NOTIFY_MAGIC);
        PTOK(pthread_mutex_destroy(&sn->mtx));
        PTOK(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);
        PTOK(pthread_mutex_lock(&sn->mtx));
        sn->hasmore = 1;
        PTOK(pthread_cond_signal(&sn->cond));
        PTOK(pthread_mutex_unlock(&sn->mtx));

}

static void
sml_notify_wait(struct sml_notify *sn)
{

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

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;

                        // null iov with the only purpose to return the resume ptr lease
                        // exception needed because assert(len > 0) in VDP_bytes()
                        if (vio->iov.iov_base == IOV_NIL.iov_base)
                                r = 0;
                        else
                                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);

#ifdef VAI_DBG
                        if (wrk->vsl)
                                VSLb(wrk->vsl, SLT_Debug, "len %zu scaret %u uu %u",
                                    vio->iov.iov_len, scaret->used, uu);
#endif

                        // 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);

#ifdef VAI_DBG
                if (wrk->vsl)
                        VSLb(wrk->vsl, SLT_Debug, "r %d nn %d uu %u",
                            r, nn, uu);
#endif

                // 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);
        }
}

		vinyl-cache/bin/vinyld/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	9288	#include "cache/cache_int.h"
36	9320
37	2132	#include "cache/cache_obj.h"
38	1445	#include "cache/cache_objhead.h"
39	9320
40		#include "storage/storage.h"
41		#include "storage/storage_simple.h"
42		#include "storage/storage_vai.h"
43
44		#include "vtim.h"
45	12200
46	24458	/* Flags for allocating memory in sml_stv_alloc */
47		#define LESS_MEM_ALLOCED_IS_OK 1
48
49		// marker pointer for sml_trimstore
50		static void *trim_once = &trim_once;
51
52	12397	/-------------------------------------------------------------------/
53
54		static struct storage *
55		objallocwithnuke(struct worker , const struct stevedore , ssize_t size,
56		int flags);
57
58		static struct storage *
59	12134	sml_stv_alloc(const struct stevedore *stv, ssize_t size, int flags)
60		{
61		struct storage *st;
62
63	12134	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
64	12134	AN(stv->sml_alloc);
65
66	12134	if (!(flags & LESS_MEM_ALLOCED_IS_OK)) {
67	2538	if (size > cache_param->fetch_maxchunksize)
68	0	return (NULL);
69		else
70	2538	return (stv->sml_alloc(stv, size));
71		}
72
73	9596	if (size > cache_param->fetch_maxchunksize)
74	0	size = cache_param->fetch_maxchunksize;
75
76	9596	assert(size <= UINT_MAX); /* field limit in struct storage */
77
78	10136	for (;;) {
79		/* try to allocate from it */
80	10136	assert(size > 0);
81	10136	st = stv->sml_alloc(stv, size);
82	10136	if (st != NULL)
83	9552	break;
84
85	584	if (size <= cache_param->fetch_chunksize)
86	44	break;
87
88	540	size /= 2;
89		}
90	9596	CHECK_OBJ_ORNULL(st, STORAGE_MAGIC);
91	9596	return (st);
92	12134	}
93
94		static void
95	14335	sml_stv_free(const struct stevedore stv, struct storage st)
96		{
97
98	14335	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
99	14335	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
100	14335	if (stv->sml_free != NULL)
101	14335	stv->sml_free(st);
102	14335	}
103
104		/*--------------------------------------------------------------------
105		* This function is called by stevedores ->allocobj() method, which
106		* very often will be SML_allocobj() below, to convert a slab
107		* of storage into object which the stevedore can then register in its
108		* internal state, before returning it to STV_NewObject().
109		* As you probably guessed: All this for persistence.
110		*/
111
112		struct object *
113	9824	SML_MkObject(const struct stevedore stv, struct objcore oc, void *ptr)
114		{
115		struct object *o;
116
117	9824	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
118	9824	AN(stv->methods);
119	9824	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
120
121	9824	assert(PAOK(ptr));
122
123	9824	o = ptr;
124	9824	INIT_OBJ(o, OBJECT_MAGIC);
125
126	9824	VTAILQ_INIT(&o->list);
127
128	9824	oc->stobj->stevedore = stv;
129	9824	oc->stobj->priv = o;
130	9824	oc->stobj->priv2 = 0;
131	9824	return (o);
132		}
133
134		/*--------------------------------------------------------------------
135		* This is the default ->allocobj() which all stevedores who do not
136		* implement persistent storage can rely on.
137		*/
138
139		int v_matchproto_(storage_allocobj_f)
140	9776	SML_allocobj(struct worker wrk, const struct stevedore stv,
141		struct objcore *oc, unsigned wsl)
142		{
143		struct object *o;
144	9776	struct storage *st = NULL;
145		unsigned ltot;
146
147	9776	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
148	9776	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
149	9776	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
150
151	9776	AN(stv->sml_alloc);
152
153	9776	ltot = sizeof(*o) + PRNDUP(wsl);
154
155	9776	do {
156	9788	st = stv->sml_alloc(stv, ltot);
157	9788	if (st != NULL && st->space < ltot) {
158	0	stv->sml_free(st);
159	0	st = NULL;
160	0	}
161	9788	} while (st == NULL && LRU_NukeOne(wrk, stv->lru));
162	9776	if (st == NULL)
163	36	return (0);
164
165	9740	CHECK_OBJ(st, STORAGE_MAGIC);
166	9740	o = SML_MkObject(stv, oc, st->ptr);
167	9740	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
168	9740	st->len = sizeof(*o);
169	9740	o->objstore = st;
170	9740	return (1);
171	9776	}
172
173		void * v_matchproto_(storage_allocbuf_t)
174	1176	SML_AllocBuf(struct worker wrk, const struct stevedore stv, size_t size,
175		uintptr_t *ppriv)
176		{
177		struct storage *st;
178
179	1176	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
180	1176	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
181	1176	AN(ppriv);
182
183	1176	if (size > UINT_MAX)
184	0	return (NULL);
185	1176	st = objallocwithnuke(wrk, stv, size, 0);
186	1176	if (st == NULL)
187	0	return (NULL);
188	1176	assert(st->space >= size);
189	1176	st->flags = STORAGE_F_BUFFER;
190	1176	st->len = size;
191	1176	*ppriv = (uintptr_t)st;
192	1176	return (st->ptr);
193	1176	}
194
195		void v_matchproto_(storage_freebuf_t)
196	1176	SML_FreeBuf(struct worker wrk, const struct stevedore stv, uintptr_t priv)
197		{
198		struct storage *st;
199
200	1176	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
201	1176	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
202
203	1176	CAST_OBJ_NOTNULL(st, (void *)priv, STORAGE_MAGIC);
204	1176	assert(st->flags == STORAGE_F_BUFFER);
205	1176	sml_stv_free(stv, st);
206	1176	}
207
208		/*---------------------------------------------------------------------
209		*/
210
211		static struct object *
212	625506	sml_getobj(struct worker wrk, struct objcore oc)
213		{
214		const struct stevedore *stv;
215		struct object *o;
216
217	625506	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
218	625506	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
219	625506	stv = oc->stobj->stevedore;
220	625506	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
221	625506	if (stv->sml_getobj != NULL)
222	1156	return (stv->sml_getobj(wrk, oc));
223	624350	if (oc->stobj->priv == NULL)
224	0	return (NULL);
225	624350	CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);
226	624350	return (o);
227	625506	}
228
229		static void v_matchproto_(objslim_f)
230	9237	sml_slim(struct worker wrk, struct objcore oc)
231		{
232		const struct stevedore *stv;
233		struct object *o;
234		struct storage st, stn;
235
236	9237	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
237
238	9237	stv = oc->stobj->stevedore;
239	9237	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
240	9237	o = sml_getobj(wrk, oc);
241	9237	CHECK_OBJ_NOTNULL(o, OBJECT_MAGIC);
242
243		#define OBJ_AUXATTR(U, l) \
244		do { \
245		if (o->aa_##l != NULL) { \
246		sml_stv_free(stv, o->aa_##l); \
247		o->aa_##l = NULL; \
248		} \
249		} while (0);
250		#include "tbl/obj_attr.h"
251
252	14005	VTAILQ_FOREACH_SAFE(st, &o->list, list, stn) {
253	4768	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
254	4768	VTAILQ_REMOVE(&o->list, st, list);
255	4768	sml_stv_free(stv, st);
256	4768	}
257		}
258
259		static void
260	9888	sml_bocfini(const struct stevedore stv, struct boc boc)
261		{
262		struct storage *st;
263
264	9888	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
265	9888	CHECK_OBJ_NOTNULL(boc, BOC_MAGIC);
266
267	9888	if (boc->stevedore_priv == NULL \|\|
268	5938	boc->stevedore_priv == trim_once)
269	9058	return;
270
271		/* Free any leftovers from Trim */
272	830	TAKE_OBJ_NOTNULL(st, &boc->stevedore_priv, STORAGE_MAGIC);
273	830	sml_stv_free(stv, st);
274	9888	}
275
276		/*
277		* called in two cases:
278		* - oc->boc == NULL: cache object on LRU freed
279		* - oc->boc != NULL: cache object replaced for backend error
280		*/
281		static void v_matchproto_(objfree_f)
282	5489	sml_objfree(struct worker wrk, struct objcore oc)
283		{
284		const struct stevedore *stv;
285		struct storage *st;
286		struct object *o;
287
288	5489	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
289	5489	CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
290	5489	stv = oc->stobj->stevedore;
291	5489	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
292	5489	CAST_OBJ_NOTNULL(o, oc->stobj->priv, OBJECT_MAGIC);
293
294	5489	sml_slim(wrk, oc);
295	5489	st = o->objstore;
296	5489	CHECK_OBJ_NOTNULL(st, STORAGE_MAGIC);
297	5489	FINI_OBJ(o);
298
299	5489	if (oc->boc != NULL)
300	92	sml_bocfini(stv, oc->boc);
301	5397	else if (stv->lru != NULL)
302	5396	LRU_Remove(oc);
303
304	5489	sml_stv_free(stv, st);
305
306	5489	memset(oc->stobj, 0, sizeof oc->stobj);
307
308	5489	wrk->stats->n_object--;
309	5489	}
310
311		// kept for reviewers - XXX remove later
312		#undef VAI_DBG
313
314		struct sml_hdl {
315		struct vai_hdl_preamble preamble;
316		#define SML_HDL_MAGIC 0x37dfd996
317		struct vai_qe qe;
318		struct pool_task task; // unfortunate
319		struct ws *ws; // NULL is malloc()
320		struct objcore *oc;
321		struct object *obj;
322		const struct stevedore *stv;
323		struct boc *boc;
324
325		struct storage *st; // updated by _lease()
326
327		// only for _lease_boc()
328		uint64_t st_off; // already returned fragment of current st
329		uint64_t avail, returned;
330		struct storage *last; // to resume, held back by _return()
331		};
332
333		static inline void
334	5909	sml_ai_viov_fill(struct viov viov, struct storage st)
335		{
336	5909	viov->iov.iov_base = TRUST_ME(st->ptr);
337	5909	viov->iov.iov_len = st->len;
338	5909	viov->lease = ptr2lease(st);
339	5909	VAI_ASSERT_LEASE(viov->lease);
340	5909	}
341
342		// sml has no mechanism to notify "I got free space again now"
343		// (we could add that, but because storage.h is used in mgt, a first attempt
344		// looks at least like this would cause some include spill for vai_q_head or
345		// something similar)
346		//
347		// So anyway, to get ahead we just implement a pretty stupid "call the notify
348		// some time later" on a thread
349		static void
350	0	sml_ai_later_task(struct worker wrk, void priv)
351		{
352		struct sml_hdl *hdl;
353	0	const vtim_dur dur = 0.0042;
354
355	0	(void)wrk;
356	0	VTIM_sleep(dur);
357	0	CAST_VAI_HDL_NOTNULL(hdl, priv, SML_HDL_MAGIC);
358	0	memset(&hdl->task, 0, sizeof hdl->task);
359	0	hdl->qe.cb(hdl, hdl->qe.priv);
360	0	}
361		static void
362	0	sml_ai_later(struct worker wrk, struct sml_hdl hdl)
363		{
364	0	AZ(hdl->task.func);
365	0	AZ(hdl->task.priv);
366	0	hdl->task.func = sml_ai_later_task;
367	0	hdl->task.priv = hdl;
368	0	AZ(Pool_Task(wrk->pool, &hdl->task, TASK_QUEUE_BG));
369	0	}
370
371
372		static int
373	64	sml_ai_buffer(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
374		{
375		const struct stevedore *stv;
376		struct sml_hdl *hdl;
377		struct storage *st;
378		struct viov *vio;
379	64	int r = 0;
380
381	64	(void) wrk;
382	64	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
383	64	stv = hdl->stv;
384	64	CHECK_OBJ_NOTNULL(stv, STEVEDORE_MAGIC);
385
386	128	VSCARAB_FOREACH(vio, scarab)
387	64	if (vio->iov.iov_len > UINT_MAX)
388	0	return (-EINVAL);
389
390	128	VSCARAB_FOREACH(vio, scarab) {
391	64	st = objallocwithnuke(wrk, stv, vio->iov.iov_len, 0);
392	64	if (st == NULL)
393	0	break;
394	64	assert(st->space >= vio->iov.iov_len);
395	64	st->flags = STORAGE_F_BUFFER;
396	64	st->len = st->space;
397
398	64	sml_ai_viov_fill(vio, st);
399	64	r++;
400	64	}
401	64	if (r == 0) {
402	0	sml_ai_later(wrk, hdl);
403	0	r = -EAGAIN;
404	0	}
405	64	return (r);
406	64	}
407
408		static int
409	9993	sml_ai_lease_simple(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
410		{
411		struct storage *st;
412		struct sml_hdl *hdl;
413		struct viov *viov;
414	9993	int r = 0;
415
416	9993	(void) wrk;
417	9993	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
418	9993	VSCARAB_CHECK_NOTNULL(scarab);
419
420	9993	AZ(hdl->st_off);
421	9993	st = hdl->st;
422	15838	while (st != NULL && (viov = VSCARAB_GET(scarab)) != NULL) {
423	5845	CHECK_OBJ(st, STORAGE_MAGIC);
424	5845	sml_ai_viov_fill(viov, st);
425	5845	r++;
426	5845	st = VTAILQ_PREV(st, storagehead, list);
427		}
428	9993	hdl->st = st;
429	9993	if (st == NULL)
430	9989	scarab->flags \|= VSCARAB_F_END;
431	9993	return (r);
432		}
433
434		/*
435		* on leases while streaming (with a boc):
436		*
437		* SML uses the lease return facility to implement the "free behind" for
438		* OC_F_TRANSIENT objects. When streaming, we also return leases on
439		* fragments of sts, but we must only "free behind" when we are done with the
440		* last fragment.
441		*
442		* So we use a magic lease to signal "this is only a fragment", which we ignore
443		* on returns
444		*/
445
446		static int
447	20896	sml_ai_lease_boc(struct worker wrk, vai_hdl vhdl, struct vscarab scarab)
448		{
449	20896	enum boc_state_e state = BOS_INVALID;
450		struct storage *next;
451		struct sml_hdl *hdl;
452		struct viov *viov;
453	20896	int r = 0;
454
455	20896	CAST_VAI_HDL_NOTNULL(hdl, vhdl, SML_HDL_MAGIC);
456	20896	VSCARAB_CHECK_NOTNULL(scarab);
457	20896	assert(hdl->boc == hdl->oc->boc);
458
459	20896	if (hdl->avail == hdl->returned) {
460	40004	hdl->avail = ObjVAIGetExtend(wrk, hdl->oc, hdl->returned,
461	20002	&state, &hdl->qe);
462	20002	assert(state >= BOS_STREAM);
463	20002	if (state == BOS_FAILED) {
464	68	hdl->last = NULL;
465	68	return (-EPIPE);
466		}
467	19934	else if (state == BOS_FINISHED)
468	3140	(void)0;
469	16794	else if (hdl->avail == hdl->returned) {
470		// ObjVAIGetExtend() has scheduled a notification
471	8298	if (hdl->boc->transit_buffer > 0)
472	1036	return (-ENOBUFS);
473		else
474	7262	return (-EAGAIN);
475		}
476		else
477	8496	assert(state < BOS_FINISHED);
478	11636	}
479	12530	Lck_Lock(&hdl->boc->mtx);
480	12530	if (hdl->st == NULL && hdl->last != NULL)
481	3629	hdl->st = VTAILQ_PREV(hdl->last, storagehead, list);
482	12530	if (hdl->last != NULL && state < BOS_FINISHED) {
483	643	viov = VSCARAB_GET(scarab);
484	643	AN(viov);
485	643	viov->iov = IOV_NIL;
486	643	viov->lease = ptr2lease(hdl->last);
487	643	r++;
488	643	}
489	12530	if (hdl->last != NULL)
490	3629	hdl->last = NULL;
491	12530	if (hdl->st == NULL && hdl->returned == 0)
492	3224	hdl->st = VTAILQ_LAST(&hdl->obj->list, storagehead);
493	12530	if (hdl->st == NULL)
494	2982	assert(hdl->avail == hdl->returned);
495
496	22626	while (hdl->avail > hdl->returned && (viov = VSCARAB_GET(scarab)) != NULL) {
497	10096	CHECK_OBJ_NOTNULL(hdl->st, STORAGE_MAGIC); // ObjVAIGetExtend ensures
498	10096	assert(hdl->boc == hdl->oc->boc);
499	10096	assert(hdl->st_off <= hdl->st->space);