iproto: don't hang on uncancellable iproto request

Currently if there is uncancellable iproto request Tarantool shutdown
will hang. Let's instead give it some time and then panic.

On this way it is good to make iproto_drop_connections() fail on timeout.
It is used in `box.ctl.iproto_lockdown` which is better to fail on
timeout than to hang indefinitely too.

In Tarantool CI which is run with TEST_BUILD set, we set the timeout to
the infinity. This is on par with current fiber_shutdown() behaviour.
We will not change the latter for a while because there is already
several tests that count on that. Also it is currently easier to test
that there is no hang than to test exit status.

Part of tarantool#8423

NO_CHANGELOG=internal
NO_DOC=internal

Author: nshy

src/box/box.cc

@@ -100,6 +100,7 @@
 #include "tt_sort.h"
 #include "event.h"
 #include "func_adapter.h"
+#include "tweaks.h"
 
 static char status[64] = "unconfigured";
 
@@ -246,6 +247,22 @@ static char box_feedback_host[BOX_FEEDBACK_HOST_MAX];
 /** Whether sending crash info to feedback URL is enabled. */
 static bool box_feedback_crash_enabled;
 
+#ifdef TEST_BUILD
+/**
+ * Set timeout to infinity in test build because first not all CI tests treat
+ * non zero exit code of Tarantool instance as failure currently. Also in
+ * luatest currently it is easier to test for no hanging rather then for
+ * Tarantool instance exit code.
+ */
+#define BOX_SHUTDOWN_TIMEOUT_DEFAULT TIMEOUT_INFINITY
+#else
+#define BOX_SHUTDOWN_TIMEOUT_DEFAULT 3.0
+#endif
+
+/** Timeout on waiting client related fibers to finish. */
+static double box_shutdown_timeout = BOX_SHUTDOWN_TIMEOUT_DEFAULT;
+TWEAK_DOUBLE(box_shutdown_timeout);
+
 static int
 box_run_on_recovery_state(enum box_recovery_state state)
 {
@@ -5976,7 +5993,10 @@ box_storage_shutdown()
 	if (!is_storage_initialized)
 		return;
 	is_storage_shutdown = true;
-	iproto_shutdown();
+	if (iproto_shutdown(box_shutdown_timeout) != 0) {
+		diag_log();
+		panic("cannot gracefully shutdown iproto");
+	}
 	box_watcher_shutdown();
 	/*
 	 * Finish client fibers after iproto_shutdown otherwise new fibers

src/box/iproto.cc

@@ -134,6 +134,25 @@ iproto_wpos_create(struct iproto_wpos *wpos, struct obuf *out)
 	wpos->svp = obuf_create_svp(out);
 }
 
+/**
+ * Message sent when iproto thread dropped all connections that requested
+ * to be dropped.
+ */
+struct iproto_drop_finished {
+	/** Base structure. */
+	struct cmsg base;
+	/**
+	 * Generation a is a sequence number of iproto_drop_connections()
+	 * invocation.
+	 *
+	 * Generation is used to handle racy situation when previous invocation
+	 * of iproto_drop_connections() was failed and there is new invocation.
+	 * Message from previous invocation may be delivired and account
+	 * iproto thread as finished dropping connection which is not true.
+	 */
+	unsigned generation;
+};
+
 struct iproto_thread {
 	/**
 	 * Slab cache used for allocating memory for output network buffers
@@ -213,7 +232,7 @@ struct iproto_thread {
 	 * Message used to notify TX thread that all connections marked
 	 * to de dropped are dropped.
 	 */
-	struct cmsg drop_finished_msg;
+	struct iproto_drop_finished drop_finished_msg;
 	/**
 	 * If set then iproto thread shutdown is started and we should not
 	 * accept new connections.
@@ -236,6 +255,12 @@ struct iproto_thread {
 static struct fiber_cond drop_finished_cond;
 /** Count of iproto threads that are not finished connections drop yet. */
 static size_t drop_pending_thread_count;
+/**
+ * Generation is a sequence number of dropping connection invocation.
+ *
+ * See also `struct iproto_drop_finished`.
+ */
+static unsigned drop_generation;
 
 /**
  * IPROTO listen URIs. Set by box.cfg.listen.
@@ -411,6 +436,13 @@ struct iproto_cfg_msg: public cbus_call_msg
 			 * NULL if the function is called not from connection.
 			 */
 			struct iproto_connection *owner;
+			/**
+			 * Generation is sequence number of dropping
+			 * connection invocation.
+			 *
+			 * See also `struct iproto_drop_finished`.
+			 */
+			unsigned generation;
 		} drop_connections;
 	};
 	struct iproto_thread *iproto_thread;
