box: implement Lua call/eval over iproto for application threads

This commit lets a network client specify a thread id to forward
a call/eval request to. To achieve that, we have to initialize
a minimal Lua state in each application threads. We do that by making
the tarantool_L global variable thread-local and factoring out
tarantool_lua_init_minimal() from tarantool_lua_init() for use in
application threads. The minimal initialization includes setting
ctype ids for all our custom Lua types, such as decimal and datetime,
because we need them for MsgPack serialization. These ctype ids are
made thread-local as well. The only exception is tuple because
tuple formats can't be used in any thread except tx. We explicitly
override the tuple decoder to return a plain table instead of a tuple
object if used in a non-tx thread, see luamp_decode_extension_box().

To safely use the box Lua call/eval infrastructure in any thread,
we have to fix a few things:

 1. execute_lua_refs and call_serializer_no_ext have to be made
    thread-local because they depend on the Lua state.
 2. Apart from setting execute_lua_refs and call_serializer_no_ext,
    box_lua_call_init() registers an internal box library and sets
    a trigger on alter of the _func system space. Apparently, we can't
    do that in application threads.
 3. get_call_serializer() uses current_session(), which creates a new
    session object if the current fiber doesn't have a session. Sessions
    aren't thread-safe and we don't set them for fibers running in
    application threads so we switch to fiber_get_session(). We also
    add an explicit check that this is an IPROTO session because for
    other sessions IPROTO feature bits are undefined. The latter is
    merely for code clarity because Lua call/eval are executed only
    from IPROTO.

This commit completes the core part of the application threads
infrastructure.

Closes #12259

NO_DOC=internal
NO_CHANGELOG=internal

Author: locker

src/box/CMakeLists.txt

@@ -297,6 +297,7 @@ set(box_sources
     lua/key_def.c
     lua/merger.c
     lua/watcher.c
+    lua/app_threads.c
     lua/iproto.c
     lua/func_adapter.c
     lua/tuple_format.c

src/box/app_threads.c

@@ -8,15 +8,21 @@
 #include <assert.h>
 #include <limits.h>
 #include <stddef.h>
+#include <stdint.h>
 #include <stdlib.h>
 #include <stdio.h>
 
 #include "diag.h"
 #include "fiber.h"
 #include "fiber_pool.h"
+#include "msgpuck.h"
+#include "port.h"
 #include "say.h"
 #include "tarantool_ev.h"
 #include "trivia/util.h"
+#include "xrow.h"
+
+#include "lua/app_threads.h"
 
 int app_thread_count;
 
@@ -27,11 +33,13 @@ static void *
 app_thread_f(void *unused)
 {
 	(void)unused;
+	app_thread_lua_init();
 	struct fiber_pool fiber_pool;
 	fiber_pool_create(&fiber_pool, cord_name(cord()), INT_MAX,
 			  FIBER_POOL_IDLE_TIMEOUT);
 	ev_run(loop(), 0);
 	fiber_pool_destroy(&fiber_pool);
+	app_thread_lua_free();
 	return NULL;
 }
 
@@ -68,3 +76,29 @@ app_threads_stop(void)
 	app_thread_cords = NULL;
 	app_thread_count = 0;
 }
+
+int
+app_thread_process_call(struct call_request *request, struct port *port)
+{
+	const char *name = request->name;
+	uint32_t name_len = mp_decode_strl(&name);
+	struct port args;
+	port_msgpack_create(&args, request->args,
+			    request->args_end - request->args);
+	int rc = app_thread_lua_call(name, name_len, &args, port);
+	port_msgpack_destroy(&args);
+	return rc;
+}
+
+int
+app_thread_process_eval(struct call_request *request, struct port *port)
+{
+	const char *expr = request->expr;
+	uint32_t expr_len = mp_decode_strl(&expr);
+	struct port args;
+	port_msgpack_create(&args, request->args,
+			    request->args_end - request->args);
+	int rc = app_thread_lua_eval(expr, expr_len, &args, port);
+	port_msgpack_destroy(&args);
+	return rc;
+}

src/box/app_threads.h

@@ -9,6 +9,9 @@
 extern "C" {
 #endif /* defined(__cplusplus) */
 
+struct call_request;
+struct port;
+
 enum {
 	APP_THREADS_MAX = 1000,
 };
@@ -32,6 +35,18 @@ app_threads_start(int thread_count);
 void
 app_threads_stop(void);
 
+/**
+ * Processes CALL request in this application thread.
+ */
+int
+app_thread_process_call(struct call_request *request, struct port *port);
+
+/**
+ * Processes EVAL request in this application thread.
+ */
+int
+app_thread_process_eval(struct call_request *request, struct port *port);
+
 #if defined(__cplusplus)
 } /* extern "C" */
 #endif /* defined(__cplusplus) */

src/box/iproto.cc

@@ -163,6 +163,8 @@ struct iproto_thread {
 		 * Used for returning request results.
 		 */
 		struct cpipe ret_pipe;
+		/** CALL/EVAL route. */
+		struct cmsg_hop call_route[2];
 		/** Route used to destroy an IPROTO connection. */
 		struct cmsg_hop destroy_route[2];
 	} *srv;
