GCC Code Coverage Report

src/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
35 / 44
79.5%
Branches:
16 / 30
53.3%
Functions:
8 / 10
80.0%
storage/io_backend/io_backend.h
Line Branch Exec Source
1 #pragma once
2
3 #include <atomic>
4 #include <boost/coroutine2/all.hpp>
5 #include <cstdint>
6 #include <cstdio>
7 #include <cstring>
8 #include <glog/logging.h>
9 #include <sys/user.h>
10 #include <vector>
11 #include "../index/index.h"
12
13 using boost::coroutines2::coroutine;
14 extern thread_local int pending;
15 extern thread_local std::vector<std::unique_ptr<coroutine<void>::pull_type>>
16 coros;
17
18 typedef uint64_t PageID_t;
19 const PageID_t INVALID_PAGE = -1;
20
21 class IOBackend {
22 public:
23 726 IOBackend(const BaseKVConfig& config) {
24
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 726 times.
 726 if (!config.json_config_.contains("page_id_offset"))
25
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ LOG(WARNING) << "IOBackend config missing 'page_id_offset'";
26
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 726 times.
 726 if (!config.json_config_.contains("queue_cnt"))
27
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ LOG(WARNING) << "IOBackend config missing 'queue_cnt'";
28
1/2
✓ Branch 1 taken 726 times.
✗ Branch 2 not taken.
 726 next_page_id.store(config.json_config_.value("page_id_offset", (PageID_t)1),
29 std::memory_order_relaxed);
30
1/2
✓ Branch 1 taken 726 times.
✗ Branch 2 not taken.
 726 queue_cnt = config.json_config_.value("queue_cnt", 512);
31 726 }
32 726 virtual ~IOBackend() {}
33 virtual void init() = 0;
34
35 PageID_t AllocatePage(coroutine<void>::push_type& sink, uint64_t index) {
36 PageID_t new_page_id = next_page_id.fetch_add(1, std::memory_order_relaxed);
37 WritePageAsync(sink, index, new_page_id, empty_page);
38 return new_page_id;
39 }
40 ✗ PageID_t AllocatePage() {
41 ✗ PageID_t new_page_id = next_page_id.fetch_add(1, std::memory_order_relaxed);
42 ✗ WritePageSync(new_page_id, empty_page);
43 ✗ return new_page_id;
44 }
45 PageID_t GetNextPageID() const {
46 return next_page_id.load(std::memory_order_relaxed);
47 }
48 void SetNextPageID(PageID_t page_id) {
49 next_page_id.store(page_id, std::memory_order_relaxed);
50 }
51
52 void ReadPage(coroutine<void>::push_type& sink,
53 uint64_t index,
54 PageID_t page_id,
55 char* buffer) {
56 ReadPageAsync(sink, index, page_id, buffer);
57 }
58 210804 void ReadPage(PageID_t page_id, char* buffer) {
59 210804 ReadPageSync(page_id, buffer);
60 210804 }
61
62 ✗ void WritePage(coroutine<void>::push_type& sink,
63 uint64_t index,
64 PageID_t page_id,
65 char* buffer) {
66 ✗ WritePageAsync(sink, index, page_id, buffer);
67 ✗ }
68 void WritePage(PageID_t page_id, char* buffer) {
69 WritePageSync(page_id, buffer);
70 }
71
72 virtual void* GetPage(
73 coroutine<void>::push_type& sink, uint64_t index, PageID_t page_id) = 0;
74 virtual void* GetPage(PageID_t page_id) = 0;
75
76 // Unpin a page, if dirty, write it back
77 virtual void
78 Unpin(coroutine<void>::push_type& sink,
79 uint64_t index,
80 PageID_t page_id,
81 void* page_data,
82 bool is_dirty) = 0;
83 virtual void Unpin(PageID_t page_id, void* page_data, bool is_dirty) = 0;
84
85 virtual char* AllocateBuffer() = 0;
86 // Allocate a contiguous buffer of page_count pages (zero-filled).
87 virtual char* AllocateBuffer(uint64_t page_count) = 0;
88 virtual void FreeBuffer(char* buf) = 0;
89
90 virtual void submit() = 0;
91 virtual void PollCompletion() = 0;
92
93 struct IOEntry {
94 PageID_t page_id; // starting page
95 char* buffer; // contiguous buffer
96 uint64_t page_count; // number of pages in this buffer
97 };
98
99 struct ReadStats {
100 uint64_t page_reads = 0;
101 uint64_t bytes_read = 0;
102 };
103
104 4 ReadStats GetReadStats() const {
105 4 return {page_reads_.load(std::memory_order_relaxed),
106 8 bytes_read_.load(std::memory_order_relaxed)};
107 }
108
109 2 void ResetReadStats() {
110 2 page_reads_.store(0, std::memory_order_relaxed);
111 2 bytes_read_.store(0, std::memory_order_relaxed);
112 2 }
113
114 // Batch write using sliding-window async IO (no coroutines).
115 // Buffers are NOT freed by this method; caller is responsible.
116 514 virtual void BatchWritePages(const std::vector<IOEntry>& entries) {
117
2/2
✓ Branch 5 taken 514 times.
✓ Branch 6 taken 514 times.
 1028 for (auto& e : entries) {
118
2/2
✓ Branch 0 taken 520 times.
✓ Branch 1 taken 514 times.
 1034 for (uint64_t i = 0; i < e.page_count; i++) {
119
1/2
✓ Branch 1 taken 520 times.
✗ Branch 2 not taken.
 520 WritePageSync(e.page_id + i, e.buffer + i * PAGE_SIZE);
120 }
121 }
122 514 }
123
124 // Batch read using sliding-window async IO (no coroutines).
125 // Buffers are NOT freed by this method; caller is responsible.
126 560 virtual void BatchReadPages(const std::vector<IOEntry>& entries) {
127 560 RecordRead(entries);
128
2/2
✓ Branch 5 taken 570 times.
✓ Branch 6 taken 560 times.
 1130 for (auto& e : entries) {
129
2/2
✓ Branch 0 taken 582 times.
✓ Branch 1 taken 570 times.
 1152 for (uint64_t i = 0; i < e.page_count; i++) {
130
1/2
✓ Branch 1 taken 582 times.
✗ Branch 2 not taken.
 582 ReadPageSync(e.page_id + i, e.buffer + i * PAGE_SIZE);
131 }
132 }
133 560 }
134
135 protected:
136 std::atomic<PageID_t> next_page_id{1};
137 char* empty_page = nullptr;
138 int queue_cnt = 512;
139 std::atomic<uint64_t> page_reads_{0};
140 std::atomic<uint64_t> bytes_read_{0};
141
142 172926 void RecordRead(const std::vector<IOEntry>& entries) {
143 172926 uint64_t pages = 0;
144
2/2
✓ Branch 4 taken 172936 times.
✓ Branch 5 taken 172926 times.
 345862 for (const auto& entry : entries) {
145 172936 pages += entry.page_count;
146 }
147 172926 page_reads_.fetch_add(pages, std::memory_order_relaxed);
148 172926 bytes_read_.fetch_add(pages * PAGE_SIZE, std::memory_order_relaxed);
149 172926 }
150
151 virtual void ReadPageAsync(coroutine<void>::push_type& sink,
152 uint64_t index,
153 PageID_t page_id,
154 char* buffer) = 0;
155 virtual void ReadPageSync(PageID_t page_id, char* buffer) = 0;
156
157 virtual void WritePageAsync(coroutine<void>::push_type& sink,
158 uint64_t index,
159 PageID_t page_id,
160 char* buffer) = 0;
161 virtual void WritePageSync(PageID_t page_id, char* buffer) = 0;
162 };
163