[tarantool-patches] [PATCH v2 1/8] Encode a dml statement to a transaction memory region

[Georgy Kirichenko <georgy@tarantool.org>]

Encode all statements to be written out to wal onto a transaction
memory region. This relaxes a relation between transaction and fiber
state and required for an autonomous transaction feature.

Prerequisites: #1254
---
 src/box/request.c |  2 +-
 src/box/txn.c     | 27 ++++++++++++++-------------
 src/box/vy_log.c  |  2 +-
 src/box/vy_stmt.c |  4 ++--
 src/box/xrow.c    | 14 ++++++++------
 src/box/xrow.h    | 12 +++++++++---
 6 files changed, 35 insertions(+), 26 deletions(-)

diff --git a/src/box/request.c b/src/box/request.c
index 041a8d548..e2a98fdf9 100644
--- a/src/box/request.c
+++ b/src/box/request.c
@@ -61,7 +61,7 @@ request_update_header(struct request *request, struct xrow_header *row)
 	if (row == NULL)
 		return 0;
 	row->type = request->type;
-	row->bodycnt = xrow_encode_dml(request, row->body);
+	row->bodycnt = xrow_encode_dml(request, &fiber()->gc, row->body);
 	if (row->bodycnt < 0)
 		return -1;
 	request->header = row;
diff --git a/src/box/txn.c b/src/box/txn.c
index 4182d5791..da749d7cc 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -49,11 +49,7 @@ fiber_set_txn(struct fiber *fiber, struct txn *txn)
 static int
 txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 {
-	stmt->row = request->header;
-	if (request->header != NULL)
-		return 0;
-
-	/* Create a redo log row for Lua requests */
+	/* Create a redo log row. */
 	struct xrow_header *row;
 	row = region_alloc_object(&txn->region, struct xrow_header);
 	if (row == NULL) {
@@ -61,14 +57,19 @@ txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 			 "region", "struct xrow_header");
 		return -1;
 	}
-	/* Initialize members explicitly to save time on memset() */
-	row->type = request->type;
-	row->replica_id = 0;
-	row->group_id = stmt->space != NULL ? space_group_id(stmt->space) : 0;
-	row->lsn = 0;
-	row->sync = 0;
-	row->tm = 0;
-	row->bodycnt = xrow_encode_dml(request, row->body);
+	if (request->header != NULL) {
+		*row = *request->header;
+	} else {
+		/* Initialize members explicitly to save time on memset() */
+		row->type = request->type;
+		row->replica_id = 0;
+		row->group_id = stmt->space != NULL ?
+				space_group_id(stmt->space) : 0;
+		row->lsn = 0;
+		row->sync = 0;
+		row->tm = 0;
+	}
+	row->bodycnt = xrow_encode_dml(request, &txn->region, row->body);
 	if (row->bodycnt < 0)
 		return -1;
 	stmt->row = row;
diff --git a/src/box/vy_log.c b/src/box/vy_log.c
index edd61b33a..bdc1cfa31 100644
--- a/src/box/vy_log.c
+++ b/src/box/vy_log.c
@@ -520,7 +520,7 @@ vy_log_record_encode(const struct vy_log_record *record,
 	req.tuple_end = pos;
 	memset(row, 0, sizeof(*row));
 	row->type = req.type;
-	row->bodycnt = xrow_encode_dml(&req, row->body);
+	row->bodycnt = xrow_encode_dml(&req, &fiber()->gc, row->body);
 	return 0;
 }
 
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index 9e20b6bfb..f936cd61f 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -653,7 +653,7 @@ vy_stmt_encode_primary(struct tuple *value, struct key_def *key_def,
 	}
 	if (vy_stmt_meta_encode(value, &request, true) != 0)
 		return -1;
-	xrow->bodycnt = xrow_encode_dml(&request, xrow->body);
+	xrow->bodycnt = xrow_encode_dml(&request, &fiber()->gc, xrow->body);
 	if (xrow->bodycnt < 0)
 		return -1;
 	return 0;
