* [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API @ 2020-01-20 22:16 Vladislav Shpilevoy 2020-01-20 23:14 ` Vladislav Shpilevoy ` (3 more replies) 0 siblings, 4 replies; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-01-20 22:16 UTC (permalink / raw) To: tarantool-patches, kostja.osipov, alexander.turenko It was a customer request. Nothing special here, the patch is trivial, with a couple of short notes: 1) Sync argument was removed. It will disappear from Lua API as well, according to #4689. Port is replaced with raw MessagePack, because port is not a part of public API. 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. 2) Dump 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. Closes #4734 --- Issue: https://github.com/tarantool/tarantool/issues/4734 Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-4734-export-box-session-push extra/exports | 1 + src/box/box.cc | 12 +++++++++++ src/box/box.h | 14 +++++++++++++ src/box/call.c | 14 ++++++++++++- test/box/function1.c | 7 +++++++ test/box/function1.result | 41 +++++++++++++++++++++++++++++++++++++ test/box/function1.test.lua | 19 +++++++++++++++++ 7 files changed, 107 insertions(+), 1 deletion(-) 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..9935c164c 100644 --- a/src/box/box.cc +++ b/src/box/box.cc @@ -1433,6 +1433,18 @@ 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); + return session_push(session, session_sync(session), base); +} + 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..336044a3b 100644 --- a/src/box/call.c +++ b/src/box/call.c @@ -64,11 +64,23 @@ 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); 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, 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/function1.result b/test/box/function1.result index b91d63c51..a57de403a 100644 --- a/test/box/function1.result +++ b/test/box/function1.result @@ -1065,3 +1065,44 @@ box.func['test'].is_multikey == true box.func['test']:drop() --- ... +-- +-- gh-4734: C API for session push. +-- +build_path = os.getenv("BUILDDIR") +--- +... +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath +--- +... +box.schema.func.create('function1.test_push', {language = 'C'}) +--- +... +box.schema.user.grant('guest', 'super') +--- +... +c = net.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() +--- +... +box.schema.user.revoke('guest', 'super') +--- +... +box.schema.func.drop('function1.test_push') +--- +... diff --git a/test/box/function1.test.lua b/test/box/function1.test.lua index b1841f3ad..4f2c1ea2d 100644 --- a/test/box/function1.test.lua +++ b/test/box/function1.test.lua @@ -380,3 +380,22 @@ box.schema.func.drop("SUM") box.schema.func.create('test', {body = "function(tuple) return tuple end", is_deterministic = true, opts = {is_multikey = true}}) box.func['test'].is_multikey == true box.func['test']:drop() + +-- +-- gh-4734: C API for session push. +-- +build_path = os.getenv("BUILDDIR") +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath + +box.schema.func.create('function1.test_push', {language = 'C'}) +box.schema.user.grant('guest', 'super') +c = net.connect(box.cfg.listen) +messages = {} +c:call('function1.test_push', \ + {1, 2, 3}, \ + {on_push = table.insert, \ + on_push_ctx = messages}) +messages +c:close() +box.schema.user.revoke('guest', 'super') +box.schema.func.drop('function1.test_push') -- 2.21.1 (Apple Git-122.3) ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-20 22:16 [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API Vladislav Shpilevoy @ 2020-01-20 23:14 ` Vladislav Shpilevoy 2020-01-21 0:24 ` Alexander Turenko ` (2 subsequent siblings) 3 siblings, 0 replies; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-01-20 23:14 UTC (permalink / raw) To: tarantool-patches, kostja.osipov, alexander.turenko Alexander noticed that the patch crashes when C push works from console. I am going to fix this by dumping pushed data in the same way as with Lua pushes - decode the data into YAML and return as a string. If nobody has something against this. Talking of YAML vs Lua output format - I am not going to fix this in scope of this issue. On 20/01/2020 23:16, Vladislav Shpilevoy wrote: > It was a customer request. Nothing special here, the patch is > trivial, with a couple of short notes: > > 1) Sync argument was removed. It will disappear from Lua API as > well, according to #4689. Port is replaced with raw MessagePack, > because port is not a part of public API. 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. > > 2) Dump 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. > > Closes #4734 > --- > Issue: https://github.com/tarantool/tarantool/issues/4734 > Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-4734-export-box-session-push > ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-20 22:16 [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API Vladislav Shpilevoy 2020-01-20 23:14 ` Vladislav Shpilevoy @ 2020-01-21 0:24 ` Alexander Turenko 2020-01-22 0:06 ` Vladislav Shpilevoy 2020-03-30 20:42 ` Vladislav Shpilevoy 2020-03-30 20:42 ` Vladislav Shpilevoy 3 siblings, 1 reply; 13+ messages in thread From: Alexander Turenko @ 2020-01-21 0:24 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches Thanks for the patch! I run the following code: | cd test/box | ../../src/tarantool | tarantool> box.cfg{} | tarantool> box.schema.func.create('function1.test_push', {language = 'C'}) | tarantool> box.func['function1.test_push']:call({1,2,3,4,5,6}) And got segmentation fault: #0 0x55fd20b2b789 in print_backtrace+9 #1 0x55fd209dd2c2 in _ZL12sig_fatal_cbiP9siginfo_tPv+1e7 #2 0x7efce59603a0 in funlockfile+80 #3 (nil) in +80 #4 0x55fd20ae6e54 in console_session_push+79 #5 0x55fd20aac23a in session_push+34 #6 0x55fd20ab1bdb in box_session_push+6e #7 0x7efce7d941cf in test_push+27 #8 0x55fd20a84bdb in func_c_call+181 #9 0x55fd20a85011 in func_call+e9 #10 0x55fd20ae4c60 in lbox_func_call+20b #11 0x55fd20b5b45b in lj_BC_FUNCC+34 #12 0x55fd20b7e23f in lua_pcall+18b #13 0x55fd20b0de3a in luaT_call+29 #14 0x55fd20b04be2 in lua_main+b9 #15 0x55fd20b05214 in run_script_f+619 #16 0x55fd209dcbc9 in _ZL16fiber_cxx_invokePFiP13__va_list_tagES0_+1e #17 0x55fd20b27bbf in fiber_loop+82 #18 0x55fd20d352ef in coro_init+4c I would expect the same output as from `box.session.push(<...>)` call in repl: a yaml document with a tag. 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. We also discussed with Vlad whether box_session_push() should validate passed msgpack. I think that it is usual for a C API (in general) to don't perform full scan of data from a user provided function (when it is C function, not something from, say, socket). So in my understanding passing of incorrect msgpack should be considered as the API misusage and should not be checked. A client may determine where the incorrect iproto packet ends, handle incorrect msgpack somehow and continue reading from the connection. Or expect that this situation will not occur due to performance reasons (which may be acceptable, say, when both client and server procedures are developed in tight). In this case incorrect iproto packet will be sent to a client, but it is possible already with box_return_tuple() in RelWithDebInfo build (Debug build however will fail on mp_tuple_assert(), called from runtime_tuple_new()). I checked it in this way: | const char *data = "\x93\x04\x05\x06\x07"; | box_tuple_format_t *fmt = box_tuple_format_default(); | box_tuple_t *tuple = box_tuple_new(fmt, data, data + 5); | return box_return_tuple(ctx, tuple); Moreover, our C API is used for cases where performance is vital. So extra traversal over msgpack data looks inappropriate here. WBR, Alexander Turenko. On Mon, Jan 20, 2020 at 11:16:05PM +0100, Vladislav Shpilevoy wrote: > It was a customer request. Nothing special here, the patch is > trivial, with a couple of short notes: > > 1) Sync argument was removed. It will disappear from Lua API as > well, according to #4689. Port is replaced with raw MessagePack, > because port is not a part of public API. 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. > > 2) Dump 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. > > Closes #4734 > --- > Issue: https://github.com/tarantool/tarantool/issues/4734 > Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-4734-export-box-session-push > > extra/exports | 1 + > src/box/box.cc | 12 +++++++++++ > src/box/box.h | 14 +++++++++++++ > src/box/call.c | 14 ++++++++++++- > test/box/function1.c | 7 +++++++ > test/box/function1.result | 41 +++++++++++++++++++++++++++++++++++++ > test/box/function1.test.lua | 19 +++++++++++++++++ > 7 files changed, 107 insertions(+), 1 deletion(-) ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-21 0:24 ` Alexander Turenko @ 2020-01-22 0:06 ` Vladislav Shpilevoy 2020-01-22 2:25 ` Alexander Turenko 0 siblings, 1 reply; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-01-22 0:06 UTC (permalink / raw) To: Alexander Turenko; +Cc: tarantool-patches Thanks for the review! > 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. I added YAML formatting. But it looks crutchy. Mainly because we don't have a MessagePack -> YAML converter. I used Lua representation as an intermediate translator. I guess we can comb plain text formatting (YAML format, Lua output format) in scope of https://github.com/tarantool/tarantool/issues/4686. ================================================================================ commit 63cdf88fbb3ed9cf7e77cd04efcee381da64cd53 Author: Vladislav Shpilevoy <v.shpilevoy@tarantool.org> 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..fbf1a76cc 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 <lua.h> #include <lauxlib.h> @@ -410,6 +411,70 @@ port_lua_dump_plain(struct port *port, uint32_t *size) return result; } +/** 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. + */ + 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. ================================================================================ + if (rc == 2) { + /* + * Nil and error object are pushed onto the stack. + */ + 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_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') ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-22 0:06 ` Vladislav Shpilevoy @ 2020-01-22 2:25 ` Alexander Turenko 2020-01-22 23:05 ` Vladislav Shpilevoy 0 siblings, 1 reply; 13+ messages in thread From: Alexander Turenko @ 2020-01-22 2:25 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches > > 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. > + /* > + * 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(). ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-22 2:25 ` Alexander Turenko @ 2020-01-22 23:05 ` Vladislav Shpilevoy 2020-04-06 23:12 ` Igor Munkin 2020-04-06 23:40 ` Alexander Turenko 0 siblings, 2 replies; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-01-22 23:05 UTC (permalink / raw) To: Alexander Turenko; +Cc: tarantool-patches 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 <v.shpilevoy@tarantool.org> 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 <lua.h> #include <lauxlib.h> @@ -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') ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-22 23:05 ` Vladislav Shpilevoy @ 2020-04-06 23:12 ` Igor Munkin 2020-04-07 23:20 ` Vladislav Shpilevoy 2020-04-06 23:40 ` Alexander Turenko 1 sibling, 1 reply; 13+ messages in thread From: Igor Munkin @ 2020-04-06 23:12 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches Vlad, Thanks for the patch! Please consider the comments below. On 23.01.20, Vladislav Shpilevoy wrote: <snipped> > > ================================================================================ > > Full patch: > > ================================================================================ > > commit 34c2d8033bcfabcab95a9aeec2ca029169969133 > Author: Vladislav Shpilevoy <v.shpilevoy@tarantool.org> > 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(). > <snipped> > 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 <snipped> > @@ -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); I see both tag handle and prefix are the same as in port_lua_dump_plain, so IMHO it would be great to introduce corresponding constants for them. Furthermore, these values looks a bit cryptic to me, could you please drop a few words nearby describing them? Can these values be changed in the future? > + 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); It looks like an additional Lua coroutine is excess here and all conversion from MessagePack to Lua can be done using the main one (i.e. tarantool_L). I changed the patch a bit (diff is below) and tests successfully passed on my machine. Did I miss something (maybe a yield underneath luamp_decode / lua_yaml_encode)? ================================================================================ diff --git a/src/box/lua/console.c b/src/box/lua/console.c index ac353d012..d5e0d193b 100644 --- a/src/box/lua/console.c +++ b/src/box/lua/console.c @@ -430,20 +430,12 @@ 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. */ + lua_State *L = tarantool_L; int top = lua_gettop(L); lua_pushcfunction(L, lua_port_msgpack_dump_plain); lua_pushlightuserdata(L, (void *) port->data); @@ -457,7 +449,7 @@ port_msgpack_dump_plain(struct port *base, uint32_t *size) */ assert(lua_isstring(L, -1)); diag_set(ClientError, ER_PROC_LUA, lua_tostring(L, -1)); - goto error; + return NULL; } assert(lua_gettop(L) - top == 1); assert(lua_isstring(L, -1)); @@ -480,15 +472,11 @@ port_msgpack_dump_plain(struct port *base, uint32_t *size) char *buffer = (char *) strdup(result); if (buffer == NULL) { diag_set(OutOfMemory, len + 1, "strdup", "buffer"); - goto error; + return NULL; } 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; } /** ================================================================================ > + 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); Lua strings can contain NUL byte in the middle of the data area and it is a good reason for using strndup here, considering you already obtain the length earlier. Otherwise, it would be nice to drop a corresponding comment about a contract guaranteeing the NUL byte doesn't occur in the middle of the result value. > + if (buffer == NULL) { > + diag_set(OutOfMemory, len + 1, "strdup", "buffer"); > + goto error; > + } > + assert(port->buffer == NULL); > + port->buffer = buffer; It seems to be more convenient to move the lines below to a separate function (e.g port_msgpack_add_buffer). I have one rationale for it: you make a strdup right here, but release the allocated memory in port_msgpack_destroy. Such resource responsibility division seems to be complex for the further maintainence. If you're against my proposal and totally OK with the current approach, feel free to ignore. > + 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. <snipped> -- Best regards, IM ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-04-06 23:12 ` Igor Munkin @ 2020-04-07 23:20 ` Vladislav Shpilevoy 0 siblings, 0 replies; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-04-07 23:20 UTC (permalink / raw) To: Igor Munkin; +Cc: tarantool-patches Hi! Thanks for the review! >> 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 >> @@ -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); > > I see both tag handle and prefix are the same as in port_lua_dump_plain, > so IMHO it would be great to introduce corresponding constants for them. > Furthermore, these values looks a bit cryptic to me, could you please > drop a few words nearby describing them? Can these values be changed in > the future? Tags are the standard YAML way of adding a type to an object. For details see lua_yaml_encode() comment, and http://yaml.org/spec/1.2/spec.html#tag/shorthand/. YAML allows to introduce your own tags, they just need to have a special syntax. For pushes we've chosen the tag you can see below. They can't be changed in future, since something may already depend on them. They are a part of the public API. Anyway, I now use the tag in only one place in v2 version, so no need to move it to a constant. >> + 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); > > It looks like an additional Lua coroutine is excess here and all > conversion from MessagePack to Lua can be done using the main one (i.e. > tarantool_L). I was afraid that in case of an OOM we will leave garbage in tarantool_L. But if you say we can use tarantool_L safely, then ok, I am not against that. It is just one another reason to start panic when the heap of Lua are out of memory, because otherwise data on tarantool_L will leak. On the other hand, the new thread will also leak, so probably this is a lose-lose situation, but with tarantool_L it is a little simpler. I rewrote that part significantly in v2. > I changed the patch a bit (diff is below) and tests > successfully passed on my machine. Did I miss something (maybe a yield > underneath luamp_decode / lua_yaml_encode)? > > ================================================================================ > > diff --git a/src/box/lua/console.c b/src/box/lua/console.c > index ac353d012..d5e0d193b 100644 > --- a/src/box/lua/console.c > +++ b/src/box/lua/console.c > @@ -457,7 +449,7 @@ port_msgpack_dump_plain(struct port *base, uint32_t *size) > */ > assert(lua_isstring(L, -1)); > diag_set(ClientError, ER_PROC_LUA, lua_tostring(L, -1)); > - goto error; > + return NULL; That is a bad idea, we still have garbage on tarantool_L. Need to drop it first. Anyway now I use cclosure, which automatically removes all leftovers from the stack. > } > assert(lua_gettop(L) - top == 1); > assert(lua_isstring(L, -1)); > ================================================================================ > >> + 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); > > Lua strings can contain NUL byte in the middle of the data area and it > is a good reason for using strndup here, considering you already obtain > the length earlier. Otherwise, it would be nice to drop a corresponding > comment about a contract guaranteeing the NUL byte doesn't occur in the > middle of the result value. YAML and Lua formatters escape 0 by writing '\0'. But anyway I agree, better use malloc + memcpy here. 'strndup' won't work, since it also stops when sees 0. Implemented in v2 and added a test. >> + if (buffer == NULL) { >> + diag_set(OutOfMemory, len + 1, "strdup", "buffer"); >> + goto error; >> + } >> + assert(port->buffer == NULL); >> + port->buffer = buffer; > > It seems to be more convenient to move the lines below to a separate > function (e.g port_msgpack_add_buffer). I have one rationale for it: you > make a strdup right here, but release the allocated memory in > port_msgpack_destroy. Such resource responsibility division seems to be > complex for the further maintainence. If you're against my proposal and > totally OK with the current approach, feel free to ignore. Agree, implemented in v2. ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-22 23:05 ` Vladislav Shpilevoy 2020-04-06 23:12 ` Igor Munkin @ 2020-04-06 23:40 ` Alexander Turenko 1 sibling, 0 replies; 13+ messages in thread From: Alexander Turenko @ 2020-04-06 23:40 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches On Thu, Jan 23, 2020 at 12:05:02AM +0100, Vladislav Shpilevoy wrote: > 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. Igor engaged me to discuss it and now I understand the idea. I'll summarize it below (just in case). Now I understood the difference you told me about. An usual return value is transferred **to a caller**, which may be a Lua function, binary protocol client, SQL VDBE, functional index. A session push value is transferred directly to a 'client' that started the session over all calling chain. The example that illustrates this: a binary protocol client calls a Lua function 'foo', which calls a Lua function 'bar', which performs pushes to a session. Those pushes are transferred directly to the binary procotol client. The nature of 'session push' action is such that it does not need box_function_ctx_t*, which in fact is the bridge to a caller. WBR, Alexander Turenko. ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-20 22:16 [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API Vladislav Shpilevoy 2020-01-20 23:14 ` Vladislav Shpilevoy 2020-01-21 0:24 ` Alexander Turenko @ 2020-03-30 20:42 ` Vladislav Shpilevoy 2020-03-30 21:03 ` Igor Munkin 2020-03-30 20:42 ` Vladislav Shpilevoy 3 siblings, 1 reply; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-03-30 20:42 UTC (permalink / raw) To: tarantool-patches, n.pettik, Igor Munkin Nikita, Igor, please, could you do a review? It seems like I am not getting any reviews here, and the patch's milestone becomes closer to the release. On 20/01/2020 23:16, Vladislav Shpilevoy wrote: > It was a customer request. Nothing special here, the patch is > trivial, with a couple of short notes: > > 1) Sync argument was removed. It will disappear from Lua API as > well, according to #4689. Port is replaced with raw MessagePack, > because port is not a part of public API. 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. > > 2) Dump 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. > > Closes #4734 > --- > Issue: https://github.com/tarantool/tarantool/issues/4734 > Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-4734-export-box-session-push > > extra/exports | 1 + > src/box/box.cc | 12 +++++++++++ > src/box/box.h | 14 +++++++++++++ > src/box/call.c | 14 ++++++++++++- > test/box/function1.c | 7 +++++++ > test/box/function1.result | 41 +++++++++++++++++++++++++++++++++++++ > test/box/function1.test.lua | 19 +++++++++++++++++ > 7 files changed, 107 insertions(+), 1 deletion(-) > > 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..9935c164c 100644 > --- a/src/box/box.cc > +++ b/src/box/box.cc > @@ -1433,6 +1433,18 @@ 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); > + return session_push(session, session_sync(session), base); > +} > + > 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..336044a3b 100644 > --- a/src/box/call.c > +++ b/src/box/call.c > @@ -64,11 +64,23 @@ 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); > > 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, > 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/function1.result b/test/box/function1.result > index b91d63c51..a57de403a 100644 > --- a/test/box/function1.result > +++ b/test/box/function1.result > @@ -1065,3 +1065,44 @@ box.func['test'].is_multikey == true > box.func['test']:drop() > --- > ... > +-- > +-- gh-4734: C API for session push. > +-- > +build_path = os.getenv("BUILDDIR") > +--- > +... > +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath > +--- > +... > +box.schema.func.create('function1.test_push', {language = 'C'}) > +--- > +... > +box.schema.user.grant('guest', 'super') > +--- > +... > +c = net.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() > +--- > +... > +box.schema.user.revoke('guest', 'super') > +--- > +... > +box.schema.func.drop('function1.test_push') > +--- > +... > diff --git a/test/box/function1.test.lua b/test/box/function1.test.lua > index b1841f3ad..4f2c1ea2d 100644 > --- a/test/box/function1.test.lua > +++ b/test/box/function1.test.lua > @@ -380,3 +380,22 @@ box.schema.func.drop("SUM") > box.schema.func.create('test', {body = "function(tuple) return tuple end", is_deterministic = true, opts = {is_multikey = true}}) > box.func['test'].is_multikey == true > box.func['test']:drop() > + > +-- > +-- gh-4734: C API for session push. > +-- > +build_path = os.getenv("BUILDDIR") > +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath > + > +box.schema.func.create('function1.test_push', {language = 'C'}) > +box.schema.user.grant('guest', 'super') > +c = net.connect(box.cfg.listen) > +messages = {} > +c:call('function1.test_push', \ > + {1, 2, 3}, \ > + {on_push = table.insert, \ > + on_push_ctx = messages}) > +messages > +c:close() > +box.schema.user.revoke('guest', 'super') > +box.schema.func.drop('function1.test_push') > ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-03-30 20:42 ` Vladislav Shpilevoy @ 2020-03-30 21:03 ` Igor Munkin 0 siblings, 0 replies; 13+ messages in thread From: Igor Munkin @ 2020-03-30 21:03 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches Vlad, OK, I'll take a look on it this week. On 30.03.20, Vladislav Shpilevoy wrote: > Nikita, Igor, please, could you do a review? It seems like I am > not getting any reviews here, and the patch's milestone becomes > closer to the release. > > On 20/01/2020 23:16, Vladislav Shpilevoy wrote: <snipped> -- Best regards, IM ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-01-20 22:16 [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API Vladislav Shpilevoy ` (2 preceding siblings ...) 2020-03-30 20:42 ` Vladislav Shpilevoy @ 2020-03-30 20:42 ` Vladislav Shpilevoy 2020-04-03 2:30 ` Nikita Pettik 3 siblings, 1 reply; 13+ messages in thread From: Vladislav Shpilevoy @ 2020-03-30 20:42 UTC (permalink / raw) To: tarantool-patches, Igor Munkin, n.pettik @ChangeLog - box_session_push() new public C API function. It takes a const char * MessagePack, and returns it to the client out of order, like Lua analogue (box.session.push()) (gh-4734). On 20/01/2020 23:16, Vladislav Shpilevoy wrote: > It was a customer request. Nothing special here, the patch is > trivial, with a couple of short notes: > > 1) Sync argument was removed. It will disappear from Lua API as > well, according to #4689. Port is replaced with raw MessagePack, > because port is not a part of public API. 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. > > 2) Dump 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. > > Closes #4734 > --- > Issue: https://github.com/tarantool/tarantool/issues/4734 > Branch: https://github.com/tarantool/tarantool/tree/gerold103/gh-4734-export-box-session-push > > extra/exports | 1 + > src/box/box.cc | 12 +++++++++++ > src/box/box.h | 14 +++++++++++++ > src/box/call.c | 14 ++++++++++++- > test/box/function1.c | 7 +++++++ > test/box/function1.result | 41 +++++++++++++++++++++++++++++++++++++ > test/box/function1.test.lua | 19 +++++++++++++++++ > 7 files changed, 107 insertions(+), 1 deletion(-) > > 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..9935c164c 100644 > --- a/src/box/box.cc > +++ b/src/box/box.cc > @@ -1433,6 +1433,18 @@ 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); > + return session_push(session, session_sync(session), base); > +} > + > 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..336044a3b 100644 > --- a/src/box/call.c > +++ b/src/box/call.c > @@ -64,11 +64,23 @@ 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); > > 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, > 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/function1.result b/test/box/function1.result > index b91d63c51..a57de403a 100644 > --- a/test/box/function1.result > +++ b/test/box/function1.result > @@ -1065,3 +1065,44 @@ box.func['test'].is_multikey == true > box.func['test']:drop() > --- > ... > +-- > +-- gh-4734: C API for session push. > +-- > +build_path = os.getenv("BUILDDIR") > +--- > +... > +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath > +--- > +... > +box.schema.func.create('function1.test_push', {language = 'C'}) > +--- > +... > +box.schema.user.grant('guest', 'super') > +--- > +... > +c = net.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() > +--- > +... > +box.schema.user.revoke('guest', 'super') > +--- > +... > +box.schema.func.drop('function1.test_push') > +--- > +... > diff --git a/test/box/function1.test.lua b/test/box/function1.test.lua > index b1841f3ad..4f2c1ea2d 100644 > --- a/test/box/function1.test.lua > +++ b/test/box/function1.test.lua > @@ -380,3 +380,22 @@ box.schema.func.drop("SUM") > box.schema.func.create('test', {body = "function(tuple) return tuple end", is_deterministic = true, opts = {is_multikey = true}}) > box.func['test'].is_multikey == true > box.func['test']:drop() > + > +-- > +-- gh-4734: C API for session push. > +-- > +build_path = os.getenv("BUILDDIR") > +package.cpath = build_path..'/test/box/?.so;'..build_path..'/test/box/?.dylib;'..package.cpath > + > +box.schema.func.create('function1.test_push', {language = 'C'}) > +box.schema.user.grant('guest', 'super') > +c = net.connect(box.cfg.listen) > +messages = {} > +c:call('function1.test_push', \ > + {1, 2, 3}, \ > + {on_push = table.insert, \ > + on_push_ctx = messages}) > +messages > +c:close() > +box.schema.user.revoke('guest', 'super') > +box.schema.func.drop('function1.test_push') > ^ permalink raw reply [flat|nested] 13+ messages in thread
* Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API 2020-03-30 20:42 ` Vladislav Shpilevoy @ 2020-04-03 2:30 ` Nikita Pettik 0 siblings, 0 replies; 13+ messages in thread From: Nikita Pettik @ 2020-04-03 2:30 UTC (permalink / raw) To: Vladislav Shpilevoy; +Cc: tarantool-patches On 30 Mar 22:42, Vladislav Shpilevoy wrote: > @ChangeLog > - box_session_push() new public C API function. It takes a > const char * MessagePack, and returns it to the client > out of order, like Lua analogue (box.session.push()) (gh-4734). LGTM. ^ permalink raw reply [flat|nested] 13+ messages in thread
end of thread, other threads:[~2020-04-07 23:20 UTC | newest] Thread overview: 13+ messages (download: mbox.gz / follow: Atom feed) -- links below jump to the message on this page -- 2020-01-20 22:16 [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API Vladislav Shpilevoy 2020-01-20 23:14 ` Vladislav Shpilevoy 2020-01-21 0:24 ` Alexander Turenko 2020-01-22 0:06 ` Vladislav Shpilevoy 2020-01-22 2:25 ` Alexander Turenko 2020-01-22 23:05 ` Vladislav Shpilevoy 2020-04-06 23:12 ` Igor Munkin 2020-04-07 23:20 ` Vladislav Shpilevoy 2020-04-06 23:40 ` Alexander Turenko 2020-03-30 20:42 ` Vladislav Shpilevoy 2020-03-30 21:03 ` Igor Munkin 2020-03-30 20:42 ` Vladislav Shpilevoy 2020-04-03 2:30 ` Nikita Pettik
This is a public inbox, see mirroring instructions for how to clone and mirror all data and code used for this inbox