box: sending arbitrary IPROTO packets

Add translation table for `box.iproto.key` constants encoding to simplify
packet assembly.

Add new `box.iproto.send` method to Lua and `box_iproto_send` function to C
API, which allow sending arbitrary IPROTO packets, using active IPROTO
sessions. Packets are sent asynchronously using Kharon.

Add `xregion_join` to the `xalloc` API.

Change gh-7894 test: instead of simply comparing `box.iproto` table to the
reference table, iterate over `box.iproto` and check that corresponding
non-{function, thread, userdata} type values exist in the reference table.

Closes #7897

@TarantoolBot document
Title: Document sending arbitrary IPROTO packets feature
For the API description and usage examples, see:
* [design document](https://www.notion.so/tarantool/box-iproto-override-44935a6ac7e04fb5a2c81ca713ed1bce#a2cc04da89d34fad8f8564c150cd9977);
* #7897.

Author: CuriousGeorgiy

changelogs/unreleased/gh-7897-sending-arbitrary-iproto-packets.md

@@ -0,0 +1,4 @@
+## feature/box
+
+* Added API for sending arbitrary IPROTO packets from server to client:
+  `box.iproto.send` in Lua and `box_iproto_send` in C (gh-7897).

extra/exports

@@ -70,6 +70,7 @@ box_index_min
 box_index_random
 box_index_tuple_position
 box_insert
+box_iproto_send
 box_iterator_free
 box_iterator_next
 box_key_def_delete

src/box/box.cc

@@ -3250,6 +3250,24 @@ box_session_id(void)
 	return current_session()->id;
 }
 