@@ -875,6 +907,12 @@ struct iproto_connection
 	struct rlist in_connections;
 	/** Set if connection is being dropped. */
 	bool is_drop_pending;
+	/**
+	 * Generation is sequence number of dropping connection invocation.
+	 *
+	 * See also `struct iproto_drop_finished`.
+	 */
+	unsigned drop_generation;
 	/**
 	 * Messaged sent to TX to cancel all inprogress requests of the
 	 * connection.
@@ -1642,21 +1680,25 @@ iproto_connection_new(struct iproto_thread *iproto_thread)
 static void
 tx_process_drop_finished(struct cmsg *m)
 {
-	(void)m;
-	assert(drop_pending_thread_count > 0);
-	if (--drop_pending_thread_count == 0)
+	struct iproto_drop_finished *drop_finished =
+					(struct iproto_drop_finished *)m;
+	if (drop_finished->generation == drop_generation &&
+	    --drop_pending_thread_count == 0)
 		fiber_cond_signal(&drop_finished_cond);
 }
 
 /** Send message to TX thread to notify that connections drop is finished. */
 static void
-iproto_send_drop_finished(struct iproto_thread *iproto_thread)
+iproto_send_drop_finished(struct iproto_thread *iproto_thread,
+			  unsigned generation)
 {
 	static const struct cmsg_hop drop_finished_route[1] =
 					{{ tx_process_drop_finished, NULL }};
 
-	cmsg_init(&iproto_thread->drop_finished_msg, drop_finished_route);
-	cpipe_push(&iproto_thread->tx_pipe, &iproto_thread->drop_finished_msg);
+	cmsg_init(&iproto_thread->drop_finished_msg.base, drop_finished_route);
+	iproto_thread->drop_finished_msg.generation = generation;
+	cpipe_push(&iproto_thread->tx_pipe,
+		   &iproto_thread->drop_finished_msg.base);
 }
 
 /** Recycle a connection. */
@@ -1684,7 +1726,8 @@ iproto_connection_delete(struct iproto_connection *con)
 
 		assert(iproto_thread->drop_pending_connection_count > 0);
 		if (--iproto_thread->drop_pending_connection_count == 0)
-			iproto_send_drop_finished(iproto_thread);
+			iproto_send_drop_finished(iproto_thread,
+						  con->drop_generation);
 	}
 	mempool_free(&con->iproto_thread->iproto_connection_pool, con);
 }
@@ -3937,6 +3980,7 @@ iproto_do_cfg_f(struct cbus_call_msg *m)
 		struct iproto_connection *con;
 		static const struct cmsg_hop cancel_route[1] =
 				{{ tx_process_cancel_inprogress, NULL }};
+		iproto_thread->drop_pending_connection_count = 0;
 		rlist_foreach_entry(con, &iproto_thread->connections,
 				    in_connections) {
 			/*
@@ -3954,6 +3998,8 @@ iproto_do_cfg_f(struct cbus_call_msg *m)
 			 */
 			if (con != cfg_msg->drop_connections.owner) {
 				con->is_drop_pending = true;
+				con->drop_generation =
+					cfg_msg->drop_connections.generation;
 				iproto_thread->drop_pending_connection_count++;
 			}
 			if (con->state != IPROTO_CONNECTION_DESTROYED) {
@@ -3963,7 +4009,9 @@ iproto_do_cfg_f(struct cbus_call_msg *m)
 			}
 		}
 		if (iproto_thread->drop_pending_connection_count == 0)
-			iproto_send_drop_finished(iproto_thread);
+			iproto_send_drop_finished(
+				iproto_thread,
+				cfg_msg->drop_connections.generation);
 		break;
 	}
 	default:
@@ -4024,27 +4072,34 @@ iproto_send_start_msg(void)
 		iproto_do_cfg(&iproto_threads[i], &cfg_msg);
 }
 
