Parallel SSSP using C++20.

sssp.cpp

#include <vector>
#include <algorithm>
#include <execution>
#include <mutex>
#include <utility>
#include <ranges>

struct frontier_t {
    // Underlying representation of frontier.
    std::vector<int> active_vertices;
    
    // Get the number of active vertices.
    int size() {
        return active_vertices.size();
    }
    // Get the active vertex at a given index.
    int get_active_vertex(int const& i) {
        return active_vertices[i];
    }
    // Add a vertex to the frontier.
    void add_vertex(int const& v) {
        active_vertices.push_back(v);
    }
};

// Compressed-Sparse Row (CSR) matrix.
struct csr_t {
    int rows;
    int cols;
    std::vector<int> row_offsets;
    std::vector<int> column_indices;
    std::vector<float> values;
};

struct graph_t : public csr_t {
    // Get edge weight for a given edge.
    float get_edge_weight(int const& e) {
        return values[e];
    }

    int get_num_vertices() { return rows; }
    int get_dest_vertex(const int& e) { return column_indices[e]; }
    auto get_edges(const int& v) { return std::ranges::iota_view{row_offsets[v], row_offsets[v+1]}; }
    auto get_vertices() { return std::ranges::iota_view{0, get_num_vertices()}; }
};

// Neighbors expand implemented in C++ 20.
template<typename expand_cond_t>
frontier_t neighbors_expand(
    graph_t& g, 
    frontier_t& f, 
    expand_cond_t condition
) {
    std::mutex m;
    frontier_t output;
    auto expand = [&] (int const& v) {
        // For all edges of vertex v.
        for (auto e : g.get_edges(v)) {
            auto n = g.get_dest_vertex(e);
            auto w = g.get_edge_weight(e);
            // If expand condition is
            // true, add the neighbor into
            // the output frontier.
            if (condition(v, n, e, w)) {
                std::lock_guard<std::mutex> 
                    guard(m);
                output.add_vertex(n);
            }
        }
    };

    // For all active vertices in the
    // frontier, process in parallel.
    std::for_each(std::execution::par, 
        f.active_vertices.begin(), 
        f.active_vertices.end(), 
        expand);
    
    // Return the new output frontier.
    return output;
}

namespace atomic {
    template<typename T>
    T min(T* a, T b) {
        T old = *a;
        T ans =std::min(old, b);
        *a = ans;
        return old;
    }
}

std::vector<float> sssp(
            graph_t& g,
            int const& source
) {
    // Initialize data.
    std::vector<float> distances(g.get_num_vertices());
    for (auto v : g.get_vertices())
        distances[v] = std::numeric_limits<float>::max();
    distances[source] = 0;
    frontier_t f;
    f.add_vertex(source);

    std::vector<std::mutex> m_locks(g.get_num_vertices());

    while(f.size() != 0) {
        // Expand the frontier.
        f = neighbors_expand(g, f,
        // User-defined condition for SSSP. 
        [&](int const& src,     // source
            int const& dst,     // destination 
            int const& edge,    // edge
            float const& weight // weight
        ) {
            float new_d = distances[src] + weight;
            // atomic::min atomically updates the distances array at dst with the minimum of new_d or its current value. And returns the old value. (eq: mutex updates)
            std::lock_guard<std::mutex> guard(m_locks[dst]);
            float curr_d = atomic::min(&distances[dst], new_d);
            return new_d < curr_d;
        });
    }
    return distances;
}

https://godbolt.org/z/noPWsz5rs (overloads)