11610324419

Committed 31 Oct 2024 10:34AM UTC coverage: 95.251% (+0.002%) from 95.249%

Build # 11610324419

Build Type

Pull #297

github

Committed by

fknorr

Commit Message

Refactor command_graph

command_graph is now an epoch_partitioned_graph, and information only
relevant for building the CDAG (like the execution front) is kept in the
CDAG generator instead.

`class abstract_command` is renamed to `class command`, constructors are
public and `get_type` member is gone, whose only remaining use was the
Tracy integration.

CDAG pruning is moved out to the scheduler, identical to IDAG pruning.

The CDAG generator maintains command pointers instead of looking up
commands by id.

Pull Request Pull Request #297: Refactor command_graph

Run Details

3004 of 3390 branches covered (88.61%)

Branch coverage included in aggregate %.

153 of 153 new or added lines in 6 files covered. (100.0%)

4 existing lines in 1 file now uncovered.

6663 of 6759 relevant lines covered (98.58%)

1479155.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.4

/src/command_graph_generator.cc

#include "command_graph_generator.h"

#include "access_modes.h"
#include "command_graph.h"
#include "grid.h"
#include "intrusive_graph.h"
#include "print_utils.h"
#include "ranges.h"
#include "recorders.h"
#include "split.h"
#include "task.h"
#include "types.h"

#include <algorithm>
#include <cstddef>


