net.box: add interactive transaction support in net.box

Implement `begin`, `commit` and `rollback` methods for stream object
in `net.box`, which allows to begin, commit and rollback transaction
accordingly.

Closes #5860

@TarantoolBot document
Title: add interactive transaction support in net.box
Implement `begin`, `commit` and `rollback` methods for stream object
in `net.box`, which allows to begin, commit and rollback transaction
accordingly. Now there are multiple ways to begin, commit and rollback
transaction from `net.box`: using appropriate stream methods, using 'call`
or 'eval' methods or using `execute` method with sql transaction syntax.
User can mix these methods, for example, start transaction using
`stream:begin()`, and commit transaction using `stream:call('box.commit')`
or stream:execute('COMMIT').
Simple example of using interactive transactions via iproto from net.box:
```lua
stream = conn:new_stream()
space = stream.space.test
space_not_from_stream = conn.space.test

stream:begin()
space:replace({1})
-- return previously inserted tuple, because request
-- belongs to transaction.
space:select({})
-- empty select, because select doesn't belongs to
-- transaction
space_not_from_stream:select({})
stream:call('box.commit')
-- now transaction was commited, so all requests
-- returns tuple.
```
Different examples of using streams you can find in
gh-5860-implement-streams-in-iproto.test.lua

Author: EvgenyMekhanik

changelogs/unreleased/gh-5860-implement-streams-in-iproto.md

@@ -0,0 +1,26 @@
+## feature/core
+
+* Streams and interactive transactions over streams are implemented
+  in iproto. Stream is associated with it's ID, which is unique within
+  one connection. All requests with same not zero stream ID belongs to
+  the same stream. All requests in stream processed synchronously. The
+  execution of the next request will not start until the previous one is
+  completed. If request has zero stream ID it does not belong to stream
+  and is processed in the old way.
+  In `net.box`, stream is an object above connection that has the same
+  methods, but allows to execute requests sequentially. ID is generated
+  on the client side automatically. If user writes his own connector and
+  wants to use streams, he must transmit stream_id over iproto protocol.
+  The main purpose of streams is transactions via iproto. Each stream
+  can start its own transaction, so they allows multiplexing several
+  transactions over one connection. There are multiple ways to begin,
+  commit and rollback transaction: using appropriate stream methods, using
+  `call` or `eval` methods or using `execute` method with sql transaction
+  syntax. User can mix these methods, for example, start transaction using
+  `stream:begin()`, and commit transaction using `stream:call('box.commit')`
+  or stream:execute('COMMIT').
+  If any request fails during the transaction, it will not affect the other
+  requests in the transaction. If disconnect occurs when there is some active
+  transaction in stream, this transaction will be rollbacked, if it does not
+  have time to commit before this moment.
+

src/box/lua/net_box.c

@@ -75,7 +75,10 @@ enum netbox_method {
 	NETBOX_MIN         = 14,
 	NETBOX_MAX         = 15,
 	NETBOX_COUNT       = 16,
-	NETBOX_INJECT      = 17,
+	NETBOX_BEGIN       = 17,
+	NETBOX_COMMIT      = 18,
+	NETBOX_ROLLBACK    = 19,
+	NETBOX_INJECT      = 20,
 	netbox_method_MAX
 };
 
@@ -916,6 +919,44 @@ netbox_encode_unprepare(lua_State *L, int idx, struct mpstream *stream,
 	netbox_encode_prepare(L, idx, stream, sync, stream_id);
 }
 
