GCOV - Varnish Cache


/*-
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006-2011 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.
 *
 * Worker thread stuff unrelated to the worker thread pools.
 *
 * --
 * signaling_note:
 *
 * note on worker wakeup signaling through the wrk condition variable (cv)
 *
 * In the general case, a cv needs to be signaled while holding the
 * corresponding mutex, otherwise the signal may be posted before the waiting
 * thread could register itself on the cv and, consequently, the signal may be
 * missed.
 *
 * In our case, any worker thread which we wake up comes from the idle queue,
 * where it put itself under the mutex, releasing that mutex implicitly via
 * Lck_CondWaitUntil() (which calls some variant of pthread_cond_wait). So we avoid
 * additional mutex contention knowing that any worker thread on the idle queue
 * is blocking on the cv.
 *
 * Except -- when it isn't, because it woke up for releasing its VCL
 * Reference. To account for this case, we check if the task function has been
 * set in the meantime, which in turn requires all of the task preparation to be
 * done holding the pool mutex. (see also #2719)
 */

#include "config.h"

#include <stdlib.h>
#include <sched.h>

#include "cache_int.h"
#include "cache_pool.h"

#include "vcli_serve.h"
#include "vtim.h"

#include "hash/hash_slinger.h"

static void Pool_Work_Thread(struct pool *pp, struct worker *wrk);

static uintmax_t reqpoolfail;

/*--------------------------------------------------------------------
 * Return all cached resources
 */

void
WRK_Cleanup(const struct worker *wrk)
{

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(wrk->wpriv, WORKER_PRIV_MAGIC);
        HSH_Cleanup(wrk);
        if (wrk->wpriv->vcl != NULL)
                VCL_Rel(&wrk->wpriv->vcl);
}

/*--------------------------------------------------------------------
 * Create and start a back-ground thread which as its own worker and
 * session data structures;
 */

struct bgthread {
        unsigned        magic;
#define BGTHREAD_MAGIC  0x23b5152b
        const char      *name;
        bgthread_t      *func;
        void            *priv;
};

static void *
wrk_bgthread(void *arg)
{
        struct bgthread *bt;
        struct worker wrk;
        struct worker_priv wpriv[1];
        struct VSC_main_wrk ds;
        void *r;

        CAST_OBJ_NOTNULL(bt, arg, BGTHREAD_MAGIC);
        THR_SetName(bt->name);
        THR_Init();
        INIT_OBJ(&wrk, WORKER_MAGIC);
        INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
        wrk.wpriv = wpriv;
        // bgthreads do not have a vpi member
        memset(&ds, 0, sizeof ds);
        wrk.stats = &ds;

        r = bt->func(&wrk, bt->priv);
        WRK_Cleanup(&wrk);
        Pool_Sumstat(&wrk);
        return (r);
}

void
WRK_BgThread(pthread_t *thr, const char *name, bgthread_t *func, void *priv)
{
        struct bgthread *bt;

        ALLOC_OBJ(bt, BGTHREAD_MAGIC);
        AN(bt);

        bt->name = name;
        bt->func = func;
        bt->priv = priv;
        PTOK(pthread_create(thr, NULL, wrk_bgthread, bt));
}

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

static void
WRK_Thread(struct pool *qp, size_t stacksize, unsigned thread_workspace)
{
        // child_signal_handler stack overflow check uses struct worker addr
        struct worker *w, ww;
        struct VSC_main_wrk ds;
        unsigned char ws[thread_workspace];
        struct worker_priv wpriv[1];
        unsigned char vpi[vpi_wrk_len];

        AN(qp);
        AN(stacksize);
        AN(thread_workspace);

        THR_SetName("cache-worker");
        w = &ww;
        INIT_OBJ(w, WORKER_MAGIC);
        INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
        w->wpriv = wpriv;
        w->lastused = NAN;
        memset(&ds, 0, sizeof ds);
        w->stats = &ds;
        THR_SetWorker(w);
        PTOK(pthread_cond_init(&w->cond, NULL));

        WS_Init(w->aws, "wrk", ws, thread_workspace);
        VPI_wrk_init(w, vpi, sizeof vpi);
        AN(w->vpi);

        VSL(SLT_WorkThread, NO_VXID, "%p start", w);

        Pool_Work_Thread(qp, w);
        AZ(w->pool);

        VSL(SLT_WorkThread, NO_VXID, "%p end", w);
        PTOK(pthread_cond_destroy(&w->cond));
        WRK_Cleanup(w);
        Pool_Sumstat(w);
}

/*--------------------------------------------------------------------
 * Summing of stats into pool counters
 */

static unsigned
wrk_addstat(const struct worker *wrk, const struct pool_task *tp, unsigned locked)
{
        struct pool *pp;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        pp = wrk->pool;
        CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
        if (locked)
                Lck_AssertHeld(&pp->mtx);

        if ((tp == NULL && wrk->stats->summs > 0) ||
            (wrk->stats->summs >= cache_param->wthread_stats_rate)) {
                if (!locked)
                        Lck_Lock(&pp->mtx);

                pp->a_stat->summs++;
                VSC_main_Summ_wrk_wrk(pp->a_stat, wrk->stats);
                memset(wrk->stats, 0, sizeof *wrk->stats);

                if (!locked)
                        Lck_Unlock(&pp->mtx);
        }

        return (tp != NULL);
}