namespace celerity::detail {

command_graph_generator::command_graph_generator(
    const size_t num_nodes, const node_id local_nid, command_graph& cdag, detail::command_recorder* const recorder, const policy_set& policy)
    : m_num_nodes(num_nodes), m_local_nid(local_nid), m_policy(policy), m_cdag(&cdag), m_recorder(recorder) {
        if(m_num_nodes > max_num_nodes) {
                throw std::runtime_error(fmt::format("Number of nodes requested ({}) exceeds compile-time maximum of {}", m_num_nodes, max_num_nodes));
        }
}

void command_graph_generator::notify_buffer_created(const buffer_id bid, const range<3>& range, bool host_initialized) {
        assert(m_epoch_for_new_commands != nullptr);
        assert(!m_buffers.contains(bid));
        // Mark contents as available locally (= don't generate await push commands) and fully replicated (= don't generate push commands).
        // This is required when tasks access host-initialized or uninitialized buffers.
        auto& buffer = m_buffers.emplace(std::piecewise_construct, std::tuple(bid), std::tuple(range, m_epoch_for_new_commands, node_bitset().set())).first->second;
        if(host_initialized && m_policy.uninitialized_read_error != error_policy::ignore) { buffer.initialized_region = box(subrange({}, range)); }
}

void command_graph_generator::notify_buffer_debug_name_changed(const buffer_id bid, const std::string& debug_name) {
        m_buffers.at(bid).debug_name = debug_name;
}

void command_graph_generator::notify_buffer_destroyed(const buffer_id bid) {
        assert(m_buffers.contains(bid));
        m_buffers.erase(bid);
}

void command_graph_generator::notify_host_object_created(const host_object_id hoid) {
        assert(m_epoch_for_new_commands != nullptr);
        assert(!m_host_objects.contains(hoid));
        m_host_objects.emplace(hoid, m_epoch_for_new_commands);
}

void command_graph_generator::notify_host_object_destroyed(const host_object_id hoid) {
        assert(m_host_objects.contains(hoid));
        m_host_objects.erase(hoid);
}

/// Returns whether an iterator range of commands is topologically sorted, i.e. sequential execution would satisfy all internal dependencies.
template <typename Iterator>
bool is_topologically_sorted(Iterator begin, Iterator end) {
        for(auto check = begin; check != end; ++check) {
                for(const auto dep : (*check)->get_dependencies()) {
                        if(std::find_if(std::next(check), end, [dep](const auto& node) { return node == dep.node; }) != end) return false;
                }
        }
        return true;
}

std::vector<const command*> command_graph_generator::build_task(const task& tsk) {
        const auto epoch_to_prune_before = m_epoch_for_new_commands;
        batch current_batch;

        switch(tsk.get_type()) {
        case task_type::epoch: generate_epoch_command(current_batch, tsk); break;
        case task_type::horizon: generate_horizon_command(current_batch, tsk); break;
        case task_type::device_compute:
        case task_type::host_compute:
        case task_type::master_node:
        case task_type::collective:
        case task_type::fence: generate_distributed_commands(current_batch, tsk); break;
        default: throw std::runtime_error("Task type NYI");
        }

        // It is currently undefined to split reduction-producer tasks into multiple chunks on the same node:
        //   - Per-node reduction intermediate results are stored with fixed access to a single backing buffer,
        //     so multiple chunks on the same node will race on this buffer access
        //   - Inputs to the final reduction command are ordered by origin node ids to guarantee bit-identical results. It is not possible to distinguish
        //     more than one chunk per node in the serialized commands, so different nodes can produce different final reduction results for non-associative
        //     or non-commutative operations
        if(!tsk.get_reductions().empty()) {
                assert(std::count_if(current_batch.begin(), current_batch.end(), [](const command* cmd) { return utils::isa<task_command>(cmd); }) <= 1);
        }

        // If a new epoch was completed in the CDAG before the current task, we can erase all tracking information from earlier commands.
        // After the epoch (or horizon) command has been executed, the scheduler will then delete all obsolete commands from the CDAG.
        if(epoch_to_prune_before != nullptr) {
                std::erase_if(m_command_buffer_reads, [=](const auto& cid_reads) { return cid_reads.first < epoch_to_prune_before->get_id(); });
        }

        // Check that all commands have been recorded
        if(is_recording()) {
                assert(std::all_of(current_batch.begin(), current_batch.end(), [this](const command* cmd) {
                        return std::any_of(m_recorder->get_graph_nodes().begin(), m_recorder->get_graph_nodes().end(),
                            [cmd](const std::unique_ptr<command_record>& rec) { return rec->id == cmd->get_id(); });
                }));
        }

        assert(is_topologically_sorted(current_batch.begin(), current_batch.end()));
        return current_batch;
}

void command_graph_generator::report_overlapping_writes(const task& tsk, const box_vector<3>& local_chunks) const {
        const chunk<3> full_chunk{tsk.get_global_offset(), tsk.get_global_size(), tsk.get_global_size()};

        // Since this check is run distributed on every node, we avoid quadratic behavior by only checking for conflicts between all local chunks and the
        // region-union of remote chunks. This way, every conflict will be reported by at least one node.
        const box<3> global_chunk(subrange(full_chunk.offset, full_chunk.range));
        auto remote_chunks = region_difference(global_chunk, region(box_vector<3>(local_chunks))).into_boxes();

        // detect_overlapping_writes takes a single box_vector, so we concatenate local and global chunks (the order does not matter)
        auto distributed_chunks = std::move(remote_chunks);
        distributed_chunks.insert(distributed_chunks.end(), local_chunks.begin(), local_chunks.end());

        if(const auto overlapping_writes = detect_overlapping_writes(tsk, distributed_chunks); !overlapping_writes.empty()) {
                auto error = fmt::format("{} has overlapping writes between multiple nodes in", print_task_debug_label(tsk, true /* title case */));
                for(const auto& [bid, overlap] : overlapping_writes) {
                        fmt::format_to(std::back_inserter(error), " {} {}", print_buffer_debug_label(bid), overlap);
                }
                error += ". Choose a non-overlapping range mapper for this write access or constrain the split via experimental::constrain_split to make the access "
                         "non-overlapping.";
                utils::report_error(m_policy.overlapping_write_error, "{}", error);
        }
}

std::vector<command_graph_generator::assigned_chunk> command_graph_generator::split_task_and_assign_chunks(const task& tsk) const {
        const chunk<3> full_chunk{tsk.get_global_offset(), tsk.get_global_size(), tsk.get_global_size()};
        const size_t num_chunks = m_num_nodes * m_test_chunk_multiplier;
        const auto chunks = ([&] {
                if(tsk.get_type() == task_type::collective || tsk.get_type() == task_type::fence) {
                        std::vector<chunk<3>> chunks;
                        for(size_t nid = 0; nid < m_num_nodes; ++nid) {
                                chunks.push_back(chunk_cast<3>(chunk<1>{id<1>{tsk.get_type() == task_type::collective ? nid : 0}, ones, {m_num_nodes}}));
                        }
                        return chunks;
                }
                if(tsk.has_variable_split()) {
                        if(tsk.get_hint<experimental::hints::split_1d>() != nullptr) {
                                // no-op, keeping this for documentation purposes
                        }
                        if(tsk.get_hint<experimental::hints::split_2d>() != nullptr) { return split_2d(full_chunk, tsk.get_granularity(), num_chunks); }
                        return split_1d(full_chunk, tsk.get_granularity(), num_chunks);
                }
                return std::vector<chunk<3>>{full_chunk};
        })();
        assert(chunks.size() <= num_chunks); // We may have created less than requested
        assert(!chunks.empty());

        // Assign each chunk to a node
        // We assign chunks next to each other to the same worker (if there is more chunks than workers), as this is likely to produce less
        // transfers between tasks than a round-robin assignment (for typical stencil codes).
        // FIXME: This only works if the number of chunks is an integer multiple of the number of workers, e.g. 3 chunks for 2 workers degrades to RR.
        const auto chunks_per_node = std::max<size_t>(1, chunks.size() / m_num_nodes);

        std::vector<assigned_chunk> assigned_chunks;
        for(size_t i = 0; i < chunks.size(); ++i) {
                const node_id nid = (i / chunks_per_node) % m_num_nodes;
                assigned_chunks.push_back({nid, chunks[i]});
        }
        return assigned_chunks;
}

command_graph_generator::buffer_requirements_list command_graph_generator::get_buffer_requirements_for_mapped_access(
    const task& tsk, const subrange<3>& sr, const range<3> global_size) const {
        buffer_requirements_list result;
        const auto& access_map = tsk.get_buffer_access_map();
        for(const buffer_id bid : access_map.get_accessed_buffers()) {
                buffer_requirements reqs{bid, {}, {}};
                for(const auto m : access_map.get_access_modes(bid)) {
                        if(detail::access::mode_traits::is_consumer(m)) {
                                reqs.consumed = region_union(reqs.consumed, access_map.get_mode_requirements(bid, m, tsk.get_dimensions(), sr, global_size));
                        }
                        // Not else: `access_mode::write` is both a consumer and a producer
                        if(detail::access::mode_traits::is_producer(m)) {
                                reqs.produced = region_union(reqs.produced, access_map.get_mode_requirements(bid, m, tsk.get_dimensions(), sr, global_size));
                        }
                }
                result.emplace_back(std::move(reqs));
        }
        return result;
}

const box<3> empty_reduction_box({0, 0, 0}, {0, 0, 0});
const box<3> scalar_reduction_box({0, 0, 0}, {1, 1, 1});

command_graph_generator::assigned_chunks_with_requirements command_graph_generator::compute_per_chunk_requirements(
    const task& tsk, const std::vector<assigned_chunk>& assigned_chunks) const {
        assigned_chunks_with_requirements result;

        for(const auto& a_chunk : assigned_chunks) {
                const node_id nid = a_chunk.executed_on;
                auto requirements = get_buffer_requirements_for_mapped_access(tsk, a_chunk.chnk, tsk.get_global_size());

                // Add read/write requirements for reductions performed in this task.
                for(const auto& reduction : tsk.get_reductions()) {
                        // task_manager verifies that there are no reduction <-> write-access conflicts
                        assert(std::none_of(
                            requirements.begin(), requirements.end(), [&](const buffer_requirements& br) { return br.bid == reduction.bid && !br.produced.empty(); }));
                        auto it = std::find_if(requirements.begin(), requirements.end(), [&](const buffer_requirements& br) { return br.bid == reduction.bid; });
                        if(it == requirements.end()) { it = requirements.insert(requirements.end(), buffer_requirements{reduction.bid, {}, {}}); }
                        it->produced = scalar_reduction_box;
                        if(nid == reduction_initializer_nid && reduction.init_from_buffer) { it->consumed = scalar_reduction_box; }
                }

                if(nid == m_local_nid) {
                        result.local_chunks.emplace_back(a_chunk, std::move(requirements));
                } else {
                        result.remote_chunks.emplace_back(a_chunk, std::move(requirements));
                }
        }

        return result;
}

void command_graph_generator::resolve_pending_reductions(
    batch& current_batch, const task& tsk, const assigned_chunks_with_requirements& chunks_with_requirements) {
        auto accessed_buffers = tsk.get_buffer_access_map().get_accessed_buffers();
        // Handle chained reductions (i.e., reductions that combine into a buffer that currently is in a pending reduction state)
        for(const auto& reduction : tsk.get_reductions()) {
                accessed_buffers.insert(reduction.bid);
        }

        for(const auto bid : accessed_buffers) {
                auto& buffer = m_buffers.at(bid);
                if(!buffer.pending_reduction.has_value()) { continue; }
                const auto& reduction = *buffer.pending_reduction;

                const auto local_last_writer_set = buffer.local_last_writer.get_region_values(scalar_reduction_box);
                assert(local_last_writer_set.size() == 1);
                const auto local_last_writer = local_last_writer_set[0].second;

                // Prepare the buffer state for after the reduction has been performed:
                // Keep the current last writer, but mark it as stale, so that if we don't generate a reduction command locally,
                // we'll know to get the data from elsewhere. If we generate a reduction command, this will be overwritten by its command id.
                auto wcs = local_last_writer;
                wcs.mark_as_stale();
                buffer_state post_reduction_state(ones, wcs, node_bitset());
                if(m_policy.uninitialized_read_error != error_policy::ignore) { post_reduction_state.initialized_region = scalar_reduction_box; }

                node_bitset participating_nodes;

                // Since the local reduction command overwrites the buffer contents that need to be pushed to other nodes, we need to process remote chunks first.
                for(const auto& [a_chunk, requirements] : chunks_with_requirements.remote_chunks) {
                        if(std::none_of(requirements.begin(), requirements.end(), [&](const buffer_requirements& br) { return br.bid == bid && !br.consumed.empty(); })) {
                                // This chunk doesn't read from the buffer
                                continue;
                        }
                        participating_nodes.set(a_chunk.executed_on);
                }

                // Generate push command to all participating nodes
                if(participating_nodes.any()) {
                        // Push an empty range if we don't have any fresh data on this node. This will then generate an empty pilot that tells the
                        // other node's receive_arbiter to not expect a send.
                        const bool notification_only = !local_last_writer.is_fresh();
                        const auto push_box = notification_only ? empty_reduction_box : scalar_reduction_box;
                        assert(participating_nodes.count() == m_num_nodes - 1 || participating_nodes.count() == 1);
                        std::vector<std::pair<node_id, region<3>>> regions;
                        for(node_id nid = 0; nid < m_num_nodes; ++nid) {
                                if(!participating_nodes.test(nid)) continue;
                                regions.push_back({nid, push_box});
                        }
                        auto* const cmd = create_command<push_command>(current_batch, transfer_id(tsk.get_id(), bid, reduction.rid), std::move(regions),
                            [&, bid = bid](const auto& record_debug_info) { record_debug_info(m_buffers.at(bid).debug_name); });
                        if(notification_only) {
                                generate_epoch_dependencies(cmd);
                        } else {
                                m_command_buffer_reads[cmd->get_id()][bid] = region_union(m_command_buffer_reads[cmd->get_id()][bid], scalar_reduction_box);
                                add_dependency(cmd, local_last_writer, dependency_kind::true_dep, dependency_origin::dataflow);
                        }

                        // Mark the reduction result as replicated so we don't generate data transfers to any of the participating nodes
                        post_reduction_state.replicated_regions.update_box(scalar_reduction_box, participating_nodes);
                }

                // We currently don't support multiple chunks on a single node for reductions (there is also -- for now -- no way to create multiple chunks,
                // as oversubscription is handled by the instruction graph).
                // NOTE: The participating_nodes.count() check below relies on this being true
                assert(chunks_with_requirements.local_chunks.size() <= 1);
                for(const auto& [a_chunk, requirements] : chunks_with_requirements.local_chunks) {
                        if(std::none_of(requirements.begin(), requirements.end(), [&](const buffer_requirements& br) { return br.bid == bid && !br.consumed.empty(); })) {
                                // This chunk doesn't read from the buffer
                                continue;
                        }

                        auto* const ap_cmd = create_command<await_push_command>(current_batch, transfer_id(tsk.get_id(), bid, reduction.rid),
                            scalar_reduction_box.get_subrange(), [&](const auto& record_debug_info) { record_debug_info(buffer.debug_name); });
                        generate_epoch_dependencies(ap_cmd);

                        auto* const reduce_cmd = create_command<reduction_command>(current_batch, reduction, local_last_writer.is_fresh() /* has_local_contribution */,
                            [&](const auto& record_debug_info) { record_debug_info(buffer.debug_name); });

                        // Only generate a true dependency on the last writer if this node participated in the intermediate result computation.
                        if(local_last_writer.is_fresh()) { add_dependency(reduce_cmd, local_last_writer, dependency_kind::true_dep, dependency_origin::dataflow); }
                        add_dependency(reduce_cmd, ap_cmd, dependency_kind::true_dep, dependency_origin::dataflow);
                        generate_anti_dependencies(tsk, bid, buffer.local_last_writer, scalar_reduction_box, reduce_cmd);

                        post_reduction_state.local_last_writer.update_box(scalar_reduction_box, reduce_cmd);
                        participating_nodes.set(m_local_nid); // We are participating
                }

                // We currently do not support generating reduction commands on only a subset of nodes, except for the special case of a single command.
                // This is because it is unclear who owns the final result in this case (normally all nodes "own" the result).
                //   => I.e., reducing and using the result on the participating nodes is actually not the problem (this works as intended); the issue only arises
                //      if the result is subsequently required in other tasks. Since we don't have a good way of detecting this condition however, we currently
                //      disallow partial reductions altogether.
                // NOTE: This check relies on the fact that we currently only generate a single chunk per node for reductions (see assertion above).
                if(participating_nodes.count() > 1 && participating_nodes.count() != m_num_nodes) {
                        utils::report_error(error_policy::panic,
                            "{} requires a reduction on {} that is not performed on all nodes. This is currently not supported. Either "
                            "ensure that all nodes receive a chunk that reads from the buffer, or reduce the data on a single node.",
                            print_task_debug_label(tsk, true /* title case */), print_buffer_debug_label(bid));
                }

                // For buffers that were in a pending reduction state and a reduction was generated
                // (i.e., the result was not discarded), set their new state.
                if(participating_nodes.count() > 0) {
                        m_completed_reductions.push_back(reduction.rid);
                        buffer = std::move(post_reduction_state);
                }
        }
}

void command_graph_generator::generate_pushes(batch& current_batch, const task& tsk, const assigned_chunks_with_requirements& chunks_with_requirements) {
        struct push_scratch {
                std::unordered_map<node_id, region<3>> target_regions;
                std::unordered_set<command*> depends_on;
        };
        std::unordered_map<buffer_id, push_scratch> per_buffer_pushes;

        for(auto& [a_chunk, requirements] : chunks_with_requirements.remote_chunks) {
                const node_id nid = a_chunk.executed_on;

                for(const auto& [bid, consumed, _] : requirements) {
                        if(consumed.empty()) continue;
                        auto& buffer = m_buffers.at(bid);

                        const auto local_sources = buffer.local_last_writer.get_region_values(consumed);
                        for(const auto& [local_box, wcs] : local_sources) {
                                if(!wcs.is_fresh() || wcs.is_replicated()) { continue; }

                                // Make sure we don't push anything we've already pushed to this node before
                                box_vector<3> non_replicated_boxes;
                                for(const auto& [replicated_box, nodes] : buffer.replicated_regions.get_region_values(local_box)) {
                                        if(nodes.test(nid)) continue;
                                        non_replicated_boxes.push_back(replicated_box);
                                }

                                if(!non_replicated_boxes.empty()) {
                                        assert(!utils::isa<await_push_command>(wcs.get_command()) && "Attempting to push non-owned data?!");
                                        auto push_region = region<3>(std::move(non_replicated_boxes));
                                        // Remember that we've replicated this region
                                        for(const auto& [replicated_box, nodes] : buffer.replicated_regions.get_region_values(push_region)) {
                                                buffer.replicated_regions.update_box(replicated_box, node_bitset{nodes}.set(nid));
                                        }
                                        auto& scratch = per_buffer_pushes[bid]; // allow default-insert
                                        if(auto it = scratch.target_regions.find(nid); it != scratch.target_regions.end()) {
                                                it->second = region_union(it->second, push_region);
                                        } else {
                                                scratch.target_regions.emplace(nid, std::move(push_region));
                                        }
                                        scratch.depends_on.insert(wcs);
                                }
                        }
                }
        }

        // Generate push command for each buffer
        for(auto& [bid, scratch] : per_buffer_pushes) {
                region<3> combined_region;
                std::vector<std::pair<node_id, region<3>>> target_regions;
                for(auto& [nid, region] : scratch.target_regions) {
                        combined_region = region_union(combined_region, region);
                        target_regions.push_back({nid, std::move(region)});
                }

                auto* const cmd = create_command<push_command>(current_batch, transfer_id(tsk.get_id(), bid, no_reduction_id), std::move(target_regions),
                    [&, bid = bid](const auto& record_debug_info) { record_debug_info(m_buffers.at(bid).debug_name); });
                for(const auto dep : scratch.depends_on) {
                        add_dependency(cmd, dep, dependency_kind::true_dep, dependency_origin::dataflow);
                }

                // Store the read access for determining anti-dependencies
                m_command_buffer_reads[cmd->get_id()].emplace(bid, std::move(combined_region));
        }
}

// TODO: We currently generate an await push command for each local chunk, whereas we only generate a single push command for all remote chunks
void command_graph_generator::generate_await_pushes(batch& current_batch, const task& tsk, const assigned_chunks_with_requirements& chunks_with_requirements) {
        for(auto& [a_chunk, requirements] : chunks_with_requirements.local_chunks) {
                for(auto& [bid, consumed, _] : requirements) {
                        if(consumed.empty()) continue;
                        auto& buffer = m_buffers.at(bid);

                        const auto local_sources = buffer.local_last_writer.get_region_values(consumed);
                        box_vector<3> missing_part_boxes;
                        for(const auto& [box, wcs] : local_sources) {
                                // Note that we initialize all buffers as fresh, so this doesn't trigger for uninitialized reads
                                if(!box.empty() && !wcs.is_fresh()) { missing_part_boxes.push_back(box); }
                        }

                        // There is data we don't yet have locally. Generate an await push command for it.
                        if(!missing_part_boxes.empty()) {
                                const region missing_parts(std::move(missing_part_boxes));
                                assert(m_num_nodes > 1);
                                auto* const ap_cmd = create_command<await_push_command>(current_batch, transfer_id(tsk.get_id(), bid, no_reduction_id), missing_parts,
                                    [&](const auto& record_debug_info) { record_debug_info(buffer.debug_name); });
                                generate_anti_dependencies(tsk, bid, buffer.local_last_writer, missing_parts, ap_cmd);
                                generate_epoch_dependencies(ap_cmd);
                                // Remember that we have this data now
                                buffer.local_last_writer.update_region(missing_parts, {ap_cmd, true /* is_replicated */});
                        }
                }
        }
}

void command_graph_generator::update_local_buffer_fresh_regions(const task& tsk, const std::unordered_map<buffer_id, region<3>>& per_buffer_local_writes) {
        auto requirements = get_buffer_requirements_for_mapped_access(tsk, subrange<3>(tsk.get_global_offset(), tsk.get_global_size()), tsk.get_global_size());
        // Add requirements for reductions
        for(const auto& reduction : tsk.get_reductions()) {
                auto it = std::find_if(requirements.begin(), requirements.end(), [&](const buffer_requirements& br) { return br.bid == reduction.bid; });
                if(it == requirements.end()) { it = requirements.insert(requirements.end(), buffer_requirements{reduction.bid, {}, {}}); }
                it->produced = scalar_reduction_box;
        }
        for(auto& [bid, _, produced] : requirements) {
                region global_writes = produced;
                auto& buffer = m_buffers.at(bid);
                if(m_policy.uninitialized_read_error != error_policy::ignore) { buffer.initialized_region = region_union(buffer.initialized_region, global_writes); }

                const auto remote_writes = ([&, bid = bid] {
                        if(auto it = per_buffer_local_writes.find(bid); it != per_buffer_local_writes.end()) {
                                const auto& local_writes = it->second;
                                assert(region_difference(local_writes, global_writes).empty()); // Local writes have to be a subset of global writes
                                return region_difference(global_writes, local_writes);
                        }
                        return std::move(global_writes);
                })(); // IIFE

                // TODO: We need a way of updating regions in place! E.g. apply_to_values(box, callback)
                auto boxes_and_cids = buffer.local_last_writer.get_region_values(remote_writes);
                for(auto& [box, wcs] : boxes_and_cids) {
                        if(wcs.is_fresh()) {
                                wcs.mark_as_stale();
                                buffer.local_last_writer.update_region(box, wcs);
                        }
                }
        }
}

void command_graph_generator::generate_distributed_commands(batch& current_batch, const task& tsk) {
        const auto chunks = split_task_and_assign_chunks(tsk);
        const auto chunks_with_requirements = compute_per_chunk_requirements(tsk, chunks);

        // Check for and report overlapping writes between local chunks, and between local and remote chunks.
        if(m_policy.overlapping_write_error != error_policy::ignore) {
                box_vector<3> local_chunks;
                for(const auto& [a_chunk, _] : chunks_with_requirements.local_chunks) {
                        local_chunks.push_back(box<3>{a_chunk.chnk});
                }
                report_overlapping_writes(tsk, local_chunks);
        }

        resolve_pending_reductions(current_batch, tsk, chunks_with_requirements);
        generate_pushes(current_batch, tsk, chunks_with_requirements);
        generate_await_pushes(current_batch, tsk, chunks_with_requirements);

        // Union of all per-buffer writes on this node, used to determine which parts of a buffer are fresh/stale later on.
        std::unordered_map<buffer_id, region<3>> per_buffer_local_writes;

        // Create command for each local chunk and resolve local data dependencies.
        for(const auto& [a_chunk, requirements] : chunks_with_requirements.local_chunks) {
                command* cmd = nullptr;
                if(tsk.get_type() == task_type::fence) {
                        cmd = create_command<fence_command>(current_batch, &tsk,
                            [&](const auto& record_debug_info) { record_debug_info(tsk, [this](const buffer_id bid) { return m_buffers.at(bid).debug_name; }); });
                } else {
                        // Go over all reductions that are to be performed *during* the execution of this chunk,
                        // not to be confused with any pending reductions that need to be finalized *before* the
                        // execution of this chunk (those have already been handled by resolve_pending_reductions).
                        // If a reduction reads the previous value of the buffer (i.e. w/o property::reduction::initialize_to_identity),
                        // we have to include it in exactly one of the per-node intermediate reductions.
                        const bool is_reduction_initializer = std::any_of(tsk.get_reductions().begin(), tsk.get_reductions().end(),
                            [&](const auto& reduction) { return m_local_nid == reduction_initializer_nid && reduction.init_from_buffer; });
                        cmd = create_command<execution_command>(current_batch, &tsk, subrange{a_chunk.chnk}, is_reduction_initializer,
                            [&](const auto& record_debug_info) { record_debug_info(tsk, [this](const buffer_id bid) { return m_buffers.at(bid).debug_name; }); });
                }

                if(tsk.get_type() == task_type::collective) {
                        // Collective host tasks have an implicit dependency on the previous task in the same collective group,
                        // which is required in order to guarantee they are executed in the same order on every node.
                        auto cgid = tsk.get_collective_group_id();
                        if(const auto cg = m_collective_groups.find(cgid); cg != m_collective_groups.end()) {
                                add_dependency(cmd, cg->second.last_collective_command, dependency_kind::true_dep, dependency_origin::collective_group_serialization);
                                cg->second.last_collective_command = cmd;
                        } else {
                                m_collective_groups.emplace(cgid, cmd);
                        }
                }

                for(const auto& [bid, consumed, produced] : requirements) {
                        auto& buffer = m_buffers.at(bid);

                        // Process consuming accesses first, so we don't add dependencies onto our own writes
                        if(!consumed.empty()) {
                                for(const auto& [box, wcs] : buffer.local_last_writer.get_region_values(consumed)) {
                                        if(box.empty()) continue;
                                        assert(wcs.is_fresh() && "Unresolved remote data dependency");
                                        add_dependency(cmd, wcs, dependency_kind::true_dep, dependency_origin::dataflow);
                                }

                                // Store the read access for determining anti-dependencies later on
                                m_command_buffer_reads[cmd->get_id()].emplace(bid, consumed);
                        }

                        if(!produced.empty()) {
                                generate_anti_dependencies(tsk, bid, buffer.local_last_writer, produced, cmd);

                                // Update last writer
                                buffer.local_last_writer.update_region(produced, cmd);
                                buffer.replicated_regions.update_region(produced, node_bitset{});

                                // In case this buffer was in a pending reduction state we discarded the result and need to remove the pending reduction.
                                if(buffer.pending_reduction.has_value()) {
                                        m_completed_reductions.push_back(buffer.pending_reduction->rid);
                                        buffer.pending_reduction = std::nullopt;
                                }

                                per_buffer_local_writes.emplace(bid, produced);
                        }

                        if(m_policy.uninitialized_read_error != error_policy::ignore
                            && !bounding_box(buffer.initialized_region).covers(bounding_box(consumed.get_boxes()))) {
                                utils::report_error(m_policy.uninitialized_read_error,
                                    "Command C{} on N{}, which executes {} of {}, reads {} {}, which has not been written by any node.", cmd->get_id(), m_local_nid,
                                    box(subrange(a_chunk.chnk.offset, a_chunk.chnk.range)), print_task_debug_label(tsk), print_buffer_debug_label(bid),
                                    region_difference(consumed, buffer.initialized_region));
                        }
                }

                for(const auto& side_effect : tsk.get_side_effect_map()) {
                        const auto [hoid, order] = side_effect;
                        auto& host_object = m_host_objects.at(hoid);

                        if(host_object.last_side_effect != nullptr) {
                                // TODO once we have different side_effect_orders, their interaction will determine the dependency kind
                                add_dependency(cmd, host_object.last_side_effect, dependency_kind::true_dep, dependency_origin::dataflow);
                        }

                        // Simplification: If there are multiple chunks per node, we generate true-dependencies between them in an arbitrary order, when all we really
                        // need is mutual exclusion (i.e. a bi-directional pseudo-dependency).
                        host_object.last_side_effect = cmd;
                }

                generate_epoch_dependencies(cmd);
        }

        // Mark any buffers that now are in a pending reduction state as such.
        // If there is only one chunk/command, it already implicitly generates the final reduced value
        // and the buffer does not need to be flagged as a pending reduction.
        for(const auto& reduction : tsk.get_reductions()) {
                if(chunks.size() > 1) {
                        m_buffers.at(reduction.bid).pending_reduction = reduction;
                } else {
                        m_completed_reductions.push_back(reduction.rid);
                }
        }
        update_local_buffer_fresh_regions(tsk, per_buffer_local_writes);
}

void command_graph_generator::generate_anti_dependencies(
    const task& tsk, const buffer_id bid, const region_map<write_command_state>& last_writers_map, const region<3>& write_req, command* const write_cmd) //
{
        const auto last_writers = last_writers_map.get_region_values(write_req);
        for(const auto& [box, wcs] : last_writers) {
                auto* const last_writer_cmd = wcs.get_command();
                assert(!utils::isa<task_command>(last_writer_cmd) || utils::as<task_command>(last_writer_cmd)->get_task() != &tsk);

                // Add anti-dependencies onto all successors of the writer
                bool has_successors = false;
                for(auto d : last_writer_cmd->get_dependents()) {
                        // Only consider true dependencies
                        if(d.kind != dependency_kind::true_dep) continue;

                        auto* const cmd = d.node;

                        // We might have already generated new commands within the same task that also depend on this; in that case, skip it
                        if(utils::isa<task_command>(cmd) && utils::as<task_command>(cmd)->get_task() == &tsk) continue;

                        // So far we don't know whether the dependent actually intersects with the subrange we're writing
                        if(const auto command_reads_it = m_command_buffer_reads.find(cmd->get_id()); command_reads_it != m_command_buffer_reads.end()) {
                                const auto& command_reads = command_reads_it->second;
                                // The task might be a dependent because of another buffer
                                if(const auto buffer_reads_it = command_reads.find(bid); buffer_reads_it != command_reads.end()) {
                                        if(!region_intersection(write_req, buffer_reads_it->second).empty()) {
                                                has_successors = true;
                                                add_dependency(write_cmd, cmd, dependency_kind::anti_dep, dependency_origin::dataflow);
                                        }
                                }
                        }
                }

                // In some cases (horizons, master node host task, weird discard_* constructs...)
                // the last writer might not have any successors. Just add the anti-dependency onto the writer itself then.
                if(!has_successors) { add_dependency(write_cmd, last_writer_cmd, dependency_kind::anti_dep, dependency_origin::dataflow); }
        }
}

void command_graph_generator::set_epoch_for_new_commands(command* const epoch_or_horizon) {
        // both an explicit epoch command and an applied horizon can be effective epochs
        assert(utils::isa<epoch_command>(epoch_or_horizon) || utils::isa<horizon_command>(epoch_or_horizon));

        for(auto& [_, buffer] : m_buffers) {
                buffer.local_last_writer.apply_to_values([epoch_or_horizon](write_command_state wcs) {
                        if(epoch_or_horizon->get_id() <= wcs.get_command()->get_id()) return wcs;
                        write_command_state new_wcs(epoch_or_horizon, wcs.is_replicated());
                        if(!wcs.is_fresh()) new_wcs.mark_as_stale();
                        return new_wcs;
                });
        }
        for(auto& [_, host_object] : m_host_objects) {
                if(host_object.last_side_effect != nullptr && host_object.last_side_effect->get_id() < epoch_or_horizon->get_id()) {
                        host_object.last_side_effect = epoch_or_horizon;
                }
        }
        for(auto& [cgid, collective_group] : m_collective_groups) {
                if(collective_group.last_collective_command->get_id() < epoch_or_horizon->get_id()) { collective_group.last_collective_command = epoch_or_horizon; }
        }

        m_epoch_for_new_commands = epoch_or_horizon;
}

void command_graph_generator::reduce_execution_front_to(command* const new_front) {
        const auto previous_execution_front = m_execution_front; // modified inside loop through add_dependency
        for(auto* const front_cmd : previous_execution_front) {
                if(front_cmd != new_front) { add_dependency(new_front, front_cmd, dependency_kind::true_dep, dependency_origin::execution_front); }
        }
        assert(m_execution_front.size() == 1 && *m_execution_front.begin() == new_front);
}

void command_graph_generator::generate_epoch_command(batch& current_batch, const task& tsk) {
        assert(tsk.get_type() == task_type::epoch);
        m_cdag->begin_epoch(tsk.get_id());
        auto* const epoch = create_command<epoch_command>(
            current_batch, &tsk, tsk.get_epoch_action(), std::move(m_completed_reductions), [&](const auto& record_debug_info) { record_debug_info(tsk); });
        set_epoch_for_new_commands(epoch);
        m_current_horizon = no_command;
        // Make the epoch depend on the previous execution front
        reduce_execution_front_to(epoch);
}

void command_graph_generator::generate_horizon_command(batch& current_batch, const task& tsk) {
        assert(tsk.get_type() == task_type::horizon);
        m_cdag->begin_epoch(tsk.get_id());
        auto* const new_horizon =
            create_command<horizon_command>(current_batch, &tsk, std::move(m_completed_reductions), [&](const auto& record_debug_info) { record_debug_info(tsk); });

        if(m_current_horizon != nullptr) {
                // Apply the previous horizon
                set_epoch_for_new_commands(m_current_horizon);
        }
        m_current_horizon = new_horizon;

        // Make the horizon depend on the previous execution front
        reduce_execution_front_to(new_horizon);
}

void command_graph_generator::generate_epoch_dependencies(command* cmd) {
        // No command must be re-ordered before its last preceding epoch to enforce the barrier semantics of epochs.
        // To guarantee that each node has a transitive true dependency (=temporal dependency) on the epoch, it is enough to add an epoch -> command dependency
        // to any command that has no other true dependencies itself and no graph traversal is necessary. This can be verified by a simple induction proof.

        // As long the first epoch is present in the graph, all transitive dependencies will be visible and the initial epoch commands (tid 0) are the only
        // commands with no true predecessor. As soon as the first epoch is pruned through the horizon mechanism however, more than one node with no true
        // predecessor can appear (when visualizing the graph). This does not violate the ordering constraint however, because all "free-floating" nodes
        // in that snapshot had a true-dependency chain to their predecessor epoch at the point they were flushed, which is sufficient for following the
        // dependency chain from the executor perspective.

        if(const auto deps = cmd->get_dependencies();
            std::none_of(deps.begin(), deps.end(), [](const command::dependency d) { return d.kind == dependency_kind::true_dep; })) {
                if(!utils::isa<epoch_command>(cmd) || utils::as<epoch_command>(cmd)->get_epoch_action() != epoch_action::init) {
                        assert(cmd != m_epoch_for_new_commands);
                        add_dependency(cmd, m_epoch_for_new_commands, dependency_kind::true_dep, dependency_origin::last_epoch);
                }
        }
}

std::string command_graph_generator::print_buffer_debug_label(const buffer_id bid) const {
        return utils::make_buffer_debug_label(bid, m_buffers.at(bid).debug_name);
}

} // namespace celerity::detail

celerity / celerity-runtime / 11610324419

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous