[tarantool-patches] [PATCH 03/10] Encode a dml statement to a transaction memory region

[Georgy Kirichenko]

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     |  2 +-
 src/box/vy_log.c  |  2 +-
 src/box/vy_stmt.c |  6 ++++--
 src/box/xrow.c    | 21 +++++++++++++++------
 src/box/xrow.h    | 12 +++++++++---
 6 files changed, 31 insertions(+), 14 deletions(-)

diff --git a/src/box/request.c b/src/box/request.c
index 9c684af73..9edd1f8b1 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, false, row->body);
 	if (row->bodycnt < 0)
 		return -1;
 	request->header = row;
diff --git a/src/box/txn.c b/src/box/txn.c
index ed5e54bb6..ffada5984 100644
--- a/src/box/txn.c
+++ b/src/box/txn.c
@@ -68,7 +68,7 @@ txn_add_redo(struct txn *txn, struct txn_stmt *stmt, struct request *request)
 	row->lsn = 0;
 	row->sync = 0;
 	row->tm = 0;
-	row->bodycnt = xrow_encode_dml(request, row->body);
+	row->bodycnt = xrow_encode_dml(request, &txn->region, true, 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 85059588e..76b21c822 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, false, row->body);
 	return 0;
 }
 
diff --git a/src/box/vy_stmt.c b/src/box/vy_stmt.c
index e1cdd293d..3bc22965c 100644
--- a/src/box/vy_stmt.c
+++ b/src/box/vy_stmt.c
@@ -681,7 +681,8 @@ 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, false,
+					xrow->body);
 	if (xrow->bodycnt < 0)
 		return -1;
 	return 0;
@@ -715,7 +716,8 @@ 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, false,
+					xrow->body);
 	if (xrow->bodycnt < 0)
 		return -1;
 	else
diff --git a/src/box/xrow.c b/src/box/xrow.c
index aaed84e38..1fc60f6d8 100644
--- a/src/box/xrow.c
+++ b/src/box/xrow.c
@@ -746,15 +746,19 @@ 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,
+		bool copy_tuple, 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 = copy_tuple ?
+			     (request->tuple_end - request->tuple) : 0;
+	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;
@@ -796,9 +800,14 @@ 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++;
+		if (copy_tuple) {
+			memcpy(pos, request->tuple, tuple_len);
+			pos += tuple_len;
+		} else {
+			iov[iovcnt].iov_base = (void *) request->tuple;
+			iov[iovcnt].iov_len = (request->tuple_end - request->tuple);
+			iovcnt++;
+		}
 		map_size++;
 	}
 
diff --git a/src/box/xrow.h b/src/box/xrow.h
index 3a8cba191..c1d737efd 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,
+		bool copy_tuple, 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,
+		   bool copy_tuple, struct iovec *iov)
 {
-	int iovcnt = xrow_encode_dml(request, iov);
+	int iovcnt = xrow_encode_dml(request, region, copy_tuple, iov);
 	if (iovcnt < 0)
 		diag_raise();
 	return iovcnt;
-- 
2.21.0