From 4a76b67fb4c9f54cfebaa6d067f82327c5d1ba93 Mon Sep 17 00:00:00 2001
From: Konstantin Osipov <kostja@tarantool.org>
Date: Mon, 8 Oct 2012 17:24:07 +0400
Subject: [PATCH] Extract cooperative IO from fiber.m.
Make all subsystems (administrative console,
replication, memcached, binary protocol) use exception-based
socket API (sio.m).
Remove all fiber members related to socket I/O.
Remove fiber->peer and fiber->cookie.
The idea of the cookie is to preserve in the WAL
the originator of the request. This information is, however,
lost when saving a snapshot. To my knowledge there is no use
of this functionality, apart from pretty-printing
the WAL.
Peer is like cookie, but is textual and is used
for logging. However, now, as long as network errors
occur outside fiber context, most of the usefulness
is lost. The only remaining case is when someone is
trying to execute a write request on a read-only port.
For all other cases, sio.m already supplies SocketError
with peer name, and this is where it's most useful.
---
include/coio.h | 97 ++++++
include/coio_buf.h | 63 ++++
include/evio.h | 6 +-
include/fiber.h | 52 +--
mod/box/box.m | 36 ++-
mod/box/memcached-grammar.m | 600 +++++++++++++++++------------------
mod/box/memcached-grammar.rl | 14 +-
mod/box/memcached.h | 4 +-
mod/box/memcached.m | 64 ++--
mod/box/txn.m | 4 +-
src/CMakeLists.txt | 2 +
src/admin.m | 136 ++++----
src/admin.rl | 30 +-
src/coio.m | 302 ++++++++++++++++++
src/coio_buf.m | 31 ++
src/evio.m | 5 +-
src/exception.m | 2 +-
src/fiber.m | 489 +---------------------------
src/iproto.m | 49 +--
src/replica.m | 138 ++++----
src/replication.m | 7 +-
src/say.m | 11 +-
src/sio.m | 2 +-
src/tarantool.m | 3 +-
src/tarantool_lua.m | 4 +-
test/box/lua.result | 8 +-
test/box/lua.test | 4 +-
27 files changed, 1068 insertions(+), 1095 deletions(-)
create mode 100644 include/coio.h
create mode 100644 include/coio_buf.h
create mode 100644 src/coio.m
create mode 100644 src/coio_buf.m
diff --git a/include/coio.h b/include/coio.h
new file mode 100644
index 0000000000..5907e8c070
--- /dev/null
+++ b/include/coio.h
@@ -0,0 +1,97 @@
+#ifndef TARANTOOL_COIO_H_INCLUDED
+#define TARANTOOL_COIO_H_INCLUDED
+/*
+ * 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 "evio.h"
+/*
+ * Co-operative I/O
+ * Yield the current fiber until IO is ready.
+ */
+struct coio
+{
+ struct ev_io ev;
+};
+
+struct coio_service
+{
+ struct evio_service evio_service;
+ /* Fiber function. */
+ void (*handler)(va_list ap);
+ /** Passed to the created fiber. */
+ void *handler_param;
+};
+
+void
+coio_clear(struct coio *coio);
+
+static inline bool
+coio_is_connected(struct coio *coio)
+{
+ return coio->ev.fd >= 0;
+}
+
+void
+coio_connect(struct coio *coio, struct sockaddr_in *addr);
+
+void
+coio_init(struct coio *coio, int fd);
+
+void
+coio_close(struct coio *coio);
+
+ssize_t
+coio_read_ahead(struct coio *coio, void *buf, size_t sz, size_t bufsiz);
+
+ssize_t
+coio_readn_ahead(struct coio *coio, void *buf, size_t sz, size_t bufsiz);
+
+static inline ssize_t
+coio_read(struct coio *coio, void *buf, size_t sz)
+{
+ return coio_read_ahead(coio, buf, sz, sz);
+}
+
+static inline ssize_t
+coio_readn(struct coio *coio, void *buf, size_t sz)
+{
+ return coio_readn_ahead(coio, buf, sz, sz);
+}
+
+void
+coio_write(struct coio *coio, const void *buf, size_t sz);
+
+ssize_t
+coio_writev(struct coio *coio, struct iovec *iov, int iovcnt);
+
+void
+coio_service_init(struct coio_service *service, const char *name,
+ const char *host, int port,
+ void (*handler)(va_list ap), void *handler_param);
+
+#endif /* TARANTOOL_COIO_H_INCLUDED */
diff --git a/include/coio_buf.h b/include/coio_buf.h
new file mode 100644
index 0000000000..8c8e8ffb5d
--- /dev/null
+++ b/include/coio_buf.h
@@ -0,0 +1,63 @@
+#ifndef TARANTOOL_COIO_BUF_H_INCLUDED
+#define TARANTOOL_COIO_BUF_H_INCLUDED
+/*
+ * 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 "tbuf.h"
+/** Buffered cooperative IO */
+
+extern int coio_readahead;
+
+static inline void
+coio_binit(int readahead)
+{
+ coio_readahead = readahead;
+}
+
+static inline ssize_t
+coio_bread(struct coio *coio, struct tbuf *buf, size_t sz)
+{
+ tbuf_ensure(buf, MAX(sz, coio_readahead));
+ ssize_t n = coio_read_ahead(coio, tbuf_end(buf),
+ sz, tbuf_unused(buf));
+ buf->size += n;
+ return n;
+}
+
+static inline ssize_t
+coio_breadn(struct coio *coio, struct tbuf *buf, size_t sz)
+{
+ tbuf_ensure(buf, MAX(sz, coio_readahead));
+ ssize_t n = coio_readn_ahead(coio, tbuf_end(buf),
+ sz, tbuf_unused(buf));
+ buf->size += n;
+ return n;
+}
+
+#endif /* TARANTOOL_COIO_BUF_H_INCLUDED */
diff --git a/include/evio.h b/include/evio.h
index 7f12c7f2ec..333ee2821c 100644
--- a/include/evio.h
+++ b/include/evio.h
@@ -78,7 +78,8 @@ struct evio_service
* when it happens, the exception is logged, and the
* accepted socket is closed.
*/
- void (*on_accept)(void *, int, struct sockaddr_in *);
+ void (*on_accept)(struct evio_service *, int,
+ struct sockaddr_in *);
void *on_accept_param;
/** libev timer object for the bind retry delay. */
@@ -91,7 +92,8 @@ struct evio_service
void
evio_service_init(struct evio_service *service, const char *name,
const char *host, int port,
- void (*on_accept)(void *, int, struct sockaddr_in *),
+ void (*on_accept)(struct evio_service *,
+ int, struct sockaddr_in *),
void *on_accept_param);
/** Set an optional callback to be invoked upon a successful bind. */
diff --git a/include/fiber.h b/include/fiber.h
index a048249266..2c196f2253 100644
--- a/include/fiber.h
+++ b/include/fiber.h
@@ -63,7 +63,6 @@
@end
struct fiber {
- ev_io io;
ev_async async;
#ifdef ENABLE_BACKTRACE
void *last_stack_frame;
@@ -73,7 +72,6 @@ struct fiber {
/* A garbage-collected memory pool. */
struct palloc_pool *gc_pool;
uint32_t fid;
- int fd;
ev_timer timer;
ev_child cw;
@@ -89,13 +87,7 @@ struct fiber {
char name[FIBER_NAME_MAXLEN];
void (*f) (va_list);
va_list f_data;
- u64 cookie;
- bool has_peer;
- /* ASCIIZ name of the peer, if there is one. */
- char peer_name[32];
-
u32 flags;
-
struct fiber *waiter;
};
@@ -111,19 +103,10 @@ extern __thread struct fiber *fiber;
void fiber_init(void);
void fiber_free(void);
-struct fiber *fiber_create(const char *name, int fd, void (*f) (va_list));
+struct fiber *fiber_create(const char *name, void (*f) (va_list));
void fiber_set_name(struct fiber *fiber, const char *name);
void wait_for_child(pid_t pid);
-void
-fiber_io_start(int fd, int events);
-
-void
-fiber_io_yield();
-
-void
-fiber_io_stop(int fd, int events);
-
void
fiber_yield(void);
@@ -132,8 +115,6 @@ fiber_yield_to(struct fiber *f);
void fiber_destroy_all();
-ssize_t fiber_bread(struct tbuf *, size_t v);
-
inline static void iov_add_unsafe(const void *buf, size_t len)
{
struct iovec *v;
@@ -164,15 +145,12 @@ inline static void iov_dup(const void *buf, size_t len)
iov_add(copy, len);
}
+struct coio;
/* Reset the fiber's iov vector. */
-ssize_t iov_flush(void);
+ssize_t iov_flush(struct coio *coio);
/* Write everything in the fiber's iov vector to fiber socket. */
void iov_reset();
-const char *fiber_peer_name(struct fiber *fiber);
-ssize_t fiber_read(void *buf, size_t count);
-ssize_t fiber_write(const void *buf, size_t count);
-int fiber_close(void);
void fiber_cleanup(void);
void fiber_gc(void);
void fiber_call(struct fiber *callee, ...);
@@ -193,29 +171,9 @@ void fiber_testcancel(void);
* fiber.
*/
void fiber_setcancelstate(bool enable);
-int fiber_connect(struct sockaddr_in *addr);
void fiber_sleep(ev_tstamp s);
void fiber_info(struct tbuf *out);
-int set_nonblock(int sock);
-
-typedef void (*fiber_server_callback)(va_list);
-
-struct fiber *fiber_server(const char *name, int port,
- fiber_server_callback callback, void *,
- void (*on_bind) (void *));
-
-/**
- * Create server socket and bind his on port. cfd.bind_ipaddr param using as IP address.
- *
- * @param type the fiber server type (TCP or UDP)
- * @param port the bind ip port.
- * @param retry the retry flag, if flag up the function will be try again to bind
- * socket after delay.
- * @param delay the bind socket retry delay in sec.
- *
- * @return on success, zero is returned. on error, -1 is returned.
- */
-int
-fiber_serv_socket(struct fiber *fiber, unsigned short port, bool retry, ev_tstamp delay);
+void
+fiber_schedule(ev_watcher *watcher, int event __attribute__((unused)));
#endif /* TARANTOOL_FIBER_H_INCLUDED */
diff --git a/mod/box/box.m b/mod/box/box.m
index b433a9c5ca..ae2d14f5db 100644
--- a/mod/box/box.m
+++ b/mod/box/box.m
@@ -47,6 +47,7 @@
#include "port.h"
#include "request.h"
#include "txn.h"
+#include "coio.h"
static void box_process_replica(struct txn *txn, struct port *port,
u32 op, struct tbuf *request_data);
@@ -518,22 +519,33 @@ mod_init(void)
}
/* run primary server */
- if (cfg.primary_port != 0)
- fiber_server("primary", cfg.primary_port,
- iproto_interact,
- iproto_primary_port_handler,
- box_leave_local_standby_mode);
+ if (cfg.primary_port != 0) {
+ static struct coio_service primary;
+ coio_service_init(&primary, "primary",
+ cfg.bind_ipaddr, cfg.primary_port,
+ iproto_interact, iproto_primary_port_handler);
+ evio_service_on_bind(&primary.evio_service,
+ box_leave_local_standby_mode, NULL);
+ evio_service_start(&primary.evio_service);
+ }
/* run secondary server */
- if (cfg.secondary_port != 0)
- fiber_server("secondary", cfg.secondary_port,
- iproto_interact,
- iproto_secondary_port_handler, NULL);
+ if (cfg.secondary_port != 0) {
+ static struct coio_service secondary;
+ coio_service_init(&secondary, "secondary",
+ cfg.bind_ipaddr, cfg.primary_port,
+ iproto_interact, iproto_secondary_port_handler);
+ evio_service_start(&secondary.evio_service);
+ }
/* run memcached server */
- if (cfg.memcached_port != 0)
- fiber_server("memcached", cfg.memcached_port,
- memcached_handler, NULL, NULL);
+ if (cfg.memcached_port != 0) {
+ static struct coio_service memcached;
+ coio_service_init(&memcached, "memcached",
+ cfg.bind_ipaddr, cfg.memcached_port,
+ memcached_handler, NULL);
+ evio_service_start(&memcached.evio_service);
+ }
}
int
diff --git a/mod/box/memcached-grammar.m b/mod/box/memcached-grammar.m
index d840bd75b9..a4a76e46a4 100644
--- a/mod/box/memcached-grammar.m
+++ b/mod/box/memcached-grammar.m
@@ -42,7 +42,7 @@ static const int memcached_en_main = 1;
static int __attribute__((noinline))
-memcached_dispatch()
+memcached_dispatch(struct coio *coio)
{
int cs;
u8 *p, *pe;
@@ -56,7 +56,6 @@ memcached_dispatch()
bool noreply = false;
u8 *data = NULL;
bool done = false;
- int r;
size_t saved_iov_cnt = fiber->iov_cnt;
uintptr_t flush_delay = 0;
size_t keys_count = 0;
@@ -67,12 +66,12 @@ memcached_dispatch()
say_debug("memcached_dispatch '%.*s'", MIN((int)(pe - p), 40) , p);
-#line 71 "mod/box/memcached-grammar.m"
+#line 70 "mod/box/memcached-grammar.m"
{
cs = memcached_start;
}
-#line 76 "mod/box/memcached-grammar.m"
+#line 75 "mod/box/memcached-grammar.m"
{
if ( p == pe )
goto _test_eof;
@@ -144,7 +143,7 @@ case 5:
goto st0;
goto tr15;
tr15:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -161,7 +160,7 @@ st6:
if ( ++p == pe )
goto _test_eof6;
case 6:
-#line 165 "mod/box/memcached-grammar.m"
+#line 164 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st7;
goto st0;
@@ -175,49 +174,49 @@ case 7:
goto tr17;
goto st0;
tr17:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st8;
st8:
if ( ++p == pe )
goto _test_eof8;
case 8:
-#line 186 "mod/box/memcached-grammar.m"
+#line 185 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr18;
if ( 48 <= (*p) && (*p) <= 57 )
goto st8;
goto st0;
tr18:
-#line 245 "mod/box/memcached-grammar.rl"
+#line 244 "mod/box/memcached-grammar.rl"
{flags = natoq(fstart, p);}
goto st9;
st9:
if ( ++p == pe )
goto _test_eof9;
case 9:
-#line 200 "mod/box/memcached-grammar.m"
+#line 199 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st9;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr21;
goto st0;
tr21:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st10;
st10:
if ( ++p == pe )
goto _test_eof10;
case 10:
-#line 214 "mod/box/memcached-grammar.m"
+#line 213 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr22;
if ( 48 <= (*p) && (*p) <= 57 )
goto st10;
goto st0;
tr22:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -228,21 +227,21 @@ st11:
if ( ++p == pe )
goto _test_eof11;
case 11:
-#line 232 "mod/box/memcached-grammar.m"
+#line 231 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st11;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr25;
goto st0;
tr25:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st12;
st12:
if ( ++p == pe )
goto _test_eof12;
case 12:
-#line 246 "mod/box/memcached-grammar.m"
+#line 245 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr26;
case 13: goto tr27;
@@ -252,18 +251,16 @@ case 12:
goto st12;
goto st0;
tr26:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -277,13 +274,13 @@ tr26:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 66 "mod/box/memcached-grammar.rl"
+#line 65 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -294,16 +291,14 @@ tr26:
}
goto st197;
tr30:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -317,13 +312,13 @@ tr30:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 66 "mod/box/memcached-grammar.rl"
+#line 65 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -334,18 +329,16 @@ tr30:
}
goto st197;
tr39:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -359,13 +352,13 @@ tr39:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 66 "mod/box/memcached-grammar.rl"
+#line 65 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -376,18 +369,16 @@ tr39:
}
goto st197;
tr58:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -401,13 +392,13 @@ tr58:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 95 "mod/box/memcached-grammar.rl"
+#line 94 "mod/box/memcached-grammar.rl"
{
struct tbuf *b;
void *value;
@@ -436,16 +427,14 @@ tr58:
}
goto st197;
tr62:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -459,13 +448,13 @@ tr62:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 95 "mod/box/memcached-grammar.rl"
+#line 94 "mod/box/memcached-grammar.rl"
{
struct tbuf *b;
void *value;
@@ -494,18 +483,16 @@ tr62:
}
goto st197;
tr71:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -519,13 +506,13 @@ tr71:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 95 "mod/box/memcached-grammar.rl"
+#line 94 "mod/box/memcached-grammar.rl"
{
struct tbuf *b;
void *value;
@@ -554,18 +541,16 @@ tr71:
}
goto st197;
tr91:
-#line 247 "mod/box/memcached-grammar.rl"
+#line 246 "mod/box/memcached-grammar.rl"
{cas = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -579,13 +564,13 @@ tr91:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 84 "mod/box/memcached-grammar.rl"
+#line 83 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -598,16 +583,14 @@ tr91:
}
goto st197;
tr95:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -621,13 +604,13 @@ tr95:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 84 "mod/box/memcached-grammar.rl"
+#line 83 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -640,18 +623,16 @@ tr95:
}
goto st197;
tr105:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -665,13 +646,13 @@ tr105:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 84 "mod/box/memcached-grammar.rl"
+#line 83 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -684,17 +665,17 @@ tr105:
}
goto st197;
tr118:
-#line 248 "mod/box/memcached-grammar.rl"
+#line 247 "mod/box/memcached-grammar.rl"
{incr = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 122 "mod/box/memcached-grammar.rl"
+#line 121 "mod/box/memcached-grammar.rl"
{
struct meta *m;
struct tbuf *b;
@@ -751,15 +732,15 @@ tr118:
}
goto st197;
tr122:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 122 "mod/box/memcached-grammar.rl"
+#line 121 "mod/box/memcached-grammar.rl"
{
struct meta *m;
struct tbuf *b;
@@ -816,17 +797,17 @@ tr122:
}
goto st197;
tr132:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 122 "mod/box/memcached-grammar.rl"
+#line 121 "mod/box/memcached-grammar.rl"
{
struct meta *m;
struct tbuf *b;
@@ -883,15 +864,15 @@ tr132:
}
goto st197;
tr141:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 177 "mod/box/memcached-grammar.rl"
+#line 176 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -911,21 +892,21 @@ tr141:
}
goto st197;
tr146:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
exptime = exptime + ev_now();
}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 177 "mod/box/memcached-grammar.rl"
+#line 176 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -945,17 +926,17 @@ tr146:
}
goto st197;
tr157:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 177 "mod/box/memcached-grammar.rl"
+#line 176 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -975,70 +956,70 @@ tr157:
}
goto st197;
tr169:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 206 "mod/box/memcached-grammar.rl"
+#line 205 "mod/box/memcached-grammar.rl"
{
- struct fiber *f = fiber_create("flush_all", -1, flush_all);
+ struct fiber *f = fiber_create("flush_all", flush_all);
if (f)
fiber_call(f, flush_delay);
iov_add("OK\r\n", 4);
}
goto st197;
tr174:
-#line 249 "mod/box/memcached-grammar.rl"
+#line 248 "mod/box/memcached-grammar.rl"
{flush_delay = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 206 "mod/box/memcached-grammar.rl"
+#line 205 "mod/box/memcached-grammar.rl"
{
- struct fiber *f = fiber_create("flush_all", -1, flush_all);
+ struct fiber *f = fiber_create("flush_all", flush_all);
if (f)
fiber_call(f, flush_delay);
iov_add("OK\r\n", 4);
}
goto st197;
tr185:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 206 "mod/box/memcached-grammar.rl"
+#line 205 "mod/box/memcached-grammar.rl"
{
- struct fiber *f = fiber_create("flush_all", -1, flush_all);
+ struct fiber *f = fiber_create("flush_all", flush_all);
if (f)
fiber_call(f, flush_delay);
iov_add("OK\r\n", 4);
}
goto st197;
tr195:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 195 "mod/box/memcached-grammar.rl"
+#line 194 "mod/box/memcached-grammar.rl"
{
@try {
memcached_get(keys_count, keys, show_cas);
@@ -1051,32 +1032,30 @@ tr195:
}
goto st197;
tr213:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 217 "mod/box/memcached-grammar.rl"
+#line 216 "mod/box/memcached-grammar.rl"
{
- return 0;
+ return -1;
}
goto st197;
tr233:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1090,13 +1069,13 @@ tr233:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 75 "mod/box/memcached-grammar.rl"
+#line 74 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -1107,16 +1086,14 @@ tr233:
}
goto st197;
tr237:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1130,13 +1107,13 @@ tr237:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 75 "mod/box/memcached-grammar.rl"
+#line 74 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -1147,18 +1124,16 @@ tr237:
}
goto st197;
tr246:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1172,13 +1147,13 @@ tr246:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 75 "mod/box/memcached-grammar.rl"
+#line 74 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
struct tuple *tuple = find(key);
@@ -1189,18 +1164,16 @@ tr246:
}
goto st197;
tr263:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1214,29 +1187,27 @@ tr263:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 61 "mod/box/memcached-grammar.rl"
+#line 60 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
STORE;
}
goto st197;
tr267:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1250,31 +1221,29 @@ tr267:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 61 "mod/box/memcached-grammar.rl"
+#line 60 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
STORE;
}
goto st197;
tr276:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 251 "mod/box/memcached-grammar.rl"
+#line 250 "mod/box/memcached-grammar.rl"
{
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -1288,28 +1257,28 @@ tr276:
goto exit;
}
}
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 61 "mod/box/memcached-grammar.rl"
+#line 60 "mod/box/memcached-grammar.rl"
{
key = read_field(keys);
STORE;
}
goto st197;
tr281:
-#line 278 "mod/box/memcached-grammar.rl"
+#line 275 "mod/box/memcached-grammar.rl"
{ p++; }
-#line 272 "mod/box/memcached-grammar.rl"
+#line 269 "mod/box/memcached-grammar.rl"
{
done = true;
stats.bytes_read += p - (u8 *)fiber->rbuf.data;
tbuf_peek(&fiber->rbuf, p - (u8 *)fiber->rbuf.data);
}
-#line 213 "mod/box/memcached-grammar.rl"
+#line 212 "mod/box/memcached-grammar.rl"
{
print_stats();
}
@@ -1318,33 +1287,33 @@ st197:
if ( ++p == pe )
goto _test_eof197;
case 197:
-#line 1322 "mod/box/memcached-grammar.m"
+#line 1291 "mod/box/memcached-grammar.m"
goto st0;
tr27:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st13;
tr40:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st13;
st13:
if ( ++p == pe )
goto _test_eof13;
case 13:
-#line 1336 "mod/box/memcached-grammar.m"
+#line 1305 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr30;
goto st0;
tr28:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st14;
st14:
if ( ++p == pe )
goto _test_eof14;
case 14:
-#line 1348 "mod/box/memcached-grammar.m"
+#line 1317 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 32: goto st14;
case 78: goto st15;
@@ -1458,18 +1427,18 @@ case 26:
goto tr45;
goto st0;
tr45:
-#line 286 "mod/box/memcached-grammar.rl"
+#line 283 "mod/box/memcached-grammar.rl"
{append = true; }
goto st27;
tr209:
-#line 287 "mod/box/memcached-grammar.rl"
+#line 284 "mod/box/memcached-grammar.rl"
{append = false;}
goto st27;
st27:
if ( ++p == pe )
goto _test_eof27;
case 27:
-#line 1473 "mod/box/memcached-grammar.m"
+#line 1442 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 13: goto st0;
case 32: goto st27;
@@ -1478,7 +1447,7 @@ case 27:
goto st0;
goto tr46;
tr46:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -1495,7 +1464,7 @@ st28:
if ( ++p == pe )
goto _test_eof28;
case 28:
-#line 1499 "mod/box/memcached-grammar.m"
+#line 1468 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st29;
goto st0;
@@ -1509,49 +1478,49 @@ case 29:
goto tr49;
goto st0;
tr49:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st30;
st30:
if ( ++p == pe )
goto _test_eof30;
case 30:
-#line 1520 "mod/box/memcached-grammar.m"
+#line 1489 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr50;
if ( 48 <= (*p) && (*p) <= 57 )
goto st30;
goto st0;
tr50:
-#line 245 "mod/box/memcached-grammar.rl"
+#line 244 "mod/box/memcached-grammar.rl"
{flags = natoq(fstart, p);}
goto st31;
st31:
if ( ++p == pe )
goto _test_eof31;
case 31:
-#line 1534 "mod/box/memcached-grammar.m"
+#line 1503 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st31;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr53;
goto st0;
tr53:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st32;
st32:
if ( ++p == pe )
goto _test_eof32;
case 32:
-#line 1548 "mod/box/memcached-grammar.m"
+#line 1517 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr54;
if ( 48 <= (*p) && (*p) <= 57 )
goto st32;
goto st0;
tr54:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -1562,21 +1531,21 @@ st33:
if ( ++p == pe )
goto _test_eof33;
case 33:
-#line 1566 "mod/box/memcached-grammar.m"
+#line 1535 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st33;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr57;
goto st0;
tr57:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st34;
st34:
if ( ++p == pe )
goto _test_eof34;
case 34:
-#line 1580 "mod/box/memcached-grammar.m"
+#line 1549 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr58;
case 13: goto tr59;
@@ -1586,30 +1555,30 @@ case 34:
goto st34;
goto st0;
tr59:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st35;
tr72:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st35;
st35:
if ( ++p == pe )
goto _test_eof35;
case 35:
-#line 1601 "mod/box/memcached-grammar.m"
+#line 1570 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr62;
goto st0;
tr60:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st36;
st36:
if ( ++p == pe )
goto _test_eof36;
case 36:
-#line 1613 "mod/box/memcached-grammar.m"
+#line 1582 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 32: goto st36;
case 78: goto st37;
@@ -1716,7 +1685,7 @@ case 47:
goto st0;
goto tr76;
tr76:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -1733,7 +1702,7 @@ st48:
if ( ++p == pe )
goto _test_eof48;
case 48:
-#line 1737 "mod/box/memcached-grammar.m"
+#line 1706 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st49;
goto st0;
@@ -1747,49 +1716,49 @@ case 49:
goto tr78;
goto st0;
tr78:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st50;
st50:
if ( ++p == pe )
goto _test_eof50;
case 50:
-#line 1758 "mod/box/memcached-grammar.m"
+#line 1727 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr79;
if ( 48 <= (*p) && (*p) <= 57 )
goto st50;
goto st0;
tr79:
-#line 245 "mod/box/memcached-grammar.rl"
+#line 244 "mod/box/memcached-grammar.rl"
{flags = natoq(fstart, p);}
goto st51;
st51:
if ( ++p == pe )
goto _test_eof51;
case 51:
-#line 1772 "mod/box/memcached-grammar.m"
+#line 1741 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st51;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr82;
goto st0;
tr82:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st52;
st52:
if ( ++p == pe )
goto _test_eof52;
case 52:
-#line 1786 "mod/box/memcached-grammar.m"
+#line 1755 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr83;
if ( 48 <= (*p) && (*p) <= 57 )
goto st52;
goto st0;
tr83:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -1800,49 +1769,49 @@ st53:
if ( ++p == pe )
goto _test_eof53;
case 53:
-#line 1804 "mod/box/memcached-grammar.m"
+#line 1773 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st53;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr86;
goto st0;
tr86:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st54;
st54:
if ( ++p == pe )
goto _test_eof54;
case 54:
-#line 1818 "mod/box/memcached-grammar.m"
+#line 1787 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr87;
if ( 48 <= (*p) && (*p) <= 57 )
goto st54;
goto st0;
tr87:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st55;
st55:
if ( ++p == pe )
goto _test_eof55;
case 55:
-#line 1832 "mod/box/memcached-grammar.m"
+#line 1801 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st55;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr90;
goto st0;
tr90:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st56;
st56:
if ( ++p == pe )
goto _test_eof56;
case 56:
-#line 1846 "mod/box/memcached-grammar.m"
+#line 1815 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr91;
case 13: goto tr92;
@@ -1852,30 +1821,30 @@ case 56:
goto st56;
goto st0;
tr106:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st57;
tr92:
-#line 247 "mod/box/memcached-grammar.rl"
+#line 246 "mod/box/memcached-grammar.rl"
{cas = natoq(fstart, p);}
goto st57;
st57:
if ( ++p == pe )
goto _test_eof57;
case 57:
-#line 1867 "mod/box/memcached-grammar.m"
+#line 1836 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr95;
goto st0;
tr93:
-#line 247 "mod/box/memcached-grammar.rl"
+#line 246 "mod/box/memcached-grammar.rl"
{cas = natoq(fstart, p);}
goto st58;
st58:
if ( ++p == pe )
goto _test_eof58;
case 58:
-#line 1879 "mod/box/memcached-grammar.m"
+#line 1848 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr95;
case 13: goto st57;
@@ -1949,14 +1918,14 @@ case 65:
}
goto st0;
tr107:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st66;
st66:
if ( ++p == pe )
goto _test_eof66;
case 66:
-#line 1960 "mod/box/memcached-grammar.m"
+#line 1929 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr95;
case 13: goto st57;
@@ -2000,18 +1969,18 @@ case 70:
goto tr113;
goto st0;
tr113:
-#line 295 "mod/box/memcached-grammar.rl"
+#line 292 "mod/box/memcached-grammar.rl"
{incr_sign = -1;}
goto st71;
tr202:
-#line 294 "mod/box/memcached-grammar.rl"
+#line 291 "mod/box/memcached-grammar.rl"
{incr_sign = 1; }
goto st71;
st71:
if ( ++p == pe )
goto _test_eof71;
case 71:
-#line 2015 "mod/box/memcached-grammar.m"
+#line 1984 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 13: goto st0;
case 32: goto st71;
@@ -2020,7 +1989,7 @@ case 71:
goto st0;
goto tr114;
tr114:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -2037,7 +2006,7 @@ st72:
if ( ++p == pe )
goto _test_eof72;
case 72:
-#line 2041 "mod/box/memcached-grammar.m"
+#line 2010 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st73;
goto st0;
@@ -2051,14 +2020,14 @@ case 73:
goto tr117;
goto st0;
tr117:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st74;
st74:
if ( ++p == pe )
goto _test_eof74;
case 74:
-#line 2062 "mod/box/memcached-grammar.m"
+#line 2031 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr118;
case 13: goto tr119;
@@ -2068,30 +2037,30 @@ case 74:
goto st74;
goto st0;
tr133:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st75;
tr119:
-#line 248 "mod/box/memcached-grammar.rl"
+#line 247 "mod/box/memcached-grammar.rl"
{incr = natoq(fstart, p);}
goto st75;
st75:
if ( ++p == pe )
goto _test_eof75;
case 75:
-#line 2083 "mod/box/memcached-grammar.m"
+#line 2052 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr122;
goto st0;
tr120:
-#line 248 "mod/box/memcached-grammar.rl"
+#line 247 "mod/box/memcached-grammar.rl"
{incr = natoq(fstart, p);}
goto st76;
st76:
if ( ++p == pe )
goto _test_eof76;
case 76:
-#line 2095 "mod/box/memcached-grammar.m"
+#line 2064 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr122;
case 13: goto st75;
@@ -2165,14 +2134,14 @@ case 83:
}
goto st0;
tr134:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st84;
st84:
if ( ++p == pe )
goto _test_eof84;
case 84:
-#line 2176 "mod/box/memcached-grammar.m"
+#line 2145 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr122;
case 13: goto st75;
@@ -2225,7 +2194,7 @@ case 89:
goto st0;
goto tr140;
tr140:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -2242,7 +2211,7 @@ st90:
if ( ++p == pe )
goto _test_eof90;
case 90:
-#line 2246 "mod/box/memcached-grammar.m"
+#line 2215 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr141;
case 13: goto st91;
@@ -2250,7 +2219,7 @@ case 90:
}
goto st0;
tr147:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -2258,14 +2227,14 @@ tr147:
}
goto st91;
tr158:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st91;
st91:
if ( ++p == pe )
goto _test_eof91;
case 91:
-#line 2269 "mod/box/memcached-grammar.m"
+#line 2238 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr141;
goto st0;
@@ -2284,14 +2253,14 @@ case 92:
goto tr144;
goto st0;
tr144:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st93;
st93:
if ( ++p == pe )
goto _test_eof93;
case 93:
-#line 2295 "mod/box/memcached-grammar.m"
+#line 2264 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr146;
case 13: goto tr147;
@@ -2301,7 +2270,7 @@ case 93:
goto st93;
goto st0;
tr148:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -2312,7 +2281,7 @@ st94:
if ( ++p == pe )
goto _test_eof94;
case 94:
-#line 2316 "mod/box/memcached-grammar.m"
+#line 2285 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr141;
case 13: goto st91;
@@ -2386,14 +2355,14 @@ case 101:
}
goto st0;
tr159:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st102;
st102:
if ( ++p == pe )
goto _test_eof102;
case 102:
-#line 2397 "mod/box/memcached-grammar.m"
+#line 2366 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr141;
case 13: goto st91;
@@ -2481,18 +2450,18 @@ case 111:
}
goto st0;
tr186:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st112;
tr175:
-#line 249 "mod/box/memcached-grammar.rl"
+#line 248 "mod/box/memcached-grammar.rl"
{flush_delay = natoq(fstart, p);}
goto st112;
st112:
if ( ++p == pe )
goto _test_eof112;
case 112:
-#line 2496 "mod/box/memcached-grammar.m"
+#line 2465 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr169;
goto st0;
@@ -2511,14 +2480,14 @@ case 113:
goto tr172;
goto st0;
tr172:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st114;
st114:
if ( ++p == pe )
goto _test_eof114;
case 114:
-#line 2522 "mod/box/memcached-grammar.m"
+#line 2491 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr174;
case 13: goto tr175;
@@ -2528,14 +2497,14 @@ case 114:
goto st114;
goto st0;
tr176:
-#line 249 "mod/box/memcached-grammar.rl"
+#line 248 "mod/box/memcached-grammar.rl"
{flush_delay = natoq(fstart, p);}
goto st115;
st115:
if ( ++p == pe )
goto _test_eof115;
case 115:
-#line 2539 "mod/box/memcached-grammar.m"
+#line 2508 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr169;
case 13: goto st112;
@@ -2609,14 +2578,14 @@ case 122:
}
goto st0;
tr187:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st123;
st123:
if ( ++p == pe )
goto _test_eof123;
case 123:
-#line 2620 "mod/box/memcached-grammar.m"
+#line 2589 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr169;
case 13: goto st112;
@@ -2652,18 +2621,18 @@ case 126:
}
goto st0;
tr191:
-#line 291 "mod/box/memcached-grammar.rl"
+#line 288 "mod/box/memcached-grammar.rl"
{show_cas = false;}
goto st127;
tr198:
-#line 292 "mod/box/memcached-grammar.rl"
+#line 289 "mod/box/memcached-grammar.rl"
{show_cas = true;}
goto st127;
st127:
if ( ++p == pe )
goto _test_eof127;
case 127:
-#line 2667 "mod/box/memcached-grammar.m"
+#line 2636 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 13: goto st0;
case 32: goto st127;
@@ -2672,7 +2641,7 @@ case 127:
goto st0;
goto tr193;
tr193:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -2689,7 +2658,7 @@ st128:
if ( ++p == pe )
goto _test_eof128;
case 128:
-#line 2693 "mod/box/memcached-grammar.m"
+#line 2662 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr195;
case 13: goto st129;
@@ -2932,7 +2901,7 @@ case 155:
goto st0;
goto tr222;
tr222:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -2949,7 +2918,7 @@ st156:
if ( ++p == pe )
goto _test_eof156;
case 156:
-#line 2953 "mod/box/memcached-grammar.m"
+#line 2922 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st157;
goto st0;
@@ -2963,49 +2932,49 @@ case 157:
goto tr224;
goto st0;
tr224:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st158;
st158:
if ( ++p == pe )
goto _test_eof158;
case 158:
-#line 2974 "mod/box/memcached-grammar.m"
+#line 2943 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr225;
if ( 48 <= (*p) && (*p) <= 57 )
goto st158;
goto st0;
tr225:
-#line 245 "mod/box/memcached-grammar.rl"
+#line 244 "mod/box/memcached-grammar.rl"
{flags = natoq(fstart, p);}
goto st159;
st159:
if ( ++p == pe )
goto _test_eof159;
case 159:
-#line 2988 "mod/box/memcached-grammar.m"
+#line 2957 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st159;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr228;
goto st0;
tr228:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st160;
st160:
if ( ++p == pe )
goto _test_eof160;
case 160:
-#line 3002 "mod/box/memcached-grammar.m"
+#line 2971 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr229;
if ( 48 <= (*p) && (*p) <= 57 )
goto st160;
goto st0;
tr229:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -3016,21 +2985,21 @@ st161:
if ( ++p == pe )
goto _test_eof161;
case 161:
-#line 3020 "mod/box/memcached-grammar.m"
+#line 2989 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st161;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr232;
goto st0;
tr232:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st162;
st162:
if ( ++p == pe )
goto _test_eof162;
case 162:
-#line 3034 "mod/box/memcached-grammar.m"
+#line 3003 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr233;
case 13: goto tr234;
@@ -3040,30 +3009,30 @@ case 162:
goto st162;
goto st0;
tr234:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st163;
tr247:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st163;
st163:
if ( ++p == pe )
goto _test_eof163;
case 163:
-#line 3055 "mod/box/memcached-grammar.m"
+#line 3024 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr237;
goto st0;
tr235:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st164;
st164:
if ( ++p == pe )
goto _test_eof164;
case 164:
-#line 3067 "mod/box/memcached-grammar.m"
+#line 3036 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 32: goto st164;
case 78: goto st165;
@@ -3172,7 +3141,7 @@ case 175:
goto st0;
goto tr252;
tr252:
-#line 222 "mod/box/memcached-grammar.rl"
+#line 221 "mod/box/memcached-grammar.rl"
{
fstart = p;
for (; p < pe && *p != ' ' && *p != '\r' && *p != '\n'; p++);
@@ -3189,7 +3158,7 @@ st176:
if ( ++p == pe )
goto _test_eof176;
case 176:
-#line 3193 "mod/box/memcached-grammar.m"
+#line 3162 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st177;
goto st0;
@@ -3203,49 +3172,49 @@ case 177:
goto tr254;
goto st0;
tr254:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st178;
st178:
if ( ++p == pe )
goto _test_eof178;
case 178:
-#line 3214 "mod/box/memcached-grammar.m"
+#line 3183 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr255;
if ( 48 <= (*p) && (*p) <= 57 )
goto st178;
goto st0;
tr255:
-#line 245 "mod/box/memcached-grammar.rl"
+#line 244 "mod/box/memcached-grammar.rl"
{flags = natoq(fstart, p);}
goto st179;
st179:
if ( ++p == pe )
goto _test_eof179;
case 179:
-#line 3228 "mod/box/memcached-grammar.m"
+#line 3197 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st179;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr258;
goto st0;
tr258:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st180;
st180:
if ( ++p == pe )
goto _test_eof180;
case 180:
-#line 3242 "mod/box/memcached-grammar.m"
+#line 3211 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto tr259;
if ( 48 <= (*p) && (*p) <= 57 )
goto st180;
goto st0;
tr259:
-#line 238 "mod/box/memcached-grammar.rl"
+#line 237 "mod/box/memcached-grammar.rl"
{
exptime = natoq(fstart, p);
if (exptime > 0 && exptime <= 60*60*24*30)
@@ -3256,21 +3225,21 @@ st181:
if ( ++p == pe )
goto _test_eof181;
case 181:
-#line 3260 "mod/box/memcached-grammar.m"
+#line 3229 "mod/box/memcached-grammar.m"
if ( (*p) == 32 )
goto st181;
if ( 48 <= (*p) && (*p) <= 57 )
goto tr262;
goto st0;
tr262:
-#line 221 "mod/box/memcached-grammar.rl"
+#line 220 "mod/box/memcached-grammar.rl"
{ fstart = p; }
goto st182;
st182:
if ( ++p == pe )
goto _test_eof182;
case 182:
-#line 3274 "mod/box/memcached-grammar.m"
+#line 3243 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 10: goto tr263;
case 13: goto tr264;
@@ -3280,30 +3249,30 @@ case 182:
goto st182;
goto st0;
tr264:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st183;
tr277:
-#line 280 "mod/box/memcached-grammar.rl"
+#line 277 "mod/box/memcached-grammar.rl"
{ noreply = true; }
goto st183;
st183:
if ( ++p == pe )
goto _test_eof183;
case 183:
-#line 3295 "mod/box/memcached-grammar.m"
+#line 3264 "mod/box/memcached-grammar.m"
if ( (*p) == 10 )
goto tr267;
goto st0;
tr265:
-#line 246 "mod/box/memcached-grammar.rl"
+#line 245 "mod/box/memcached-grammar.rl"
{bytes = natoq(fstart, p);}
goto st184;
st184:
if ( ++p == pe )
goto _test_eof184;
case 184:
-#line 3307 "mod/box/memcached-grammar.m"
+#line 3276 "mod/box/memcached-grammar.m"
switch( (*p) ) {
case 32: goto st184;
case 78: goto st185;
@@ -3618,7 +3587,7 @@ case 196:
_out: {}
}
-#line 305 "mod/box/memcached-grammar.rl"
+#line 302 "mod/box/memcached-grammar.rl"
if (!done) {
@@ -3643,7 +3612,6 @@ case 196:
fiber->iov_cnt = saved_iov_cnt;
fiber->iov.size = saved_iov_cnt * sizeof(struct iovec);
}
-
return 1;
}
diff --git a/mod/box/memcached-grammar.rl b/mod/box/memcached-grammar.rl
index 03294a6bf7..70046b2a77 100644
--- a/mod/box/memcached-grammar.rl
+++ b/mod/box/memcached-grammar.rl
@@ -33,7 +33,7 @@
}%%
static int __attribute__((noinline))
-memcached_dispatch()
+memcached_dispatch(struct coio *coio)
{
int cs;
u8 *p, *pe;
@@ -47,7 +47,6 @@ memcached_dispatch()
bool noreply = false;
u8 *data = NULL;
bool done = false;
- int r;
size_t saved_iov_cnt = fiber->iov_cnt;
uintptr_t flush_delay = 0;
size_t keys_count = 0;
@@ -204,7 +203,7 @@ memcached_dispatch()
}
action flush_all {
- struct fiber *f = fiber_create("flush_all", -1, flush_all);
+ struct fiber *f = fiber_create("flush_all", flush_all);
if (f)
fiber_call(f, flush_delay);
iov_add("OK\r\n", 4);
@@ -215,7 +214,7 @@ memcached_dispatch()
}
action quit {
- return 0;
+ return -1;
}
action fstart { fstart = p; }
@@ -251,10 +250,8 @@ memcached_dispatch()
action read_data {
size_t parsed = p - (u8 *)fiber->rbuf.data;
while (fiber->rbuf.size - parsed < bytes + 2) {
- if ((r = fiber_bread(&fiber->rbuf, bytes + 2 - (pe - p))) <= 0) {
- say_debug("read returned %i, closing connection", r);
- return 0;
- }
+ if (coio_bread(coio, &fiber->rbuf, bytes + 2 - (pe - p)) <= 0)
+ return -1;
}
p = fiber->rbuf.data + parsed;
@@ -326,7 +323,6 @@ memcached_dispatch()
fiber->iov_cnt = saved_iov_cnt;
fiber->iov.size = saved_iov_cnt * sizeof(struct iovec);
}
-
return 1;
}
diff --git a/mod/box/memcached.h b/mod/box/memcached.h
index bd53bfb3cd..6563841d6e 100644
--- a/mod/box/memcached.h
+++ b/mod/box/memcached.h
@@ -28,6 +28,8 @@
* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
+#include <stdarg.h>
+
struct tarantool_cfg;
void
@@ -45,6 +47,6 @@ memcached_check_config(struct tarantool_cfg *conf);
void memcached_start_expire();
void memcached_stop_expire();
-void memcached_handler(va_list /* ap */);
+void memcached_handler(va_list ap);
#endif /* TARANTOOL_MEMCACHED_H_INCLUDED */
diff --git a/mod/box/memcached.m b/mod/box/memcached.m
index 61af54a016..71a5ee5c5f 100644
--- a/mod/box/memcached.m
+++ b/mod/box/memcached.m
@@ -26,6 +26,7 @@
* THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
+#include "memcached.h"
#include "tarantool.h"
#include <limits.h>
@@ -43,6 +44,7 @@
#include "pickle.h"
#include "space.h"
#include "port.h"
+#include "coio_buf.h"
#define STAT(_) \
_(MEMC_GET, 1) \
@@ -333,59 +335,69 @@ do { \
#include "memcached-grammar.m"
void
-memcached_handler(va_list ap __attribute__((unused)))
+memcached_loop(struct coio *coio)
{
- stats.total_connections++;
- stats.curr_connections++;
- int r, p;
+ int rc;
+ int bytes_written;
int batch_count;
for (;;) {
batch_count = 0;
- if ((r = fiber_bread(&fiber->rbuf, 1)) <= 0) {
- say_debug("read returned %i, closing connection", r);
- goto exit;
- }
+ if (coio_bread(coio, &fiber->rbuf, 1) <= 0)
+ return;
dispatch:
- p = memcached_dispatch();
- if (p < 0) {
+ rc = memcached_dispatch(coio);
+ if (rc < 0) {
say_debug("negative dispatch, closing connection");
- goto exit;
+ return;
}
- if (p == 0 && batch_count == 0) /* we havn't successfully parsed any requests */
+ if (rc == 0 && batch_count == 0) /* we haven't successfully parsed any requests */
continue;
- if (p == 1) {
+ if (rc == 1) {
batch_count++;
/* some unparsed commands remain and batch count less than 20 */
if (fiber->rbuf.size > 0 && batch_count < 20)
goto dispatch;
}
- r = iov_flush();
- if (r < 0) {
- say_debug("flush_output failed, closing connection");
- goto exit;
- }
+ bytes_written = iov_flush(coio);
- stats.bytes_written += r;
+ stats.bytes_written += bytes_written;
fiber_gc();
- if (p == 1 && fiber->rbuf.size > 0) {
+ if (rc == 1 && fiber->rbuf.size > 0) {
batch_count = 0;
goto dispatch;
}
}
-exit:
- iov_flush();
- fiber_sleep(0.01);
- say_debug("exit");
- stats.curr_connections--; /* FIXME: nonlocal exit via exception will leak this counter */
}
+void memcached_handler(va_list ap)
+{
+ struct coio *coio = va_arg(ap, struct coio *);
+ stats.total_connections++;
+ stats.curr_connections++;
+
+ @try {
+ memcached_loop(coio);
+ iov_flush(coio);
+ } @catch (FiberCancelException *e) {
+ @throw;
+ } @catch (tnt_Exception *e) {
+ [e log];
+ } @finally {
+ iov_reset();
+ fiber_sleep(0.01);
+ stats.curr_connections--;
+ coio_close(coio);
+ free(coio);
+ }
+}
+
int
memcached_check_config(struct tarantool_cfg *conf)
{
@@ -556,7 +568,7 @@ void memcached_start_expire()
return;
assert(memcached_expire == NULL);
- memcached_expire = fiber_create("memcached_expire", -1,
+ memcached_expire = fiber_create("memcached_expire",
memcached_expire_loop);
if (memcached_expire == NULL)
say_error("can't start the expire fiber");
diff --git a/mod/box/txn.m b/mod/box/txn.m
index 369ca6ed6b..6adab7c56b 100644
--- a/mod/box/txn.m
+++ b/mod/box/txn.m
@@ -100,10 +100,8 @@ txn_commit(struct txn *txn)
if (txn->op == 0) /* Nothing to do. */
return;
if (! (txn->txn_flags & BOX_NOT_STORE)) {
- fiber_peer_name(fiber); /* fill the cookie */
-
int64_t lsn = next_lsn(recovery_state);
- int res = wal_write(recovery_state, lsn, fiber->cookie,
+ int res = wal_write(recovery_state, lsn, 0,
txn->op, &txn->req);
confirm_lsn(recovery_state, lsn, res == 0);
if (res)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 91f012a418..cf0c4ec67d 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -85,6 +85,8 @@ set (common_sources
util.m
sio.m
evio.m
+ coio.m
+ coio_buf.m
salloc.m
pickle.m
coro.m
diff --git a/src/admin.m b/src/admin.m
index 7d0fb92e5e..5236363682 100644
--- a/src/admin.m
+++ b/src/admin.m
@@ -46,6 +46,7 @@
#include <tbuf.h>
#include <util.h>
#include <errinj.h>
+#include "coio_buf.h"
#include "lua.h"
#include "lauxlib.h"
@@ -71,7 +72,7 @@ static const char *help =
static const char *unknown_command = "unknown command. try typing help." CRLF;
-#line 75 "src/admin.m"
+#line 76 "src/admin.m"
static const int admin_start = 1;
static const int admin_first_final = 135;
static const int admin_error = 0;
@@ -79,7 +80,7 @@ static const int admin_error = 0;
static const int admin_en_main = 1;
-#line 74 "src/admin.rl"
+#line 75 "src/admin.rl"
struct salloc_stat_admin_cb_ctx {
@@ -197,7 +198,7 @@ show_stat(struct tbuf *buf)
}
static int
-admin_dispatch(lua_State *L)
+admin_dispatch(struct coio *coio, lua_State *L)
{
struct tbuf *out = tbuf_alloc(fiber->gc_pool);
struct tbuf *err = tbuf_alloc(fiber->gc_pool);
@@ -207,20 +208,20 @@ admin_dispatch(lua_State *L)
bool state;
while ((pe = memchr(fiber->rbuf.data, '\n', fiber->rbuf.size)) == NULL) {
- if (fiber_bread(&fiber->rbuf, 1) <= 0)
- return 0;
+ if (coio_bread(coio, &fiber->rbuf, 1) <= 0)
+ return -1;
}
pe++;
p = fiber->rbuf.data;
-#line 219 "src/admin.m"
+#line 220 "src/admin.m"
{
cs = admin_start;
}
-#line 224 "src/admin.m"
+#line 225 "src/admin.m"
{
if ( p == pe )
goto _test_eof;
@@ -283,15 +284,15 @@ case 6:
}
goto st0;
tr13:
-#line 318 "src/admin.rl"
+#line 319 "src/admin.rl"
{slab_validate(); ok(out);}
goto st135;
tr20:
-#line 306 "src/admin.rl"
+#line 307 "src/admin.rl"
{return 0;}
goto st135;
tr25:
-#line 233 "src/admin.rl"
+#line 234 "src/admin.rl"
{
start(out);
tbuf_append(out, help, strlen(help));
@@ -299,9 +300,9 @@ tr25:
}
goto st135;
tr36:
-#line 292 "src/admin.rl"
+#line 293 "src/admin.rl"
{strend = p;}
-#line 239 "src/admin.rl"
+#line 240 "src/admin.rl"
{
strstart[strend-strstart]='\0';
start(out);
@@ -310,7 +311,7 @@ tr36:
}
goto st135;
tr43:
-#line 246 "src/admin.rl"
+#line 247 "src/admin.rl"
{
if (reload_cfg(err))
fail(out, err);
@@ -319,11 +320,11 @@ tr43:
}
goto st135;
tr67:
-#line 316 "src/admin.rl"
+#line 317 "src/admin.rl"
{coredump(60); ok(out);}
goto st135;
tr76:
-#line 253 "src/admin.rl"
+#line 254 "src/admin.rl"
{
int ret = snapshot(NULL, 0);
@@ -338,9 +339,9 @@ tr76:
}
goto st135;
tr98:
-#line 302 "src/admin.rl"
+#line 303 "src/admin.rl"
{ state = false; }
-#line 266 "src/admin.rl"
+#line 267 "src/admin.rl"
{
strstart[strend-strstart] = '\0';
if (errinj_set_byname(strstart, state)) {
@@ -352,9 +353,9 @@ tr98:
}
goto st135;
tr101:
-#line 301 "src/admin.rl"
+#line 302 "src/admin.rl"
{ state = true; }
-#line 266 "src/admin.rl"
+#line 267 "src/admin.rl"
{
strstart[strend-strstart] = '\0';
if (errinj_set_byname(strstart, state)) {
@@ -366,7 +367,7 @@ tr101:
}
goto st135;
tr117:
-#line 209 "src/admin.rl"
+#line 210 "src/admin.rl"
{
tarantool_cfg_iterator_t *i;
char *key, *value;
@@ -386,15 +387,15 @@ tr117:
}
goto st135;
tr131:
-#line 309 "src/admin.rl"
+#line 310 "src/admin.rl"
{start(out); fiber_info(out); end(out);}
goto st135;
tr137:
-#line 308 "src/admin.rl"
+#line 309 "src/admin.rl"
{start(out); tarantool_info(out); end(out);}
goto st135;
tr146:
-#line 227 "src/admin.rl"
+#line 228 "src/admin.rl"
{
start(out);
errinj_info(out);
@@ -402,33 +403,33 @@ tr146:
}
goto st135;
tr152:
-#line 312 "src/admin.rl"
+#line 313 "src/admin.rl"
{start(out); palloc_stat(out); end(out);}
goto st135;
tr160:
-#line 311 "src/admin.rl"
+#line 312 "src/admin.rl"
{start(out); show_slab(out); end(out);}
goto st135;
tr164:
-#line 313 "src/admin.rl"
+#line 314 "src/admin.rl"
{start(out); show_stat(out);end(out);}
goto st135;
st135:
if ( ++p == pe )
goto _test_eof135;
case 135:
-#line 421 "src/admin.m"
+#line 422 "src/admin.m"
goto st0;
tr14:
-#line 318 "src/admin.rl"
+#line 319 "src/admin.rl"
{slab_validate(); ok(out);}
goto st7;
tr21:
-#line 306 "src/admin.rl"
+#line 307 "src/admin.rl"
{return 0;}
goto st7;
tr26:
-#line 233 "src/admin.rl"
+#line 234 "src/admin.rl"
{
start(out);
tbuf_append(out, help, strlen(help));
@@ -436,9 +437,9 @@ tr26:
}
goto st7;
tr37:
-#line 292 "src/admin.rl"
+#line 293 "src/admin.rl"
{strend = p;}
-#line 239 "src/admin.rl"
+#line 240 "src/admin.rl"
{
strstart[strend-strstart]='\0';
start(out);
@@ -447,7 +448,7 @@ tr37:
}
goto st7;
tr44:
-#line 246 "src/admin.rl"
+#line 247 "src/admin.rl"
{
if (reload_cfg(err))
fail(out, err);
@@ -456,11 +457,11 @@ tr44:
}
goto st7;
tr68:
-#line 316 "src/admin.rl"
+#line 317 "src/admin.rl"
{coredump(60); ok(out);}
goto st7;
tr77:
-#line 253 "src/admin.rl"
+#line 254 "src/admin.rl"
{
int ret = snapshot(NULL, 0);
@@ -475,9 +476,9 @@ tr77:
}
goto st7;
tr99:
-#line 302 "src/admin.rl"
+#line 303 "src/admin.rl"
{ state = false; }
-#line 266 "src/admin.rl"
+#line 267 "src/admin.rl"
{
strstart[strend-strstart] = '\0';
if (errinj_set_byname(strstart, state)) {
@@ -489,9 +490,9 @@ tr99:
}
goto st7;
tr102:
-#line 301 "src/admin.rl"
+#line 302 "src/admin.rl"
{ state = true; }
-#line 266 "src/admin.rl"
+#line 267 "src/admin.rl"
{
strstart[strend-strstart] = '\0';
if (errinj_set_byname(strstart, state)) {
@@ -503,7 +504,7 @@ tr102:
}
goto st7;
tr118:
-#line 209 "src/admin.rl"
+#line 210 "src/admin.rl"
{
tarantool_cfg_iterator_t *i;
char *key, *value;
@@ -523,15 +524,15 @@ tr118:
}
goto st7;
tr132:
-#line 309 "src/admin.rl"
+#line 310 "src/admin.rl"
{start(out); fiber_info(out); end(out);}
goto st7;
tr138:
-#line 308 "src/admin.rl"
+#line 309 "src/admin.rl"
{start(out); tarantool_info(out); end(out);}
goto st7;
tr147:
-#line 227 "src/admin.rl"
+#line 228 "src/admin.rl"
{
start(out);
errinj_info(out);
@@ -539,22 +540,22 @@ tr147:
}
goto st7;
tr153:
-#line 312 "src/admin.rl"
+#line 313 "src/admin.rl"
{start(out); palloc_stat(out); end(out);}
goto st7;
tr161:
-#line 311 "src/admin.rl"
+#line 312 "src/admin.rl"
{start(out); show_slab(out); end(out);}
goto st7;
tr165:
-#line 313 "src/admin.rl"
+#line 314 "src/admin.rl"
{start(out); show_stat(out);end(out);}
goto st7;
st7:
if ( ++p == pe )
goto _test_eof7;
case 7:
-#line 558 "src/admin.m"
+#line 559 "src/admin.m"
if ( (*p) == 10 )
goto st135;
goto st0;
@@ -707,28 +708,28 @@ case 23:
}
goto tr33;
tr33:
-#line 292 "src/admin.rl"
+#line 293 "src/admin.rl"
{strstart = p;}
goto st24;
st24:
if ( ++p == pe )
goto _test_eof24;
case 24:
-#line 718 "src/admin.m"
+#line 719 "src/admin.m"
switch( (*p) ) {
case 10: goto tr36;
case 13: goto tr37;
}
goto st24;
tr34:
-#line 292 "src/admin.rl"
+#line 293 "src/admin.rl"
{strstart = p;}
goto st25;
st25:
if ( ++p == pe )
goto _test_eof25;
case 25:
-#line 732 "src/admin.m"
+#line 733 "src/admin.m"
switch( (*p) ) {
case 10: goto tr36;
case 13: goto tr37;
@@ -1178,28 +1179,28 @@ case 73:
goto tr91;
goto st0;
tr91:
-#line 300 "src/admin.rl"
+#line 301 "src/admin.rl"
{ strstart = p; }
goto st74;
st74:
if ( ++p == pe )
goto _test_eof74;
case 74:
-#line 1189 "src/admin.m"
+#line 1190 "src/admin.m"
if ( (*p) == 32 )
goto tr92;
if ( 33 <= (*p) && (*p) <= 126 )
goto st74;
goto st0;
tr92:
-#line 300 "src/admin.rl"
+#line 301 "src/admin.rl"
{ strend = p; }
goto st75;
st75:
if ( ++p == pe )
goto _test_eof75;
case 75:
-#line 1203 "src/admin.m"
+#line 1204 "src/admin.m"
switch( (*p) ) {
case 32: goto st75;
case 111: goto st76;
@@ -1891,7 +1892,7 @@ case 134:
_out: {}
}
-#line 324 "src/admin.rl"
+#line 325 "src/admin.rl"
tbuf_ltrim(&fiber->rbuf, (void *)pe - (void *)fiber->rbuf.data);
@@ -1902,37 +1903,38 @@ case 134:
end(out);
}
- return fiber_write(out->data, out->size);
+ coio_write(coio, out->data, out->size);
+ return 0;
}
static void
-admin_handler(va_list ap __attribute__((unused)))
+admin_handler(va_list ap)
{
+ struct coio *coio = va_arg(ap, struct coio *);
lua_State *L = lua_newthread(tarantool_L);
int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX);
@try {
for (;;) {
- if (admin_dispatch(L) <= 0)
+ if (admin_dispatch(coio, L) < 0)
return;
fiber_gc();
}
} @finally {
luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
+ coio_close(coio);
+ free(coio);
}
}
-int
+void
admin_init(void)
{
- if (fiber_server("admin", cfg.admin_port, admin_handler, NULL, NULL) == NULL) {
- say_syserror("can't bind to %d", cfg.admin_port);
- return -1;
- }
- return 0;
+ static struct coio_service admin;
+ coio_service_init(&admin, "admin", cfg.bind_ipaddr,
+ cfg.admin_port, admin_handler, NULL);
+ evio_service_start(&admin.evio_service);
}
-
-
/*
* Local Variables:
* mode: c
diff --git a/src/admin.rl b/src/admin.rl
index 1eacd2f38e..66847749ef 100644
--- a/src/admin.rl
+++ b/src/admin.rl
@@ -44,6 +44,7 @@
#include <tbuf.h>
#include <util.h>
#include <errinj.h>
+#include "coio_buf.h"
#include "lua.h"
#include "lauxlib.h"
@@ -188,7 +189,7 @@ show_stat(struct tbuf *buf)
}
static int
-admin_dispatch(lua_State *L)
+admin_dispatch(struct coio *coio, lua_State *L)
{
struct tbuf *out = tbuf_alloc(fiber->gc_pool);
struct tbuf *err = tbuf_alloc(fiber->gc_pool);
@@ -198,8 +199,8 @@ admin_dispatch(lua_State *L)
bool state;
while ((pe = memchr(fiber->rbuf.data, '\n', fiber->rbuf.size)) == NULL) {
- if (fiber_bread(&fiber->rbuf, 1) <= 0)
- return 0;
+ if (coio_bread(coio, &fiber->rbuf, 1) <= 0)
+ return -1;
}
pe++;
@@ -331,37 +332,38 @@ admin_dispatch(lua_State *L)
end(out);
}
- return fiber_write(out->data, out->size);
+ coio_write(coio, out->data, out->size);
+ return 0;
}
static void
-admin_handler(va_list ap __attribute__((unused)))
+admin_handler(va_list ap)
{
+ struct coio *coio = va_arg(ap, struct coio *);
lua_State *L = lua_newthread(tarantool_L);
int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX);
@try {
for (;;) {
- if (admin_dispatch(L) <= 0)
+ if (admin_dispatch(coio, L) < 0)
return;
fiber_gc();
}
} @finally {
luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref);
+ coio_close(coio);
+ free(coio);
}
}
-int
+void
admin_init(void)
{
- if (fiber_server("admin", cfg.admin_port, admin_handler, NULL, NULL) == NULL) {
- say_syserror("can't bind to %d", cfg.admin_port);
- return -1;
- }
- return 0;
+ static struct coio_service admin;
+ coio_service_init(&admin, "admin", cfg.bind_ipaddr,
+ cfg.admin_port, admin_handler, NULL);
+ evio_service_start(&admin.evio_service);
}
-
-
/*
* Local Variables:
* mode: c
diff --git a/src/coio.m b/src/coio.m
new file mode 100644
index 0000000000..8e1ee0d4d2
--- /dev/null
+++ b/src/coio.m
@@ -0,0 +1,302 @@
+/*
+ * 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 <stdio.h>
+
+#include "fiber.h"
+#include "sio.h"
+
+void
+coio_clear(struct coio *coio)
+{
+ ev_init(&coio->ev, (void *) fiber_schedule);
+ coio->ev.data = fiber;
+ coio->ev.fd = -1;
+}
+
+/** Note: this function does not throw */
+void
+coio_init(struct coio *coio, int fd)
+{
+ assert(fd >= 0);
+
+ /* Prepare for ev events. */
+ coio->ev.data = fiber;
+ ev_init(&coio->ev, (void *) fiber_schedule);
+ coio->ev.fd = fd;
+}
+
+/** Note: this function does not throw. */
+void
+coio_close(struct coio *coio)
+{
+ /* Stop I/O events. Safe to do even if not started. */
+ ev_io_stop(&coio->ev);
+
+ /* Close the socket. */
+ close(coio->ev.fd);
+ /* Make sure coio_is_connected() returns a proper value. */
+ coio->ev.fd = -1;
+}
+
+/**
+ * Connect to a host and initialize coio with connected
+ * socket.
+ */
+void
+coio_connect(struct coio *coio, struct sockaddr_in *addr)
+{
+ int fd = sio_socket();
+ @try {
+ coio_init(coio, fd);
+
+ int on = 1;
+ /* libev is non-blocking */
+ sio_setfl(fd, O_NONBLOCK, on);
+
+ /*
+ * SO_KEEPALIVE to ensure connections don't hang
+ * around for too long when a link goes away
+ */
+ sio_setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
+ &on, sizeof(on));
+ /*
+ * Lower latency is more important than higher
+ * bandwidth, and we usually write entire
+ * request/response in a single syscall.
+ */
+ sio_setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
+ &on, sizeof(on));
+
+ if (sio_connect(fd, addr, sizeof(*addr)) < 0) {
+ assert(errno == EINPROGRESS);
+ /* Wait until socket is ready for writing. */
+ ev_io_set(&coio->ev, fd, EV_WRITE);
+ ev_io_start(&coio->ev);
+ fiber_yield();
+ ev_io_stop(&coio->ev);
+ fiber_testcancel();
+
+ int error = EINPROGRESS;
+ socklen_t sz = sizeof(error);
+ sio_getsockopt(fd, SOL_SOCKET, SO_ERROR,
+ &error, &sz);
+ if (error != 0) {
+ errno = error;
+ tnt_raise(SocketError, :fd in:"connect");
+ }
+ }
+ } @catch (tnt_Exception *e) {
+ coio_close(coio);
+ @throw;
+ }
+}
+
+/**
+ * Read at least sz bytes from socket with readahead.
+ *
+ * In case of EOF returns 0.
+ * Can read up to bufsiz bytes.
+ *
+ * Returns the number of bytes read.
+ */
+ssize_t
+coio_read_ahead(struct coio *coio, void *buf, size_t sz, size_t bufsiz)
+{
+ assert(sz <= bufsiz);
+
+ ssize_t to_read = (ssize_t) sz;
+ @try {
+ while (true) {
+ /*
+ * Sic: assume the socket is ready: since
+ * the user called read(), some data must
+ * be expected.
+ */
+ ssize_t nrd = sio_read(coio->ev.fd, buf, bufsiz);
+ if (nrd > 0) {
+ to_read -= nrd;
+ if (to_read <= 0)
+ return sz - to_read;
+ buf += nrd;
+ bufsiz -= nrd;
+ } else if (nrd == 0) {
+ return 0;
+ }
+ /* The socket is not ready, yield */
+ if (! ev_is_active(&coio->ev)) {
+ ev_io_set(&coio->ev, coio->ev.fd, EV_READ);
+ ev_io_start(&coio->ev);
+ }
+ fiber_yield();
+ fiber_testcancel();
+ }
+ } @finally {
+ ev_io_stop(&coio->ev);
+ }
+}
+
+/**
+ * Read at least sz bytes, with readahead.
+ *
+ * Treats EOF as an error, and throws an exception.
+ *
+ * @retval the number of bytes read, > 0.
+ */
+ssize_t
+coio_readn_ahead(struct coio *coio, void *buf, size_t sz, size_t bufsiz)
+{
+ ssize_t nrd = coio_read_ahead(coio, buf, sz, bufsiz);
+ if (nrd < sz) {
+ errno = EPIPE;
+ tnt_raise(SocketError, :coio->ev.fd in:"unexpected EOF when reading "
+ "from socket");
+ }
+ 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.
+ */
+void
+coio_write(struct coio *coio, const void *buf, size_t sz)
+{
+ @try {
+ while (true) {
+ /*
+ * Sic: write as much data as possible,
+ * assuming the socket is ready.
+ */
+ ssize_t nwr = sio_write(coio->ev.fd, buf, sz);
+ if (nwr > 0) {
+ /* Go past the data just written. */
+ if (nwr >= sz)
+ return;
+ sz -= nwr;
+ buf += nwr;
+ }
+ if (! ev_is_active(&coio->ev)) {
+ ev_io_set(&coio->ev, coio->ev.fd, EV_WRITE);
+ ev_io_start(&coio->ev);
+ }
+ /* Yield control to other fibers. */
+ fiber_yield();
+ fiber_testcancel();
+ }
+ } @finally {
+ ev_io_stop(&coio->ev);
+ }
+}
+
+ssize_t
+coio_writev(struct coio *coio, struct iovec *iov, int iovcnt)
+{
+ ssize_t total = 0;
+ @try {
+ /* Avoid a syscall in case of 0 iovcnt. */
+ while (iovcnt) {
+ /* Write as much data as possible. */
+ ssize_t nwr = sio_writev(coio->ev.fd, iov, iovcnt);
+ if (nwr >= 0) {
+ total += nwr;
+ iov = sio_advance_iov(iov, &iovcnt, nwr);
+ if (iovcnt == 0)
+ break;
+ }
+ if (! ev_is_active(&coio->ev)) {
+ ev_io_set(&coio->ev, coio->ev.fd, EV_WRITE);
+ ev_io_start(&coio->ev);
+ }
+ /* Yield control to other fibers. */
+ fiber_yield();
+ fiber_testcancel();
+ }
+ } @finally {
+ ev_io_stop(&coio->ev);
+ }
+ return total;
+}
+
+void
+coio_service_on_accept(struct evio_service *evio_service,
+ int fd, struct sockaddr_in *addr)
+{
+ struct coio_service *service = evio_service->on_accept_param;
+ struct coio *coio = malloc(sizeof(struct coio));
+
+ if (coio == NULL)
+ goto error;
+
+ coio_init(coio, fd);
+
+ /* Set connection name. */
+ char name[SERVICE_NAME_MAXLEN];
+ snprintf(name, sizeof(name),
+ "%s/%s", evio_service->name, sio_strfaddr(addr));
+
+ /* Create the worker fiber. */
+ struct fiber *f = fiber_create(name, service->handler);
+ if (f == NULL) {
+ free(coio);
+ goto error;
+ }
+ /*
+ * The coio is passed on to the created fiber, reset the
+ * libev callback param to point at it.
+ */
+ coio->ev.data = f;
+ /*
+ * Start the created fiber. It becomes the coio object owner
+ * and will have to close it and free before termination.
+ */
+ fiber_call(f, coio, service->handler_param);
+ return;
+
+error:
+ say_error("can't create a handler fiber, dropping client connection");
+ close(fd);
+}
+
+void
+coio_service_init(struct coio_service *service, const char *name,
+ const char *host, int port,
+ void (*handler)(va_list ap), void *handler_param)
+{
+ evio_service_init(&service->evio_service, name, host, port,
+ coio_service_on_accept, service);
+ service->handler = handler;
+ service->handler_param = handler_param;
+}
+
diff --git a/src/coio_buf.m b/src/coio_buf.m
new file mode 100644
index 0000000000..ac0e5d5c73
--- /dev/null
+++ b/src/coio_buf.m
@@ -0,0 +1,31 @@
+/*
+ * 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_buf.h"
+
+int coio_readahead;
diff --git a/src/evio.m b/src/evio.m
index 891f516b6a..4d09745de6 100644
--- a/src/evio.m
+++ b/src/evio.m
@@ -79,7 +79,7 @@ evio_service_accept_cb(ev_io *watcher,
* Invoke the callback and pass it the accepted
* socket.
*/
- service->on_accept(service->on_accept_param, fd, &addr);
+ service->on_accept(service, fd, &addr);
} @catch (tnt_Exception *e) {
if (fd >= 0)
@@ -155,7 +155,8 @@ evio_service_timer_cb(ev_timer *watcher, int revents __attribute__((unused)))
void
evio_service_init(struct evio_service *service, const char *name,
const char *host, int port,
- void (*on_accept)(void *, int, struct sockaddr_in *),
+ void (*on_accept)(struct evio_service *, int,
+ struct sockaddr_in *),
void *on_accept_param)
{
memset(service, 0, sizeof(struct evio_service));
diff --git a/src/exception.m b/src/exception.m
index e233b8a433..2ca41a0ab9 100644
--- a/src/exception.m
+++ b/src/exception.m
@@ -83,7 +83,7 @@
- (void) log
{
- say(S_ERROR, strerror(errnum), "%s", errmsg);
+ say(S_ERROR, strerror(errnum), "%s in %s", object_getClassName(self), errmsg);
}
- (const char *) errmsg
diff --git a/src/fiber.m b/src/fiber.m
index 8076c81ee8..14b349c05f 100644
--- a/src/fiber.m
+++ b/src/fiber.m
@@ -56,11 +56,12 @@
#include <util.h>
#include <stat.h>
#include <pickle.h>
+#include "coio_buf.h"
@implementation FiberCancelException
@end
-#define FIBER_CALL_STACK 16
+enum { FIBER_CALL_STACK = 16 };
static struct fiber sched;
__thread struct fiber *fiber = &sched;
@@ -268,46 +269,8 @@ wait_for_child(pid_t pid)
fiber_testcancel();
}
-
-void
-fiber_io_start(int fd, int events)
-{
- ev_io *io = &fiber->io;
-
- assert (!ev_is_active(io));
-
- ev_io_set(io, fd, events);
- ev_io_start(io);
-}
-
-/** @note: this is a cancellation point.
- */
-
-void
-fiber_io_yield()
-{
- assert(ev_is_active(&fiber->io));
-
- fiber_yield();
-
- if (fiber_is_cancelled()) {
- ev_io_stop(&fiber->io);
- fiber_testcancel();
- }
-}
-
void
-fiber_io_stop(int fd __attribute__((unused)), int events __attribute__((unused)))
-{
- ev_io *io = &fiber->io;
-
- assert(ev_is_active(io) && io->fd == fd && (io->events & events));
-
- ev_io_stop(io);
-}
-
-static void
-ev_schedule(ev_watcher *watcher, int event __attribute__((unused)))
+fiber_schedule(ev_watcher *watcher, int event __attribute__((unused)))
{
assert(fiber == &sched);
fiber_call(watcher->data);
@@ -425,16 +388,14 @@ fiber_loop(void *data __attribute__((unused)))
assert(fiber != NULL && fiber->f != NULL && fiber->fid != 0);
@try {
fiber->f(fiber->f_data);
- }
- @catch (FiberCancelException *e) {
+ } @catch (FiberCancelException *e) {
say_info("fiber `%s' has been cancelled", fiber->name);
say_info("fiber `%s': exiting", fiber->name);
- }
- @catch (id e) {
+ } @catch (id e) {
say_error("fiber `%s': exception `%s'", fiber->name, object_getClassName(e));
panic("fiber `%s': exiting", fiber->name);
}
- fiber_close();
+ tbuf_reset(&fiber->rbuf);
fiber_zombificate();
fiber_yield(); /* give control back to scheduler */
}
@@ -456,7 +417,7 @@ fiber_set_name(struct fiber *fiber, const char *name)
/* fiber never dies, just become zombie */
struct fiber *
-fiber_create(const char *name, int fd, void (*f) (va_list))
+fiber_create(const char *name, void (*f) (va_list))
{
struct fiber *fiber = NULL;
@@ -475,19 +436,16 @@ fiber_create(const char *name, int fd, void (*f) (va_list))
fiber->gc_pool = palloc_create_pool("");
fiber_alloc(fiber);
- ev_init(&fiber->io, (void *)ev_schedule);
- ev_async_init(&fiber->async, (void *)ev_schedule);
+ ev_async_init(&fiber->async, (void *)fiber_schedule);
ev_async_start(&fiber->async);
- ev_init(&fiber->timer, (void *)ev_schedule);
- ev_init(&fiber->cw, (void *)ev_schedule);
- fiber->io.data = fiber->async.data = fiber->timer.data = fiber->cw.data = fiber;
+ ev_init(&fiber->timer, (void *)fiber_schedule);
+ ev_init(&fiber->cw, (void *)fiber_schedule);
+ fiber->async.data = fiber->timer.data = fiber->cw.data = fiber;
SLIST_INSERT_HEAD(&fibers, fiber, link);
}
- fiber->fd = fd;
fiber->f = f;
- memset(&fiber->f_data, 0, sizeof(fiber->f_data)); /* safety */
while (++last_used_fid <= 100) ; /* fids from 0 to 100 are reserved */
fiber->fid = last_used_fid;
fiber->flags = 0;
@@ -524,90 +482,6 @@ fiber_destroy_all()
fiber_destroy(f);
}
-
-const char *
-fiber_peer_name(struct fiber *fiber)
-{
- struct sockaddr_in peer;
- socklen_t peer_len = sizeof(peer);
-
- if (!fiber->has_peer || fiber->fd < 3)
- return NULL;
-
- if (fiber->peer_name[0] != 0)
- return fiber->peer_name;
-
- memset(&peer, 0, peer_len);
- if (getpeername(fiber->fd, (struct sockaddr *)&peer, &peer_len) < 0)
- return NULL;
-
- uint32_t zero = 0;
- if (memcmp(&peer.sin_addr, &zero, sizeof(zero)) == 0)
- return NULL;
-
- snprintf(fiber->peer_name, sizeof(fiber->peer_name),
- "%s:%d", inet_ntoa(peer.sin_addr), ntohs(peer.sin_port));
-
- fiber->cookie = 0;
- memcpy(&fiber->cookie, &peer, MIN(sizeof(peer), sizeof(fiber->cookie)));
- return fiber->peer_name;
-}
-
-int
-fiber_close(void)
-{
- if (fiber->fd < 0)
- return 0;
-
- /* We don't know if IO is active if there was an error. */
- if (ev_is_active(&fiber->io))
- fiber_io_stop(fiber->fd, -1);
-
- int r = close(fiber->fd);
-
- fiber->fd = -1;
- fiber->has_peer = false;
- fiber->peer_name[0] = 0;
- tbuf_reset(&fiber->rbuf);
-
- return r;
-}
-
-/**
- * Read at least at_least bytes from a socket.
- *
- * @retval 0 socket is closed by the sender
- * @reval -1 a system error
- * @retval >0 success, size of the last read chunk is returned
- *
- * @note: this is a cancellation point.
- */
-
-ssize_t
-fiber_bread(struct tbuf *buf, size_t at_least)
-{
- ssize_t r = 0;
- tbuf_ensure(buf, MAX(cfg.readahead, at_least));
- size_t stop_at = buf->size + at_least;
-
- fiber_io_start(fiber->fd, EV_READ);
-
- while (buf->size < stop_at) {
- fiber_io_yield();
-
- r = read(fiber->fd, buf->data + buf->size, buf->capacity - buf->size);
- if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK))
- continue;
- else if (r <= 0)
- break;
-
- buf->size += r;
- }
- fiber_io_stop(fiber->fd, EV_READ);
-
- return r;
-}
-
void
iov_reset()
{
@@ -620,345 +494,14 @@ iov_reset()
*/
ssize_t
-iov_flush(void)
+iov_flush(struct coio *coio)
{
- ssize_t result, r = 0, bytes = 0;
struct iovec *iov = iovec(&fiber->iov);
size_t iov_cnt = fiber->iov_cnt;
- fiber_io_start(fiber->fd, EV_WRITE);
- while (iov_cnt > 0) {
- fiber_io_yield();
- bytes += r = writev(fiber->fd, iov, MIN(iov_cnt, IOV_MAX));
- if (r <= 0) {
- if (errno == EAGAIN || errno == EWOULDBLOCK)
- continue;
- else
- break;
- }
-
- while (iov_cnt > 0) {
- if (iov->iov_len > r) {
- iov->iov_base += r;
- iov->iov_len -= r;
- break;
- } else {
- r -= iov->iov_len;
- iov++;
- iov_cnt--;
- }
- }
- }
- fiber_io_stop(fiber->fd, EV_WRITE);
-
- if (r < 0) {
- size_t rem = 0;
- for (int i = 0; i < iov_cnt; i++)
- rem += iov[i].iov_len;
-
- say_syserror("client unexpectedly gone, %" PRI_SZ " bytes unwritten", rem);
- result = r;
- } else
- result = bytes;
-
+ ssize_t nwr = coio_writev(coio, iov, iov_cnt);
iov_reset();
- return result;
-}
-
-/**
- * @note: this is a cancellation point.
- */
-
-ssize_t
-fiber_read(void *buf, size_t count)
-{
- ssize_t r, done = 0;
-
- fiber_io_start(fiber->fd, EV_READ);
- while (count != done) {
-
- fiber_io_yield();
-
- if ((r = read(fiber->fd, buf + done, count - done)) <= 0) {
- if (errno == EAGAIN || errno == EWOULDBLOCK)
- continue;
- else
- break;
- }
- done += r;
- }
- fiber_io_stop(fiber->fd, EV_READ);
-
- return done;
-}
-
-/**
- * @note: this is a cancellation point.
- */
-
-ssize_t
-fiber_write(const void *buf, size_t count)
-{
- int r;
- unsigned int done = 0;
-
- fiber_io_start(fiber->fd, EV_WRITE);
-
- while (count != done) {
- fiber_io_yield();
- if ((r = write(fiber->fd, buf + done, count - done)) == -1) {
- if (errno == EAGAIN || errno == EWOULDBLOCK)
- continue;
- else
- break;
- }
- done += r;
- }
- fiber_io_stop(fiber->fd, EV_WRITE);
-
- return done;
-}
-
-/**
- * @note: this is a cancellation point.
- */
-
-int
-fiber_connect(struct sockaddr_in *addr)
-{
- fiber->fd = socket(AF_INET, SOCK_STREAM, 0);
- if (fiber->fd < 0)
- goto error;
-
- if (set_nonblock(fiber->fd) < 0)
- goto error;
-
- /* set SO_KEEPALIVE flag */
- int keepalive = 1;
- if (setsockopt(fiber->fd, SOL_SOCKET, SO_KEEPALIVE,
- &keepalive, sizeof(int)) != 0)
- /* just print error, it's not critical error */
- say_syserror("setsockopt()");
-
- if (connect(fiber->fd, (struct sockaddr *)addr, sizeof(*addr)) < 0) {
-
- if (errno != EINPROGRESS)
- goto error;
-
- fiber_io_start(fiber->fd, EV_WRITE);
- fiber_io_yield();
- fiber_io_stop(fiber->fd, EV_WRITE);
-
- int error;
- socklen_t error_size = sizeof(error);
-
- if (getsockopt(fiber->fd, SOL_SOCKET, SO_ERROR,
- &error, &error_size) < 0)
- goto error;
-
- assert(error_size == sizeof(error));
-
- if (error != 0) {
- errno = error;
- goto error;
- }
- }
-
- return fiber->fd;
-
- error:
- fiber_close();
- return fiber->fd;
-}
-
-int
-set_nonblock(int sock)
-{
- int flags;
- if ((flags = fcntl(sock, F_GETFL, 0)) < 0 || fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0)
- return -1;
- return sock;
-}
-
-static void
-tcp_server_handler(va_list ap)
-{
- struct fiber_server *server = va_arg(ap, struct fiber_server *);
- struct fiber *h;
- char name[FIBER_NAME_MAXLEN];
- int fd;
- int one = 1;
-
- if (fiber_serv_socket(fiber, server->port, true, 0.1) != 0) {
- say_error("init server socket on port %i fail", server->port);
- exit(EX_OSERR);
- }
-
- if (server->on_bind != NULL) {
- server->on_bind(server->data);
- }
-
- fiber_io_start(fiber->fd, EV_READ);
- for (;;) {
- fiber_io_yield();
-
- while ((fd = accept(fiber->fd, NULL, NULL)) > 0) {
- if (set_nonblock(fd) == -1) {
- say_error("can't set nonblock");
- close(fd);
- continue;
- }
- if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
- &one, sizeof(one)) == -1) {
- say_syserror("setsockopt failed");
- /* Do nothing, not a fatal error. */
- }
-
- snprintf(name, sizeof(name), "%i/handler", server->port);
- h = fiber_create(name, fd, server->handler);
- if (h == NULL) {
- say_error("can't create handler fiber, dropping client connection");
- close(fd);
- continue;
- }
-
- h->has_peer = true;
- fiber_call(h, server->data);
- }
- if (fd < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
- say_syserror("accept");
- continue;
- }
- }
- fiber_io_stop(fiber->fd, EV_READ);
-}
-
-struct fiber *
-fiber_server(const char *name, int port, void (*handler) (va_list ap), void *data,
- void (*on_bind) (void *data))
-{
- char server_name[FIBER_NAME_MAXLEN];
- struct fiber_server *server;
- struct fiber *s;
-
- snprintf(server_name, sizeof(server_name), "%i/%s", port, name);
- server = palloc(eter_pool, sizeof(struct fiber_server));
- assert(server != NULL);
- server->data = data;
- server->port = port;
- server->handler = handler;
- server->on_bind = on_bind;
- s = fiber_create(server_name, -1, tcp_server_handler);
-
- fiber_call(s, server); /* give a handler a chance */
- return s;
-}
-
-/** create new fiber's socket and set standat options. */
-static int
-create_socket(struct fiber *fiber)
-{
- if (fiber->fd != -1) {
- say_error("fiber is already has socket");
- goto create_socket_fail;
- }
-
- fiber->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
- if (fiber->fd == -1) {
- say_syserror("socket");
- goto create_socket_fail;
- }
-
- int one = 1;
- if (setsockopt(fiber->fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
- say_syserror("setsockopt");
- goto create_socket_fail;
- }
-
- struct linger ling = { 0, 0 };
- if (setsockopt(fiber->fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one)) != 0 ||
- setsockopt(fiber->fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0 ||
- setsockopt(fiber->fd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)) != 0) {
- say_syserror("setsockopt");
- goto create_socket_fail;
- }
-
- if (set_nonblock(fiber->fd) == -1) {
- goto create_socket_fail;
- }
-
- return 0;
-
-create_socket_fail:
-
- if (fiber->fd != -1) {
- close(fiber->fd);
- }
- return -1;
-}
-
-/** Create server socket and bind his on port. */
-int
-fiber_serv_socket(struct fiber *fiber, unsigned short port, bool retry, ev_tstamp delay)
-{
- const ev_tstamp min_delay = 0.001; /* minimal delay is 1 msec */
- struct sockaddr_in sin;
- bool warning_said = false;
-
- if (delay < min_delay) {
- delay = min_delay;
- }
-
- if (create_socket(fiber) != 0) {
- return -1;
- }
-
- /* clean sockaddr_in struct */
- memset(&sin, 0, sizeof(struct sockaddr_in));
-
- /* fill sockaddr_in struct */
- sin.sin_family = AF_INET;
- sin.sin_port = htons(port);
- if (strcmp(cfg.bind_ipaddr, "INADDR_ANY") == 0) {
- sin.sin_addr.s_addr = INADDR_ANY;
- } else {
- if (!inet_aton(cfg.bind_ipaddr, &sin.sin_addr)) {
- say_syserror("inet_aton");
- return -1;
- }
- }
-
- while (true) {
- if (bind(fiber->fd, (struct sockaddr *)&sin, sizeof(sin)) != 0) {
- if (retry && (errno == EADDRINUSE)) {
- /* retry mode, try, to bind after delay */
- goto sleep_and_retry;
- }
- say_syserror("bind");
- return -1;
- }
- if (listen(fiber->fd, cfg.backlog) != 0) {
- if (retry && (errno == EADDRINUSE)) {
- /* retry mode, try, to bind after delay */
- goto sleep_and_retry;
- }
- say_syserror("listen");
- return -1;
- }
-
- say_info("bound to port %i", port);
- break;
-
- sleep_and_retry:
- if (!warning_said) {
- say_warn("port %i is already in use, "
- "will retry binding after %lf seconds.", port, delay);
- warning_said = true;
- }
- fiber_sleep(delay);
- }
-
- return 0;
+ return nwr;
}
void
@@ -973,8 +516,6 @@ fiber_info(struct tbuf *out)
tbuf_printf(out, " - fid: %4i" CRLF, fiber->fid);
tbuf_printf(out, " csw: %i" CRLF, fiber->csw);
tbuf_printf(out, " name: %s" CRLF, fiber->name);
- tbuf_printf(out, " fd: %4i" CRLF, fiber->fd);
- tbuf_printf(out, " peer: %s" CRLF, fiber_peer_name(fiber));
tbuf_printf(out, " stack: %p" CRLF, stack_top);
#ifdef ENABLE_BACKTRACE
tbuf_printf(out, " backtrace:" CRLF "%s",
@@ -1000,6 +541,8 @@ fiber_init(void)
sp = call_stack;
fiber = &sched;
last_used_fid = 100;
+
+ coio_binit(cfg.readahead);
}
void
diff --git a/src/iproto.m b/src/iproto.m
index 446c64c92c..4495dd8e04 100644
--- a/src/iproto.m
+++ b/src/iproto.m
@@ -37,6 +37,7 @@
#include <fiber.h>
#include <tbuf.h>
#include <say.h>
+#include "coio_buf.h"
const uint32_t msg_ping = 0xff00;
@@ -45,52 +46,52 @@ static void iproto_reply(iproto_callback callback, struct tbuf *request);
static void
iproto_validate_header(struct iproto_header *header);
-inline static int
-iproto_flush(ssize_t to_read)
+inline static void
+iproto_flush(struct coio *coio, ssize_t to_read)
{
/*
* Flush output and garbage collect before reading
* next header.
*/
if (to_read > 0) {
- if (iov_flush() < 0) {
- say_warn("io_error: %s", strerror(errno));
- return -1;
- }
+ iov_flush(coio);
fiber_gc();
}
- return 0;
}
void
iproto_interact(va_list ap)
{
+ struct coio *coio = va_arg(ap, struct coio *);
iproto_callback callback = va_arg(ap, iproto_callback);
struct tbuf *in = &fiber->rbuf;
ssize_t to_read = sizeof(struct iproto_header);
+ @try {
+ for (;;) {
+ if (to_read > 0 && coio_bread(coio, in, to_read) <= 0)
+ break;
- for (;;) {
- if (to_read > 0 && fiber_bread(in, to_read) <= 0)
- break;
+ /* validating iproto package header */
+ iproto_validate_header(iproto(in));
- /* validating iproto package header */
- iproto_validate_header(iproto(in));
+ ssize_t request_len = sizeof(struct iproto_header)
+ + iproto(in)->len;
+ to_read = request_len - in->size;
- ssize_t request_len = sizeof(struct iproto_header)
- + iproto(in)->len;
- to_read = request_len - in->size;
+ iproto_flush(coio, to_read);
- if (iproto_flush(to_read) == -1)
- break;
- if (to_read > 0 && fiber_bread(in, to_read) <= 0)
- break;
+ if (to_read > 0 && coio_bread(coio, in, to_read) <= 0)
+ break;
- struct tbuf *request = tbuf_split(in, request_len);
- iproto_reply(callback, request);
+ struct tbuf *request = tbuf_split(in, request_len);
+ iproto_reply(callback, request);
- to_read = sizeof(struct iproto_header) - in->size;
- if (iproto_flush(to_read) == -1)
- break;
+ to_read = sizeof(struct iproto_header) - in->size;
+ iproto_flush(coio, to_read);
+ }
+ } @finally {
+ coio_close(coio);
+ free(coio);
}
}
diff --git a/src/replica.m b/src/replica.m
index c36384d002..fbf56435ec 100644
--- a/src/replica.m
+++ b/src/replica.m
@@ -35,112 +35,96 @@
#include "log_io.h"
#include "fiber.h"
#include "pickle.h"
+#include "coio_buf.h"
-static int
+static void
remote_apply_row(struct recovery_state *r, struct tbuf *row);
static struct tbuf *
-remote_row_reader_v11()
+remote_read_row(struct coio *coio)
{
ssize_t to_read = sizeof(struct header_v11) - fiber->rbuf.size;
- if (to_read > 0 && fiber_bread(&fiber->rbuf, to_read) <= 0)
- goto error;
+ if (to_read > 0)
+ coio_breadn(coio, &fiber->rbuf, to_read);
ssize_t request_len = header_v11(&fiber->rbuf)->len + sizeof(struct header_v11);
to_read = request_len - fiber->rbuf.size;
- if (to_read > 0 && fiber_bread(&fiber->rbuf, to_read) <= 0)
- goto error;
+ if (to_read > 0)
+ coio_breadn(coio, &fiber->rbuf, to_read);
- say_debug("read row bytes:%" PRI_SSZ, request_len);
return tbuf_split(&fiber->rbuf, request_len);
-error:
- say_error("unexpected eof reading row header");
- return NULL;
}
-static struct tbuf *
-remote_read_row(struct sockaddr_in *remote_addr, i64 initial_lsn)
+static void
+remote_connect(struct coio *coio, struct sockaddr_in *remote_addr,
+ i64 initial_lsn, const char **err)
{
- struct tbuf *row;
- bool warning_said = false;
- const int reconnect_delay = 1;
- const char *err = NULL;
- u32 version;
-
- for (;;) {
- if (fiber->fd < 0) {
- if (fiber_connect(remote_addr) < 0) {
- err = "can't connect to master";
- goto err;
- }
+ *err = "can't connect to master";
+ coio_connect(coio, remote_addr);
- if (fiber_write(&initial_lsn, sizeof(initial_lsn)) != sizeof(initial_lsn)) {
- err = "can't write version";
- goto err;
- }
-
- if (fiber_read(&version, sizeof(version)) != sizeof(version)) {
- err = "can't read version";
- goto err;
- }
-
- if (version != default_version) {
- err = "remote version mismatch";
- goto err;
- }
-
- say_crit("successfully connected to master");
- say_crit("starting replication from lsn:%" PRIi64, initial_lsn);
-
- warning_said = false;
- err = NULL;
- }
+ *err = "can't write version";
+ coio_write(coio, &initial_lsn, sizeof(initial_lsn));
- row = remote_row_reader_v11();
- if (row == NULL) {
- err = "can't read row";
- goto err;
- }
-
- return row;
-
- err:
- if (err != NULL && !warning_said) {
- say_info("%s", err);
- say_info("will retry every %i second", reconnect_delay);
- warning_said = true;
- }
- fiber_close();
- fiber_sleep(reconnect_delay);
- }
+ u32 version;
+ *err = "can't read version";
+ coio_readn(coio, &version, sizeof(version));
+ *err = NULL;
+ if (version != default_version)
+ tnt_raise(SystemError, :"remote version mismatch");
+
+ say_crit("successfully connected to master");
+ say_crit("starting replication from lsn: %" PRIi64, initial_lsn);
}
static void
pull_from_remote(va_list ap)
{
struct recovery_state *r = va_arg(ap, struct recovery_state *);
- struct tbuf *row;
+ struct coio coio;
+ bool warning_said = false;
+ const int reconnect_delay = 1;
+ coio_clear(&coio);
for (;;) {
- fiber_setcancelstate(true);
- row = remote_read_row(&r->remote->addr, r->confirmed_lsn + 1);
- fiber_setcancelstate(false);
-
- r->remote->recovery_lag = ev_now() - header_v11(row)->tm;
- r->remote->recovery_last_update_tstamp = ev_now();
+ const char *err = NULL;
+ @try {
+ fiber_setcancelstate(true);
+ if (! coio_is_connected(&coio)) {
+ remote_connect(&coio, &r->remote->addr,
+ r->confirmed_lsn + 1, &err);
+ warning_said = false;
+ }
+ err = "can't read row";
+ struct tbuf *row = remote_read_row(&coio);
+ fiber_setcancelstate(false);
+ err = NULL;
- if (remote_apply_row(r, row) < 0) {
- fiber_close();
- continue;
+ r->remote->recovery_lag = ev_now() - header_v11(row)->tm;
+ r->remote->recovery_last_update_tstamp = ev_now();
+
+ remote_apply_row(r, row);
+
+ fiber_gc();
+ } @catch (FiberCancelException *e) {
+ coio_close(&coio);
+ @throw;
+ } @catch (tnt_Exception *e) {
+ [e log];
+ if (! warning_said) {
+ if (err != NULL)
+ say_info("%s", err);
+ say_info("will retry every %i second", reconnect_delay);
+ warning_said = true;
+ }
+ coio_close(&coio);
+ fiber_sleep(reconnect_delay);
}
-
- fiber_gc();
}
}
-static int
+static void
remote_apply_row(struct recovery_state *r, struct tbuf *row)
{
struct tbuf *data;
@@ -166,8 +150,6 @@ remote_apply_row(struct recovery_state *r, struct tbuf *row)
panic("replication failure: can't write row to WAL");
set_lsn(r, lsn);
-
- return 0;
}
void
@@ -185,7 +167,7 @@ recovery_follow_remote(struct recovery_state *r, const char *addr)
say_crit("initializing the replica, WAL master %s", addr);
snprintf(name, sizeof(name), "replica/%s", addr);
- f = fiber_create(name, -1, pull_from_remote);
+ f = fiber_create(name, pull_from_remote);
if (f == NULL)
return;
diff --git a/src/replication.m b/src/replication.m
index 6c2fc08637..557de40e58 100644
--- a/src/replication.m
+++ b/src/replication.m
@@ -81,7 +81,7 @@ static int master_to_spawner_socket;
* to the spawner.
*/
static void
-replication_on_accept(void *data __attribute__((unused)),
+replication_on_accept(struct evio_service *service __attribute__((unused)),
int fd, struct sockaddr_in *addr __attribute__((unused)));
/** Send a file descriptor to replication relay spawner.
@@ -225,8 +225,9 @@ replication_init()
/** Replication acceptor fiber handler. */
static void
-replication_on_accept(void *data __attribute__((unused)),
- int fd, struct sockaddr_in *addr __attribute__((unused)))
+replication_on_accept(struct evio_service *service __attribute__((unused)),
+ int fd,
+ struct sockaddr_in *addr __attribute__((unused)))
{
/*
* Drop the O_NONBLOCK flag, which was possibly
diff --git a/src/say.m b/src/say.m
index 79c72c4e73..6d08187349 100644
--- a/src/say.m
+++ b/src/say.m
@@ -41,6 +41,7 @@
#include <fiber.h>
#include TARANTOOL_CONFIG
#include "tarantool.h"
+#include "sio.h"
int sayfd = STDERR_FILENO;
pid_t logger_pid;
@@ -112,7 +113,7 @@ say_logger_init(int nonblock)
}
out:
if (nonblock)
- set_nonblock(sayfd);
+ sio_setfl(sayfd, O_NONBLOCK, 1);
setvbuf(stderr, NULL, _IONBF, 0);
}
@@ -120,7 +121,6 @@ say_logger_init(int nonblock)
void
vsay(int level, const char *filename, int line, const char *error, const char *format, va_list ap)
{
- const char *peer_name = fiber_peer_name(fiber);
size_t p = 0, len = PIPE_BUF;
const char *f;
static __thread char buf[PIPE_BUF];
@@ -136,9 +136,6 @@ vsay(int level, const char *filename, int line, const char *error, const char *f
ev_now_update();
- if (peer_name == NULL)
- peer_name = "_";
-
for (f = filename; *f; f++)
if (*f == '/' && *(f + 1) != '\0')
filename = f + 1;
@@ -152,8 +149,8 @@ vsay(int level, const char *filename, int line, const char *error, const char *f
p += snprintf(buf + p, len - p, ":%06.3f",
ev_now() - now + tm.tm_sec);
- p += snprintf(buf + p, len - p, " [%i] %i/%s %s", getpid(),
- fiber->fid, fiber->name, peer_name);
+ p += snprintf(buf + p, len - p, " [%i] %i/%s", getpid(),
+ fiber->fid, fiber->name);
if (level == S_WARN || level == S_ERROR)
p += snprintf(buf + p, len - p, " %s:%i", filename, line);
diff --git a/src/sio.m b/src/sio.m
index f548dc82e0..dfb39b8189 100644
--- a/src/sio.m
+++ b/src/sio.m
@@ -73,7 +73,7 @@ sio_socketname(int fd)
va_end(ap);
const char *socketname = sio_socketname(fd);
errno = save_errno;
- self = [self init: "in %s, called on %s", buf, socketname];
+ self = [self init: "%s, called on %s", buf, socketname];
return self;
}
@end
diff --git a/src/tarantool.m b/src/tarantool.m
index 2c29e568b3..7227784cba 100644
--- a/src/tarantool.m
+++ b/src/tarantool.m
@@ -829,7 +829,8 @@ main(int argc, char **argv)
start_time = ev_now();
ev_loop(0);
} @catch (tnt_Exception *e) {
- panic("%s", [e errmsg]);
+ [e log];
+ panic("%s", "Fatal error, exiting loop");
}
say_crit("exiting loop");
/* freeing resources */
diff --git a/src/tarantool_lua.m b/src/tarantool_lua.m
index e98759392e..f7584b79c4 100644
--- a/src/tarantool_lua.m
+++ b/src/tarantool_lua.m
@@ -781,7 +781,7 @@ lbox_fiber_create(struct lua_State *L)
luaL_error(L, "fiber.create(function): recursion limit"
" reached");
- struct fiber *f = fiber_create("lua", -1, box_lua_fiber_run);
+ struct fiber *f = fiber_create("lua", box_lua_fiber_run);
/* Initially the fiber is cancellable */
f->flags |= FIBER_USER_MODE | FIBER_CANCELLABLE;
@@ -1484,7 +1484,7 @@ tarantool_lua_load_init_script(struct lua_State *L)
* To work this problem around we must run init script in
* a separate fiber.
*/
- struct fiber *loader = fiber_create(TARANTOOL_LUA_INIT_SCRIPT, -1,
+ struct fiber *loader = fiber_create(TARANTOOL_LUA_INIT_SCRIPT,
load_init_script);
fiber_call(loader, L);
/* Outside the startup file require() or ffi are not
diff --git a/test/box/lua.result b/test/box/lua.result
index e6749acf02..5b9bf3338a 100644
--- a/test/box/lua.result
+++ b/test/box/lua.result
@@ -1682,13 +1682,13 @@ error: ''
lua old_name = box.fiber.name()
---
...
-lua box.fiber.name()
+lua box.fiber.name() == old_name
---
- - 33015/handler
+ - true
...
-lua box.fiber.self():name()
+lua box.fiber.self():name() == old_name
---
- - 33015/handler
+ - true
...
lua box.fiber.name('hello fiber')
---
diff --git a/test/box/lua.test b/test/box/lua.test
index 78fe92535c..e9cab5c50a 100644
--- a/test/box/lua.test
+++ b/test/box/lua.test
@@ -561,8 +561,8 @@ print """# A test case for Bug#1043804 lua error() -> server crash"""
exec admin "lua error()"
print """# Test box.fiber.name()"""
exec admin "lua old_name = box.fiber.name()"
-exec admin "lua box.fiber.name()"
-exec admin "lua box.fiber.self():name()"
+exec admin "lua box.fiber.name() == old_name"
+exec admin "lua box.fiber.self():name() == old_name"
exec admin "lua box.fiber.name('hello fiber')"
exec admin "lua box.fiber.name()"
exec admin "lua box.fiber.self():name('bye fiber')"
--
GitLab