11326255795

Committed 14 Oct 2024 11:02AM UTC coverage: 87.252% (-0.004%) from 87.256%

Build # 11326255795

Build Type

push

github

Committed by

locker

Commit Message

box: fix SIGSEGV on unaligned access to a struct with extended alignment

All structures with a non-default alignment (set by `alignas()`) must be
allocated by `aligned_alloc()`, otherwise an access to such a structure
member fill crash, e.g. if compiled with AVX-512 support.

Closes #10215
Part of #10631

NO_DOC=bugfix
NO_TEST=tested by debug_asan_clang workflow
NO_CHANGELOG=fix is actually not user-visible, because tarantool still
             doesn't work with enabled AVX-512 (#10671)

(cherry picked from commit a60ec82d4)

Run Details

68486 of 121715 branches covered (56.27%)

4 of 4 new or added lines in 3 files covered. (100.0%)

80 existing lines in 21 files now uncovered.

101080 of 115848 relevant lines covered (87.25%)

2454114.93 hits per line

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

85.25

/src/lib/core/coio.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 "coio.h"

#include <sys/socket.h>
#include <sys/un.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <arpa/inet.h>
#include <errno.h>

#include "iostream.h"
#include "sio.h"
#include "coio_task.h" /* coio_resolve() */
#include "uri/uri.h"

typedef void (*ev_stat_cb)(ev_loop *, ev_stat *, int);

/**
 * Connect to a host with a specified timeout.
 * @retval socket fd
 * @retval -1 error
 */
static int
coio_connect_addr(const struct sockaddr *addr, socklen_t len, ev_tstamp timeout)
{
        int fd = sio_socket(addr->sa_family, SOCK_STREAM, 0);
        if (fd < 0)
                return -1;
        if (evio_setsockopt_client(fd, addr->sa_family, SOCK_STREAM) != 0)
                goto err;
        if (sio_connect(fd, addr, len) == 0)
                return fd;
        if (errno != EINPROGRESS)
                goto err;
        /*
         * Wait until socket is ready for writing or
         * timed out.
         */
        int revents = coio_wait(fd, EV_WRITE, timeout);
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                goto err;
        }
        if (revents == 0) {
                diag_set(TimedOut);
                goto err;
        }
        int error = EINPROGRESS;
        socklen_t sz = sizeof(error);
        if (sio_getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &sz))
                goto err;
        if (error != 0) {
                errno = error;
                diag_set(SocketError, sio_socketname(fd), "connect");
                goto err;
        }
        return fd;
err:
        close(fd);
        return -1;
}

static int
coio_fill_addrinfo(struct addrinfo *ai_local, const char *host,
                   const char *service, int host_hint)
{
        int rc;
        ai_local->ai_next = NULL;
        if (host_hint == 1) { // IPv4
                ai_local->ai_addrlen = sizeof(struct sockaddr_in);
                ai_local->ai_addr = xmalloc(ai_local->ai_addrlen);
                memset(ai_local->ai_addr, 0, ai_local->ai_addrlen);
                struct sockaddr_in *addr = (void *)ai_local->ai_addr;
                addr->sin_family = AF_INET;
                addr->sin_port = htons((uint16_t)atoi(service));
                rc = inet_pton(AF_INET, host, &addr->sin_addr);
        } else { // IPv6
                ai_local->ai_addrlen = sizeof(struct sockaddr_in6);
                ai_local->ai_addr = xmalloc(ai_local->ai_addrlen);
                memset(ai_local->ai_addr, 0, ai_local->ai_addrlen);
                struct sockaddr_in6 *addr = (void *)ai_local->ai_addr;
                addr->sin6_family = AF_INET6;
                addr->sin6_port = htons((uint16_t)atoi(service));
                rc = inet_pton(AF_INET6, host, &addr->sin6_addr);
        }
        if (rc != 1) {
                free(ai_local->ai_addr);
                ai_local->ai_addr = NULL;
                diag_set(IllegalParams, "Invalid host name: %s", host);
                return -1;
        }
        return 0;
}

/**
 * Resolve \a host and \a service with optional \a host_hint and connect to
 * the first available address with a specified timeout.
 *
 * If \a addr is not NULL the function provides resolved address on success.
 * In this case, \a addr_len is a value-result argument. It should be
 * initialized to the size of the buffer associated with \a addr. Upon return,
 * \a addr_len is updated to contain the actual size of the source address.
 * The returned address is truncated if the buffer provided is too small;
 * in this case, addrlen will return a value greater than was supplied to the
 * call.
 *
 * This function also supports UNIX domain sockets: if \a host is 'unix/',
 * it will treat \a service as a path to a socket file.
 *
 * @retval socket fd
 * @retval -1 error
 */
