GCOV - Varnish Cache


/*-
 * Copyright (c) 2006 Verdens Gang AS
 * Copyright (c) 2006-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.
 *
 */

#include "config.h"

#include <stdlib.h>
#include <netinet/in.h>
#include <netinet/tcp.h>

#include "cache/cache_int.h"

#include "acceptor/cache_acceptor.h"
#include "acceptor/acceptor_priv.h"
#include "acceptor/acceptor_tcp.h"

#include "cache/cache_transport.h"
#include "cache/cache_pool.h"
#include "common/heritage.h"

#include "vcli_serve.h"
#include "vsa.h"
#include "vtcp.h"
#include "vtim.h"

/*--------------------------------------------------------------------
 * TCP options we want to control
 */

static struct sock_opt sock_opts[] = {
        /* Note: Setting the mod counter to something not-zero is needed
         * to force the setsockopt() calls on startup */

        SOCK_OPT(SOL_SOCKET, SO_LINGER, struct linger)
        SOCK_OPT(SOL_SOCKET, SO_KEEPALIVE, int)
        SOCK_OPT(SOL_SOCKET, SO_SNDTIMEO, struct timeval)
        SOCK_OPT(SOL_SOCKET, SO_RCVTIMEO, struct timeval)

        SOCK_OPT(IPPROTO_TCP, TCP_NODELAY, int)

#if defined(HAVE_TCP_KEEP)
        SOCK_OPT(IPPROTO_TCP, TCP_KEEPIDLE, int)
        SOCK_OPT(IPPROTO_TCP, TCP_KEEPCNT, int)
        SOCK_OPT(IPPROTO_TCP, TCP_KEEPINTVL, int)
#elif defined(HAVE_TCP_KEEPALIVE)
        SOCK_OPT(IPPROTO_TCP, TCP_KEEPALIVE, int)
#endif

#undef SOCK_OPT
};

static const int n_sock_opts = vcountof(sock_opts);

/*--------------------------------------------------------------------
 * We want to get out of any kind of trouble-hit TCP connections as fast
 * as absolutely possible, so we set them LINGER disabled, so that even if
 * there are outstanding write data on the socket, a close(2) will return
 * immediately.
 */
static const struct linger disable_so_linger = {
        .l_onoff        =       0,
};

/*
 * We turn on keepalives by default to assist in detecting clients that have
 * hung up on connections returning from waitinglists
 */
static const unsigned enable_so_keepalive = 1;

/* We disable Nagle's algorithm in favor of low latency setups.
 */
static const unsigned enable_tcp_nodelay = 1;

/*--------------------------------------------------------------------
 * Some kernels have bugs/limitations with respect to which options are
 * inherited from the accept/listen socket, so we have to keep track of
 * which, if any, sockopts we have to set on the accepted socket.
 */

