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/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) { CMDS_MAGIC, #n, cmd_tunnel_##n },
        CMD_TUNNEL(pause)
        CMD_TUNNEL(send)
        CMD_TUNNEL(recv)
        CMD_TUNNEL(resume)
#undef CMD_TUNNEL
        { CMDS_MAGIC, 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));
}

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