Re: [Tarantool-patches] [PATCH 1/1] box: export box_session_push to the public C API

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

@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')
>