core: fix crashes after altering trigger list while it is run

This patch fixes a number of issues with trigger_clear() while the
trigger list is being run:
1) clearing the next-to-be-run trigger doesn't prevent it from being run
2) clearing the next-to-be-run trigger causes an infinite loop or a
   crash
3) swapping trigger list head before the last trigger is run causes an
   infinite loop or a crash (see space_swap_triggers() in alter.cc, which
   had worked all this time by miracle: space _space on_replace trigger
   swaps its own head during local recovery, and that had only worked
   because the trigger by luck was the last to run)

This is fixed by adding triggers in a separate run list on trigger_run.
This list may be iterated by `rlist_shift_entry`, which doesn't suffer
from any of the problems mentioned above.

While being bad in a number of ways, old approach supported practically
unlimited number of concurrent trigger_runs for the same trigger list.
The new approach requires the trigger to be in as many run lists as
there are concurrent trigger_runs, which results in quite a big
refactoring.

Add a luatest-based test and a unit test.

Closes #4264

NO_DOC=bugfix

changelogs/unreleased/gh-4264-trigger-crash-on-nontrivial-change.md

+## bugfix/core
+
+* Fixed crashes or undefined behaviour with triggers clearing other triggers
+  (gh-4264).

src/lib/core/fiber.c

@@ -1525,6 +1525,7 @@ cord_create(struct cord *cord, const char *name)
 	}
 	cord_set_name(name);
 
+	trigger_init_in_thread();
 #if ENABLE_ASAN
 	/* Record stack extents */
 	tt_pthread_attr_getstack(cord->id, &cord->sched.stack,
@@ -1565,6 +1566,7 @@ cord_exit(struct cord *cord)
 {
 	assert(cord == cord());
 	(void)cord;
+	trigger_free_in_thread();
 }
 
 void

src/lib/core/trigger.cc

@@ -30,10 +30,9 @@
  */
 
 #include "trigger.h"
-
 #include "fiber.h"
-
 #include <small/region.h>
+#include <small/mempool.h>
 
 static void
 trigger_fiber_run_timeout(ev_loop *loop, ev_timer *watcher, int revents)
@@ -54,30 +53,135 @@ trigger_fiber_f(va_list ap)
 	return trigger->run(trigger, event);
 }
 
