GCOV - Varnish Cache


/*-
 * Copyright (c) 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.
 *
 * (TCP|UDS) connection pools.
 *
 */

#include "config.h"

#include <stdlib.h>

#include "cache_int.h"

#include "vsa.h"
#include "vsha256.h"
#include "vtcp.h"
#include "vus.h"
#include "vtim.h"
#include "waiter/waiter.h"

#include "cache_conn_pool.h"
#include "cache_pool.h"

#include "VSC_vcp.h"

struct conn_pool;
static inline int vcp_cmp(const struct conn_pool *a, const struct conn_pool *b);

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

struct pfd {
        unsigned                magic;
#define PFD_MAGIC               0x0c5e6593
        int                     fd;
        VTAILQ_ENTRY(pfd)       list;
        VCL_IP                  addr;
        uint8_t                 state;
        struct waited           waited[1];
        struct conn_pool        *conn_pool;

        pthread_cond_t          *cond;

        vtim_mono               created;
        uint64_t                reused;
};

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

typedef int cp_open_f(const struct conn_pool *, vtim_dur tmo, VCL_IP *ap);
typedef void cp_close_f(struct pfd *);
typedef void cp_name_f(const struct pfd *, char *, unsigned, char *, unsigned);

struct cp_methods {
        cp_open_f                               *open;
        cp_close_f                              *close;
        cp_name_f                               *local_name;
        cp_name_f                               *remote_name;
};

struct conn_pool {
        unsigned                                magic;
#define CONN_POOL_MAGIC                         0x85099bc3

        const struct cp_methods                 *methods;

        struct vrt_endpoint                     *endpoint;
        char                                    ident[VSHA256_DIGEST_LENGTH];

        VRBT_ENTRY(conn_pool)                   entry;
        int                                     refcnt;
        struct lock                             mtx;

        VTAILQ_HEAD(, pfd)                      connlist;
        int                                     n_conn;

        int                                     n_kill;

        int                                     n_used;

        vtim_mono                               holddown;
        int                                     holddown_errno;
};

static struct lock conn_pools_mtx;
static struct lock dead_pools_mtx;
static struct VSC_vcp *vsc;

VRBT_HEAD(vrb, conn_pool);
VRBT_GENERATE_REMOVE_COLOR(vrb, conn_pool, entry, static)
VRBT_GENERATE_REMOVE(vrb, conn_pool, entry, static)
VRBT_GENERATE_INSERT_COLOR(vrb, conn_pool, entry, static)
VRBT_GENERATE_INSERT_FINISH(vrb, conn_pool, entry, static)
VRBT_GENERATE_INSERT(vrb, conn_pool, entry, vcp_cmp, static)
VRBT_GENERATE_NEXT(vrb, conn_pool, entry, static)
VRBT_GENERATE_MINMAX(vrb, conn_pool, entry, static)

static struct vrb conn_pools = VRBT_INITIALIZER(&conn_pools);
static struct vrb dead_pools = VRBT_INITIALIZER(&dying_cps);

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

unsigned
PFD_State(const struct pfd *p)
{
        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        return (p->state);
}

int *
PFD_Fd(struct pfd *p)
{
        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        return (&(p->fd));
}

vtim_dur
PFD_Age(const struct pfd *p)
{
        vtim_mono t_mono;

        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        t_mono = VTIM_mono();
        assert(t_mono >= p->created);

        return (t_mono - p->created);
}

uint64_t
PFD_Reused(const struct pfd *p)
{
        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        return (p->reused);
}

void
PFD_LocalName(const struct pfd *p, char *abuf, unsigned alen, char *pbuf,
              unsigned plen)
{
        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        CHECK_OBJ_NOTNULL(p->conn_pool, CONN_POOL_MAGIC);
        p->conn_pool->methods->local_name(p, abuf, alen, pbuf, plen);
}

void
PFD_RemoteName(const struct pfd *p, char *abuf, unsigned alen, char *pbuf,
               unsigned plen)
{
        CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
        CHECK_OBJ_NOTNULL(p->conn_pool, CONN_POOL_MAGIC);
        p->conn_pool->methods->remote_name(p, abuf, alen, pbuf, plen);
}

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

static inline int
vcp_cmp(const struct conn_pool *a, const struct conn_pool *b)
{
        return (memcmp(a->ident, b->ident, sizeof b->ident));
}

/*--------------------------------------------------------------------
 * Waiter-handler
 */

static void  v_matchproto_(waiter_handle_f)
vcp_handle(struct waited *w, enum wait_event ev, vtim_real now)
{
        struct pfd *pfd;
        struct conn_pool *cp;

        CHECK_OBJ_NOTNULL(w, WAITED_MAGIC);
        CAST_OBJ_NOTNULL(pfd, w->priv1, PFD_MAGIC);
        (void)ev;
        (void)now;
        CHECK_OBJ_NOTNULL(pfd->conn_pool, CONN_POOL_MAGIC);
        cp = pfd->conn_pool;

        Lck_Lock(&cp->mtx);

        switch (pfd->state) {
        case PFD_STATE_STOLEN:
                pfd->state = PFD_STATE_USED;
                VTAILQ_REMOVE(&cp->connlist, pfd, list);
                AN(pfd->cond);
                PTOK(pthread_cond_signal(pfd->cond));
                break;
        case PFD_STATE_AVAIL:
                cp->methods->close(pfd);
                VTAILQ_REMOVE(&cp->connlist, pfd, list);
                cp->n_conn--;
                FREE_OBJ(pfd);
                break;
        case PFD_STATE_CLEANUP:
                cp->methods->close(pfd);
                cp->n_kill--;
                memset(pfd, 0x11, sizeof *pfd);
                free(pfd);
                break;
        default:
                WRONG("Wrong pfd state");
        }
        Lck_Unlock(&cp->mtx);
}


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

