neoblizz/sssp.cpp

## 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;
}
	#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;
	}