+API_EXPORT int
+box_iproto_send(uint64_t sid,
+		const char *header, const char *header_end,
+		const char *body, const char *body_end)
+{
+	struct session *session = session_find(sid);
+	if (session == NULL) {
+		diag_set(ClientError, ER_NO_SUCH_SESSION, sid);
+		return -1;
+	}
+	if (session->type != SESSION_TYPE_BINARY) {
+		diag_set(ClientError, ER_WRONG_SESSION_TYPE,
+			 session_type_strs[session->type]);
+		return -1;
+	}
+	return iproto_session_send(session, header, header_end, body, body_end);
+}
+
 static inline void
 box_register_replica(uint32_t id, const struct tt_uuid *uuid)
 {

src/box/box.h

@@ -601,6 +601,25 @@ box_session_push(const char *data, const char *data_end);
 API_EXPORT uint64_t
 box_session_id(void);
 
+/**
+ * Sends a packet with the given header and body over the IPROTO session's
+ * socket.
+ *
+ * NB: yields.
+ *
+ * \param sid IPROTO session identifier
+ * \param header MsgPack encoded header
+ * \param header_end MsgPack encoded header end
+ * \param body MsgPack encoded body
+ * \param body_end MsgPack encoded body end
+ * \retval -1 on error (check box_error_last())
+ * \retval 0 on success
+ */
+API_EXPORT int
+box_iproto_send(uint64_t sid,
+		const char *header, const char *header_end,
+		const char *body, const char *body_end);
+
 /** \endcond public */
 
 /**

src/box/errcode.h

@@ -306,6 +306,8 @@ struct errcode_record {
 	/*251 */_(ER_INVALID_AUTH_REQUEST,	"Invalid '%s' request: %s") \
 	/*252 */_(ER_WEAK_PASSWORD,		"Password doesn't meet security requirements: %s") \
 	/*253 */_(ER_OLD_PASSWORD,		"Password must differ from last %d passwords") \
+	/*254 */_(ER_NO_SUCH_SESSION,		"Session %llu does not exist") \
+	/*255 */_(ER_WRONG_SESSION_TYPE,	"Session '%s' is not supported") \
 
 /*
  * !IMPORTANT! Please follow instructions at start of the file

src/box/iproto.cc

@@ -3299,6 +3299,42 @@ iproto_set_msg_max(int new_iproto_msg_max)
 	}
 }
 
+int
+iproto_session_send(struct session *session,
+		    const char *header, const char *header_end,
+		    const char *body, const char *body_end)
+{
+	assert(session->type == SESSION_TYPE_BINARY);
+	struct iproto_connection *con =
+		(struct iproto_connection *)session->meta.connection;
+	if (con->state != IPROTO_CONNECTION_ALIVE) {
+		diag_set(ClientError, ER_SESSION_CLOSED);
+		return -1;
+	}
+
+	ptrdiff_t header_size = header_end - header;
+	ptrdiff_t body_size = body_end - body;
+	ptrdiff_t packet_size = 5 + header_size + body_size;
+	char *buf = (char *)obuf_alloc(con->tx.p_obuf, packet_size);
+	if (buf == NULL) {
+		diag_set(OutOfMemory, packet_size, "obuf_alloc", "buf");
+		return -1;
+	}
+	char *p = buf;
+	*(p++) = INT8_C(0xce);
+	p = mp_store_u32(p, packet_size - 5);
+	memcpy(p, header, header_size);
+	p += header_size;
+	memcpy(p, body, body_size);
+	tx_push(con);
+	/*
+	 * The control yield is solely for enforcing the fact this function
+	 * yields — in the future we may implement back pressure based on this.
+	 */
+	fiber_sleep(0);
+	return 0;
+}
+
 void
 iproto_free(void)
 {

src/box/iproto.h

@@ -34,6 +34,7 @@
 #include <stddef.h>
 
 struct uri_set;
+struct session;
 
 #if defined(__cplusplus)
 extern "C" {
@@ -131,6 +132,18 @@ iproto_listen(const struct uri_set *uri_set);
 void
 iproto_set_msg_max(int iproto_msg_max);
 
+/**
+ * Sends a packet with the given header and body over the IPROTO session's
+ * socket.
+ * On success, a packet is written to the session's output buffer, which is
+ * flushed asynchronously using Kharon.
+ * Returns 0 on success, a non-zero value otherwise (diagnostic is set).
+ */
+int
+iproto_session_send(struct session *session,
+		    const char *header, const char *header_end,
+		    const char *body, const char *body_end);
+
 void
 iproto_free(void);
 

src/box/lua/init.c

@@ -585,3 +585,9 @@ box_lua_init(struct lua_State *L)
 
 	assert(lua_gettop(L) == 0);
 }
+
+void
+box_lua_free(void)
+{
+	box_lua_iproto_free();
+}

src/box/lua/init.h

@@ -39,6 +39,12 @@ struct lua_State;
 void
 box_lua_init(struct lua_State *L);
 
+/**
+ * Frees `box` module.
+ */
+void
+box_lua_free(void);
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */

src/box/lua/iproto.c

@@ -6,10 +6,29 @@
 
 #include "box/lua/iproto.h"
 
+#include "box/box.h"
 #include "box/iproto_constants.h"
 #include "box/iproto_features.h"
 
-#include <lua.h>
+#include "core/assoc.h"
+#include "core/fiber.h"
+#include "core/tt_static.h"
+
+#include "lua/msgpack.h"
+#include "lua/utils.h"
+
+#include "mpstream/mpstream.h"
+
+#include "small/region.h"
+
+#include <ctype.h>
+#include <lauxlib.h>
+
+/**
+ * Translation table for `box.iproto.key` constants encoding: used in
+ * `luamp_encode_with_translation`.
+ */
+static struct mh_strnu32_t *iproto_key_translation;
 
 /**
  * IPROTO constant from `src/box/iproto_{constants, features}.h`.
@@ -119,6 +138,22 @@ push_iproto_key_enum(struct lua_State *L)
 		{"VCLOCK_SYNC", IPROTO_VCLOCK_SYNC},
 	};
 	push_iproto_constant_subnamespace(L, "key", keys, lengthof(keys));
+	for (size_t i = 0; i < lengthof(keys); ++i) {
+		size_t len = strlen(keys[i].name);
+		char *lowercase = strtolowerdup(keys[i].name);
+		struct mh_strnu32_node_t translation = {
+			.str = lowercase,
+			.len = len,
+			.hash = lua_hash(lowercase, len),
+			.val = keys[i].val,
+		};
+		mh_strnu32_put(iproto_key_translation, &translation,
+			       NULL, NULL);
+		translation.str = xstrdup(keys[i].name);
+		translation.hash = lua_hash(translation.str, len);
+		mh_strnu32_put(iproto_key_translation, &translation,
+			       NULL, NULL);
+	}
 }
 
 /**
@@ -268,16 +303,112 @@ push_iproto_protocol_features(struct lua_State *L)
 					  lengthof(features));
 }
 
+/**
+ * Encodes a packet header/body argument to MsgPack: if the argument is a
+ * string, then no encoding is needed — otherwise the argument must be a Lua
+ * table. The Lua table is encoded to MsgPack using IPROTO key translation
+ * table.
+ * In both cases, the result is stored on the fiber region.
+ */
+static const char *
+encode_packet(struct lua_State *L, int idx, size_t *mp_len)
+{
+	int packet_part_type = lua_type(L, idx);
+	struct region *gc = &fiber()->gc;
+	if (packet_part_type == LUA_TSTRING) {
+		const char *arg = lua_tolstring(L, idx, mp_len);
+		char *mp = xregion_alloc(gc, *mp_len);
+		return memcpy(mp, arg, *mp_len);
+	}
+	assert(packet_part_type == LUA_TTABLE);
+	/*
+	 * FIXME(gh-7939): `luamp_error` and `luamp_encode_with_translation` can
+	 * throw a Lua exception, which we cannot catch, and cause the fiber
+	 * region to leak.
+	 */
+	struct mpstream stream;
+	mpstream_init(&stream, gc, region_reserve_cb, region_alloc_cb,
+		      luamp_error, L);
+	size_t used = region_used(gc);
+	luamp_encode_with_translation(L, luaL_msgpack_default, &stream, idx,
+				      iproto_key_translation);
+	mpstream_flush(&stream);
+	*mp_len = region_used(gc) - used;
+	return xregion_join(gc, *mp_len);
+}
+
+/**
+ * Sends an IPROTO packet consisting of a header (second argument) and an
+ * optional body (third argument) over the IPROTO session identified by first
+ * argument.
+ */
+static int
+lbox_iproto_send(struct lua_State *L)
+{
+	int n_args = lua_gettop(L);
+	if (n_args < 2 || n_args > 3)
+		return luaL_error(L, "Usage: "
+				     "box.iproto.send(sid, header[, body])");
+	uint64_t sid = luaL_checkuint64(L, 1);
+	int header_type = lua_type(L, 2);
+	if (header_type != LUA_TSTRING && header_type != LUA_TTABLE)
+		return luaL_error(L, "expected table or string as 2 argument");
+	if (n_args == 3) {
+		int body_type = lua_type(L, 3);
+		if (body_type != LUA_TSTRING && body_type != LUA_TTABLE)
+			return luaL_error(
+				L, "expected table or string as 3 argument");
+	}
+
+	struct region *gc = &fiber()->gc;
+	size_t used = region_used(gc);
+	size_t header_len;
+	const char *header = encode_packet(L, 2, &header_len);
+	size_t body_len = 0;
+	const char *body = NULL;
+	if (n_args == 3)
+		body = encode_packet(L, 3, &body_len);
+	int rc = box_iproto_send(sid,
+				 header, header + header_len,
+				 body, body + body_len);
+	region_truncate(gc, used);
+	return (rc == 0) ? 0 : luaT_error(L);
+}
+
 /**
  * Initializes module for working with Tarantool's network subsystem.
  */
 void
 box_lua_iproto_init(struct lua_State *L)
 {
+	iproto_key_translation = mh_strnu32_new();
+
 	lua_getfield(L, LUA_GLOBALSINDEX, "box");
 	lua_newtable(L);
+
 	push_iproto_constants(L);
 	push_iproto_protocol_features(L);
+
+	const struct luaL_Reg iproto_methods[] = {
+		{"send", lbox_iproto_send},
+		{NULL, NULL}
+	};
+	luaL_register(L, NULL, iproto_methods);
+
 	lua_setfield(L, -2, "iproto");
 	lua_pop(L, 1);
 }
+
+/**
+ * Deletes the IPROTO key translation and all its dynamically allocated key
+ * strings.
+ */
+void
+box_lua_iproto_free(void)
+{
+	struct mh_strnu32_t *h = iproto_key_translation;
+	mh_int_t k;
+	mh_foreach(h, k)
+		free((void *)mh_strnu32_node(h, k)->str);
+	mh_strnu32_delete(iproto_key_translation);
+}

src/box/lua/iproto.h

@@ -12,9 +12,19 @@ extern "C" {
 
 struct lua_State;
 
+/**
+ * Initializes `box.iproto` submodule for working with Tarantool network
+ * subsystem.
+ */
 void
 box_lua_iproto_init(struct lua_State *L);
 
+/**
+ * Frees `box.iproto` submodule.
+ */
+void
+box_lua_iproto_free(void);
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */

src/main.cc

@@ -557,6 +557,7 @@ tarantool_free(void)
 	/* Shutdown worker pool. Waits until threads terminate. */
 	coio_shutdown();
 
+	box_lua_free();
 	box_free();
 
 	title_free(main_argc, main_argv);

src/trivia/util.h

@@ -133,6 +133,7 @@ strnindex(const char *const *haystack, const char *needle, uint32_t len,
 #define xregion_alloc_array(p, T, count, size)				\
 	xalloc_impl(sizeof(T) * (count), region_alloc_array, (p), T,	\
 		    (count), (size))
+#define xregion_join(p, size)	xalloc_impl((size), region_join, (p), (size))
 #define xibuf_alloc(p, size)	xalloc_impl((size), ibuf_alloc, (p), (size))
 #define xibuf_reserve(p, size)	xalloc_impl((size), ibuf_reserve, (p), (size))