[PATCH 3/4] sql: decode bind parameters in TX thread

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Before this patch bind parameters were decoded in
iproto thread in an attempt to save some TX thread
time. But they were stored in an array allocated on
iproto thread memory. Lack of iproto_msg destructor
lead to leak of this array.

This patch makes sql_request store only raw undecoded
MessagePack and moves decoding to TX thread where in
a single function the parameters are decoded, applied
and freed.

There were a couple of alternatives allowing to keep
decoding in iproto thread and to move even more
decoding there, for example, parameters from
IPROTO_UPDATE:

* decode on malloc and add a virtual destructor to
  struct iproto_msg;

* decode on malloc and add a virtual destructor to
  struct cmsg. By the way, it could allow to reuse
  this destructor for struct cbus_call_msg and
  struct relay_gc_msg.

The way how is it solved by this patch has been chosen
thanks to its simplicity.

Closes #3828
---
 src/box/execute.c | 60 +++++++++++++++--------------------------------
 src/box/execute.h | 37 +++++++++++++++++++++--------
 src/box/iproto.cc | 14 +++++++++--
 3 files changed, 58 insertions(+), 53 deletions(-)

diff --git a/src/box/execute.c b/src/box/execute.c
index 3c4898216..25a594d7d 100644
--- a/src/box/execute.c
+++ b/src/box/execute.c
@@ -194,27 +194,9 @@ sql_bind_decode(struct sql_bind *bind, int i, const char **packet)
 	return 0;
 }
 
-/**
- * Parse MessagePack array of SQL parameters and store a result
- * into the @request->bind, bind_count.
- * @param request Request to save decoded parameters.
- * @param data MessagePack array of parameters. Each parameter
- *        either must have scalar type, or must be a map with the
- *        following format: {name: value}. Name - string name of
- *        the named parameter, value - scalar value of the
- *        parameter. Named and positioned parameters can be mixed.
- *        For more details
- *        @sa https://www.sqlite.org/lang_expr.html#varparam.
- * @param region Allocator.
- *
- * @retval  0 Success.
- * @retval -1 Client or memory error.
- */
-static inline int
-sql_bind_list_decode(struct sql_request *request, const char *data,
-		     struct region *region)
+int
+sql_bind_list_decode(const char *data, struct sql_bind **out_bind)
 {
-	assert(request != NULL);
 	assert(data != NULL);
 	if (mp_typeof(*data) != MP_ARRAY) {
 		diag_set(ClientError, ER_INVALID_MSGPACK, "SQL parameter list");
@@ -228,6 +210,7 @@ sql_bind_list_decode(struct sql_request *request, const char *data,
 			 (int) bind_count);
 		return -1;
 	}
+	struct region *region = &fiber()->gc;
 	uint32_t used = region_used(region);
 	size_t size = sizeof(struct sql_bind) * bind_count;
 	struct sql_bind *bind = (struct sql_bind *) region_alloc(region, size);
@@ -241,14 +224,12 @@ sql_bind_list_decode(struct sql_request *request, const char *data,
 			return -1;
 		}
 	}
-	request->bind_count = bind_count;
-	request->bind = bind;
-	return 0;
+	*out_bind = bind;
+	return bind_count;
 }
 
 int
-xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
-		struct region *region)
+xrow_decode_sql(const struct xrow_header *row, struct sql_request *request)
 {
 	if (row->bodycnt == 0) {
 		diag_set(ClientError, ER_INVALID_MSGPACK, "missing request body");
@@ -268,7 +249,6 @@ error:
 	uint32_t map_size = mp_decode_map(&data);
 	request->sql_text = NULL;
 	request->bind = NULL;
-	request->bind_count = 0;
 	for (uint32_t i = 0; i < map_size; ++i) {
 		uint8_t key = *data;
 		if (key != IPROTO_SQL_BIND && key != IPROTO_SQL_TEXT) {
@@ -279,12 +259,10 @@ error:
 		const char *value = ++data;     /* skip the key */
 		if (mp_check(&data, end) != 0)  /* check the value */
 			goto error;
-		if (key == IPROTO_SQL_BIND) {
-			if (sql_bind_list_decode(request, value, region) != 0)
-				return -1;
-		} else {
+		if (key == IPROTO_SQL_BIND)
+			request->bind = value;
+		else
 			request->sql_text = value;
-		}
 	}
 	if (request->sql_text == NULL) {
 		diag_set(ClientError, ER_MISSING_REQUEST_FIELD,
@@ -500,19 +478,21 @@ sql_bind_column(struct sqlite3_stmt *stmt, const struct sql_bind *p,
 
 /**
  * Bind parameter values to the prepared statement.
- * @param request Parsed SQL request.
  * @param stmt Prepared statement.
+ * @param bind Parameters to bind.
+ * @param bind_count Length of @a bind.
  *
  * @retval  0 Success.
  * @retval -1 Client or memory error.
  */
 static inline int
-sql_bind(const struct sql_request *request, struct sqlite3_stmt *stmt)
+sql_bind(struct sqlite3_stmt *stmt, const struct sql_bind *bind,
+	 uint32_t bind_count)
 {
 	assert(stmt != NULL);
 	uint32_t pos = 1;
-	for (uint32_t i = 0; i < request->bind_count; pos = ++i + 1) {
-		if (sql_bind_column(stmt, &request->bind[i], pos) != 0)
+	for (uint32_t i = 0; i < bind_count; pos = ++i + 1) {
+		if (sql_bind_column(stmt, &bind[i], pos) != 0)
 			return -1;
 	}
 	return 0;
@@ -595,12 +575,10 @@ sql_execute(sqlite3 *db, struct sqlite3_stmt *stmt, struct port *port,
 }
 
 int
-sql_prepare_and_execute(const struct sql_request *request,
-			struct sql_response *response, struct region *region)
+sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
+			uint32_t bind_count, struct sql_response *response,
+			struct region *region)
 {
-	const char *sql = request->sql_text;
-	uint32_t len;
-	sql = mp_decode_str(&sql, &len);
 	struct sqlite3_stmt *stmt;
 	sqlite3 *db = sql_get();
 	if (db == NULL) {
@@ -614,7 +592,7 @@ sql_prepare_and_execute(const struct sql_request *request,
 	assert(stmt != NULL);
 	port_tuple_create(&response->port);
 	response->prep_stmt = stmt;
-	if (sql_bind(request, stmt) == 0 &&
+	if (sql_bind(stmt, bind, bind_count) == 0 &&
 	    sql_execute(db, stmt, &response->port, region) == 0)
 		return 0;
 	port_destroy(&response->port);
diff --git a/src/box/execute.h b/src/box/execute.h
index 266566260..025695ea9 100644
--- a/src/box/execute.h
+++ b/src/box/execute.h
@@ -57,10 +57,8 @@ struct xrow_header;
 struct sql_request {
 	/** SQL statement text. */
 	const char *sql_text;
-	/** Array of parameters. */
-	struct sql_bind *bind;
-	/** Length of the @bind. */
-	uint32_t bind_count;
+	/** MessagePack array of parameters. */
+	const char *bind;
 };
 
 /** Response on EXECUTE request. */
@@ -71,6 +69,23 @@ struct sql_response {
 	void *prep_stmt;
 };
 
+/**
+ * Parse MessagePack array of SQL parameters.
+ * @param data MessagePack array of parameters. Each parameter
+ *        either must have scalar type, or must be a map with the
+ *        following format: {name: value}. Name - string name of
+ *        the named parameter, value - scalar value of the
+ *        parameter. Named and positioned parameters can be mixed.
+ *        For more details
+ *        @sa https://www.sqlite.org/lang_expr.html#varparam.
+ * @param[out] out_bind Pointer to save decoded parameters.
+ *
+ * @retval  >= 0 Number of decoded parameters.
+ * @retval -1 Client or memory error.
+ */
+int
+sql_bind_list_decode(const char *data, struct sql_bind **out_bind);
+
 /**
  * Dump a built response into @an out buffer. The response is
  * destroyed.
@@ -112,18 +127,19 @@ sql_response_dump(struct sql_response *response, int *keys, struct obuf *out);
  * Parse the EXECUTE request.
  * @param row Encoded data.
  * @param[out] request Request to decode to.
- * @param region Allocator.
  *
  * @retval  0 Sucess.
  * @retval -1 Format or memory error.
  */
 int
-xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
-		struct region *region);
+xrow_decode_sql(const struct xrow_header *row, struct sql_request *request);
 
 /**
  * Prepare and execute an SQL statement.
- * @param request IProto request.
+ * @param sql SQL statement.
+ * @param len Length of @a sql.
+ * @param bind Array of parameters.
+ * @param bind_count Length of @a bind.
  * @param[out] response Response to store result.
  * @param region Runtime allocator for temporary objects
  *        (columns, tuples ...).
@@ -132,8 +148,9 @@ xrow_decode_sql(const struct xrow_header *row, struct sql_request *request,
  * @retval -1 Client or memory error.
  */
 int
-sql_prepare_and_execute(const struct sql_request *request,
-			struct sql_response *response, struct region *region);
+sql_prepare_and_execute(const char *sql, int len, const struct sql_bind *bind,
+			uint32_t bind_count, struct sql_response *response,
+			struct region *region);
 
 #if defined(__cplusplus)
 } /* extern "C" { */
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index f24a56e4e..3ebe4ddad 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -1156,7 +1156,7 @@ iproto_msg_decode(struct iproto_msg *msg, const char **pos, const char *reqend,
 		cmsg_init(&msg->base, call_route);
 		break;
 	case IPROTO_EXECUTE:
-		if (xrow_decode_sql(&msg->header, &msg->sql, &fiber()->gc))
+		if (xrow_decode_sql(&msg->header, &msg->sql) != 0)
 			goto error;
 		cmsg_init(&msg->base, sql_route);
 		break;
@@ -1579,6 +1579,10 @@ tx_process_sql(struct cmsg *m)
 	struct iproto_msg *msg = tx_accept_msg(m);
 	struct obuf *out;
 	struct sql_response response;
+	struct sql_bind *bind;
+	int bind_count;
+	const char *sql;
+	uint32_t len;
 
 	tx_fiber_init(msg->connection->session, msg->header.sync);
 
@@ -1586,7 +1590,13 @@ tx_process_sql(struct cmsg *m)
 		goto error;
 	assert(msg->header.type == IPROTO_EXECUTE);
 	tx_inject_delay();
-	if (sql_prepare_and_execute(&msg->sql, &response, &fiber()->gc) != 0)
+	bind_count = sql_bind_list_decode(msg->sql.bind, &bind);
+	if (bind_count < 0)
+		goto error;
+	sql = msg->sql.sql_text;
+	sql = mp_decode_str(&sql, &len);
+	if (sql_prepare_and_execute(sql, len, bind, bind_count, &response,
+				    &fiber()->gc) != 0)
 		goto error;
 	/*
 	 * Take an obuf only after execute(). Else the buffer can
-- 
2.17.2 (Apple Git-113)