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_varnishd.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;

/*--------------------------------------------------------------------
 * 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);
        HSH_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);
        if (w->wpriv->vcl != NULL)
                VCL_Rel(&w->wpriv->vcl);
        PTOK(pthread_cond_destroy(&w->cond));
        HSH_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);
}

		varnish-cache/bin/varnishd/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_varnishd.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		* Create and start a back-ground thread which as its own worker and
73		* session data structures;
74		*/
75
76		struct bgthread {
77		unsigned magic;
78		#define BGTHREAD_MAGIC 0x23b5152b
79		const char *name;
80		bgthread_t *func;
81		void *priv;
82		};
83
84		static void *
85	26946	wrk_bgthread(void *arg)
86		{
87		struct bgthread *bt;
88		struct worker wrk;
89		struct worker_priv wpriv[1];
90		struct VSC_main_wrk ds;
91		void *r;
92
93	26946	CAST_OBJ_NOTNULL(bt, arg, BGTHREAD_MAGIC);
94	26946	THR_SetName(bt->name);
95	26946	THR_Init();
96	26946	INIT_OBJ(&wrk, WORKER_MAGIC);
97	26946	INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
98	26946	wrk.wpriv = wpriv;
99		// bgthreads do not have a vpi member
100	26946	memset(&ds, 0, sizeof ds);
101	26946	wrk.stats = &ds;
102
103	26946	r = bt->func(&wrk, bt->priv);
104	26946	HSH_Cleanup(&wrk);
105	26946	Pool_Sumstat(&wrk);
106	26946	return (r);
107		}
108
109		void
110	26946	WRK_BgThread(pthread_t thr, const char name, bgthread_t func, void priv)
111		{
112		struct bgthread *bt;
113
114	26946	ALLOC_OBJ(bt, BGTHREAD_MAGIC);
115	26946	AN(bt);
116
117	26946	bt->name = name;
118	26946	bt->func = func;
119	26946	bt->priv = priv;
120	26946	PTOK(pthread_create(thr, NULL, wrk_bgthread, bt));
121	26946	}
122
123		/--------------------------------------------------------------------/
124
125		static void
126	167	WRK_Thread(struct pool *qp, size_t stacksize, unsigned thread_workspace)
127		{
128		// child_signal_handler stack overflow check uses struct worker addr
129		struct worker *w, ww;
130		struct VSC_main_wrk ds;
131	167	unsigned char ws[thread_workspace];
132		struct worker_priv wpriv[1];
133	167	unsigned char vpi[vpi_wrk_len];
134
135	167	AN(qp);
136	167	AN(stacksize);
137	167	AN(thread_workspace);
138
139	167	THR_SetName("cache-worker");
140	167	w = &ww;
141	167	INIT_OBJ(w, WORKER_MAGIC);
142	167	INIT_OBJ(wpriv, WORKER_PRIV_MAGIC);
143	167	w->wpriv = wpriv;
144	167	w->lastused = NAN;
145	167	memset(&ds, 0, sizeof ds);
146	167	w->stats = &ds;
147	167	THR_SetWorker(w);
148	167	PTOK(pthread_cond_init(&w->cond, NULL));
149
150	167	WS_Init(w->aws, "wrk", ws, thread_workspace);
151	167	VPI_wrk_init(w, vpi, sizeof vpi);
152	167	AN(w->vpi);
153
154	167	VSL(SLT_WorkThread, NO_VXID, "%p start", w);
155
156	167	Pool_Work_Thread(qp, w);
157	167	AZ(w->pool);
158
159	167	VSL(SLT_WorkThread, NO_VXID, "%p end", w);
160	167	if (w->wpriv->vcl != NULL)
161	14	VCL_Rel(&w->wpriv->vcl);
162	167	PTOK(pthread_cond_destroy(&w->cond));
163	167	HSH_Cleanup(w);
164	167	Pool_Sumstat(w);
165	167	}
166
167		/*--------------------------------------------------------------------
168		* Summing of stats into pool counters
169		*/
170
171		static unsigned
172	206834	wrk_addstat(const struct worker wrk, const struct pool_task tp, unsigned locked)
173		{
174		struct pool *pp;
175
176	206834	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
177	206834	pp = wrk->pool;
178	206834	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
179	206834	if (locked)
180	206834	Lck_AssertHeld(&pp->mtx);
181
182	206836	if ((tp == NULL && wrk->stats->summs > 0) \|\|
183	53	(wrk->stats->summs >= cache_param->wthread_stats_rate)) {
184	377861	if (!locked)
185	0	Lck_Lock(&pp->mtx);
186
187	35705	pp->a_stat->summs++;
188	35705	VSC_main_Summ_wrk_wrk(pp->a_stat, wrk->stats);
189	35705	memset(wrk->stats, 0, sizeof *wrk->stats);
190
191	35705	if (!locked)
192	0	Lck_Unlock(&pp->mtx);
193	35705	}
194
195	206730	return (tp != NULL);
196		}
197
198		void
199	0	WRK_AddStat(const struct worker *wrk)
200		{
201
202	0	(void)wrk_addstat(wrk, wrk->task, 0);
203	0	wrk->stats->summs++;
204	0	}
205
206		/*--------------------------------------------------------------------
207		* Pool reserve calculation
208		*/
209
210		static unsigned
211	255950	pool_reserve(void)
212		{
213		unsigned lim;
214
215	255950	if (cache_param->wthread_reserve == 0) {
216	255770	lim = cache_param->wthread_min / 20 + 1;
217	255770	} else {
218	180	lim = cache_param->wthread_min * 950 / 1000;
219	180	if (cache_param->wthread_reserve < lim)
220	90	lim = cache_param->wthread_reserve;
221		}
222	255950	if (lim < TASK_QUEUE_RESERVE)
223	253004	return (TASK_QUEUE_RESERVE);
224	2946	return (lim);
225	255950	}
226
227		/--------------------------------------------------------------------/
228
229		static struct worker *
230	49132	pool_getidleworker(struct pool *pp, enum task_prio prio)
231		{
232	49132	struct pool_task *pt = NULL;
233		struct worker *wrk;
234
235	49132	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
236	49132	Lck_AssertHeld(&pp->mtx);
237	49132	if (pp->nidle > (pool_reserve() * prio / TASK_QUEUE_RESERVE)) {
238	49046	pt = VTAILQ_FIRST(&pp->idle_queue);
239	49046	if (pt == NULL)
240	0	AZ(pp->nidle);
241	49046	}
242
243	49132	if (pt == NULL)
244	86	return (NULL);
245
246	49046	AZ(pt->func);
247	49046	CAST_OBJ_NOTNULL(wrk, pt->priv, WORKER_MAGIC);
248
249	49046	AN(pp->nidle);
250	49046	VTAILQ_REMOVE(&pp->idle_queue, wrk->task, list);
251	49046	pp->nidle--;
252
253	49046	return (wrk);
254	49132	}
255
256		/*--------------------------------------------------------------------
257		* Special scheduling: If no thread can be found, the current thread
258		* will be prepared for rescheduling instead.
259		* The selected threads workspace is reserved and the argument put there.
260		* Return one if another thread was scheduled, otherwise zero.
261		*/
262
263		int
264	15536	Pool_Task_Arg(struct worker wrk, enum task_prio prio, task_func_t func,
265		const void *arg, size_t arg_len)
266		{
267		struct pool *pp;
268		struct worker *wrk2;
269		int retval;
270
271	15536	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
272	15536	AN(arg);
273	15536	AN(arg_len);
274	15536	pp = wrk->pool;
275	15536	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
276
277	15536	Lck_Lock(&pp->mtx);
278	15536	wrk2 = pool_getidleworker(pp, prio);
279	15536	if (wrk2 != NULL)
280	15515	retval = 1;
281		else {
282	21	wrk2 = wrk;
283	21	retval = 0;
284		}
285	15536	AZ(wrk2->task->func);
286	15536	assert(arg_len <= WS_ReserveSize(wrk2->aws, arg_len));
287	15536	memcpy(WS_Reservation(wrk2->aws), arg, arg_len);
288	15536	wrk2->task->func = func;
289	15536	wrk2->task->priv = WS_Reservation(wrk2->aws);
290	15536	Lck_Unlock(&pp->mtx);
291		// see signaling_note at the top for explanation
292	15536	if (retval)
293	15517	PTOK(pthread_cond_signal(&wrk2->cond));
294	15536	return (retval);
295		}
296
297		/*--------------------------------------------------------------------
298		* Enter a new task to be done
299		*/
300
301		int
302	33601	Pool_Task(struct pool pp, struct pool_task task, enum task_prio prio)
303		{
304		struct worker *wrk;
305	33601	int retval = 0;
306	33601	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
307	33601	AN(task);
308	33601	AN(task->func);
309	33601	assert(prio < TASK_QUEUE__END);
310
311	33601	if (prio == TASK_QUEUE_REQ && reqpoolfail) {
312	7	retval = reqpoolfail & 1;
313	7	reqpoolfail >>= 1;
314	7	if (retval) {
315	14	VSL(SLT_Debug, NO_VXID,
316		"Failing due to reqpoolfail (next= 0x%jx)",
317	7	reqpoolfail);
318	7	return (retval);
319		}
320	0	}
321
322	33594	Lck_Lock(&pp->mtx);
323
324		/* The common case first: Take an idle thread, do it. */
325
326	33594	wrk = pool_getidleworker(pp, prio);
327	33594	if (wrk != NULL) {
328	33529	AZ(wrk->task->func);
329	33529	wrk->task->func = task->func;
330	33529	wrk->task->priv = task->priv;
331	33529	Lck_Unlock(&pp->mtx);
332		// see signaling_note at the top for explanation
333	33529	PTOK(pthread_cond_signal(&wrk->cond));
334	33529	return (0);
335		}
336
337		/* Vital work is always queued. Only priority classes that can
338		* fit under the reserve capacity are eligible to queuing.
339		*/
340	65	if (prio >= TASK_QUEUE_RESERVE) {
341	7	retval = -1;
342	65	} else if (!TASK_QUEUE_LIMITED(prio) \|\|
343	116	pp->lqueue + pp->nthr < cache_param->wthread_max +
344	58	cache_param->wthread_queue_limit) {
345	51	pp->stats->sess_queued++;
346	51	pp->lqueue++;
347	51	VTAILQ_INSERT_TAIL(&pp->queues[prio], task, list);
348	51	PTOK(pthread_cond_signal(&pp->herder_cond));
349	51	} else {
350		/* NB: This is counter-intuitive but when we drop a REQ
351		* task, it is an HTTP/1 request and we effectively drop
352		* the whole session. It is otherwise an h2 stream with
353		* STR priority in which case we are dropping a request.
354		*/
355	7	if (prio == TASK_QUEUE_REQ)
356	0	pp->stats->sess_dropped++;
357		else
358	7	pp->stats->req_dropped++;
359	7	retval = -1;
360		}
361	65	Lck_Unlock(&pp->mtx);
362	65	return (retval);
363	33601	}
364
365		/*--------------------------------------------------------------------
366		* Empty function used as a pointer value for the thread exit condition.
367		*/
368
369		static void v_matchproto_(task_func_t)
370	0	pool_kiss_of_death(struct worker wrk, void priv)
371		{
372	0	(void)wrk;
373	0	(void)priv;
374	0	}
375
376
377		/*--------------------------------------------------------------------
378		* This is the work function for worker threads in the pool.
379		*/
380
381		static void
382	202997	Pool_Work_Thread(struct pool pp, struct worker wrk)
383		{
384		struct pool_task *tp;
385		struct pool_task tpx, tps;
386		vtim_real tmo, now;
387		unsigned i, reserve;
388
389	202997	CHECK_OBJ_NOTNULL(pp, POOL_MAGIC);
390	202997	wrk->pool = pp;
391	206835	while (1) {
392	206835	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
393	206835	tp = NULL;
394
395	206835	WS_Rollback(wrk->aws, 0);
396	206835	AZ(wrk->vsl);
397
398	206835	Lck_Lock(&pp->mtx);
399	206835	reserve = pool_reserve();
400
401	1107197	for (i = 0; i < TASK_QUEUE_RESERVE; i++) {
402	954453	if (pp->nidle < (reserve * i / TASK_QUEUE_RESERVE))
403	54040	break;
404	900413	tp = VTAILQ_FIRST(&pp->queues[i]);
405	900413	if (tp != NULL) {
406	51	pp->lqueue--;
407	51	pp->ndequeued--;
408	51	VTAILQ_REMOVE(&pp->queues[i], tp, list);
409	51	break;
410		}
411	900362	}
412
413	206835	if (wrk_addstat(wrk, tp, 1)) {
414	51	wrk->stats->summs++;
415	51	AN(tp);
416	206835	} else if (pp->b_stat != NULL && pp->a_stat->summs) {
417		/* Nothing to do, push pool stats into global pool */
418	35701	tps.func = pool_stat_summ;
419	35701	tps.priv = pp->a_stat;
420	35701	pp->a_stat = pp->b_stat;
421	35701	pp->b_stat = NULL;
422	35701	tp = &tps;
423	35701	} else {
424		/* Nothing to do: To sleep, perchance to dream ... */
425	171083	if (isnan(wrk->lastused))
426	135376	wrk->lastused = VTIM_real();
427	171083	wrk->task->func = NULL;
428	171083	wrk->task->priv = wrk;
429	171083	VTAILQ_INSERT_HEAD(&pp->idle_queue, wrk->task, list);
430	171083	pp->nidle++;
431	171083	now = wrk->lastused;
432	171083	do {
433		// see signaling_note at the top for explanation
434	179908	if (DO_DEBUG(DBG_VCLREL) &&
435	7779	pp->b_stat == NULL && pp->a_stat->summs)
436		/* We've released the VCL, but
437		* there are pool stats not pushed
438		* to the global stats and some
439		* thread is busy pushing
440		* stats. Set a 1 second timeout
441		* so that we'll wake up and get a
442		* chance to push stats. */
443	0	tmo = now + 1.;
444	179908	else if (wrk->wpriv->vcl == NULL)
445	164466	tmo = INFINITY;
446	15442	else if (DO_DEBUG(DBG_VTC_MODE))
447	15442	tmo = now + 1.;
448		else
449	0	tmo = now + 60.;
450	179908	(void)Lck_CondWaitUntil(
451	179908	&wrk->cond, &pp->mtx, tmo);
452	179908	if (wrk->task->func != NULL) {
453		/* We have been handed a new task */
454	171082	tpx = *wrk->task;
455	171082	tp = &tpx;
456	171082	wrk->stats->summs++;
457	179908	} else if (pp->b_stat != NULL &&
458	8826	pp->a_stat->summs) {
459		/* Woken up to release the VCL,
460		* and noticing that there are
461		* pool stats not pushed to the
462		* global stats and no active
463		* thread currently doing
464		* it. Remove ourself from the
465		* idle queue and take on the
466		* task. */
467	0	assert(pp->nidle > 0);
468	0	VTAILQ_REMOVE(&pp->idle_queue,
469		wrk->task, list);
470	0	pp->nidle--;
471	0	tps.func = pool_stat_summ;
472	0	tps.priv = pp->a_stat;
473	0	pp->a_stat = pp->b_stat;
474	0	pp->b_stat = NULL;
475	0	tp = &tps;
476	0	} else {
477		// Presumably ETIMEDOUT but we do not
478		// assert this because pthread condvars
479		// are not airtight.
480	8826	if (wrk->wpriv->vcl)
481	8826	VCL_Rel(&wrk->wpriv->vcl);
482	8826	now = VTIM_real();
483		}
484	179908	} while (tp == NULL);
485		}
486	206835	Lck_Unlock(&pp->mtx);
487
488	206835	if (tp->func == pool_kiss_of_death)
489	167	break;
490
491	206668	do {
492	308083	memset(wrk->task, 0, sizeof wrk->task);
493	308083	assert(wrk->pool == pp);
494	105253	AN(tp->func);
495	105253	tp->func(wrk, tp->priv);
496	105253	if (DO_DEBUG(DBG_VCLREL) && wrk->wpriv->vcl != NULL)
497	160	VCL_Rel(&wrk->wpriv->vcl);
498	105253	tpx = *wrk->task;
499	105253	tp = &tpx;