+/**
+ * A single link in a list of triggers scheduled for execution. Can't be part of
+ * struct trigger, because one trigger might be part of an unlimited number of
+ * run lists.
+ */
+struct run_link {
+	/** A link in the run list belonging to one trigger_run() call.  */
+	struct rlist in_run;
+	/**
+	 * A link in a list of all links referencing that trigger. Used to
+	 * remove the trigger from whatever list it's on during trigger_clear().
+	 */
+	struct rlist in_trigger;
+	/** The trigger to be executed. */
+	struct trigger *trigger;
+};
+
+/** A memory pool for trigger run link allocation. */
+static __thread struct mempool run_link_pool;
+
+/** Add the trigger to the run list. */
+static int
+run_list_put_trigger(struct rlist *list, struct trigger *trigger)
+{
+	struct run_link *run_link =
+		(struct run_link *)mempool_alloc(&run_link_pool);
+	if (run_link == NULL) {
+		diag_set(OutOfMemory, sizeof(struct run_link),
+			 "mempool_alloc", "trigger run link");
+		return -1;
+	}
+	run_link->trigger = trigger;
+	rlist_add_tail_entry(list, run_link, in_run);
+	rlist_add_tail_entry(&trigger->run_links, run_link, in_trigger);
+	return 0;
+}
+
+/** Take the next trigger from the run list and free the corresponding link. */
+static struct trigger *
+run_list_take_trigger(struct rlist *list)
+{
+	struct run_link *link = rlist_shift_entry(list, struct run_link,
+						  in_run);
+	struct trigger *trigger = link->trigger;
+	rlist_del_entry(link, in_trigger);
+	mempool_free(&run_link_pool, link);
+	return trigger;
+}
+
+/** Empty the run list and free all the links allocated for it. */
+static void
+run_list_clear(struct rlist *list)
+{
+	while (!rlist_empty(list))
+		run_list_take_trigger(list);
+}
+
+/** Execute the triggers in an order specified by \a list. */
+static int
+trigger_run_list(struct rlist *list, void *event)
+{
+	while (!rlist_empty(list)) {
+		struct trigger *trigger = run_list_take_trigger(list);
+		if (trigger->run(trigger, event) != 0) {
+			run_list_clear(list);
+			return -1;
+		}
+	}
+	return 0;
+}
+
 int
 trigger_run(struct rlist *list, void *event)
 {
-	struct trigger *trigger, *tmp;
-	rlist_foreach_entry_safe(trigger, list, link, tmp)
-		if (trigger->run(trigger, event) != 0)
+	/*
+	 * A list holding all triggers scheduled for execution. It's important
+	 * to save all triggers in a separate run_list and iterate over it
+	 * instead of the passed list, because the latter might change anyhow
+	 * while the triggers are run:
+	 *  * the current element might be removed from the list
+	 *    (rlist_foreach_entry_safe helps with that)
+	 *  * the next element might be removed from the list
+	 *    (rlist_foreach_entry_safe fails to handle that, but plain
+	 *    rlist_foreach_entry works just fine)
+	 *  * trigger lists might be swapped (for example see
+	 *    space_swap_triggers() in alter.cc), in which case neither
+	 *    rlist_foreach_entry nor rlist_foreach_entry_safe can help
+	 */
+	RLIST_HEAD(run_list);
+	struct trigger *trigger;
+	rlist_foreach_entry(trigger, list, link) {
+		if (run_list_put_trigger(&run_list, trigger) != 0) {
+			run_list_clear(&run_list);
 			return -1;
-	return 0;
+		}
+	}
+	return trigger_run_list(&run_list, event);
 }
 
 int
 trigger_run_reverse(struct rlist *list, void *event)
 {
-	struct trigger *trigger, *tmp;
-	rlist_foreach_entry_safe_reverse(trigger, list, link, tmp)
-		if (trigger->run(trigger, event) != 0)
+	RLIST_HEAD(run_list);
+	struct trigger *trigger;
+	rlist_foreach_entry_reverse(trigger, list, link) {
+		if (run_list_put_trigger(&run_list, trigger) != 0) {
+			run_list_clear(&run_list);
 			return -1;
-	return 0;
+		}
+	}
+	return trigger_run_list(&run_list, event);
+}
+
+void
+trigger_clear(struct trigger *trigger)
+{
+	rlist_del_entry(trigger, link);
+	struct run_link *link, *tmp;
+	rlist_foreach_entry_safe(link, &trigger->run_links, in_trigger, tmp) {
+		rlist_del_entry(link, in_run);
+		rlist_del_entry(link, in_trigger);
+		mempool_free(&run_link_pool, link);
+	}
 }
 
 int
 trigger_fiber_run(struct rlist *list, void *event, double timeout)
 {
-	struct trigger *trigger, *tmp;
+	struct trigger *trigger;
 	unsigned trigger_count = 0;
 	struct region *region = &fiber()->gc;
 	RegionGuard guard(region);
@@ -95,6 +199,14 @@ trigger_fiber_run(struct rlist *list, void *event, double timeout)
 		return -1;
 	}
 
+	RLIST_HEAD(run_list);
+	rlist_foreach_entry(trigger, list, link) {
+		if (run_list_put_trigger(&run_list, trigger) != 0) {
+			run_list_clear(&run_list);
+			return -1;
+		}
+	}
+
 	bool expired = false;
 	struct ev_timer timer;
 	ev_timer_init(&timer, trigger_fiber_run_timeout, timeout, 0);
@@ -108,7 +220,8 @@ trigger_fiber_run(struct rlist *list, void *event, double timeout)
 	ev_timer_start(loop(), &timer);
 
 	unsigned current_fiber = 0;
-	rlist_foreach_entry_safe(trigger, list, link, tmp) {
+	while (!rlist_empty(&run_list)) {
+		trigger = run_list_take_trigger(&run_list);
 		char name[FIBER_NAME_INLINE];
 		snprintf(name, FIBER_NAME_INLINE,
 			 "trigger_fiber%d", current_fiber);
@@ -119,6 +232,7 @@ trigger_fiber_run(struct rlist *list, void *event, double timeout)
 			current_fiber++;
 		} else {
 			ev_timer_stop(loop(), &timer);
+			run_list_clear(&run_list);
 			return -1;
 		}
 	}
@@ -140,3 +254,15 @@ trigger_fiber_run(struct rlist *list, void *event, double timeout)
 	ev_timer_stop(loop(), &timer);
 	return 0;
 }
+
+void
+trigger_init_in_thread(void)
+{
+	mempool_create(&run_link_pool, &cord()->slabc, sizeof(struct run_link));
+}
+
+void
+trigger_free_in_thread(void)
+{
+	mempool_destroy(&run_link_pool);
+}

src/lib/core/trigger.h

@@ -45,7 +45,10 @@ typedef void (*trigger_f0)(struct trigger *trigger);
 
 struct trigger
 {
+	/** A link in a list of all registered triggers. */
 	struct rlist link;
+	/** All execution lists which this trigger is part of. */
+	struct rlist run_links;
 	trigger_f run;
 	/**
 	 * Lua ref in case the trigger is used in Lua,
@@ -59,21 +62,23 @@ struct trigger
 	trigger_f0 destroy;
 };
 
-#define TRIGGER_INITIALIZER(run) {\
+#define TRIGGER_INITIALIZER(name, run) {\
 	RLIST_LINK_INITIALIZER,\
+	RLIST_HEAD_INITIALIZER((name).run_links),\
 	(run),\
 	NULL,\
 	NULL\
 }
 
 #define TRIGGER(name, run)\
-	struct trigger name = TRIGGER_INITIALIZER(run)
+	struct trigger name = TRIGGER_INITIALIZER(name, run)
 
 static inline void
 trigger_create(struct trigger *trigger, trigger_f run, void *data,
 	       trigger_f0 destroy)
 {
 	rlist_create(&trigger->link);
+	rlist_create(&trigger->run_links);
 	trigger->run = run;
 	trigger->data = data;
 	trigger->destroy = destroy;
@@ -105,12 +110,9 @@ trigger_add_unique(struct rlist *list, struct trigger *trigger)
 	trigger_add(list, trigger);
 }
 
-static inline void
-trigger_clear(struct trigger *trigger)
-{
-	rlist_del_entry(trigger, link);
-}
-
+/** Remove the trigger from any list it's in. */
+void
+trigger_clear(struct trigger *trigger);
 
 static inline void
 trigger_destroy(struct rlist *list)
@@ -152,6 +154,14 @@ trigger_run_reverse(struct rlist *list, void *event);
 int
 trigger_fiber_run(struct rlist *list, void *event, double timeout);
 
+/** Initialize the thread-dependent part of trigger subsystem. */
+void
+trigger_init_in_thread(void);
+
+/** Free the thread-dependent part of trigger subsystem. */
+void
+trigger_free_in_thread(void);
+
 #if defined(__cplusplus)
 } /* extern "C" */
 

test/box-luatest/gh_4264_recursive_trigger_test.lua

 local t = require('luatest')
 local server = require('test.luatest_helpers.server')
+local fio = require('fio')
 
 local g = t.group('gh-4264')
 
@@ -41,3 +42,66 @@ g.test_recursive_trigger_invocation = function(cg)
     t.assert_equals(order, {11, 21, 31, 32, 22, 12},
                     "Correct recursive trigger invocation")
 end