static int
vca_tcp_sockopt_init(void)
{
        struct sock_opt *so;
        union sock_arg tmp;
        int n, chg = 0;
        size_t sz;

        memset(&tmp, 0, sizeof tmp);

        for (n = 0; n < n_sock_opts; n++) {
                so = &sock_opts[n];

#define SET_VAL(nm, so, fld, val)                                       \
        do {                                                            \
                if (!vstrcmp(#nm, so->strname)) {                       \
                        assert(so->sz == sizeof so->arg->fld);          \
                        so->arg->fld = (val);                           \
                }                                                       \
        } while (0)

#define NEW_VAL(nm, so, fld, val)                                       \
        do {                                                            \
                if (!vstrcmp(#nm, so->strname)) {                       \
                        sz = sizeof tmp.fld;                            \
                        assert(so->sz == sz);                           \
                        tmp.fld = (val);                                \
                        if (vmemcmp(&so->arg->fld, &(tmp.fld), sz)) {   \
                                vmemcpy(&so->arg->fld, &(tmp.fld), sz); \
                                so->mod++;                              \
                                chg = 1;                                \
                        }                                               \
                }                                                       \
        } while (0)

                SET_VAL(SO_LINGER, so, lg, disable_so_linger);
                SET_VAL(SO_KEEPALIVE, so, i, enable_so_keepalive);
                NEW_VAL(SO_SNDTIMEO, so, tv,
                    VTIM_timeval_sock(cache_param->idle_send_timeout));
                NEW_VAL(SO_RCVTIMEO, so, tv,
                    VTIM_timeval_sock(cache_param->timeout_idle));
                SET_VAL(TCP_NODELAY, so, i, enable_tcp_nodelay);
#if defined(HAVE_TCP_KEEP)
                NEW_VAL(TCP_KEEPIDLE, so, i,
                    (int)cache_param->tcp_keepalive_time);
                NEW_VAL(TCP_KEEPCNT, so, i,
                    (int)cache_param->tcp_keepalive_probes);
                NEW_VAL(TCP_KEEPINTVL, so, i,
                    (int)cache_param->tcp_keepalive_intvl);
#elif defined(HAVE_TCP_KEEPALIVE)
                NEW_VAL(TCP_KEEPALIVE, so, i,
                    (int)cache_param->tcp_keepalive_time);
#endif
        }

        return (chg);
}

static void
vca_tcp_sockopt_test(const struct listen_sock *ls, const struct sess *sp)
{
        struct conn_heritage *ch;
        struct sock_opt *so;
        union sock_arg tmp;
        socklen_t l;
        int i, n;

        CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
        CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);

        for (n = 0; n < n_sock_opts; n++) {
                so = &sock_opts[n];
                ch = &ls->conn_heritage[n];

                if (ch->sess_set) {
                        VSL(SLT_Debug, sp->vxid,
                            "sockopt: Not testing nonhereditary %s for %s=%s",
                            so->strname, ls->name, ls->endpoint);
                        continue;
                }

                memset(&tmp, 0, sizeof tmp);
                l = so->sz;
                i = getsockopt(sp->fd, so->level, so->optname, &tmp, &l);

                if (i == 0 && vmemcmp(&tmp, so->arg, so->sz)) {
                        VSL(SLT_Debug, sp->vxid,
                            "sockopt: Test confirmed %s non heredity for %s=%s",
                            so->strname, ls->name, ls->endpoint);
                        ch->sess_set = 1;
                }

                if (i && errno != ENOPROTOOPT)
                        VTCP_Assert(i);
        }
}

static void
vca_tcp_sockopt_set(const struct listen_sock *ls, const struct sess *sp)
{
        struct conn_heritage *ch;
        struct sock_opt *so;
        vxid_t vxid;
        int n, sock;

        CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);

        if (sp != NULL) {
                CHECK_OBJ(sp, SESS_MAGIC);
                sock = sp->fd;
                vxid = sp->vxid;
        } else {
                sock = ls->sock;
                vxid = NO_VXID;
        }

        for (n = 0; n < n_sock_opts; n++) {
                so = &sock_opts[n];
                ch = &ls->conn_heritage[n];

                if (sp == NULL && ch->listen_mod == so->mod) {
                        VSL(SLT_Debug, vxid,
                            "sockopt: Not setting unmodified %s for %s=%s",
                            so->strname, ls->name, ls->endpoint);
                        continue;
                }

                if  (sp != NULL && !ch->sess_set) {
                        VSL(SLT_Debug, sp->vxid,
                            "sockopt: %s may be inherited for %s=%s",
                            so->strname, ls->name, ls->endpoint);
                        continue;
                }

                VSL(SLT_Debug, vxid,
                    "sockopt: Setting %s for %s=%s",
                    so->strname, ls->name, ls->endpoint);
                VTCP_Assert(setsockopt(sock,
                    so->level, so->optname, so->arg, so->sz));

                if (sp == NULL)
                        ch->listen_mod = so->mod;
        }
}

static void
vca_tcp_init(void)
{

}

static int
vca_tcp_listen(struct cli *cli, struct listen_sock *ls)
{

        CHECK_OBJ_NOTNULL(ls->transport, TRANSPORT_MAGIC);
        assert (ls->sock > 0);  // We know where stdin is

        if (cache_param->tcp_fastopen &&
            VTCP_fastopen(ls->sock, cache_param->listen_depth))
                VSL(SLT_Error, NO_VXID,
                    "Kernel TCP Fast Open: sock=%d, errno=%d %s",
                    ls->sock, errno, VAS_errtxt(errno));

        if (listen(ls->sock, cache_param->listen_depth)) {
                VCLI_SetResult(cli, CLIS_CANT);
                VCLI_Out(cli, "Listen failed on socket '%s': %s",
                    ls->endpoint, VAS_errtxt(errno));
                return (-1);
        }

        AZ(ls->conn_heritage);
        ls->conn_heritage = calloc(n_sock_opts,
            sizeof *ls->conn_heritage);
        AN(ls->conn_heritage);

        ls->test_heritage = 1;
        vca_tcp_sockopt_set(ls, NULL);

        if (cache_param->accept_filter && VTCP_filter_http(ls->sock))
                VSL(SLT_Error, NO_VXID,
                    "Kernel filtering: sock=%d, errno=%d %s",
                    ls->sock, errno, VAS_errtxt(errno));

        return (0);
}

static void
vca_tcp_start(struct cli *cli)
{
        struct listen_sock *ls;

        ASSERT_CLI();

        (void)vca_tcp_sockopt_init();

        VTAILQ_FOREACH(ls, &TCP_acceptor.socks, vcalist) {
                CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);

                if (vca_tcp_listen(cli, ls))
                        return;
        }
}

static void
vca_tcp_event(struct cli *cli, struct listen_sock *ls, enum vca_event event)
{
        char h[VTCP_ADDRBUFSIZE], p[VTCP_PORTBUFSIZE];

        (void) ls; // XXX const?
        switch (event) {
        case VCA_EVENT_LADDR:
                VTCP_myname(ls->sock, h, sizeof h, p, sizeof p);
                VCLI_Out(cli, "%s %s %s\n", ls->name, h, p);
                break;
        default:
                WRONG("INVALID VCA_EVENT");
        }
}