int
coio_connect_timeout(const char *host, const char *service, int host_hint,
                     struct sockaddr *addr, socklen_t *addr_len,
                     ev_tstamp timeout)
{
        int fd = -1;
        /* try to resolve a hostname */
        struct ev_loop *loop = loop();
        ev_tstamp start, delay;
        evio_timeout_init(loop, &start, &delay, timeout);

        if (host != NULL && service != NULL &&
            strcmp(host, URI_HOST_UNIX) == 0) {
                /* UNIX socket */
                struct sockaddr_un un;
                snprintf(un.sun_path, sizeof(un.sun_path), "%s", service);
                un.sun_family = AF_UNIX;
                fd = coio_connect_addr((struct sockaddr *)&un, sizeof(un),
                                       delay);
                if (fd < 0)
                        return -1;
                if (addr != NULL) {
                        assert(addr_len != NULL);
                        *addr_len = MIN(sizeof(un), *addr_len);
                        memcpy(addr, &un, *addr_len);
                }
                return fd;
        }

        struct addrinfo *ai;
        struct addrinfo *ai_resolve = NULL;
        struct addrinfo ai_local = { .ai_addr = NULL };
        if (host != NULL && service != NULL && host_hint != 0) {
                if (coio_fill_addrinfo(&ai_local, host, service,
                                       host_hint) != 0)
                        return -1;
                ai = &ai_local;
        } else {
                struct addrinfo hints;
                memset(&hints, 0, sizeof(struct addrinfo));
                hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */
                hints.ai_socktype = SOCK_STREAM;
                hints.ai_flags = AI_ADDRCONFIG;
                int rc = coio_getaddrinfo(host, service, &hints, &ai_resolve,
                                          delay);
                if (rc != 0)
                        return -1;
                ai = ai_resolve;
        }
        evio_timeout_update(loop(), &start, &delay);
        coio_timeout_init(&start, &delay, timeout);
        while (ai) {
                fd = coio_connect_addr(ai->ai_addr, ai->ai_addrlen, delay);
                if (fd >= 0) {
                        if (addr != NULL) {
                                assert(addr_len != NULL);
                                *addr_len = MIN(ai->ai_addrlen, *addr_len);
                                memcpy(addr, ai->ai_addr, *addr_len);
                        }
                        goto out; /* connected */
                }
                if (ai->ai_next == NULL)
                        goto out;
                /* Ignore the error and try the next address. */
                ai = ai->ai_next;
                ev_now_update(loop);
                coio_timeout_update(&start, &delay);
        }
        diag_set(SocketError, sio_socketname(fd), "connection failed");
        return fd;
out:
        if (ai_resolve != NULL)
                freeaddrinfo(ai_resolve);
        if (ai_local.ai_addr != NULL)
                free(ai_local.ai_addr);
        return fd;
}

/**
 * Wait a client connection on a server socket until
 * timedout.
 * @retval socket fd
 * @retval -1 error
 */
int
coio_accept(int sfd, struct sockaddr *addr, socklen_t addrlen,
            ev_tstamp timeout)
{
        ev_tstamp start, delay;
        coio_timeout_init(&start, &delay, timeout);

        while (true) {
                /* Assume that there are waiting clients
                 * available */
                int fd = sio_accept(sfd, addr, &addrlen);
                if (fd >= 0) {
                        if (evio_setsockopt_client(fd, addr->sa_family,
                                                   SOCK_STREAM) != 0) {
                                close(fd);
                                return -1;
                        }
                        return fd;
                }
                if (!sio_wouldblock(errno))
                        return -1;
                if (delay <= 0) {
                        diag_set(TimedOut);
                        return -1;
                }
                /*
                 * Yield control to other fibers until the
                 * timeout is reached.
                 */
                coio_wait(sfd, EV_READ, delay);
                if (fiber_is_cancelled()) {
                        diag_set(FiberIsCancelled);
                        return -1;
                }
                coio_timeout_update(&start, &delay);
        }
}

/**
 * Read at least sz bytes from socket with readahead.
 *
 * In case of EOF returns the amount read until eof (possibly 0).
 * Can read up to bufsiz bytes.
 *
 * @retval the number of bytes read.
 * @retval -1 error.
 */
ssize_t
coio_read_ahead_timeout(struct iostream *io, void *buf, size_t sz,
                        size_t bufsiz, ev_tstamp timeout)
{
        assert(sz <= bufsiz);
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        ev_tstamp start, delay;
        coio_timeout_init(&start, &delay, timeout);
        ssize_t to_read = (ssize_t) sz;
        while (true) {
                /*
                 * Sic: assume the socket is ready: since
                 * the user called read(), some data must
                 * be expected.
                 */
                ssize_t nrd = iostream_read(io, buf, bufsiz);
                if (nrd > 0) {
                        to_read -= nrd;
                        if (to_read <= 0)
                                return sz - to_read;
                        buf = (char *) buf + nrd;
                        bufsiz -= nrd;
                        continue;
                } else if (nrd == 0) {
                        return sz - to_read;
                } else if (nrd == IOSTREAM_ERROR) {
                        return -1;
                }
                if (delay <= 0) {
                        diag_set(TimedOut);
                        return -1;
                }
                /*
                 * Yield control to other fibers until the
                 * timeout is being reached.
                 */
                coio_wait(io->fd, iostream_status_to_events(nrd), delay);
                if (fiber_is_cancelled()) {
                        diag_set(FiberIsCancelled);
                        return -1;
                }
                coio_timeout_update(&start, &delay);
        }
}

