From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from smtp56.i.mail.ru (smtp56.i.mail.ru [217.69.128.36]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (No client certificate requested) by dev.tarantool.org (Postfix) with ESMTPS id C74E746970E for ; Thu, 23 Jan 2020 02:05:05 +0300 (MSK) References: <1065f691482e681b779abd3698ec4489267c11fd.1579558507.git.v.shpilevoy@tarantool.org> <20200121002441.sysri34odzjrpjzu@tkn_work_nb> <20200122022537.tygaowsn5remjee4@tkn_work_nb> From: Vladislav Shpilevoy Message-ID: <5925f9e8-e148-9d68-eb39-db87e9e631eb@tarantool.org> Date: Thu, 23 Jan 2020 00:05:02 +0100 MIME-Version: 1.0 In-Reply-To: <20200122022537.tygaowsn5remjee4@tkn_work_nb> Content-Type: text/plain; charset=utf-8 Content-Language: en-US Content-Transfer-Encoding: 7bit Subject: Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API List-Id: Tarantool development patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: Alexander Turenko Cc: tarantool-patches@dev.tarantool.org On 22/01/2020 03:25, Alexander Turenko wrote: >>> I see that box_return_tuple() receives (box_function_ctx_t *) as the >>> first argument and wonder why box_session_push() does not. If we able to >>> determine where to and how to send a push w/o the context, then the >>> context is not necessary in box_return_tuple() too? I don't very >>> familiar with this part of codebase, so it is just question for now. >>> I'll look around soon if time permits. >> >> Context of session push is a session. It is available in fiber, which >> is a global variable. >> >> Context of a function is its port - a storage for returned values. Port >> is not available in the fiber, and therefore is not stored anywhere >> globally. It needs to be passed explicitly. We pass it as an opaque >> pointer box_function_ctx_t*. >> >> We can add a context to the push, but then we need to create a public >> version of struct session. Something like 'box_session_t', which would >> be an alias for 'struct session'. And have a function like >> 'box_session_current()', which in turn won't have a context. >> >> I think we need to introduce box_session_t only if we are going to >> allow users to access foreign sessions. Otherwise it is always the >> user's own session and is available in the fiber anyway. > > But box_return_tuple() can create a port on demand, just like you do in > box_session_push(), right? I don't see any difference between those > functions in context of the question. > > If box_return_tuple() could be implemented with or without explicit > (box_function_ctx_t *) argument in the past, then someone made the > decision to implement it with the argument and there is (hopefully) some > reasoning under this decision. If so, it may be applicable to > box_session_push(). Or there is a reason why it is not applicable here. Well, I recommend you to look at the code and think about it. You can't create a port on demand in box_return_tuple(). Because where would you put it then? box_return_tuple()'s context is a link to the stack frame called the user's function, where the port is stored. You can't omit it in box_return_tuple(). There is just no way. By the same reason SQL functions have sql_context and pass it to sql_result_*(). And this is why box_function_ctx_t* is an opaque pointer, which can't be created by a user. It is created by us for the user. The only way to avoid passing a context is to always return a result using 'return' statement. In that case you return directly to the caller. But it does not allow to do multireturn, and to do something after return. For these cases you need a context. And this is our case with box_return_tuple(). In box_session_push() you don't need an upper stack frame or whatever else what is not available in the fiber. And you don't have any special caller's context. Push() does not return to anywhere. It pushes out of bound. I don't know how to explain this else. >> + /* >> + * Create a new thread and pop it immediately from the >> + * main stack. So as it would not stay there in case >> + * something would throw in this function. >> + */ >> + int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX); >> + /* >> + * MessagePack -> Lua -> YAML. The middle is not really >> + * needed here, but there is no MessagePack -> YAML >> + * encoder yet. >> + */ >> + const char *data = port->data; >> + luamp_decode(L, luaL_msgpack_default, &data); >> + int rc = lua_yaml_encode(L, luaL_yaml_default, "!push!", >> + "tag:tarantool.io/push,2018"); >> ================================================================================ >> >> luamp_decode and lua_yaml_encode can throw a Lua error. Don't know how to >> fix that except introduction of a MessagePack -> YAML converter, which seems >> a pretty big and independent task. >> >> ================================================================================ > > It seems that it is usually handled with lua_pushcfunction() / > lua_pushcclosure() + lua_pcall(). Thanks for the tip. Here is the diff: ================================================================================ diff --git a/src/box/lua/console.c b/src/box/lua/console.c index fbf1a76cc..51f08404f 100644 --- a/src/box/lua/console.c +++ b/src/box/lua/console.c @@ -411,6 +411,20 @@ port_lua_dump_plain(struct port *port, uint32_t *size) return result; } +/** + * A helper for port_msgpack_dump_plain() to safely work with Lua, + * being called via pcall(). + */ +static int +lua_port_msgpack_dump_plain(struct lua_State *L) +{ + const char *data = lua_touserdata(L, 1); + lua_pop(L, 1); + luamp_decode(L, luaL_msgpack_default, &data); + return lua_yaml_encode(L, luaL_yaml_default, "!push!", + "tag:tarantool.io/push,2018"); +} + /** The same as port_lua plain dump, but from raw MessagePack. */ const char * port_msgpack_dump_plain(struct port *base, uint32_t *size) @@ -430,20 +444,22 @@ port_msgpack_dump_plain(struct port *base, uint32_t *size) * needed here, but there is no MessagePack -> YAML * encoder yet. */ - const char *data = port->data; - luamp_decode(L, luaL_msgpack_default, &data); - int rc = lua_yaml_encode(L, luaL_yaml_default, "!push!", - "tag:tarantool.io/push,2018"); - if (rc == 2) { + int top = lua_gettop(L); + lua_pushcfunction(L, lua_port_msgpack_dump_plain); + lua_pushlightuserdata(L, (void *) port->data); + if (lua_pcall(L, 1, LUA_MULTRET, 0) != 0 || + lua_gettop(L) - top == 2) { /* - * Nil and error object are pushed onto the stack. + * Nil and error string are pushed onto the stack + * if that was a YAML error. Only error string is + * pushed in case it was a Lua error. In both + * cases top of the stack is a string. */ - assert(lua_isnil(L, -2)); assert(lua_isstring(L, -1)); diag_set(ClientError, ER_PROC_LUA, lua_tostring(L, -1)); goto error; } - assert(rc == 1); + assert(lua_gettop(L) - top == 1); assert(lua_isstring(L, -1)); size_t len; const char *result = lua_tolstring(L, -1, &len); ================================================================================ Full patch: ================================================================================ commit 34c2d8033bcfabcab95a9aeec2ca029169969133 Author: Vladislav Shpilevoy Date: Mon Jan 20 23:08:58 2020 +0100 box: export box_session_push to the public C API API is different from box.session.push() - sync argument was removed. It will disappear from Lua API as well, according to a part of public API, and because it just is not needed here. Session is omitted as well. Indeed, a user can't push to a foreign session, and the current session can be obtained inside box_session_push(). And anyway session is not in the public C API. Internally dump into iproto is done using obuf_dup(), just like tuple_to_obuf() does. obuf_alloc() would be a bad call here, because it wouldn't be able to split the pushed data into several obuf chunks, and would cause obuf fragmentation. Dump into plain text behaves just like a Lua push - it returns a YAML formatted text. But to turn MessagePack into YAML an intermediate Lua representation is used, because there is no a MessagePack -> YAML translator yet. Closes #4734 @TarantoolBot document Title: box_session_push() C API There is a new function in the public C API: int box_session_push(const char *data, const char *data_end); It takes raw MessagePack, and behaves just like Lua box.session.push(). diff --git a/extra/exports b/extra/exports index 7b84a1452..bc3759244 100644 --- a/extra/exports +++ b/extra/exports @@ -216,6 +216,7 @@ box_truncate box_sequence_next box_sequence_set box_sequence_reset +box_session_push box_index_iterator box_iterator_next box_iterator_free diff --git a/src/box/box.cc b/src/box/box.cc index 1b2b27d61..a32c8cdbd 100644 --- a/src/box/box.cc +++ b/src/box/box.cc @@ -1433,6 +1433,20 @@ box_sequence_reset(uint32_t seq_id) return sequence_data_delete(seq_id); } +int +box_session_push(const char *data, const char *data_end) +{ + struct session *session = current_session(); + if (session == NULL) + return -1; + struct port_msgpack port; + struct port *base = (struct port *) &port; + port_msgpack_create(base, data, data_end - data); + int rc = session_push(session, session_sync(session), base); + port_msgpack_destroy(base); + return rc; +} + static inline void box_register_replica(uint32_t id, const struct tt_uuid *uuid) { diff --git a/src/box/box.h b/src/box/box.h index a212e6510..93773cedd 100644 --- a/src/box/box.h +++ b/src/box/box.h @@ -442,6 +442,20 @@ box_sequence_set(uint32_t seq_id, int64_t value); API_EXPORT int box_sequence_reset(uint32_t seq_id); +/** + * Push MessagePack data into a session data channel - socket, + * console or whatever is behind the session. Note, that + * successful push does not guarantee delivery in case it was sent + * into the network. Just like with write()/send() system calls. + * + * \param data begin of MessagePack to push + * \param data_end end of MessagePack to push + * \retval -1 on error (check box_error_last()) + * \retval 0 on success + */ +API_EXPORT int +box_session_push(const char *data, const char *data_end); + /** \endcond public */ /** diff --git a/src/box/call.c b/src/box/call.c index bcaa453ea..00aa3845b 100644 --- a/src/box/call.c +++ b/src/box/call.c @@ -64,17 +64,40 @@ port_msgpack_get_msgpack(struct port *base, uint32_t *size) return port->data; } +static int +port_msgpack_dump_msgpack(struct port *base, struct obuf *out) +{ + struct port_msgpack *port = (struct port_msgpack *) base; + assert(port->vtab == &port_msgpack_vtab); + size_t size = port->data_sz; + if (obuf_dup(out, port->data, size) == size) + return 0; + diag_set(OutOfMemory, size, "obuf_dup", "port->data"); + return -1; +} + extern void port_msgpack_dump_lua(struct port *base, struct lua_State *L, bool is_flat); +extern const char * +port_msgpack_dump_plain(struct port *base, uint32_t *size); + +void +port_msgpack_destroy(struct port *base) +{ + struct port_msgpack *port = (struct port_msgpack *) base; + assert(port->vtab == &port_msgpack_vtab); + free(port->buffer); +} + static const struct port_vtab port_msgpack_vtab = { - .dump_msgpack = NULL, + .dump_msgpack = port_msgpack_dump_msgpack, .dump_msgpack_16 = NULL, .dump_lua = port_msgpack_dump_lua, - .dump_plain = NULL, + .dump_plain = port_msgpack_dump_plain, .get_msgpack = port_msgpack_get_msgpack, .get_vdbemem = NULL, - .destroy = NULL, + .destroy = port_msgpack_destroy, }; int diff --git a/src/box/lua/console.c b/src/box/lua/console.c index 57e7e7f4f..ac353d012 100644 --- a/src/box/lua/console.c +++ b/src/box/lua/console.c @@ -37,6 +37,7 @@ #include "lua/fiber.h" #include "fiber.h" #include "coio.h" +#include "lua/msgpack.h" #include "lua-yaml/lyaml.h" #include #include @@ -410,6 +411,86 @@ port_lua_dump_plain(struct port *port, uint32_t *size) return result; } +/** + * A helper for port_msgpack_dump_plain() to safely work with Lua, + * being called via pcall(). + */ +static int +lua_port_msgpack_dump_plain(struct lua_State *L) +{ + const char *data = lua_touserdata(L, 1); + lua_pop(L, 1); + luamp_decode(L, luaL_msgpack_default, &data); + return lua_yaml_encode(L, luaL_yaml_default, "!push!", + "tag:tarantool.io/push,2018"); +} + +/** The same as port_lua plain dump, but from raw MessagePack. */ +const char * +port_msgpack_dump_plain(struct port *base, uint32_t *size) +{ + struct port_msgpack *port = (struct port_msgpack *) base; + struct lua_State *L = luaT_newthread(tarantool_L); + if (L == NULL) + return NULL; + /* + * Create a new thread and pop it immediately from the + * main stack. So as it would not stay there in case + * something would throw in this function. + */ + int coro_ref = luaL_ref(tarantool_L, LUA_REGISTRYINDEX); + /* + * MessagePack -> Lua -> YAML. The middle is not really + * needed here, but there is no MessagePack -> YAML + * encoder yet. + */ + int top = lua_gettop(L); + lua_pushcfunction(L, lua_port_msgpack_dump_plain); + lua_pushlightuserdata(L, (void *) port->data); + if (lua_pcall(L, 1, LUA_MULTRET, 0) != 0 || + lua_gettop(L) - top == 2) { + /* + * Nil and error string are pushed onto the stack + * if that was a YAML error. Only error string is + * pushed in case it was a Lua error. In both + * cases top of the stack is a string. + */ + assert(lua_isstring(L, -1)); + diag_set(ClientError, ER_PROC_LUA, lua_tostring(L, -1)); + goto error; + } + assert(lua_gettop(L) - top == 1); + assert(lua_isstring(L, -1)); + size_t len; + const char *result = lua_tolstring(L, -1, &len); + *size = (uint32_t) len; + /* + * The result string should be copied, because the stack, + * keeping the string, is going to be destroyed in the + * next lines. + * Can't use region, because somebody should free it, and + * its purpose is to free many objects at once. In this + * case only one string should be allocated and freed + * right after usage. Heap is fine for that. + * Can't use the global tarantool_lua_ibuf, because the + * plain dumper is used by session push, which yields in + * coio. And during that yield the ibuf can be reset by + * another fiber. + */ + char *buffer = (char *) strdup(result); + if (buffer == NULL) { + diag_set(OutOfMemory, len + 1, "strdup", "buffer"); + goto error; + } + assert(port->buffer == NULL); + port->buffer = buffer; + luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref); + return buffer; +error: + luaL_unref(tarantool_L, LUA_REGISTRYINDEX, coro_ref); + return NULL; +} + /** * Push a tagged YAML document into a console socket. * @param session Console session. diff --git a/src/box/port.h b/src/box/port.h index 9d3d02b3c..8a1650424 100644 --- a/src/box/port.h +++ b/src/box/port.h @@ -86,6 +86,12 @@ struct port_msgpack { const struct port_vtab *vtab; const char *data; uint32_t data_sz; + /** + * Buffer for dump_*() functions. In particular, it is + * used by the plain dumper to save a result string and + * free it when the port is destroyed. + */ + char *buffer; }; static_assert(sizeof(struct port_msgpack) <= sizeof(struct port), @@ -95,6 +101,10 @@ static_assert(sizeof(struct port_msgpack) <= sizeof(struct port), void port_msgpack_create(struct port *port, const char *data, uint32_t data_sz); +/** Destroy a MessagePack port. */ +void +port_msgpack_destroy(struct port *base); + /** Port for storing the result of a Lua CALL/EVAL. */ struct port_lua { const struct port_vtab *vtab; diff --git a/test/box/function1.c b/test/box/function1.c index 87062d6a8..b2ce752a9 100644 --- a/test/box/function1.c +++ b/test/box/function1.c @@ -245,3 +245,10 @@ test_sleep(box_function_ctx_t *ctx, const char *args, const char *args_end) fiber_sleep(0); return 0; } + +int +test_push(box_function_ctx_t *ctx, const char *args, const char *args_end) +{ + (void) ctx; + return box_session_push(args, args_end); +} diff --git a/test/box/push.result b/test/box/push.result index aebcb7501..6d85a484b 100644 --- a/test/box/push.result +++ b/test/box/push.result @@ -563,3 +563,95 @@ box.schema.func.drop('do_long_and_push') box.session.on_disconnect(nil, on_disconnect) --- ... +-- +-- gh-4734: C API for session push. +-- +build_path = os.getenv("BUILDDIR") +--- +... +old_cpath = package.cpath +--- +... +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..old_cpath +--- +... +box.schema.func.create('function1.test_push', {language = 'C'}) +--- +... +box.schema.user.grant('guest', 'super') +--- +... +c = netbox.connect(box.cfg.listen) +--- +... +messages = {} +--- +... +c:call('function1.test_push', \ + {1, 2, 3}, \ + {on_push = table.insert, \ + on_push_ctx = messages}) +--- +- [] +... +messages +--- +- - [1, 2, 3] +... +c:close() +--- +... +-- +-- C can push to the console. +-- +console = require('console') +--- +... +fio = require('fio') +--- +... +socket = require('socket') +--- +... +sock_path = fio.pathjoin(fio.cwd(), 'console.sock') +--- +... +_ = fio.unlink(sock_path) +--- +... +server = console.listen(sock_path) +--- +... +client = socket.tcp_connect('unix/', sock_path) +--- +... +_ = client:read({chunk = 128}) +--- +... +_ = client:write("box.func['function1.test_push']:call({1, 2, 3})\n") +--- +... +client:read("\n...\n") +--- +- '%TAG !push! tag:tarantool.io/push,2018 + + --- [1, 2, 3] + + ... + + ' +... +client:close() +--- +- true +... +server:close() +--- +- true +... +box.schema.user.revoke('guest', 'super') +--- +... +box.schema.func.drop('function1.test_push') +--- +... diff --git a/test/box/push.test.lua b/test/box/push.test.lua index 7ae6f4a86..d47422cef 100644 --- a/test/box/push.test.lua +++ b/test/box/push.test.lua @@ -264,3 +264,39 @@ chan_push:put(true) chan_push:get() box.schema.func.drop('do_long_and_push') box.session.on_disconnect(nil, on_disconnect) + +-- +-- gh-4734: C API for session push. +-- +build_path = os.getenv("BUILDDIR") +old_cpath = package.cpath +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..old_cpath + +box.schema.func.create('function1.test_push', {language = 'C'}) +box.schema.user.grant('guest', 'super') +c = netbox.connect(box.cfg.listen) +messages = {} +c:call('function1.test_push', \ + {1, 2, 3}, \ + {on_push = table.insert, \ + on_push_ctx = messages}) +messages +c:close() +-- +-- C can push to the console. +-- +console = require('console') +fio = require('fio') +socket = require('socket') +sock_path = fio.pathjoin(fio.cwd(), 'console.sock') +_ = fio.unlink(sock_path) +server = console.listen(sock_path) +client = socket.tcp_connect('unix/', sock_path) +_ = client:read({chunk = 128}) +_ = client:write("box.func['function1.test_push']:call({1, 2, 3})\n") +client:read("\n...\n") +client:close() +server:close() + +box.schema.user.revoke('guest', 'super') +box.schema.func.drop('function1.test_push')