GCOV - Varnish Cache


/*-
 * Copyright (c) 2020 Varnish Software
 * All rights reserved.
 *
 * Author: Dridi Boukelmoune <dridi.boukelmoune@gmail.com>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */

#include "config.h"

#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>

#include <errno.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "vtc.h"

#include "vsa.h"
#include "vtcp.h"

/* SECTION: tunnel tunnel
 *
 * The goal of a tunnel is to help control the data transfer between two
 * parties, for example to trigger socket timeouts in the middle of protocol
 * frames, without the need to change how both parties are implemented.
 *
 * A tunnel accepts a connection and then connects on behalf of the source to
 * the desired destination. Once both connections are established the tunnel
 * will transfer bytes unchanged between the source and destination. Transfer
 * can be interrupted, usually with the help of synchronization methods like
 * barriers. Once the transfer is paused, it is possible to let a specific
 * amount of bytes move in either direction.
 *
 * SECTION: tunnel.args Arguments
 *
 * \-start
 *        Start the tunnel in background, processing the last given
 *        specification.
 *
 * \-start+pause
 *        Start the tunnel, but already paused.
 *
 * \-wait
 *        Block until the thread finishes.
 *
 * \-listen STRING
 *        Dictate the listening socket for the server. STRING is of the form
 *        "IP PORT", or "HOST PORT".
 *
 *        Listens by defaults to a local random port.
 *
 * \-connect STRING
 *        Indicate the server to connect to. STRING is also of the form
 *        "IP PORT", or "HOST PORT".
 *
 *        Connects by default to a varnish instance called ``v1``.
 *
 * SECTION: tunnel.spec Specification
 *
 * The specification contains a list of tunnel commands that can be combined
 * with barriers and delays. For example::
 *
 *     tunnel t1 {
 *         barrier b1 sync
 *         pause
 *         delay 1
 *         send 42
 *         barrier b2 sync
 *         resume
 *     } -start
 *
 * If one end of the tunnel is closed before the end of the specification
 * the test case will fail. A specification that ends in a paused state will
 * implicitly resume the tunnel.
 */

enum tunnel_state_e {
        TUNNEL_ACCEPT,
        TUNNEL_RUNNING,
        TUNNEL_PAUSED,
        TUNNEL_SPEC_DONE,
        TUNNEL_POLL_DONE,
        TUNNEL_STOPPED,
};

struct tunnel_lane {
        char                    buf[1024];
        ssize_t                 buf_len;
        size_t                  wrk_len;
        int                     *rfd;
        int                     *wfd;
};

struct tunnel {
        unsigned                magic;
#define TUNNEL_MAGIC            0x7f59913d
        char                    *name;
        struct vtclog           *vl;
        VTAILQ_ENTRY(tunnel)    list;
        enum tunnel_state_e     state;
        unsigned                start_paused;

        char                    *spec;

        char                    connect[256];
        int                     csock;

        char                    listen[256];
        int                     lsock;
        char                    laddr[VTCP_ADDRBUFSIZE];
        char                    lport[VTCP_PORTBUFSIZE];

        int                     asock;

        struct tunnel_lane      send_lane[1];
        struct tunnel_lane      recv_lane[1];

        pthread_mutex_t         mtx;            /* state and lanes->*_len */
        pthread_cond_t          cond;
        pthread_t               tspec;
        pthread_t               tpoll;
};

static pthread_mutex_t          tunnel_mtx;

static VTAILQ_HEAD(, tunnel)    tunnels = VTAILQ_HEAD_INITIALIZER(tunnels);

/**********************************************************************
 * Is the tunnel still operating?
 */

static unsigned
tunnel_is_open(struct tunnel *t)
{
        unsigned is_open;

        PTOK(pthread_mutex_lock(&t->mtx));
        is_open = (t->send_lane->buf_len >= 0 && t->recv_lane->buf_len >= 0);
        PTOK(pthread_mutex_unlock(&t->mtx));
        return (is_open);
}

/**********************************************************************
 * SECTION: tunnel.spec.pause
 *
 * pause
 *         Wait for in-flight bytes to be transferred and pause the tunnel.
 *
 *         The tunnel must be running.
 */

static void
cmd_tunnel_pause(CMD_ARGS)
{
        struct tunnel *t;

        CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
        AZ(av[1]);

        if (!tunnel_is_open(t))
                vtc_fatal(vl, "Tunnel already closed");

        PTOK(pthread_mutex_lock(&t->mtx));
        if (t->state == TUNNEL_PAUSED) {
                PTOK(pthread_mutex_unlock(&t->mtx));
                vtc_fatal(vl, "Tunnel already paused");
        }
        assert(t->state == TUNNEL_RUNNING);
        t->state = TUNNEL_PAUSED;
        PTOK(pthread_cond_signal(&t->cond));
        PTOK(pthread_cond_wait(&t->cond, &t->mtx));
        PTOK(pthread_mutex_unlock(&t->mtx));
}

/**********************************************************************
 * SECTION: tunnel.spec.send
 *
 * send NUMBER
 *         Wait until NUMBER bytes are transferred from source to
 *         destination.
 *
 *         The tunnel must be paused, it remains paused afterwards.
 */

static void
cmd_tunnel_send(CMD_ARGS)
{
        struct tunnel *t;
        unsigned len;

        CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
        AN(av[1]);
        AZ(av[2]);

        len = atoi(av[1]);

        if (!tunnel_is_open(t))
                vtc_fatal(vl, "Tunnel already closed");

        PTOK(pthread_mutex_lock(&t->mtx));
        if (t->state == TUNNEL_RUNNING) {
                PTOK(pthread_mutex_unlock(&t->mtx));
                vtc_fatal(vl, "Tunnel still running");
        }
        assert(t->state == TUNNEL_PAUSED);
        AZ(t->send_lane->wrk_len);
        AZ(t->recv_lane->wrk_len);
        if (!strcmp(av[0], "send"))
                t->send_lane->wrk_len = len;
        else
                t->recv_lane->wrk_len = len;
        PTOK(pthread_cond_signal(&t->cond));
        PTOK(pthread_cond_wait(&t->cond, &t->mtx));
        PTOK(pthread_mutex_unlock(&t->mtx));
}

/**********************************************************************
 * SECTION: tunnel.spec.recv
 *
 * recv NUMBER
 *         Wait until NUMBER bytes are transferred from destination to
 *         source.
 *
 *         The tunnel must be paused, it remains paused afterwards.
 */

static void
cmd_tunnel_recv(CMD_ARGS)
{

        cmd_tunnel_send(av, priv, vl);
}

/**********************************************************************
 * SECTION: tunnel.spec.resume
 *
 * resume
 *         Resume the transfer of bytes in both directions.
 *
 *         The tunnel must be paused.
 */

static void
cmd_tunnel_resume(CMD_ARGS)
{
        struct tunnel *t;

        CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
        AZ(av[1]);

        if (!tunnel_is_open(t))
                vtc_fatal(vl, "Tunnel already closed");

        PTOK(pthread_mutex_lock(&t->mtx));
        if (t->state == TUNNEL_RUNNING) {
                PTOK(pthread_mutex_unlock(&t->mtx));
                vtc_fatal(vl, "Tunnel already running");
        }
        assert(t->state == TUNNEL_PAUSED);
        t->state = TUNNEL_RUNNING;
        PTOK(pthread_cond_signal(&t->cond));
        PTOK(pthread_mutex_unlock(&t->mtx));
}

static const struct cmds tunnel_cmds[] = {
#define CMD_TUNNEL(n) { #n, cmd_tunnel_##n },
        CMD_TUNNEL(pause)
        CMD_TUNNEL(send)
        CMD_TUNNEL(recv)
        CMD_TUNNEL(resume)
#undef CMD_TUNNEL
        { NULL, NULL }
};

/**********************************************************************
 * Tunnel poll thread
 */

static void
tunnel_read(struct tunnel *t, struct vtclog *vl, const struct pollfd *pfd,
    struct tunnel_lane *lane)
{
        size_t len;
        ssize_t res;
        enum tunnel_state_e state;

        assert(pfd->fd == *lane->rfd);
        if (!(pfd->revents & POLLIN))
                return;

        PTOK(pthread_mutex_lock(&t->mtx));
        AZ(lane->buf_len);
        len = lane->wrk_len;
        state = t->state;
        PTOK(pthread_mutex_unlock(&t->mtx));

        if (len == 0 && state == TUNNEL_PAUSED)
                return;

        if (len == 0 || len > sizeof lane->buf)
                len = sizeof lane->buf;

        res = read(pfd->fd, lane->buf, len);
        if (res < 0)
                vtc_fatal(vl, "Read failed: %s", strerror(errno));

        PTOK(pthread_mutex_lock(&t->mtx));
        lane->buf_len = (res == 0) ? -1 : res;
        PTOK(pthread_mutex_unlock(&t->mtx));
}

static void
tunnel_write(struct tunnel *t, struct vtclog *vl, struct tunnel_lane *lane,
    const char *action)
{
        const char *p;
        ssize_t res, l;

        p = lane->buf;
        l = lane->buf_len;

        if (l > 0)
                vtc_log(vl, 3, "%s %zd bytes", action, l);
        while (l > 0) {
                res = write(*lane->wfd, p, l);
                if (res <= 0)
                        vtc_fatal(vl, "Write failed: %s", strerror(errno));
                l -= res;
                p += res;
        }

        PTOK(pthread_mutex_lock(&t->mtx));
        if (lane->wrk_len > 0 && lane->buf_len != -1) {
                assert(lane->buf_len >= 0);
                assert(lane->wrk_len >= (size_t)lane->buf_len);
                lane->wrk_len -= lane->buf_len;
        }
        lane->buf_len = l;
        PTOK(pthread_mutex_unlock(&t->mtx));
}