void
WRK_AddStat(const struct worker *wrk)
{

        (void)wrk_addstat(wrk, wrk->task, 0);
        wrk->stats->summs++;
}

/*--------------------------------------------------------------------
 * Pool reserve calculation
 */

static unsigned
pool_reserve(void)
{
        unsigned lim;

        if (cache_param->wthread_reserve == 0) {
                lim = cache_param->wthread_min / 20 + 1;
        } else {
                lim = cache_param->wthread_min * 950 / 1000;
                if (cache_param->wthread_reserve < lim)
                        lim = cache_param->wthread_reserve;
        }
        if (lim < TASK_QUEUE_RESERVE)
                return (TASK_QUEUE_RESERVE);
        return (lim);
}

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

static struct worker *
pool_getidleworker(struct pool *pp, enum task_prio prio)
{
        struct pool_task *pt = NULL;
        struct worker *wrk;

        CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
        Lck_AssertHeld(&pp->mtx);
        if (pp->nidle > (pool_reserve() * prio / TASK_QUEUE_RESERVE)) {
                pt = VTAILQ_FIRST(&pp->idle_queue);
                if (pt == NULL)
                        AZ(pp->nidle);
        }

        if (pt == NULL)
                return (NULL);

        AZ(pt->func);
        CAST_OBJ_NOTNULL(wrk, pt->priv, WORKER_MAGIC);

        AN(pp->nidle);
        VTAILQ_REMOVE(&pp->idle_queue, wrk->task, list);
        pp->nidle--;

        return (wrk);
}

/*--------------------------------------------------------------------
 * Special scheduling:  If no thread can be found, the current thread
 * will be prepared for rescheduling instead.
 * The selected threads workspace is reserved and the argument put there.
 * Return one if another thread was scheduled, otherwise zero.
 */

int
Pool_Task_Arg(struct worker *wrk, enum task_prio prio, task_func_t *func,
    const void *arg, size_t arg_len)
{
        struct pool *pp;
        struct worker *wrk2;
        int retval;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        AN(arg);
        AN(arg_len);
        pp = wrk->pool;
        CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);

        Lck_Lock(&pp->mtx);
        wrk2 = pool_getidleworker(pp, prio);
        if (wrk2 != NULL)
                retval = 1;
        else {
                wrk2 = wrk;
                retval = 0;
        }
        AZ(wrk2->task->func);
        assert(arg_len <= WS_ReserveSize(wrk2->aws, arg_len));
        memcpy(WS_Reservation(wrk2->aws), arg, arg_len);
        wrk2->task->func = func;
        wrk2->task->priv = WS_Reservation(wrk2->aws);
        Lck_Unlock(&pp->mtx);
        // see signaling_note at the top for explanation
        if (retval)
                PTOK(pthread_cond_signal(&wrk2->cond));
        return (retval);
}

/*--------------------------------------------------------------------
 * Enter a new task to be done
 */

int
Pool_Task(struct pool *pp, struct pool_task *task, enum task_prio prio)
{
        struct worker *wrk;
        int retval = 0;
        CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
        AN(task);
        AN(task->func);
        assert(prio < TASK_QUEUE__END);

        if (prio == TASK_QUEUE_REQ && reqpoolfail) {
                retval = reqpoolfail & 1;
                reqpoolfail >>= 1;
                if (retval) {
                        VSL(SLT_Debug, NO_VXID,
                            "Failing due to reqpoolfail (next= 0x%jx)",
                            reqpoolfail);
                        return (retval);
                }
        }

        Lck_Lock(&pp->mtx);

        /* The common case first:  Take an idle thread, do it. */

        wrk = pool_getidleworker(pp, prio);
        if (wrk != NULL) {
                AZ(wrk->task->func);
                wrk->task->func = task->func;
                wrk->task->priv = task->priv;
                Lck_Unlock(&pp->mtx);
                // see signaling_note at the top for explanation
                PTOK(pthread_cond_signal(&wrk->cond));
                return (0);
        }

        /* Vital work is always queued. Only priority classes that can
         * fit under the reserve capacity are eligible to queuing.
         */
        if (prio >= TASK_QUEUE_RESERVE) {
                retval = -1;
        } else if (!TASK_QUEUE_LIMITED(prio) ||
            pp->lqueue + pp->nthr < cache_param->wthread_max +
            cache_param->wthread_queue_limit) {
                pp->stats->sess_queued++;
                pp->lqueue++;
                VTAILQ_INSERT_TAIL(&pp->queues[prio], task, list);
                PTOK(pthread_cond_signal(&pp->herder_cond));
        } else {
                /* NB: This is counter-intuitive but when we drop a REQ
                 * task, it is an HTTP/1 request and we effectively drop
                 * the whole session. It is otherwise an h2 stream with
                 * STR priority in which case we are dropping a request.
                 */
                if (prio == TASK_QUEUE_REQ)
                        pp->stats->sess_dropped++;
                else
                        pp->stats->req_dropped++;
                retval = -1;
        }
        Lck_Unlock(&pp->mtx);
        return (retval);
}

/*--------------------------------------------------------------------
 * Empty function used as a pointer value for the thread exit condition.
 */

static void v_matchproto_(task_func_t)
pool_kiss_of_death(struct worker *wrk, void *priv)
{
        (void)wrk;
        (void)priv;
}