void
VCP_AddRef(struct conn_pool *cp)
{
        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);

        Lck_Lock(&conn_pools_mtx);
        assert(cp->refcnt > 0);
        cp->refcnt++;
        Lck_Unlock(&conn_pools_mtx);
}

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

static void
vcp_destroy(struct conn_pool **cpp)
{
        struct conn_pool *cp;

        TAKE_OBJ_NOTNULL(cp, cpp, CONN_POOL_MAGIC);
        AZ(cp->n_conn);
        AZ(cp->n_kill);
        Lck_Delete(&cp->mtx);
        FREE_OBJ(cp->endpoint);
        FREE_OBJ(cp);
}

/*--------------------------------------------------------------------
 * Release Conn pool, destroy or stash for future destruction if last
 * reference.
 */

void
VCP_Rel(struct conn_pool **cpp)
{
        struct conn_pool *cp;
        struct pfd *pfd, *pfd2;
        int n_kill;

        TAKE_OBJ_NOTNULL(cp, cpp, CONN_POOL_MAGIC);

        Lck_Lock(&conn_pools_mtx);
        assert(cp->refcnt > 0);
        if (--cp->refcnt > 0) {
                Lck_Unlock(&conn_pools_mtx);
                return;
        }
        AZ(cp->n_used);
        VRBT_REMOVE(vrb, &conn_pools, cp);
        Lck_Unlock(&conn_pools_mtx);

        Lck_Lock(&cp->mtx);
        VTAILQ_FOREACH_SAFE(pfd, &cp->connlist, list, pfd2) {
                VTAILQ_REMOVE(&cp->connlist, pfd, list);
                cp->n_conn--;
                assert(pfd->state == PFD_STATE_AVAIL);
                pfd->state = PFD_STATE_CLEANUP;
                (void)shutdown(pfd->fd, SHUT_RDWR);
                cp->n_kill++;
        }
        n_kill = cp->n_kill;
        Lck_Unlock(&cp->mtx);
        if (n_kill == 0) {
                vcp_destroy(&cp);
                return;
        }
        Lck_Lock(&dead_pools_mtx);
        /*
         * Here we reuse cp's entry but it will probably not be correctly
         * indexed because of the hack in VCP_RelPoll
         */
        VRBT_INSERT(vrb, &dead_pools, cp);
        Lck_Unlock(&dead_pools_mtx);
}

void
VCP_RelPoll(void)
{
        struct vrb dead;
        struct conn_pool *cp, *cp2;
        int n_kill;

        ASSERT_CLI();

        Lck_Lock(&dead_pools_mtx);
        if (VRBT_EMPTY(&dead_pools)) {
                Lck_Unlock(&dead_pools_mtx);
                return;
        }
        dead = dead_pools;
        VRBT_INIT(&dead_pools);
        Lck_Unlock(&dead_pools_mtx);

        VRBT_FOREACH_SAFE(cp, vrb, &dead, cp2) {
                CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
                Lck_Lock(&cp->mtx);
                n_kill = cp->n_kill;
                Lck_Unlock(&cp->mtx);
                if (n_kill > 0)
                        continue;
                VRBT_REMOVE(vrb, &dead, cp);
                vcp_destroy(&cp);
        }

        if (VRBT_EMPTY(&dead))
                return;

        Lck_Lock(&dead_pools_mtx);
        /*
         * The following insertion will most likely result in an
         * unordered tree, but in this case it does not matter
         * as we just want to iterate over all the elements
         * in the tree in order to delete them.
         */
        VRBT_INSERT(vrb, &dead_pools, dead.rbh_root);
        Lck_Unlock(&dead_pools_mtx);
}

/*--------------------------------------------------------------------
 * Recycle a connection.
 */

void
VCP_Recycle(const struct worker *wrk, struct pfd **pfdp)
{
        struct pfd *pfd;
        struct conn_pool *cp;
        int i = 0;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        TAKE_OBJ_NOTNULL(pfd, pfdp, PFD_MAGIC);
        cp = pfd->conn_pool;
        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);

        assert(pfd->state == PFD_STATE_USED);
        assert(pfd->fd > 0);

        Lck_Lock(&cp->mtx);
        cp->n_used--;

        pfd->waited->priv1 = pfd;
        pfd->waited->fd = pfd->fd;
        pfd->waited->idle = VTIM_real();
        pfd->state = PFD_STATE_AVAIL;
        pfd->waited->func = vcp_handle;
        pfd->waited->tmo = cache_param->backend_idle_timeout;
        if (Wait_Enter(wrk->pool->waiter, pfd->waited)) {
                cp->methods->close(pfd);
                memset(pfd, 0x33, sizeof *pfd);
                free(pfd);
                // XXX: stats
                pfd = NULL;
        } else {
                VTAILQ_INSERT_HEAD(&cp->connlist, pfd, list);
                i++;
        }

        if (pfd != NULL)
                cp->n_conn++;
        Lck_Unlock(&cp->mtx);

        if (i && DO_DEBUG(DBG_VTC_MODE)) {
                /*
                 * In vinyltest we do not have the luxury of using
                 * multiple backend connections, so whenever we end up
                 * in the "pending" case, take a short nap to let the
                 * waiter catch up and put the pfd back into circulations.
                 *
                 * In particular ESI:include related tests suffer random
                 * failures without this.
                 *
                 * In normal operation, the only effect is that we will
                 * have N+1 backend connections rather than N, which is
                 * entirely harmless.
                 */
                VTIM_sleep(0.01);
        }
}

