eloqstore-sys 1.1.1

Low-level Rust FFI bindings for EloqStore
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133


#pragma once

#include <cassert>
#include <limits>
#include <memory>
#include <vector>

#include "tasks/background_write.h"
#include "tasks/batch_write_task.h"
#include "tasks/list_object_task.h"
#include "tasks/list_standby_partition_task.h"
#include "tasks/read_task.h"
#include "tasks/reopen_task.h"
#include "tasks/scan_task.h"
#include "types.h"

namespace eloqstore
{
struct KvOptions;

class TaskManager
{
public:
    explicit TaskManager(const KvOptions *opts);

    static void SetPoolSizesForTest(uint32_t batch_write_pool_size,
                                    uint32_t background_write_pool_size,
                                    uint32_t read_pool_size,
                                    uint32_t scan_pool_size,
                                    uint32_t list_object_pool_size,
                                    uint32_t reopen_pool_size);

    BatchWriteTask *GetBatchWriteTask(const TableIdent &tbl_id);
    BackgroundWrite *GetBackgroundWrite(const TableIdent &tbl_id);
    ReadTask *GetReadTask();
    ScanTask *GetScanTask();
    ListObjectTask *GetListObjectTask();
    ListStandbyPartitionTask *GetListStandbyPartitionTask();
    ReopenTask *GetReopenTask(const TableIdent &tbl_id);
    void FreeTask(KvTask *task);

    void AddExternalTask();
    void FinishExternalTask();

    size_t NumActive() const;

    void Shutdown();

private:
    template <typename T>
    class TaskPool
    {
    public:
        TaskPool(uint32_t size, bool allow_growth = true)
            : init_pool_size_(size), allow_growth_(allow_growth)
        {
            if (size > 0)
            {
                init_pool_ = std::make_unique<T[]>(size);
                for (uint32_t i = 0; i < size; i++)
                {
                    FreeTask(&init_pool_[i]);
                }
            }
        };

        T *GetTask()
        {
            if (free_head_ != nullptr)
            {
                // Reuse a free task.
                T *task = free_head_;
                free_head_ = static_cast<T *>(task->next_);
                task->next_ = nullptr;
                assert(task->status_ == TaskStatus::Idle);
                return task;
            }
            if (!allow_growth_)
            {
                return nullptr;
            }
            auto &task = ext_pool_.emplace_back(std::make_unique<T>());
            return task.get();
        }

        void FreeTask(T *task)
        {
            task->status_ = TaskStatus::Idle;
            task->next_ = free_head_;
            free_head_ = task;
        }

        template <typename F>
        void ForEachTask(F &&visitor)
        {
            for (uint32_t i = 0; i < init_pool_size_; ++i)
            {
                visitor(&init_pool_[i]);
            }
            for (auto &task : ext_pool_)
            {
                visitor(task.get());
            }
        }

        void Clear()
        {
            free_head_ = nullptr;
            ext_pool_.clear();
            init_pool_.reset();
        }

    private:
        std::unique_ptr<T[]> init_pool_{nullptr};
        std::vector<std::unique_ptr<T>> ext_pool_;
        T *free_head_{nullptr};
        uint32_t init_pool_size_{0};
        bool allow_growth_{true};
    };

    TaskPool<BatchWriteTask> batch_write_pool_;
    TaskPool<BackgroundWrite> bg_write_pool_;
    TaskPool<ReadTask> read_pool_;
    TaskPool<ScanTask> scan_pool_;
    TaskPool<ListObjectTask> list_object_pool_;
    TaskPool<ListStandbyPartitionTask> list_standby_partition_pool_;
    TaskPool<ReopenTask> reopen_pool_;
    size_t num_active_{0};
    size_t num_active_write_{0};
    size_t max_write_concurrency_{std::numeric_limits<size_t>::max()};
};
}  // namespace eloqstore