/*--------------------------------------------------------------------
 * The pool-task for a newly accepted session
 *
 * Called from assigned worker thread
 */

static void
vca_mk_tcp(const struct wrk_accept *wa,
    struct sess *sp, char *laddr, char *lport, char *raddr, char *rport)
{
        struct suckaddr *sa = NULL;
        ssize_t sz;

        AN(SES_Reserve_remote_addr(sp, &sa, &sz));
        AN(sa);
        assert(sz == vsa_suckaddr_len);
        AN(VSA_Build(sa, &wa->acceptaddr, wa->acceptaddrlen));
        sp->sattr[SA_CLIENT_ADDR] = sp->sattr[SA_REMOTE_ADDR];

        VTCP_name(sa, raddr, VTCP_ADDRBUFSIZE, rport, VTCP_PORTBUFSIZE);
        AN(SES_Set_String_Attr(sp, SA_CLIENT_IP, raddr));
        AN(SES_Set_String_Attr(sp, SA_CLIENT_PORT, rport));


        AN(SES_Reserve_local_addr(sp, &sa, &sz));
        AN(VSA_getsockname(sp->fd, sa, sz));
        sp->sattr[SA_SERVER_ADDR] = sp->sattr[SA_LOCAL_ADDR];
        VTCP_name(sa, laddr, VTCP_ADDRBUFSIZE, lport, VTCP_PORTBUFSIZE);
}

static void v_matchproto_(task_func_t)
vca_tcp_make_session(struct worker *wrk, void *arg)
{
        char laddr[VTCP_ADDRBUFSIZE];
        char lport[VTCP_PORTBUFSIZE];
        char raddr[VTCP_ADDRBUFSIZE];
        char rport[VTCP_PORTBUFSIZE];
        struct wrk_accept *wa;
        struct sess *sp;
        struct req *req;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CAST_OBJ_NOTNULL(wa, arg, WRK_ACCEPT_MAGIC);

        VTCP_blocking(wa->acceptsock);

        /* Turn accepted socket into a session */
        AN(WS_Reservation(wrk->aws));
        sp = SES_New(wrk->pool);
        CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
        wrk->stats->s_sess++;

        sp->t_open = VTIM_real();
        sp->t_idle = sp->t_open;
        sp->vxid = VXID_Get(wrk, VSL_CLIENTMARKER);

        sp->fd = wa->acceptsock;
        wa->acceptsock = -1;
        sp->listen_sock = wa->acceptlsock;

        assert((size_t)wa->acceptaddrlen <= vsa_suckaddr_len);

        vca_mk_tcp(wa, sp, laddr, lport, raddr, rport);

        AN(wa->acceptlsock->name);
        VSL(SLT_Begin, sp->vxid, "sess 0 %s",
            wa->acceptlsock->transport->name);
        VSL(SLT_SessOpen, sp->vxid, "%s %s %s %s %s %.6f %d",
            raddr, rport, wa->acceptlsock->name, laddr, lport,
            sp->t_open, sp->fd);

        vca_pace_good();
        wrk->stats->sess_conn++;

        if (wa->acceptlsock->test_heritage) {
                vca_tcp_sockopt_test(wa->acceptlsock, sp);
                wa->acceptlsock->test_heritage = 0;
        }

        vca_tcp_sockopt_set(wa->acceptlsock, sp);

        req = Req_New(sp, NULL);
        CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
        req->htc->rfd = &sp->fd;

        SES_SetTransport(wrk, sp, req, wa->acceptlsock->transport);
        WS_Release(wrk->aws, 0);
}

/*--------------------------------------------------------------------
 * This function accepts on a single socket for a single thread pool.
 *
 * As long as we can stick the accepted connection to another thread
 * we do so, otherwise we put the socket back on the "BACK" pool
 * and handle the new connection ourselves.
 */