static void *
tunnel_poll_thread(void *priv)
{
        struct tunnel *t;
        struct vtclog *vl;
        struct pollfd pfd[2];
        enum tunnel_state_e state;
        int res;

        CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);

        vl = vtc_logopen("%s", t->name);
        pthread_cleanup_push(vtc_logclose, vl);

        while (tunnel_is_open(t) && !vtc_stop) {
                PTOK(pthread_mutex_lock(&t->mtx));
                /* NB: can be woken up by `tunnel tX -wait` */
                while (t->state == TUNNEL_ACCEPT && !vtc_stop)
                        PTOK(pthread_cond_wait(&t->cond, &t->mtx));
                state = t->state;
                PTOK(pthread_mutex_unlock(&t->mtx));

                if (vtc_stop)
                        break;

                assert(state < TUNNEL_POLL_DONE);

                memset(pfd, 0, sizeof pfd);
                pfd[0].fd = *t->send_lane->rfd;
                pfd[1].fd = *t->recv_lane->rfd;
                pfd[0].events = POLLIN;
                pfd[1].events = POLLIN;
                res = poll(pfd, 2, 100);
                if (res == -1)
                        vtc_fatal(vl, "Poll failed: %s", strerror(errno));

                tunnel_read(t, vl, &pfd[0], t->send_lane);
                tunnel_read(t, vl, &pfd[1], t->recv_lane);

                PTOK(pthread_mutex_lock(&t->mtx));
                if (t->state == TUNNEL_PAUSED && t->send_lane->wrk_len == 0 &&
                    t->recv_lane->wrk_len == 0) {
                        AZ(t->send_lane->buf_len);
                        AZ(t->recv_lane->buf_len);
                        PTOK(pthread_cond_signal(&t->cond));
                        PTOK(pthread_cond_wait(&t->cond, &t->mtx));
                }
                PTOK(pthread_mutex_unlock(&t->mtx));

                if (vtc_stop)
                        break;

                tunnel_write(t, vl, t->send_lane, "Sending");
                tunnel_write(t, vl, t->recv_lane, "Receiving");
        }

        PTOK(pthread_mutex_lock(&t->mtx));
        if (t->state != TUNNEL_SPEC_DONE && !vtc_stop) {
                PTOK(pthread_cond_signal(&t->cond));
                PTOK(pthread_cond_wait(&t->cond, &t->mtx));
        }
        PTOK(pthread_mutex_unlock(&t->mtx));

        pthread_cleanup_pop(0);
        vtc_logclose(vl);
        t->state = TUNNEL_POLL_DONE;
        return (NULL);
}

/**********************************************************************
 * Tunnel spec thread
 */

static void
tunnel_accept(struct tunnel *t, struct vtclog *vl)
{
        struct vsb *vsb;
        const char *addr, *err;
        int afd, cfd;

        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        assert(t->lsock >= 0);
        assert(t->asock < 0);
        assert(t->csock < 0);
        assert(t->state == TUNNEL_ACCEPT);

        vtc_log(vl, 4, "Accepting");
        afd = accept(t->lsock, NULL, NULL);
        if (afd < 0)
                vtc_fatal(vl, "Accept failed: %s", strerror(errno));
        vtc_log(vl, 3, "Accepted socket fd is %d", afd);

        vsb = macro_expand(vl, t->connect);
        AN(vsb);
        addr = VSB_data(vsb);

        cfd = VTCP_open(addr, NULL, 10., &err);
        if (cfd < 0)
                vtc_fatal(vl, "Failed to open %s: %s", addr, err);
        vtc_log(vl, 3, "Connected socket fd is %d", cfd);
        VSB_destroy(&vsb);

        VTCP_blocking(afd);
        VTCP_blocking(cfd);

        PTOK(pthread_mutex_lock(&t->mtx));
        t->asock = afd;
        t->csock = cfd;
        t->send_lane->buf_len = 0;
        t->send_lane->wrk_len = 0;
        t->recv_lane->buf_len = 0;
        t->recv_lane->wrk_len = 0;
        if (t->start_paused) {
                t->state = TUNNEL_PAUSED;
                t->start_paused = 0;
        } else
                t->state = TUNNEL_RUNNING;
        PTOK(pthread_cond_signal(&t->cond));
        PTOK(pthread_mutex_unlock(&t->mtx));
}

static void *
tunnel_spec_thread(void *priv)
{
        struct tunnel *t;
        struct vtclog *vl;
        enum tunnel_state_e state;

        CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
        AN(*t->connect);

        vl = vtc_logopen("%s", t->name);
        vtc_log_set_cmd(vl, tunnel_cmds);
        pthread_cleanup_push(vtc_logclose, vl);

        tunnel_accept(t, vl);
        parse_string(vl, t, t->spec);

        PTOK(pthread_mutex_lock(&t->mtx));
        state = t->state;
        PTOK(pthread_mutex_unlock(&t->mtx));

        if (state == TUNNEL_PAUSED && !vtc_stop)
                parse_string(vl, t, "resume");

        PTOK(pthread_mutex_lock(&t->mtx));
        t->state = TUNNEL_SPEC_DONE;
        PTOK(pthread_cond_signal(&t->cond));
        PTOK(pthread_mutex_unlock(&t->mtx));

        vtc_log(vl, 2, "Ending");
        pthread_cleanup_pop(0);
        vtc_logclose(vl);
        return (NULL);
}

