10899922320

Committed 17 Sep 2024 09:04AM UTC coverage: 85.938% (+0.04%) from 85.896%

Build # 10899922320

Build Type

push

github

Committed by

sergepetrenko

Commit Message

memtx: reference tuples when rolling back statements without story

During the latest rework of DDL in MVCC, the new helper
`memtx_tx_history_rollback_empty_stmt` was introduced - it is used for
statements without stories (such statements can appear, for example,
when DDL removes all stories). By mistake, we forgot to unreference the
new tuple and reference the old one there - the commit fixes this
embarrassing mistake.

Follow-up #10146

NO_CHANGELOG=bugfix for unreleased patch
NO_DOC=bugfix

(cherry picked from commit 32797f703)

Run Details

63033 of 114297 branches covered (55.15%)

4 of 4 new or added lines in 1 file covered. (100.0%)

51 existing lines in 11 files now uncovered.

93931 of 109301 relevant lines covered (85.94%)

2469194.24 hits per line

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

84.01

/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);
        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)
{
        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)
{
        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)	64,998✔
54	{
55	int fd = sio_socket(addr->sa_family, SOCK_STREAM, 0);	64,998!
56	if (fd < 0)	64,998!
57	return -1;	64,998✔
58	if (evio_setsockopt_client(fd, addr->sa_family, SOCK_STREAM) != 0)	64,998!
59	goto err;	×
60	if (sio_connect(fd, addr, len) == 0)	64,998!
61	return fd;	20,062✔
62	if (errno != EINPROGRESS)	44,936✔
63	goto err;	44,908✔
64	/*
65	* Wait until socket is ready for writing or
66	* timed out.
67	*/
68	int revents = coio_wait(fd, EV_WRITE, timeout);	28✔
69	if (fiber_is_cancelled()) {	28!
UNCOV 70	diag_set(FiberIsCancelled);	×
UNCOV 71	goto err;	×
72	}
73	if (revents == 0) {	28✔
74	diag_set(TimedOut);	1!
75	goto err;	1✔
76	}
77	int error = EINPROGRESS;	27✔
78	socklen_t sz = sizeof(error);	27✔
79	if (sio_getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &sz))	27!
80	goto err;	×
81	if (error != 0) {	27✔
82	errno = error;	16✔
83	diag_set(SocketError, sio_socketname(fd), "connect");	16!
84	goto err;	16✔
85	}
86	return fd;	11✔
87	err:	44,925✔
88	close(fd);	44,925!
89	return -1;	44,925✔
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,	65,001✔
144	struct sockaddr addr, socklen_t addr_len,
145	ev_tstamp timeout)
146	{
147	int fd = -1;	65,001✔
148	/* try to resolve a hostname */
149	struct ev_loop *loop = loop();	65,001!
150	ev_tstamp start, delay;
151	evio_timeout_init(loop, &start, &delay, timeout);	65,001!
152
153	if (host != NULL && service != NULL &&	65,001!
154	strcmp(host, URI_HOST_UNIX) == 0) {	65,000✔
155	/* UNIX socket */
156	struct sockaddr_un un;
157	snprintf(un.sun_path, sizeof(un.sun_path), "%s", service);	64,970✔
158	un.sun_family = AF_UNIX;	64,970✔
159	fd = coio_connect_addr((struct sockaddr *)&un, sizeof(un),	64,970!
160	delay);
161	if (fd < 0)	64,970✔
162	return -1;	64,970✔
163	if (addr != NULL) {	20,062✔
164	assert(addr_len != NULL);	16,606!
165	addr_len = MIN(sizeof(un), addr_len);	16,606✔
166	memcpy(addr, &un, *addr_len);	16,606✔
167	}
168	return fd;	20,062✔
169	}
170
171	struct addrinfo *ai;
172	struct addrinfo *ai_resolve = NULL;	31✔
173	struct addrinfo ai_local = { .ai_addr = NULL };	31✔
174	if (host != NULL && service != NULL && host_hint != 0) {	31!
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));	19✔
182	hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */	19✔
183	hints.ai_socktype = SOCK_STREAM;	19✔
184	hints.ai_flags = AI_ADDRCONFIG;	19✔
185	int rc = coio_getaddrinfo(host, service, &hints, &ai_resolve,	19!
186	delay);
187	if (rc != 0)	19✔
188	return -1;	3✔
189	ai = ai_resolve;	16✔
190	}
191	evio_timeout_update(loop(), &start, &delay);	26!
192	coio_timeout_init(&start, &delay, timeout);	26!
193	while (ai) {	28!
194	fd = coio_connect_addr(ai->ai_addr, ai->ai_addrlen, delay);	28!
195	if (fd >= 0) {	28✔
196	if (addr != NULL) {	11!
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 */	11✔
202	}
203	if (ai->ai_next == NULL)	17✔
204	goto out;	15✔
205	/* Ignore the error and try the next address. */
206	ai = ai->ai_next;	2✔
207	ev_now_update(loop);	2!
208	coio_timeout_update(&start, &delay);	2!
209	}
210	diag_set(SocketError, sio_socketname(fd), "connection failed");	×
211	return fd;	×
212	out:	26✔
213	if (ai_resolve != NULL)	26✔
214	freeaddrinfo(ai_resolve);	16✔
215	if (ai_local.ai_addr != NULL)	26✔
216	free(ai_local.ai_addr);	10✔
217	return fd;	26✔
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,	136,370✔
275	size_t bufsiz, ev_tstamp timeout)
276	{
277	assert(sz <= bufsiz);	136,370!
278	ev_tstamp start, delay;
279	coio_timeout_init(&start, &delay, timeout);	136,370!
280	ssize_t to_read = (ssize_t) sz;	136,370✔
281	while (true) {	148,563✔
282	/*
283	* Sic: assume the socket is ready: since
284	* the user called read(), some data must
285	* be expected.
286	*/
287	ssize_t nrd = iostream_read(io, buf, bufsiz);	284,933!
288	if (nrd > 0) {	284,933✔
289	to_read -= nrd;	126,366✔
290	if (to_read <= 0)	126,366✔
291	return sz - to_read;	135,516✔
292	buf = (char *) buf + nrd;	5,550✔
293	bufsiz -= nrd;	5,550✔
294	continue;	5,550✔
295	} else if (nrd == 0) {	158,567✔
296	return sz - to_read;	1,317✔
297	} else if (nrd == IOSTREAM_ERROR) {	157,250!
298	return -1;	×
299	}
300	if (delay <= 0) {	157,250✔
301	diag_set(TimedOut);	840!
302	return -1;	840✔
303	}
304	/*
305	* Yield control to other fibers until the
306	* timeout is being reached.
307	*/
308	coio_wait(io->fd, iostream_status_to_events(nrd), delay);	156,410!
309	if (fiber_is_cancelled()) {	155,556✔
310	diag_set(FiberIsCancelled);	12,543!
311	return -1;	12,543✔
312	}
313	coio_timeout_update(&start, &delay);	143,013!
314	}
315	}
316
317	/**
318	* Read at least sz bytes, with readahead.
319	*
320	* Treats EOF as an error, and throws an exception.
321	*
322	* @retval the number of bytes read, > 0.
323	* @retval -1 error.
324	*/
325	ssize_t
326	coio_readn_ahead(struct iostream io, void buf, size_t sz, size_t bufsiz)	24,207✔
327	{
328	ssize_t nrd = coio_read_ahead(io, buf, sz, bufsiz);	24,207✔
329	if (nrd >= 0 && nrd < (ssize_t)sz) {	24,183✔
330	errno = EPIPE;	6✔
331	diag_set(SocketError, sio_socketname(io->fd),	6!
332	"unexpected EOF when reading from socket");
333	return -1;	6✔
334	}
335	return nrd;	24,177✔
336	}
337
338	/**
339	* Read at least sz bytes, with readahead and timeout.
340	*
341	* Treats EOF as an error, and throws an exception.
342	*
343	* @retval the number of bytes read, > 0.
344	* @retval -1 error.
345	*/
346	ssize_t
347	coio_readn_ahead_timeout(struct iostream io, void buf, size_t sz,	110,637✔
348	size_t bufsiz, ev_tstamp timeout)
349	{
350	ssize_t nrd = coio_read_ahead_timeout(io, buf, sz, bufsiz, timeout);	110,637✔
351	if (nrd >= 0 && nrd < (ssize_t)sz) { /* EOF. */	109,807✔
352	errno = EPIPE;	1,213✔
353	diag_set(SocketError, sio_socketname(io->fd),	1,213!
354	"unexpected EOF when reading from socket");
355	return -1;	1,213✔
356	}
357	return nrd;	108,594✔
358	}
359
360	/**
361	* Write sz bytes to socket.
362	*
363	* Throws SocketError in case of write error. If
364	* the socket is not ready, yields the current
365	* fiber until the socket becomes ready, until
366	* all data is written.
367	*
368	* @retval the number of bytes written. Always
369	* equal to @a sz.
370	* @retval -1 error.
371	*/
372	ssize_t
373	coio_write_timeout(struct iostream io, const void buf, size_t sz,	417✔
374	ev_tstamp timeout)
375	{
376	ssize_t towrite = sz;	417✔
377	ev_tstamp start, delay;
378	coio_timeout_init(&start, &delay, timeout);	417!
379	while (true) {	2✔
380	/*
381	* Sic: write as much data as possible,
382	* assuming the socket is ready.
383	*/
384	ssize_t nwr = iostream_write(io, buf, towrite);	419!
385	if (nwr >= 0) {	419✔
386	/* Go past the data just written. */
387	if (nwr >= towrite)	417!
388	return sz;	417✔
389	towrite -= nwr;	×
390	buf = (char *) buf + nwr;	×
391	continue;	×
392	} else if (nwr == IOSTREAM_ERROR) {	2!
393	return -1;	×
394	}
395	if (delay <= 0) {	2!
396	diag_set(TimedOut);	×
397	return -1;	×
398	}
399	/*
400	* Yield control to other fibers until the
401	* timeout is reached or the socket is
402	* ready.
403	*/
404	coio_wait(io->fd, iostream_status_to_events(nwr), delay);	2!
405	if (fiber_is_cancelled()) {	2!
406	diag_set(FiberIsCancelled);	×
407	return -1;	×
408	}
409	coio_timeout_update(&start, &delay);	2!
410	}
411	}
412
413	/*
414	* Write iov using iostream API.
415	* Put in an own function to workaround gcc bug with @finally
416	*/
417	static inline ssize_t
418	coio_flush(struct iostream io, struct iovec iov, ssize_t offset, int iovcnt)	331,837✔
419	{
420	sio_add_to_iov(iov, -offset);	331,837✔
421	ssize_t nwr = iostream_writev(io, iov, iovcnt);	331,837✔
422	sio_add_to_iov(iov, offset);	331,835✔
423	return nwr;	331,835✔
424	}
425
426	ssize_t
427	coio_writev_timeout(struct iostream io, struct iovec iov, int iovcnt,	331,189✔
428	size_t size_hint, ev_tstamp timeout)
429	{
430	size_t total = 0;	331,189✔
431	size_t iov_len = 0;	331,189✔
432	struct iovec *end = iov + iovcnt;	331,189✔
433	ev_tstamp start, delay;
434	coio_timeout_init(&start, &delay, timeout);	331,189!
435
436	/* Avoid a syscall in case of 0 iovcnt. */
437	while (iov < end) {	331,837!
438	/* Write as much data as possible. */
439	ssize_t nwr = coio_flush(io, iov, iov_len, end - iov);	331,837✔
440	if (nwr >= 0) {	331,835✔
441	total += nwr;	330,748✔
442	/*
443	* If there was a hint for the total size
444	* of the vector, use it.
445	*/
446	if (size_hint > 0 && size_hint == total)	330,748!
447	break;	×
448
449	iov += sio_move_iov(iov, nwr, &iov_len);	330,748!
450	if (iov == end) {	330,748!
451	assert(iov_len == 0);	330,748!
452	break;	330,748✔
453	}
454	continue;	×
455	} else if (nwr == IOSTREAM_ERROR) {	1,087✔
456	return -1;	331,187✔
457	}
458	if (delay <= 0) {	650!
459	diag_set(TimedOut);	×
460	return -1;	×
461	}
462	/*
463	* Yield control to other fibers until the
464	* timeout is reached or the socket is
465	* ready.
466	*/
467	coio_wait(io->fd, iostream_status_to_events(nwr), delay);	650!
468	if (fiber_is_cancelled()) {	650✔
469	diag_set(FiberIsCancelled);	2!
470	return -1;	2✔
471	}
472	coio_timeout_update(&start, &delay);	648!
473	}
474	return total;	330,748✔
475	}
476
477	void
478	coio_stat_init(ev_stat stat, const char path)	2✔
479	{
480	ev_stat_init(stat, (ev_stat_cb) fiber_schedule_cb, path, 0.0);	2✔
481	}	2✔
482
483	int
484	coio_stat_stat_timeout(ev_stat *stat, ev_tstamp timeout)	2✔
485	{
486	stat->data = fiber();	2!
487	ev_stat_start(loop(), stat);	2!
488	ev_tstamp start, delay;
489	coio_timeout_init(&start, &delay, timeout);	2!
490	fiber_yield_timeout(delay);	2!
491	ev_stat_stop(loop(), stat);	2!
492	if (fiber_is_cancelled()) {	2!
493	diag_set(FiberIsCancelled);	×
494	return -1;	2✔
495	}
496	return 0;	2✔
497	}
498
499	/**
500	* The process status change callback.
501	* Similar to fiber_schedule_cb, but also set watcher->data to NULL to indicate
502	* that the fiber has been woken up on the child process termination.
503	*/
504	static void
505	coio_status_change_cb(ev_loop loop, ev_child watcher, int revents)	1✔
506	{
507	(void)loop;
508	(void)revents;
509	struct fiber *fiber = watcher->data;	1✔
510	assert(fiber() == &cord()->sched);	1!
511	watcher->data = NULL;	1✔
512	fiber_wakeup(fiber);	1✔
513	}	1✔
514
515	int
516	coio_waitpid(pid_t pid, int *status)	1✔
517	{
518	assert(cord_is_main());	1!
519	ev_child cw;
520	ev_init(&cw, coio_status_change_cb);	1✔
521	ev_child_set(&cw, pid, 0);	1✔
522	cw.data = fiber();	1!
523	ev_child_start(loop(), &cw);	1!
524
525	do {
526	fiber_yield();	2✔
527	} while (cw.data != NULL);	2✔
528
529	ev_child_stop(loop(), &cw);	1!
530	if (fiber_is_cancelled()) {	1!
531	diag_set(FiberIsCancelled);	×
532	return -1;	1✔
533	}
534	*status = cw.rstatus;	1✔
535	return 0;	1✔
536	}
537
538	/* Values of COIO_READ(WRITE) must equal to EV_READ(WRITE) */
539	static_assert(COIO_READ == (int) EV_READ, "TNT_IO_READ");
540	static_assert(COIO_WRITE == (int) EV_WRITE, "TNT_IO_WRITE");
541
542	struct coio_wdata {
543	struct fiber *fiber;
544	int revents;
545	};
546
547	static void
548	coio_wait_cb(struct ev_loop loop, ev_io watcher, int revents)	321,745✔
549	{
550	(void) loop;
551	struct coio_wdata wdata = (struct coio_wdata ) watcher->data;	321,745✔
552	wdata->revents = revents;	321,745✔
553	fiber_wakeup(wdata->fiber);	321,745✔
554	}	321,745✔
555
556	API_EXPORT int
557	coio_wait(int fd, int events, double timeout)	770,969✔
558	{
559	if (fiber_is_cancelled())	770,969✔
560	return 0;	762,319✔
561	struct ev_io io;
562	ev_io_init(&io, coio_wait_cb, fd, events);	770,915✔
563	struct coio_wdata wdata = {	770,915✔
564	/* .fiber = */ fiber(),	770,915!
565	/* .revents = */ 0
566	};
567	io.data = &wdata;	770,915✔
568
569	/* A special hack to work with zero timeout */
570	ev_set_priority(&io, EV_MAXPRI);	770,915✔
571	ev_io_start(loop(), &io);	770,915!
572
573	fiber_yield_timeout(timeout);	770,915!
574
575	ev_io_stop(loop(), &io);	762,265!
576	return wdata.revents & (EV_READ \| EV_WRITE);	762,265✔
577	}
578
579	API_EXPORT int
580	coio_close(int fd)	72,842✔
581	{
582	ev_io_closing(loop(), fd);	72,842!
583	return close(fd);	72,842✔
584	}

tarantool / tarantool / 10899922320

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