15920993143

Committed 27 Jun 2025 07:42AM UTC coverage: 87.418%. First build

Build # 15920993143

Build Type

Pull #11607

github

Committed by

web-flow

Commit Message

Merge eca61e599 into 504668286

Pull Request Pull Request #11607: [backport 3.2] core: bind to all addresses

Run Details

68963 of 122267 branches covered (56.4%)

76 of 83 new or added lines in 1 file covered. (91.57%)

101652 of 116283 relevant lines covered (87.42%)

1955861.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.91

/src/lib/core/evio.c

/*
 * Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
 *
 * 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 <COPYRIGHT HOLDER> ``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
 * <COPYRIGHT HOLDER> 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 "evio.h"
#include <stdio.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>

#include <trivia/util.h>
#include "exception.h"
#include "iostream.h"
#include "tt_strerror.h"
#include "uri/uri.h"

struct evio_service_entry {
        /** Bind URI */
        struct uri uri;
        /** Interface/port to bind to */
        union {
                struct sockaddr addr;
                struct sockaddr_storage addrstorage;
        };
        socklen_t addr_len;
        /** IO stream context. */
        struct iostream_ctx io_ctx;
        /** libev io object for the acceptor socket. */
        struct ev_io ev;
        /** Pointer to the root evio_service, which contains this object */
        struct evio_service *service;
        /** Link to other entries */
        struct rlist link;
};

static int
evio_setsockopt_keepalive(int fd)
{
        int on = 1;
        /*
         * SO_KEEPALIVE to ensure connections don't hang
         * around for too long when a link goes away.
         */
        if (sio_setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
                       &on, sizeof(on)))
                return -1;
#ifdef __linux__
        /*
         * On Linux, we are able to fine-tune keepalive
         * intervals. Set smaller defaults, since the system-wide
         * defaults are in days.
         */
        int keepcnt = 5;
        if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &keepcnt,
                       sizeof(int)))
                return -1;
        int keepidle = 30;

        if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &keepidle,
                       sizeof(int)))
                return -1;

        int keepintvl = 60;
        if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &keepintvl,
                       sizeof(int)))
                return -1;
#endif
        return 0;
}

/** Set common client socket options. */
int
evio_setsockopt_client(int fd, int family, int type)
{
        int on = 1;
        /* In case this throws, the socket is not leaked. */
        if (sio_setfl(fd, O_NONBLOCK, on))
                return -1;
        if (type == SOCK_STREAM && family != AF_UNIX) {
                /*
                 * SO_KEEPALIVE to ensure connections don't hang
                 * around for too long when a link goes away.
                 */
                if (evio_setsockopt_keepalive(fd) != 0)
                        return -1;
                /*
                 * Lower latency is more important than higher
                 * bandwidth, and we usually write entire
                 * request/response in a single syscall.
                 */
                if (sio_setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
                                   &on, sizeof(on)))
                        return -1;
        }
        return 0;
}

int
evio_setsockopt_server(int fd, int family, int type)
{
        int on = 1;
        /* In case this throws, the socket is not leaked. */
        if (sio_setfl(fd, O_NONBLOCK, on))
                return -1;
        /* Allow reuse local adresses. */
        if (sio_setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,
                       &on, sizeof(on)))
                return -1;
        /*
         * Allow binding to the same port for ipv4 and ipv6 simultaneously on
         * platforms with dualstack sockets support.
         */
        if (family == AF_INET6 && sio_setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,
                                                 &on, sizeof(on)))
                return -1;

#ifndef TARANTOOL_WSL1_WORKAROUND_ENABLED
        /* Send all buffered messages on socket before take
         * control out from close(2) or shutdown(2). */
        struct linger linger = { 0, 0 };

        if (sio_setsockopt(fd, SOL_SOCKET, SO_LINGER,
                       &linger, sizeof(linger)))
                return -1;
#endif
        if (type == SOCK_STREAM && family != AF_UNIX &&
            evio_setsockopt_keepalive(fd) != 0)
                return -1;
        return 0;
}

static inline const char *
evio_service_name(struct evio_service *service)
{
        return service->name;
}

static void
evio_service_entry_detach(struct evio_service_entry *entry);

static void
evio_service_entry_stop(struct evio_service_entry *entry);

/**
 * A callback invoked by libev when acceptor socket is ready.
 * Accept the socket, initialize it and pass to the on_accept
 * callback.
 */
static void
evio_service_entry_accept_cb(ev_loop *loop, ev_io *watcher, int events)
{
        (void) loop;
        (void) events;
        struct evio_service_entry *entry =
                (struct evio_service_entry *)watcher->data;
        int fd;
        while (1) {
                /*
                 * Accept all pending connections from backlog during event
                 * loop iteration. Significally speed up acceptor with enabled
                 * io_collect_interval.
                 */
                struct sockaddr_storage addr;
                socklen_t addrlen = sizeof(addr);
                fd = sio_accept(entry->ev.fd, (struct sockaddr *)&addr,
                                &addrlen);

                if (fd < 0) {
                        if (! sio_wouldblock(errno))
                                break;
                        return;
                }
                if (evio_setsockopt_client(fd, entry->addr.sa_family,
                                           SOCK_STREAM) != 0)
                        break;
                struct iostream io;
                if (iostream_create(&io, fd, &entry->io_ctx) != 0)
                        break;
                entry->service->on_accept(entry->service, &io,
                                          (struct sockaddr *)&addr, addrlen);
                /* Must be moved by the callback. */
                assert(!iostream_is_initialized(&io));
        }
        if (fd >= 0)
                close(fd);
        diag_log();
}

