added celerity blockchain for task divergence checking

CMakeLists.txt

@@ -187,6 +187,7 @@ set(SOURCES
   src/command_graph.cc
   src/config.cc
   src/device_queue.cc
+  src/divergence_block_chain.cc
   src/executor.cc
   src/distributed_graph_generator.cc
   src/graph_serializer.cc

include/divergence_block_chain.h

@@ -0,0 +1,179 @@
+#pragma once
+
+#include "recorders.h"
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace celerity::detail {
+// in c++23 replace this with mdspan
+template <typename T>
+struct mpi_2d_send_wrapper {
+  public:
+	const T& operator[](std::pair<int, int> ij) const {
+		assert(ij.first * m_width + ij.second < m_data.size());
+		return m_data[ij.first * m_width + ij.second];
+	}
+
+	T* data() { return m_data.data(); }
+
+	mpi_2d_send_wrapper(size_t width, size_t height) : m_data(width * height), m_width(width){};
+
+  private:
+	std::vector<T> m_data;
+	const size_t m_width;
+};
+
+// Probably replace this in c++20 with span
+template <typename T>
+struct window {
+  public:
+	window(const std::vector<T>& value) : m_value(value) {}
+
+	const T& operator[](size_t i) const {
+		assert(i >= 0 && i < m_width);
+		return m_value[m_offset + i];
+	}
+
+	size_t size() {
+		m_width = m_value.size() - m_offset;
+		return m_width;
+	}
+
+	void slide(size_t i) {
+		assert(i == 0 || (i >= 0 && i <= m_width));
+		m_offset += i;
+		m_width -= i;
+	}
+
+  private:
+	const std::vector<T>& m_value;
+	size_t m_offset = 0;
+	size_t m_width = 0;
+};
+
+using task_hash = size_t;
+using task_hash_data = mpi_2d_send_wrapper<task_hash>;
+using divergence_map = std::unordered_map<task_hash, std::vector<node_id>>;
+
+class abstract_block_chain {
+	friend struct abstract_block_chain_testspy;
+
+  public:
+	virtual void stop() { m_is_running = false; };
+
+	abstract_block_chain(const abstract_block_chain&) = delete;
+	abstract_block_chain& operator=(const abstract_block_chain&) = delete;
+	abstract_block_chain& operator=(abstract_block_chain&&) = delete;
+
+	abstract_block_chain(abstract_block_chain&&) = default;
+
+	abstract_block_chain(size_t num_nodes, node_id local_nid, const std::vector<task_record>& task_recorder, MPI_Comm comm)
+	    : m_local_nid(local_nid), m_num_nodes(num_nodes), m_sizes(num_nodes), m_task_recorder_window(task_recorder), m_comm(comm) {}
+
+	virtual ~abstract_block_chain() = default;
+
+  protected:
+	void start() { m_is_running = true; };
+
+	virtual void run() = 0;
+
+	virtual void divergence_out(const divergence_map& check_map, const int task_num) = 0;
+
+	void add_new_hashes();
+	void clear(const int min_progress);
+	virtual void allgather_sizes();
+	virtual void allgather_hashes(const int max_size, task_hash_data& data);
+	std::pair<int, int> collect_sizes();
+	task_hash_data collect_hashes(const int max_size);
+	divergence_map create_check_map(const task_hash_data& task_graphs, const int task_num) const;
+
+	void check_for_deadlock() const;
+
+	static void print_node_divergences(const divergence_map& check_map, const int task_num);
+
+	static void print_task_record(const divergence_map& check_map, const task_record& task, const task_hash hash);
+
+	virtual void dedub_print_task_record(const divergence_map& check_map, const int task_num) const;
+
+	bool check_for_divergence();
+
+  protected:
+	node_id m_local_nid;
+	size_t m_num_nodes;
+
+	std::vector<task_hash> m_hashes;
+	std::vector<int> m_sizes;
+
+	bool m_is_running = true;
+
+	window<task_record> m_task_recorder_window;
+
+	std::chrono::time_point<std::chrono::steady_clock> m_last_cleared = std::chrono::steady_clock::now();
+
+	MPI_Comm m_comm;
+};
+
+class single_node_test_divergence_block_chain : public abstract_block_chain {
+  public:
+	single_node_test_divergence_block_chain(size_t num_nodes, node_id local_nid, const std::vector<task_record>& task_recorder, MPI_Comm comm,
+	    const std::vector<std::reference_wrapper<const std::vector<task_record>>>& other_task_records)
+	    : abstract_block_chain(num_nodes, local_nid, task_recorder, comm), m_other_hashes(other_task_records.size()) {
+		for(auto& tsk_rcd : other_task_records) {
+			m_other_task_records.push_back(window<task_record>(tsk_rcd));
+		}
+	}
+
+  private:
+	void run() override {}
+
+	void divergence_out(const divergence_map& check_map, const int task_num) override;
+	void allgather_sizes() override;
+	void allgather_hashes(const int max_size, task_hash_data& data) override;
+
+	void dedub_print_task_record(const divergence_map& check_map, const int task_num) const override;
+
+	std::vector<std::vector<task_hash>> m_other_hashes;
+	std::vector<window<task_record>> m_other_task_records;
+
+	int m_injected_delete_size = 0;
+};
+
+class distributed_test_divergence_block_chain : public abstract_block_chain {
+  public:
+	distributed_test_divergence_block_chain(size_t num_nodes, node_id local_nid, const std::vector<task_record>& task_record, MPI_Comm comm)
+	    : abstract_block_chain(num_nodes, local_nid, task_record, comm) {}
+
+  private:
+	void run() override {}
+
+	void divergence_out(const divergence_map& check_map, const int task_num) override;
+};
+
+class divergence_block_chain : public abstract_block_chain {
+  public:
+	void start();
+	void stop() override;
+
+	divergence_block_chain(size_t num_nodes, node_id local_nid, const std::vector<task_record>& task_record, MPI_Comm comm)
+	    : abstract_block_chain(num_nodes, local_nid, task_record, comm) {
+		divergence_block_chain::start();
+	}
+
+	divergence_block_chain(const divergence_block_chain&) = delete;
+	divergence_block_chain& operator=(const divergence_block_chain&) = delete;
+	divergence_block_chain& operator=(divergence_block_chain&&) = delete;
+
+	divergence_block_chain(divergence_block_chain&&) = default;
+
+	~divergence_block_chain() override { divergence_block_chain::stop(); }
+
+  private:
+	void run() override;
+
+	void divergence_out(const divergence_map& check_map, const int task_num) override;
+
+  private:
+	std::thread m_thread;
+};
+} // namespace celerity::detail

include/grid.h

@@ -8,6 +8,7 @@
 #include <gch/small_vector.hpp>
 
 #include "ranges.h"
+#include "utils.h"
 #include "workaround.h"
 
 namespace celerity::detail {
@@ -257,6 +258,27 @@ class region {
 
 } // namespace celerity::detail
 
+template <int Dims>
+struct std::hash<celerity::detail::box<Dims>> {
+	std::size_t operator()(const celerity::detail::box<Dims> r) {
+		std::size_t seed = 0;
+		celerity::detail::utils::hash_combine(seed, std::hash<celerity::id<Dims>>{}(r.get_min()), std::hash<celerity::id<Dims>>{}(r.get_max()));
+		return seed;
+	};
+};
+
+template <int Dims>
+struct std::hash<celerity::detail::region<Dims>> {
+	std::size_t operator()(const celerity::detail::region<Dims> r) {
+		std::size_t seed = 0;
+		for(auto box : r.get_boxes()) {
+			celerity::detail::utils::hash_combine(seed, std::hash<celerity::detail::box<Dims>>{}(box));
+		}
+		return seed;
+	};
+};
+
+
 namespace celerity::detail::grid_detail {
 
 // forward-declaration for tests (explicitly instantiated)

include/print_utils.h

@@ -70,4 +70,4 @@ struct fmt::formatter<celerity::chunk<Dims>> : fmt::formatter<celerity::subrange
 		out = formatter<celerity::id<Dims>>::format(celerity::id(chunk.global_size), ctx); // cast to id to avoid multiple inheritance
 		return out;
 	}
-};
+};

include/ranges.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "sycl_wrappers.h"
+#include "utils.h"
 #include "workaround.h"
 
 namespace celerity {
@@ -229,6 +230,17 @@ struct ones_t {
 
 }; // namespace celerity::detail
 
+template <typename Interface, int Dims>
+struct std::hash<celerity::detail::coordinate<Interface, Dims>> {
+	std::size_t operator()(const celerity::detail::coordinate<Interface, Dims>& r) const noexcept {
+		std::size_t seed = 0;
+		for(int i = 0; i < Dims; ++i) {
+			celerity::detail::utils::hash_combine(seed, std::hash<int>{}(r[i]));
+		}
+		return seed;
+	};
+};
+
 namespace celerity {
 
 template <int Dims>
@@ -401,6 +413,17 @@ nd_range(range<3> global_range, range<3> local_range)->nd_range<3>;
 
 } // namespace celerity
 
+
+template <int Dims>
+struct std::hash<celerity::range<Dims>> {
+	std::size_t operator()(const celerity::range<Dims>& r) const noexcept { return std::hash<celerity::detail::coordinate<celerity::range<Dims>, Dims>>{}(r); };
+};
+
+template <int Dims>
+struct std::hash<celerity::id<Dims>> {
+	std::size_t operator()(const celerity::id<Dims>& r) const noexcept { return std::hash<celerity::detail::coordinate<celerity::id<Dims>, Dims>>{}(r); };
+};
+
 namespace celerity {
 namespace detail {
 

include/recorders.h

@@ -114,3 +114,85 @@ class command_recorder {
 };
 
 } // namespace celerity::detail
+
+template <>
+struct std::hash<celerity::detail::reduction_record> {
+	std::size_t operator()(const celerity::detail::reduction_record& r) const noexcept {
+		std::size_t seed = 0;
+		celerity::detail::utils::hash_combine(seed, std::hash<celerity::detail::reduction_id>{}(r.rid), std::hash<celerity::detail::buffer_id>{}(r.bid),
+		    std::hash<std::string>{}(r.buffer_name), std::hash<bool>{}(r.init_from_buffer));
+		return seed;
+	};
+};
+
+template <>
+struct std::hash<celerity::detail::access_record> {
+	std::size_t operator()(const celerity::detail::access_record& r) {
+		std::size_t seed = 0;
+		celerity::detail::utils::hash_combine(seed, std::hash<celerity::detail::buffer_id>{}(r.bid), std::hash<std::string>{}(r.buffer_name),
+		    std::hash<celerity::access_mode>{}(r.mode), std::hash<celerity::detail::region<3>>{}(r.req));
+		return seed;
+	};
+};
+
+template <typename IdType>
+struct std::hash<celerity::detail::dependency_record<IdType>> {
+	std::size_t operator()(const celerity::detail::dependency_record<IdType>& r) const noexcept {
+		std::size_t seed = 0;
+		celerity::detail::utils::hash_combine(seed, std::hash<IdType>{}(r.node), std::hash<celerity::detail::dependency_kind>{}(r.kind),
+		    std::hash<celerity::detail::dependency_origin>{}(r.origin));
+		return seed;
+	};
+};
+
+template <>
+struct std::hash<celerity::detail::side_effect_map> {
+	std::size_t operator()(const celerity::detail::side_effect_map& m) const noexcept {
+		std::size_t seed = 0;
+		for(auto& [hoid, order] : m) {
+			celerity::detail::utils::hash_combine(
+			    seed, std::hash<celerity::detail::host_object_id>{}(hoid), std::hash<celerity::experimental::side_effect_order>{}(order));
+		}
+		return seed;
+	};
+};
+
+template <>
+struct std::hash<celerity::detail::task_record> {
+	std::size_t operator()(const celerity::detail::task_record& t) const noexcept {
+		std::size_t seed = 0;
+		celerity::detail::utils::hash_combine(seed, std::hash<celerity::detail::task_id>{}(t.tid), std::hash<std::string>{}(t.debug_name),
+		    std::hash<celerity::detail::collective_group_id>{}(t.cgid), std::hash<celerity::detail::task_type>{}(t.type),
+		    std::hash<celerity::detail::task_geometry>{}(t.geometry), celerity::detail::utils::vector_hash{}(t.reductions),
+		    celerity::detail::utils::vector_hash{}(t.accesses), std::hash<celerity::detail::side_effect_map>{}(t.side_effect_map),
+		    celerity::detail::utils::vector_hash{}(t.dependencies));
+
+		return seed;
+	};
+};
+
+template <>
+struct fmt::formatter<celerity::detail::dependency_kind> : fmt::formatter<std::string> {
+	static format_context::iterator format(const celerity::detail::dependency_kind& dk, format_context& ctx) {
+		auto out = ctx.out();
+		switch(dk) {
+		case celerity::detail::dependency_kind::anti_dep: out = std::copy_n("anti-dep", 8, out); break;
+		case celerity::detail::dependency_kind::true_dep: out = std::copy_n("true-dep", 8, out); break;
+		}
+		return out;
+	}
+};
+
+template <>
+struct fmt::formatter<celerity::detail::dependency_origin> : fmt::formatter<std::string> {
+	static format_context::iterator format(const celerity::detail::dependency_origin& dk, format_context& ctx) {
+		auto out = ctx.out();
+		switch(dk) {
+		case celerity::detail::dependency_origin::dataflow: out = std::copy_n("dataflow", 8, out); break;
+		case celerity::detail::dependency_origin::collective_group_serialization: out = std::copy_n("collective-group-serialization", 31, out); break;
+		case celerity::detail::dependency_origin::execution_front: out = std::copy_n("execution-front", 15, out); break;
+		case celerity::detail::dependency_origin::last_epoch: out = std::copy_n("last-epoch", 10, out); break;
+		}
+		return out;
+	}
+};

include/runtime.h

@@ -7,6 +7,7 @@
 #include "command.h"
 #include "config.h"
 #include "device_queue.h"
+#include "divergence_block_chain.h"
 #include "frame.h"
 #include "host_queue.h"
 #include "recorders.h"
@@ -101,6 +102,8 @@ namespace detail {
 		size_t m_num_nodes;
 		node_id m_local_nid;
 
+		std::unique_ptr<abstract_block_chain> m_divergence_check;
+
 		// These management classes are only constructed on the master node.
 		std::unique_ptr<command_graph> m_cdag;
 		std::unique_ptr<scheduler> m_schdlr;