sux 0.12.3

A pure Rust implementation of succinct and compressed data structures
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

#pragma once

#include <thread>
#include <memory>
#include <deque>

#include "configuration.hpp"
#include "util.hpp"

namespace ds2i {

    class semiasync_queue {
    public:

        semiasync_queue(double work_per_thread)
            : m_expected_work(0)
            , m_work_per_thread(work_per_thread)
        {
            m_max_threads = configuration::get().worker_threads;
            logger() << "semiasync_queue using " << m_max_threads
                     << " worker threads" << std::endl;
        }

        class job {
        public:
            virtual void prepare() = 0;
            virtual void commit() = 0;
        };

        typedef std::shared_ptr<job> job_ptr_type;

        void add_job(job_ptr_type j, double expected_work)
        {
            if (m_max_threads) {
                m_next_thread.first.push_back(j);
                m_expected_work += expected_work;
                if (m_expected_work >= m_work_per_thread) {
                    spawn_next_thread();
                }
            } else { // all in main thread
                j->prepare();
                j->commit();
                j.reset();
            }
        }

        void complete()
        {
            if (!m_next_thread.first.empty()) {
                spawn_next_thread();
            }
            while (!m_running_threads.empty()) {
                commit_thread();
            }
        }

    private:

        void spawn_next_thread()
        {
            if (m_running_threads.size() == m_max_threads) {
                commit_thread();
            }

            m_running_threads.emplace_back();
            std::swap(m_next_thread, m_running_threads.back());

            std::vector<job_ptr_type> const& cur_queue = m_running_threads.back().first;
            m_running_threads.back().second = std::thread([&]() {
                    for (auto const& j: cur_queue) {
                        j->prepare();
                    }
                });

            m_expected_work = 0;
        }

        void commit_thread()
        {
            assert(!m_running_threads.empty());
            m_running_threads.front().second.join();
            for (auto& j: m_running_threads.front().first) {
                j->commit();
                j.reset();
            }
            m_running_threads.pop_front();
        }

        typedef std::pair<std::vector<job_ptr_type>, std::thread> thread_t;
        thread_t m_next_thread;
        std::deque<thread_t> m_running_threads;

        size_t m_expected_work;
        double m_work_per_thread;
        size_t m_max_threads;
    };

}