/*
 * Check if the UNIX socket exists and no one is
 * listening on it. Unlink the file if it's the case.
 */
static int
evio_service_entry_reuse_addr(const struct uri *u)
{
        if (u->host != NULL && strcmp(u->host, URI_HOST_UNIX) != 0)
                return 0;

        struct sockaddr_un un = { 0 };
        assert(u->service != NULL);
        strlcpy(un.sun_path, u->service, sizeof(un.sun_path));
        un.sun_family = AF_UNIX;

        int cl_fd = sio_socket(un.sun_family, SOCK_STREAM, 0);
        if (cl_fd < 0)
                return -1;

        if (connect(cl_fd, (struct sockaddr *)&un, sizeof(un)) == 0)
                goto err;

        if (errno == ECONNREFUSED && unlink(un.sun_path) != 0)
                goto err;

        close(cl_fd);
        return 0;
err:
        errno = EADDRINUSE;
        diag_set(SocketError, sio_socketname(cl_fd), "unlink");
        close(cl_fd);
        return -1;
}

/**
 * Try to bind on the configured port.
 *
 * Throws an exception if error.
 */
static int
evio_service_entry_bind_addr(struct evio_service_entry *entry)
{
        say_debug("%s: binding to %s...",
                  evio_service_name(entry->service),
                  sio_strfaddr(&entry->addr, entry->addr_len));
        /* Create a socket. */
        int fd = sio_socket(entry->addr.sa_family,
                            SOCK_STREAM, IPPROTO_TCP);
        if (fd < 0)
                return -1;

        if (evio_setsockopt_server(fd, entry->addr.sa_family,
                                   SOCK_STREAM) != 0)
                goto error;

        if (sio_bind(fd, &entry->addr, entry->addr_len) != 0)
                goto error;

        /*
         * After binding a result address may be different. For
         * example, if a port was 0.
         */
        if (sio_getsockname(fd, &entry->addr, &entry->addr_len) != 0)
                goto error;

        say_info("%s: bound to %s",
                 evio_service_name(entry->service),
                 sio_strfaddr(&entry->addr, entry->addr_len));

        /* Register the socket in the event loop. */
        ev_io_set(&entry->ev, fd, EV_READ);
        return 0;
error:
        close(fd);
        return -1;
}

/**
 * Listen on bounded port.
 *
 * @retval 0 for success
 */
static int
evio_service_entry_listen(struct evio_service_entry *entry)
{
        say_debug("%s: listening on %s...",
                  evio_service_name(entry->service),
                  sio_strfaddr(&entry->addr, entry->addr_len));

        int fd = entry->ev.fd;
        if (sio_listen(fd))
                return -1;
        if (entry->service->on_accept != NULL)
                ev_io_start(entry->service->loop, &entry->ev);
        return 0;
}

static void
evio_service_entry_create(struct evio_service_entry *entry,
                          struct evio_service *service)
{
        uri_create(&entry->uri, NULL);
        memset(&entry->addrstorage, 0, sizeof(entry->addrstorage));
        entry->addr_len = 0;
        iostream_ctx_clear(&entry->io_ctx);
        /*
         * Initialize libev objects to be able to detect if they
         * are active or not in evio_service_entry_stop().
         */
        ev_init(&entry->ev, evio_service_entry_accept_cb);
        ev_io_set(&entry->ev, -1, 0);
        entry->ev.data = entry;
        entry->service = service;
        rlist_create(&entry->link);
}

/**
 * Add new entry to the service.
 */
static void
evio_service_add_entry(struct evio_service *dst,
                       struct evio_service_entry *entry)
{
        rlist_add_entry(&dst->entries, entry, link);
        ++dst->entry_count;
}

/**
 * Remove the entry from service and free the memory.
 */
static void
evio_service_delete_entry(struct evio_service_entry *entry)
{
        struct evio_service *service = entry->service;
        --service->entry_count;
        rlist_del_entry(entry, link);
        free(entry);
}

/**
 * Try to bind.
 */
