Collective host task order-dependencies must respect horizons

When a new horizon is applied, tracked last collective group task ids must be replaced by the previous horizon.
This follows the same rule as buffer last-writers.

src/graph_generator.cc

@@ -494,6 +494,10 @@ namespace detail {
 						return {std::max(prev_hid, *cid)};
 					});
 				}
+				for(auto& [nid_cgid, cid] : last_collective_commands) {
+					const auto [nid, cgid] = nid_cgid;
+					if(nid == node) { cid = std::max(prev_hid, cid); }
+				}
 				// update lowest previous horizon id (for later command deletion)
 				if(lowest_prev_hid == 0) {
 					lowest_prev_hid = prev_hid;

src/task_manager.cc

@@ -215,6 +215,9 @@ namespace detail {
 						return {std::max(prev_hid, *tid)};
 					});
 				}
+				for(auto& [cgid, tid] : last_collective_tasks) {
+					tid = std::max(prev_hid, tid);
+				}
 
 				// We also use the previous horizon as the new init task for host-initialized buffers
 				current_init_task_id = prev_hid;

test/graph_compaction_tests.cc

@@ -333,5 +333,48 @@ namespace detail {
 		maybe_print_graphs(ctx);
 	}
 
+	TEST_CASE("commands for collective host tasks do not order-depend on their predecessor if it is shadowed by a horizon",
+	    "[graph_generator][command-graph][horizon]") {
+		// Regression test: the order-dependencies between host tasks in the same collective group are built by tracking the last task command in each
+		// collective group. Once a horizon is inserted, commands for new collective host tasks must order-depend on that horizon command instead.
+
+		const size_t num_nodes = 1;
+		test_utils::cdag_test_context ctx(num_nodes);
+		auto& tm = ctx.get_task_manager();
+		tm.set_horizon_step(2);
+
+		const auto first_collective = test_utils::build_and_flush(ctx, test_utils::add_host_task(tm, experimental::collective, [&](handler& cgh) {}));
+
+		// generate exactly two horizons
+		auto& ggen = ctx.get_graph_generator();
+		test_utils::mock_buffer_factory mbf(&tm, &ggen);
+		auto buf = mbf.create_buffer(range<1>(1));
+		for(int i = 0; i < 5; ++i) {
+			test_utils::build_and_flush(
+			    ctx, test_utils::add_host_task(tm, on_master_node, [&](handler& cgh) { buf.get_access<access_mode::discard_write>(cgh, all{}); }));
+		}
+
+		// This must depend on the first horizon, not first_collective
+		const auto second_collective = test_utils::build_and_flush(ctx, test_utils::add_host_task(tm, experimental::collective, [&](handler& cgh) {}));
+
+		const auto& inspector = ctx.get_inspector();
+		auto& cdag = ctx.get_command_graph();
+		const auto first_commands = inspector.get_commands(first_collective, std::nullopt, std::nullopt);
+		const auto second_commands = inspector.get_commands(second_collective, std::nullopt, std::nullopt);
+		for(const auto second_cid : second_commands) {
+			for(const auto first_cid : first_commands) {
+				CHECK(!inspector.has_dependency(second_cid, first_cid));
+			}
+			const auto second_deps = cdag.get(second_cid)->get_dependencies();
+			CHECK(std::distance(second_deps.begin(), second_deps.end()) == 1);
+			for(const auto& dep : second_deps) {
+				CHECK(dep.kind == dependency_kind::ORDER_DEP);
+				CHECK(dynamic_cast<const horizon_command*>(dep.node));
+			}
+		}
+
+		maybe_print_graphs(ctx);
+	}
+
 } // namespace detail
 } // namespace celerity

test/task_graph_tests.cc

@@ -505,5 +505,33 @@ namespace detail {
 		maybe_print_graph(tm);
 	}
 
+	TEST_CASE("collective host tasks do not order-depend on their predecessor if it is shadowed by a horizon", "[task_manager][task-graph][task-horizon]") {
+		// Regression test: the order-dependencies between host tasks in the same collective group are built by tracking the last task in each collective group.
+		// Once a horizon is inserted, new collective host tasks must order-depend on that horizon instead.
+
+		task_manager tm{1, nullptr, nullptr};
+		tm.set_horizon_step(2);
+
+		const auto first_collective = test_utils::add_host_task(tm, experimental::collective, [&](handler& cgh) {});
+
+		// generate exactly two horizons
+		test_utils::mock_buffer_factory mbf(&tm);
+		auto buf = mbf.create_buffer(range<1>(1));
+		for(int i = 0; i < 5; ++i) {
+			test_utils::add_host_task(tm, on_master_node, [&](handler& cgh) { buf.get_access<access_mode::discard_write>(cgh, all{}); });
+		}
+
+		// This must depend on the first horizon, not first_collective
+		const auto second_collective = test_utils::add_host_task(tm, experimental::collective, [&](handler& cgh) {});
+
+		for(const auto& dep : tm.get_task(second_collective)->get_dependencies()) {
+			const auto type = dep.node->get_type();
+			CHECK(type == task_type::HORIZON);
+			CHECK(dep.kind == dependency_kind::ORDER_DEP);
+		}
+
+		maybe_print_graph(tm);
+	}
+
 } // namespace detail
 } // namespace celerity

test/test_utils.h

@@ -121,7 +121,7 @@ namespace test_utils {
 			return result;
 		}
 
-		bool has_dependency(detail::command_id dependent, detail::command_id dependency) {
+		bool has_dependency(detail::command_id dependent, detail::command_id dependency) const {
 			const auto& deps = commands.at(dependent).dependencies;
 			return std::find(deps.cbegin(), deps.cend(), dependency) != deps.cend();
 		}