+
+g.test_trigger_clear_from_trigger = function(cg)
+    local order = cg.server:exec(function()
+        local order = {}
+        local s = box.space.test
+        local f2 = function()
+            table.insert(order, 2)
+        end
+        local f1
+        f1 = function()
+            table.insert(order, 1)
+            s:on_replace(nil, f2)
+            s:on_replace(nil, f1)
+        end
+        s:on_replace(f2)
+        s:on_replace(f1)
+        s:replace{2}
+        return order
+    end)
+    t.assert_equals(order, {1}, "Correct trigger invocation when 1st trigger \
+                                 clears 2nd")
+end
+
+local g2 = t.group('gh-4264-on-shutdown')
+
+g2.before_test('test_on_shutdown_trigger_clear', function(cg)
+    cg.server = server:new{alias = 'server'}
+    cg.server:start()
+end)
+
+g2.after_test('test_on_shutdown_trigger_clear', function(cg)
+    cg.server:cleanup()
+end)
+
+g2.test_on_shutdown_trigger_clear = function(cg)
+    cg.server:exec(function()
+        local log = require('log')
+        local f2 = function()
+            log.info('trigger 2 executed')
+        end
+        local f1
+        f1 = function()
+            log.info('trigger 1 executed')
+            box.ctl.on_shutdown(nil, f2)
+            box.ctl.on_shutdown(nil, f1)
+        end
+        box.ctl.on_shutdown(f2)
+        box.ctl.on_shutdown(f1)
+        -- Killing the server with luatest doesn't trigger the issue. Need
+        -- os.exit for that.
+        require('fiber').new(os.exit)
+    end)
+
+    local logfile = fio.pathjoin(cg.server.workdir, 'server.log')
+    t.helpers.retrying({}, function()
+        local found = cg.server:grep_log('trigger 1 executed', nil,
+                                         {filename = logfile})
+        t.assert(found ~= nil, 'first on_shutdown trigger is executed')
+    end)
+    local found = cg.server:grep_log('trigger 2 executed', nil,
+                                     {filename = logfile})
+    t.assert(found == nil, 'second on_shutdown trigger is not executed')
+end

test/unit/CMakeLists.txt

@@ -104,6 +104,8 @@ add_executable(prbuf.test prbuf.c)
 target_link_libraries(prbuf.test unit core)
 add_executable(clock_lowres.test clock_lowres.c core_test_utils.c)
 target_link_libraries(clock_lowres.test unit core)
+add_executable(trigger.test trigger.c core_test_utils.c)
+target_link_libraries(trigger.test unit core)
 
 if (NOT ENABLE_GCOV)
     # This test is known to be broken with GCOV

test/unit/trigger.c