/*--------------------------------------------------------------------
 * This is the work function for worker threads in the pool.
 */

static void
Pool_Work_Thread(struct pool *pp, struct worker *wrk)
{
        struct pool_task *tp;
        struct pool_task tpx, tps;
        vtim_real tmo, now;
        unsigned i, reserve;

        CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
        wrk->pool = pp;
        while (1) {
                CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
                tp = NULL;

                WS_Rollback(wrk->aws, 0);
                AZ(wrk->vsl);

                Lck_Lock(&pp->mtx);
                reserve = pool_reserve();

                for (i = 0; i < TASK_QUEUE_RESERVE; i++) {
                        if (pp->nidle < (reserve * i / TASK_QUEUE_RESERVE))
                                break;
                        tp = VTAILQ_FIRST(&pp->queues[i]);
                        if (tp != NULL) {
                                pp->lqueue--;
                                pp->ndequeued--;
                                VTAILQ_REMOVE(&pp->queues[i], tp, list);
                                break;
                        }
                }

                if (wrk_addstat(wrk, tp, 1)) {
                        wrk->stats->summs++;
                        AN(tp);
                } else if (pp->b_stat != NULL && pp->a_stat->summs) {
                        /* Nothing to do, push pool stats into global pool */
                        tps.func = pool_stat_summ;
                        tps.priv = pp->a_stat;
                        pp->a_stat = pp->b_stat;
                        pp->b_stat = NULL;
                        tp = &tps;
                } else {
                        /* Nothing to do: To sleep, perchance to dream ... */
                        if (isnan(wrk->lastused))
                                wrk->lastused = VTIM_real();
                        wrk->task->func = NULL;
                        wrk->task->priv = wrk;
                        VTAILQ_INSERT_HEAD(&pp->idle_queue, wrk->task, list);
                        pp->nidle++;
                        now = wrk->lastused;
                        do {
                                // see signaling_note at the top for explanation
                                if (DO_DEBUG(DBG_VCLREL) &&
                                    pp->b_stat == NULL && pp->a_stat->summs)
                                        /* We've released the VCL, but
                                         * there are pool stats not pushed
                                         * to the global stats and some
                                         * thread is busy pushing
                                         * stats. Set a 1 second timeout
                                         * so that we'll wake up and get a
                                         * chance to push stats. */
                                        tmo = now + 1.;
                                else if (wrk->wpriv->vcl == NULL)
                                        tmo = INFINITY;
                                else if (DO_DEBUG(DBG_VTC_MODE))
                                        tmo = now + 1.;
                                else
                                        tmo = now + 60.;
                                (void)Lck_CondWaitUntil(
                                    &wrk->cond, &pp->mtx, tmo);
                                if (wrk->task->func != NULL) {
                                        /* We have been handed a new task */
                                        tpx = *wrk->task;
                                        tp = &tpx;
                                        wrk->stats->summs++;
                                } else if (pp->b_stat != NULL &&
                                    pp->a_stat->summs) {
                                        /* Woken up to release the VCL,
                                         * and noticing that there are
                                         * pool stats not pushed to the
                                         * global stats and no active
                                         * thread currently doing
                                         * it. Remove ourself from the
                                         * idle queue and take on the
                                         * task. */
                                        assert(pp->nidle > 0);
                                        VTAILQ_REMOVE(&pp->idle_queue,
                                            wrk->task, list);
                                        pp->nidle--;
                                        tps.func = pool_stat_summ;
                                        tps.priv = pp->a_stat;
                                        pp->a_stat = pp->b_stat;
                                        pp->b_stat = NULL;
                                        tp = &tps;
                                } else {
                                        // Presumably ETIMEDOUT but we do not
                                        // assert this because pthread condvars
                                        // are not airtight.
                                        if (wrk->wpriv->vcl)
                                                VCL_Rel(&wrk->wpriv->vcl);
                                        now = VTIM_real();
                                }
                        } while (tp == NULL);
                }
                Lck_Unlock(&pp->mtx);

                if (tp->func == pool_kiss_of_death)
                        break;

                do {
                        memset(wrk->task, 0, sizeof wrk->task);
                        assert(wrk->pool == pp);
                        AN(tp->func);
                        tp->func(wrk, tp->priv);
                        if (DO_DEBUG(DBG_VCLREL) && wrk->wpriv->vcl != NULL)
                                VCL_Rel(&wrk->wpriv->vcl);
                        tpx = *wrk->task;
                        tp = &tpx;
                } while (tp->func != NULL);

                if (WS_Overflowed(wrk->aws))
                        wrk->stats->ws_thread_overflow++;
                /* cleanup for next task */
                wrk->seen_methods = 0;
        }
        wrk->pool = NULL;
}

/*--------------------------------------------------------------------
 * Create another worker thread.
 */

struct pool_info {
        unsigned                magic;
#define POOL_INFO_MAGIC         0x4e4442d3
        size_t                  stacksize;
        struct pool             *qp;
};

static void *
pool_thread(void *priv)
{
        struct pool_info *pi;

        CAST_OBJ_NOTNULL(pi, priv, POOL_INFO_MAGIC);
        THR_Init();
        WRK_Thread(pi->qp, pi->stacksize, cache_param->workspace_thread);
        FREE_OBJ(pi);
        return (NULL);
}

