12436013294

Committed 20 Dec 2024 05:47PM UTC coverage: 87.379% (+0.01%) from 87.367%

Build # 12436013294

Build Type

push

github

Committed by

Totktonada

Commit Message

lua: fix flaky custom allocator test again

This patch fixes the flaky test verifying `used()` and `unused()`
functions of the `internal.alloc` module in the `alloc_test` suite.

The test compares the value of the used memory with the one calculated
by the builtin Lua functions. The executed and the real values has been
close to each other though not equal. This false negative result likely
happens because of the details of the `collectgarbage()` implementation.

This patch fixes it by adding arbitrary margin equal to 16Kb of memory
when comparing the value calculated using the LuaJIT call with the one
provided by the module function.

The same problem has been fixed in terms of the commit 7731d13d0af6a1
("lua: fix flaky custom allocator test"). The previous
commit fixed corresponding checks on the `unused()` value. This one
fixes `used()` checks that seem to be not precise in coverage workflows.

NO_CHANGELOG=flaky tests
NO_DOC=flaky tests

(cherry picked from commit dd5039a1d)

Run Details

69685 of 123524 branches covered (56.41%)

102630 of 117454 relevant lines covered (87.38%)

3271784.23 hits per line

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

84.53

/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)	83,707✔
54	{
55	int fd = sio_socket(addr->sa_family, SOCK_STREAM, 0);	83,707!
56	if (fd < 0)	83,707!
57	return -1;	83,707✔
58	if (evio_setsockopt_client(fd, addr->sa_family, SOCK_STREAM) != 0)	83,707!
59	goto err;	×
60	if (sio_connect(fd, addr, len) == 0)	83,707!
61	return fd;	26,856✔
62	if (errno != EINPROGRESS)	56,851✔
63	goto err;	56,775✔
64	/*
65	* Wait until socket is ready for writing or
66	* timed out.
67	*/
68	int revents = coio_wait(fd, EV_WRITE, timeout);	76✔
69	if (fiber_is_cancelled()) {	76!
70	diag_set(FiberIsCancelled);	×
71	goto err;	×
72	}
73	if (revents == 0) {	76✔
74	diag_set(TimedOut);	1!
75	goto err;	1✔
76	}
77	int error = EINPROGRESS;	75✔
78	socklen_t sz = sizeof(error);	75✔
79	if (sio_getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &sz))	75!
80	goto err;	×
81	if (error != 0) {	75✔
82	errno = error;	57✔
83	diag_set(SocketError, sio_socketname(fd), "connect");	57!
84	goto err;	57✔
85	}
86	return fd;	18✔
87	err:	56,833✔
88	close(fd);	56,833!
89	return -1;	56,833✔
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,	83,691✔
144	struct sockaddr addr, socklen_t addr_len,
145	ev_tstamp timeout)
146	{
147	int fd = -1;	83,691✔
148	/* try to resolve a hostname */
149	struct ev_loop *loop = loop();	83,691!
150	ev_tstamp start, delay;
151	evio_timeout_init(loop, &start, &delay, timeout);	83,691!
152
153	if (host != NULL && service != NULL &&	83,691!
154	strcmp(host, URI_HOST_UNIX) == 0) {	83,688✔
155	/* UNIX socket */
156	struct sockaddr_un un;
157	snprintf(un.sun_path, sizeof(un.sun_path), "%s", service);	83,631✔
158	un.sun_family = AF_UNIX;	83,631✔
159	fd = coio_connect_addr((struct sockaddr *)&un, sizeof(un),	83,631!
160	delay);
161	if (fd < 0)	83,631✔
162	return -1;	83,631✔
163	if (addr != NULL) {	26,856✔
164	assert(addr_len != NULL);	20,555!
165	addr_len = MIN(sizeof(un), addr_len);	20,555✔
166	memcpy(addr, &un, *addr_len);	20,555✔
167	}
168	return fd;	26,856✔
169	}
170
171	struct addrinfo *ai;
172	struct addrinfo *ai_resolve = NULL;	60✔
173	struct addrinfo ai_local = { .ai_addr = NULL };	60✔
174	if (host != NULL && service != NULL && host_hint != 0) {	60!
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));	48✔
182	hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */	48✔
183	hints.ai_socktype = SOCK_STREAM;	48✔
184	hints.ai_flags = AI_ADDRCONFIG;	48✔
185	int rc = coio_getaddrinfo(host, service, &hints, &ai_resolve,	48!
186	delay);
187	if (rc != 0)	48✔
188	return -1;	5✔
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) {	76!
194	fd = coio_connect_addr(ai->ai_addr, ai->ai_addrlen, delay);	76!
195	if (fd >= 0) {	76✔
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)	58✔
204	goto out;	35✔
205	/* Ignore the error and try the next address. */
206	ai = ai->ai_next;	23✔
207	ev_now_update(loop);	23!
208	coio_timeout_update(&start, &delay);	23!
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,	160,097✔
275	size_t bufsiz, ev_tstamp timeout)
276	{
277	assert(sz <= bufsiz);	160,097!
278	if (fiber_is_cancelled()) {	160,097!
279	diag_set(FiberIsCancelled);	×
280	return -1;	160,023✔
281	}
282	ev_tstamp start, delay;
283	coio_timeout_init(&start, &delay, timeout);	160,095!
284	ssize_t to_read = (ssize_t) sz;	160,098✔
285	while (true) {	172,882✔
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);	332,980!
292	if (nrd > 0) {	332,977✔
293	to_read -= nrd;	145,363✔
294	if (to_read <= 0)	145,363✔
295	return sz - to_read;	139,847✔
296	buf = (char *) buf + nrd;	5,516✔
297	bufsiz -= nrd;	5,516✔
298	continue;	5,516✔
299	} else if (nrd == 0) {	187,614✔
300	return sz - to_read;	3,065✔
301	} else if (nrd == IOSTREAM_ERROR) {	184,549✔
302	return -1;	1✔
303	}
304	if (delay <= 0) {	184,548✔
305	diag_set(TimedOut);	1,245!
306	return -1;	1,245✔
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);	183,303!
313	if (fiber_is_cancelled()) {	183,227✔
314	diag_set(FiberIsCancelled);	15,864!
315	return -1;	15,865✔
316	}
317	coio_timeout_update(&start, &delay);	167,365!
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)	34,957✔
331	{
332	ssize_t nrd = coio_read_ahead(io, buf, sz, bufsiz);	34,957✔
333	if (nrd >= 0 && nrd < (ssize_t)sz) {	34,868✔
334	errno = EPIPE;	20✔
335	diag_set(SocketError, sio_socketname(io->fd),	20!
336	"unexpected EOF when reading from socket");
337	return -1;	20✔
338	}
339	return nrd;	34,848✔
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,	120,029✔
352	size_t bufsiz, ev_tstamp timeout)
353	{
354	ssize_t nrd = coio_read_ahead_timeout(io, buf, sz, bufsiz, timeout);	120,029✔
355	if (nrd >= 0 && nrd < (ssize_t)sz) { /* EOF. */	120,035✔
356	errno = EPIPE;	2,058✔
357	diag_set(SocketError, sio_socketname(io->fd),	2,058!
358	"unexpected EOF when reading from socket");
359	return -1;	2,058✔
360	}
361	return nrd;	117,977✔
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,	743✔
378	ev_tstamp timeout)
379	{
380	if (fiber_is_cancelled()) {	743!
381	diag_set(FiberIsCancelled);	×
382	return -1;	743✔
383	}
384	ssize_t towrite = sz;	743✔
385	ev_tstamp start, delay;
386	coio_timeout_init(&start, &delay, timeout);	743!
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);	745!
393	if (nwr >= 0) {	745✔
394	/* Go past the data just written. */
395	if (nwr >= towrite)	743!
396	return sz;	743✔
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)	125,765✔
427	{
428	sio_add_to_iov(iov, -offset);	125,765✔
429	ssize_t nwr = iostream_writev(io, iov, iovcnt);	125,756✔
430	sio_add_to_iov(iov, offset);	125,822✔
431	return nwr;	125,817✔
432	}
433
434	ssize_t
435	coio_writev_timeout(struct iostream io, struct iovec iov, int iovcnt,	128,174✔
436	size_t size_hint, ev_tstamp timeout)
437	{
438	if (fiber_is_cancelled()) {	128,174✔
439	diag_set(FiberIsCancelled);	2,477!
440	return -1;	128,206✔
441	}
442	size_t total = 0;	125,693✔
443	size_t iov_len = 0;	125,693✔
444	struct iovec *end = iov + iovcnt;	125,693✔
445	ev_tstamp start, delay;
446	coio_timeout_init(&start, &delay, timeout);	125,693!
447
448	/* Avoid a syscall in case of 0 iovcnt. */
449	while (iov < end) {	125,763!
450	/* Write as much data as possible. */
451	ssize_t nwr = coio_flush(io, iov, iov_len, end - iov);	125,765!
452	if (nwr >= 0) {	125,814✔
453	total += nwr;	125,728✔
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)	125,728!
459	break;	×
460
461	iov += sio_move_iov(iov, nwr, &iov_len);	125,728!
462	if (iov == end) {	125,719✔
463	assert(iov_len == 0);	125,709!
464	break;	125,709✔
465	}
466	continue;	10✔
467	} else if (nwr == IOSTREAM_ERROR) {	86✔
468	return -1;	21✔
469	}
470	if (delay <= 0) {	65!
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);	65!
480	if (fiber_is_cancelled()) {	72✔
481	diag_set(FiberIsCancelled);	1!
482	return -1;	1✔
483	}
484	coio_timeout_update(&start, &delay);	71!
485	}
486	return total;	125,707✔
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)	686,588✔
561	{
562	(void) loop;
563	struct coio_wdata wdata = (struct coio_wdata ) watcher->data;	686,588✔
564	wdata->revents = revents;	686,588✔
565	fiber_wakeup(wdata->fiber);	686,588✔
566	}	686,586✔
567
568	API_EXPORT int
569	coio_wait(int fd, int events, double timeout)	1,785,790✔
570	{
571	if (fiber_is_cancelled())	1,785,790✔
572	return 0;	1,785,650✔
573	struct ev_io io;
574	ev_io_init(&io, coio_wait_cb, fd, events);	1,785,790✔
575	struct coio_wdata wdata = {	1,785,790✔
576	/* .fiber = */ fiber(),	1,785,790!
577	/* .revents = */ 0
578	};
579	io.data = &wdata;	1,785,790✔
580
581	/* A special hack to work with zero timeout */
582	ev_set_priority(&io, EV_MAXPRI);	1,785,790✔
583	ev_io_start(loop(), &io);	1,785,790!
584
585	fiber_yield_timeout(timeout);	1,785,780!
586
587	ev_io_stop(loop(), &io);	1,785,640!
588	return wdata.revents & (EV_READ \| EV_WRITE);	1,785,640✔
589	}
590
591	API_EXPORT int
592	coio_close(int fd)	40,820✔
593	{
594	ev_io_closing(loop(), fd);	40,820!
595	return close(fd);	40,820✔
596	}

tarantool / tarantool / 12436013294

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