diff --git a/src/coio.cc b/src/coio.cc index 64298c745d4a0eccf53a393b36ff45b515038cc6..c803a79eaeeb041d057698ed770614135ab48874 100644 --- a/src/coio.cc +++ b/src/coio.cc @@ -705,3 +705,42 @@ coio_waitpid(pid_t pid) return status; } +/* 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_call(wdata->fiber); +} + +int +coio_wait(int fd, int events, double timeout) +{ + struct ev_io io; + coio_init(&io, fd); + 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; +} diff --git a/src/coio.h b/src/coio.h index 3fdc8579d72d5ccae0c5e83bfa4c1c12a442247d..f43811c9e143256efe9a3272524b5b91c71e8351 100644 --- a/src/coio.h +++ b/src/coio.h @@ -32,6 +32,7 @@ */ #include "evio.h" #include "fiber.h" +#include "trivia/util.h" /** * Co-operative I/O @@ -183,5 +184,28 @@ coio_stat_stat_timeout(ev_stat *stat, ev_tstamp delay); int coio_waitpid(pid_t pid); +/** \cond public */ + +enum { + /** READ event */ + COIO_READ = 0x1, + /** WRITE event */ + COIO_WRITE = 0x2, +}; + +/** + * Wait until READ or WRITE event on socket (\a fd). Yields. + * \param fd - non-blocking socket file description + * \param events - requested events to wait. + * Combination of TNT_IO_READ | TNT_IO_WRITE bit flags. + * \param timeoout - timeout in seconds. + * \retval 0 - timeout + * \retval >0 - returned events. Combination of TNT_IO_READ | TNT_IO_WRITE + * bit flags. + */ +API_EXPORT int +coio_wait(int fd, int event, double timeout); + +/** \endcond public */ #endif /* TARANTOOL_COIO_H_INCLUDED */ diff --git a/src/lua/bsdsocket.cc b/src/lua/bsdsocket.cc index db5a6866e390ab1a1838952e57d05ad352b02010..79a2a14fbc1ce6711adab6dbad9e6a02ee089677 100644 --- a/src/lua/bsdsocket.cc +++ b/src/lua/bsdsocket.cc @@ -49,6 +49,7 @@ extern "C" { #include <lualib.h> } +#include <coio.h> #include <coeio.h> #include <fiber.h> #include <scoped_guard.h> @@ -379,21 +380,6 @@ bsdsocket_nonblock(int fh, int mode) return mode ? 1 : 0; } -struct bsdsocket_io_wdata { - struct fiber *fiber; - int io; -}; - -static void -bsdsocket_io(struct ev_loop *loop, ev_io *watcher, int revents) -{ - (void) loop; - struct bsdsocket_io_wdata *wdata = - (struct bsdsocket_io_wdata *)watcher->data; - wdata->io = revents; - fiber_wakeup(wdata->fiber); -} - static int lbox_bsdsocket_iowait(struct lua_State *L) { @@ -401,35 +387,7 @@ lbox_bsdsocket_iowait(struct lua_State *L) int events = lua_tointeger(L, 2); ev_tstamp timeout = lua_tonumber(L, 3); - switch (events) { - case 0: - events = EV_READ; - break; - case 1: - events = EV_WRITE; - break; - case 2: - events = EV_READ | EV_WRITE; - break; - default: - assert(false); - } - - struct ev_io io; - ev_io_init(&io, bsdsocket_io, fh, events); - struct bsdsocket_io_wdata wdata = { fiber(), 0 }; - io.data = &wdata; - ev_set_priority(&io, EV_MAXPRI); - ev_io_start(loop(), &io); - - fiber_yield_timeout(timeout); - ev_io_stop(loop(), &io); - - int ret = 0; - if (wdata.io & EV_READ) - ret |= 1; - if (wdata.io & EV_WRITE) - ret |= 2; + int ret = coio_wait(fh, events, timeout); lua_pushinteger(L, ret); return 1; diff --git a/src/lua/bsdsocket.lua b/src/lua/bsdsocket.lua index cb0b11e51a799bf7533193d9d8bf5dffe121eb51..edffec7014da911b0d8b96ea0b1efcf828a31679 100644 --- a/src/lua/bsdsocket.lua +++ b/src/lua/bsdsocket.lua @@ -270,20 +270,20 @@ local function wait_safely(self, what, timeout) self.waiters[fid] = true local res = internal.iowait(fd, what, timeout) self.waiters[fid] = nil - fiber.testcancel() if res == 0 then self._errno = boxerrno.ETIMEDOUT return 0 end + fiber.testcancel() return res end socket_methods.readable = function(self, timeout) - return wait_safely(self, 0, timeout) ~= 0 + return wait_safely(self, 1, timeout) ~= 0 end socket_methods.wait = function(self, timeout) - local wres = wait_safely(self, 2, timeout) + local wres = wait_safely(self, 3, timeout) local res = '' if bit.band(wres, 1) ~= 0 then res = res .. 'R' @@ -295,7 +295,7 @@ socket_methods.wait = function(self, timeout) end socket_methods.writable = function(self, timeout) - return wait_safely(self, 1, timeout) ~= 0 + return wait_safely(self, 2, timeout) ~= 0 end socket_methods.listen = function(self, backlog) diff --git a/src/trivia/CMakeLists.txt b/src/trivia/CMakeLists.txt index f92d73e8a16c6f2115501567fc0ab162e4fb30d3..dce4d39c342f75b905f225862dda2066c9fc81b9 100644 --- a/src/trivia/CMakeLists.txt +++ b/src/trivia/CMakeLists.txt @@ -1,6 +1,7 @@ set(api_headers ${CMAKE_CURRENT_BINARY_DIR}/config.h ${CMAKE_SOURCE_DIR}/src/say.h + ${CMAKE_SOURCE_DIR}/src/coio.h ${CMAKE_SOURCE_DIR}/src/coeio.h ${CMAKE_SOURCE_DIR}/src/lua/utils.h ${CMAKE_SOURCE_DIR}/src/box/txn.h