From: Serge Petrenko via Tarantool-patches <tarantool-patches@dev.tarantool.org> To: Vladislav Shpilevoy <v.shpilevoy@tarantool.org>, vdavydov@tarantool.org Cc: tarantool-patches@dev.tarantool.org Subject: Re: [Tarantool-patches] [PATCH] coio: handle spurious wakeup correctly Date: Fri, 19 Nov 2021 13:30:14 +0300 [thread overview] Message-ID: <73c8ec91-322f-7f67-c822-e5409ce2e999@tarantool.org> (raw) In-Reply-To: <e2fe6cf4-c5d3-dbe0-2b72-55be70f8f170@tarantool.org> 18.11.2021 02:11, Vladislav Shpilevoy пишет: >>> On 16.11.2021 11:17, Serge Petrenko via Tarantool-patches wrote: >>>> coio_accept, coio_read, coio_write, coio_writev used to handle spurious >>>> wakeups correctly: if the timeout hasn't passed yet, they would simply >>>> retry reading (or writing) and fall asleep once again if no data is >>>> ready. >>>> >>>> This behaviour changed in the following patches: >>>> 577a640a7fdec986d19101ed04d2afa80e951c78 ("coio: pass fd to >>>> coio_accept") and 4f84859dcdd6126b0bdcda810b7f5f58386bdac6 ("Introduce >>>> iostream wrapper for socket I/O"). >>>> >>>> Now the functions timeout on the very first spurious wakeup. >>>> >>>> Fix this, add the appropriate unit tests and a test_iostream >>>> implementation for the ease of testing. >>> Don't we have the same problem with coio_connect_addr() (used in >>> coio_connect_timeout())? >> Not really. Neither coio_connect_addr() nor coio_connect_timeout() retry >> the connection. So even the previous version would throw an error after >> a spurious wakeup. Just the error would be different. >> Before the change it would throw SocketError, not TimedOut, but I don't think >> this matters much. Does it? >> >> By the change I mean (2db0741b) "coio: return fd from coio_connect". > Ok, then never mind. I will wait for fixes of Vova's comments regarding the > tests, but the bugfix itself LGTM. Thanks for your answer! Here's the incremental diff after the fixes. Long story short, I've deleted the test_iostream wrapper. There wasn't much use for it. I've force-pushed the new version on the branch. =============================== diff --git a/test/unit/CMakeLists.txt b/test/unit/CMakeLists.txt index 71c8ba981..d8b1aac0d 100644 --- a/test/unit/CMakeLists.txt +++ b/test/unit/CMakeLists.txt @@ -121,7 +121,7 @@ if (GLIBC_USED) target_link_libraries(cbus_hang.test core unit stat) endif () -add_executable(coio.test coio.cc core_test_utils.c test_iostream.c) +add_executable(coio.test coio.cc core_test_utils.c) target_link_libraries(coio.test core eio bit uri unit) if (ENABLE_BUNDLED_MSGPUCK) diff --git a/test/unit/coio.cc b/test/unit/coio.cc index 194503fda..d55bc2347 100644 --- a/test/unit/coio.cc +++ b/test/unit/coio.cc @@ -4,7 +4,11 @@ #include "coio_task.h" #include "fio.h" #include "unit.h" -#include "test_iostream.h" +#include "iostream.h" + +#include <fcntl.h> +#include <sys/uio.h> +#include <sys/errno.h> int touch_f(va_list ap) @@ -137,7 +141,7 @@ static int test_read_f(va_list ap) { struct iostream *io = va_arg(ap, struct iostream *); - char buf[5]; + char buf[1024]; int rc = coio_read(io, buf, sizeof(buf)); if (rc < (ssize_t)sizeof(buf)) return -1; @@ -148,9 +152,9 @@ static int test_write_f(va_list ap) { struct iostream *io = va_arg(ap, struct iostream *); - const char str[] = "test"; - int rc = coio_write_timeout(io, str, sizeof(str), TIMEOUT_INFINITY); - if (rc < (ssize_t)sizeof(str)) + char buf[1024] = ""; + int rc = coio_write_timeout(io, buf, sizeof(buf), TIMEOUT_INFINITY); + if (rc < (ssize_t)sizeof(buf)) return -1; return 0; } @@ -159,14 +163,45 @@ static int test_writev_f(va_list ap) { struct iostream *io = va_arg(ap, struct iostream *); - const char str[] = "test"; - struct iovec iov = {(void *)str, sizeof(str)}; + char buf[1024] = ""; + struct iovec iov = {(void *)buf, sizeof(buf)}; int rc = coio_writev(io, &iov, 1, 0); - if (rc < (ssize_t)sizeof(str)) + if (rc < (ssize_t)sizeof(buf)) return -1; return 0; } +static void +fill_pipe(int fd) +{ + char buf[1024] = ""; + int rc = 0; + while (rc >= 0 || errno == EINTR) + rc = write(fd, buf, sizeof(buf)); + fail_unless(errno == EAGAIN || errno == EWOULDBLOCK); +} + +static void +empty_pipe(int fd) +{ + char buf[1024]; + int rc = 0; + while (rc >= 0 || errno == EINTR) + rc = read(fd, buf, sizeof(buf)); + fail_unless(errno == EAGAIN || errno == EWOULDBLOCK); +} + +static void +create_pipe(int fds[2]) +{ + int rc = pipe(fds); + fail_unless(rc >= 0); + rc = fcntl(fds[0], F_SETFL, O_NONBLOCK); + fail_unless(rc >= 0); + rc = fcntl(fds[1], F_SETFL, O_NONBLOCK); + fail_unless(rc >= 0); +} + static void read_write_test(void) { @@ -186,25 +221,36 @@ read_write_test(void) int num_tests = sizeof(test_funcs) / sizeof(test_funcs[0]); plan(2 * num_tests); - struct test_stream s; - test_stream_create(&s, 0, 0); + int fds[2]; + create_pipe(fds); for (int i = 0; i < num_tests; i++) { - test_stream_reset(&s, 0, 0); + struct iostream io; + if (i == 0) { + /* A non-readable fd, since the pipe is empty. */ + iostream_create(&io, fds[0]); + } else { + iostream_create(&io, fds[1]); + /* Make the fd non-writable. */ + fill_pipe(fds[1]); + } struct fiber *f = fiber_new_xc("rw_test", test_funcs[i]); fiber_set_joinable(f, true); - fiber_start(f, &s.io); + fiber_start(f, &io); fiber_wakeup(f); fiber_sleep(0); ok(!fiber_is_dead(f), "coio_%s handle spurious wakeup", descr[i]); - test_stream_reset(&s, INT_MAX, INT_MAX); - fiber_wakeup(f); + if (i == 0) + fill_pipe(fds[1]); + else + empty_pipe(fds[0]); int rc = fiber_join(f); ok(rc == 0, "coio_%s success after a spurious wakeup", descr[i]); + iostream_destroy(&io); } - test_stream_destroy(&s); - + close(fds[0]); + close(fds[1]); check_plan(); footer(); } diff --git a/test/unit/test_iostream.c b/test/unit/test_iostream.c deleted file mode 100644 index c50b63944..000000000 --- a/test/unit/test_iostream.c +++ /dev/null @@ -1,125 +0,0 @@ -/* - * SPDX-License-Identifier: BSD-2-Clause - * - * Copyright 2010-2021, Tarantool AUTHORS, please see AUTHORS file. - */ - -#include <sys/uio.h> -#include <fcntl.h> -#include <sys/errno.h> - -#include "test_iostream.h" -#include "trivia/util.h" - -static ssize_t -test_stream_read(struct iostream *io, void *buf, size_t count) -{ - struct test_stream_ctx *ctx = io->ctx; - io->fd = ctx->rfd; - (void)buf; - ssize_t nrd = 0; - if (ctx->avail_rd == 0) - return IOSTREAM_WANT_READ; - nrd = MIN(count, ctx->avail_rd); - ctx->avail_rd -= nrd; - return nrd; -} - -static ssize_t -test_stream_write(struct iostream *io, const void *buf, size_t count) -{ - assert(io->fd >= 0); - struct test_stream_ctx *ctx = io->ctx; - io->fd = ctx->wfd; - (void)buf; - ssize_t nwr = 0; - if (ctx->avail_wr == 0) - return IOSTREAM_WANT_WRITE; - nwr = MIN(count, ctx->avail_wr); - ctx->avail_wr -= nwr; - return nwr; -} - -static ssize_t -test_stream_writev(struct iostream *io, const struct iovec *iov, int iovcnt) -{ - assert(io->fd >= 0); - struct test_stream_ctx *ctx = io->ctx; - io->fd = ctx->wfd; - if (ctx->avail_wr == 0) - return IOSTREAM_WANT_WRITE; - ssize_t start_wr = ctx->avail_wr; - for (int i = 0; i < iovcnt && ctx->avail_wr > 0; i++) { - size_t nwr = MIN(iov[i].iov_len, ctx->avail_wr); - ctx->avail_wr -= nwr; - } - return start_wr - ctx->avail_wr; -} - -static void -test_stream_delete_ctx(void *ptr) -{ - struct test_stream_ctx *ctx = ptr; - close(ctx->rfds[0]); - close(ctx->rfds[1]); - close(ctx->wfds[0]); - close(ctx->wfds[1]); -} - -static const struct iostream_vtab test_stream_vtab = { - test_stream_delete_ctx, - test_stream_read, - test_stream_write, - test_stream_writev, -}; - -static void -fill_pipe(int fd) -{ - char buf[4096]; - int rc = 0; - rc = fcntl(fd, F_SETFL, O_NONBLOCK); - assert(rc >= 0); - while (rc >= 0 || errno == EINTR) - rc = write(fd, buf, sizeof(buf)); - assert(errno == EAGAIN || errno == EWOULDBLOCK); -} - -static void -test_stream_ctx_create(struct test_stream_ctx *ctx, size_t maxrd, size_t maxwr) -{ - ctx->avail_rd = maxrd; - ctx->avail_wr = maxwr; - int rc = pipe(ctx->rfds); - assert(rc == 0); - rc = pipe(ctx->wfds); - assert(rc == 0); - /* rfd is never readable, since we didn't write anything to the pipe. */ - ctx->rfd = ctx->rfds[0]; - ctx->wfd = ctx->wfds[1]; - /* Make wfd never writeable by filling the pipe. */ - fill_pipe(ctx->wfd); -} - -void -test_stream_create(struct test_stream *s, size_t maxrd, size_t maxwr) -{ - test_stream_ctx_create(&s->ctx, maxrd, maxwr); - s->io.ctx = &s->ctx; - s->io.vtab = &test_stream_vtab; - s->io.fd = -1; /* Initialized on demand. */ -} - -void -test_stream_reset(struct test_stream *s, size_t maxrd, size_t maxwr) -{ - struct test_stream_ctx *ctx = s->io.ctx; - ctx->avail_rd = maxrd; - ctx->avail_wr = maxwr; -} - -void -test_stream_destroy(struct test_stream *s) -{ - iostream_destroy(&s->io); -} diff --git a/test/unit/test_iostream.h b/test/unit/test_iostream.h deleted file mode 100644 index fd8b998aa..000000000 --- a/test/unit/test_iostream.h +++ /dev/null @@ -1,69 +0,0 @@ -/* - * SPDX-License-Identifier: BSD-2-Clause - * - * Copyright 2010-2021, Tarantool AUTHORS, please see AUTHORS file. - */ - -#pragma once - -#include "core/iostream.h" -/** - * A testing stream implementation. Doesn't write anything anywhere, but is - * useful for testing coio behaviour when waiting for stream to become readable - * or writeable. - */ - -#if defined (__cplusplus) -extern "C" { -#endif - -struct test_stream_ctx { - /** How many bytes can be written before the stream would block. */ - size_t avail_wr; - /** How many bytes can be read before the stream would block. */ - size_t avail_rd; - /** - * A file descriptor which is never ready for reads. Points to rfds[0], - * the reading end of a pipe which is never written to. - */ - int rfd; - /** A pipe used to simulate a fd never ready for reads. */ - int rfds[2]; - /** - * A file descriptor which is never ready for writes. Points to - * wfds[1], the writing end of a filled up pipe. - */ - int wfd; - /** A pipe to simulate a fd never ready for writes. */ - int wfds[2]; -}; - -struct test_stream { - struct iostream io; - struct test_stream_ctx ctx; -}; - -/** - * Create an instance of a testing iostream. - * The stream will allow reading up to \a maxrd and writing up to \a maxwr - * bytes before blocking. - */ -void -test_stream_create(struct test_stream *s, size_t maxrd, size_t maxwr); - -/** - * Reset the stream. Allow additional \a maxrd and \a maxwr bytes of input / - * output respectively. - */ -void -test_stream_reset(struct test_stream *s, size_t maxrd, size_t maxwr); - -/** - * Destroy the stream. - */ -void -test_stream_destroy(struct test_stream *s); - -#if defined (__cplusplus) -} /* extern "C" */ -#endif =============================== -- Serge Petrenko
next prev parent reply other threads:[~2021-11-19 10:30 UTC|newest] Thread overview: 9+ messages / expand[flat|nested] mbox.gz Atom feed top 2021-11-16 10:17 Serge Petrenko via Tarantool-patches 2021-11-16 11:29 ` Vladimir Davydov via Tarantool-patches 2021-11-16 11:36 ` Serge Petrenko via Tarantool-patches 2021-11-16 22:49 ` Vladislav Shpilevoy via Tarantool-patches 2021-11-17 7:33 ` Serge Petrenko via Tarantool-patches 2021-11-17 23:11 ` Vladislav Shpilevoy via Tarantool-patches 2021-11-19 10:30 ` Serge Petrenko via Tarantool-patches [this message] 2021-11-21 14:24 ` Vladislav Shpilevoy via Tarantool-patches 2021-11-22 6:47 ` Serge Petrenko via Tarantool-patches
Reply instructions: You may reply publicly to this message via plain-text email using any one of the following methods: * Save the following mbox file, import it into your mail client, and reply-to-all from there: mbox Avoid top-posting and favor interleaved quoting: https://en.wikipedia.org/wiki/Posting_style#Interleaved_style * Reply using the --to, --cc, and --in-reply-to switches of git-send-email(1): git send-email \ --in-reply-to=73c8ec91-322f-7f67-c822-e5409ce2e999@tarantool.org \ --to=tarantool-patches@dev.tarantool.org \ --cc=sergepetrenko@tarantool.org \ --cc=v.shpilevoy@tarantool.org \ --cc=vdavydov@tarantool.org \ --subject='Re: [Tarantool-patches] [PATCH] coio: handle spurious wakeup correctly' \ /path/to/YOUR_REPLY https://kernel.org/pub/software/scm/git/docs/git-send-email.html * If your mail client supports setting the In-Reply-To header via mailto: links, try the mailto: link
This is a public inbox, see mirroring instructions for how to clone and mirror all data and code used for this inbox