[Tarantool-patches] [PATCH v3 3/8] iproto: implement stream id in binary iproto protocol
mechanik20051988
mechanik20051988 at tarantool.org
Wed Aug 11 11:56:53 MSK 2021
From: mechanik20051988 <mechanik20.05.1988 at 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 3d78ce2bb..b9498868c 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 a61c6e345..7df1af4ab 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 48b8b55f5..cb83fddff 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
More information about the Tarantool-patches
mailing list