/**
 * Read at least sz bytes, with readahead.
 *
 * Treats EOF as an error, and throws an exception.
 *
 * @retval the number of bytes read, > 0.
 * @retval -1 error.
 */
ssize_t
coio_readn_ahead(struct iostream *io, void *buf, size_t sz, size_t bufsiz)
{
        ssize_t nrd = coio_read_ahead(io, buf, sz, bufsiz);
        if (nrd >= 0 && nrd < (ssize_t)sz) {
                errno = EPIPE;
                diag_set(SocketError, sio_socketname(io->fd),
                         "unexpected EOF when reading from socket");
                return -1;
        }
        return nrd;
}

/**
 * Read at least sz bytes, with readahead and timeout.
 *
 * Treats EOF as an error, and throws an exception.
 *
 * @retval the number of bytes read, > 0.
 * @retval -1 error.
 */
ssize_t
coio_readn_ahead_timeout(struct iostream *io, void *buf, size_t sz,
                         size_t bufsiz, ev_tstamp timeout)
{
        ssize_t nrd = coio_read_ahead_timeout(io, buf, sz, bufsiz, timeout);
        if (nrd >= 0 && nrd < (ssize_t)sz) { /* EOF. */
                errno = EPIPE;
                diag_set(SocketError, sio_socketname(io->fd),
                         "unexpected EOF when reading from socket");
                return -1;
        }
        return nrd;
}

/**
 * Write sz bytes to socket.
 *
 * Throws SocketError in case of write error. If
 * the socket is not ready, yields the current
 * fiber until the socket becomes ready, until
 * all data is written.
 *
 * @retval the number of bytes written. Always
 * equal to @a sz.
 * @retval -1 error.
 */
ssize_t
coio_write_timeout(struct iostream *io, const void *buf, size_t sz,
                   ev_tstamp timeout)
{
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        ssize_t towrite = sz;
        ev_tstamp start, delay;
        coio_timeout_init(&start, &delay, timeout);
        while (true) {
                /*
                 * Sic: write as much data as possible,
                 * assuming the socket is ready.
                 */
                ssize_t nwr = iostream_write(io, buf, towrite);
                if (nwr >= 0) {
                        /* Go past the data just written. */
                        if (nwr >= towrite)
                                return sz;
                        towrite -= nwr;
                        buf = (char *) buf + nwr;
                        continue;
                } else if (nwr == IOSTREAM_ERROR) {
                        return -1;
                }
                if (delay <= 0) {
                        diag_set(TimedOut);
                        return -1;
                }
                /*
                 * Yield control to other fibers until the
                 * timeout is reached or the socket is
                 * ready.
                 */
                coio_wait(io->fd, iostream_status_to_events(nwr), delay);
                if (fiber_is_cancelled()) {
                        diag_set(FiberIsCancelled);
                        return -1;
                }
                coio_timeout_update(&start, &delay);
        }
}

/*
 * Write iov using iostream API.
 * Put in an own function to workaround gcc bug with @finally
 */
static inline ssize_t
coio_flush(struct iostream *io, struct iovec *iov, ssize_t offset, int iovcnt)
{
        sio_add_to_iov(iov, -offset);
        ssize_t nwr = iostream_writev(io, iov, iovcnt);
        sio_add_to_iov(iov, offset);
        return nwr;
}

ssize_t
coio_writev_timeout(struct iostream *io, struct iovec *iov, int iovcnt,
                    size_t size_hint, ev_tstamp timeout)
{
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        size_t total = 0;
        size_t iov_len = 0;
        struct iovec *end = iov + iovcnt;
        ev_tstamp start, delay;
        coio_timeout_init(&start, &delay, timeout);

        /* Avoid a syscall in case of 0 iovcnt. */
        while (iov < end) {
                /* Write as much data as possible. */
                ssize_t nwr = coio_flush(io, iov, iov_len, end - iov);
                if (nwr >= 0) {
                        total += nwr;
                        /*
                         * If there was a hint for the total size
                         * of the vector, use it.
                         */
                        if (size_hint > 0 && size_hint == total)
                                break;

                        iov += sio_move_iov(iov, nwr, &iov_len);
                        if (iov == end) {
                                assert(iov_len == 0);
                                break;
                        }
                        continue;
                } else if (nwr == IOSTREAM_ERROR) {
                        return -1;
                }
                if (delay <= 0) {
                        diag_set(TimedOut);
                        return -1;
                }
                /*
                 * Yield control to other fibers until the
                 * timeout is reached or the socket is
                 * ready.
                 */
                coio_wait(io->fd, iostream_status_to_events(nwr), delay);
                if (fiber_is_cancelled()) {
                        diag_set(FiberIsCancelled);
                        return -1;
                }
                coio_timeout_update(&start, &delay);
        }
        return total;
}