+#define UNIT_TAP_COMPATIBLE 1
+#include "unit.h"
+#include <stdbool.h>
+#include <assert.h>
+#include "memory.h"
+#include "fiber.h"
+#include "trigger.h"
+
+/* Length of trigger chains to test. */
+#define TEST_LENGTH 5
+#define FUNC_COUNT (TEST_LENGTH + 3)
+
+struct test_trigger {
+	struct trigger base;
+	int id;
+	int target_id;
+};
+
+/* How many times each trigger was run. */
+static int was_run[TEST_LENGTH];
+/* Function codes for each trigger. */
+static int funcs[TEST_LENGTH];
+static struct test_trigger triggers[TEST_LENGTH];
+
+static RLIST_HEAD(list_a);
+static RLIST_HEAD(list_b);
+
+static int
+trigger_nop_f(struct trigger *trigger, void *event)
+{
+	struct test_trigger *trig = (struct test_trigger *)trigger;
+	was_run[trig->id]++;
+	return 0;
+}
+
+static int
+trigger_err_f(struct trigger *trigger, void *event)
+{
+	trigger_nop_f(trigger, event);
+	return -1;
+}
+
+static int
+trigger_swap_f(struct trigger *trigger, void *event)
+{
+	trigger_nop_f(trigger, event);
+	rlist_swap(&list_a, &list_b);
+	return 0;
+}
+
+static int
+trigger_clear_f(struct trigger *trigger, void *event)
+{
+	trigger_nop_f(trigger, event);
+	struct test_trigger *trig = (struct test_trigger *)trigger;
+	int clear_id = trig->target_id;
+	trigger_clear(&triggers[clear_id].base);
+	return 0;
+}
+
+/**
+ * Types of trigger functions which might harm trigger_run one way or another.
+ */
+enum func_type {
+	/** Do nothing. */
+	FUNC_TYPE_NOP,
+	/** Error. */
+	FUNC_TYPE_ERR,
+	/** Swap trigger list heads. */
+	FUNC_TYPE_SWAP,
+	/** Clear one of the triggers: self or other. */
+	FUNC_TYPE_CLEAR,
+};
+
+static int
+func_type_by_no(int func_no)
+{
+	assert(func_no >= 0 && func_no < FUNC_COUNT);
+	if (func_no < TEST_LENGTH)
+		return FUNC_TYPE_CLEAR;
+	else if (func_no == TEST_LENGTH)
+		return FUNC_TYPE_ERR;
+	else if (func_no == TEST_LENGTH + 1)
+		return FUNC_TYPE_NOP;
+	else /* if (func_no == TEST_LENGTH + 2) */
+		return FUNC_TYPE_SWAP;
+}
+
+static void
+fill_trigger_list(struct rlist *list, bool *should_run, int direction)
+{
+	for (int i = 0; i < TEST_LENGTH; i++)
+		should_run[i] = true;
+	rlist_create(list);
+	/*
+	 * Create triggers and deduce which should and which shouldn't run based
+	 * on trigger functions and run direction.
+	 */
+	for (int i = (TEST_LENGTH - 1) * (1 - direction) / 2;
+	     i >= 0 && i < TEST_LENGTH; i += direction) {
+		was_run[i] = 0;
+		triggers[i].id = i;
+		int func_no = funcs[i];
+		switch (func_type_by_no(func_no)) {
+		case FUNC_TYPE_CLEAR:
+			trigger_create(&triggers[i].base,
+				       trigger_clear_f, NULL, NULL);
+			triggers[i].target_id = func_no;
+			if (!should_run[i])
+				break;
+			int target_id = funcs[i];
+			if (direction * i < direction * target_id)
+				should_run[target_id] = false;
+			break;
+		case FUNC_TYPE_ERR:
+			trigger_create(&triggers[i].base, trigger_err_f,
+				       NULL, NULL);
+			if (!should_run[i])
+				break;
+			for (int j = i + direction; j >= 0 && j < TEST_LENGTH;
+			     j += direction) {
+				should_run[j] = false;
+			}
+			break;
+		case FUNC_TYPE_NOP:
+			trigger_create(&triggers[i].base, trigger_nop_f,
+				       NULL, NULL);
+			break;
+		case FUNC_TYPE_SWAP:
+			trigger_create(&triggers[i].base,
+				       trigger_swap_f, NULL, NULL);
+			break;
+		}
+	}
+	/*
+	 * Add triggers in reverse order, so that trigger[0] is first to run in
+	 * direct order and last to run in reverse order.
+	 */
+	for (int i = TEST_LENGTH - 1; i >= 0; i--)
+		trigger_add(list, &triggers[i].base);
+}
+
+static void
+test_trigger_one(void)
+{
+	plan(2 * TEST_LENGTH);
+	for (int direction = -1; direction <= 1; direction += 2) {
+		bool should_run[TEST_LENGTH];
+		fill_trigger_list(&list_a, should_run, direction);
+		if (direction == 1)
+			trigger_run(&list_a, NULL);
+		else
+			trigger_run_reverse(&list_a, NULL);
+		for (int i = 0; i < TEST_LENGTH; i++) {
+			ok((should_run[i] && was_run[i] == 1) ||
+			   (!should_run[i] && was_run[i] == 0),
+			   "Triggers ran correctly");
+		}
+	}
+	check_plan();
+}
+
+static void
+test_trigger(int pos, bool had_clear)
+{
+	if (pos == TEST_LENGTH)
+		return test_trigger_one();
+	int i = had_clear ? TEST_LENGTH : 0;
+	plan(FUNC_COUNT - i);
+	for (i; i < FUNC_COUNT; i++) {
+		funcs[pos] = i;
+		test_trigger(pos + 1, had_clear || func_type_by_no(i) ==
+						   FUNC_TYPE_CLEAR);
+	}
+	check_plan();
+}
+
+static int
+test_trigger_clear_during_run(void)
+{
+	header();
+	plan(1);
+
+	test_trigger(0, false);
+
+	footer();
+	return check_plan();
+}
+
+int
+main(void)
+{
+	memory_init();
+	fiber_init(fiber_c_invoke);
+
+	plan(1);
+	test_trigger_clear_during_run();
+
+	fiber_free();
+	memory_free();
+	return check_plan();
+}