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
134
135
136
137
138
139
140
141

#pragma once

#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "direct_io_buffer.h"
#include "error.h"
#include "storage/data_page.h"
#include "storage/index_page_manager.h"
#include "storage/object_store.h"
#include "storage/page_mapper.h"
#include "storage/root_meta.h"
#include "tasks/task.h"
#include "tasks/write_buffer_aggregator.h"
#include "types.h"

namespace eloqstore
{
class WriteTask : public KvTask
{
public:
    struct UploadState
    {
        DirectIoBuffer buffer;
        std::string filename;
        uint64_t start_offset{0};
        uint64_t end_offset{0};
        bool initialized{false};
        bool invalid{false};

        void ResetMetadata()
        {
            filename.clear();
            start_offset = 0;
            end_offset = 0;
            initialized = false;
            invalid = false;
            buffer.clear();
        }
    };

    WriteTask() = default;
    WriteTask(const WriteTask &) = delete;

    void Abort() override;
    virtual void Reset(const TableIdent &tbl_id);
    const TableIdent &TableId() const;

    /**
     * @brief The index/data page has been flushed.
     * Enqueues the page into the cache replacement list (so that the page is
     * allowed to be evicted) if it is a index page.
     */
    void WritePageCallback(VarPage page, KvError err);

    UploadState &MutableUploadState()
    {
        return upload_state_;
    }
    const UploadState &GetUploadState() const
    {
        return upload_state_;
    }
    void ResetUploadState();
    void EnsureUploadStateBuffer();
    void AddInflightUploadTask();
    void CompletePendingUploadTask(ObjectStore::UploadTask *task);
    void AddPendingUploadTask(std::unique_ptr<ObjectStore::UploadTask> task)
    {
        pending_upload_tasks_.emplace_back(std::move(task));
    }

    KvError WaitWrite();
    KvError DeleteArchive(uint64_t term, std::string_view tag);
    void ReleaseDirBusy();
    bool NeedWaitDirEviction() const
    {
        return need_wait_dir_eviction_;
    }
    void ClearNeedWaitDirEviction()
    {
        need_wait_dir_eviction_ = false;
    }
    // write_err_ record the result of the last failed write
    // request.
    KvError write_err_{KvError::NoError};

protected:
    KvError FlushManifest();
    KvError UpdateMeta();

    /**
     * @brief Request shard to create a compaction task if space amplification
     * factor is too big.
     */
    void CompactIfNeeded(PageMapper *mapper) const;
    void TriggerTTL();
    void TriggerFileGC() const;
    KvError TriggerLocalFileGC() const;

    std::pair<DataPage, KvError> LoadDataPage(PageId page_id);
    std::pair<OverflowPage, KvError> LoadOverflowPage(PageId page_id);

    std::pair<PageId, FilePageId> AllocatePage(PageId page_id);
    std::string_view TaskTypeName() const;
    void FreePage(PageId page_id);

    FilePageId ToFilePage(PageId page_id);

    TableIdent tbl_ident_;

    CowRootMeta cow_meta_;
    ManifestBuilder wal_builder_;

    KvError WritePage(DataPage &&page);
    KvError WritePage(OverflowPage &&page);
    KvError WritePage(MemIndexPage::Handle &page);
    KvError WritePage(MemIndexPage::Handle &page, FilePageId file_page_id);
    KvError WritePage(VarPage page, FilePageId file_page_id);
    KvError AppendWritePage(VarPage page, FilePageId file_page_id);
    void FlushAppendWrites();
    KvError WaitPendingUploads();
    KvError ConsumePendingUploadResults();
    std::pair<FileId, uint32_t> ConvFilePageId(FilePageId file_page_id) const;
    virtual DirectIoBuffer AcquireUploadStateBuffer();
    virtual void ReleaseUploadStateBuffer(DirectIoBuffer buffer);

    std::optional<FileId> last_append_file_id_;
    WriteBufferAggregator append_aggregator_{0};
    UploadState upload_state_;
    uint32_t inflight_upload_tasks_{0};
    WaitingSeat upload_waiting_;
    std::vector<std::unique_ptr<ObjectStore::UploadTask>> pending_upload_tasks_;
    bool dir_busy_registered_{false};
    bool need_wait_dir_eviction_{false};
};

}  // namespace eloqstore