static void v_matchproto_(task_func_t)
vca_tcp_accept_task(struct worker *wrk, void *arg)
{
        char laddr[VTCP_ADDRBUFSIZE];
        char lport[VTCP_PORTBUFSIZE];
        struct listen_sock *ls;
        struct wrk_accept wa;
        struct poolsock *ps;
        int i;

        CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
        CAST_OBJ_NOTNULL(ps, arg, POOLSOCK_MAGIC);
        ls = ps->lsock;
        CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
        CHECK_OBJ_NOTNULL(ps->pool, POOL_MAGIC);

        /* Return any cached resources from previous task */
        WRK_Cleanup(wrk);

        while (!pool_accepting && ls->sock != -2 && !ps->pool->die)
                VTIM_sleep(.1);

        ps->thread = pthread_self();
        VCA_USR1_Unblock();
        while (ls->sock != -2 && !ps->pool->die) {
                INIT_OBJ(&wa, WRK_ACCEPT_MAGIC);
                wa.acceptlsock = ls;

                vca_pace_check();

                wa.acceptaddrlen = sizeof wa.acceptaddr;
                do {
                        i = accept(ls->sock, (void*)&wa.acceptaddr,
                            &wa.acceptaddrlen);
                } while (i < 0 && errno == EAGAIN && !ps->pool->die);

                if (i < 0 && (ls->sock == -2 || ps->pool->die))
                        break;

                if (i < 0) {
                        switch (errno) {
                        case ECONNABORTED:
                                wrk->stats->sess_fail_econnaborted++;
                                break;
                        case EINTR:
                                wrk->stats->sess_fail_eintr++;
                                break;
                        case EMFILE:
                                wrk->stats->sess_fail_emfile++;
                                vca_pace_bad();
                                break;
                        case EBADF:
                                wrk->stats->sess_fail_ebadf++;
                                vca_pace_bad();
                                break;
                        case ENOBUFS:
                        case ENOMEM:
                                wrk->stats->sess_fail_enomem++;
                                vca_pace_bad();
                                break;
                        default:
                                wrk->stats->sess_fail_other++;
                                vca_pace_bad();
                                break;
                        }

                        i = errno;
                        wrk->stats->sess_fail++;

                        VTCP_myname(ls->sock, laddr, VTCP_ADDRBUFSIZE,
                            lport, VTCP_PORTBUFSIZE);

                        VSL(SLT_SessError, NO_VXID, "%s %s %s %d %d \"%s\"",
                            wa.acceptlsock->name, laddr, lport,
                            ls->sock, i, VAS_errtxt(i));
                        (void)Pool_TrySumstat(wrk);
                        continue;
                }

                wa.acceptsock = i;

                if (!Pool_Task_Arg(wrk, TASK_QUEUE_REQ,
                    vca_tcp_make_session, &wa, sizeof wa)) {
                        ps->thread = cli_thread;
                        /*
                         * We couldn't get another thread, so we will handle
                         * the request in this worker thread, but first we
                         * must reschedule the listening task so it will be
                         * taken up by another thread again.
                         */
                        if (!ps->pool->die) {
                                AZ(Pool_Task(wrk->pool, ps->task,
                                    TASK_QUEUE_VCA));
                                VCA_USR1_Block();
                                return;
                        }
                }
                if (!ps->pool->die && DO_DEBUG(DBG_SLOW_ACCEPTOR))
                        VTIM_sleep(2.0);

        }
        ps->thread = cli_thread;
        VCA_USR1_Block();

        VSL(SLT_Debug, NO_VXID, "XXX Accept thread dies %p", ps);
        Lck_Lock(&ps->pool->mtx);
        VTAILQ_REMOVE(&ps->pool->poolsocks, ps, list);
        Lck_Unlock(&ps->pool->mtx);
        FREE_OBJ(ps);
}

static void
vca_tcp_accept(struct pool *pp)
{
        struct listen_sock *ls;
        struct poolsock *ps;

        VTAILQ_FOREACH(ls, &TCP_acceptor.socks, vcalist) {
                CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);

                ALLOC_OBJ(ps, POOLSOCK_MAGIC);
                AN(ps);
                ps->lsock = ls;
                ps->task->func = vca_tcp_accept_task;
                ps->task->priv = ps;
                ps->pool = pp;
                // we use cli_thread as a marker for "not accepting"
                ps->thread = cli_thread;
                Lck_Lock(&pp->mtx);
                VTAILQ_INSERT_TAIL(&pp->poolsocks, ps, list);
                Lck_Unlock(&pp->mtx);
                AZ(Pool_Task(pp, ps->task, TASK_QUEUE_VCA));
        }
}

static void
vca_tcp_update(struct lock *shut_mtx)
{
        struct listen_sock *ls;

        if (!vca_tcp_sockopt_init())
                return;

        Lck_Lock(shut_mtx);

        VTAILQ_FOREACH(ls, &TCP_acceptor.socks, vcalist) {
                CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);

                if (ls->sock == -2)
                        continue;       // VCA_Shutdown
                assert (ls->sock > 0);
                vca_tcp_sockopt_set(ls, NULL);
                /* If one of the options on a socket has
                 * changed, also force a retest of whether
                 * the values are inherited to the
                 * accepted sockets. This should then
                 * catch any false positives from previous
                 * tests that could happen if the set
                 * value of an option happened to just be
                 * the OS default for that value, and
                 * wasn't actually inherited from the
                 * listening socket. */
                ls->test_heritage = 1;
        }

        Lck_Unlock(shut_mtx);
}

static void
vca_tcp_shutdown(void)
{
        struct listen_sock *ls;
        int i;

        VTAILQ_FOREACH(ls, &TCP_acceptor.socks, vcalist) {
                CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);

                i = ls->sock;
                ls->sock = -2;
                (void)close(i);
        }
}

struct acceptor TCP_acceptor = {
        .magic          = ACCEPTOR_MAGIC,
        .name           = "tcp",
        .config         = vca_tcp_config,
        .init           = vca_tcp_init,
        .open           = vca_tcp_open,
        .reopen         = vca_tcp_reopen,
        .start          = vca_tcp_start,
        .event          = vca_tcp_event,
        .accept         = vca_tcp_accept,
        .update         = vca_tcp_update,
        .shutdown       = vca_tcp_shutdown,
};

		vinyl-cache/bin/vinyld/acceptor/cache_acceptor_tcp.c