static void
pool_breed(struct pool *qp)
{
        pthread_t tp;
        pthread_attr_t tp_attr;
        struct pool_info *pi;

        PTOK(pthread_attr_init(&tp_attr));
        PTOK(pthread_attr_setdetachstate(&tp_attr, PTHREAD_CREATE_DETACHED));

        /* Set the stacksize for worker threads we create */
        if (cache_param->wthread_stacksize != UINT_MAX)
                PTOK(pthread_attr_setstacksize(&tp_attr, cache_param->wthread_stacksize));

        ALLOC_OBJ(pi, POOL_INFO_MAGIC);
        AN(pi);
        PTOK(pthread_attr_getstacksize(&tp_attr, &pi->stacksize));
        pi->qp = qp;

        errno = pthread_create(&tp, &tp_attr, pool_thread, pi);
        if (errno) {
                FREE_OBJ(pi);
                VSL(SLT_Debug, NO_VXID, "Create worker thread failed %d %s",
                    errno, VAS_errtxt(errno));
                Lck_Lock(&pool_mtx);
                VSC_C_main->threads_failed++;
                Lck_Unlock(&pool_mtx);
                VTIM_sleep(cache_param->wthread_fail_delay);
        } else {
                qp->nthr++;
                Lck_Lock(&pool_mtx);
                VSC_C_main->threads++;
                VSC_C_main->threads_created++;
                Lck_Unlock(&pool_mtx);
                if (cache_param->wthread_add_delay > 0.0)
                        VTIM_sleep(cache_param->wthread_add_delay);
                else
                        (void)sched_yield();
        }

        PTOK(pthread_attr_destroy(&tp_attr));
}

/*--------------------------------------------------------------------
 * Herd a single pool
 *
 * This thread wakes up every thread_pool_timeout seconds, whenever a pool
 * queues and when threads need to be destroyed
 *
 * The trick here is to not be too aggressive about creating threads.  In
 * pool_breed(), we sleep whenever we create a thread and a little while longer
 * whenever we fail to, hopefully missing a lot of cond_signals in the meantime.
 *
 * Idle threads are destroyed at a rate determined by wthread_destroy_delay
 *
 * XXX: probably need a lot more work.
 *
 */

void*
pool_herder(void *priv)
{
        struct pool *pp;
        struct pool_task *pt;
        double t_idle;
        struct worker *wrk;
        double delay;
        unsigned wthread_min;
        uintmax_t dq = (1ULL << 31);
        vtim_mono dqt = 0;
        int r = 0;

        CAST_OBJ_NOTNULL(pp, priv, POOL_MAGIC);

        THR_SetName("pool_herder");
        THR_Init();

        while (!pp->die || pp->nthr > 0) {
                /*
                 * If the worker pool is configured too small, we can
                 * end up deadlocking it (see #2418 for details).
                 *
                 * Recovering from this would require a lot of complicated
                 * code, and fundamentally, either people configured their
                 * pools wrong, in which case we want them to notice, or
                 * they are under DoS, in which case recovering gracefully
                 * is unlikely be a major improvement.
                 *
                 * Instead we implement a watchdog and kill the worker if
                 * nothing has been dequeued for that long.
                 */
                if (VTAILQ_EMPTY(&pp->queues[TASK_QUEUE_HIGHEST_PRIORITY])) {
                        /* Watchdog only applies to no movement on the
                         * highest priority queue (TASK_QUEUE_BO) */
                        dq = pp->ndequeued + 1;
                } else if (dq != pp->ndequeued) {
                        dq = pp->ndequeued;
                        dqt = VTIM_mono();
                } else if (VTIM_mono() - dqt > cache_param->wthread_watchdog) {
                        VSL(SLT_Error, NO_VXID,
                            "Pool Herder: Queue does not move ql=%u dt=%f",
                            pp->lqueue, VTIM_mono() - dqt);
                        WRONG("Worker Pool Queue does not move"
                              " - see thread_pool_watchdog parameter");
                }
                wthread_min = cache_param->wthread_min;
                if (pp->die)
                        wthread_min = 0;

                /* Make more threads if needed and allowed */
                if (pp->nthr < wthread_min ||
                    (pp->lqueue > 0 && pp->nthr < cache_param->wthread_max)) {
                        pool_breed(pp);
                        continue;
                }

                delay = cache_param->wthread_timeout;
                assert(pp->nthr >= wthread_min);

                if (pp->nthr > wthread_min) {

                        t_idle = VTIM_real() - cache_param->wthread_timeout;

                        Lck_Lock(&pp->mtx);
                        wrk = NULL;
                        pt = VTAILQ_LAST(&pp->idle_queue, taskhead);
                        if (pt != NULL) {
                                AN(pp->nidle);
                                AZ(pt->func);
                                CAST_OBJ_NOTNULL(wrk, pt->priv, WORKER_MAGIC);

                                if (pp->die || wrk->lastused < t_idle ||
                                    pp->nthr > cache_param->wthread_max) {
                                        /* Give it a kiss on the cheek... */
                                        VTAILQ_REMOVE(&pp->idle_queue,
                                            wrk->task, list);
                                        pp->nidle--;
                                        wrk->task->func = pool_kiss_of_death;
                                        PTOK(pthread_cond_signal(&wrk->cond));
                                } else {
                                        delay = wrk->lastused - t_idle;
                                        wrk = NULL;
                                }
                        }
                        Lck_Unlock(&pp->mtx);

                        if (wrk != NULL) {
                                pp->nthr--;
                                Lck_Lock(&pool_mtx);
                                VSC_C_main->threads--;
                                VSC_C_main->threads_destroyed++;
                                Lck_Unlock(&pool_mtx);
                                delay = cache_param->wthread_destroy_delay;
                        } else
                                delay = vmax(delay,
                                    cache_param->wthread_destroy_delay);
                }

                if (pp->die) {
                        if (delay < 2)
                                delay = .01;
                        else
                                delay = 1;
                        VTIM_sleep(delay);
                        continue;
                }
                Lck_Lock(&pp->mtx);
                if (pp->lqueue == 0) {
                        if (DO_DEBUG(DBG_VTC_MODE))
                                delay = 0.5;
                        r = Lck_CondWaitTimeout(
                            &pp->herder_cond, &pp->mtx, delay);
                } else if (pp->nthr >= cache_param->wthread_max) {
                        /* XXX: unsafe counters */
                        if (r != ETIMEDOUT)
                                VSC_C_main->threads_limited++;
                        r = Lck_CondWaitTimeout(
                            &pp->herder_cond, &pp->mtx, 1.0);
                }
                Lck_Unlock(&pp->mtx);
        }
        return (NULL);
}

