Box.session.push() allows to send a message to a client with no
finishing a main request.
Tarantool supports two push types: via text protocol and via
binary protocol. Text protocol push message contains "push:"
prefix followed by a YAML formatted text, where as a final
response always has '---' prefix. To catch pushed messages on a
client side use console.connect() on_push options which takes
a function with a single argument - message.
IProto message is encoded just like a regular response, but
instead of IPROTO_DATA it has IPROTO_PUSH with a single element -
pushed MessagePack encoded data.
Text push is trivial - it is just blocking write into a socket
with a prefix. Binary push is more complex.
TX thread to notify IProto thread about new data in obuf sends
a message 'push_msg'. IProto thread, got this message, notifies
libev about new data, and then sends 'push_msg' back with
updated write position. TX thread, received the message back,
updates its version of a write position. If IProto will not send
a write position, then TX will write to the same obuf again
and again, because it can not know that IProto already flushed
another obuf.
To avoid multiple 'push_msg' in fly between IProto and TX, the
only one 'push_msg' per connection is used. To deliver pushes,
appeared when 'push_msg' was in fly, TX thread sets a flag every
time when sees, that 'push_msg' is sent, and there is a new push.
When 'push_msg' returns, it checks this flag, and if it is set,
then the IProto is notified again.
Closes #2677
Signed-off-by: Vladislav Shpilevoy <v.shpilevoy@xxxxxxxxxxxxx>
---
src/box/iproto.cc | 235 +++++++++++++++++---
src/box/iproto_constants.c | 3 +-
src/box/iproto_constants.h | 8 +
src/box/lua/call.c | 1 +
src/box/lua/console.c | 2 +-
src/box/lua/console.h | 8 +
src/box/lua/console.lua | 6 +-
src/box/lua/net_box.c | 37 ++++
src/box/lua/net_box.lua | 97 ++++++--
src/box/lua/session.c | 171 +++++++++++++++
src/box/port.c | 7 +
src/box/port.h | 15 ++
src/box/session.cc | 14 ++
src/box/session.h | 17 ++
src/box/xrow.c | 40 +++-
src/box/xrow.h | 26 ++-
src/fio.c | 12 +
src/fio.h | 16 ++
test/app-tap/console.test.lua | 9 +-
test/box/net.box.result | 2 +-
test/box/net.box.test.lua | 2 +-
test/box/push.result | 364 +++++++++++++++++++++++++++++++
test/box/push.test.lua | 163 ++++++++++++++
test/replication/before_replace.result | 14 ++
test/replication/before_replace.test.lua | 11 +
25 files changed, 1211 insertions(+), 69 deletions(-)
create mode 100644 test/box/push.result
create mode 100644 test/box/push.test.lua
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index ad4b1a757..4408293c1 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -63,6 +63,45 @@
enum { IPROTO_SALT_SIZE = 32 };
+/**
+ * This structure represents a position in the output.
+ * Since we use rotating buffers to recycle memory,
+ * it includes not only a position in obuf, but also
+ * a pointer to obuf the position is for.
+ */
+struct iproto_wpos {
+ struct obuf *obuf;
+ struct obuf_svp svp;
+};
+
+static void
+iproto_wpos_create(struct iproto_wpos *wpos, struct obuf *out)
+{
+ wpos->obuf = out;
+ wpos->svp = obuf_create_svp(out);
+}
+
+/* {{{ IPROTO_PUSH declarations. */
+
+/**
+ * Message to notify IProto thread about new data in an output
+ * buffer. Struct iproto_msg is not used here, because push
+ * notification can be much more compact: it does not have
+ * request, ibuf, length, flags ...
+ */
+struct iproto_push_msg {
+ struct cmsg base;
+ /**
+ * Before sending to IProto thread, the wpos is set to a
+ * current position in an output buffer. Before IProto
+ * returns the message to TX, it sets wpos to the last
+ * flushed position (works like iproto_msg.wpos).
+ */
+ struct iproto_wpos wpos;
+ /** IProto connection to push into. */
+ struct iproto_connection *connection;
+};
+
/** Owner of binary IProto sessions. */
struct iproto_session_owner {
struct session_owner base;
@@ -70,6 +109,37 @@ struct iproto_session_owner {
char salt[IPROTO_SALT_SIZE];
/** IProto connection. */
struct iproto_connection *connection;
+ /**
+ * Is_push_in_progress is set, when a push_msg is sent to
+ * IProto thread, and reset, when the message is returned
+ * to TX. If a new push sees, that a push_msg is already
+ * sent to IProto, then has_new_pushes is set. After push
+ * notification is returned to TX, it checks
+ * has_new_pushes. If it is set, then the notification is
+ * sent again. This ping-pong continues, until TX stopped
+ * pushing. It allows to
+ * 1) avoid multiple push_msg from one session in fly,
+ * 2) do not block push() until a previous push() is
+ * finished.
+ *
+ * IProto TX
+ * -------------------------------------------------------
+ * + [push message]
+ * start socket <--- notification ----
+ * write
+ * + [push message]
+ * + [push message]
+ * ...
+ * end socket
+ * write ----------------> check for new
+ * pushes - found
+ * <--- notification ---
+ * ....
+ */
+ bool has_new_pushes;
+ bool is_push_in_progress;
+ /** Push notification for IProto thread. */
+ struct iproto_push_msg push_msg;
};
static struct session_owner *
@@ -78,10 +148,27 @@ iproto_session_owner_dup(struct session_owner *owner);
static int
iproto_session_owner_fd(const struct session_owner *owner);
+/**
+ * Push a message from @a port to a remote client.
+ * @param owner IProto session owner.
+ * @param sync Message sync. Must be the same as a request sync to
+ * be able to detect their tie on a client side.
+ * @param port Port with data to send.
+ *
+ * @retval -1 Memory error.
+ * @retval 0 Success, a message is wrote to an output buffer. But
+ * it is not guaranteed, that it will be sent
+ * successfully.
+ */
+static int
+iproto_session_owner_push(struct session_owner *owner, uint64_t sync,
+ struct port *port);
+
static const struct session_owner_vtab iproto_session_owner_vtab = {
/* .dup = */ iproto_session_owner_dup,
/* .delete = */ (void (*)(struct session_owner *)) free,
/* .fd = */ iproto_session_owner_fd,
+ /* .push = */ iproto_session_owner_push,
};
static struct session_owner *
@@ -105,6 +192,9 @@ iproto_session_owner_create(struct iproto_session_owner
*owner,
owner->base.type = SESSION_TYPE_BINARY;
owner->base.vtab = &iproto_session_owner_vtab;
owner->connection = connection;
+ owner->has_new_pushes = false;
+ owner->is_push_in_progress = false;
+ owner->push_msg.connection = connection;
random_bytes(owner->salt, IPROTO_SALT_SIZE);
}
@@ -117,6 +207,8 @@ iproto_session_salt(struct session *session)
return session_owner->salt;
}
+/** }}} */
+
/* The number of iproto messages in flight */
enum { IPROTO_MSG_MAX = 768 };
@@ -164,24 +256,6 @@ iproto_reset_input(struct ibuf *ibuf)
}
}
-/**
- * This structure represents a position in the output.
- * Since we use rotating buffers to recycle memory,
- * it includes not only a position in obuf, but also
- * a pointer to obuf the position is for.
- */
-struct iproto_wpos {
- struct obuf *obuf;
- struct obuf_svp svp;
-};
-
-static void
-iproto_wpos_create(struct iproto_wpos *wpos, struct obuf *out)
-{
- wpos->obuf = out;
- wpos->svp = obuf_create_svp(out);
-}
-
/* {{{ iproto_msg - declaration */
/**
@@ -1168,15 +1242,16 @@ tx_discard_input(struct iproto_msg *msg)
* not, the empty buffer is selected.
* - if neither of the buffers are empty, the function
* does not rotate the buffer.
+ *
+ * @param con IProto connection.
+ * @param wpos Last flushed write position, received from IProto
+ * thread.
*/
-static struct iproto_msg *
-tx_accept_msg(struct cmsg *m)
+static void
+tx_accept_msg(struct iproto_connection *con, struct iproto_wpos *wpos)
{
- struct iproto_msg *msg = (struct iproto_msg *) m;
- struct iproto_connection *con = msg->connection;
-
struct obuf *prev = &con->obuf[con->tx.p_obuf == con->obuf];
- if (msg->wpos.obuf == con->tx.p_obuf) {
+ if (wpos->obuf == con->tx.p_obuf) {
/*
* We got a message advancing the buffer which
* is being appended to. The previous buffer is
@@ -1194,7 +1269,6 @@ tx_accept_msg(struct cmsg *m)
*/
con->tx.p_obuf = prev;
}
- return msg;
}
/**
@@ -1217,7 +1291,8 @@ tx_reply_error(struct iproto_msg *msg)
static void
tx_reply_iproto_error(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
+ struct iproto_msg *msg = (struct iproto_msg *) m;
+ tx_accept_msg(msg->connection, &msg->wpos);
struct obuf *out = msg->connection->tx.p_obuf;
iproto_reply_error(out, diag_last_error(&msg->diag),
msg->header.sync, ::schema_version);
@@ -1227,8 +1302,8 @@ tx_reply_iproto_error(struct cmsg *m)
static void
tx_process1(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
- struct obuf *out = msg->connection->tx.p_obuf;
+ struct iproto_msg *msg = (struct iproto_msg *) m;
+ tx_accept_msg(msg->connection, &msg->wpos);
tx_fiber_init(msg->connection->session, msg->header.sync);
if (tx_check_schema(msg->header.schema_version))
@@ -1236,8 +1311,11 @@ tx_process1(struct cmsg *m)
struct tuple *tuple;
struct obuf_svp svp;
- if (box_process1(&msg->dml, &tuple) ||
- iproto_prepare_select(out, &svp))
+ struct obuf *out;
+ if (box_process1(&msg->dml, &tuple) != 0)
+ goto error;
+ out = msg->connection->tx.p_obuf;
+ if (iproto_prepare_select(out, &svp) != 0)
goto error;
if (tuple && tuple_to_obuf(tuple, out))
goto error;
@@ -1252,8 +1330,9 @@ error:
static void
tx_process_select(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
- struct obuf *out = msg->connection->tx.p_obuf;
+ struct iproto_msg *msg = (struct iproto_msg *) m;
+ tx_accept_msg(msg->connection, &msg->wpos);
+ struct obuf *out;
struct obuf_svp svp;
struct port port;
int count;
@@ -1270,6 +1349,7 @@ tx_process_select(struct cmsg *m)
req->key, req->key_end, &port);
if (rc < 0)
goto error;
+ out = msg->connection->tx.p_obuf;
if (iproto_prepare_select(out, &svp) != 0) {
port_destroy(&port);
goto error;
@@ -1305,7 +1385,8 @@ tx_process_call_on_yield(struct trigger *trigger, void
*event)
static void
tx_process_call(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
+ struct iproto_msg *msg = (struct iproto_msg *) m;
+ tx_accept_msg(msg->connection, &msg->wpos);
tx_fiber_init(msg->connection->session, msg->header.sync);
@@ -1387,7 +1468,8 @@ error:
static void
tx_process_misc(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
+ struct iproto_msg *msg = (struct iproto_msg *) m;
+ tx_accept_msg(msg->connection, &msg->wpos);
struct obuf *out = msg->connection->tx.p_obuf;
struct session *session = msg->connection->session;
@@ -1428,8 +1510,9 @@ error:
static void
tx_process_join_subscribe(struct cmsg *m)
{
- struct iproto_msg *msg = tx_accept_msg(m);
+ struct iproto_msg *msg = (struct iproto_msg *) m;
struct iproto_connection *con = msg->connection;
+ tx_accept_msg(con, &msg->wpos);
tx_fiber_init(con->session, msg->header.sync);
@@ -1612,6 +1695,88 @@ static const struct cmsg_hop connect_route[] = {
/** }}} */
+/** {{{ IPROTO_PUSH implementation. */
+
+/**
+ * Send to IProto thread a notification about new pushes.
+ * @param owner IProto session owner.
+ */
+static void
+tx_begin_push(struct iproto_session_owner *owner);
+
+/**
+ * Create an event to send push.
+ * @param m IProto push message.
+ */
+static void
+net_push_msg(struct cmsg *m)
+{
+ struct iproto_push_msg *msg = (struct iproto_push_msg *) m;
+ struct iproto_connection *con = msg->connection;
+ con->wend = msg->wpos;
+ msg->wpos = con->wpos;
+ if (evio_has_fd(&con->output) && !ev_is_active(&con->output))
+ ev_feed_event(con->loop, &con->output, EV_WRITE);
+}
+
+/**
+ * After a message notifies IProto thread about pushed data, TX
+ * thread can already have a new push in one of obufs. This
+ * function checks for new pushes and possibly re-sends push
+ * notification to IProto thread.
+ */
+static void
+tx_check_for_new_push(struct cmsg *m)
+{
+ struct iproto_push_msg *msg = (struct iproto_push_msg *) m;
+ struct iproto_session_owner *owner =
+ container_of(msg, struct iproto_session_owner, push_msg);
+ tx_accept_msg(msg->connection, &msg->wpos);
+ owner->is_push_in_progress = false;
+ if (owner->has_new_pushes)
+ tx_begin_push(owner);
+}
+
+static const struct cmsg_hop push_route[] = {
+ { net_push_msg, &tx_pipe },
+ { tx_check_for_new_push, NULL }
+};
+
+static void
+tx_begin_push(struct iproto_session_owner *owner)
+{
+ assert(! owner->is_push_in_progress);
+ cmsg_init((struct cmsg *) &owner->push_msg, push_route);
+ iproto_wpos_create(&owner->push_msg.wpos, owner->connection->tx.p_obuf);
+ owner->has_new_pushes = false;
+ owner->is_push_in_progress = true;
+ cpipe_push(&net_pipe, (struct cmsg *) &owner->push_msg);
+}
+
+static int
+iproto_session_owner_push(struct session_owner *session_owner, uint64_t sync,
+ struct port *port)
+{
+ struct iproto_session_owner *owner =
+ (struct iproto_session_owner *) session_owner;
+ struct iproto_connection *con = owner->connection;
+ struct obuf_svp svp;
+ if (iproto_prepare_push(con->tx.p_obuf, &svp) != 0)
+ return -1;
+ if (port_dump(port, con->tx.p_obuf) != 0) {
+ obuf_rollback_to_svp(con->tx.p_obuf, &svp);
+ return -1;
+ }
+ iproto_reply_push(con->tx.p_obuf, &svp, sync, ::schema_version);
+ if (! owner->is_push_in_progress)
+ tx_begin_push(owner);
+ else
+ owner->has_new_pushes = true;
+ return 0;
+}
+
+/** }}} */
+
/**
* Create a connection and start input.
*/
diff --git a/src/box/iproto_constants.c b/src/box/iproto_constants.c
index cd7b1d03b..893275c7b 100644
--- a/src/box/iproto_constants.c
+++ b/src/box/iproto_constants.c
@@ -174,7 +174,8 @@ const char *iproto_key_strs[IPROTO_KEY_MAX] = {
NULL, /* 0x2e */
NULL, /* 0x2f */
"data", /* 0x30 */
- "error" /* 0x31 */
+ "error", /* 0x31 */
+ "push", /* 0x32 */
};
const char *vy_page_info_key_strs[VY_PAGE_INFO_KEY_MAX] = {
diff --git a/src/box/iproto_constants.h b/src/box/iproto_constants.h
index 951842485..05d262a0b 100644
--- a/src/box/iproto_constants.h
+++ b/src/box/iproto_constants.h
@@ -79,6 +79,14 @@ enum iproto_key {
/* Leave a gap between request keys and response keys */
IPROTO_DATA = 0x30,
IPROTO_ERROR = 0x31,
+ /**
+ * Tarantool supports two push types: binary and text.
+ * A text push can be distinguished from a response by a
+ * prefix "push:".
+ * Binary push is encoded using IPROTO_PUSH key in a
+ * message body, which replaces IPROTO_DATA.
+ */
+ IPROTO_PUSH = 0x32,
IPROTO_KEY_MAX
};
diff --git a/src/box/lua/call.c b/src/box/lua/call.c
index be13812aa..fc70ed430 100644
--- a/src/box/lua/call.c
+++ b/src/box/lua/call.c
@@ -418,6 +418,7 @@ port_lua_destroy(struct port *base)
static const struct port_vtab port_lua_vtab = {
.dump = port_lua_dump,
.dump_16 = port_lua_dump_16,
+ .dump_raw = NULL,
.destroy = port_lua_destroy,
};
diff --git a/src/box/lua/console.c b/src/box/lua/console.c
index 450745c90..3bc4b6425 100644
--- a/src/box/lua/console.c
+++ b/src/box/lua/console.c
@@ -329,7 +329,7 @@ lbox_console_add_history(struct lua_State *L)
return 0;
}
-static int
+int
lbox_console_format(struct lua_State *L)
{
int arg_count = lua_gettop(L);
diff --git a/src/box/lua/console.h b/src/box/lua/console.h
index 208b31490..92a4af035 100644
--- a/src/box/lua/console.h
+++ b/src/box/lua/console.h
@@ -39,6 +39,14 @@ struct lua_State;
void
tarantool_lua_console_init(struct lua_State *L);
+/**
+ * Encode Lua object into YAML string.
+ * @param Lua object to encode on top of a stack.
+ * @retval Lua string.
+ */
+int
+lbox_console_format(struct lua_State *L);
+
#if defined(__cplusplus)
} /* extern "C" */
#endif /* defined(__cplusplus) */
diff --git a/src/box/lua/console.lua b/src/box/lua/console.lua
index b4199ef85..295bfcbaa 100644
--- a/src/box/lua/console.lua
+++ b/src/box/lua/console.lua
@@ -59,7 +59,7 @@ end
--
-- Evaluate command on remote instance
--
-local function remote_eval(self, line)
+local function remote_eval(self, line, opts)
if not line or self.remote.state ~= 'active' then
local err = self.remote.error
self.remote:close()
@@ -74,7 +74,7 @@ local function remote_eval(self, line)
--
-- execute line
--
- local ok, res = pcall(self.remote.eval, self.remote, line)
+ local ok, res = pcall(self.remote.eval, self.remote, line, opts)
return ok and res or format(false, res)
end
@@ -310,7 +310,7 @@ local function connect(uri, opts)
-- override methods
self.remote = remote
- self.eval = remote_eval
+ self.eval = function(s, l) return remote_eval(s, l, {on_push =
opts.on_push}) end
self.prompt = string.format("%s:%s", self.remote.host, self.remote.port)
self.completion = function (str, pos1, pos2)
local c = string.format(
diff --git a/src/box/lua/net_box.c b/src/box/lua/net_box.c
index db2d2dbb4..9fa7935f7 100644
--- a/src/box/lua/net_box.c
+++ b/src/box/lua/net_box.c
@@ -554,6 +554,41 @@ handle_error:
return 2;
}
+/**
+ * Search for a "push:" prefix in a message, received from a
+ * server using a text protocol.
+ */
+static int
+netbox_text_is_push(struct lua_State *L)
+{
+ assert(lua_gettop(L) == 2);
+ uint32_t ctypeid;
+ const char *text = *(const char **)luaL_checkcdata(L, 1, &ctypeid);
+ assert(ctypeid == luaL_ctypeid(L, "char *"));
+ uint32_t len = (uint32_t) lua_tonumber(L, 2);
+ uint32_t push_len = strlen("push:");
+ lua_pushboolean(L, len >= 5 && memcmp(text, "push:", push_len) == 0);
+ return 1;
+}
+
+/**
+ * Search for IPROTO_PUSH key in a MessagePack encoded response
+ * body. It is needed without entire message decoding, when a user
+ * wants to store raw responses and pushes in its own buffer.
+ */
+static int
+netbox_body_is_push(struct lua_State *L)
+{
+ uint32_t ctypeid;
+ const char *body = *(const char **)luaL_checkcdata(L, 1, &ctypeid);
+ assert(ctypeid == luaL_ctypeid(L, "char *"));
+ assert(mp_typeof(*body) == MP_MAP);
+ lua_pushboolean(L, mp_decode_map(&body) == 1 &&
+ mp_typeof(*body) == MP_UINT &&
+ mp_decode_uint(&body) == IPROTO_PUSH);
+ return 1;
+}
+
int
luaopen_net_box(struct lua_State *L)
{
@@ -571,6 +606,8 @@ luaopen_net_box(struct lua_State *L)
{ "encode_auth", netbox_encode_auth },
{ "decode_greeting",netbox_decode_greeting },
{ "communicate", netbox_communicate },
+ { "text_is_push", netbox_text_is_push },
+ { "body_is_push", netbox_body_is_push },
{ NULL, NULL}
};
/* luaL_register_module polutes _G */
diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
index 87c8c548b..7ef6a6a2d 100644
--- a/src/box/lua/net_box.lua
+++ b/src/box/lua/net_box.lua
@@ -26,6 +26,8 @@ local communicate = internal.communicate
local encode_auth = internal.encode_auth
local encode_select = internal.encode_select
local decode_greeting = internal.decode_greeting
+local text_is_push = internal.text_is_push
+local body_is_push = internal.body_is_push
local sequence_mt = { __serialize = 'sequence' }
local TIMEOUT_INFINITY = 500 * 365 * 86400
@@ -38,6 +40,7 @@ local IPROTO_SYNC_KEY = 0x01
local IPROTO_SCHEMA_VERSION_KEY = 0x05
local IPROTO_DATA_KEY = 0x30
local IPROTO_ERROR_KEY = 0x31
+local IPROTO_PUSH_KEY = 0x32
local IPROTO_GREETING_SIZE = 128
-- select errors from box.error
@@ -229,7 +232,8 @@ local function create_transport(host, port, user, password,
callback)
end
-- REQUEST/RESPONSE --
- local function perform_request(timeout, buffer, method, schema_version,
...)
+ local function perform_request(timeout, buffer, method, on_push,
+ schema_version, ...)
if state ~= 'active' then
return last_errno or E_NO_CONNECTION, last_error
end
@@ -242,7 +246,7 @@ local function create_transport(host, port, user, password,
callback)
local id = next_request_id
method_codec[method](send_buf, id, schema_version, ...)
next_request_id = next_id(id)
- local request = table_new(0, 6) -- reserve space for 6 keys
+ local request = table_new(0, 7) -- reserve space for 7 keys
request.client = fiber_self()
request.method = method
request.schema_version = schema_version
@@ -254,6 +258,17 @@ local function create_transport(host, port, user,
password, callback)
requests[id] = nil
return E_TIMEOUT, 'Timeout exceeded'
end
+ if request.messages then
+ -- Multiple push messages can appear on a single
+ -- event loop iteration.
+ local messages = request.messages
+ request.messages = nil
+ if on_push then
+ for _, m in pairs(messages) do
+ on_push(m)
+ end
+ end
+ end
until requests[id] == nil -- i.e. completed (beware spurious wakeups)
return request.errno, request.response
end
@@ -270,12 +285,12 @@ local function create_transport(host, port, user,
password, callback)
if request == nil then -- nobody is waiting for the response
return
end
- requests[id] = nil
local status = hdr[IPROTO_STATUS_KEY]
local body, body_end_check
if status ~= 0 then
-- Handle errors
+ requests[id] = nil
body, body_end_check = decode(body_rpos)
assert(body_end == body_end_check, "invalid xrow length")
request.errno = band(status, IPROTO_ERRNO_MASK)
@@ -290,7 +305,16 @@ local function create_transport(host, port, user,
password, callback)
local body_len = body_end - body_rpos
local wpos = buffer:alloc(body_len)
ffi.copy(wpos, body_rpos, body_len)
- request.response = tonumber(body_len)
+ if body_is_push(body_rpos) then
+ if request.messages then
+ table.insert(request.messages, tonumber(body_len))
+ else
+ request.messages = {tonumber(body_len)}
+ end
+ else
+ requests[id] = nil
+ request.response = tonumber(body_len)
+ end
wakeup_client(request.client)
return
end
@@ -298,7 +322,18 @@ local function create_transport(host, port, user,
password, callback)
-- Decode xrow.body[DATA] to Lua objects
body, body_end_check = decode(body_rpos)
assert(body_end == body_end_check, "invalid xrow length")
- request.response = body[IPROTO_DATA_KEY]
+ if body[IPROTO_PUSH_KEY] then
+ assert(#body[IPROTO_PUSH_KEY] == 1)
+ assert(not body[IPROTO_DATA_KEY])
+ if request.messages then
+ table.insert(request.messages, body[IPROTO_PUSH_KEY][1])
+ else
+ request.messages = {body[IPROTO_PUSH_KEY][1]}
+ end
+ else
+ requests[id] = nil
+ request.response = body[IPROTO_DATA_KEY]
+ end
wakeup_client(request.client)
end
@@ -345,9 +380,16 @@ local function create_transport(host, port, user,
password, callback)
if err then
return err, delim_pos
else
- local response = ffi.string(recv_buf.rpos, delim_pos + #delim)
+ local response
+ local is_push = text_is_push(recv_buf.rpos, delim_pos + #delim)
+ if not is_push then
+ response = ffi.string(recv_buf.rpos, delim_pos + #delim)
+ else
+ -- 5 - len of 'push:' prefix of a message.
+ response = ffi.string(recv_buf.rpos + 5, delim_pos + #delim -
5)
+ end
recv_buf.rpos = recv_buf.rpos + delim_pos + #delim
- return nil, response
+ return nil, response, is_push
end
end
@@ -408,18 +450,26 @@ local function create_transport(host, port, user,
password, callback)
console_sm = function(rid)
local delim = '\n...\n'
- local err, response = send_and_recv_console()
+ local err, response, is_push = send_and_recv_console()
if err then
return error_sm(err, response)
else
local request = requests[rid]
- if request == nil then -- nobody is waiting for the response
- return
+ if request then
+ if is_push then
+ -- In a console mode it is impossible to
+ -- get multiple pushes on a single event loop
+ -- iteration.
+ assert(not request.messages)
+ request.messages = {response}
+ else
+ requests[rid] = nil
+ request.response = response
+ rid = next_id(rid)
+ end
+ wakeup_client(request.client)
+ return console_sm(rid)
end
- requests[rid] = nil
- request.response = response
- wakeup_client(request.client)
- return console_sm(next_id(rid))
end
end
@@ -761,7 +811,8 @@ function remote_methods:_request(method, opts, ...)
timeout = deadline and max(0, deadline - fiber_clock())
end
err, res = perform_request(timeout, buffer, method,
- self.schema_version, ...)
+ opts and opts.on_push, self.schema_version,
+ ...)
if not err and buffer ~= nil then
return res -- the length of xrow.body
elseif not err then
@@ -791,7 +842,7 @@ function remote_methods:ping(opts)
timeout = deadline and max(0, deadline - fiber_clock())
or (opts and opts.timeout)
end
- local err = self._transport.perform_request(timeout, nil, 'ping',
+ local err = self._transport.perform_request(timeout, nil, 'ping', nil,
self.schema_version)
return not err or err == E_WRONG_SCHEMA_VERSION
end
@@ -956,8 +1007,16 @@ console_methods.on_disconnect =
remote_methods.on_disconnect
console_methods.on_connect = remote_methods.on_connect
console_methods.is_connected = remote_methods.is_connected
console_methods.wait_state = remote_methods.wait_state
-function console_methods:eval(line, timeout)
+function console_methods:eval(line, opts)
check_remote_arg(self, 'eval')
+ local timeout
+ local on_push
+ if type(opts) == 'table' then
+ timeout = opts.timeout or TIMEOUT_INFINITY
+ on_push = opts.on_push
+ else
+ timeout = opts
+ end
local err, res
local transport = self._transport
local pr = transport.perform_request
@@ -968,10 +1027,10 @@ function console_methods:eval(line, timeout)
end
if self.protocol == 'Binary' then
local loader = 'return require("console").eval(...)'
- err, res = pr(timeout, nil, 'eval', nil, loader, {line})
+ err, res = pr(timeout, nil, 'eval', on_push, nil, loader, {line})
else
assert(self.protocol == 'Lua console')
- err, res = pr(timeout, nil, 'inject', nil, line..'$EOF$\n')
+ err, res = pr(timeout, nil, 'inject', on_push, nil, line..'$EOF$\n')
end
if err then
box.error({code = err, reason = res})
diff --git a/src/box/lua/session.c b/src/box/lua/session.c
index af8411068..73cc9da8f 100644
--- a/src/box/lua/session.c
+++ b/src/box/lua/session.c
@@ -31,6 +31,7 @@
#include "session.h"
#include "lua/utils.h"
#include "lua/trigger.h"
+#include "lua/msgpack.h"
#include <lua.h>
#include <lauxlib.h>
@@ -41,6 +42,10 @@
#include "box/session.h"
#include "box/user.h"
#include "box/schema.h"
+#include "box/port.h"
+#include "box/lua/console.h"
+#include "fio.h"
+#include "small/obuf.h"
/** Owner of a console session. */
struct console_session_owner {
@@ -55,10 +60,26 @@ console_session_owner_dup(struct session_owner *owner);
static int
console_session_owner_fd(const struct session_owner *owner);
+/**
+ * Send "push:" prefix + message in a blocking mode, with no
+ * yields, to a console socket.
+ * @param owner Console session owner.
+ * @param sync Sync. It is unused since a text protocol has no
+ * syncs.
+ * @param port Port with text to dump.
+ *
+ * @retval -1 Memory or IO error.
+ * @retval 0 Success.
+ */
+static int
+console_session_owner_push(struct session_owner *owner, uint64_t sync,
+ struct port *port);
+
static const struct session_owner_vtab console_session_owner_vtab = {
/* .dup = */ console_session_owner_dup,
/* .delete = */ (void (*)(struct session_owner *)) free,
/* .fd = */ console_session_owner_fd,
+ /* .push = */ console_session_owner_push,
};
static struct session_owner *
@@ -90,6 +111,45 @@ console_session_owner_create(struct console_session_owner
*owner, int fd)
owner->fd = fd;
}
+/**
+ * Write @a text into @a fd in a blocking mode, ignoring transient
+ * socket errors.
+ * @param fd Console descriptor.
+ * @param text Text to send.
+ * @param len Length of @a text.
+ */
+static inline int
+console_do_push(int fd, const char *text, uint32_t len)
+{
+ while (len > 0) {
+ int written = fio_write_silent(fd, text, len);
+ if (written < 0)
+ return -1;
+ assert((uint32_t) written <= len);
+ len -= written;
+ text += written;
+ }
+ return 0;
+}
+
+static int
+console_session_owner_push(struct session_owner *owner, uint64_t sync,
+ struct port *port)
+{
+ /* Console has no sync. */
+ (void) sync;
+ assert(owner->vtab == &console_session_owner_vtab);
+ int fd = console_session_owner_fd(owner);
+ uint32_t text_len;
+ const char *text = port_dump_raw(port, &text_len);
+ if (text == NULL ||
+ console_do_push(fd, "push:", strlen("push:")) != 0 ||
+ console_do_push(fd, text, text_len) != 0)
+ return -1;
+ else
+ return 0;
+}
+
static const char *sessionlib_name = "box.session";
/* Create session and pin it to fiber */
@@ -418,6 +478,116 @@ lbox_push_on_access_denied_event(struct lua_State *L,
void *event)
return 3;
}
+/**
+ * Port to push a message from Lua.
+ */
+struct lua_push_port {
+ const struct port_vtab *vtab;
+ /**
+ * Lua state, containing data to dump on top of the stack.
+ */
+ struct lua_State *L;
+};
+
+/**
+ * Lua push port supports two dump types: usual and raw. Raw dump
+ * encodes a message as a YAML formatted text, usual dump encodes
+ * the message as MessagePack right into an output buffer.
+ */
+static int
+lua_push_port_dump_msgpack(struct port *port, struct obuf *out);
+
+static const char *
+lua_push_port_dump_text(struct port *port, uint32_t *size);
+
+static const struct port_vtab lua_push_port_vtab = {
+ .dump = lua_push_port_dump_msgpack,
+ /*
+ * Dump_16 has no sense, since push appears since 1.10
+ * protocol.
+ */
+ .dump_16 = NULL,
+ .dump_raw = lua_push_port_dump_text,
+ .destroy = NULL,
+};
+
+static void
+obuf_error_cb(void *ctx)
+{
+ *((int *)ctx) = -1;
+}
+
+static int
+lua_push_port_dump_msgpack(struct port *port, struct obuf *out)
+{
+ struct lua_push_port *lua_port = (struct lua_push_port *) port;
+ assert(lua_port->vtab == &lua_push_port_vtab);
+ struct mpstream stream;
+ int rc = 0;
+ /*
+ * Do not use luamp_error to allow a caller to clear the
+ * obuf, if it already has allocated something (for
+ * example, iproto push reserves memory for a header).
+ */
+ mpstream_init(&stream, out, obuf_reserve_cb, obuf_alloc_cb,
+ obuf_error_cb, &rc);
+ luamp_encode(lua_port->L, luaL_msgpack_default, &stream, 1);
+ if (rc != 0)
+ return -1;
+ mpstream_flush(&stream);
+ return 0;
+}
+
+static const char *
+lua_push_port_dump_text(struct port *port, uint32_t *size)
+{
+ struct lua_push_port *lua_port = (struct lua_push_port *) port;
+ assert(lua_port->vtab == &lua_push_port_vtab);
+ lbox_console_format(lua_port->L);
+ assert(lua_isstring(lua_port->L, -1));
+ size_t len;
+ const char *result = lua_tolstring(lua_port->L, -1, &len);
+ *size = (uint32_t) len;
+ return result;
+}
+
+/**
+ * Push a message using a protocol, depending on a session type.
+ * @param data Data to push, first argument on a stack.
+ * @param opts Options. Now requires a single possible option -
+ * sync. Second argument on a stack.
+ */
+static int
+lbox_session_push(struct lua_State *L)
+{
+ if (lua_gettop(L) != 2 || !lua_istable(L, 2)) {
+usage_error:
+ return luaL_error(L, "Usage: box.session.push(data, opts)");
+ }
+ lua_getfield(L, 2, "sync");
+ if (! lua_isnumber(L, 3))
+ goto usage_error;
+ double lua_sync = lua_tonumber(L, 3);
+ lua_pop(L, 1);
+ uint64_t sync = (uint64_t) lua_sync;
+ if (lua_sync != sync)
+ goto usage_error;
+ struct lua_push_port port;
+ port.vtab = &lua_push_port_vtab;
+ port.L = L;
+ /*
+ * Pop the opts - they must not be pushed. Leave only data
+ * on a stack.
+ */
+ lua_remove(L, 2);
+ if (session_push(current_session(), sync, (struct port *) &port) != 0) {
+ return luaT_error(L);
+ } else {
+ lua_pushboolean(L, true);
+ return 1;
+ }
+}
+
/**
* Sets trigger on_access_denied.
* For test purposes only.
@@ -489,6 +659,7 @@ box_lua_session_init(struct lua_State *L)
{"on_disconnect", lbox_session_on_disconnect},
{"on_auth", lbox_session_on_auth},
{"on_access_denied", lbox_session_on_access_denied},
+ {"push", lbox_session_push},
{NULL, NULL}
};
luaL_register_module(L, sessionlib_name, sessionlib);
diff --git a/src/box/port.c b/src/box/port.c
index 03f6be79d..7f3cea5e7 100644
--- a/src/box/port.c
+++ b/src/box/port.c
@@ -143,6 +143,12 @@ port_dump_16(struct port *port, struct obuf *out)
return port->vtab->dump_16(port, out);
}
+const char *
+port_dump_raw(struct port *port, uint32_t *size)
+{
+ return port->vtab->dump_raw(port, size);
+}
+
void
port_init(void)
{
@@ -159,5 +165,6 @@ port_free(void)
const struct port_vtab port_tuple_vtab = {
.dump = port_tuple_dump,
.dump_16 = port_tuple_dump_16,
+ .dump_raw = NULL,
.destroy = port_tuple_destroy,
};
diff --git a/src/box/port.h b/src/box/port.h
index 7cf3339b5..d1db74909 100644
--- a/src/box/port.h
+++ b/src/box/port.h
@@ -76,6 +76,11 @@ struct port_vtab {
* format.
*/
int (*dump_16)(struct port *port, struct obuf *out);
+ /**
+ * Same as dump, but find a memory for an output buffer
+ * for itself.
+ */
+ const char *(*dump_raw)(struct port *port, uint32_t *size);
/**
* Destroy a port and release associated resources.
*/
@@ -158,6 +163,16 @@ port_dump(struct port *port, struct obuf *out);
int
port_dump_16(struct port *port, struct obuf *out);
+/**
+ * Same as port_dump(), but find a memory for an output buffer for
+ * itself.
+ * @param port Port to dump.
+ * @param[out] size Size of the data.
+ * @retval Data.
+ */
+const char *
+port_dump_raw(struct port *port, uint32_t *size);
+
void
port_init(void);
diff --git a/src/box/session.cc b/src/box/session.cc
index 908ec9c4e..793393fc8 100644
--- a/src/box/session.cc
+++ b/src/box/session.cc
@@ -63,6 +63,7 @@ static const struct session_owner_vtab
generic_session_owner_vtab = {
/* .dup = */ generic_session_owner_dup,
/* .delete = */ (void (*)(struct session_owner *)) free,
/* .fd = */ generic_session_owner_fd,
+ /* .push = */ generic_session_owner_push,
};
static struct session_owner *
@@ -95,6 +96,19 @@ session_owner_create(struct session_owner *owner, enum
session_type type)
owner->vtab = &generic_session_owner_vtab;
}
+int
+generic_session_owner_push(struct session_owner *owner, uint64_t sync,
+ struct port *port)
+{
+ (void) owner;
+ (void) sync;
+ (void) port;
+ const char *session =
+ tt_sprintf("Session '%s'", session_type_strs[owner->type]);
+ diag_set(ClientError, ER_UNSUPPORTED, session, "push()");
+ return -1;
+}
+
static inline uint64_t
sid_max()
{
diff --git a/src/box/session.h b/src/box/session.h
index 105dcab17..b5f4955d8 100644
--- a/src/box/session.h
+++ b/src/box/session.h
@@ -59,6 +59,7 @@ enum session_type {
extern const char *session_type_strs[];
struct session_owner_vtab;
+struct port;
/**
* Object to store session type specific data. For example, IProto
@@ -78,8 +79,18 @@ struct session_owner_vtab {
void (*free)(struct session_owner *);
/** Get the descriptor of an owner, if has. Else -1. */
int (*fd)(const struct session_owner *);
+ /** Push a port data into a session owner's channel. */
+ int (*push)(struct session_owner *, uint64_t, struct port *);
};
+/**
+ * In a common case, a session does not support push. This
+ * function always returns -1 and sets ER_UNSUPPORTED error.
+ */
+int
+generic_session_owner_push(struct session_owner *owner, uint64_t sync,
+ struct port *port);
+
static inline struct session_owner *
session_owner_dup(struct session_owner *owner)
{
@@ -142,6 +153,12 @@ session_fd(const struct session *session)
return session->owner->vtab->fd(session->owner);
}
+static inline int
+session_push(struct session *session, uint64_t sync, struct port *port)
+{
+ return session->owner->vtab->push(session->owner, sync, port);
+}
+
/**
* Find a session by id.
*/
diff --git a/src/box/xrow.c b/src/box/xrow.c
index f48525645..f3e929f69 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -43,6 +43,9 @@
#include "scramble.h"
#include "iproto_constants.h"
+static_assert(IPROTO_DATA < 0x7f && IPROTO_PUSH < 0x7f,
+ "encoded IPROTO_BODY keys must fit into one byte");
+
int
xrow_header_decode(struct xrow_header *header, const char **pos,
const char *end)
@@ -231,6 +234,9 @@ struct PACKED iproto_body_bin {
uint32_t v_data_len; /* string length of array size */
};
+static_assert(sizeof(struct iproto_body_bin) + IPROTO_HEADER_LEN ==
+ IPROTO_SELECT_HEADER_LEN, "size of the prepared select");
+
static const struct iproto_body_bin iproto_body_bin = {
0x81, IPROTO_DATA, 0xdd, 0
};
@@ -239,6 +245,19 @@ static const struct iproto_body_bin iproto_error_bin = {
0x81, IPROTO_ERROR, 0xdb, 0
};
+struct PACKED iproto_body_push_bin {
+ uint8_t m_body; /* MP_MAP */
+ uint8_t k_data; /* IPROTO_PUSH */
+ uint8_t v_data; /* 1-size MP_ARRAY */
+};
+
+static_assert(sizeof(struct iproto_body_push_bin) + IPROTO_HEADER_LEN ==
+ IPROTO_PUSH_HEADER_LEN, "size of the prepared push");
+
+static const struct iproto_body_push_bin iproto_push_bin = {
+ 0x81, IPROTO_PUSH, 0x91
+};
+
/** Return a 4-byte numeric error code, with status flags. */
static inline uint32_t
iproto_encode_error(uint32_t error)
@@ -337,23 +356,21 @@ iproto_write_error(int fd, const struct error *e,
uint32_t schema_version,
(void) write(fd, e->errmsg, msg_len);
}
-enum { SVP_SIZE = IPROTO_HEADER_LEN + sizeof(iproto_body_bin) };
-
int
-iproto_prepare_select(struct obuf *buf, struct obuf_svp *svp)
+iproto_prepare_header(struct obuf *buf, struct obuf_svp *svp, size_t size)
{
/**
* Reserve memory before taking a savepoint.
* This ensures that we get a contiguous chunk of memory
* and the savepoint is pointing at the beginning of it.
*/
- void *ptr = obuf_reserve(buf, SVP_SIZE);
+ void *ptr = obuf_reserve(buf, size);
if (ptr == NULL) {
- diag_set(OutOfMemory, SVP_SIZE, "obuf", "reserve");
+ diag_set(OutOfMemory, size, "obuf_reserve", "ptr");
return -1;
}
*svp = obuf_create_svp(buf);
- ptr = obuf_alloc(buf, SVP_SIZE);
+ ptr = obuf_alloc(buf, size);
assert(ptr != NULL);
return 0;
}
@@ -373,6 +390,17 @@ iproto_reply_select(struct obuf *buf, struct obuf_svp
*svp, uint64_t sync,
memcpy(pos + IPROTO_HEADER_LEN, &body, sizeof(body));
}
+void
+iproto_reply_push(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
+ uint32_t schema_version)
+{
+ char *pos = (char *) obuf_svp_to_ptr(buf, svp);
+ iproto_header_encode(pos, IPROTO_OK, sync, schema_version,
+ obuf_size(buf) - svp->used - IPROTO_HEADER_LEN);
+ memcpy(pos + IPROTO_HEADER_LEN, &iproto_push_bin,
+ sizeof(iproto_push_bin));
+}
+
int
xrow_decode_dml(struct xrow_header *row, struct request *request,
uint64_t key_map)
diff --git a/src/box/xrow.h b/src/box/xrow.h
index d407d151b..d2ec394c5 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -50,6 +50,10 @@ enum {
XROW_HEADER_LEN_MAX = 40,
XROW_BODY_LEN_MAX = 128,
IPROTO_HEADER_LEN = 28,
+ /** 7 = sizeof(iproto_body_bin). */
+ IPROTO_SELECT_HEADER_LEN = IPROTO_HEADER_LEN + 7,
+ /** 3 = sizeof(iproto_body_push_bin). */
+ IPROTO_PUSH_HEADER_LEN = IPROTO_HEADER_LEN + 3,
};
struct xrow_header {
@@ -344,8 +348,18 @@ iproto_header_encode(char *data, uint32_t type, uint64_t
sync,
struct obuf;
struct obuf_svp;
+/**
+ * Reserve obuf space for a header, which depends on a response
+ * size.
+ */
int
-iproto_prepare_select(struct obuf *buf, struct obuf_svp *svp);
+iproto_prepare_header(struct obuf *buf, struct obuf_svp *svp, size_t size);
+
+static inline int
+iproto_prepare_select(struct obuf *buf, struct obuf_svp *svp)
+{
+ return iproto_prepare_header(buf, svp, IPROTO_SELECT_HEADER_LEN);
+}
/**
* Write select header to a preallocated buffer.
@@ -355,6 +369,16 @@ void
iproto_reply_select(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
uint32_t schema_version, uint32_t count);
+static inline int
+iproto_prepare_push(struct obuf *buf, struct obuf_svp *svp)
+{
+ return iproto_prepare_header(buf, svp, IPROTO_PUSH_HEADER_LEN);
+}
+
+void
+iproto_reply_push(struct obuf *buf, struct obuf_svp *svp, uint64_t sync,
+ uint32_t schema_version);
+
/**
* Encode iproto header with IPROTO_OK response code.
* @param out Encode to.
diff --git a/src/fio.c b/src/fio.c
index b79d3d058..a813b7760 100644
--- a/src/fio.c
+++ b/src/fio.c
@@ -134,6 +134,18 @@ fio_writen(int fd, const void *buf, size_t count)
return 0;
}
+int
+fio_write_silent(int fd, const void *buf, size_t count)
+{
+ ssize_t nwr = write(fd, buf, count);
+ if (nwr >= 0)
+ return nwr;
+ if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
+ return 0;
+ else
+ return -1;
+}
+
ssize_t
fio_writev(int fd, struct iovec *iov, int iovcnt)
{
diff --git a/src/fio.h b/src/fio.h
index 12749afcb..c88e19258 100644
--- a/src/fio.h
+++ b/src/fio.h
@@ -108,6 +108,22 @@ fio_pread(int fd, void *buf, size_t count, off_t offset);
int
fio_writen(int fd, const void *buf, size_t count);
+/**
+ * Write the given buffer with no retrying for partial writes,
+ * and ignore transient write errors: EINTR, EWOULDBLOCK and
+ * EAGAIN - these errors are not logged, and the function returns
+ * 0.
+ * @param fd File descriptor to write to.
+ * @param buf Buffer to write.
+ * @param count Size of @a buf.
+ *
+ * @retval -1 Non-transient error.
+ * @retval 0 Nothing to write, or a transient error.
+ * @retval > 0 Written byte count.
+ */
+int
+fio_write_silent(int fd, const void *buf, size_t count);
+
/**
* A simple wrapper around writev().
* Re-tries write in case of EINTR.
diff --git a/test/app-tap/console.test.lua b/test/app-tap/console.test.lua
index 48d28bd6d..c91576dc2 100755
--- a/test/app-tap/console.test.lua
+++ b/test/app-tap/console.test.lua
@@ -21,7 +21,7 @@ local EOL = "\n...\n"
test = tap.test("console")
-test:plan(59)
+test:plan(61)
-- Start console and connect to it
local server = console.listen(CONSOLE_SOCKET)
@@ -31,6 +31,13 @@ local handshake = client:read{chunk = 128}
test:ok(string.match(handshake, '^Tarantool .*console') ~= nil, 'Handshake')
test:ok(client ~= nil, "connect to console")
+--
+-- gh-2677: box.session.push, text protocol support.
+--
+client:write('box.session.push(200, {sync = 0})\n')
+test:is(client:read(EOL):find('push:'), 1, "read pushed value")
+test:is(client:read(EOL):find('true'), 7, "read box.session.push result")
+
-- Execute some command
client:write("1\n")
test:is(yaml.decode(client:read(EOL))[1], 1, "eval")
diff --git a/test/box/net.box.result b/test/box/net.box.result
index 46d85b327..ecf86af0d 100644
--- a/test/box/net.box.result
+++ b/test/box/net.box.result
@@ -28,7 +28,7 @@ function x_select(cn, space_id, index_id, iterator, offset,
limit, key, opts)
return cn:_request('select', opts, space_id, index_id, iterator,
offset, limit, key)
end
-function x_fatal(cn) cn._transport.perform_request(nil, nil, 'inject', nil,
'\x80') end
+function x_fatal(cn) cn._transport.perform_request(nil, nil, 'inject', nil,
nil, '\x80') end
test_run:cmd("setopt delimiter ''");
---
...
diff --git a/test/box/net.box.test.lua b/test/box/net.box.test.lua
index 87e26f84c..2c19ee479 100644
--- a/test/box/net.box.test.lua
+++ b/test/box/net.box.test.lua
@@ -11,7 +11,7 @@ function x_select(cn, space_id, index_id, iterator, offset,
limit, key, opts)
return cn:_request('select', opts, space_id, index_id, iterator,
offset, limit, key)
end
-function x_fatal(cn) cn._transport.perform_request(nil, nil, 'inject', nil,
'\x80') end
+function x_fatal(cn) cn._transport.perform_request(nil, nil, 'inject', nil,
nil, '\x80') end
test_run:cmd("setopt delimiter ''");
LISTEN = require('uri').parse(box.cfg.listen)
diff --git a/test/box/push.result b/test/box/push.result
new file mode 100644
index 000000000..747503f59
--- /dev/null
+++ b/test/box/push.result
@@ -0,0 +1,364 @@
+--
+-- gh-2677: box.session.push.
+--
+--
+-- Usage.
+--
+box.session.push()
+---
+- error: 'Usage: box.session.push(data, opts)'
+...
+box.session.push(100)
+---
+- error: 'Usage: box.session.push(data, opts)'
+...
+box.session.push(100, {sync = -200})
+---
+- error: 'Usage: box.session.push(data, opts)'
+...
+--
+-- Test text protocol.
+--
+test_run = require('test_run').new()
+---
+...
+console = require('console')
+---
+...
+netbox = require('net.box')
+---
+...
+fiber = require('fiber')
+---
+...
+s = console.listen(3434)
+---
+...
+c = netbox.connect(3434, {console = true})
+---
+...
+c:eval('100')
+---
+- '---
+
+ - 100
+
+ ...
+
+'
+...
+messages = {}
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+function on_push(message)
+ table.insert(messages, message)
+end;
+---
+...
+function do_pushes()
+ local sync = box.session.sync()
+ for i = 1, 5 do
+ box.session.push(i, {sync = sync})
+ fiber.sleep(0.01)
+ end
+ return 300
+end;
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+-- Ensure a pushed message can ignored on a client.
+c:eval('do_pushes()')
+---
+- '---
+
+ - 300
+
+ ...
+
+'
+...
+-- Now start catching pushes.
+c:eval('do_pushes()', {on_push = on_push})
+---
+- '---
+
+ - 300
+
+ ...
+
+'
+...
+messages
+---
+- - '---
+
+ - 1
+
+ ...
+
+'
+ - '---
+
+ - 2
+
+ ...
+
+'
+ - '---
+
+ - 3
+
+ ...
+
+'
+ - '---
+
+ - 4
+
+ ...
+
+'
+ - '---
+
+ - 5
+
+ ...
+
+'
+...
+c:close()
+---
+...
+s:close()
+---
+- true
+...
+--
+-- Test binary protocol.
+--
+box.schema.user.grant('guest','read,write,execute','universe')
+---
+...
+c = netbox.connect(box.cfg.listen)
+---
+...
+c:ping()
+---
+- true
+...
+messages = {}
+---
+...
+c:call('do_pushes', {}, {on_push = on_push})
+---
+- 300
+...
+messages
+---
+- - 1
+ - 2
+ - 3
+ - 4
+ - 5
+...
+-- Add a little stress: many pushes with different syncs, from
+-- different fibers and DML/DQL requests.
+catchers = {}
+---
+...
+started = 0
+---
+...
+finished = 0
+---
+...
+s = box.schema.create_space('test', {format = {{'field1', 'integer'}}})
+---
+...
+pk = s:create_index('pk')
+---
+...
+c:reload_schema()
+---
+...
+test_run:cmd("setopt delimiter ';'")
+---
+- true
+...
+function dml_push_and_dml(key)
+ local sync = box.session.sync()
+ box.session.push('started dml', {sync = sync})
+ s:replace{key}
+ box.session.push('continued dml', {sync = sync})
+ s:replace{-key}
+ box.session.push('finished dml', {sync = sync})
+ return key
+end;
+---
+...
+function do_pushes(val)
+ local sync = box.session.sync()
+ for i = 1, 5 do
+ box.session.push(i, {sync = sync})
+ fiber.yield()
+ end
+ return val
+end;
+---
+...
+function push_catcher_f()
+ fiber.yield()
+ started = started + 1
+ local catcher = {messages = {}, retval = nil, is_dml = false}
+ catcher.retval = c:call('do_pushes', {started}, {on_push =
function(message)
+ table.insert(catcher.messages, message)
+ end})
+ table.insert(catchers, catcher)
+ finished = finished + 1
+end;
+---
+...
+function dml_push_and_dml_f()
+ fiber.yield()
+ started = started + 1
+ local catcher = {messages = {}, retval = nil, is_dml = true}
+ catcher.retval = c:call('dml_push_and_dml', {started}, {on_push =
function(message)
+ table.insert(catcher.messages, message)
+ end})
+ table.insert(catchers, catcher)
+ finished = finished + 1
+end;
+---
+...
+-- At first check that a pushed message can be ignored in a binary
+-- protocol too.
+c:call('do_pushes', {300});
+---
+- 300
+...
+-- Then do stress.
+for i = 1, 200 do
+ fiber.create(dml_push_and_dml_f)
+ fiber.create(push_catcher_f)
+end;
+---
+...
+while finished ~= 400 do fiber.sleep(0.1) end;
+---
+...
+for _, c in pairs(catchers) do
+ if c.is_dml then
+ assert(#c.messages == 3)
+ assert(c.messages[1] == 'started dml')
+ assert(c.messages[2] == 'continued dml')
+ assert(c.messages[3] == 'finished dml')
+ assert(s:get{c.retval})
+ assert(s:get{-c.retval})
+ else
+ assert(c.retval)
+ assert(#c.messages == 5)
+ for k, v in pairs(c.messages) do
+ assert(k == v)
+ end
+ end
+end;
+---
+...
+test_run:cmd("setopt delimiter ''");
+---
+- true
+...
+#s:select{}
+---
+- 400
+...
+--
+-- Test binary pushes.
+--
+ibuf = require('buffer').ibuf()
+---
+...
+msgpack = require('msgpack')
+---
+...
+messages = {}
+---
+...
+resp_len = c:call('do_pushes', {300}, {on_push = on_push, buffer = ibuf})
+---
+...
+resp_len
+---
+- 10
+...
+messages
+---
+- - 4
+ - 4
+ - 4
+ - 4
+ - 4
+...
+decoded = {}
+---
+...
+r = nil
+---
+...
+for i = 1, #messages do r, ibuf.rpos = msgpack.decode_unchecked(ibuf.rpos)
table.insert(decoded, r) end
+---
+...
+decoded
+---
+- - {50: [1]}
+ - {50: [2]}
+ - {50: [3]}
+ - {50: [4]}
+ - {50: [5]}
+...
+r, _ = msgpack.decode_unchecked(ibuf.rpos)
+---
+...
+r
+---
+- {48: [300]}
+...
+c:close()
+---
+...
+s:drop()
+---
+...
+box.schema.user.revoke('guest', 'read,write,execute', 'universe')
+---
+...
+--
+-- Ensure can not push in background.
+--
+ok = nil
+---
+...
+err = nil
+---
+...
+function back_push_f() ok, err = pcall(box.session.push, 100, {sync = 100}) end
+---
+...
+f = fiber.create(back_push_f)
+---
+...
+while f:status() ~= 'dead' do fiber.sleep(0.01) end
+---
+...
+ok, err
+---
+- false
+- Session 'background' does not support push()
+...
diff --git a/test/box/push.test.lua b/test/box/push.test.lua
new file mode 100644
index 000000000..02d571876
--- /dev/null
+++ b/test/box/push.test.lua
@@ -0,0 +1,163 @@
+--
+-- gh-2677: box.session.push.
+--
+
+--
+-- Usage.
+--
+box.session.push()
+box.session.push(100)
+box.session.push(100, {sync = -200})
+
+--
+-- Test text protocol.
+--
+test_run = require('test_run').new()
+console = require('console')
+netbox = require('net.box')
+fiber = require('fiber')
+s = console.listen(3434)
+c = netbox.connect(3434, {console = true})
+c:eval('100')
+messages = {}
+test_run:cmd("setopt delimiter ';'")
+function on_push(message)
+ table.insert(messages, message)
+end;
+
+function do_pushes()
+ local sync = box.session.sync()
+ for i = 1, 5 do
+ box.session.push(i, {sync = sync})
+ fiber.sleep(0.01)
+ end
+ return 300
+end;
+test_run:cmd("setopt delimiter ''");
+-- Ensure a pushed message can ignored on a client.
+c:eval('do_pushes()')
+-- Now start catching pushes.
+c:eval('do_pushes()', {on_push = on_push})
+messages
+
+c:close()
+s:close()
+
+--
+-- Test binary protocol.
+--
+box.schema.user.grant('guest','read,write,execute','universe')
+
+c = netbox.connect(box.cfg.listen)
+c:ping()
+messages = {}
+c:call('do_pushes', {}, {on_push = on_push})
+messages
+
+-- Add a little stress: many pushes with different syncs, from
+-- different fibers and DML/DQL requests.
+
+catchers = {}
+started = 0
+finished = 0
+s = box.schema.create_space('test', {format = {{'field1', 'integer'}}})
+pk = s:create_index('pk')
+c:reload_schema()
+test_run:cmd("setopt delimiter ';'")
+function dml_push_and_dml(key)
+ local sync = box.session.sync()
+ box.session.push('started dml', {sync = sync})
+ s:replace{key}
+ box.session.push('continued dml', {sync = sync})
+ s:replace{-key}
+ box.session.push('finished dml', {sync = sync})
+ return key
+end;
+function do_pushes(val)
+ local sync = box.session.sync()
+ for i = 1, 5 do
+ box.session.push(i, {sync = sync})
+ fiber.yield()
+ end
+ return val
+end;
+function push_catcher_f()
+ fiber.yield()
+ started = started + 1
+ local catcher = {messages = {}, retval = nil, is_dml = false}
+ catcher.retval = c:call('do_pushes', {started}, {on_push =
function(message)
+ table.insert(catcher.messages, message)
+ end})
+ table.insert(catchers, catcher)
+ finished = finished + 1
+end;
+function dml_push_and_dml_f()
+ fiber.yield()
+ started = started + 1
+ local catcher = {messages = {}, retval = nil, is_dml = true}
+ catcher.retval = c:call('dml_push_and_dml', {started}, {on_push =
function(message)
+ table.insert(catcher.messages, message)
+ end})
+ table.insert(catchers, catcher)
+ finished = finished + 1
+end;
+-- At first check that a pushed message can be ignored in a binary
+-- protocol too.
+c:call('do_pushes', {300});
+-- Then do stress.
+for i = 1, 200 do
+ fiber.create(dml_push_and_dml_f)
+ fiber.create(push_catcher_f)
+end;
+while finished ~= 400 do fiber.sleep(0.1) end;
+
+for _, c in pairs(catchers) do
+ if c.is_dml then
+ assert(#c.messages == 3)
+ assert(c.messages[1] == 'started dml')
+ assert(c.messages[2] == 'continued dml')
+ assert(c.messages[3] == 'finished dml')
+ assert(s:get{c.retval})
+ assert(s:get{-c.retval})
+ else
+ assert(c.retval)
+ assert(#c.messages == 5)
+ for k, v in pairs(c.messages) do
+ assert(k == v)
+ end
+ end
+end;
+test_run:cmd("setopt delimiter ''");
+
+#s:select{}
+
+--
+-- Test binary pushes.
+--
+ibuf = require('buffer').ibuf()
+msgpack = require('msgpack')
+messages = {}
+resp_len = c:call('do_pushes', {300}, {on_push = on_push, buffer = ibuf})
+resp_len
+messages
+decoded = {}
+r = nil
+for i = 1, #messages do r, ibuf.rpos = msgpack.decode_unchecked(ibuf.rpos)
table.insert(decoded, r) end
+decoded
+r, _ = msgpack.decode_unchecked(ibuf.rpos)
+r
+
+c:close()
+s:drop()
+
+box.schema.user.revoke('guest', 'read,write,execute', 'universe')
+
+--
+-- Ensure can not push in background.
+--
+ok = nil
+err = nil
+function back_push_f() ok, err = pcall(box.session.push, 100, {sync = 100}) end
+f = fiber.create(back_push_f)
+while f:status() ~= 'dead' do fiber.sleep(0.01) end
+ok, err
diff --git a/test/replication/before_replace.result
b/test/replication/before_replace.result
index d561b4813..7b0f2993f 100644
--- a/test/replication/before_replace.result
+++ b/test/replication/before_replace.result
@@ -49,7 +49,17 @@ test_run:cmd("switch autobootstrap3");
---
- true
...
+--
+-- gh-2677 - test that an applier can not push() messages. Applier
+-- session is available in Lua, so the test is here instead of
+-- box/push.test.lua.
+--
+push_ok = nil
+push_err = nil
_ = box.space.test:before_replace(function(old, new)
+ if box.session.type() == 'applier' and not push_err then
+ push_ok, push_err = pcall(box.session.push, 100, {sync = 100})
+ end
if old ~= nil and new ~= nil then
return new[2] > old[2] and new or old
end
@@ -187,6 +197,10 @@ box.space.test:select()
- [9, 90]
- [10, 100]
...
+push_err
+---
+- Session 'applier' does not support push()
+...
test_run:cmd('restart server autobootstrap3')
box.space.test:select()
---
diff --git a/test/replication/before_replace.test.lua
b/test/replication/before_replace.test.lua
index 2c6912d06..04e5a2172 100644
--- a/test/replication/before_replace.test.lua
+++ b/test/replication/before_replace.test.lua
@@ -26,7 +26,17 @@ _ = box.space.test:before_replace(function(old, new)
end
end);
test_run:cmd("switch autobootstrap3");
+--
+-- gh-2677 - test that an applier can not push() messages. Applier
+-- session is available in Lua, so the test is here instead of
+-- box/push.test.lua.
+--
+push_ok = nil
+push_err = nil
_ = box.space.test:before_replace(function(old, new)
+ if box.session.type() == 'applier' and not push_err then
+ push_ok, push_err = pcall(box.session.push, 100, {sync = 100})
+ end
if old ~= nil and new ~= nil then
return new[2] > old[2] and new or old
end
@@ -62,6 +72,7 @@ test_run:cmd('restart server autobootstrap2')
box.space.test:select()
test_run:cmd("switch autobootstrap3")
box.space.test:select()
+push_err
test_run:cmd('restart server autobootstrap3')
box.space.test:select()
--
2.14.3 (Apple Git-98)