void
coio_stat_init(ev_stat *stat, const char *path)
{
        ev_stat_init(stat, (ev_stat_cb) fiber_schedule_cb, path, 0.0);
}

int
coio_stat_stat_timeout(ev_stat *stat, ev_tstamp timeout)
{
        stat->data = fiber();
        ev_stat_start(loop(), stat);
        ev_tstamp start, delay;
        coio_timeout_init(&start, &delay, timeout);
        fiber_yield_timeout(delay);
        ev_stat_stop(loop(), stat);
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        return 0;
}

/**
 * The process status change callback.
 * Similar to fiber_schedule_cb, but also set watcher->data to NULL to indicate
 * that the fiber has been woken up on the child process termination.
 */
static void
coio_status_change_cb(ev_loop *loop, ev_child *watcher, int revents)
{
        (void)loop;
        (void)revents;
        struct fiber *fiber = watcher->data;
        assert(fiber() == &cord()->sched);
        watcher->data = NULL;
        fiber_wakeup(fiber);
}

int
coio_waitpid(pid_t pid, int *status)
{
        assert(cord_is_main());
        ev_child cw;
        ev_init(&cw, coio_status_change_cb);
        ev_child_set(&cw, pid, 0);
        cw.data = fiber();
        ev_child_start(loop(), &cw);

        do {
                fiber_yield();
        } while (cw.data != NULL);

        ev_child_stop(loop(), &cw);
        if (fiber_is_cancelled()) {
                diag_set(FiberIsCancelled);
                return -1;
        }
        *status = cw.rstatus;
        return 0;
}

/* Values of COIO_READ(WRITE) must equal to EV_READ(WRITE) */
static_assert(COIO_READ == (int) EV_READ, "TNT_IO_READ");
static_assert(COIO_WRITE == (int) EV_WRITE, "TNT_IO_WRITE");

struct coio_wdata {
        struct fiber *fiber;
        int revents;
};

static void
coio_wait_cb(struct ev_loop *loop, ev_io *watcher, int revents)
{
        (void) loop;
        struct coio_wdata *wdata = (struct coio_wdata *) watcher->data;
        wdata->revents = revents;
        fiber_wakeup(wdata->fiber);
}

API_EXPORT int
coio_wait(int fd, int events, double timeout)
{
        if (fiber_is_cancelled())
                return 0;
        struct ev_io io;
        ev_io_init(&io, coio_wait_cb, fd, events);
        struct coio_wdata wdata = {
                /* .fiber =   */ fiber(),
                /* .revents = */ 0
        };
        io.data = &wdata;

        /* A special hack to work with zero timeout */
        ev_set_priority(&io, EV_MAXPRI);
        ev_io_start(loop(), &io);

        fiber_yield_timeout(timeout);

        ev_io_stop(loop(), &io);
        return wdata.revents & (EV_READ | EV_WRITE);
}