/*--------------------------------------------------------------------
 * Debugging aids
 */

static void v_matchproto_(cli_func_t)
debug_reqpoolfail(struct cli *cli, const char * const *av, void *priv)
{
        uintmax_t u = 1;
        const char *p;

        (void)priv;
        (void)cli;
        reqpoolfail = 0;
        for (p = av[2]; *p != '\0'; p++) {
                if (*p == 'F' || *p == 'f')
                        reqpoolfail |= u;
                u <<= 1;
        }
}

static struct cli_proto debug_cmds[] = {
        { CLICMD_DEBUG_REQPOOLFAIL,             "d", debug_reqpoolfail },
        { NULL }
};

void
WRK_Log(enum VSL_tag_e tag, const char *fmt, ...)
{
        struct worker *wrk;
        va_list ap;

        AN(fmt);

        wrk = THR_GetWorker();
        CHECK_OBJ_ORNULL(wrk, WORKER_MAGIC);

        va_start(ap, fmt);
        if (wrk != NULL && wrk->vsl != NULL)
                VSLbv(wrk->vsl, tag, fmt, ap);
        else
                VSLv(tag, NO_VXID, fmt, ap);
        va_end(ap);
}

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

void
WRK_Init(void)
{
        assert(cache_param->wthread_min >= TASK_QUEUE_RESERVE);
        CLI_AddFuncs(debug_cmds);
}

		vinyl-cache/bin/vinyld/cache/cache_wrk.c