/*--------------------------------------------------------------------
 * Open a new connection from pool.
 */

int
VCP_Open(struct conn_pool *cp, vtim_dur tmo, VCL_IP *ap, int *err)
{
        int r;
        vtim_mono h;

        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
        AN(err);

        while (cp->holddown > 0) {
                Lck_Lock(&cp->mtx);
                if (cp->holddown == 0) {
                        Lck_Unlock(&cp->mtx);
                        break;
                }

                if (VTIM_mono() >= cp->holddown) {
                        cp->holddown = 0;
                        Lck_Unlock(&cp->mtx);
                        break;
                }

                *err = 0;
                errno = cp->holddown_errno;
                Lck_Unlock(&cp->mtx);
                return (-1);
        }

        *err = errno = 0;
        r = cp->methods->open(cp, tmo, ap);

        if (r >= 0 && errno == 0 && cp->endpoint->preamble != NULL &&
             cp->endpoint->preamble->len > 0) {
                CHECK_OBJ(cp->endpoint->preamble, VRT_BLOB_MAGIC);
                if (write(r, cp->endpoint->preamble->blob,
                    cp->endpoint->preamble->len) !=
                    cp->endpoint->preamble->len) {
                        *err = errno;
                        closefd(&r);
                }
        } else {
                *err = errno;
        }

        if (r >= 0)
                return (r);

        h = 0;

        switch (errno) {
        case EACCES:
        case EPERM:
                h = cache_param->backend_local_error_holddown;
                break;
        case EADDRNOTAVAIL:
                h = cache_param->backend_local_error_holddown;
                break;
        case ECONNREFUSED:
                h = cache_param->backend_remote_error_holddown;
                break;
        case ENETUNREACH:
                h = cache_param->backend_remote_error_holddown;
                break;
        default:
                break;
        }

        if (h == 0)
                return (r);

        Lck_Lock(&cp->mtx);
        h += VTIM_mono();
        if (cp->holddown == 0 || h < cp->holddown) {
                cp->holddown = h;
                cp->holddown_errno = errno;
        }

        Lck_Unlock(&cp->mtx);

        return (r);
}

/*--------------------------------------------------------------------
 * Close a connection.
 */

void
VCP_Close(struct pfd **pfdp)
{
        struct pfd *pfd;
        struct conn_pool *cp;

        TAKE_OBJ_NOTNULL(pfd, pfdp, PFD_MAGIC);
        cp = pfd->conn_pool;
        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);

        assert(pfd->fd > 0);

        Lck_Lock(&cp->mtx);
        assert(pfd->state == PFD_STATE_USED || pfd->state == PFD_STATE_STOLEN);
        cp->n_used--;
        if (pfd->state == PFD_STATE_STOLEN) {
                (void)shutdown(pfd->fd, SHUT_RDWR);
                VTAILQ_REMOVE(&cp->connlist, pfd, list);
                pfd->state = PFD_STATE_CLEANUP;
                cp->n_kill++;
        } else {
                assert(pfd->state == PFD_STATE_USED);
                cp->methods->close(pfd);
                memset(pfd, 0x44, sizeof *pfd);
                free(pfd);
        }
        Lck_Unlock(&cp->mtx);
}

/*--------------------------------------------------------------------
 * Get a connection, possibly recycled
 */

struct pfd *
VCP_Get(struct conn_pool *cp, vtim_dur tmo, struct worker *wrk,
    unsigned force_fresh, int *err)
{
        struct pfd *pfd;

        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        AN(err);

        *err = 0;
        Lck_Lock(&cp->mtx);
        pfd = VTAILQ_FIRST(&cp->connlist);
        CHECK_OBJ_ORNULL(pfd, PFD_MAGIC);
        if (force_fresh || pfd == NULL || pfd->state == PFD_STATE_STOLEN) {
                pfd = NULL;
        } else {
                assert(pfd->conn_pool == cp);
                assert(pfd->state == PFD_STATE_AVAIL);
                VTAILQ_REMOVE(&cp->connlist, pfd, list);
                VTAILQ_INSERT_TAIL(&cp->connlist, pfd, list);
                cp->n_conn--;
                VSC_C_main->backend_reuse++;
                pfd->state = PFD_STATE_STOLEN;
                pfd->cond = &wrk->cond;
                pfd->reused++;
        }
        cp->n_used++;                   // Opening mostly works
        Lck_Unlock(&cp->mtx);

        if (pfd != NULL)
                return (pfd);

        ALLOC_OBJ(pfd, PFD_MAGIC);
        AN(pfd);
        INIT_OBJ(pfd->waited, WAITED_MAGIC);
        pfd->state = PFD_STATE_USED;
        pfd->conn_pool = cp;
        pfd->fd = VCP_Open(cp, tmo, &pfd->addr, err);
        if (pfd->fd < 0) {
                FREE_OBJ(pfd);
                Lck_Lock(&cp->mtx);
                cp->n_used--;           // Nope, didn't work after all.
                Lck_Unlock(&cp->mtx);
        } else {
                pfd->created = VTIM_mono();
                VSC_C_main->backend_conn++;
        }

        return (pfd);
}

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

int
VCP_Wait(struct worker *wrk, struct pfd *pfd, vtim_real when)
{
        struct conn_pool *cp;
        int r;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CHECK_OBJ_NOTNULL(pfd, PFD_MAGIC);
        cp = pfd->conn_pool;
        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
        assert(pfd->cond == &wrk->cond);
        Lck_Lock(&cp->mtx);
        while (pfd->state == PFD_STATE_STOLEN) {
                r = Lck_CondWaitUntil(&wrk->cond, &cp->mtx, when);
                if (r != 0) {
                        if (r == EINTR)
                                continue;
                        assert(r == ETIMEDOUT);
                        Lck_Unlock(&cp->mtx);
                        return (1);
                }
        }
        assert(pfd->state == PFD_STATE_USED);
        pfd->cond = NULL;
        Lck_Unlock(&cp->mtx);

        return (0);
}

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

VCL_IP
VCP_GetIp(struct pfd *pfd)
{

        CHECK_OBJ_NOTNULL(pfd, PFD_MAGIC);
        return (pfd->addr);
}

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

static void
vcp_panic_endpoint(struct vsb *vsb, const struct vrt_endpoint *vep)
{
        char h[VTCP_ADDRBUFSIZE];
        char p[VTCP_PORTBUFSIZE];

        if (PAN_dump_struct(vsb, vep, VRT_ENDPOINT_MAGIC, "vrt_endpoint"))
                return;
        if (vep->uds_path)
                VSB_printf(vsb, "uds_path = %s,\n", vep->uds_path);
        if (vep->ipv4 && VSA_Sane(vep->ipv4)) {
                VTCP_name(vep->ipv4, h, sizeof h, p, sizeof p);
                VSB_printf(vsb, "ipv4 = %s, ", h);
                VSB_printf(vsb, "port = %s,\n", p);
        }
        if (vep->ipv6 && VSA_Sane(vep->ipv6)) {
                VTCP_name(vep->ipv6, h, sizeof h, p, sizeof p);
                VSB_printf(vsb, "ipv6 = %s, ", h);
                VSB_printf(vsb, "port = %s,\n", p);
        }
        VSB_indent(vsb, -2);
        VSB_cat(vsb, "},\n");
}

void
VCP_Panic(struct vsb *vsb, struct conn_pool *cp)
{

        if (PAN_dump_struct(vsb, cp, CONN_POOL_MAGIC, "conn_pool"))
                return;
        VSB_cat(vsb, "ident = ");
        VSB_quote(vsb, cp->ident, VSHA256_DIGEST_LENGTH, VSB_QUOTE_HEX);
        VSB_cat(vsb, ",\n");
        vcp_panic_endpoint(vsb, cp->endpoint);
        VSB_indent(vsb, -2);
        VSB_cat(vsb, "},\n");
}

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

void
VCP_Init(void)
{
        Lck_New(&conn_pools_mtx, lck_conn_pool);
        Lck_New(&dead_pools_mtx, lck_dead_pool);

        AZ(vsc);
        vsc = VSC_vcp_New(NULL, NULL, "");
        AN(vsc);
}

/**********************************************************************/

static inline int
tmo2msec(vtim_dur tmo)
{
        return ((int)floor(tmo * 1000.0));
}

static int v_matchproto_(cp_open_f)
vtp_open(const struct conn_pool *cp, vtim_dur tmo, VCL_IP *ap)
{
        int s;
        int msec;

        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);

        msec = tmo2msec(tmo);
        if (cache_param->prefer_ipv6) {
                *ap = cp->endpoint->ipv6;
                s = VTCP_connect(*ap, msec);
                if (s >= 0)
                        return (s);
        }
        *ap = cp->endpoint->ipv4;
        s = VTCP_connect(*ap, msec);
        if (s >= 0)
                return (s);
        if (!cache_param->prefer_ipv6) {
                *ap = cp->endpoint->ipv6;
                s = VTCP_connect(*ap, msec);
        }
        return (s);
}


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

static void v_matchproto_(cp_close_f)
vtp_close(struct pfd *pfd)
{

        CHECK_OBJ_NOTNULL(pfd, PFD_MAGIC);
        VTCP_close(&pfd->fd);
}

static void v_matchproto_(cp_name_f)
vtp_local_name(const struct pfd *pfd, char *addr, unsigned alen, char *pbuf,
               unsigned plen)
{
        CHECK_OBJ_NOTNULL(pfd, PFD_MAGIC);
        VTCP_myname(pfd->fd, addr, alen, pbuf, plen);
}

static void v_matchproto_(cp_name_f)
vtp_remote_name(const struct pfd *pfd, char *addr, unsigned alen, char *pbuf,
                unsigned plen)
{
        CHECK_OBJ_NOTNULL(pfd, PFD_MAGIC);
        VTCP_hisname(pfd->fd, addr, alen, pbuf, plen);
}

static const struct cp_methods vtp_methods = {
        .open = vtp_open,
        .close = vtp_close,
        .local_name = vtp_local_name,
        .remote_name = vtp_remote_name,
};

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

static int v_matchproto_(cp_open_f)
vus_open(const struct conn_pool *cp, vtim_dur tmo, VCL_IP *ap)
{
        int s;
        int msec;

        CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
        AN(cp->endpoint->uds_path);

        msec = tmo2msec(tmo);
        *ap = bogo_ip;
        s = VUS_connect(cp->endpoint->uds_path, msec);
        return (s);
}

static void v_matchproto_(cp_name_f)
vus_name(const struct pfd *pfd, char *addr, unsigned alen, char *pbuf,
         unsigned plen)
{
        (void) pfd;
        assert(alen > vstrlen("0.0.0.0"));
        assert(plen > 1);
        strcpy(addr, "0.0.0.0");
        strcpy(pbuf, "0");
}

static const struct cp_methods vus_methods = {
        .open = vus_open,
        .close = vtp_close,
        .local_name = vus_name,
        .remote_name = vus_name,
};

/*--------------------------------------------------------------------
 * Reference a TCP pool given by {ip4, ip6} pair or a UDS.  Create if
 * it doesn't exist already.
 */

struct conn_pool *
VCP_Ref(const struct vrt_endpoint *vep, const char *ident)
{
        struct conn_pool *cp, *cp2;
        struct VSHA256Context cx[1];
        unsigned char digest[VSHA256_DIGEST_LENGTH];

        CHECK_OBJ_NOTNULL(vep, VRT_ENDPOINT_MAGIC);
        AN(ident);
        AN(vsc);

        VSHA256_Init(cx);
        VSHA256_Update(cx, ident, vstrlen(ident) + 1); // include \0
        if (vep->uds_path != NULL) {
                AZ(vep->ipv4);
                AZ(vep->ipv6);
                VSHA256_Update(cx, "UDS", 4); // include \0
                VSHA256_Update(cx, vep->uds_path, vstrlen(vep->uds_path));
        } else {
                assert(vep->ipv4 != NULL || vep->ipv6 != NULL);
                if (vep->ipv4 != NULL) {
                        assert(VSA_Sane(vep->ipv4));
                        VSHA256_Update(cx, "IP4", 4); // include \0
                        VSHA256_Update(cx, vep->ipv4, vsa_suckaddr_len);
                }
                if (vep->ipv6 != NULL) {
                        assert(VSA_Sane(vep->ipv6));
                        VSHA256_Update(cx, "IP6", 4); // include \0
                        VSHA256_Update(cx, vep->ipv6, vsa_suckaddr_len);
                }
        }
        CHECK_OBJ_ORNULL(vep->preamble, VRT_BLOB_MAGIC);
        if (vep->preamble != NULL && vep->preamble->len > 0) {
                VSHA256_Update(cx, "PRE", 4); // include \0
                VSHA256_Update(cx, vep->preamble->blob, vep->preamble->len);
        }
        VSHA256_Final(digest, cx);

        ALLOC_OBJ(cp, CONN_POOL_MAGIC);
        AN(cp);
        cp->refcnt = 1;
        cp->holddown = 0;
        cp->endpoint = VRT_Endpoint_Clone(vep);
        CHECK_OBJ_NOTNULL(cp->endpoint, VRT_ENDPOINT_MAGIC);
        memcpy(cp->ident, digest, sizeof cp->ident);
        if (vep->uds_path != NULL)
                cp->methods = &vus_methods;
        else
                cp->methods = &vtp_methods;
        Lck_New(&cp->mtx, lck_conn_pool);
        VTAILQ_INIT(&cp->connlist);

        Lck_Lock(&conn_pools_mtx);
        cp2 = VRBT_INSERT(vrb, &conn_pools, cp);
        if (cp2 == NULL) {
                vsc->ref_miss++;
                Lck_Unlock(&conn_pools_mtx);
                return (cp);
        }

        CHECK_OBJ(cp2, CONN_POOL_MAGIC);
        assert(cp2->refcnt > 0);
        cp2->refcnt++;
        vsc->ref_hit++;
        Lck_Unlock(&conn_pools_mtx);

        vcp_destroy(&cp);
        return (cp2);
}

		vinyl-cache/bin/vinyld/cache/cache_conn_pool.c
0		/*-
1		* Copyright (c) 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		* (TCP\|UDS) connection pools.
30		*
31		*/
32
33		#include "config.h"
34
35		#include <stdlib.h>
36
37		#include "cache_int.h"
38
39		#include "vsa.h"
40		#include "vsha256.h"
41		#include "vtcp.h"
42		#include "vus.h"
43		#include "vtim.h"
44		#include "waiter/waiter.h"
45
46		#include "cache_conn_pool.h"
47		#include "cache_pool.h"
48
49		#include "VSC_vcp.h"
50
51		struct conn_pool;
52		static inline int vcp_cmp(const struct conn_pool a, const struct conn_pool b);
53
54		/*--------------------------------------------------------------------
55		*/
56
57		struct pfd {
58		unsigned magic;
59		#define PFD_MAGIC 0x0c5e6593
60		int fd;
61		VTAILQ_ENTRY(pfd) list;
62		VCL_IP addr;
63		uint8_t state;
64		struct waited waited[1];
65		struct conn_pool *conn_pool;
66
67		pthread_cond_t *cond;
68
69		vtim_mono created;
70		uint64_t reused;
71		};
72
73		/*--------------------------------------------------------------------
74		*/
75
76		typedef int cp_open_f(const struct conn_pool , vtim_dur tmo, VCL_IP ap);
77		typedef void cp_close_f(struct pfd *);
78		typedef void cp_name_f(const struct pfd , char , unsigned, char *, unsigned);
79
80		struct cp_methods {
81		cp_open_f *open;
82		cp_close_f *close;
83		cp_name_f *local_name;
84		cp_name_f *remote_name;
85		};
86
87		struct conn_pool {
88		unsigned magic;
89		#define CONN_POOL_MAGIC 0x85099bc3
90
91		const struct cp_methods *methods;
92
93		struct vrt_endpoint *endpoint;
94		char ident[VSHA256_DIGEST_LENGTH];
95
96		VRBT_ENTRY(conn_pool) entry;
97		int refcnt;
98		struct lock mtx;
99
100		VTAILQ_HEAD(, pfd) connlist;
101		int n_conn;
102
103		int n_kill;
104
105		int n_used;
106
107		vtim_mono holddown;
108		int holddown_errno;
109		};
110
111		static struct lock conn_pools_mtx;
112		static struct lock dead_pools_mtx;
113		static struct VSC_vcp *vsc;
114
115		VRBT_HEAD(vrb, conn_pool);
116	64	VRBT_GENERATE_REMOVE_COLOR(vrb, conn_pool, entry, static)
117	167	VRBT_GENERATE_REMOVE(vrb, conn_pool, entry, static)
118	883	VRBT_GENERATE_INSERT_COLOR(vrb, conn_pool, entry, static)
119	4238	VRBT_GENERATE_INSERT_FINISH(vrb, conn_pool, entry, static)
120	6421	VRBT_GENERATE_INSERT(vrb, conn_pool, entry, vcp_cmp, static)
121	10	VRBT_GENERATE_NEXT(vrb, conn_pool, entry, static)
122	20	VRBT_GENERATE_MINMAX(vrb, conn_pool, entry, static)
123
124		static struct vrb conn_pools = VRBT_INITIALIZER(&conn_pools);
125		static struct vrb dead_pools = VRBT_INITIALIZER(&dying_cps);
126
127		/*--------------------------------------------------------------------
128		*/
129
130		unsigned
131	42691	PFD_State(const struct pfd *p)
132		{
133	42691	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
134	42691	return (p->state);
135		}
136
137		int *
138	18030	PFD_Fd(struct pfd *p)
139		{
140	18030	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
141	18030	return (&(p->fd));
142		}
143
144		vtim_dur
145	3292	PFD_Age(const struct pfd *p)
146		{
147		vtim_mono t_mono;
148
149	3292	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
150	3292	t_mono = VTIM_mono();
151	3292	assert(t_mono >= p->created);
152
153	3292	return (t_mono - p->created);
154		}
155
156		uint64_t
157	3292	PFD_Reused(const struct pfd *p)
158		{
159	3292	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
160	3292	return (p->reused);
161		}
162
163		void
164	8968	PFD_LocalName(const struct pfd p, char abuf, unsigned alen, char *pbuf,
165		unsigned plen)
166		{
167	8968	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
168	8968	CHECK_OBJ_NOTNULL(p->conn_pool, CONN_POOL_MAGIC);
169	8968	p->conn_pool->methods->local_name(p, abuf, alen, pbuf, plen);
170	8968	}
171
172		void
173	8968	PFD_RemoteName(const struct pfd p, char abuf, unsigned alen, char *pbuf,
174		unsigned plen)
175		{
176	8968	CHECK_OBJ_NOTNULL(p, PFD_MAGIC);
177	8968	CHECK_OBJ_NOTNULL(p->conn_pool, CONN_POOL_MAGIC);
178	8968	p->conn_pool->methods->remote_name(p, abuf, alen, pbuf, plen);
179	8968	}
180
181		/*--------------------------------------------------------------------
182		*/
183
184		static inline int
185	2183	vcp_cmp(const struct conn_pool a, const struct conn_pool b)
186		{
187	2183	return (memcmp(a->ident, b->ident, sizeof b->ident));
188		}
189
190		/*--------------------------------------------------------------------
191		* Waiter-handler
192		*/
193
194		static void v_matchproto_(waiter_handle_f)
195	7252	vcp_handle(struct waited *w, enum wait_event ev, vtim_real now)
196		{
197		struct pfd *pfd;
198		struct conn_pool *cp;
199
200	7252	CHECK_OBJ_NOTNULL(w, WAITED_MAGIC);
201	7252	CAST_OBJ_NOTNULL(pfd, w->priv1, PFD_MAGIC);
202	7252	(void)ev;
203	7252	(void)now;
204	7252	CHECK_OBJ_NOTNULL(pfd->conn_pool, CONN_POOL_MAGIC);
205	7252	cp = pfd->conn_pool;
206
207	7252	Lck_Lock(&cp->mtx);
208
209	7252	switch (pfd->state) {
210		case PFD_STATE_STOLEN:
211	3284	pfd->state = PFD_STATE_USED;
212	3284	VTAILQ_REMOVE(&cp->connlist, pfd, list);
213	3284	AN(pfd->cond);
214	3284	PTOK(pthread_cond_signal(pfd->cond));
215	3284	break;
216		case PFD_STATE_AVAIL:
217	3952	cp->methods->close(pfd);
218	3952	VTAILQ_REMOVE(&cp->connlist, pfd, list);
219	3952	cp->n_conn--;
220	3952	FREE_OBJ(pfd);
221	3952	break;
222		case PFD_STATE_CLEANUP:
223	16	cp->methods->close(pfd);
224	16	cp->n_kill--;
225	16	memset(pfd, 0x11, sizeof *pfd);
226	16	free(pfd);
227	16	break;
228		default:
229	0	WRONG("Wrong pfd state");
230	0	}
231	7252	Lck_Unlock(&cp->mtx);
232	7252	}
233
234
235		/*--------------------------------------------------------------------
236		*/
237
238		void
239	152	VCP_AddRef(struct conn_pool *cp)
240		{
241	152	CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
242
243	152	Lck_Lock(&conn_pools_mtx);
244	152	assert(cp->refcnt > 0);
245	152	cp->refcnt++;
246	152	Lck_Unlock(&conn_pools_mtx);
247	152	}
248
249		/*--------------------------------------------------------------------
250		*/
251
252		static void
253	1340	vcp_destroy(struct conn_pool **cpp)
254		{
255		struct conn_pool *cp;
256
257	1340	TAKE_OBJ_NOTNULL(cp, cpp, CONN_POOL_MAGIC);
258	1340	AZ(cp->n_conn);
259	1340	AZ(cp->n_kill);
260	1340	Lck_Delete(&cp->mtx);
261	1340	FREE_OBJ(cp->endpoint);
262	1340	FREE_OBJ(cp);
263	1340	}
264
265		/*--------------------------------------------------------------------
266		* Release Conn pool, destroy or stash for future destruction if last
267		* reference.
268		*/
269
270		void
271	348	VCP_Rel(struct conn_pool **cpp)
272		{
273		struct conn_pool *cp;
274		struct pfd pfd, pfd2;
275		int n_kill;
276
277	348	TAKE_OBJ_NOTNULL(cp, cpp, CONN_POOL_MAGIC);
278
279	348	Lck_Lock(&conn_pools_mtx);
280	348	assert(cp->refcnt > 0);
281	348	if (--cp->refcnt > 0) {
282	240	Lck_Unlock(&conn_pools_mtx);
283	240	return;
284		}
285	108	AZ(cp->n_used);
286	108	VRBT_REMOVE(vrb, &conn_pools, cp);
287	108	Lck_Unlock(&conn_pools_mtx);
288
289	108	Lck_Lock(&cp->mtx);
290	116	VTAILQ_FOREACH_SAFE(pfd, &cp->connlist, list, pfd2) {
291	8	VTAILQ_REMOVE(&cp->connlist, pfd, list);
292	8	cp->n_conn--;
293	8	assert(pfd->state == PFD_STATE_AVAIL);
294	8	pfd->state = PFD_STATE_CLEANUP;
295	8	(void)shutdown(pfd->fd, SHUT_RDWR);
296	8	cp->n_kill++;
297	8	}
298	108	n_kill = cp->n_kill;
299	108	Lck_Unlock(&cp->mtx);
300	108	if (n_kill == 0) {
301	100	vcp_destroy(&cp);
302	100	return;
303		}
304	8	Lck_Lock(&dead_pools_mtx);
305		/*
306		* Here we reuse cp's entry but it will probably not be correctly
307		* indexed because of the hack in VCP_RelPoll
308		*/
309	8	VRBT_INSERT(vrb, &dead_pools, cp);
310	8	Lck_Unlock(&dead_pools_mtx);
311	348	}
312
313		void
314	31828	VCP_RelPoll(void)
315		{
316		struct vrb dead;
317		struct conn_pool cp, cp2;
318		int n_kill;
319
320	31828	ASSERT_CLI();
321
322	31828	Lck_Lock(&dead_pools_mtx);
323	31828	if (VRBT_EMPTY(&dead_pools)) {
324	31818	Lck_Unlock(&dead_pools_mtx);
325	31818	return;
326		}
327	10	dead = dead_pools;
328	10	VRBT_INIT(&dead_pools);
329	10	Lck_Unlock(&dead_pools_mtx);
330
331	20	VRBT_FOREACH_SAFE(cp, vrb, &dead, cp2) {
332	10	CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
333	10	Lck_Lock(&cp->mtx);
334	10	n_kill = cp->n_kill;
335	10	Lck_Unlock(&cp->mtx);
336	10	if (n_kill > 0)
337	2	continue;
338	8	VRBT_REMOVE(vrb, &dead, cp);
339	8	vcp_destroy(&cp);
340	8	}
341
342	10	if (VRBT_EMPTY(&dead))
343	8	return;
344
345	2	Lck_Lock(&dead_pools_mtx);
346		/*
347		* The following insertion will most likely result in an
348		* unordered tree, but in this case it does not matter
349		* as we just want to iterate over all the elements
350		* in the tree in order to delete them.
351		*/
352	2	VRBT_INSERT(vrb, &dead_pools, dead.rbh_root);
353	2	Lck_Unlock(&dead_pools_mtx);
354	31828	}
355
356		/*--------------------------------------------------------------------
357		* Recycle a connection.
358		*/
359
360		void
361	7348	VCP_Recycle(const struct worker wrk, struct pfd *pfdp)
362		{
363		struct pfd *pfd;
364		struct conn_pool *cp;
365	7348	int i = 0;
366
367	7348	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
368	7348	TAKE_OBJ_NOTNULL(pfd, pfdp, PFD_MAGIC);
369	7348	cp = pfd->conn_pool;
370	7348	CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
371
372	7348	assert(pfd->state == PFD_STATE_USED);
373	7348	assert(pfd->fd > 0);
374
375	7348	Lck_Lock(&cp->mtx);
376	7348	cp->n_used--;
377
378	7348	pfd->waited->priv1 = pfd;
379	7348	pfd->waited->fd = pfd->fd;
380	7348	pfd->waited->idle = VTIM_real();
381	7348	pfd->state = PFD_STATE_AVAIL;
382	7348	pfd->waited->func = vcp_handle;
383	7348	pfd->waited->tmo = cache_param->backend_idle_timeout;
384	7348	if (Wait_Enter(wrk->pool->waiter, pfd->waited)) {
385	0	cp->methods->close(pfd);
386	0	memset(pfd, 0x33, sizeof *pfd);
387	0	free(pfd);
388		// XXX: stats
389	0	pfd = NULL;
390	0	} else {
391	7348	VTAILQ_INSERT_HEAD(&cp->connlist, pfd, list);
392	7348	i++;
393		}
394
395	7348	if (pfd != NULL)
396	7348	cp->n_conn++;
397	7348	Lck_Unlock(&cp->mtx);
398
399	7348	if (i && DO_DEBUG(DBG_VTC_MODE)) {
400		/*
401		* In vinyltest we do not have the luxury of using
402		* multiple backend connections, so whenever we end up
403		* in the "pending" case, take a short nap to let the
404		* waiter catch up and put the pfd back into circulations.
405		*
406		* In particular ESI:include related tests suffer random
407		* failures without this.
408		*
409		* In normal operation, the only effect is that we will
410		* have N+1 backend connections rather than N, which is
411		* entirely harmless.
412		*/
413	7348	VTIM_sleep(0.01);
414	7348	}
415	7348	}
416
417		/*--------------------------------------------------------------------
418		* Open a new connection from pool.
419		*/
420
421		int
422	6599	VCP_Open(struct conn_pool cp, vtim_dur tmo, VCL_IP ap, int *err)
423		{
424		int r;
425		vtim_mono h;
426
427	6599	CHECK_OBJ_NOTNULL(cp, CONN_POOL_MAGIC);
428	6599	AN(err);
429
430	6599	while (cp->holddown > 0) {
431	89	Lck_Lock(&cp->mtx);
432	89	if (cp->holddown == 0) {
433	0	Lck_Unlock(&cp->mtx);
434	0	break;
435		}
436
437	89	if (VTIM_mono() >= cp->holddown) {
438	10	cp->holddown = 0;
439	10	Lck_Unlock(&cp->mtx);
440	10	break;
441		}
442
443	79	*err = 0;
444	79	errno = cp->holddown_errno;
445	79	Lck_Unlock(&cp->mtx);
446	79	return (-1);
447		}
448
449	6520	*err = errno = 0;
450	6520	r = cp->methods->open(cp, tmo, ap);
451
452	6520	if (r >= 0 && errno == 0 && cp->endpoint->preamble != NULL &&
453	36	cp->endpoint->preamble->len > 0) {
454	36	CHECK_OBJ(cp->endpoint->preamble, VRT_BLOB_MAGIC);
455	108	if (write(r, cp->endpoint->preamble->blob,
456	72	cp->endpoint->preamble->len) !=
457	36	cp->endpoint->preamble->len) {
458	0	*err = errno;
459	0	closefd(&r);
460	0	}
461	36	} else {
462	6484	*err = errno;
463		}
464
465	6520	if (r >= 0)
466	6413	return (r);
467
468	107	h = 0;
469
470	107	switch (errno) {
471		case EACCES:
472		case EPERM:
473	0	h = cache_param->backend_local_error_holddown;
474	0	break;
475		case EADDRNOTAVAIL:
476	0	h = cache_param->backend_local_error_holddown;
477	0	break;
478		case ECONNREFUSED:
479	107	h = cache_param->backend_remote_error_holddown;
480	107	break;
481		case ENETUNREACH:
482	0	h = cache_param->backend_remote_error_holddown;
483	0	break;
484		default:
485	0	break;
486		}
487
488	107	if (h == 0)
489	0	return (r);
490
491	107	Lck_Lock(&cp->mtx);
492	107	h += VTIM_mono();
493	107	if (cp->holddown == 0 \|\| h < cp->holddown) {
494	98	cp->holddown = h;
495	98	cp->holddown_errno = errno;
496	98	}
497
498	107	Lck_Unlock(&cp->mtx);
499