/**********************************************************************
 * Tunnel management
 */

static struct tunnel *
tunnel_new(const char *name)
{
        struct tunnel *t;

        ALLOC_OBJ(t, TUNNEL_MAGIC);
        AN(t);
        REPLACE(t->name, name);
        t->vl = vtc_logopen("%s", name);
        AN(t->vl);

        t->state = TUNNEL_STOPPED;
        bprintf(t->connect, "%s", "${v1_sock}");
        bprintf(t->listen, "%s", default_listen_addr);
        t->csock = -1;
        t->lsock = -1;
        t->asock = -1;
        t->send_lane->rfd = &t->asock;
        t->send_lane->wfd = &t->csock;
        t->recv_lane->rfd = &t->csock;
        t->recv_lane->wfd = &t->asock;
        PTOK(pthread_mutex_init(&t->mtx, NULL));
        PTOK(pthread_cond_init(&t->cond, NULL));
        PTOK(pthread_mutex_lock(&tunnel_mtx));
        VTAILQ_INSERT_TAIL(&tunnels, t, list);
        PTOK(pthread_mutex_unlock(&tunnel_mtx));
        return (t);
}

static void
tunnel_delete(struct tunnel *t)
{

        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        assert(t->asock < 0);
        assert(t->csock < 0);
        if (t->lsock >= 0)
                VTCP_close(&t->lsock);
        macro_undef(t->vl, t->name, "addr");
        macro_undef(t->vl, t->name, "port");
        macro_undef(t->vl, t->name, "sock");
        vtc_logclose(t->vl);
        (void)pthread_mutex_destroy(&t->mtx);
        (void)pthread_cond_destroy(&t->cond);
        free(t->name);
        FREE_OBJ(t);
}

/**********************************************************************
 * Tunnel listen
 */

static void
tunnel_listen(struct tunnel *t)
{
        char buf[vsa_suckaddr_len];
        const struct suckaddr *sua;
        const char *err;

        if (t->lsock >= 0)
                VTCP_close(&t->lsock);
        t->lsock = VTCP_listen_on(t->listen, "0", 1, &err);
        if (err != NULL)
                vtc_fatal(t->vl,
                    "Tunnel listen address (%s) cannot be resolved: %s",
                    t->listen, err);
        assert(t->lsock > 0);
        sua = VSA_getsockname(t->lsock, buf, sizeof buf);
        AN(sua);
        VTCP_name(sua, t->laddr, sizeof t->laddr, t->lport, sizeof t->lport);

        /* Record the actual port, and reuse it on subsequent starts */
        if (VSA_Get_Proto(sua) == AF_INET)
                bprintf(t->listen, "%s:%s", t->laddr, t->lport);
        else
                bprintf(t->listen, "[%s]:%s", t->laddr, t->lport);

        macro_def(t->vl, t->name, "addr", "%s", t->laddr);
        macro_def(t->vl, t->name, "port", "%s", t->lport);
        macro_def(t->vl, t->name, "sock", "%s %s", t->laddr, t->lport);
}

/**********************************************************************
 * Start the tunnel thread
 */

static void
tunnel_start(struct tunnel *t)
{

        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        vtc_log(t->vl, 2, "Starting tunnel");
        tunnel_listen(t);
        vtc_log(t->vl, 1, "Listen on %s", t->listen);
        assert(t->state == TUNNEL_STOPPED);
        t->state = TUNNEL_ACCEPT;
        t->send_lane->buf_len = 0;
        t->send_lane->wrk_len = 0;
        t->recv_lane->buf_len = 0;
        t->recv_lane->wrk_len = 0;
        PTOK(pthread_create(&t->tpoll, NULL, tunnel_poll_thread, t));
        PTOK(pthread_create(&t->tspec, NULL, tunnel_spec_thread, t));
}

static void
tunnel_start_pause(struct tunnel *t)
{

        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        t->start_paused = 1;
        tunnel_start(t);
}

/**********************************************************************
 * Wait for tunnel thread to stop
 */

static void
tunnel_wait(struct tunnel *t)
{
        void *res;

        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        vtc_log(t->vl, 2, "Waiting for tunnel");

        PTOK(pthread_cond_signal(&t->cond));

        PTOK(pthread_join(t->tspec, &res));
        if (res != NULL && !vtc_stop)
                vtc_fatal(t->vl, "Tunnel spec returned \"%p\"", res);

        PTOK(pthread_join(t->tpoll, &res));
        if (res != NULL && !vtc_stop)
                vtc_fatal(t->vl, "Tunnel poll returned \"%p\"", res);

        if (t->csock >= 0)
                VTCP_close(&t->csock);
        if (t->asock >= 0)
                VTCP_close(&t->asock);
        t->tspec = 0;
        t->tpoll = 0;
        t->state = TUNNEL_STOPPED;
}

/**********************************************************************
 * Reap tunnel
 */

