From d731fd1bd9400eab4ee9bf11bda9f64951fc4d89 Mon Sep 17 00:00:00 2001 From: CurlyMoo Date: Fri, 9 Sep 2016 00:42:43 +0200 Subject: [PATCH] poll: add support for OOB TCP and GPIO interrupts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Out-of-band TCP messages are used for TCP data transmission outside (outband) the inbound TCP data. These packets are sent with an "urgent pointer", but previously discarded. Additionally, when using (e)poll a POLLPRI is triggered when an interrupt signal is received on GPIO capable systems such as the Raspberry Pi. PR-URL: https://github.com/libuv/libuv/pull/1040 Reviewed-By: Ben Noordhuis Reviewed-By: Saúl Ibarra Corretgé --- Makefile.am | 3 +- docs/src/poll.rst | 15 ++-- include/uv.h | 3 +- src/unix/core.c | 8 +- src/unix/internal.h | 6 ++ src/unix/kqueue.c | 35 ++++++++ src/unix/linux-core.c | 2 +- src/unix/poll.c | 24 ++++- test/test-list.h | 9 ++ test/test-poll-oob.c | 205 ++++++++++++++++++++++++++++++++++++++++++ uv.gyp | 1 + 11 files changed, 295 insertions(+), 16 deletions(-) create mode 100644 test/test-poll-oob.c diff --git a/Makefile.am b/Makefile.am index 404674ba..6bf7d234 100644 --- a/Makefile.am +++ b/Makefile.am @@ -212,10 +212,11 @@ test_run_tests_SOURCES = test/blackhole-server.c \ test/test-pipe-close-stdout-read-stdin.c \ test/test-pipe-set-non-blocking.c \ test/test-platform-output.c \ + test/test-poll.c \ test/test-poll-close.c \ test/test-poll-close-doesnt-corrupt-stack.c \ test/test-poll-closesocket.c \ - test/test-poll.c \ + test/test-poll-oob.c \ test/test-process-title.c \ test/test-queue-foreach-delete.c \ test/test-ref.c \ diff --git a/docs/src/poll.rst b/docs/src/poll.rst index 004ff4b9..aba89158 100644 --- a/docs/src/poll.rst +++ b/docs/src/poll.rst @@ -54,7 +54,8 @@ Data types enum uv_poll_event { UV_READABLE = 1, UV_WRITABLE = 2, - UV_DISCONNECT = 4 + UV_DISCONNECT = 4, + UV_PRIORITIZED = 8 }; @@ -84,10 +85,13 @@ API .. c:function:: int uv_poll_start(uv_poll_t* handle, int events, uv_poll_cb cb) - Starts polling the file descriptor. `events` is a bitmask consisting made up - of UV_READABLE, UV_WRITABLE and UV_DISCONNECT. As soon as an event is detected - the callback will be called with `status` set to 0, and the detected events set on the - `events` field. + Starts polling the file descriptor. `events` is a bitmask made up of + UV_READABLE, UV_WRITABLE, UV_PRIORITIZED and UV_DISCONNECT. As soon as an + event is detected the callback will be called with `status` set to 0, and the + detected events set on the `events` field. + + The UV_PRIORITIZED event is used to watch for sysfs interrupts or TCP out-of-band + messages. The UV_DISCONNECT event is optional in the sense that it may not be reported and the user is free to ignore it, but it can help optimize the shutdown @@ -108,6 +112,7 @@ API on the `events` field in the callback. .. versionchanged:: 1.9.0 Added the UV_DISCONNECT event. + .. versionchanged:: 1.14.0 Added the UV_PRIORITIZED event. .. c:function:: int uv_poll_stop(uv_poll_t* poll) diff --git a/include/uv.h b/include/uv.h index f076094c..af500cb0 100644 --- a/include/uv.h +++ b/include/uv.h @@ -719,7 +719,8 @@ struct uv_poll_s { enum uv_poll_event { UV_READABLE = 1, UV_WRITABLE = 2, - UV_DISCONNECT = 4 + UV_DISCONNECT = 4, + UV_PRIORITIZED = 8 }; UV_EXTERN int uv_poll_init(uv_loop_t* loop, uv_poll_t* handle, int fd); diff --git a/src/unix/core.c b/src/unix/core.c index a0513451..bee641cb 100644 --- a/src/unix/core.c +++ b/src/unix/core.c @@ -838,7 +838,7 @@ void uv__io_init(uv__io_t* w, uv__io_cb cb, int fd) { void uv__io_start(uv_loop_t* loop, uv__io_t* w, unsigned int events) { - assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP))); + assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI))); assert(0 != events); assert(w->fd >= 0); assert(w->fd < INT_MAX); @@ -866,7 +866,7 @@ void uv__io_start(uv_loop_t* loop, uv__io_t* w, unsigned int events) { void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events) { - assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP))); + assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI))); assert(0 != events); if (w->fd == -1) @@ -898,7 +898,7 @@ void uv__io_stop(uv_loop_t* loop, uv__io_t* w, unsigned int events) { void uv__io_close(uv_loop_t* loop, uv__io_t* w) { - uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP); + uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI); QUEUE_REMOVE(&w->pending_queue); /* Remove stale events for this file descriptor */ @@ -913,7 +913,7 @@ void uv__io_feed(uv_loop_t* loop, uv__io_t* w) { int uv__io_active(const uv__io_t* w, unsigned int events) { - assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP))); + assert(0 == (events & ~(POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI))); assert(0 != events); return 0 != (w->pevents & events); } diff --git a/src/unix/internal.h b/src/unix/internal.h index 2e3afa6c..283c1dfa 100644 --- a/src/unix/internal.h +++ b/src/unix/internal.h @@ -110,6 +110,12 @@ int uv__pthread_sigmask(int how, const sigset_t* set, sigset_t* oset); # define UV__POLLRDHUP 0x2000 #endif +#ifdef POLLPRI +# define UV__POLLPRI POLLPRI +#else +# define UV__POLLPRI 0 +#endif + #if !defined(O_CLOEXEC) && defined(__FreeBSD__) /* * It may be that we are just missing `__POSIX_VISIBLE >= 200809`. diff --git a/src/unix/kqueue.c b/src/unix/kqueue.c index 6bc60bbe..300bac07 100644 --- a/src/unix/kqueue.c +++ b/src/unix/kqueue.c @@ -34,6 +34,17 @@ #include #include +/* + * Required on + * - Until at least FreeBSD 11.0 + * - Older versions of Mac OS X + * + * http://www.boost.org/doc/libs/1_61_0/boost/asio/detail/kqueue_reactor.hpp + */ +#ifndef EV_OOBAND +#define EV_OOBAND EV_FLAG1 +#endif + static void uv__fs_event(uv_loop_t* loop, uv__io_t* w, unsigned int fflags); @@ -166,6 +177,16 @@ void uv__io_poll(uv_loop_t* loop, int timeout) { } } + if ((w->events & UV__POLLPRI) == 0 && (w->pevents & UV__POLLPRI) != 0) { + EV_SET(events + nevents, w->fd, EV_OOBAND, EV_ADD, 0, 0, 0); + + if (++nevents == ARRAY_SIZE(events)) { + if (kevent(loop->backend_fd, events, nevents, NULL, 0, NULL)) + abort(); + nevents = 0; + } + } + w->events = w->pevents; } @@ -275,6 +296,20 @@ void uv__io_poll(uv_loop_t* loop, int timeout) { } } + if (ev->filter == EV_OOBAND) { + if (w->pevents & UV__POLLPRI) { + revents |= UV__POLLPRI; + w->rcount = ev->data; + } else { + /* TODO batch up */ + struct kevent events[1]; + EV_SET(events + 0, fd, ev->filter, EV_DELETE, 0, 0, 0); + if (kevent(loop->backend_fd, events, 1, NULL, 0, NULL)) + if (errno != ENOENT) + abort(); + } + } + if (ev->filter == EVFILT_WRITE) { if (w->pevents & POLLOUT) { revents |= POLLOUT; diff --git a/src/unix/linux-core.c b/src/unix/linux-core.c index 2866e938..24ae9490 100644 --- a/src/unix/linux-core.c +++ b/src/unix/linux-core.c @@ -388,7 +388,7 @@ void uv__io_poll(uv_loop_t* loop, int timeout) { * free when we switch over to edge-triggered I/O. */ if (pe->events == POLLERR || pe->events == POLLHUP) - pe->events |= w->pevents & (POLLIN | POLLOUT); + pe->events |= w->pevents & (POLLIN | POLLOUT | UV__POLLPRI); if (pe->events != 0) { /* Run signal watchers last. This also affects child process watchers diff --git a/src/unix/poll.c b/src/unix/poll.c index 370994bd..2001fc8c 100644 --- a/src/unix/poll.c +++ b/src/unix/poll.c @@ -33,8 +33,19 @@ static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { handle = container_of(w, uv_poll_t, io_watcher); - if (events & POLLERR) { - uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP); + /* + * As documented in the kernel source fs/kernfs/file.c #780 + * poll will return POLLERR|POLLPRI in case of sysfs + * polling. This does not happen in case of out-of-band + * TCP messages. + * + * The above is the case on (at least) FreeBSD and Linux. + * + * So to properly determine a POLLPRI or a POLLERR we need + * to check for both. + */ + if ((events & POLLERR) && !(events & UV__POLLPRI)) { + uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI); uv__handle_stop(handle); handle->poll_cb(handle, -EBADF, 0); return; @@ -43,6 +54,8 @@ static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) { pevents = 0; if (events & POLLIN) pevents |= UV_READABLE; + if (events & UV__POLLPRI) + pevents |= UV_PRIORITIZED; if (events & POLLOUT) pevents |= UV_WRITABLE; if (events & UV__POLLRDHUP) @@ -86,7 +99,7 @@ int uv_poll_init_socket(uv_loop_t* loop, uv_poll_t* handle, static void uv__poll_stop(uv_poll_t* handle) { uv__io_stop(handle->loop, &handle->io_watcher, - POLLIN | POLLOUT | UV__POLLRDHUP); + POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI); uv__handle_stop(handle); } @@ -101,7 +114,8 @@ int uv_poll_stop(uv_poll_t* handle) { int uv_poll_start(uv_poll_t* handle, int pevents, uv_poll_cb poll_cb) { int events; - assert((pevents & ~(UV_READABLE | UV_WRITABLE | UV_DISCONNECT)) == 0); + assert((pevents & ~(UV_READABLE | UV_WRITABLE | UV_DISCONNECT | + UV_PRIORITIZED)) == 0); assert(!uv__is_closing(handle)); uv__poll_stop(handle); @@ -112,6 +126,8 @@ int uv_poll_start(uv_poll_t* handle, int pevents, uv_poll_cb poll_cb) { events = 0; if (pevents & UV_READABLE) events |= POLLIN; + if (pevents & UV_PRIORITIZED) + events |= UV__POLLPRI; if (pevents & UV_WRITABLE) events |= POLLOUT; if (pevents & UV_DISCONNECT) diff --git a/test/test-list.h b/test/test-list.h index 0c32d84d..848d0b5d 100644 --- a/test/test-list.h +++ b/test/test-list.h @@ -328,6 +328,10 @@ TEST_DECLARE (thread_rwlock_trylock) TEST_DECLARE (thread_create) TEST_DECLARE (thread_equal) TEST_DECLARE (dlerror) +#if (defined(__unix__) || (defined(__APPLE__) && defined(__MACH__))) && \ + !defined(__sun) +TEST_DECLARE (poll_oob) +#endif TEST_DECLARE (poll_duplex) TEST_DECLARE (poll_unidirectional) TEST_DECLARE (poll_close) @@ -681,6 +685,11 @@ TASK_LIST_START TEST_ENTRY (poll_unidirectional) TEST_ENTRY (poll_close) TEST_ENTRY (poll_bad_fdtype) +#if (defined(__unix__) || (defined(__APPLE__) && defined(__MACH__))) && \ + !defined(__sun) + TEST_ENTRY (poll_oob) +#endif + #ifdef __linux__ TEST_ENTRY (poll_nested_epoll) #endif diff --git a/test/test-poll-oob.c b/test/test-poll-oob.c new file mode 100644 index 00000000..2a6da843 --- /dev/null +++ b/test/test-poll-oob.c @@ -0,0 +1,205 @@ +/* Copyright libuv project contributors. All rights reserved. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to + * deal in the Software without restriction, including without limitation the + * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or + * sell copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS + * IN THE SOFTWARE. + */ + +#if !defined(_WIN32) + +#include "uv.h" +#include "task.h" + +#include +#include +#include +#include +#include + +static uv_tcp_t server_handle; +static uv_tcp_t client_handle; +static uv_tcp_t peer_handle; +static uv_poll_t poll_req[2]; +static uv_idle_t idle; +static uv_os_fd_t client_fd; +static uv_os_fd_t server_fd; +static int ticks; +static const int kMaxTicks = 10; +static int cli_pr_check = 0; +static int cli_rd_check = 0; +static int srv_rd_check = 0; + +static int got_eagain(void) { + return errno == EAGAIN + || errno == EINPROGRESS +#ifdef EWOULDBLOCK + || errno == EWOULDBLOCK +#endif + ; +} + +static void idle_cb(uv_idle_t* idle) { + uv_sleep(100); + if (++ticks < kMaxTicks) + return; + + uv_poll_stop(&poll_req[0]); + uv_poll_stop(&poll_req[1]); + uv_close((uv_handle_t*) &server_handle, NULL); + uv_close((uv_handle_t*) &client_handle, NULL); + uv_close((uv_handle_t*) &peer_handle, NULL); + uv_close((uv_handle_t*) idle, NULL); +} + +static void poll_cb(uv_poll_t* handle, int status, int events) { + char buffer[5]; + int n; + int fd; + + ASSERT(0 == uv_fileno((uv_handle_t*)handle, &fd)); + memset(buffer, 0, 5); + + if (events & UV_PRIORITIZED) { + do + n = recv(client_fd, &buffer, 5, MSG_OOB); + while (n == -1 && errno == EINTR); + ASSERT(n >= 0 || errno != EINVAL); + cli_pr_check = 1; + ASSERT(0 == uv_poll_stop(&poll_req[0])); + ASSERT(0 == uv_poll_start(&poll_req[0], + UV_READABLE | UV_WRITABLE, + poll_cb)); + } + if (events & UV_READABLE) { + if (fd == client_fd) { + do + n = recv(client_fd, &buffer, 5, 0); + while (n == -1 && errno == EINTR); + ASSERT(n >= 0 || errno != EINVAL); + if (cli_rd_check == 1) { + ASSERT(strncmp(buffer, "world", n) == 0); + ASSERT(5 == n); + cli_rd_check = 2; + } + if (cli_rd_check == 0) { + ASSERT(n == 4); + ASSERT(strncmp(buffer, "hello", n) == 0); + cli_rd_check = 1; + do { + do + n = recv(server_fd, &buffer, 5, 0); + while (n == -1 && errno == EINTR); + if (n > 0) { + ASSERT(n == 5); + ASSERT(strncmp(buffer, "world", n) == 0); + cli_rd_check = 2; + } + } while (n > 0); + + ASSERT(got_eagain()); + } + } + if (fd == server_fd) { + do + n = recv(server_fd, &buffer, 3, 0); + while (n == -1 && errno == EINTR); + ASSERT(n >= 0 || errno != EINVAL); + ASSERT(3 == n); + ASSERT(strncmp(buffer, "foo", n) == 0); + srv_rd_check = 1; + uv_poll_stop(&poll_req[1]); + } + } + if (events & UV_WRITABLE) { + do { + n = send(client_fd, "foo", 3, 0); + } while (n < 0 && errno == EINTR); + ASSERT(3 == n); + } +} + +static void connection_cb(uv_stream_t* handle, int status) { + int r; + + ASSERT(0 == status); + ASSERT(0 == uv_accept(handle, (uv_stream_t*) &peer_handle)); + ASSERT(0 == uv_fileno((uv_handle_t*) &peer_handle, &server_fd)); + ASSERT(0 == uv_poll_init_socket(uv_default_loop(), &poll_req[0], client_fd)); + ASSERT(0 == uv_poll_init_socket(uv_default_loop(), &poll_req[1], server_fd)); + ASSERT(0 == uv_poll_start(&poll_req[0], + UV_PRIORITIZED | UV_READABLE | UV_WRITABLE, + poll_cb)); + ASSERT(0 == uv_poll_start(&poll_req[1], + UV_READABLE, + poll_cb)); + do { + r = send(server_fd, "hello", 5, MSG_OOB); + } while (r < 0 && errno == EINTR); + ASSERT(5 == r); + + do { + r = send(server_fd, "world", 5, 0); + } while (r < 0 && errno == EINTR); + ASSERT(5 == r); + + ASSERT(0 == uv_idle_start(&idle, idle_cb)); +} + + +TEST_IMPL(poll_oob) { + struct sockaddr_in addr; + int r = 0; + uv_loop_t* loop; + + ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &addr)); + loop = uv_default_loop(); + + ASSERT(0 == uv_tcp_init(loop, &server_handle)); + ASSERT(0 == uv_tcp_init(loop, &client_handle)); + ASSERT(0 == uv_tcp_init(loop, &peer_handle)); + ASSERT(0 == uv_idle_init(loop, &idle)); + ASSERT(0 == uv_tcp_bind(&server_handle, (const struct sockaddr*) &addr, 0)); + ASSERT(0 == uv_listen((uv_stream_t*) &server_handle, 1, connection_cb)); + + /* Ensure two separate packets */ + ASSERT(0 == uv_tcp_nodelay(&client_handle, 1)); + + client_fd = socket(PF_INET, SOCK_STREAM, 0); + ASSERT(client_fd >= 0); + do { + errno = 0; + r = connect(client_fd, (const struct sockaddr*)&addr, sizeof(addr)); + } while (r == -1 && errno == EINTR); + ASSERT(r == 0); + + ASSERT(0 == uv_run(loop, UV_RUN_DEFAULT)); + + ASSERT(ticks == kMaxTicks); + + /* Did client receive the POLLPRI message */ + ASSERT(cli_pr_check == 1); + /* Did client receive the POLLIN message */ + ASSERT(cli_rd_check == 2); + /* Could we write with POLLOUT and did the server receive our POLLOUT message + * through POLLIN. + */ + ASSERT(srv_rd_check == 1); + + MAKE_VALGRIND_HAPPY(); + return 0; +} +#endif diff --git a/uv.gyp b/uv.gyp index 6f61d725..825edd9f 100644 --- a/uv.gyp +++ b/uv.gyp @@ -405,6 +405,7 @@ 'test/test-poll-close.c', 'test/test-poll-close-doesnt-corrupt-stack.c', 'test/test-poll-closesocket.c', + 'test/test-poll-oob.c', 'test/test-process-title.c', 'test/test-queue-foreach-delete.c', 'test/test-ref.c',