static int
evio_service_bind_uri(struct evio_service *service, const struct uri *u)
{
        assert(u->service != NULL);
        if (u->host != NULL && strcmp(u->host, URI_HOST_UNIX) == 0) {
                struct evio_service_entry *entry =
                        xmalloc(sizeof(struct evio_service_entry));
                evio_service_entry_create(entry, service);
                evio_service_add_entry(service, entry);
                assert(!ev_is_active(&entry->ev));

                if (iostream_ctx_create(&entry->io_ctx,
                                        IOSTREAM_SERVER, u) != 0) {
                        evio_service_delete_entry(entry);
                        return -1;
                }

                uri_destroy(&entry->uri);
                uri_copy(&entry->uri, u);

                /* UNIX domain socket */
                struct sockaddr_un *un = (struct sockaddr_un *)&entry->addr;
                entry->addr_len = sizeof(*un);
                strlcpy(un->sun_path, u->service, sizeof(un->sun_path));
                un->sun_family = AF_UNIX;
                return evio_service_entry_bind_addr(entry);
        }
        /* IP socket */
        struct addrinfo hints, *res;
        memset(&hints, 0, sizeof(hints));
        hints.ai_family = AF_UNSPEC;
        hints.ai_socktype = SOCK_STREAM;
        hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;

        if (getaddrinfo(u->host, u->service, &hints, &res) != 0 ||
            res == NULL) {
                diag_set(SocketError, sio_socketname(-1),
                         "can't resolve uri for bind");
                return -1;
        }
        for (struct addrinfo *ai = res; ai != NULL; ai = ai->ai_next) {
                struct evio_service_entry *iter_entry;
                rlist_foreach_entry(iter_entry, &service->entries,
                                    link) {
                        if (iter_entry->addr_len == ai->ai_addrlen &&
                            memcmp(&iter_entry->addr, ai->ai_addr,
                                   iter_entry->addr_len) == 0) {
                                say_debug("Duplicate address %s returned by"
                                          " getaddrinfo is omitted",
                                          sio_strfaddr(ai->ai_addr,
                                                       ai->ai_addrlen));
                                goto next;
                        }
                }
                struct evio_service_entry *entry =
                        xmalloc(sizeof(struct evio_service_entry));
                evio_service_entry_create(entry, service);
                evio_service_add_entry(service, entry);
                assert(!ev_is_active(&entry->ev));

                if (iostream_ctx_create(&entry->io_ctx,
                                        IOSTREAM_SERVER, u) != 0) {
                        evio_service_entry_stop(entry);
                        evio_service_delete_entry(entry);
                        freeaddrinfo(res);
                        return -1;
                }

                uri_destroy(&entry->uri);
                uri_copy(&entry->uri, u);
                memcpy(&entry->addr, ai->ai_addr, ai->ai_addrlen);
                entry->addr_len = ai->ai_addrlen;
                if (evio_service_entry_bind_addr(entry) != 0) {
                        say_error("%s: failed to bind on %s: %s",
                                  evio_service_name(entry->service),
                                  sio_strfaddr(ai->ai_addr, ai->ai_addrlen),
                                  diag_last_error(diag_get())->errmsg);
                        evio_service_entry_stop(entry);
                        evio_service_delete_entry(entry);
                        freeaddrinfo(res);
                        diag_set(SocketError, sio_socketname(-1),
                                 "%s: failed to bind",
                                 evio_service_name(service));
                        return -1;
                }
next:
                ;
        }
        freeaddrinfo(res);
        return 0;
}

/** Destroy the entry uri, ctx and stop everything related to ev. **/
static void
evio_service_entry_detach(struct evio_service_entry *entry)
{
        iostream_ctx_destroy(&entry->io_ctx);
        if (ev_is_active(&entry->ev)) {
                ev_io_stop(entry->service->loop, &entry->ev);
                entry->addr_len = 0;
        }
        ev_io_set(&entry->ev, -1, 0);
        uri_destroy(&entry->uri);
}

/** It's safe to stop a service entry which is not started yet. */
static void
evio_service_entry_stop(struct evio_service_entry *entry)
{
        int service_fd = entry->ev.fd;
        evio_service_entry_detach(entry);
        if (service_fd < 0)
                return;

        if (close(service_fd) < 0)
                say_error("Failed to close socket: %s", tt_strerror(errno));

        if (entry->addr.sa_family != AF_UNIX)
                return;

        if (unlink(((struct sockaddr_un *)&entry->addr)->sun_path) < 0) {
                say_error("Failed to unlink unix "
                          "socket path: %s", tt_strerror(errno));
        }
}

/**
 * Updates dst entry socket settings according to ones in the src.
 */
static void
evio_service_entry_attach(struct evio_service_entry *dst,
                          const struct evio_service_entry *src)
{
        assert(!ev_is_active(&dst->ev));
        uri_destroy(&dst->uri);
        uri_copy(&dst->uri, &src->uri);
        dst->addrstorage = src->addrstorage;
        dst->addr_len = src->addr_len;
        iostream_ctx_copy(&dst->io_ctx, &src->io_ctx);
        ev_io_set(&dst->ev, src->ev.fd, EV_READ);
        ev_io_start(dst->service->loop, &dst->ev);
}

/** Recreate the IO stream contexts from the service entry URI. */
static int
evio_service_entry_reload_uri(struct evio_service_entry *entry)
{
        struct iostream_ctx io_ctx;
        if (iostream_ctx_create(&io_ctx, IOSTREAM_SERVER, &entry->uri) != 0)
                return -1;
        iostream_ctx_destroy(&entry->io_ctx);
        iostream_ctx_move(&entry->io_ctx, &io_ctx);
        return 0;
}

static inline int
evio_service_reuse_addr(const struct uri_set *uri_set)
{
        for (int i = 0; i < uri_set->uri_count; i++) {
                const struct uri *uri = &uri_set->uris[i];
                if (evio_service_entry_reuse_addr(uri) != 0)
                        return -1;
        }
        return 0;
}

int
evio_service_count(const struct evio_service *service)
{
        return service->entry_count;
}

const struct sockaddr *
evio_service_addr(const struct evio_service *service, int idx, socklen_t *size)
{
        assert(idx < service->entry_count && idx >= 0);
        struct evio_service_entry *entry;
        int i = 0;
        rlist_foreach_entry(entry, &service->entries, link) {
                if (i++ == idx) {
                        *size = entry->addr_len;
                        return &entry->addr;
                }
        }
        unreachable();
}

void
evio_service_create(struct ev_loop *loop, struct evio_service *service,
                    const char *name, evio_accept_f on_accept,
                    void *on_accept_param)
{
        memset(service, 0, sizeof(struct evio_service));
        snprintf(service->name, sizeof(service->name), "%s", name);
        service->loop = loop;
        service->on_accept = on_accept;
        service->on_accept_param = on_accept_param;
        service->entry_count = 0;
        rlist_create(&service->entries);
}