API_EXPORT int
coio_close(int fd)
{
        ev_io_closing(loop(), fd);
        return close(fd);
}

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 "coio.h"
32
33	#include <sys/socket.h>
34	#include <sys/un.h>
35	#include <netinet/tcp.h>
36	#include <stdio.h>
37	#include <arpa/inet.h>
38	#include <errno.h>
39
40	#include "iostream.h"
41	#include "sio.h"
42	#include "coio_task.h" /* coio_resolve() */
43	#include "uri/uri.h"
44
45	typedef void (ev_stat_cb)(ev_loop , ev_stat *, int);
46
47	/**
48	* Connect to a host with a specified timeout.
49	* @retval socket fd
50	* @retval -1 error
51	*/
52	static int
53	coio_connect_addr(const struct sockaddr *addr, socklen_t len, ev_tstamp timeout)	84,326✔
54	{
55	int fd = sio_socket(addr->sa_family, SOCK_STREAM, 0);	84,326!
56	if (fd < 0)	84,326!
57	return -1;	84,326✔
58	if (evio_setsockopt_client(fd, addr->sa_family, SOCK_STREAM) != 0)	84,326!
59	goto err;	×
60	if (sio_connect(fd, addr, len) == 0)	84,326!
61	return fd;	24,560✔
62	if (errno != EINPROGRESS)	59,766✔
63	goto err;	59,691✔
64	/*
65	* Wait until socket is ready for writing or
66	* timed out.
67	*/
68	int revents = coio_wait(fd, EV_WRITE, timeout);	75✔
69	if (fiber_is_cancelled()) {	75✔
70	diag_set(FiberIsCancelled);	1!
71	goto err;	1✔
72	}
73	if (revents == 0) {	74✔
74	diag_set(TimedOut);	1!
75	goto err;	1✔
76	}
77	int error = EINPROGRESS;	73✔
78	socklen_t sz = sizeof(error);	73✔
79	if (sio_getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &sz))	73!
80	goto err;	×
81	if (error != 0) {	73✔
82	errno = error;	55✔
83	diag_set(SocketError, sio_socketname(fd), "connect");	55!
84	goto err;	55✔
85	}
86	return fd;	18✔
87	err:	59,748✔
88	close(fd);	59,748!
89	return -1;	59,748✔
90	}
91
92	static int
93	coio_fill_addrinfo(struct addrinfo ai_local, const char host,	12✔
94	const char *service, int host_hint)
95	{
96	int rc;
97	ai_local->ai_next = NULL;	12✔
98	if (host_hint == 1) { // IPv4	12✔
99	ai_local->ai_addrlen = sizeof(struct sockaddr_in);	11✔
100	ai_local->ai_addr = xmalloc(ai_local->ai_addrlen);	11!
101	memset(ai_local->ai_addr, 0, ai_local->ai_addrlen);	11✔
102	struct sockaddr_in addr = (void )ai_local->ai_addr;	11✔
103	addr->sin_family = AF_INET;	11✔
104	addr->sin_port = htons((uint16_t)atoi(service));	11✔
105	rc = inet_pton(AF_INET, host, &addr->sin_addr);	11✔
106	} else { // IPv6
107	ai_local->ai_addrlen = sizeof(struct sockaddr_in6);	1✔
108	ai_local->ai_addr = xmalloc(ai_local->ai_addrlen);	1!
109	memset(ai_local->ai_addr, 0, ai_local->ai_addrlen);	1✔
110	struct sockaddr_in6 addr = (void )ai_local->ai_addr;	1✔
111	addr->sin6_family = AF_INET6;	1✔
112	addr->sin6_port = htons((uint16_t)atoi(service));	1✔
113	rc = inet_pton(AF_INET6, host, &addr->sin6_addr);	1✔
114	}
115	if (rc != 1) {	12✔
116	free(ai_local->ai_addr);	2✔
117	ai_local->ai_addr = NULL;	2✔
118	diag_set(IllegalParams, "Invalid host name: %s", host);	2!
119	return -1;	2✔
120	}
121	return 0;	10✔
122	}
123
124	/**
125	* Resolve \a host and \a service with optional \a host_hint and connect to
126	* the first available address with a specified timeout.
127	*
128	* If \a addr is not NULL the function provides resolved address on success.
129	* In this case, \a addr_len is a value-result argument. It should be
130	* initialized to the size of the buffer associated with \a addr. Upon return,
131	* \a addr_len is updated to contain the actual size of the source address.
132	* The returned address is truncated if the buffer provided is too small;
133	* in this case, addrlen will return a value greater than was supplied to the
134	* call.
135	*
136	* This function also supports UNIX domain sockets: if \a host is 'unix/',
137	* it will treat \a service as a path to a socket file.
138	*
139	* @retval socket fd
140	* @retval -1 error
141	*/
142	int
143	coio_connect_timeout(const char host, const char service, int host_hint,	84,310✔
144	struct sockaddr addr, socklen_t addr_len,
145	ev_tstamp timeout)
146	{
147	int fd = -1;	84,310✔
148	/* try to resolve a hostname */
149	struct ev_loop *loop = loop();	84,310!
150	ev_tstamp start, delay;
151	evio_timeout_init(loop, &start, &delay, timeout);	84,310!
152
153	if (host != NULL && service != NULL &&	84,310!
154	strcmp(host, URI_HOST_UNIX) == 0) {	84,307✔
155	/* UNIX socket */
156	struct sockaddr_un un;
157	snprintf(un.sun_path, sizeof(un.sun_path), "%s", service);	84,251✔
158	un.sun_family = AF_UNIX;	84,251✔
159	fd = coio_connect_addr((struct sockaddr *)&un, sizeof(un),	84,251!
160	delay);
161	if (fd < 0)	84,251✔
162	return -1;	84,251✔
163	if (addr != NULL) {	24,560✔
164	assert(addr_len != NULL);	19,072!
165	addr_len = MIN(sizeof(un), addr_len);	19,072✔
166	memcpy(addr, &un, *addr_len);	19,072✔
167	}
168	return fd;	24,560✔
169	}
170
171	struct addrinfo *ai;
172	struct addrinfo *ai_resolve = NULL;	59✔
173	struct addrinfo ai_local = { .ai_addr = NULL };	59✔
174	if (host != NULL && service != NULL && host_hint != 0) {	59!
175	if (coio_fill_addrinfo(&ai_local, host, service,	12!
176	host_hint) != 0)
177	return -1;	2✔
178	ai = &ai_local;	10✔
179	} else {
180	struct addrinfo hints;
181	memset(&hints, 0, sizeof(struct addrinfo));	47✔
182	hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */	47✔
183	hints.ai_socktype = SOCK_STREAM;	47✔
184	hints.ai_flags = AI_ADDRCONFIG;	47✔
185	int rc = coio_getaddrinfo(host, service, &hints, &ai_resolve,	47!
186	delay);
187	if (rc != 0)	47✔
188	return -1;	4✔
189	ai = ai_resolve;	43✔
190	}
191	evio_timeout_update(loop(), &start, &delay);	53!
192	coio_timeout_init(&start, &delay, timeout);	53!
193	while (ai) {	75!
194	fd = coio_connect_addr(ai->ai_addr, ai->ai_addrlen, delay);	75!
195	if (fd >= 0) {	75✔
196	if (addr != NULL) {	18!
197	assert(addr_len != NULL);	×
198	addr_len = MIN(ai->ai_addrlen, addr_len);	×
199	memcpy(addr, ai->ai_addr, *addr_len);	×
200	}
201	goto out; /* connected */	18✔
202	}
203	if (ai->ai_next == NULL)	57✔
204	goto out;	35✔
205	/* Ignore the error and try the next address. */
206	ai = ai->ai_next;	22✔
207	ev_now_update(loop);	22!
208	coio_timeout_update(&start, &delay);	22!
209	}
210	diag_set(SocketError, sio_socketname(fd), "connection failed");	×
211	return fd;	×
212	out:	53✔
213	if (ai_resolve != NULL)	53✔
214	freeaddrinfo(ai_resolve);	43✔
215	if (ai_local.ai_addr != NULL)	53✔
216	free(ai_local.ai_addr);	10✔
217	return fd;	53✔
218	}
219
220	/**
221	* Wait a client connection on a server socket until
222	* timedout.
223	* @retval socket fd
224	* @retval -1 error
225	*/
226	int
227	coio_accept(int sfd, struct sockaddr *addr, socklen_t addrlen,	×
228	ev_tstamp timeout)
229	{
230	ev_tstamp start, delay;
231	coio_timeout_init(&start, &delay, timeout);	×
232
233	while (true) {	×
234	/* Assume that there are waiting clients
235	* available */
236	int fd = sio_accept(sfd, addr, &addrlen);	×
237	if (fd >= 0) {	×
238	if (evio_setsockopt_client(fd, addr->sa_family,	×
239	SOCK_STREAM) != 0) {
240	close(fd);	×
241	return -1;	×
242	}
243	return fd;	×
244	}
245	if (!sio_wouldblock(errno))	×
246	return -1;	×
247	if (delay <= 0) {	×
248	diag_set(TimedOut);	×
249	return -1;	×
250	}
251	/*
252	* Yield control to other fibers until the
253	* timeout is reached.
254	*/
255	coio_wait(sfd, EV_READ, delay);	×
256	if (fiber_is_cancelled()) {	×
257	diag_set(FiberIsCancelled);	×
258	return -1;	×
259	}
260	coio_timeout_update(&start, &delay);	×
261	}
262	}
263
264	/**
265	* Read at least sz bytes from socket with readahead.
266	*
267	* In case of EOF returns the amount read until eof (possibly 0).
268	* Can read up to bufsiz bytes.
269	*
270	* @retval the number of bytes read.
271	* @retval -1 error.
272	*/
273	ssize_t
274	coio_read_ahead_timeout(struct iostream io, void buf, size_t sz,	137,472✔
275	size_t bufsiz, ev_tstamp timeout)
276	{
277	assert(sz <= bufsiz);	137,472!
278	if (fiber_is_cancelled()) {	137,472!
UNCOV 279	diag_set(FiberIsCancelled);	×
280	return -1;	137,471✔
281	}
282	ev_tstamp start, delay;
283	coio_timeout_init(&start, &delay, timeout);	137,471!
284	ssize_t to_read = (ssize_t) sz;	137,469✔
285	while (true) {	143,906✔
286	/*
287	* Sic: assume the socket is ready: since
288	* the user called read(), some data must
289	* be expected.
290	*/
291	ssize_t nrd = iostream_read(io, buf, bufsiz);	281,375!
292	if (nrd > 0) {	281,435✔
293	to_read -= nrd;	124,396✔
294	if (to_read <= 0)	124,396✔
295	return sz - to_read;	118,884✔
296	buf = (char *) buf + nrd;	5,512✔
297	bufsiz -= nrd;	5,512✔
298	continue;	5,512✔
299	} else if (nrd == 0) {	157,039✔
300	return sz - to_read;	2,687✔
301	} else if (nrd == IOSTREAM_ERROR) {	154,352✔
302	return -1;	1✔
303	}
304	if (delay <= 0) {	154,351✔
305	diag_set(TimedOut);	640!
306	return -1;	640✔
307	}
308	/*
309	* Yield control to other fibers until the
310	* timeout is being reached.
311	*/
312	coio_wait(io->fd, iostream_status_to_events(nrd), delay);	153,711!
313	if (fiber_is_cancelled()) {	153,663✔
314	diag_set(FiberIsCancelled);	15,258!
315	return -1;	15,259✔
316	}
317	coio_timeout_update(&start, &delay);	138,402!
318	}
319	}
320
321	/**
322	* Read at least sz bytes, with readahead.
323	*
324	* Treats EOF as an error, and throws an exception.
325	*
326	* @retval the number of bytes read, > 0.
327	* @retval -1 error.
328	*/
329	ssize_t
330	coio_readn_ahead(struct iostream io, void buf, size_t sz, size_t bufsiz)	32,066✔
331	{
332	ssize_t nrd = coio_read_ahead(io, buf, sz, bufsiz);	32,066✔
333	if (nrd >= 0 && nrd < (ssize_t)sz) {	32,014✔
334	errno = EPIPE;	13✔
335	diag_set(SocketError, sio_socketname(io->fd),	13!
336	"unexpected EOF when reading from socket");
337	return -1;	13✔
338	}
339	return nrd;	32,001✔
340	}
341
342	/**
343	* Read at least sz bytes, with readahead and timeout.
344	*
345	* Treats EOF as an error, and throws an exception.
346	*
347	* @retval the number of bytes read, > 0.
348	* @retval -1 error.
349	*/
350	ssize_t
351	coio_readn_ahead_timeout(struct iostream io, void buf, size_t sz,	101,337✔
352	size_t bufsiz, ev_tstamp timeout)
353	{
354	ssize_t nrd = coio_read_ahead_timeout(io, buf, sz, bufsiz, timeout);	101,337✔
355	if (nrd >= 0 && nrd < (ssize_t)sz) { /* EOF. */	101,387✔
356	errno = EPIPE;	1,842✔
357	diag_set(SocketError, sio_socketname(io->fd),	1,842!
358	"unexpected EOF when reading from socket");
359	return -1;	1,842✔
360	}
361	return nrd;	99,545✔
362	}
363
364	/**
365	* Write sz bytes to socket.
366	*
367	* Throws SocketError in case of write error. If
368	* the socket is not ready, yields the current
369	* fiber until the socket becomes ready, until
370	* all data is written.
371	*
372	* @retval the number of bytes written. Always
373	* equal to @a sz.
374	* @retval -1 error.
375	*/
376	ssize_t
377	coio_write_timeout(struct iostream io, const void buf, size_t sz,	645✔
378	ev_tstamp timeout)
379	{
380	if (fiber_is_cancelled()) {	645!
381	diag_set(FiberIsCancelled);	×
382	return -1;	645✔
383	}
384	ssize_t towrite = sz;	645✔
385	ev_tstamp start, delay;
386	coio_timeout_init(&start, &delay, timeout);	645!
387	while (true) {	2✔
388	/*
389	* Sic: write as much data as possible,
390	* assuming the socket is ready.
391	*/
392	ssize_t nwr = iostream_write(io, buf, towrite);	647!
393	if (nwr >= 0) {	647✔
394	/* Go past the data just written. */
395	if (nwr >= towrite)	645!
396	return sz;	645✔
397	towrite -= nwr;	×
398	buf = (char *) buf + nwr;	×
399	continue;	×
400	} else if (nwr == IOSTREAM_ERROR) {	2!
401	return -1;	×
402	}
403	if (delay <= 0) {	2!
404	diag_set(TimedOut);	×
405	return -1;	×
406	}
407	/*
408	* Yield control to other fibers until the
409	* timeout is reached or the socket is
410	* ready.
411	*/
412	coio_wait(io->fd, iostream_status_to_events(nwr), delay);	2!
413	if (fiber_is_cancelled()) {	2!
414	diag_set(FiberIsCancelled);	×
415	return -1;	×
416	}
417	coio_timeout_update(&start, &delay);	2!
418	}
419	}
420
421	/*
422	* Write iov using iostream API.
423	* Put in an own function to workaround gcc bug with @finally
424	*/
425	static inline ssize_t
426	coio_flush(struct iostream io, struct iovec iov, ssize_t offset, int iovcnt)	107,808✔
427	{
428	sio_add_to_iov(iov, -offset);	107,808✔
429	ssize_t nwr = iostream_writev(io, iov, iovcnt);	107,792✔
430	sio_add_to_iov(iov, offset);	107,894✔
431	return nwr;	107,893✔
432	}
433
434	ssize_t
435	coio_writev_timeout(struct iostream io, struct iovec iov, int iovcnt,	109,967✔
436	size_t size_hint, ev_tstamp timeout)
437	{
438	if (fiber_is_cancelled()) {	109,967✔
439	diag_set(FiberIsCancelled);	2,223!
440	return -1;	109,978✔
441	}
442	size_t total = 0;	107,719✔
443	size_t iov_len = 0;	107,719✔
444	struct iovec *end = iov + iovcnt;	107,719✔
445	ev_tstamp start, delay;
446	coio_timeout_init(&start, &delay, timeout);	107,719!
447
448	/* Avoid a syscall in case of 0 iovcnt. */
449	while (iov < end) {	107,801!
450	/* Write as much data as possible. */
451	ssize_t nwr = coio_flush(io, iov, iov_len, end - iov);	107,807!
452	if (nwr >= 0) {	107,893✔
453	total += nwr;	107,781✔
454	/*
455	* If there was a hint for the total size
456	* of the vector, use it.
457	*/
458	if (size_hint > 0 && size_hint == total)	107,781!
459	break;	×
460
461	iov += sio_move_iov(iov, nwr, &iov_len);	107,781!
462	if (iov == end) {	107,774✔
463	assert(iov_len == 0);	107,749!
464	break;	107,749✔
465	}
466	continue;	25✔
467	} else if (nwr == IOSTREAM_ERROR) {	112✔
468	return -1;	11✔
469	}
470	if (delay <= 0) {	101!
471	diag_set(TimedOut);	×
472	return -1;	×
473	}
474	/*
475	* Yield control to other fibers until the
476	* timeout is reached or the socket is
477	* ready.
478	*/
479	coio_wait(io->fd, iostream_status_to_events(nwr), delay);	101!
480	if (fiber_is_cancelled()) {	102✔
481	diag_set(FiberIsCancelled);	1!
482	return -1;	1✔
483	}
484	coio_timeout_update(&start, &delay);	101!
485	}
486	return total;	107,743✔
487	}
488
489	void
490	coio_stat_init(ev_stat stat, const char path)	2✔
491	{
492	ev_stat_init(stat, (ev_stat_cb) fiber_schedule_cb, path, 0.0);	2✔
493	}	2✔
494
495	int
496	coio_stat_stat_timeout(ev_stat *stat, ev_tstamp timeout)	2✔
497	{
498	stat->data = fiber();	2!
499	ev_stat_start(loop(), stat);	2!
500	ev_tstamp start, delay;
501	coio_timeout_init(&start, &delay, timeout);	2!
502	fiber_yield_timeout(delay);	2!
503	ev_stat_stop(loop(), stat);	2!
504	if (fiber_is_cancelled()) {	2!
505	diag_set(FiberIsCancelled);	×
506	return -1;	2✔
507	}
508	return 0;	2✔
509	}
510
511	/**
512	* The process status change callback.
513	* Similar to fiber_schedule_cb, but also set watcher->data to NULL to indicate
514	* that the fiber has been woken up on the child process termination.
515	*/
516	static void
517	coio_status_change_cb(ev_loop loop, ev_child watcher, int revents)	1✔
518	{
519	(void)loop;
520	(void)revents;
521	struct fiber *fiber = watcher->data;	1✔
522	assert(fiber() == &cord()->sched);	1!
523	watcher->data = NULL;	1✔
524	fiber_wakeup(fiber);	1✔
525	}	1✔
526
527	int
528	coio_waitpid(pid_t pid, int *status)	1✔
529	{
530	assert(cord_is_main());	1!
531	ev_child cw;
532	ev_init(&cw, coio_status_change_cb);	1✔
533	ev_child_set(&cw, pid, 0);	1✔
534	cw.data = fiber();	1!
535	ev_child_start(loop(), &cw);	1!
536
537	do {
538	fiber_yield();	2✔
539	} while (cw.data != NULL);	2✔
540
541	ev_child_stop(loop(), &cw);	1!
542	if (fiber_is_cancelled()) {	1!
543	diag_set(FiberIsCancelled);	×
544	return -1;	1✔
545	}
546	*status = cw.rstatus;	1✔
547	return 0;	1✔
548	}
549
550	/* Values of COIO_READ(WRITE) must equal to EV_READ(WRITE) */
551	static_assert(COIO_READ == (int) EV_READ, "TNT_IO_READ");
552	static_assert(COIO_WRITE == (int) EV_WRITE, "TNT_IO_WRITE");
553
554	struct coio_wdata {
555	struct fiber *fiber;
556	int revents;
557	};
558
559	static void
560	coio_wait_cb(struct ev_loop loop, ev_io watcher, int revents)	339,210✔
561	{
562	(void) loop;
563	struct coio_wdata wdata = (struct coio_wdata ) watcher->data;	339,210✔
564	wdata->revents = revents;	339,210✔
565	fiber_wakeup(wdata->fiber);	339,210✔
566	}	339,210✔
567
568	API_EXPORT int
569	coio_wait(int fd, int events, double timeout)	887,012✔
570	{
571	if (fiber_is_cancelled())	887,012✔
572	return 0;	886,907✔
573	struct ev_io io;
574	ev_io_init(&io, coio_wait_cb, fd, events);	887,005✔
575	struct coio_wdata wdata = {	887,005✔
576	/* .fiber = */ fiber(),	887,005!
577	/* .revents = */ 0
578	};
579	io.data = &wdata;	887,005✔
580
581	/* A special hack to work with zero timeout */
582	ev_set_priority(&io, EV_MAXPRI);	887,005✔
583	ev_io_start(loop(), &io);	887,005!
584
585	fiber_yield_timeout(timeout);	886,996!
586
587	ev_io_stop(loop(), &io);	886,908!
588	return wdata.revents & (EV_READ \| EV_WRITE);	886,905✔
589	}
590
591	API_EXPORT int
592	coio_close(int fd)	38,315✔
593	{
594	ev_io_closing(loop(), fd);	38,315!
595	return close(fd);	38,315✔
596	}

tarantool / tarantool / 11326255795

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