static void
tunnel_reset(void)
{
        struct tunnel *t;

        while (1) {
                PTOK(pthread_mutex_lock(&tunnel_mtx));
                t = VTAILQ_FIRST(&tunnels);
                CHECK_OBJ_ORNULL(t, TUNNEL_MAGIC);
                if (t != NULL)
                        VTAILQ_REMOVE(&tunnels, t, list);
                PTOK(pthread_mutex_unlock(&tunnel_mtx));
                if (t == NULL)
                        break;

                if (t->state != TUNNEL_STOPPED)
                        tunnel_wait(t);
                tunnel_delete(t);
        }
}

/**********************************************************************
 * Tunnel command dispatch
 */

void
cmd_tunnel(CMD_ARGS)
{
        struct tunnel *t;

        (void)priv;

        if (av == NULL) {
                /* Reset and free */
                tunnel_reset();
                return;
        }

        AZ(strcmp(av[0], "tunnel"));
        av++;

        VTC_CHECK_NAME(vl, av[0], "Tunnel", 't');

        PTOK(pthread_mutex_lock(&tunnel_mtx));
        VTAILQ_FOREACH(t, &tunnels, list)
                if (!strcmp(t->name, av[0]))
                        break;
        PTOK(pthread_mutex_unlock(&tunnel_mtx));
        if (t == NULL)
                t = tunnel_new(av[0]);
        CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
        av++;

        for (; *av != NULL; av++) {
                if (vtc_error)
                        break;
                if (!strcmp(*av, "-wait")) {
                        if (t->state == TUNNEL_STOPPED)
                                vtc_fatal(t->vl, "Tunnel not -started");
                        tunnel_wait(t);
                        continue;
                }

                /* Don't mess with a running tunnel */
                if (t->state != TUNNEL_STOPPED)
                        tunnel_wait(t);

                assert(t->state == TUNNEL_STOPPED);
                if (!strcmp(*av, "-connect")) {
                        bprintf(t->connect, "%s", av[1]);
                        av++;
                        continue;
                }
                if (!strcmp(*av, "-listen")) {
                        bprintf(t->listen, "%s", av[1]);
                        av++;
                        continue;
                }
                if (!strcmp(*av, "-start")) {
                        tunnel_start(t);
                        continue;
                }
                if (!strcmp(*av, "-start+pause")) {
                        tunnel_start_pause(t);
                        continue;
                }
                if (**av == '-')
                        vtc_fatal(t->vl, "Unknown tunnel argument: %s", *av);
                t->spec = *av;
        }
}

void
init_tunnel(void)
{

        PTOK(pthread_mutex_init(&tunnel_mtx, NULL));
}

		varnish-cache/bin/varnishtest/vtest2/src/vtc_tunnel.c
0		/*-
1		* Copyright (c) 2020 Varnish Software
2		* All rights reserved.
3		*
4		* Author: Dridi Boukelmoune <dridi.boukelmoune@gmail.com>
5		*
6		* SPDX-License-Identifier: BSD-2-Clause
7		*
8		* Redistribution and use in source and binary forms, with or without
9		* modification, are permitted provided that the following conditions
10		* are met:
11		* 1. Redistributions of source code must retain the above copyright
12		* notice, this list of conditions and the following disclaimer.
13		* 2. Redistributions in binary form must reproduce the above copyright
14		* notice, this list of conditions and the following disclaimer in the
15		* documentation and/or other materials provided with the distribution.
16		*
17		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
18		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
21		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
22		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
23		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
24		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
26		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
27		* SUCH DAMAGE.
28		*/
29
30		#include "config.h"
31
32		#include <sys/socket.h>
33		#include <sys/types.h>
34		#include <sys/stat.h>
35
36		#include <errno.h>
37		#include <poll.h>
38		#include <stdio.h>
39		#include <stdlib.h>
40		#include <string.h>
41		#include <unistd.h>
42
43		#include "vtc.h"
44
45		#include "vsa.h"
46		#include "vtcp.h"
47
48		/* SECTION: tunnel tunnel
49		*
50		* The goal of a tunnel is to help control the data transfer between two
51		* parties, for example to trigger socket timeouts in the middle of protocol
52		* frames, without the need to change how both parties are implemented.
53		*
54		* A tunnel accepts a connection and then connects on behalf of the source to
55		* the desired destination. Once both connections are established the tunnel
56		* will transfer bytes unchanged between the source and destination. Transfer
57		* can be interrupted, usually with the help of synchronization methods like
58		* barriers. Once the transfer is paused, it is possible to let a specific
59		* amount of bytes move in either direction.
60		*
61		* SECTION: tunnel.args Arguments
62		*
63		* \-start
64		* Start the tunnel in background, processing the last given
65		* specification.
66		*
67		* \-start+pause
68		* Start the tunnel, but already paused.
69		*
70		* \-wait
71		* Block until the thread finishes.
72		*
73		* \-listen STRING
74		* Dictate the listening socket for the server. STRING is of the form
75		* "IP PORT", or "HOST PORT".
76		*
77		* Listens by defaults to a local random port.
78		*
79		* \-connect STRING
80		* Indicate the server to connect to. STRING is also of the form
81		* "IP PORT", or "HOST PORT".
82		*
83		* Connects by default to a varnish instance called ``v1``.
84		*
85		* SECTION: tunnel.spec Specification
86		*
87		* The specification contains a list of tunnel commands that can be combined
88		* with barriers and delays. For example::
89		*
90		* tunnel t1 {
91		* barrier b1 sync
92		* pause
93		* delay 1
94		* send 42
95		* barrier b2 sync
96		* resume
97		* } -start
98		*
99		* If one end of the tunnel is closed before the end of the specification
100		* the test case will fail. A specification that ends in a paused state will
101		* implicitly resume the tunnel.
102		*/
103
104		enum tunnel_state_e {
105		TUNNEL_ACCEPT,
106		TUNNEL_RUNNING,
107		TUNNEL_PAUSED,
108		TUNNEL_SPEC_DONE,
109		TUNNEL_POLL_DONE,
110		TUNNEL_STOPPED,
111		};
112
113		struct tunnel_lane {
114		char buf[1024];
115		ssize_t buf_len;
116		size_t wrk_len;
117		int *rfd;
118		int *wfd;
119		};
120
121		struct tunnel {
122		unsigned magic;
123		#define TUNNEL_MAGIC 0x7f59913d
124		char *name;
125		struct vtclog *vl;
126		VTAILQ_ENTRY(tunnel) list;
127		enum tunnel_state_e state;
128		unsigned start_paused;
129
130		char *spec;
131
132		char connect[256];
133		int csock;
134
135		char listen[256];
136		int lsock;
137		char laddr[VTCP_ADDRBUFSIZE];
138		char lport[VTCP_PORTBUFSIZE];
139
140		int asock;
141
142		struct tunnel_lane send_lane[1];
143		struct tunnel_lane recv_lane[1];
144
145		pthread_mutex_t mtx; /* state and lanes->_len /
146		pthread_cond_t cond;
147		pthread_t tspec;
148		pthread_t tpoll;
149		};
150
151		static pthread_mutex_t tunnel_mtx;
152
153		static VTAILQ_HEAD(, tunnel) tunnels = VTAILQ_HEAD_INITIALIZER(tunnels);
154
155		/**********************************************************************
156		* Is the tunnel still operating?
157		*/
158
159	2618	static unsigned
160		tunnel_is_open(struct tunnel *t)
161		{
162		unsigned is_open;
163	2618
164	2618	PTOK(pthread_mutex_lock(&t->mtx));
165	2618	is_open = (t->send_lane->buf_len >= 0 && t->recv_lane->buf_len >= 0);
166	2618	PTOK(pthread_mutex_unlock(&t->mtx));
167		return (is_open);
168		}
169
170		/**********************************************************************
171		* SECTION: tunnel.spec.pause
172		*
173		* pause
174		* Wait for in-flight bytes to be transferred and pause the tunnel.
175		*
176		* The tunnel must be running.
177		*/
178
179	240	static void
180		cmd_tunnel_pause(CMD_ARGS)
181		{
182		struct tunnel *t;
183	240
184	240	CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
185		AZ(av[1]);
186	240
187	0	if (!tunnel_is_open(t))
188		vtc_fatal(vl, "Tunnel already closed");
189	240
190	240	PTOK(pthread_mutex_lock(&t->mtx));
191	0	if (t->state == TUNNEL_PAUSED) {
192	0	PTOK(pthread_mutex_unlock(&t->mtx));
193		vtc_fatal(vl, "Tunnel already paused");
194	240	}
195	240	assert(t->state == TUNNEL_RUNNING);
196	240	t->state = TUNNEL_PAUSED;
197	240	PTOK(pthread_cond_signal(&t->cond));
198	240	PTOK(pthread_cond_wait(&t->cond, &t->mtx));
199	240	PTOK(pthread_mutex_unlock(&t->mtx));
200		}
201
202		/**********************************************************************
203		* SECTION: tunnel.spec.send
204		*
205		* send NUMBER
206		* Wait until NUMBER bytes are transferred from source to
207		* destination.
208		*
209		* The tunnel must be paused, it remains paused afterwards.
210		*/
211
212	400	static void
213		cmd_tunnel_send(CMD_ARGS)
214		{
215		struct tunnel *t;
216		unsigned len;
217	400
218	400	CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
219	400	AN(av[1]);
220		AZ(av[2]);
221	400
222		len = atoi(av[1]);
223	400
224	0	if (!tunnel_is_open(t))
225		vtc_fatal(vl, "Tunnel already closed");
226	400
227	400	PTOK(pthread_mutex_lock(&t->mtx));
228	0	if (t->state == TUNNEL_RUNNING) {
229	0	PTOK(pthread_mutex_unlock(&t->mtx));
230		vtc_fatal(vl, "Tunnel still running");
231	400	}
232	400	assert(t->state == TUNNEL_PAUSED);
233	400	AZ(t->send_lane->wrk_len);
234	400	AZ(t->recv_lane->wrk_len);
235	200	if (!strcmp(av[0], "send"))
236		t->send_lane->wrk_len = len;
237	200	else
238	400	t->recv_lane->wrk_len = len;
239	400	PTOK(pthread_cond_signal(&t->cond));
240	400	PTOK(pthread_cond_wait(&t->cond, &t->mtx));
241	400	PTOK(pthread_mutex_unlock(&t->mtx));
242		}
243
244		/**********************************************************************
245		* SECTION: tunnel.spec.recv
246		*
247		* recv NUMBER
248		* Wait until NUMBER bytes are transferred from destination to
249		* source.
250		*
251		* The tunnel must be paused, it remains paused afterwards.
252		*/
253
254	200	static void
255		cmd_tunnel_recv(CMD_ARGS)
256		{
257	200
258	200	cmd_tunnel_send(av, priv, vl);
259		}
260
261		/**********************************************************************
262		* SECTION: tunnel.spec.resume
263		*
264		* resume
265		* Resume the transfer of bytes in both directions.
266		*
267		* The tunnel must be paused.
268		*/
269
270	280	static void
271		cmd_tunnel_resume(CMD_ARGS)
272		{
273		struct tunnel *t;
274	280
275	280	CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
276		AZ(av[1]);
277	280
278	0	if (!tunnel_is_open(t))
279		vtc_fatal(vl, "Tunnel already closed");
280	280
281	280	PTOK(pthread_mutex_lock(&t->mtx));
282	0	if (t->state == TUNNEL_RUNNING) {
283	0	PTOK(pthread_mutex_unlock(&t->mtx));
284		vtc_fatal(vl, "Tunnel already running");
285	280	}
286	280	assert(t->state == TUNNEL_PAUSED);
287	280	t->state = TUNNEL_RUNNING;
288	280	PTOK(pthread_cond_signal(&t->cond));
289	280	PTOK(pthread_mutex_unlock(&t->mtx));
290		}
291
292		static const struct cmds tunnel_cmds[] = {
293		#define CMD_TUNNEL(n) { #n, cmd_tunnel_##n },
294		CMD_TUNNEL(pause)
295		CMD_TUNNEL(send)
296		CMD_TUNNEL(recv)
297		CMD_TUNNEL(resume)
298		#undef CMD_TUNNEL
299		{ NULL, NULL }
300		};
301
302		/**********************************************************************
303		* Tunnel poll thread
304		*/
305
306	3076	static void
307		tunnel_read(struct tunnel t, struct vtclog vl, const struct pollfd *pfd,
308		struct tunnel_lane *lane)
309		{
310		size_t len;
311		ssize_t res;
312		enum tunnel_state_e state;
313	3076
314	3076	assert(pfd->fd == *lane->rfd);
315	1796	if (!(pfd->revents & POLLIN))
316		return;
317	1280
318	1280	PTOK(pthread_mutex_lock(&t->mtx));
319	1280	AZ(lane->buf_len);
320	1280	len = lane->wrk_len;
321	1280	state = t->state;
322		PTOK(pthread_mutex_unlock(&t->mtx));
323	1280
324	400	if (len == 0 && state == TUNNEL_PAUSED)
325		return;
326	880
327	480	if (len == 0 \|\| len > sizeof lane->buf)
328		len = sizeof lane->buf;
329	880
330	880	res = read(pfd->fd, lane->buf, len);
331	0	if (res < 0)
332		vtc_fatal(vl, "Read failed: %s", strerror(errno));
333	880
334	880	PTOK(pthread_mutex_lock(&t->mtx));
335	880	lane->buf_len = (res == 0) ? -1 : res;
336	3076	PTOK(pthread_mutex_unlock(&t->mtx));
337		}
338
339	3076	static void
340		tunnel_write(struct tunnel t, struct vtclog vl, struct tunnel_lane *lane,
341		const char *action)
342		{
343		const char *p;
344		ssize_t res, l;
345	3076
346	3076	p = lane->buf;
347		l = lane->buf_len;
348	3076
349	720	if (l > 0)
350	3796	vtc_log(vl, 3, "%s %zd bytes", action, l);
351	720	while (l > 0) {
352	720	res = write(*lane->wfd, p, l);
353	0	if (res <= 0)
354	720	vtc_fatal(vl, "Write failed: %s", strerror(errno));
355	720	l -= res;
356		p += res;
357		}
358	3076
359	3076	PTOK(pthread_mutex_lock(&t->mtx));
360	800	if (lane->wrk_len > 0 && lane->buf_len != -1) {
361	800	assert(lane->buf_len >= 0);
362	800	assert(lane->wrk_len >= (size_t)lane->buf_len);
363	800	lane->wrk_len -= lane->buf_len;
364	3076	}
365	3076	lane->buf_len = l;
366	3076	PTOK(pthread_mutex_unlock(&t->mtx));
367		}
368
369	160	static void *
370		tunnel_poll_thread(void *priv)
371		{
372		struct tunnel *t;
373		struct vtclog *vl;
374		struct pollfd pfd[2];
375		enum tunnel_state_e state;
376		int res;
377	160
378		CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
379	160
380	160	vl = vtc_logopen("%s", t->name);
381		pthread_cleanup_push(vtc_logclose, vl);
382	1698
383	1538	while (tunnel_is_open(t) && !vtc_stop) {
384		PTOK(pthread_mutex_lock(&t->mtx));
385	1704	/* NB: can be woken up by `tunnel tX -wait` */
386	166	while (t->state == TUNNEL_ACCEPT && !vtc_stop)
387	1538	PTOK(pthread_cond_wait(&t->cond, &t->mtx));
388	1538	state = t->state;
389		PTOK(pthread_mutex_unlock(&t->mtx));
390	1538
391	0	if (vtc_stop)
392		break;
393	1538
394		assert(state < TUNNEL_POLL_DONE);
395	1538
396	1538	memset(pfd, 0, sizeof pfd);
397	1538	pfd[0].fd = *t->send_lane->rfd;
398	1538	pfd[1].fd = *t->recv_lane->rfd;
399	1538	pfd[0].events = POLLIN;
400	1538	pfd[1].events = POLLIN;
401	1538	res = poll(pfd, 2, 100);
402	0	if (res == -1)
403		vtc_fatal(vl, "Poll failed: %s", strerror(errno));
404	1538
405	1538	tunnel_read(t, vl, &pfd[0], t->send_lane);
406		tunnel_read(t, vl, &pfd[1], t->recv_lane);
407	1538
408	1538	PTOK(pthread_mutex_lock(&t->mtx));
409	840	if (t->state == TUNNEL_PAUSED && t->send_lane->wrk_len == 0 &&
410	640	t->recv_lane->wrk_len == 0) {
411	640	AZ(t->send_lane->buf_len);
412	640	AZ(t->recv_lane->buf_len);
413	640	PTOK(pthread_cond_signal(&t->cond));
414	640	PTOK(pthread_cond_wait(&t->cond, &t->mtx));
415	1538	}
416		PTOK(pthread_mutex_unlock(&t->mtx));
417	1538
418	0	if (vtc_stop)
419		break;
420	1538
421	1538	tunnel_write(t, vl, t->send_lane, "Sending");
422		tunnel_write(t, vl, t->recv_lane, "Receiving");
423		}
424	160
425	160	PTOK(pthread_mutex_lock(&t->mtx));
426	0	if (t->state != TUNNEL_SPEC_DONE && !vtc_stop) {
427	0	PTOK(pthread_cond_signal(&t->cond));
428	0	PTOK(pthread_cond_wait(&t->cond, &t->mtx));
429	160	}
430		PTOK(pthread_mutex_unlock(&t->mtx));
431	160
432	160	pthread_cleanup_pop(0);
433	160	vtc_logclose(vl);
434	160	t->state = TUNNEL_POLL_DONE;
435		return (NULL);
436		}
437
438		/**********************************************************************
439		* Tunnel spec thread
440		*/
441
442	160	static void
443		tunnel_accept(struct tunnel t, struct vtclog vl)
444		{
445		struct vsb *vsb;
446		const char addr, err;
447		int afd, cfd;
448	160
449	160	CHECK_OBJ_NOTNULL(t, TUNNEL_MAGIC);
450	160	assert(t->lsock >= 0);
451	160	assert(t->asock < 0);
452	160	assert(t->csock < 0);
453		assert(t->state == TUNNEL_ACCEPT);
454	160
455	160	vtc_log(vl, 4, "Accepting");
456	160	afd = accept(t->lsock, NULL, NULL);
457	0	if (afd < 0)
458	160	vtc_fatal(vl, "Accept failed: %s", strerror(errno));
459		vtc_log(vl, 3, "Accepted socket fd is %d", afd);
460	160
461	160	vsb = macro_expand(vl, t->connect);
462	160	AN(vsb);
463		addr = VSB_data(vsb);
464	160
465	160	cfd = VTCP_open(addr, NULL, 10., &err);
466	0	if (cfd < 0)
467	160	vtc_fatal(vl, "Failed to open %s: %s", addr, err);
468	160	vtc_log(vl, 3, "Connected socket fd is %d", cfd);
469		VSB_destroy(&vsb);
470	160
471	160	VTCP_blocking(afd);
472		VTCP_blocking(cfd);
473	160
474	160	PTOK(pthread_mutex_lock(&t->mtx));
475	160	t->asock = afd;
476	160	t->csock = cfd;
477	160	t->send_lane->buf_len = 0;
478	160	t->send_lane->wrk_len = 0;
479	160	t->recv_lane->buf_len = 0;
480	160	t->recv_lane->wrk_len = 0;
481	40	if (t->start_paused) {
482	40	t->state = TUNNEL_PAUSED;
483	40	t->start_paused = 0;
484	120	} else
485	160	t->state = TUNNEL_RUNNING;
486	160	PTOK(pthread_cond_signal(&t->cond));
487	160	PTOK(pthread_mutex_unlock(&t->mtx));
488		}
489
490	160	static void *
491		tunnel_spec_thread(void *priv)
492		{
493		struct tunnel *t;
494		struct vtclog *vl;
495		enum tunnel_state_e state;
496	160
497	160	CAST_OBJ_NOTNULL(t, priv, TUNNEL_MAGIC);
498		AN(*t->connect);
499	160