void
evio_service_attach(struct evio_service *dst, const struct evio_service *src)
{
        assert(dst->entry_count == 0);
        struct evio_service_entry *src_entry;
        rlist_foreach_entry(src_entry, &src->entries, link) {
                struct evio_service_entry *dst_entry =
                        xmalloc(sizeof(struct evio_service_entry));
                evio_service_entry_create(dst_entry, dst);
                evio_service_entry_attach(dst_entry, src_entry);
                evio_service_add_entry(dst, dst_entry);
        }
}

void
evio_service_detach(struct evio_service *service)
{
        if (service->entry_count == 0)
                return;
        struct evio_service_entry *entry, *tmp;
        rlist_foreach_entry_safe(entry, &service->entries, link, tmp) {
                evio_service_entry_detach(entry);
                evio_service_delete_entry(entry);
        }
        assert(service->entry_count == 0);
}

/** Listen on bound socket. */
static int
evio_service_listen(struct evio_service *service)
{
        struct evio_service_entry *entry;
        rlist_foreach_entry(entry, &service->entries, link) {
                if (evio_service_entry_listen(entry) != 0)
                        return -1;
        }
        return 0;
}

void
evio_service_stop(struct evio_service *service)
{
        if (service->entry_count == 0)
                return;
        say_info("%s: stopped", evio_service_name(service));
        struct evio_service_entry *entry, *tmp;
        rlist_foreach_entry_safe(entry, &service->entries, link, tmp) {
                evio_service_entry_stop(entry);
                evio_service_delete_entry(entry);
        }
        assert(service->entry_count == 0);
}

/** Bind service to specified URI. */
static int
evio_service_bind(struct evio_service *service, const struct uri_set *uri_set)
{
        if (evio_service_reuse_addr(uri_set) != 0)
                return -1;
        for (int i = 0; i < uri_set->uri_count; i++) {
                const struct uri *uri = &uri_set->uris[i];
                if (evio_service_bind_uri(service, uri) != 0)
                        return -1;
        }
        return 0;
}

int
evio_service_start(struct evio_service *service, const struct uri_set *uri_set)
{
        if (evio_service_bind(service, uri_set) != 0)
                return -1;
        if (evio_service_listen(service) != 0)
                return -1;
        return 0;
}

int
evio_service_reload_uris(struct evio_service *service)
{
        struct evio_service_entry *entry;
        rlist_foreach_entry(entry, &service->entries, link) {
                if (evio_service_entry_reload_uri(entry) != 0)
                        return -1;
        }
        return 0;
}