0		/*-
1		* Copyright (c) 2006 Verdens Gang AS
2		* Copyright (c) 2006-2011 Varnish Software AS
3		* All rights reserved.
4		*
5		* Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
6		*
7		* SPDX-License-Identifier: BSD-2-Clause
8		*
9		* Redistribution and use in source and binary forms, with or without
10		* modification, are permitted provided that the following conditions
11		* are met:
12		* 1. Redistributions of source code must retain the above copyright
13		* notice, this list of conditions and the following disclaimer.
14		* 2. Redistributions in binary form must reproduce the above copyright
15		* notice, this list of conditions and the following disclaimer in the
16		* documentation and/or other materials provided with the distribution.
17		*
18		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
19		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
22		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
24		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
27		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
28		* SUCH DAMAGE.
29		*
30		* Worker thread stuff unrelated to the worker thread pools.
31		*
32		* --
33		* signaling_note:
34		*
35		* note on worker wakeup signaling through the wrk condition variable (cv)
36		*
37		* In the general case, a cv needs to be signaled while holding the
38		* corresponding mutex, otherwise the signal may be posted before the waiting
39		* thread could register itself on the cv and, consequently, the signal may be
40		* missed.
41		*
42		* In our case, any worker thread which we wake up comes from the idle queue,
43		* where it put itself under the mutex, releasing that mutex implicitly via
44		* Lck_CondWaitUntil() (which calls some variant of pthread_cond_wait). So we avoid
45		* additional mutex contention knowing that any worker thread on the idle queue
46		* is blocking on the cv.
47		*
48		* Except -- when it isn't, because it woke up for releasing its VCL
49		* Reference. To account for this case, we check if the task function has been
50		* set in the meantime, which in turn requires all of the task preparation to be
51		* done holding the pool mutex. (see also #2719)
52		*/
53
54		#include "config.h"
55
56		#include <stdlib.h>
57		#include <sched.h>
58
59		#include "cache_int.h"
60		#include "cache_pool.h"
61
62		#include "vcli_serve.h"
63		#include "vtim.h"
64
65		#include "hash/hash_slinger.h"
66
67		static void Pool_Work_Thread(struct pool pp, struct worker wrk);
68
69		static uintmax_t reqpoolfail;
70
71		/*--------------------------------------------------------------------
72		* Return all cached resources
73		*/
74
75		void
76	39204	WRK_Cleanup(const struct worker *wrk)
77		{
78
79	39204	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
80	39204	CHECK_OBJ_NOTNULL(wrk->wpriv, WORKER_PRIV_MAGIC);
81	39204	HSH_Cleanup(wrk);
82	39204	if (wrk->wpriv->vcl != NULL)
83	2482	VCL_Rel(&wrk->wpriv->vcl);
84	39204	}
85
86		/*--------------------------------------------------------------------
87		* Create and start a back-ground thread which as its own worker and
88		* session data structures;
89		*/
90
91		struct bgthread {
92		unsigned magic;
93		#define BGTHREAD_MAGIC 0x23b5152b
94		const char *name;
95		bgthread_t *func;
96		void *priv;
97		};
98
99		static void *
100	15704	wrk_bgthread(void *arg)
101		{
102		struct bgthread *bt;
103		struct worker wrk;
104		struct worker_priv wpriv[1];
105		struct VSC_main_wrk ds;
106		void *r;
107
108	15704	CAST_OBJ_NOTNULL(bt, arg, BGTHREAD_MAGIC);
109	15704	THR_SetName(bt->name);
110	15704	THR_Init();
111	15704	INIT_OBJ(&wrk, WORKER_MAGIC);
112	15704	INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
113	15704	wrk.wpriv = wpriv;
114		// bgthreads do not have a vpi member
115	15704	memset(&ds, 0, sizeof ds);
116	15704	wrk.stats = &ds;
117
118	15704	r = bt->func(&wrk, bt->priv);
119	15704	WRK_Cleanup(&wrk);
120	15704	Pool_Sumstat(&wrk);
121	15704	return (r);
122		}
123
124		void
125	15704	WRK_BgThread(pthread_t thr, const char name, bgthread_t func, void priv)
126		{
127		struct bgthread *bt;
128
129	15704	ALLOC_OBJ(bt, BGTHREAD_MAGIC);
130	15704	AN(bt);
131
132	15704	bt->name = name;
133	15704	bt->func = func;
134	15704	bt->priv = priv;
135	15704	PTOK(pthread_create(thr, NULL, wrk_bgthread, bt));
136	15704	}
137
138		/--------------------------------------------------------------------/
139
140		static void
141	27604	WRK_Thread(struct pool *qp, size_t stacksize, unsigned thread_workspace)
142		{
143		// child_signal_handler stack overflow check uses struct worker addr
144		struct worker *w, ww;
145		struct VSC_main_wrk ds;
146	27604	unsigned char ws[thread_workspace];
147		struct worker_priv wpriv[1];
148	27604	unsigned char vpi[vpi_wrk_len];
149
150	27604	AN(qp);
151	27604	AN(stacksize);
152	27604	AN(thread_workspace);
153
154	27604	THR_SetName("cache-worker");
155	27604	w = &ww;
156	27604	INIT_OBJ(w, WORKER_MAGIC);
157	27604	INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
158	27604	w->wpriv = wpriv;
159	27604	w->lastused = NAN;
160	27604	memset(&ds, 0, sizeof ds);
161	27604	w->stats = &ds;
162	27604	THR_SetWorker(w);
163	27604	PTOK(pthread_cond_init(&w->cond, NULL));
164
165	27604	WS_Init(w->aws, "wrk", ws, thread_workspace);
166	27604	VPI_wrk_init(w, vpi, sizeof vpi);
167	27604	AN(w->vpi);
168
169	27604	VSL(SLT_WorkThread, NO_VXID, "%p start", w);
170
171	27604	Pool_Work_Thread(qp, w);
172	27604	AZ(w->pool);
173
174	27604	VSL(SLT_WorkThread, NO_VXID, "%p end", w);
175	27604	PTOK(pthread_cond_destroy(&w->cond));
176	27604	WRK_Cleanup(w);
177	27604	Pool_Sumstat(w);
178	27604	}
179
180		/*--------------------------------------------------------------------
181		* Summing of stats into pool counters
182		*/
183
184		static unsigned
185	121224	wrk_addstat(const struct worker wrk, const struct pool_task tp, unsigned locked)
186		{
187		struct pool *pp;
188
189	121224	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
190	121224	pp = wrk->pool;
191	121224	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
192	121224	if (locked)
193	121227	Lck_AssertHeld(&pp->mtx);
194
195	121230	if ((tp == NULL && wrk->stats->summs > 0) \|\|
196	29	(wrk->stats->summs >= cache_param->wthread_stats_rate)) {
197	220932	if (!locked)
198	0	Lck_Lock(&pp->mtx);
199
200	21470	pp->a_stat->summs++;
201	21470	VSC_main_Summ_wrk_wrk(pp->a_stat, wrk->stats);
202	21470	memset(wrk->stats, 0, sizeof *wrk->stats);
203
204	21470	if (!locked)
205	0	Lck_Unlock(&pp->mtx);
206	21470	}
207
208	121172	return (tp != NULL);
209		}
210
211		void
212	0	WRK_AddStat(const struct worker *wrk)
213		{
214
215	0	(void)wrk_addstat(wrk, wrk->task, 0);
216	0	wrk->stats->summs++;
217	0	}
218
219		/*--------------------------------------------------------------------
220		* Pool reserve calculation
221		*/
222
223		static unsigned
224	150503	pool_reserve(void)
225		{
226		unsigned lim;
227
228	150503	if (cache_param->wthread_reserve == 0) {
229	150323	lim = cache_param->wthread_min / 20 + 1;
230	150323	} else {
231	180	lim = cache_param->wthread_min * 950 / 1000;
232	180	if (cache_param->wthread_reserve < lim)
233	0	lim = cache_param->wthread_reserve;
234		}
235	150503	if (lim < TASK_QUEUE_RESERVE)
236	149507	return (TASK_QUEUE_RESERVE);
237	996	return (lim);
238	150503	}
239
240		/--------------------------------------------------------------------/
241
242		static struct worker *
243	29287	pool_getidleworker(struct pool *pp, enum task_prio prio)
244		{
245	29287	struct pool_task *pt = NULL;
246		struct worker *wrk;
247
248	29287	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
249	29287	Lck_AssertHeld(&pp->mtx);
250	29287	if (pp->nidle > (pool_reserve() * prio / TASK_QUEUE_RESERVE)) {
251	29237	pt = VTAILQ_FIRST(&pp->idle_queue);
252	29237	if (pt == NULL)
253	0	AZ(pp->nidle);
254	29237	}
255
256	29287	if (pt == NULL)
257	51	return (NULL);
258
259	29236	AZ(pt->func);
260	29236	CAST_OBJ_NOTNULL(wrk, pt->priv, WORKER_MAGIC);
261
262	29236	AN(pp->nidle);
263	29236	VTAILQ_REMOVE(&pp->idle_queue, wrk->task, list);
264	29236	pp->nidle--;
265
266	29236	return (wrk);
267	29287	}
268
269		/*--------------------------------------------------------------------
270		* Special scheduling: If no thread can be found, the current thread
271		* will be prepared for rescheduling instead.
272		* The selected threads workspace is reserved and the argument put there.
273		* Return one if another thread was scheduled, otherwise zero.
274		*/
275
276		int
277	9131	Pool_Task_Arg(struct worker wrk, enum task_prio prio, task_func_t func,
278		const void *arg, size_t arg_len)
279		{
280		struct pool *pp;
281		struct worker *wrk2;
282		int retval;
283
284	9131	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
285	9131	AN(arg);
286	9131	AN(arg_len);
287	9131	pp = wrk->pool;
288	9131	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
289
290	9131	Lck_Lock(&pp->mtx);
291	9131	wrk2 = pool_getidleworker(pp, prio);
292	9131	if (wrk2 != NULL)
293	9114	retval = 1;
294		else {
295	17	wrk2 = wrk;
296	17	retval = 0;
297		}
298	9131	AZ(wrk2->task->func);
299	9131	assert(arg_len <= WS_ReserveSize(wrk2->aws, arg_len));
300	9131	memcpy(WS_Reservation(wrk2->aws), arg, arg_len);
301	9131	wrk2->task->func = func;
302	9131	wrk2->task->priv = WS_Reservation(wrk2->aws);
303	9131	Lck_Unlock(&pp->mtx);
304		// see signaling_note at the top for explanation
305	9131	if (retval)
306	9115	PTOK(pthread_cond_signal(&wrk2->cond));
307	9131	return (retval);
308		}
309
310		/*--------------------------------------------------------------------
311		* Enter a new task to be done
312		*/
313
314		int
315	20160	Pool_Task(struct pool pp, struct pool_task task, enum task_prio prio)
316		{
317		struct worker *wrk;
318	20160	int retval = 0;
319	20160	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
320	20160	AN(task);
321	20160	AN(task->func);
322	20160	assert(prio < TASK_QUEUE__END);
323
324	20160	if (prio == TASK_QUEUE_REQ && reqpoolfail) {
325	4	retval = reqpoolfail & 1;
326	4	reqpoolfail >>= 1;
327	4	if (retval) {
328	8	VSL(SLT_Debug, NO_VXID,
329		"Failing due to reqpoolfail (next= 0x%jx)",
330	4	reqpoolfail);
331	4	return (retval);
332		}
333	0	}
334
335	20156	Lck_Lock(&pp->mtx);
336
337		/* The common case first: Take an idle thread, do it. */
338
339	20156	wrk = pool_getidleworker(pp, prio);
340	20156	if (wrk != NULL) {
341	20122	AZ(wrk->task->func);
342	20122	wrk->task->func = task->func;
343	20122	wrk->task->priv = task->priv;
344	20122	Lck_Unlock(&pp->mtx);
345		// see signaling_note at the top for explanation
346	20122	PTOK(pthread_cond_signal(&wrk->cond));
347	20122	return (0);
348		}
349
350		/* Vital work is always queued. Only priority classes that can
351		* fit under the reserve capacity are eligible to queuing.
352		*/
353	34	if (prio >= TASK_QUEUE_RESERVE) {
354	4	retval = -1;
355	34	} else if (!TASK_QUEUE_LIMITED(prio) \|\|
356	60	pp->lqueue + pp->nthr < cache_param->wthread_max +
357	30	cache_param->wthread_queue_limit) {
358	26	pp->stats->sess_queued++;
359	26	pp->lqueue++;
360	26	VTAILQ_INSERT_TAIL(&pp->queues[prio], task, list);
361	26	PTOK(pthread_cond_signal(&pp->herder_cond));
362	26	} else {
363		/* NB: This is counter-intuitive but when we drop a REQ
364		* task, it is an HTTP/1 request and we effectively drop
365		* the whole session. It is otherwise an h2 stream with
366		* STR priority in which case we are dropping a request.
367		*/
368	4	if (prio == TASK_QUEUE_REQ)
369	0	pp->stats->sess_dropped++;
370		else
371	4	pp->stats->req_dropped++;
372	4	retval = -1;
373		}
374	34	Lck_Unlock(&pp->mtx);
375	34	return (retval);
376	20160	}
377
378		/*--------------------------------------------------------------------
379		* Empty function used as a pointer value for the thread exit condition.
380		*/
381
382		static void v_matchproto_(task_func_t)
383	0	pool_kiss_of_death(struct worker wrk, void priv)
384		{
385	0	(void)wrk;
386	0	(void)priv;
387	0	}
388
389
390		/*--------------------------------------------------------------------
391		* This is the work function for worker threads in the pool.
392		*/
393
394		static void
395	89244	Pool_Work_Thread(struct pool pp, struct worker wrk)
396		{
397		struct pool_task *tp;
398		struct pool_task tpx, tps;
399		vtim_real tmo, now;
400		unsigned i, reserve;
401
402	89244	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
403	89244	wrk->pool = pp;
404	121228	while (1) {
405	121228	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
406	121228	tp = NULL;
407
408	121228	WS_Rollback(wrk->aws, 0);
409	121228	AZ(wrk->vsl);
410
411	121228	Lck_Lock(&pp->mtx);
412	121228	reserve = pool_reserve();
413
414	649087	for (i = 0; i < TASK_QUEUE_RESERVE; i++) {
415	559557	if (pp->nidle < (reserve * i / TASK_QUEUE_RESERVE))
416	31672	break;
417	527885	tp = VTAILQ_FIRST(&pp->queues[i]);
418	527885	if (tp != NULL) {
419	26	pp->lqueue--;
420	26	pp->ndequeued--;
421	26	VTAILQ_REMOVE(&pp->queues[i], tp, list);
422	26	break;
423		}
424	527859	}
425
426	121228	if (wrk_addstat(wrk, tp, 1)) {
427	26	wrk->stats->summs++;
428	26	AN(tp);
429	121228	} else if (pp->b_stat != NULL && pp->a_stat->summs) {
430		/* Nothing to do, push pool stats into global pool */
431	21463	tps.func = pool_stat_summ;
432	21463	tps.priv = pp->a_stat;
433	21463	pp->a_stat = pp->b_stat;
434	21463	pp->b_stat = NULL;
435	21463	tp = &tps;
436	21463	} else {
437		/* Nothing to do: To sleep, perchance to dream ... */
438	99739	if (isnan(wrk->lastused))
439	78270	wrk->lastused = VTIM_real();
440	99739	wrk->task->func = NULL;
441	99739	wrk->task->priv = wrk;
442	99739	VTAILQ_INSERT_HEAD(&pp->idle_queue, wrk->task, list);
443	99739	pp->nidle++;
444	99739	now = wrk->lastused;
445	99739	do {
446		// see signaling_note at the top for explanation
447	102445	if (DO_DEBUG(DBG_VCLREL) &&
448	2048	pp->b_stat == NULL && pp->a_stat->summs)
449		/* We've released the VCL, but
450		* there are pool stats not pushed
451		* to the global stats and some
452		* thread is busy pushing
453		* stats. Set a 1 second timeout
454		* so that we'll wake up and get a
455		* chance to push stats. */
456	0	tmo = now + 1.;
457	102445	else if (wrk->wpriv->vcl == NULL)
458	93338	tmo = INFINITY;
459	9107	else if (DO_DEBUG(DBG_VTC_MODE))
460	9107	tmo = now + 1.;
461		else
462	0	tmo = now + 60.;
463	102445	(void)Lck_CondWaitUntil(
464	102445	&wrk->cond, &pp->mtx, tmo);
465	102445	if (wrk->task->func != NULL) {
466		/* We have been handed a new task */
467	99739	tpx = *wrk->task;
468	99739	tp = &tpx;
469	99739	wrk->stats->summs++;
470	102445	} else if (pp->b_stat != NULL &&
471	2706	pp->a_stat->summs) {
472		/* Woken up to release the VCL,
473		* and noticing that there are
474		* pool stats not pushed to the
475		* global stats and no active
476		* thread currently doing
477		* it. Remove ourself from the
478		* idle queue and take on the
479		* task. */
480	0	assert(pp->nidle > 0);
481	0	VTAILQ_REMOVE(&pp->idle_queue,
482		wrk->task, list);
483	0	pp->nidle--;
484	0	tps.func = pool_stat_summ;
485	0	tps.priv = pp->a_stat;
486	0	pp->a_stat = pp->b_stat;
487	0	pp->b_stat = NULL;
488	0	tp = &tps;
489	0	} else {
490		// Presumably ETIMEDOUT but we do not
491		// assert this because pthread condvars
492		// are not airtight.
493	2706	if (wrk->wpriv->vcl)
494	2706	VCL_Rel(&wrk->wpriv->vcl);
495	2706	now = VTIM_real();
496		}
497	102445	} while (tp == NULL);
498		}
499	121228	Lck_Unlock(&pp->mtx);