+static inline void
+netbox_encode_txn(lua_State *L, enum iproto_type type, int idx,
+		  struct mpstream *stream, uint64_t sync,
+		  uint64_t stream_id)
+{
+	(void)L;
+	(void) idx;
+	assert(type == IPROTO_BEGIN ||
+	       type == IPROTO_COMMIT ||
+	       type == IPROTO_ROLLBACK);
+	size_t svp = netbox_begin_encode(stream, sync, type, stream_id);
+	netbox_end_encode(stream, svp);
+}
+
+static void
+netbox_encode_begin(struct lua_State *L, int idx, struct mpstream *stream,
+		    uint64_t sync, uint64_t stream_id)
+{
+	return netbox_encode_txn(L, IPROTO_BEGIN, idx, stream,
+				 sync, stream_id);
+}
+
+static void
+netbox_encode_commit(struct lua_State *L, int idx, struct mpstream *stream,
+		     uint64_t sync, uint64_t stream_id)
+{
+	return netbox_encode_txn(L, IPROTO_COMMIT, idx, stream,
+				 sync, stream_id);
+}
+
+static void
+netbox_encode_rollback(struct lua_State *L, int idx, struct mpstream *stream,
+		       uint64_t sync, uint64_t stream_id)
+{
+	return netbox_encode_txn(L, IPROTO_ROLLBACK, idx, stream,
+				 sync, stream_id);
+}
+
 static void
 netbox_encode_inject(struct lua_State *L, int idx, struct mpstream *stream,
 		     uint64_t sync, uint64_t stream_id)
@@ -959,6 +1000,9 @@ netbox_encode_method(struct lua_State *L, int idx, enum netbox_method method,
 		[NETBOX_MIN]		= netbox_encode_select,
 		[NETBOX_MAX]		= netbox_encode_select,
 		[NETBOX_COUNT]		= netbox_encode_call,
+		[NETBOX_BEGIN]          = netbox_encode_begin,
+		[NETBOX_COMMIT]         = netbox_encode_commit,
+		[NETBOX_ROLLBACK]       = netbox_encode_rollback,
 		[NETBOX_INJECT]		= netbox_encode_inject,
 	};
 	struct mpstream stream;
@@ -1330,6 +1374,9 @@ netbox_decode_method(struct lua_State *L, enum netbox_method method,
 		[NETBOX_MIN]		= netbox_decode_tuple,
 		[NETBOX_MAX]		= netbox_decode_tuple,
 		[NETBOX_COUNT]		= netbox_decode_value,
+		[NETBOX_BEGIN]          = netbox_decode_nil,
+		[NETBOX_COMMIT]         = netbox_decode_nil,
+		[NETBOX_ROLLBACK]       = netbox_decode_nil,
 		[NETBOX_INJECT]		= netbox_decode_table,
 	};
 	method_decoder[method](L, data, data_end, format);

src/box/lua/net_box.lua

@@ -51,8 +51,11 @@ local M_GET         = 13
 local M_MIN         = 14
 local M_MAX         = 15
 local M_COUNT       = 16
+local M_BEGIN       = 17
+local M_COMMIT      = 18
+local M_ROLLBACK    = 19
 -- Injects raw data into connection. Used by console and tests.
-local M_INJECT      = 17
+local M_INJECT      = 20
 
 -- utility tables
 local is_final_state         = {closed = 1, error = 1}
@@ -754,11 +757,38 @@ local function stream_new_stream(stream)
     return stream._conn:new_stream()
 end
 
+local function stream_begin(stream, opts)
+    check_remote_arg(stream, 'begin')
+    local res = stream:_request(M_BEGIN, opts, nil, stream._stream_id)
+    if opts and opts.is_async then
+        return res
+    end
+end
+
+local function stream_commit(stream, opts)
+    check_remote_arg(stream, 'commit')
+    local res = stream:_request(M_COMMIT, opts, nil, stream._stream_id)
+    if opts and opts.is_async then
+        return res
+    end
+end
+
+local function stream_rollback(stream, opts)
+    check_remote_arg(stream, 'rollback')
+    local res = stream:_request(M_ROLLBACK, opts, nil, stream._stream_id)
+    if opts and opts.is_async then
+        return res
+    end
+end
+
 function remote_methods:new_stream()
     check_remote_arg(self, 'new_stream')
     self._last_stream_id = self._last_stream_id + 1
     local stream = setmetatable({
         new_stream = stream_new_stream,
+        begin = stream_begin,
+        commit = stream_commit,
+        rollback = stream_rollback,
         _stream_id = self._last_stream_id,
         space = setmetatable({
             _stream_space_cache = {},
@@ -1243,6 +1273,9 @@ local this_module = {
         min         = M_MIN,
         max         = M_MAX,
         count       = M_COUNT,
+        begin       = M_BEGIN,
+        commit      = M_COMMIT,
+        rollback    = M_ROLLBACK,
         inject      = M_INJECT,
     }
 }