[Tarantool-patches] [PATCH 1/7] iproto: implement stream id in binary iproto protocol

[mechanik20051988 via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

From: mechanik20051988 <mechanik20.05.1988@gmail.com>

For further implementation of streams, we need to separate
requests belonging to and not belonging to streams. For this
purpose, the stream ID field was added to the iproto binary
protocol. For requests that do not belong to stream, this field
is omitted or equal to zero. For requests belonging to stream,
we use this field to determine which stream the request belongs to.

Part of #5860

@TarantoolBot document
Title: new field in binary iproto protocol

Add new field to binary iproto protocol.
`IPROTO_STREAM_ID 0x0a` determines whether a request
belongs to a stream or not. If this field is omited
or equal to zero this request doesn't belongs to stream.
---
 src/box/iproto_constants.c |   4 +-
 src/box/iproto_constants.h |   1 +
 src/box/xrow.c             |   8 ++
 src/box/xrow.h             |   5 ++
 test/unit/xrow.cc          |   7 +-
 test/unit/xrow.result      | 168 +++++++++++++++++++------------------
 6 files changed, 109 insertions(+), 84 deletions(-)

diff --git a/src/box/iproto_constants.c b/src/box/iproto_constants.c
index addda39dc..f2902946a 100644
--- a/src/box/iproto_constants.c
+++ b/src/box/iproto_constants.c
@@ -43,10 +43,10 @@ const unsigned char iproto_key_type[IPROTO_KEY_MAX] =
 		/* 0x07 */	MP_UINT,   /* IPROTO_GROUP_ID */
 		/* 0x08 */	MP_UINT,   /* IPROTO_TSN */
 		/* 0x09 */	MP_UINT,   /* IPROTO_FLAGS */
+		/* 0x0a */	MP_UINT,   /* IPROTO_STREAM_ID */
 	/* }}} */
 
 	/* {{{ unused */
-		/* 0x0a */	MP_UINT,
 		/* 0x0b */	MP_UINT,
 		/* 0x0c */	MP_UINT,
 		/* 0x0d */	MP_UINT,
@@ -198,7 +198,7 @@ const char *iproto_key_strs[IPROTO_KEY_MAX] = {
 	"group id",         /* 0x07 */
 	"tsn",              /* 0x08 */
 	"flags",            /* 0x09 */
-	NULL,               /* 0x0a */
+	"stream_id",        /* 0x0a */
 	NULL,               /* 0x0b */
 	NULL,               /* 0x0c */
 	NULL,               /* 0x0d */
diff --git a/src/box/iproto_constants.h b/src/box/iproto_constants.h
index ea7290da6..59e8574f3 100644
--- a/src/box/iproto_constants.h
+++ b/src/box/iproto_constants.h
@@ -72,6 +72,7 @@ enum iproto_key {
 	IPROTO_GROUP_ID = 0x07,
 	IPROTO_TSN = 0x08,
 	IPROTO_FLAGS = 0x09,
+	IPROTO_STREAM_ID = 0x0a,
 	/* Leave a gap for other keys in the header. */
 	IPROTO_SPACE_ID = 0x10,
 	IPROTO_INDEX_ID = 0x11,
diff --git a/src/box/xrow.c b/src/box/xrow.c
index e5d9d5be5..44a75616a 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -186,6 +186,9 @@ error:
 			flags = mp_decode_uint(pos);
 			header->flags = flags;
 			break;
+		case IPROTO_STREAM_ID:
+			header->stream_id = mp_decode_uint(pos);
+			break;
 		default:
 			/* unknown header */
 			mp_next(pos);
@@ -319,6 +322,11 @@ xrow_header_encode(const struct xrow_header *header, uint64_t sync,
 			flags_to_encode |= IPROTO_FLAG_COMMIT;
 		}
 	}
+	if (header->stream_id != 0) {
+		d = mp_encode_uint(d, IPROTO_STREAM_ID);
+		d = mp_encode_uint(d, header->stream_id);
+		map_size++;
+	}
 	if (flags_to_encode != 0) {
 		d = mp_encode_uint(d, IPROTO_FLAGS);
 		d = mp_encode_uint(d, flags_to_encode);
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 000cd6b88..44f9cbdd6 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -81,6 +81,11 @@ struct xrow_header {
 	 * transaction.
 	 */
 	int64_t tsn;
+	/**
+	 * Stream id. Used in iproto binary protocol to identify stream.
+	 * Zero if stream is not used.
+	 */
+	uint64_t stream_id;
 	/** Transaction meta flags set only in the last transaction row. */
 	union {
 		uint8_t flags;
diff --git a/test/unit/xrow.cc b/test/unit/xrow.cc
index b6018eed9..2c0dd88b6 100644
--- a/test/unit/xrow.cc
+++ b/test/unit/xrow.cc
@@ -220,8 +220,10 @@ test_xrow_header_encode_decode()
 	header.bodycnt = 0;
 	header.tsn = header.lsn;
 	uint64_t sync = 100500;
+	uint64_t stream_id = 1;
 	for (int opt_idx = 0; opt_idx < bit_comb_count; opt_idx++) {
-		plan(12);
+		plan(13);
+		header.stream_id = stream_id++;
 		header.is_commit = opt_idx & 0x01;
 		header.wait_sync = opt_idx >> 1 & 0x01;
 		header.wait_ack = opt_idx >> 2 & 0x01;
@@ -229,7 +231,7 @@ test_xrow_header_encode_decode()
 		is(1, xrow_header_encode(&header, sync, vec, 200), "encode");
 		int fixheader_len = 200;
 		pos = (char *)vec[0].iov_base + fixheader_len;
-		uint32_t exp_map_size = 5;
+		uint32_t exp_map_size = 6;
 		/*
 		 * header.is_commit flag isn't encoded, since this row looks
 		 * like a single-statement transaction.
@@ -249,6 +251,7 @@ test_xrow_header_encode_decode()
 		end += vec[0].iov_len;
 		is(xrow_header_decode(&decoded_header, &begin, end, true), 0,
 		   "header decode");
+		is(header.stream_id, decoded_header.stream_id, "decoded stream_id");
 		is(header.is_commit, decoded_header.is_commit, "decoded is_commit");
 		is(header.wait_sync, decoded_header.wait_sync, "decoded wait_sync");
 		is(header.wait_ack, decoded_header.wait_ack, "decoded wait_ack");
diff --git a/test/unit/xrow.result b/test/unit/xrow.result
index 3b705d5ba..1ca222d37 100644
--- a/test/unit/xrow.result
+++ b/test/unit/xrow.result
@@ -43,117 +43,125 @@
 ok 1 - subtests
     1..9
     ok 1 - bad msgpack end
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 2 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 3 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 4 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 5 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 6 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 7 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 8 - subtests
-        1..12
+        1..13
         ok 1 - encode
         ok 2 - header map size
         ok 3 - header decode
-        ok 4 - decoded is_commit
-        ok 5 - decoded wait_sync
-        ok 6 - decoded wait_ack
-        ok 7 - decoded type
-        ok 8 - decoded replica_id
-        ok 9 - decoded lsn
-        ok 10 - decoded tm
-        ok 11 - decoded sync
-        ok 12 - decoded bodycnt
+        ok 4 - decoded stream_id
+        ok 5 - decoded is_commit
+        ok 6 - decoded wait_sync
+        ok 7 - decoded wait_ack
+        ok 8 - decoded type
+        ok 9 - decoded replica_id
+        ok 10 - decoded lsn
+        ok 11 - decoded tm
+        ok 12 - decoded sync
+        ok 13 - decoded bodycnt
     ok 9 - subtests
 ok 2 - subtests
     1..1
-- 
2.20.1