@@ -688,7 +688,7 @@ vy_stmt_encode_secondary(struct tuple *value, struct key_def *cmp_def,
 	}
 	if (vy_stmt_meta_encode(value, &request, false) != 0)
 		return -1;
-	xrow->bodycnt = xrow_encode_dml(&request, xrow->body);
+	xrow->bodycnt = xrow_encode_dml(&request, &fiber()->gc, xrow->body);
 	if (xrow->bodycnt < 0)
 		return -1;
 	else
diff --git a/src/box/xrow.c b/src/box/xrow.c
index 886fb57fe..0ae5271c1 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -752,15 +752,18 @@ request_str(const struct request *request)
 }
 
 int
-xrow_encode_dml(const struct request *request, struct iovec *iov)
+xrow_encode_dml(const struct request *request, struct region *region,
+		struct iovec *iov)
 {
 	int iovcnt = 1;
 	const int MAP_LEN_MAX = 40;
 	uint32_t key_len = request->key_end - request->key;
 	uint32_t ops_len = request->ops_end - request->ops;
 	uint32_t tuple_meta_len = request->tuple_meta_end - request->tuple_meta;
-	uint32_t len = MAP_LEN_MAX + key_len + ops_len + tuple_meta_len;
-	char *begin = (char *) region_alloc(&fiber()->gc, len);
+	uint32_t tuple_len = request->tuple_end - request->tuple;
+	uint32_t len = MAP_LEN_MAX + key_len + ops_len + tuple_meta_len +
+		       tuple_len;
+	char *begin = (char *) region_alloc(region, len);
 	if (begin == NULL) {
 		diag_set(OutOfMemory, len, "region_alloc", "begin");
 		return -1;
@@ -802,9 +805,8 @@ xrow_encode_dml(const struct request *request, struct iovec *iov)
 	}
 	if (request->tuple) {
 		pos = mp_encode_uint(pos, IPROTO_TUPLE);
-		iov[iovcnt].iov_base = (void *) request->tuple;
-		iov[iovcnt].iov_len = (request->tuple_end - request->tuple);
-		iovcnt++;
+		memcpy(pos, request->tuple, tuple_len);
+		pos += tuple_len;
 		map_size++;
 	}
 
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 3a8cba191..35ec06dc0 100644
--- a/src/box/xrow.h
+++ b/src/box/xrow.h
@@ -54,6 +54,8 @@ enum {
 	IPROTO_SELECT_HEADER_LEN = IPROTO_HEADER_LEN + 7,
 };
 
+struct region;
+
 struct xrow_header {
 	/* (!) Please update txn_add_redo() after changing members */
 
@@ -195,12 +197,15 @@ xrow_decode_dml(struct xrow_header *xrow, struct request *request,
 /**
  * Encode the request fields to iovec using region_alloc().
  * @param request request to encode
+ * @param region region to encode
+ * @param copy_tuple if true then tuple is going to be copied to the region
  * @param iov[out] iovec to fill
  * @retval -1 on error, see diag
  * @retval > 0 the number of iovecs used
  */
 int
-xrow_encode_dml(const struct request *request, struct iovec *iov);
+xrow_encode_dml(const struct request *request, struct region *region,
+		struct iovec *iov);
 
 /**
  * CALL/EVAL request.
@@ -713,9 +718,10 @@ xrow_decode_dml_xc(struct xrow_header *row, struct request *request,
 
 /** @copydoc xrow_encode_dml. */
 static inline int
-xrow_encode_dml_xc(const struct request *request, struct iovec *iov)
+xrow_encode_dml_xc(const struct request *request, struct region *region,
+		   struct iovec *iov)
 {
-	int iovcnt = xrow_encode_dml(request, iov);
+	int iovcnt = xrow_encode_dml(request, region, iov);
 	if (iovcnt < 0)
 		diag_raise();
 	return iovcnt;
-- 
2.21.0