<commit_before>#include "brandes.h"

#include <queue>
#include <stack>
#include <algorithm>
#include <thread>

template<typename T>
void Brandes<T>::process(T vertex_id, std::unordered_map<T, fType> *BC_local)
{
    std::stack<T> S;
    std::unordered_map<T, std::vector<T> > P;
    std::unordered_map<T, int> sigma;
    std::unordered_map<T, int> d;
    std::unordered_map<T, fType> delta;

    for (T w : graph_.get_vertex_ids()) {
        P[w] = std::vector<T>();
        sigma[w] = 0;
        d[w] = -1;
        delta[w] = 0;
    }

    sigma[vertex_id] = 1;
    d[vertex_id] = 0;

    std::queue<T> Q;
    Q.push(vertex_id);

    while (!Q.empty()) {
        T v = Q.front();
        Q.pop();
        S.push(v);

        for (auto w : graph_.get_neighbors(v)) {
            if (d[w] < 0) {
                Q.push(w);
                d[w] = d[v] + 1;
            }

            if (d[w] == d[v] + 1) {
                sigma[w] += sigma[v];
                P[w].emplace_back(v);
            }
        }
    }

    while (!S.empty()) {
        T v = S.top(); S.pop();

        for (T p : P[v]) {
            double result = (fType(sigma[p]) / sigma[v]) * (1.0 + delta[v]);
            delta[p] += result;
        }

        if (v != vertex_id) {
            (*BC_local)[v] += delta[v];
        }
    }

}

template<typename T>
void Brandes<T>::run_worker(const std::vector<T> &jobs, std::atomic<int> *idx) {
    std::unordered_map<T, fType> BC_local;

    for (T w : graph_.get_vertex_ids()) {
        BC_local[w] = 0;
    }

    while(true) {
        int my_index = (*idx)--;

        if (my_index < 0)
            break;

        process(jobs[my_index], &BC_local);
    }

    // Synchronized section
    {
        std::lock_guard<std::mutex> guard(bc_mutex_);
        for (auto it : BC_local) {
            BC_[it.first] += it.second;
        }
    }
}

template<typename T>
void Brandes<T>::run(const size_t thread_num) {
    std::vector<std::thread> threads;

    std::vector<T> jobs;
    std::atomic<int> index;

    // add jobs
    for (auto id : graph_.get_vertex_ids())
        jobs.emplace_back(id);

    index.store(jobs.size() - 1);

    // run threads
    for (size_t i = 0; i < thread_num; i++)
        threads.emplace_back(std::thread( [this, &jobs, &index] { run_worker(jobs, &index); }));

    // wait for them to finish
    for (auto& thread : threads)
        thread.join();
}

template<typename T>
const std::vector<std::pair<T, typename Brandes<T>::fType>>
Brandes<T>::get_result_vector() const {
    std::vector<std::pair<T, fType>> results;
    results.reserve(BC_.size());

    for (auto it : BC_) {<commit_msg>Modify algorithm to reduce number of threads<commit_after>#include "brandes.h"

#include <queue>
#include <stack>
#include <algorithm>
#include <thread>

template<typename T>
void Brandes<T>::process(T vertex_id, std::unordered_map<T, fType> *BC_local)
{
    std::stack<T> S;
    std::unordered_map<T, std::vector<T> > P;
    std::unordered_map<T, int> sigma;
    std::unordered_map<T, int> d;
    std::unordered_map<T, fType> delta;

    for (T w : graph_.get_vertex_ids()) {
        P[w] = std::vector<T>();
        sigma[w] = 0;
        d[w] = -1;
        delta[w] = 0;
    }

    sigma[vertex_id] = 1;
    d[vertex_id] = 0;

    std::queue<T> Q;
    Q.push(vertex_id);

    while (!Q.empty()) {
        T v = Q.front();
        Q.pop();
        S.push(v);

        for (auto w : graph_.get_neighbors(v)) {
            if (d[w] < 0) {
                Q.push(w);
                d[w] = d[v] + 1;
            }

            if (d[w] == d[v] + 1) {
                sigma[w] += sigma[v];
                P[w].emplace_back(v);
            }
        }
    }

    while (!S.empty()) {
        T v = S.top(); S.pop();

        for (T p : P[v]) {
            double result = (fType(sigma[p]) / sigma[v]) * (1.0 + delta[v]);
            delta[p] += result;
        }

        if (v != vertex_id) {
            (*BC_local)[v] += delta[v];
        }
    }

}

template<typename T>
void Brandes<T>::run_worker(const std::vector<T> &jobs, std::atomic<int> *idx) {
    std::unordered_map<T, fType> BC_local;

    for (T w : graph_.get_vertex_ids()) {
        BC_local[w] = 0;
    }

    while(true) {
        int my_index = (*idx)--;

        if (my_index < 0)
            break;

        process(jobs[my_index], &BC_local);
    }

    // Synchronized section
    {
        std::lock_guard<std::mutex> guard(bc_mutex_);
        for (auto it : BC_local) {
            BC_[it.first] += it.second;
        }
    }
}

template<typename T>
void Brandes<T>::run(const size_t thread_num) {
    std::vector<std::thread> threads;

    std::vector<T> jobs;
    std::atomic<int> index;

    // add jobs
    for (auto id : graph_.get_vertex_ids())
        jobs.emplace_back(id);

    index.store(jobs.size() - 1);

    // run threads
    for (size_t i = 0; i < thread_num-1; i++)
        threads.emplace_back(std::thread( [this, &jobs, &index] { run_worker(jobs, &index); }));

    run_worker(jobs, &index);

    // wait for others to finish
    for (auto& thread : threads)
        thread.join();
}

template<typename T>
const std::vector<std::pair<T, typename Brandes<T>::fType>>
Brandes<T>::get_result_vector() const {
    std::vector<std::pair<T, fType>> results;
    results.reserve(BC_.size());

    for (auto it : BC_) {
{'fraction_non_alphanumeric': 0.17640449438202246, 'fraction_numerical': 0.0054307116104868915, 'mean_word_length': 2.1362301820317087, 'pattern_counts': {'":': 0, '<': 75, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6293103', 'n_tokens_mistral': 2051, 'n_tokens_neox': 1952, 'n_words': 404}