0		/*-
1		* Copyright (c) 2006 Verdens Gang AS
2		* Copyright (c) 2006-2015 Varnish Software AS
3		* All rights reserved.
4		*
5		* Author: Poul-Henning Kamp <phk@phk.freebsd.dk>
6		*
7		* SPDX-License-Identifier: BSD-2-Clause
8		*
9		* Redistribution and use in source and binary forms, with or without
10		* modification, are permitted provided that the following conditions
11		* are met:
12		* 1. Redistributions of source code must retain the above copyright
13		* notice, this list of conditions and the following disclaimer.
14		* 2. Redistributions in binary form must reproduce the above copyright
15		* notice, this list of conditions and the following disclaimer in the
16		* documentation and/or other materials provided with the distribution.
17		*
18		* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
19		* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
20		* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
21		* ARE DISCLAIMED. IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
22		* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23		* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
24		* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
25		* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
26		* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
27		* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
28		* SUCH DAMAGE.
29		*
30		*/
31
32		#include "config.h"
33
34		#include <stdlib.h>
35		#include <netinet/in.h>
36		#include <netinet/tcp.h>
37
38		#include "cache/cache_int.h"
39
40		#include "acceptor/cache_acceptor.h"
41		#include "acceptor/acceptor_priv.h"
42		#include "acceptor/acceptor_tcp.h"
43
44		#include "cache/cache_transport.h"
45		#include "cache/cache_pool.h"
46		#include "common/heritage.h"
47
48		#include "vcli_serve.h"
49		#include "vsa.h"
50		#include "vtcp.h"
51		#include "vtim.h"
52
53		/*--------------------------------------------------------------------
54		* TCP options we want to control
55		*/
56
57		static struct sock_opt sock_opts[] = {
58		/* Note: Setting the mod counter to something not-zero is needed
59		* to force the setsockopt() calls on startup */
60
61		SOCK_OPT(SOL_SOCKET, SO_LINGER, struct linger)
62		SOCK_OPT(SOL_SOCKET, SO_KEEPALIVE, int)
63		SOCK_OPT(SOL_SOCKET, SO_SNDTIMEO, struct timeval)
64		SOCK_OPT(SOL_SOCKET, SO_RCVTIMEO, struct timeval)
65
66		SOCK_OPT(IPPROTO_TCP, TCP_NODELAY, int)
67
68		#if defined(HAVE_TCP_KEEP)
69		SOCK_OPT(IPPROTO_TCP, TCP_KEEPIDLE, int)
70		SOCK_OPT(IPPROTO_TCP, TCP_KEEPCNT, int)
71		SOCK_OPT(IPPROTO_TCP, TCP_KEEPINTVL, int)
72		#elif defined(HAVE_TCP_KEEPALIVE)
73		SOCK_OPT(IPPROTO_TCP, TCP_KEEPALIVE, int)
74		#endif
75
76		#undef SOCK_OPT
77		};
78
79		static const int n_sock_opts = vcountof(sock_opts);
80
81		/*--------------------------------------------------------------------
82		* We want to get out of any kind of trouble-hit TCP connections as fast
83		* as absolutely possible, so we set them LINGER disabled, so that even if
84		* there are outstanding write data on the socket, a close(2) will return
85		* immediately.
86		*/
87		static const struct linger disable_so_linger = {
88		.l_onoff = 0,
89		};
90
91		/*
92		* We turn on keepalives by default to assist in detecting clients that have
93		* hung up on connections returning from waitinglists
94		*/
95		static const unsigned enable_so_keepalive = 1;
96
97		/* We disable Nagle's algorithm in favor of low latency setups.
98		*/
99		static const unsigned enable_tcp_nodelay = 1;
100
101		/*--------------------------------------------------------------------
102		* Some kernels have bugs/limitations with respect to which options are
103		* inherited from the accept/listen socket, so we have to keep track of
104		* which, if any, sockopts we have to set on the accepted socket.
105		*/
106
107		static int
108	61487	vca_tcp_sockopt_init(void)
109		{
110		struct sock_opt *so;
111		union sock_arg tmp;
112	61487	int n, chg = 0;
113		size_t sz;
114
115	61487	memset(&tmp, 0, sizeof tmp);
116
117	553383	for (n = 0; n < n_sock_opts; n++) {
118	491896	so = &sock_opts[n];
119
120		#define SET_VAL(nm, so, fld, val) \
121		do { \
122		if (!vstrcmp(#nm, so->strname)) { \
123		assert(so->sz == sizeof so->arg->fld); \
124		so->arg->fld = (val); \
125		} \
126		} while (0)
127
128		#define NEW_VAL(nm, so, fld, val) \
129		do { \
130		if (!vstrcmp(#nm, so->strname)) { \
131		sz = sizeof tmp.fld; \
132		assert(so->sz == sz); \
133		tmp.fld = (val); \
134		if (vmemcmp(&so->arg->fld, &(tmp.fld), sz)) { \
135		vmemcpy(&so->arg->fld, &(tmp.fld), sz); \
136		so->mod++; \
137		chg = 1; \
138		} \
139		} \
140		} while (0)
141
142	491896	SET_VAL(SO_LINGER, so, lg, disable_so_linger);
143	491896	SET_VAL(SO_KEEPALIVE, so, i, enable_so_keepalive);
144	491896	NEW_VAL(SO_SNDTIMEO, so, tv,
145		VTIM_timeval_sock(cache_param->idle_send_timeout));
146	491896	NEW_VAL(SO_RCVTIMEO, so, tv,
147		VTIM_timeval_sock(cache_param->timeout_idle));
148	491896	SET_VAL(TCP_NODELAY, so, i, enable_tcp_nodelay);
149		#if defined(HAVE_TCP_KEEP)
150	491896	NEW_VAL(TCP_KEEPIDLE, so, i,
151		(int)cache_param->tcp_keepalive_time);
152	491896	NEW_VAL(TCP_KEEPCNT, so, i,
153		(int)cache_param->tcp_keepalive_probes);
154	491896	NEW_VAL(TCP_KEEPINTVL, so, i,
155		(int)cache_param->tcp_keepalive_intvl);
156		#elif defined(HAVE_TCP_KEEPALIVE)
157		NEW_VAL(TCP_KEEPALIVE, so, i,
158		(int)cache_param->tcp_keepalive_time);
159		#endif
160	491896	}
161
162	61487	return (chg);
163		}
164
165		static void
166	18951	vca_tcp_sockopt_test(const struct listen_sock ls, const struct sess sp)
167		{
168		struct conn_heritage *ch;
169		struct sock_opt *so;
170		union sock_arg tmp;
171		socklen_t l;
172		int i, n;
173
174	18951	CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
175	18951	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
176
177	170506	for (n = 0; n < n_sock_opts; n++) {
178	151639	so = &sock_opts[n];
179	151639	ch = &ls->conn_heritage[n];
180
181	151639	if (ch->sess_set) {
182	1244	VSL(SLT_Debug, sp->vxid,
183		"sockopt: Not testing nonhereditary %s for %s=%s",
184	622	so->strname, ls->name, ls->endpoint);
185	622	continue;
186		}
187
188	151017	memset(&tmp, 0, sizeof tmp);
189	151017	l = so->sz;
190	151017	i = getsockopt(sp->fd, so->level, so->optname, &tmp, &l);
191
192	151017	if (i == 0 && vmemcmp(&tmp, so->arg, so->sz)) {
193	74560	VSL(SLT_Debug, sp->vxid,
194		"sockopt: Test confirmed %s non heredity for %s=%s",
195	37280	so->strname, ls->name, ls->endpoint);
196	37280	ch->sess_set = 1;
197	37280	}
198
199	151017	if (i && errno != ENOPROTOOPT)
200	0	VTCP_Assert(i);
201	150933	}
202	18867	}
203
204		static void
205	62415	vca_tcp_sockopt_set(const struct listen_sock ls, const struct sess sp)
206		{
207		struct conn_heritage *ch;
208		struct sock_opt *so;
209		vxid_t vxid;
210		int n, sock;
211
212	62415	CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
213
214	62415	if (sp != NULL) {
215	42000	CHECK_OBJ(sp, SESS_MAGIC);
216	42000	sock = sp->fd;
217	42000	vxid = sp->vxid;
218	42000	} else {
219	20415	sock = ls->sock;
220	20415	vxid = NO_VXID;
221		}
222
223	561689	for (n = 0; n < n_sock_opts; n++) {
224	499274	so = &sock_opts[n];
225	499274	ch = &ls->conn_heritage[n];
226
227	499274	if (sp == NULL && ch->listen_mod == so->mod) {
228	609	VSL(SLT_Debug, vxid,
229		"sockopt: Not setting unmodified %s for %s=%s",
230	609	so->strname, ls->name, ls->endpoint);
231	609	continue;
232		}
233
234	498665	if (sp != NULL && !ch->sess_set) {
235	503918	VSL(SLT_Debug, sp->vxid,
236		"sockopt: %s may be inherited for %s=%s",
237	251959	so->strname, ls->name, ls->endpoint);
238	251959	continue;
239		}
240
241	246706	VSL(SLT_Debug, vxid,
242		"sockopt: Setting %s for %s=%s",
243	246706	so->strname, ls->name, ls->endpoint);
244	246706	VTCP_Assert(setsockopt(sock,
245		so->level, so->optname, so->arg, so->sz));
246
247	246706	if (sp == NULL)
248	162711	ch->listen_mod = so->mod;
249	246706	}
250	62415	}
251
252		static void
253	20884	vca_tcp_init(void)
254		{
255
256	20884	}
257
258		static int
259	20328	vca_tcp_listen(struct cli cli, struct listen_sock ls)
260		{
261
262	20328	CHECK_OBJ_NOTNULL(ls->transport, TRANSPORT_MAGIC);
263	20328	assert (ls->sock > 0); // We know where stdin is
264
265	20328	if (cache_param->tcp_fastopen &&
266	0	VTCP_fastopen(ls->sock, cache_param->listen_depth))
267	0	VSL(SLT_Error, NO_VXID,
268		"Kernel TCP Fast Open: sock=%d, errno=%d %s",
269	0	ls->sock, errno, VAS_errtxt(errno));
270
271	20328	if (listen(ls->sock, cache_param->listen_depth)) {
272	0	VCLI_SetResult(cli, CLIS_CANT);
273	0	VCLI_Out(cli, "Listen failed on socket '%s': %s",
274	0	ls->endpoint, VAS_errtxt(errno));
275	0	return (-1);
276		}
277
278	20328	AZ(ls->conn_heritage);
279	20328	ls->conn_heritage = calloc(n_sock_opts,
280		sizeof *ls->conn_heritage);
281	20328	AN(ls->conn_heritage);
282
283	20328	ls->test_heritage = 1;
284	20328	vca_tcp_sockopt_set(ls, NULL);
285
286	20328	if (cache_param->accept_filter && VTCP_filter_http(ls->sock))
287	40446	VSL(SLT_Error, NO_VXID,
288		"Kernel filtering: sock=%d, errno=%d %s",
289	20223	ls->sock, errno, VAS_errtxt(errno));
290
291	20328	return (0);
292	20328	}
293
294		static void
295	20790	vca_tcp_start(struct cli *cli)
296		{
297		struct listen_sock *ls;
298
299	20790	ASSERT_CLI();
300
301	20790	(void)vca_tcp_sockopt_init();
302
303	41118	VTAILQ_FOREACH(ls, &TCP_acceptor.socks, vcalist) {
304	20328	CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
305
306	20328	if (vca_tcp_listen(cli, ls))
307	0	return;
308	20328	}
309	20790	}
310
311		static void
312	40299	vca_tcp_event(struct cli cli, struct listen_sock ls, enum vca_event event)
313		{
314		char h[VTCP_ADDRBUFSIZE], p[VTCP_PORTBUFSIZE];
315
316	40299	(void) ls; // XXX const?
317	40299	switch (event) {
318		case VCA_EVENT_LADDR:
319	40299	VTCP_myname(ls->sock, h, sizeof h, p, sizeof p);
320	40299	VCLI_Out(cli, "%s %s %s\n", ls->name, h, p);
321	40299	break;
322		default:
323	0	WRONG("INVALID VCA_EVENT");
324	0	}
325	40299	}
326
327		/*--------------------------------------------------------------------
328		* The pool-task for a newly accepted session
329		*
330		* Called from assigned worker thread
331		*/
332
333		static void
334	42000	vca_mk_tcp(const struct wrk_accept *wa,
335		struct sess sp, char laddr, char lport, char raddr, char *rport)
336		{
337	42000	struct suckaddr *sa = NULL;
338		ssize_t sz;
339
340	42000	AN(SES_Reserve_remote_addr(sp, &sa, &sz));
341	42000	AN(sa);
342	42000	assert(sz == vsa_suckaddr_len);
343	42000	AN(VSA_Build(sa, &wa->acceptaddr, wa->acceptaddrlen));
344	42000	sp->sattr[SA_CLIENT_ADDR] = sp->sattr[SA_REMOTE_ADDR];
345
346	42000	VTCP_name(sa, raddr, VTCP_ADDRBUFSIZE, rport, VTCP_PORTBUFSIZE);
347	42000	AN(SES_Set_String_Attr(sp, SA_CLIENT_IP, raddr));
348	42000	AN(SES_Set_String_Attr(sp, SA_CLIENT_PORT, rport));
349
350
351	42000	AN(SES_Reserve_local_addr(sp, &sa, &sz));
352	42000	AN(VSA_getsockname(sp->fd, sa, sz));
353	42000	sp->sattr[SA_SERVER_ADDR] = sp->sattr[SA_LOCAL_ADDR];
354	42000	VTCP_name(sa, laddr, VTCP_ADDRBUFSIZE, lport, VTCP_PORTBUFSIZE);
355	42000	}
356
357		static void v_matchproto_(task_func_t)
358	42002	vca_tcp_make_session(struct worker wrk, void arg)
359		{
360		char laddr[VTCP_ADDRBUFSIZE];
361		char lport[VTCP_PORTBUFSIZE];
362		char raddr[VTCP_ADDRBUFSIZE];
363		char rport[VTCP_PORTBUFSIZE];
364		struct wrk_accept *wa;
365		struct sess *sp;
366		struct req *req;
367
368	42002	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
369	42002	CAST_OBJ_NOTNULL(wa, arg, WRK_ACCEPT_MAGIC);
370
371	42002	VTCP_blocking(wa->acceptsock);
372
373		/* Turn accepted socket into a session */
374	42002	AN(WS_Reservation(wrk->aws));
375	42002	sp = SES_New(wrk->pool);
376	42002	CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
377	42002	wrk->stats->s_sess++;
378
379	42002	sp->t_open = VTIM_real();
380	42002	sp->t_idle = sp->t_open;
381	42002	sp->vxid = VXID_Get(wrk, VSL_CLIENTMARKER);
382
383	42002	sp->fd = wa->acceptsock;
384	42002	wa->acceptsock = -1;
385	42002	sp->listen_sock = wa->acceptlsock;
386
387	42002	assert((size_t)wa->acceptaddrlen <= vsa_suckaddr_len);
388
389	42002	vca_mk_tcp(wa, sp, laddr, lport, raddr, rport);
390
391	42002	AN(wa->acceptlsock->name);
392	84004	VSL(SLT_Begin, sp->vxid, "sess 0 %s",
393	42002	wa->acceptlsock->transport->name);
394	84004	VSL(SLT_SessOpen, sp->vxid, "%s %s %s %s %s %.6f %d",
395	42002	raddr, rport, wa->acceptlsock->name, laddr, lport,
396	42002	sp->t_open, sp->fd);
397
398	42002	vca_pace_good();
399	42002	wrk->stats->sess_conn++;
400
401	42002	if (wa->acceptlsock->test_heritage) {
402	18951	vca_tcp_sockopt_test(wa->acceptlsock, sp);
403	18951	wa->acceptlsock->test_heritage = 0;
404	18951	}
405
406	42002	vca_tcp_sockopt_set(wa->acceptlsock, sp);
407
408	42002	req = Req_New(sp, NULL);
409	42002	CHECK_OBJ_NOTNULL(req, REQ_MAGIC);
410	42002	req->htc->rfd = &sp->fd;
411
412	42002	SES_SetTransport(wrk, sp, req, wa->acceptlsock->transport);
413	42002	WS_Release(wrk->aws, 0);
414	42002	}
415
416		/*--------------------------------------------------------------------
417		* This function accepts on a single socket for a single thread pool.
418		*
419		* As long as we can stick the accepted connection to another thread
420		* we do so, otherwise we put the socket back on the "BACK" pool
421		* and handle the new connection ourselves.
422		*/
423
424		static void v_matchproto_(task_func_t)
425	40690	vca_tcp_accept_task(struct worker wrk, void arg)
426		{
427		char laddr[VTCP_ADDRBUFSIZE];
428		char lport[VTCP_PORTBUFSIZE];
429		struct listen_sock *ls;
430		struct wrk_accept wa;
431		struct poolsock *ps;
432		int i;
433
434	40690	CHECK_OBJ_NOTNULL(wrk, WORKER_MAGIC);
435	40690	CAST_OBJ_NOTNULL(ps, arg, POOLSOCK_MAGIC);
436	40690	ls = ps->lsock;
437	40690	CHECK_OBJ_NOTNULL(ls, LISTEN_SOCK_MAGIC);
438	40690	CHECK_OBJ_NOTNULL(ps->pool, POOL_MAGIC);
439
440		/* Return any cached resources from previous task */
441	40690	WRK_Cleanup(wrk);
442
443	77428	while (!pool_accepting && ls->sock != -2 && !ps->pool->die)
444	36738	VTIM_sleep(.1);
445
446	40690	ps->thread = pthread_self();
447	40690	VCA_USR1_Unblock();
448	82658	while (ls->sock != -2 && !ps->pool->die) {
449	81965	INIT_OBJ(&wa, WRK_ACCEPT_MAGIC);
450	81965	wa.acceptlsock = ls;
451
452	81965	vca_pace_check();
453
454	81965	wa.acceptaddrlen = sizeof wa.acceptaddr;
455	81965	do {
456	164680	i = accept(ls->sock, (void*)&wa.acceptaddr,
457	82340	&wa.acceptaddrlen);
458	82340	} while (i < 0 && errno == EAGAIN && !ps->pool->die);
459
460	81965	if (i < 0 && (ls->sock == -2 \|\| ps->pool->die))
461	39963	break;
462
463	42002	if (i < 0) {
464	0	switch (errno) {
465		case ECONNABORTED:
466	0	wrk->stats->sess_fail_econnaborted++;
467	0	break;
468		case EINTR:
469	0	wrk->stats->sess_fail_eintr++;
470	0	break;
471		case EMFILE:
472	0	wrk->stats->sess_fail_emfile++;
473	0	vca_pace_bad();
474	0	break;
475		case EBADF:
476	0	wrk->stats->sess_fail_ebadf++;
477	0	vca_pace_bad();
478	0	break;
479		case ENOBUFS:
480		case ENOMEM:
481	0	wrk->stats->sess_fail_enomem++;
482	0	vca_pace_bad();
483	0	break;
484		default:
485	0	wrk->stats->sess_fail_other++;
486	0	vca_pace_bad();
487	0	break;
488		}
489
490	0	i = errno;
491	0	wrk->stats->sess_fail++;
492
493	0	VTCP_myname(ls->sock, laddr, VTCP_ADDRBUFSIZE,
494	0	lport, VTCP_PORTBUFSIZE);
495
496	0	VSL(SLT_SessError, NO_VXID, "%s %s %s %d %d \"%s\"",
497	0	wa.acceptlsock->name, laddr, lport,
498	0	ls->sock, i, VAS_errtxt(i));
499	0	(void)Pool_TrySumstat(wrk);