-void
-iproto_drop_connections(void)
+int
+iproto_drop_connections(double timeout)
 {
 	static struct latch latch = LATCH_INITIALIZER(latch);
 	latch_lock(&latch);
 	struct iproto_connection *owner = NULL;
 	struct session *session = fiber_get_session(fiber());
 	if (session != NULL && session->type == SESSION_TYPE_BINARY)
 		owner = (struct iproto_connection *)session->meta.connection;
+	drop_generation++;
 	drop_pending_thread_count = iproto_threads_count;
 	for (int i = 0; i < iproto_threads_count; i++) {
 		struct iproto_cfg_msg *cfg_msg =
 			(struct iproto_cfg_msg *)xmalloc(sizeof(*cfg_msg));
 		iproto_cfg_msg_create(cfg_msg, IPROTO_CFG_DROP_CONNECTIONS);
 		cfg_msg->drop_connections.owner = owner;
+		cfg_msg->drop_connections.generation = drop_generation;
 		iproto_do_cfg_async(&iproto_threads[i], cfg_msg);
 	}
 
-	while (drop_pending_thread_count != 0)
-		fiber_cond_wait(&drop_finished_cond);
+	double deadline = ev_monotonic_now(loop()) + timeout;
+	while (drop_pending_thread_count != 0) {
+		if (fiber_cond_wait_deadline(&drop_finished_cond,
+					     deadline) != 0)
+			break;
+	}
 	latch_unlock(&latch);
+	return drop_pending_thread_count == 0 ? 0 : -1;
 }
 
 /** Send IPROTO_CFG_RESTART to all threads. */
@@ -4233,11 +4288,11 @@ iproto_session_send(struct session *session,
 	return 0;
 }
 
-void
-iproto_shutdown(void)
+int
+iproto_shutdown(double timeout)
 {
 	assert(iproto_is_shutting_down);
-	iproto_drop_connections();
+	return iproto_drop_connections(timeout);
 }
 
 void

src/box/iproto.h

@@ -193,16 +193,30 @@ iproto_free(void);
  * Drop all current connections. That is stop IO and cancel all inprogress
  * requests. Return when the requests are finished and connection is freed.
  * Concurrent calls are serialized.
+ *
+ * Drop can be interrupted by cancelling fiber or on timeout. In this case
+ * failure result code is returned. The function can be called again after
+ * failure.
+ *
+ * Return:
+ *   0 - success
+ *  -1 - failure (diag is set)
  */
-void
-iproto_drop_connections(void);
+int
+iproto_drop_connections(double timeout);
 
 /**
  * Prepare for freeing resources in iproto_free while TX event loop is still
  * running.
+ *
+ * If not finished in given timeout then failure result code is returned.
+ *
+ * Return:
+ *   0 - success
+ *  -1 - failure (diag is set)
  */
-void
-iproto_shutdown(void);
+int
+iproto_shutdown(double timeout);
 
 #if defined(__cplusplus)
 } /* extern "C" */

src/box/lua/ctl.c

@@ -164,7 +164,8 @@ lbox_ctl_set_iproto_lockdown(struct lua_State *L)
 		lua_error(L);
 	}
 	bool new_val = lua_toboolean(L, 1);
-	security_set_iproto_lockdown(new_val);
+	if (security_set_iproto_lockdown(new_val) != 0)
+		return luaT_error(L);
 #else
 	lua_pushstring(L, "box.ctl.iproto_lockdown() is available only in "
 		       "Enterprise Edition builds.");

test/box-luatest/shutdown_test.lua

@@ -2,6 +2,7 @@ local server = require('luatest.server')
 local utils = require('luatest.utils')
 local fio = require('fio')
 local popen = require('popen')
+local fiber = require('fiber')
 local t = require('luatest')
 
 -- Luatest server currently does not allow to check process exit code.
@@ -72,3 +73,49 @@ g_crash.test_shutdown_during_snapshot_on_signal = function(cg)
     t.assert_equals(status.state, 'exited')
     t.assert_equals(status.exit_code, 0)
 end
+
+local g = t.group()
+
+g.before_each(function(cg)
+    cg.server = server:new()
+    cg.server:start()
+end)
+
+g.after_each(function(cg)
+    if cg.server ~= nil then
+        cg.server:drop()
+    end
+end)
+
+local function test_no_hang_on_shutdown(server)
+    local channel = fiber.channel()
+    fiber.create(function()
+        server:stop()
+        channel:put('finished')
+    end)
+    t.assert(channel:get(60) ~= nil)
+end
+
+-- Test shutdown does not hang if there is iproto request that
+-- cannot be cancelled.
+g.test_shutdown_of_hanging_iproto_request = function(cg)
+    fiber.new(function()
+        cg.server:exec(function()
+            local log = require('log')
+            local fiber = require('fiber')
+            local tweaks = require('internal.tweaks')
+            tweaks.box_shutdown_timeout = 1.0
+            log.info('going to sleep for test')
+            while true do
+                pcall(fiber.sleep, 1000)
+            end
+        end)
+    end)
+    t.helpers.retrying({}, function()
+        t.assert(cg.server:grep_log('going to sleep for test'))
+    end)
+    local log = fio.pathjoin(cg.server.workdir, cg.server.alias .. '.log')
+    test_no_hang_on_shutdown(cg.server)
+    t.assert(cg.server:grep_log('cannot gracefully shutdown iproto', nil,
+             {filename = log}))
+end