@@ -177,7 +179,6 @@ struct iproto_thread {
 	struct cmsg_hop rollback_on_disconnect_route[2];
 	struct cmsg_hop disconnect_route[2];
 	struct cmsg_hop misc_route[2];
-	struct cmsg_hop call_route[2];
 	struct cmsg_hop select_route[2];
 	struct cmsg_hop process1_route[2];
 	struct cmsg_hop sql_route[2];
@@ -1940,7 +1941,8 @@ iproto_msg_prepare(struct iproto_msg *msg, const char **pos, const char *reqend)
 		diag_set(ClientError, ER_NO_SUCH_THREAD, thread_id);
 		goto error;
 	}
-	if (thread_id != XROW_THREAD_UNSPEC && thread_id != 0) {
+	if (thread_id != XROW_THREAD_UNSPEC && thread_id != 0 &&
+	    type != IPROTO_CALL && type != IPROTO_EVAL) {
 		diag_set(ClientError, ER_UNABLE_TO_PROCESS_IN_THREAD,
 			 iproto_type_name(type), thread_id);
 		goto error;
@@ -1990,6 +1992,9 @@ iproto_msg_prepare(struct iproto_msg *msg, const char **pos, const char *reqend)
 		return;
 	}
 
+	if (thread_id != XROW_THREAD_UNSPEC)
+		msg->srv_id = thread_id;
+
 	rc = iproto_msg_decode(msg, &route);
 	if (rc == 0) {
 		assert(route != NULL);
@@ -2054,7 +2059,7 @@ iproto_msg_decode(struct iproto_msg *msg, struct cmsg_hop **route)
 	case IPROTO_CALL_16:
 	case IPROTO_CALL:
 	case IPROTO_EVAL:
-		*route = iproto_thread->call_route;
+		*route = iproto_thread->srv[msg->srv_id].call_route;
 		if (xrow_decode_call(&msg->header, &msg->call))
 			return -1;
 		return 0;
@@ -2772,7 +2777,7 @@ tx_process_select(struct cmsg *m)
 }
 
 static int
-tx_process_call_on_yield(struct trigger *trigger, void *event)
+srv_process_call_on_yield(struct trigger *trigger, void *event)
 {
 	(void)event;
 	struct iproto_msg *msg = (struct iproto_msg *)trigger->data;
@@ -2810,7 +2815,7 @@ tx_process_call(struct cmsg *m)
 	 * a long polling request.
 	 */
 	struct trigger fiber_on_yield;
-	trigger_create(&fiber_on_yield, tx_process_call_on_yield, msg, NULL);
+	trigger_create(&fiber_on_yield, srv_process_call_on_yield, msg, NULL);
 	trigger_add(&fiber()->on_yield, &fiber_on_yield);
 
 	int rc;
@@ -2885,6 +2890,60 @@ tx_process_call(struct cmsg *m)
 	tx_end_msg(msg, &svp);
 }
 
+/**
+ * Process a CALL/EVAL request in a serving thread.
+ */
+static void
+srv_process_call(struct cmsg *m)
+{
+	struct iproto_msg *msg = srv_accept_msg(m);
+	struct obuf *out;
+	struct obuf_svp svp;
+	struct port port;
+	int count;
+	int rc;
+
+	struct trigger fiber_on_yield;
+	trigger_create(&fiber_on_yield, srv_process_call_on_yield, msg, NULL);
+	trigger_add(&fiber()->on_yield, &fiber_on_yield);
+
+	switch (msg->header.type) {
+	case IPROTO_CALL:
+		rc = app_thread_process_call(&msg->call, &port);
+		break;
+	case IPROTO_EVAL:
+		rc = app_thread_process_eval(&msg->call, &port);
+		break;
+	default:
+		unreachable();
+	}
+
+	trigger_clear(&fiber_on_yield);
+
+	if (rc != 0)
+		goto error;
+
+	out = iproto_msg_obuf(msg);
+	iproto_prepare_select(out, &svp);
+	count = port_dump_msgpack(&port, out);
+	port_destroy(&port);
+	if (count < 0) {
+		obuf_rollback_to_svp(out, &svp);
+		goto error;
+	}
+	iproto_reply_select(out, &svp, msg->header.sync, /*schema_version=*/0,
+			    count, /*box_tuple_as_ext=*/false);
+	iproto_wpos_create(&msg->wpos, out);
+	srv_end_msg(msg);
+	return;
+error:
+	out = iproto_msg_obuf(msg);
+	iproto_reply_error(out, diag_last_error(&fiber()->diag),
+			   msg->header.sync, /*schema_version=*/0);
+	iproto_wpos_create(&msg->wpos, out);
+	srv_end_msg(msg);
+}
+
 static void
 tx_process_id(struct iproto_connection *con, const struct id_request *id)
 {
@@ -3879,8 +3938,6 @@ iproto_thread_init_routes(struct iproto_thread *iproto_thread)
 	iproto_thread->disconnect_route[1] = {net_finish_disconnect, NULL};
 	iproto_thread->misc_route[0] = {tx_process_misc, net_pipe};
 	iproto_thread->misc_route[1] = {net_send_msg, NULL};
-	iproto_thread->call_route[0] = {tx_process_call, net_pipe};
-	iproto_thread->call_route[1] = {net_send_msg, NULL};
 	iproto_thread->select_route[0] = {tx_process_select, net_pipe};
 	iproto_thread->select_route[1] = {net_send_msg, NULL};
 	iproto_thread->process1_route[0] = {tx_process1, net_pipe};
@@ -3897,30 +3954,27 @@ iproto_thread_init_routes(struct iproto_thread *iproto_thread)
 	iproto_thread->push_route[1] = {tx_end_push, NULL};
 
 	struct cmsg_hop **dml_route = iproto_thread->dml_route;
-	assert(dml_route[IPROTO_OK] == NULL);
 	dml_route[IPROTO_SELECT] = iproto_thread->select_route;
 	dml_route[IPROTO_INSERT] = iproto_thread->process1_route;
 	dml_route[IPROTO_REPLACE] = iproto_thread->process1_route;
 	dml_route[IPROTO_UPDATE] = iproto_thread->process1_route;
 	dml_route[IPROTO_DELETE] = iproto_thread->process1_route;
-	dml_route[IPROTO_CALL_16] = iproto_thread->call_route;
-	dml_route[IPROTO_AUTH] = iproto_thread->misc_route;
-	dml_route[IPROTO_EVAL] = iproto_thread->call_route;
 	dml_route[IPROTO_UPSERT] = iproto_thread->process1_route;
-	dml_route[IPROTO_CALL] = iproto_thread->call_route;
-	dml_route[IPROTO_EXECUTE] = iproto_thread->sql_route;
-	assert(dml_route[IPROTO_NOP] == NULL);
-	dml_route[IPROTO_PREPARE] = iproto_thread->sql_route;
-	assert(dml_route[IPROTO_BEGIN] == NULL);
-	assert(dml_route[IPROTO_COMMIT] == NULL);
-	assert(dml_route[IPROTO_ROLLBACK] == NULL);
 	dml_route[IPROTO_INSERT_ARROW] = iproto_thread->process1_route;
 
 	iproto_thread->connect_route[0] = {tx_process_connect, net_pipe};
 	iproto_thread->connect_route[1] = {net_send_greeting, NULL};
 	iproto_thread->override_route[0] = {tx_process_override, net_pipe};
 	iproto_thread->override_route[1] = {net_send_msg, NULL};
 
+	iproto_thread->srv[0].call_route[0] = {tx_process_call, net_pipe};
+	iproto_thread->srv[0].call_route[1] = {net_send_msg, NULL};
+	for (int i = 1; i < iproto_thread->srv_count; i++) {
+		iproto_thread->srv[i].call_route[0] =
+			{srv_process_call, &iproto_thread->srv[i].ret_pipe};
+		iproto_thread->srv[i].call_route[1] =
+			{net_send_msg, NULL};
+	}
 	for (int i = 0; i < iproto_thread->srv_count; i++) {
 		iproto_thread->srv[i].destroy_route[0] =
 			{srv_process_destroy, &iproto_thread->srv[i].ret_pipe};

src/box/lua/app_threads.c

@@ -0,0 +1,40 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2026, Tarantool AUTHORS, please see AUTHORS file.
+ */
+#include "box/lua/app_threads.h"
+
+#include <lua.h>
+#include <stdint.h>
+
+#include "lua/init.h"
+#include "box/lua/call.h"
+
+void
+app_thread_lua_init(void)
+{
+	tarantool_lua_init_minimal();
+	box_lua_call_init(tarantool_L);
+}
+
+void
+app_thread_lua_free(void)
+{
+	lua_close(tarantool_L);
+	tarantool_L = NULL;
+}
+
+int
+app_thread_lua_call(const char *name, uint32_t name_len,
+		    struct port *args, struct port *ret)
+{
+	return box_lua_call(name, name_len, args, ret);
+}
+
+int
+app_thread_lua_eval(const char *expr, uint32_t expr_len,
+		    struct port *args, struct port *ret)
+{
+	return box_lua_eval(expr, expr_len, args, ret);
+}

src/box/lua/app_threads.h

@@ -0,0 +1,38 @@
+/*
+ * SPDX-License-Identifier: BSD-2-Clause
+ *
+ * Copyright 2010-2026, Tarantool AUTHORS, please see AUTHORS file.
+ */
+#pragma once
+
+#include <stdint.h>
+
+#if defined(__cplusplus)
+extern "C" {
+#endif /* defined(__cplusplus) */
+
+struct port;
+
+void
+app_thread_lua_init(void);
+
+void
+app_thread_lua_free(void);
+
+/**
+ * Executes a Lua function in this application thread.
+ */
+int
+app_thread_lua_call(const char *name, uint32_t name_len,
+		    struct port *args, struct port *ret);
+
+/**
+ * Executes a Lua expression in this application thread.
+ */
+int
+app_thread_lua_eval(const char *expr, uint32_t expr_len,
+		    struct port *args, struct port *ret);
+
+#if defined(__cplusplus)
+} /* extern "C" */
+#endif /* defined(__cplusplus) */