1	/*
2	* Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
3	*
4	* Redistribution and use in source and binary forms, with or
5	* without modification, are permitted provided that the following
6	* conditions are met:
7	*
8	* 1. Redistributions of source code must retain the above
9	* copyright notice, this list of conditions and the
10	* following disclaimer.
11	*
12	* 2. Redistributions in binary form must reproduce the above
13	* copyright notice, this list of conditions and the following
14	* disclaimer in the documentation and/or other materials
15	* provided with the distribution.
16	*
17	* THIS SOFTWARE IS PROVIDED BY <COPYRIGHT HOLDER> ``AS IS'' AND
18	* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
19	* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
20	* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
21	* <COPYRIGHT HOLDER> OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
22	* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
23	* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
24	* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
25	* BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
26	* LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27	* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
28	* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
29	* SUCH DAMAGE.
30	*/
31	#include "evio.h"
32	#include <stdio.h>
33	#include <sys/socket.h>
34	#include <sys/un.h>
35	#include <netinet/in.h>
36	#include <netinet/tcp.h>
37	#include <arpa/inet.h>
38
39	#include <trivia/util.h>
40	#include "exception.h"
41	#include "iostream.h"
42	#include "tt_strerror.h"
43	#include "uri/uri.h"
44
45	struct evio_service_entry {
46	/** Bind URI */
47	struct uri uri;
48	/** Interface/port to bind to */
49	union {
50	struct sockaddr addr;
51	struct sockaddr_storage addrstorage;
52	};
53	socklen_t addr_len;
54	/** IO stream context. */
55	struct iostream_ctx io_ctx;
56	/** libev io object for the acceptor socket. */
57	struct ev_io ev;
58	/** Pointer to the root evio_service, which contains this object */
59	struct evio_service *service;
60	/** Link to other entries */
61	struct rlist link;
62	};
63
64	static int
65	evio_setsockopt_keepalive(int fd)	152✔
66	{
67	int on = 1;	152✔
68	/*
69	* SO_KEEPALIVE to ensure connections don't hang
70	* around for too long when a link goes away.
71	*/
72	if (sio_setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,	152!
73	&on, sizeof(on)))
74	return -1;	152✔
75	#ifdef __linux__
76	/*
77	* On Linux, we are able to fine-tune keepalive
78	* intervals. Set smaller defaults, since the system-wide
79	* defaults are in days.
80	*/
81	int keepcnt = 5;	152✔
82	if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &keepcnt,	152!
83	sizeof(int)))
84	return -1;	×
85	int keepidle = 30;	152✔
86
87	if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &keepidle,	152!
88	sizeof(int)))
89	return -1;	×
90
91	int keepintvl = 60;	152✔
92	if (sio_setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &keepintvl,	152!
93	sizeof(int)))
94	return -1;	×
95	#endif
96	return 0;	152✔
97	}
98
99	/** Set common client socket options. */
100	int
101	evio_setsockopt_client(int fd, int family, int type)	103,368✔
102	{
103	int on = 1;	103,368✔
104	/* In case this throws, the socket is not leaked. */
105	if (sio_setfl(fd, O_NONBLOCK, on))	103,368!
106	return -1;	103,368✔
107	if (type == SOCK_STREAM && family != AF_UNIX) {	103,368!
108	/*
109	* SO_KEEPALIVE to ensure connections don't hang
110	* around for too long when a link goes away.
111	*/
112	if (evio_setsockopt_keepalive(fd) != 0)	121!
113	return -1;	×
114	/*
115	* Lower latency is more important than higher
116	* bandwidth, and we usually write entire
117	* request/response in a single syscall.
118	*/
119	if (sio_setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,	121!
120	&on, sizeof(on)))
121	return -1;	×
122	}
123	return 0;	103,368✔
124	}
125
126	int
127	evio_setsockopt_server(int fd, int family, int type)	3,691✔
128	{
129	int on = 1;	3,691✔
130	/* In case this throws, the socket is not leaked. */
131	if (sio_setfl(fd, O_NONBLOCK, on))	3,691!
132	return -1;	3,691✔
133	/* Allow reuse local adresses. */
134	if (sio_setsockopt(fd, SOL_SOCKET, SO_REUSEADDR,	3,691!
135	&on, sizeof(on)))
136	return -1;	×
137	/*
138	* Allow binding to the same port for ipv4 and ipv6 simultaneously on
139	* platforms with dualstack sockets support.
140	*/
141	if (family == AF_INET6 && sio_setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY,	3,691!
142	&on, sizeof(on)))
NEW 143	return -1;	×
144
145	#ifndef TARANTOOL_WSL1_WORKAROUND_ENABLED
146	/* Send all buffered messages on socket before take
147	* control out from close(2) or shutdown(2). */
148	struct linger linger = { 0, 0 };	3,691✔
149
150	if (sio_setsockopt(fd, SOL_SOCKET, SO_LINGER,	3,691!
151	&linger, sizeof(linger)))
152	return -1;	×
153	#endif
154	if (type == SOCK_STREAM && family != AF_UNIX &&	3,691✔
155	evio_setsockopt_keepalive(fd) != 0)	31!
156	return -1;	×
157	return 0;	3,691✔
158	}
159
160	static inline const char *
161	evio_service_name(struct evio_service *service)	7,200✔
162	{
163	return service->name;	7,200✔
164	}
165
166	static void
167	evio_service_entry_detach(struct evio_service_entry *entry);
168
169	static void
170	evio_service_entry_stop(struct evio_service_entry *entry);
171
172	/**
173	* A callback invoked by libev when acceptor socket is ready.
174	* Accept the socket, initialize it and pass to the on_accept
175	* callback.
176	*/
177	static void
178	evio_service_entry_accept_cb(ev_loop loop, ev_io watcher, int events)	23,560✔
179	{
180	(void) loop;
181	(void) events;
182	struct evio_service_entry *entry =	23,560✔
183	(struct evio_service_entry *)watcher->data;
184	int fd;
185	while (1) {	25,130✔
186	/*
187	* Accept all pending connections from backlog during event
188	* loop iteration. Significally speed up acceptor with enabled
189	* io_collect_interval.
190	*/
191	struct sockaddr_storage addr;
192	socklen_t addrlen = sizeof(addr);	48,690✔
193	fd = sio_accept(entry->ev.fd, (struct sockaddr *)&addr,	48,690!
194	&addrlen);
195
196	if (fd < 0) {	48,690✔
197	if (! sio_wouldblock(errno))	23,560!
198	break;	×
199	return;	23,560✔
200	}
201	if (evio_setsockopt_client(fd, entry->addr.sa_family,	25,130!
202	SOCK_STREAM) != 0)
203	break;	×
204	struct iostream io;
205	if (iostream_create(&io, fd, &entry->io_ctx) != 0)	25,130!
206	break;	×
207	entry->service->on_accept(entry->service, &io,	25,130!
208	(struct sockaddr *)&addr, addrlen);
209	/* Must be moved by the callback. */
210	assert(!iostream_is_initialized(&io));	25,130!
211	}
212	if (fd >= 0)	×
213	close(fd);	×
214	diag_log();	×
215	}
216
217	/*
218	* Check if the UNIX socket exists and no one is
219	* listening on it. Unlink the file if it's the case.
220	*/
221	static int
222	evio_service_entry_reuse_addr(const struct uri *u)	3,662✔
223	{
224	if (u->host != NULL && strcmp(u->host, URI_HOST_UNIX) != 0)	3,662✔
225	return 0;	3,662✔
226
227	struct sockaddr_un un = { 0 };	3,641✔
228	assert(u->service != NULL);	3,641!
229	strlcpy(un.sun_path, u->service, sizeof(un.sun_path));	3,641!
230	un.sun_family = AF_UNIX;	3,641✔
231
232	int cl_fd = sio_socket(un.sun_family, SOCK_STREAM, 0);	3,641!
233	if (cl_fd < 0)	3,641!
234	return -1;	×
235
236	if (connect(cl_fd, (struct sockaddr *)&un, sizeof(un)) == 0)	3,641!
237	goto err;	×
238
239	if (errno == ECONNREFUSED && unlink(un.sun_path) != 0)	3,641!
240	goto err;	×
241
242	close(cl_fd);	3,641!
243	return 0;	3,641✔
244	err:	×
245	errno = EADDRINUSE;	×
246	diag_set(SocketError, sio_socketname(cl_fd), "unlink");	×
247	close(cl_fd);	×
248	return -1;	×
249	}
250
251	/**
252	* Try to bind on the configured port.
253	*
254	* Throws an exception if error.
255	*/
256	static int
257	evio_service_entry_bind_addr(struct evio_service_entry *entry)	3,661✔
258	{
259	say_debug("%s: binding to %s...",	3,661!
260	evio_service_name(entry->service),
261	sio_strfaddr(&entry->addr, entry->addr_len));
262	/* Create a socket. */
263	int fd = sio_socket(entry->addr.sa_family,	3,661✔
264	SOCK_STREAM, IPPROTO_TCP);
265	if (fd < 0)	3,661!
266	return -1;	×
267
268	if (evio_setsockopt_server(fd, entry->addr.sa_family,	3,661!
269	SOCK_STREAM) != 0)
270	goto error;	×
271
272	if (sio_bind(fd, &entry->addr, entry->addr_len) != 0)	3,661✔
273	goto error;	1✔
274
275	/*
276	* After binding a result address may be different. For
277	* example, if a port was 0.
278	*/
279	if (sio_getsockname(fd, &entry->addr, &entry->addr_len) != 0)	3,660!
280	goto error;	×
281
282	say_info("%s: bound to %s",	3,660✔
283	evio_service_name(entry->service),
284	sio_strfaddr(&entry->addr, entry->addr_len));
285
286	/* Register the socket in the event loop. */
287	ev_io_set(&entry->ev, fd, EV_READ);	3,660✔
288	return 0;	3,660✔
289	error:	1✔
290	close(fd);	1✔
291	return -1;	1✔
292	}
293
294	/**
295	* Listen on bounded port.
296	*
297	* @retval 0 for success
298	*/
299	static int
300	evio_service_entry_listen(struct evio_service_entry *entry)	3,654✔
301	{
302	say_debug("%s: listening on %s...",	3,654!
303	evio_service_name(entry->service),
304	sio_strfaddr(&entry->addr, entry->addr_len));
305
306	int fd = entry->ev.fd;	3,654✔
307	if (sio_listen(fd))	3,654!
308	return -1;	×
309	if (entry->service->on_accept != NULL)	3,654!
310	ev_io_start(entry->service->loop, &entry->ev);	×
311	return 0;	3,654✔
312	}
313
314	static void
315	evio_service_entry_create(struct evio_service_entry *entry,	7,608✔
316	struct evio_service *service)
317	{
318	uri_create(&entry->uri, NULL);	7,608✔
319	memset(&entry->addrstorage, 0, sizeof(entry->addrstorage));	7,608✔
320	entry->addr_len = 0;	7,608✔
321	iostream_ctx_clear(&entry->io_ctx);	7,608✔
322	/*
323	* Initialize libev objects to be able to detect if they
324	* are active or not in evio_service_entry_stop().
325	*/
326	ev_init(&entry->ev, evio_service_entry_accept_cb);	7,608✔
327	ev_io_set(&entry->ev, -1, 0);	7,608✔
328	entry->ev.data = entry;	7,608✔
329	entry->service = service;	7,608✔
330	rlist_create(&entry->link);	7,608✔
331	}	7,608✔
332
333	/**
334	* Add new entry to the service.
335	*/
336	static void
337	evio_service_add_entry(struct evio_service *dst,	7,608✔
338	struct evio_service_entry *entry)
339	{
340	rlist_add_entry(&dst->entries, entry, link);	7,608✔
341	++dst->entry_count;	7,608✔
342	}	7,608✔
343
344	/**
345	* Remove the entry from service and free the memory.
346	*/
347	static void
348	evio_service_delete_entry(struct evio_service_entry *entry)	7,470✔
349	{
350	struct evio_service *service = entry->service;	7,470✔
351	--service->entry_count;	7,470✔
352	rlist_del_entry(entry, link);	7,470✔
353	free(entry);	7,470✔
354	}	7,470✔
355
356	/**
357	* Try to bind.
358	*/
359	static int
360	evio_service_bind_uri(struct evio_service service, const struct uri u)	3,660✔
361	{
362	assert(u->service != NULL);	3,660!
363	if (u->host != NULL && strcmp(u->host, URI_HOST_UNIX) == 0) {	3,660✔
364	struct evio_service_entry *entry =	3,632✔
365	xmalloc(sizeof(struct evio_service_entry));	3,632!
366	evio_service_entry_create(entry, service);	3,632!
367	evio_service_add_entry(service, entry);	3,632!
368	assert(!ev_is_active(&entry->ev));	3,632!
369
370	if (iostream_ctx_create(&entry->io_ctx,	3,632!
371	IOSTREAM_SERVER, u) != 0) {
372	evio_service_delete_entry(entry);	2!
373	return -1;	3,660✔
374	}
375
376	uri_destroy(&entry->uri);	3,630!
377	uri_copy(&entry->uri, u);	3,630!
378
379	/* UNIX domain socket */
380	struct sockaddr_un un = (struct sockaddr_un )&entry->addr;	3,630✔
381	entry->addr_len = sizeof(*un);	3,630✔
382	strlcpy(un->sun_path, u->service, sizeof(un->sun_path));	3,630!
383	un->sun_family = AF_UNIX;	3,630✔
384	return evio_service_entry_bind_addr(entry);	3,630!
385	}
386	/* IP socket */
387	struct addrinfo hints, *res;
388	memset(&hints, 0, sizeof(hints));	28✔
389	hints.ai_family = AF_UNSPEC;	28✔
390	hints.ai_socktype = SOCK_STREAM;	28✔
391	hints.ai_flags = AI_PASSIVE \| AI_ADDRCONFIG;	28✔
392
393	if (getaddrinfo(u->host, u->service, &hints, &res) != 0 \|\|	28!
394	res == NULL) {	26!
395	diag_set(SocketError, sio_socketname(-1),	2!
396	"can't resolve uri for bind");
397	return -1;	2✔
398	}
399	for (struct addrinfo *ai = res; ai != NULL; ai = ai->ai_next) {	60✔
400	struct evio_service_entry *iter_entry;
401	rlist_foreach_entry(iter_entry, &service->entries,	118✔
402	link) {
403	if (iter_entry->addr_len == ai->ai_addrlen &&	27✔
404	memcmp(&iter_entry->addr, ai->ai_addr,	21✔
405	iter_entry->addr_len) == 0) {	21✔
406	say_debug("Duplicate address %s returned by"	3!
407	" getaddrinfo is omitted",
408	sio_strfaddr(ai->ai_addr,
409	ai->ai_addrlen));
410	goto next;	3✔
411	}
412	}
413	struct evio_service_entry *entry =	32✔
414	xmalloc(sizeof(struct evio_service_entry));	32!
415	evio_service_entry_create(entry, service);	32!
416	evio_service_add_entry(service, entry);	32!
417	assert(!ev_is_active(&entry->ev));	32!
418
419	if (iostream_ctx_create(&entry->io_ctx,	32!
420	IOSTREAM_SERVER, u) != 0) {
421	evio_service_entry_stop(entry);	1!
422	evio_service_delete_entry(entry);	1!
423	freeaddrinfo(res);	1✔
424	return -1;	1✔
425	}
426
427	uri_destroy(&entry->uri);	31!
428	uri_copy(&entry->uri, u);	31!
429	memcpy(&entry->addr, ai->ai_addr, ai->ai_addrlen);	31✔
430	entry->addr_len = ai->ai_addrlen;	31✔
431	if (evio_service_entry_bind_addr(entry) != 0) {	31!
NEW 432	say_error("%s: failed to bind on %s: %s",	×
433	evio_service_name(entry->service),
434	sio_strfaddr(ai->ai_addr, ai->ai_addrlen),
435	diag_last_error(diag_get())->errmsg);
NEW 436	evio_service_entry_stop(entry);	×
NEW 437	evio_service_delete_entry(entry);	×
438	freeaddrinfo(res);	×
NEW 439	diag_set(SocketError, sio_socketname(-1),	×
440	"%s: failed to bind",
441	evio_service_name(service));
NEW 442	return -1;	×
443	}
444	next:	34✔
445	;
446	}
447	freeaddrinfo(res);	25✔
448	return 0;	25✔
449	}
450
451	/ Destroy the entry uri, ctx and stop everything related to ev. /
452	static void
453	evio_service_entry_detach(struct evio_service_entry *entry)	7,468✔
454	{
455	iostream_ctx_destroy(&entry->io_ctx);	7,468✔
456	if (ev_is_active(&entry->ev)) {	7,468✔
457	ev_io_stop(entry->service->loop, &entry->ev);	3,875✔
458	entry->addr_len = 0;	3,875✔
459	}
460	ev_io_set(&entry->ev, -1, 0);	7,468✔
461	uri_destroy(&entry->uri);	7,468✔
462	}	7,468✔
463
464	/** It's safe to stop a service entry which is not started yet. */
465	static void
466	evio_service_entry_stop(struct evio_service_entry *entry)	3,593✔
467	{
468	int service_fd = entry->ev.fd;	3,593✔
469	evio_service_entry_detach(entry);	3,593✔
470	if (service_fd < 0)	3,593✔
471	return;	2✔
472
473	if (close(service_fd) < 0)	3,591!
474	say_error("Failed to close socket: %s", tt_strerror(errno));	×
475
476	if (entry->addr.sa_family != AF_UNIX)	3,591✔
477	return;	31✔
478
479	if (unlink(((struct sockaddr_un *)&entry->addr)->sun_path) < 0) {	3,560✔
480	say_error("Failed to unlink unix "	6!
481	"socket path: %s", tt_strerror(errno));
482	}
483	}
484
485	/**
486	* Updates dst entry socket settings according to ones in the src.
487	*/
488	static void
489	evio_service_entry_attach(struct evio_service_entry *dst,	3,944✔
490	const struct evio_service_entry *src)
491	{
492	assert(!ev_is_active(&dst->ev));	3,944!
493	uri_destroy(&dst->uri);	3,944✔
494	uri_copy(&dst->uri, &src->uri);	3,944✔
495	dst->addrstorage = src->addrstorage;	3,944✔
496	dst->addr_len = src->addr_len;	3,944✔
497	iostream_ctx_copy(&dst->io_ctx, &src->io_ctx);	3,944✔
498	ev_io_set(&dst->ev, src->ev.fd, EV_READ);	3,944✔
499	ev_io_start(dst->service->loop, &dst->ev);	3,944✔
500	}	3,944✔
501
502	/** Recreate the IO stream contexts from the service entry URI. */
503	static int
504	evio_service_entry_reload_uri(struct evio_service_entry *entry)	273✔
505	{
506	struct iostream_ctx io_ctx;
507	if (iostream_ctx_create(&io_ctx, IOSTREAM_SERVER, &entry->uri) != 0)	273!
508	return -1;	273✔
509	iostream_ctx_destroy(&entry->io_ctx);	273!
510	iostream_ctx_move(&entry->io_ctx, &io_ctx);	273!
511	return 0;	273✔
512	}
513
514	static inline int
515	evio_service_reuse_addr(const struct uri_set *uri_set)	3,645✔
516	{
517	for (int i = 0; i < uri_set->uri_count; i++) {	7,307✔
518	const struct uri *uri = &uri_set->uris[i];	3,662✔
519	if (evio_service_entry_reuse_addr(uri) != 0)	3,662!
520	return -1;	×
521	}
522	return 0;	3,645✔
523	}
524
525	int
526	evio_service_count(const struct evio_service *service)	225✔
527	{
528	return service->entry_count;	225✔
529	}
530
531	const struct sockaddr *
532	evio_service_addr(const struct evio_service service, int idx, socklen_t size)	271✔
533	{
534	assert(idx < service->entry_count && idx >= 0);	271!
535	struct evio_service_entry *entry;
536	int i = 0;	271✔
537	rlist_foreach_entry(entry, &service->entries, link) {	776!
538	if (i++ == idx) {	388✔
539	*size = entry->addr_len;	271✔
540	return &entry->addr;	271✔
541	}
542	}
NEW 543	unreachable();	×
544	}
545
546	void
547	evio_service_create(struct ev_loop loop, struct evio_service service,	8,509✔
548	const char *name, evio_accept_f on_accept,
549	void *on_accept_param)
550	{
551	memset(service, 0, sizeof(struct evio_service));	8,509✔
552	snprintf(service->name, sizeof(service->name), "%s", name);	8,509✔
553	service->loop = loop;	8,509✔
554	service->on_accept = on_accept;	8,509✔
555	service->on_accept_param = on_accept_param;	8,509✔
556	service->entry_count = 0;	8,509✔
557	rlist_create(&service->entries);	8,509✔
558	}	8,509✔
559
560	void
561	evio_service_attach(struct evio_service dst, const struct evio_service src)	4,625✔
562	{
563	assert(dst->entry_count == 0);	4,625!
564	struct evio_service_entry *src_entry;
565	rlist_foreach_entry(src_entry, &src->entries, link) {	17,138✔
566	struct evio_service_entry *dst_entry =	3,944✔
567	xmalloc(sizeof(struct evio_service_entry));	3,944!
568	evio_service_entry_create(dst_entry, dst);	3,944✔
569	evio_service_entry_attach(dst_entry, src_entry);	3,944✔
570	evio_service_add_entry(dst, dst_entry);	3,944✔
571	}
572	}	4,625✔
573
574	void
575	evio_service_detach(struct evio_service *service)	17,154✔
576	{
577	if (service->entry_count == 0)	17,154✔
578	return;	13,321✔
579	struct evio_service_entry entry, tmp;
580	rlist_foreach_entry_safe(entry, &service->entries, link, tmp) {	15,416!
581	evio_service_entry_detach(entry);	3,875✔
582	evio_service_delete_entry(entry);	3,875✔
583	}
584	assert(service->entry_count == 0);	3,833!
585	}
586
587	/** Listen on bound socket. */
588	static int
589	evio_service_listen(struct evio_service *service)	3,639✔
590	{
591	struct evio_service_entry *entry;
592	rlist_foreach_entry(entry, &service->entries, link) {	14,586✔
593	if (evio_service_entry_listen(entry) != 0)	3,654!
594	return -1;	×
595	}
596	return 0;	3,639✔
597	}
598
599	void
600	evio_service_stop(struct evio_service *service)	11,934✔
601	{
602	if (service->entry_count == 0)	11,934✔
603	return;	8,376✔
604	say_info("%s: stopped", evio_service_name(service));	3,558✔
605	struct evio_service_entry entry, tmp;
606	rlist_foreach_entry_safe(entry, &service->entries, link, tmp) {	14,300!
607	evio_service_entry_stop(entry);	3,592✔
608	evio_service_delete_entry(entry);	3,592✔
609	}
610	assert(service->entry_count == 0);	3,558!
611	}
612
613	/** Bind service to specified URI. */
614	static int
615	evio_service_bind(struct evio_service service, const struct uri_set uri_set)	3,645✔
616	{
617	if (evio_service_reuse_addr(uri_set) != 0)	3,645!
618	return -1;	×
619	for (int i = 0; i < uri_set->uri_count; i++) {	7,299✔
620	const struct uri *uri = &uri_set->uris[i];	3,660✔
621	if (evio_service_bind_uri(service, uri) != 0)	3,660✔
622	return -1;	6✔
623	}
624	return 0;	3,639✔
625	}
626
627	int
628	evio_service_start(struct evio_service service, const struct uri_set uri_set)	3,645✔
629	{
630	if (evio_service_bind(service, uri_set) != 0)	3,645✔
631	return -1;	6✔
632	if (evio_service_listen(service) != 0)	3,639!
633	return -1;	×
634	return 0;	3,639✔
635	}
636
637	int
638	evio_service_reload_uris(struct evio_service *service)	954✔
639	{
640	struct evio_service_entry *entry;
641	rlist_foreach_entry(entry, &service->entries, link) {	2,454✔
642	if (evio_service_entry_reload_uri(entry) != 0)	273!
643	return -1;	×
644	}
645	return 0;	954✔
646	}

tarantool / tarantool / 15920993143

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous