[Tarantool-patches] [PATCH 5/7] net.box: add stream support to net.box
Vladimir Davydov
vdavydov at tarantool.org
Fri Aug 6 15:03:26 MSK 2021
On Thu, Aug 05, 2021 at 09:17:43PM +0300, mechanik20051988 wrote:
> From: mechanik20051988 <mechanik20.05.1988 at gmail.com>
>
> Add stream support to `net.box`. In "net.box", stream
> is an object over connection that has the same methods,
> but all requests from it sends with non-zero stream ID.
> Since there can be a lot of streams, we do not copy the
> spaces from the connection to the stream immediately when
> creating a stream, but do it only when we first access space.
> Also, when updating the schema, we update the spaces only for
> those streams that have had at least one access to the space.
>
> Part of #5860
>
> @TarantoolBot document
> Title: stream support was added to net.box
> In "net.box", stream is an object over connection that
> has the same methods, but all requests from it sends
> with non-zero stream ID.
I see some differences between a stream and a connection while we want
them to be completely interchangeable.
1. They are serialized differently:
tarantool> c
---
- peer_uuid: 140674d0-38e7-4fbd-9a54-7133dc9151af
schema_version: 95
protocol: Binary
state: active
peer_version_id: 133376
port: '3301'
...
tarantool> s
---
- &0
_schema_version: 95
commit: 'function: 0x414d32e0'
begin: 'function: 0x414d3278'
rollback: 'function: 0x414d3390'
space:
_space: []
_stream: *0
_stream_id: 1
new_stream: 'function: 0x414d31f8'
_conn:
peer_uuid: 140674d0-38e7-4fbd-9a54-7133dc9151af
schema_version: 95
protocol: Binary
state: active
peer_version_id: 133376
port: '3301'
...
2. When typed into a console, c.space lists all spaces while s.space
shows this:
tarantool> s.space
---
- &0
_space: []
_stream: &1
_schema_version: 95
commit: 'function: 0x414d32e0'
begin: 'function: 0x414d3278'
rollback: 'function: 0x414d3390'
space: *0
_stream_id: 1
new_stream: 'function: 0x414d31f8'
_conn:
peer_uuid: 140674d0-38e7-4fbd-9a54-7133dc9151af
schema_version: 95
protocol: Binary
state: active
peer_version_id: 133376
port: '3301'
...
There are other similar differences in serialization. For example,
compare s.space._vfunc and c.space._vfunc or s.space._vfunc.index
and c.space._vfunc.index.
3. c.space._space refers to the corresponding system space while
s.space._space doesn't.
4. Autocompletion doesn't work for s.space.<TAB> and
s.space._vfunc.<TAB> and s.space._vfunc.index.<TAB>
Some of them (most notably 3) can and must be fixed. I'm not so sure
about autocompletion though. Please check.
Also, I think that stream_id should be visible (without a leading
underscore) when you dump a stream to the console.
> Stream ID is generated on the client side in two ways:
> automatically or manually.
Manual stream ids are not supported anymore. Please update the comment.
> User can choose any of two methods, but can not mix them.
> Simple example of stream creation using net.box:
> ```lua
> -- automatically generated stream_id
> stream = conn:stream()
Now it's called conn:new_stream().
> -- manually chosen stream_id
> stream = conn:stream(1)
> ```
> ---
> src/box/lua/net_box.c | 101 ++--
> src/box/lua/net_box.lua | 191 ++++--
> test/box/access.result | 6 +-
> test/box/access.test.lua | 6 +-
> ...net.box_console_connections_gh-2677.result | 2 +-
> ...t.box_console_connections_gh-2677.test.lua | 2 +-
> .../net.box_incorrect_iterator_gh-841.result | 4 +-
> ...net.box_incorrect_iterator_gh-841.test.lua | 4 +-
> test/box/net.box_iproto_hangs_gh-3464.result | 2 +-
> .../box/net.box_iproto_hangs_gh-3464.test.lua | 2 +-
> .../net.box_long-poll_input_gh-3400.result | 8 +-
> .../net.box_long-poll_input_gh-3400.test.lua | 8 +-
> test/box/stream.lua | 13 +
> test/box/stream.result | 553 ++++++++++++++++++
> test/box/stream.test.lua | 207 +++++++
> test/box/suite.ini | 2 +-
> 16 files changed, 1011 insertions(+), 100 deletions(-)
> create mode 100644 test/box/stream.lua
> create mode 100644 test/box/stream.result
> create mode 100644 test/box/stream.test.lua
>
> diff --git a/src/box/lua/net_box.c b/src/box/lua/net_box.c
> index 82efc483d..ec850cd9f 100644
> --- a/src/box/lua/net_box.c
> +++ b/src/box/lua/net_box.c
> @@ -128,20 +132,21 @@ netbox_encode_request(struct mpstream *stream, size_t initial_size)
>
> static void
> netbox_encode_ping(lua_State *L, int idx, struct mpstream *stream,
> - uint64_t sync)
> + uint64_t sync, uint64_t stream_id)
> {
> (void)L;
> (void)idx;
> - size_t svp = netbox_prepare_request(stream, sync, IPROTO_PING);
> + size_t svp = netbox_prepare_request(stream, sync,
> + IPROTO_PING, stream_id);
> netbox_encode_request(stream, svp);
> }
>
> static int
> netbox_encode_auth(lua_State *L)
> {
> - if (lua_gettop(L) < 5) {
> + if (lua_gettop(L) < 6) {
> return luaL_error(L, "Usage: netbox.encode_update(ibuf, sync, "
> - "user, password, greeting)");
> + "stream_id, user, password, greeting)");
IPROTO_AUTH shouldn't support streams so this is not necessary. Looking
at net_box.lua, you always pass nil for stream_id anyway. Please revert
this part.
> }
> struct ibuf *ibuf = (struct ibuf *)lua_topointer(L, 1);
> uint64_t sync = luaL_touint64(L, 2);
> @@ -149,14 +154,14 @@ netbox_encode_auth(lua_State *L)
> struct mpstream stream;
> mpstream_init(&stream, ibuf, ibuf_reserve_cb, ibuf_alloc_cb,
> luamp_error, L);
> - size_t svp = netbox_prepare_request(&stream, sync, IPROTO_AUTH);
> + size_t svp = netbox_prepare_request(&stream, sync, IPROTO_AUTH, 0);
>
> size_t user_len;
> - const char *user = lua_tolstring(L, 3, &user_len);
> + const char *user = lua_tolstring(L, 4, &user_len);
> size_t password_len;
> - const char *password = lua_tolstring(L, 4, &password_len);
> + const char *password = lua_tolstring(L, 5, &password_len);
> size_t salt_len;
> - const char *salt = lua_tolstring(L, 5, &salt_len);
> + const char *salt = lua_tolstring(L, 6, &salt_len);
> if (salt_len < SCRAMBLE_SIZE)
> return luaL_error(L, "Invalid salt");
>
> diff --git a/src/box/lua/net_box.lua b/src/box/lua/net_box.lua
> index 9af6028eb..bf6a89e15 100644
> --- a/src/box/lua/net_box.lua
> +++ b/src/box/lua/net_box.lua
> @@ -930,6 +930,8 @@ local function remote_serialize(self)
> }
> end
>
> +local stream_methods = {}
> +
This table is not really needed. Please remove.
> local remote_methods = {}
> local remote_mt = {
> __index = remote_methods, __serialize = remote_serialize,
> @@ -942,6 +944,90 @@ local console_mt = {
> __metatable = false
> }
>
> +local stream_index_mt = {
> + __index = function(self, key)
> + return self._src[key]
> + end
> +}
> +
> +-- Create stream space index, which is same as connection space
> +-- index, but have non zero stream ID.
> +local function stream_wrap_index(stream_id, src)
> + return setmetatable({
> + _stream_id = stream_id,
> + _src = src,
> + }, stream_index_mt)
> +end
> +
> +-- Metatable for stream space indexes. When stream space being
> +-- created there are no indexes in it. When accessing the space
> +-- index, we look for corresponding space index in corresponding
> +-- connection space. If it is found we create same index for the
> +-- stream space but with corresponding stream ID. We do not need
> +-- to compare stream _schema_version and connection schema_version,
> +-- because all access to index is carried out through it's space.
> +-- So we update schema_version when we access space.
> +local stream_indexes_mt = {
> + __index = function(self, key)
> + local _space = self._space
> + local src = _space._src.index[key]
> + if not src then
> + return nil
> + end
> + local res = stream_wrap_index(_space._stream_id, src)
> + self[key] = res
> + return res
> + end
> +}
> +
> +local stream_space_mt = {
> + __index = function(self, key)
> + return self._src[key]
> + end
> +}
> +
> +-- Create stream space, which is same as connection space,
> +-- but have non zero stream ID.
> +local function stream_wrap_space(stream, src)
> + local res = setmetatable({
> + _stream_id = stream._stream_id,
> + _src = src,
> + index = setmetatable({
> + _space = nil,
> + }, stream_indexes_mt)
> + }, stream_space_mt)
> + res.index._space = res
> + return res
> +end
> +
> +-- Metatable for stream spaces. When stream being created there
> +-- are no spaces in it. When user try to access some space in
> +-- stream, we first of all compare _schema_version of stream with
> +-- schema_version from connection and if they are not equal, we
> +-- clear stream space cache and update it's schema_version. Then
> +-- we look for corresponding space in the connection. If it is
> +-- found we create same space for the stream but with corresponding
> +-- stream ID.
> +local stream_spaces_mt = {
> + __index = function(self, key)
> + local stream = self._stream
> + if stream._schema_version ~= stream._conn.schema_version then
> + stream._schema_version = stream._conn.schema_version
> + self._space = {}
> + end
> + if self._space[key] then
> + return self._space[key]
> + end
> + local src = stream._conn.space[key]
> + if not src then
> + return nil
> + end
> + local res = stream_wrap_space(stream, src)
> + self._space[key] = res
> + return res
> + end
> +}
> +
> local space_metatable, index_metatable
>
> local function new_sm(host, port, opts, connection, greeting)
> @@ -1021,6 +1107,9 @@ local function new_sm(host, port, opts, connection, greeting)
> if opts.wait_connected ~= false then
> remote._transport.wait_state('active', tonumber(opts.wait_connected))
> end
> + -- Last stream ID used for this connection
> + remote._last_stream_id = 0
> + remote._streams = setmetatable({}, {__mode = 'v'})
This table is never read from. Please remove.
> return remote
> end
>
> @@ -1078,6 +1167,29 @@ local function check_eval_args(args)
> end
> end
>
> +function stream_methods:new_stream()
Can be inlined.
> + check_remote_arg(self, 'stream')
'new_stream'
> + box.error(E_PROC_LUA, "Unsupported for stream");
Why? Let's create a new stream here:
return self._conn:new_stream()
> +end
> +
> +function remote_methods:new_stream()
> + check_remote_arg(self, 'stream')
'new_stream'
> + self._last_stream_id = self._last_stream_id + 1
> + local stream = setmetatable({
> + new_stream = stream_methods.new_stream,
> + _stream_id = self._last_stream_id,
> + space = setmetatable({
> + _space = {},
> + _stream = nil,
> + }, stream_spaces_mt),
> + _conn = self,
> + _schema_version = self.schema_version,
> + }, { __index = self })
> + stream.space._stream = stream
> + self._streams[self._last_stream_id] = stream
> + return stream
> +end
> +
> function remote_methods:close()
> check_remote_arg(self, 'close')
> self._transport.stop()
> @@ -1108,7 +1220,7 @@ function remote_methods:wait_connected(timeout)
> return self._transport.wait_state('active', timeout)
> end
>
> -function remote_methods:_request(method, opts, format, ...)
> +function remote_methods:_request(method, opts, format, stream_id, ...)
> local transport = self._transport
> local on_push, on_push_ctx, buffer, skip_header, deadline
> -- Extract options, set defaults, check if the request is
> @@ -1123,7 +1235,7 @@ function remote_methods:_request(method, opts, format, ...)
> local res, err =
> transport.perform_async_request(buffer, skip_header, method,
> table.insert, {}, format,
> - ...)
> + stream_id, ...)
> if err then
> box.error(err)
> end
> @@ -1145,7 +1257,7 @@ function remote_methods:_request(method, opts, format, ...)
> end
> local res, err = transport.perform_request(timeout, buffer, skip_header,
> method, on_push, on_push_ctx,
> - format, ...)
> + format, stream_id, ...)
> if err then
> box.error(err)
> end
> @@ -1161,7 +1273,7 @@ end
>
> function remote_methods:ping(opts)
> check_remote_arg(self, 'ping')
> - return (pcall(self._request, self, M_PING, opts))
> + return (pcall(self._request, self, M_PING, opts, nil, self._stream_id))
> end
>
> function remote_methods:reload_schema()
> @@ -1172,14 +1284,16 @@ end
> -- @deprecated since 1.7.4
> function remote_methods:call_16(func_name, ...)
> check_remote_arg(self, 'call')
> - return (self:_request(M_CALL_16, nil, nil, tostring(func_name), {...}))
> + return (self:_request(M_CALL_16, nil, nil, self._stream_id,
> + tostring(func_name), {...}))
> end
>
> function remote_methods:call(func_name, args, opts)
> check_remote_arg(self, 'call')
> check_call_args(args)
> args = args or {}
> - local res = self:_request(M_CALL_17, opts, nil, tostring(func_name), args)
> + local res = self:_request(M_CALL_17, opts, nil, self._stream_id,
> + tostring(func_name), args)
Rather than passing self._stream_id through all the net.box methods,
can't you simply set it in _request?
More information about the Tarantool-patches
mailing list