dbengine extent cache (#10293)
* Add initial extent cache implementation * Integrate extent cache with dbengine.
Markos Fountoulakis committed
Dec 3, 2020 at 12:06 UTC
d1a3c48ce522b0477655491d184be981b39e567c
3 files changed
+288
-10
database/engine/rrdengine.c
+236
-10
@@ -27,10 +27,188 @@ static void sanity_check(void)
27
/* page count must fit in 8 bits */
28
BUILD_BUG_ON(MAX_PAGES_PER_EXTENT > 255);
29
30
+ /* extent cache count must fit in 32 bits */
31
+ BUILD_BUG_ON(MAX_CACHED_EXTENTS > 32);
32
+
33
/* page info scratch space must be able to hold 2 32-bit integers */
34
BUILD_BUG_ON(sizeof(((struct rrdeng_page_info *)0)->scratch) < 2 * sizeof(uint32_t));
35
}
36
37
+/* always inserts into tail */
38
+static inline void xt_cache_replaceQ_insert(struct rrdengine_worker_config* wc,
39
+ struct extent_cache_element *xt_cache_elem)
40
+{
41
+ struct extent_cache *xt_cache = &wc->xt_cache;
42
+
43
+ xt_cache_elem->prev = NULL;
44
+ xt_cache_elem->next = NULL;
45
+
46
+ if (likely(NULL != xt_cache->replaceQ_tail)) {
47
+ xt_cache_elem->prev = xt_cache->replaceQ_tail;
48
+ xt_cache->replaceQ_tail->next = xt_cache_elem;
49
+ }
50
+ if (unlikely(NULL == xt_cache->replaceQ_head)) {
51
+ xt_cache->replaceQ_head = xt_cache_elem;
52
+ }
53
+ xt_cache->replaceQ_tail = xt_cache_elem;
54
+}
55
+
56
+static inline void xt_cache_replaceQ_delete(struct rrdengine_worker_config* wc,
57
+ struct extent_cache_element *xt_cache_elem)
58
+{
59
+ struct extent_cache *xt_cache = &wc->xt_cache;
60
+ struct extent_cache_element *prev, *next;
61
+
62
+ prev = xt_cache_elem->prev;
63
+ next = xt_cache_elem->next;
64
+
65
+ if (likely(NULL != prev)) {
66
+ prev->next = next;
67
+ }
68
+ if (likely(NULL != next)) {
69
+ next->prev = prev;
70
+ }
71
+ if (unlikely(xt_cache_elem == xt_cache->replaceQ_head)) {
72
+ xt_cache->replaceQ_head = next;
73
+ }
74
+ if (unlikely(xt_cache_elem == xt_cache->replaceQ_tail)) {
75
+ xt_cache->replaceQ_tail = prev;
76
+ }
77
+ xt_cache_elem->prev = xt_cache_elem->next = NULL;
78
+}
79
+
80
+static inline void xt_cache_replaceQ_set_hot(struct rrdengine_worker_config* wc,
81
+ struct extent_cache_element *xt_cache_elem)
82
+{
83
+ xt_cache_replaceQ_delete(wc, xt_cache_elem);
84
+ xt_cache_replaceQ_insert(wc, xt_cache_elem);
85
+}
86
+
87
+/* Returns the index of the cached extent if it was successfully inserted in the extent cache, otherwise -1 */
88
+static int try_insert_into_xt_cache(struct rrdengine_worker_config* wc, struct extent_info *extent)
89
+{
90
+ struct extent_cache *xt_cache = &wc->xt_cache;
91
+ struct extent_cache_element *xt_cache_elem;
92
+ unsigned idx;
93
+ int ret;
94
+
95
+ ret = find_first_zero(xt_cache->allocation_bitmap);
96
+ if (-1 == ret || ret >= MAX_CACHED_EXTENTS) {
97
+ for (xt_cache_elem = xt_cache->replaceQ_head ; NULL != xt_cache_elem ; xt_cache_elem = xt_cache_elem->next) {
98
+ idx = xt_cache_elem - xt_cache->extent_array;
99
+ if (!check_bit(xt_cache->inflight_bitmap, idx)) {
100
+ xt_cache_replaceQ_delete(wc, xt_cache_elem);
101
+ break;
102
+ }
103
+ }
104
+ if (NULL == xt_cache_elem)
105
+ return -1;
106
+ } else {
107
+ idx = (unsigned)ret;
108
+ xt_cache_elem = &xt_cache->extent_array[idx];
109
+ }
110
+ xt_cache_elem->extent = extent;
111
+ xt_cache_elem->fileno = extent->datafile->fileno;
112
+ xt_cache_elem->inflight_io_descr = NULL;
113
+ xt_cache_replaceQ_insert(wc, xt_cache_elem);
114
+ modify_bit(&xt_cache->allocation_bitmap, idx, 1);
115
+
116
+ return (int)idx;
117
+}
118
+
119
+/**
120
+ * Returns 0 if the cached extent was found in the extent cache, 1 otherwise.
121
+ * Sets *idx to point to the position of the extent inside the cache.
122
+ **/
123
+static uint8_t lookup_in_xt_cache(struct rrdengine_worker_config* wc, struct extent_info *extent, unsigned *idx)
124
+{
125
+ struct extent_cache *xt_cache = &wc->xt_cache;
126
+ struct extent_cache_element *xt_cache_elem;
127
+ unsigned i;
128
+
129
+ for (i = 0 ; i < MAX_CACHED_EXTENTS ; ++i) {
130
+ xt_cache_elem = &xt_cache->extent_array[i];
131
+ if (check_bit(xt_cache->allocation_bitmap, i) && xt_cache_elem->extent == extent &&
132
+ xt_cache_elem->fileno == extent->datafile->fileno) {
133
+ *idx = i;
134
+ return 0;
135
+ }
136
+ }
137
+ return 1;
138
+}
139
+
140
+#if 0 /* disabled code */
141
+static void delete_from_xt_cache(struct rrdengine_worker_config* wc, unsigned idx)
142
+{
143
+ struct extent_cache *xt_cache = &wc->xt_cache;
144
+ struct extent_cache_element *xt_cache_elem;
145
+
146
+ xt_cache_elem = &xt_cache->extent_array[idx];
147
+ xt_cache_replaceQ_delete(wc, xt_cache_elem);
148
+ xt_cache_elem->extent = NULL;
149
+ modify_bit(&wc->xt_cache.allocation_bitmap, idx, 0); /* invalidate it */
150
+ modify_bit(&wc->xt_cache.inflight_bitmap, idx, 0); /* not in-flight anymore */
151
+}
152
+#endif
153
+
154
+void enqueue_inflight_read_to_xt_cache(struct rrdengine_worker_config* wc, unsigned idx,
155
+ struct extent_io_descriptor *xt_io_descr)
156
+{
157
+ struct extent_cache *xt_cache = &wc->xt_cache;
158
+ struct extent_cache_element *xt_cache_elem;
159
+ struct extent_io_descriptor *old_next;
160
+
161
+ xt_cache_elem = &xt_cache->extent_array[idx];
162
+ old_next = xt_cache_elem->inflight_io_descr->next;
163
+ xt_cache_elem->inflight_io_descr->next = xt_io_descr;
164
+ xt_io_descr->next = old_next;
165
+}
166
+
167
+void read_cached_extent_cb(struct rrdengine_worker_config* wc, unsigned idx, struct extent_io_descriptor *xt_io_descr)
168
+{
169
+ unsigned i, j, page_offset;
170
+ struct rrdengine_instance *ctx = wc->ctx;
171
+ struct rrdeng_page_descr *descr;
172
+ struct page_cache_descr *pg_cache_descr;
173
+ void *page;
174
+ struct extent_info *extent = xt_io_descr->descr_array[0]->extent;
175
+
176
+ for (i = 0 ; i < xt_io_descr->descr_count; ++i) {
177
+ page = mallocz(RRDENG_BLOCK_SIZE);
178
+ descr = xt_io_descr->descr_array[i];
179
+ for (j = 0, page_offset = 0 ; j < extent->number_of_pages ; ++j) {
180
+ /* care, we don't hold the descriptor mutex */
181
+ if (!uuid_compare(*extent->pages[j]->id, *descr->id) &&
182
+ extent->pages[j]->page_length == descr->page_length &&
183
+ extent->pages[j]->start_time == descr->start_time &&
184
+ extent->pages[j]->end_time == descr->end_time) {
185
+ break;
186
+ }
187
+ page_offset += extent->pages[j]->page_length;
188
+
189
+ }
190
+ /* care, we don't hold the descriptor mutex */
191
+ (void) memcpy(page, wc->xt_cache.extent_array[idx].pages + page_offset, descr->page_length);
192
+
193
+ rrdeng_page_descr_mutex_lock(ctx, descr);
194
+ pg_cache_descr = descr->pg_cache_descr;
195
+ pg_cache_descr->page = page;
196
+ pg_cache_descr->flags |= RRD_PAGE_POPULATED;
197
+ pg_cache_descr->flags &= ~RRD_PAGE_READ_PENDING;
198
+ rrdeng_page_descr_mutex_unlock(ctx, descr);
199
+ pg_cache_replaceQ_insert(ctx, descr);
200
+ if (xt_io_descr->release_descr) {
201
+ pg_cache_put(ctx, descr);
202
+ } else {
203
+ debug(D_RRDENGINE, "%s: Waking up waiters.", __func__);
204
+ pg_cache_wake_up_waiters(ctx, descr);
205
+ }
206
+ }
207
+ if (xt_io_descr->completion)
208
+ complete(xt_io_descr->completion);
209
+ freez(xt_io_descr);
210
+}
211
+
212
void read_extent_cb(uv_fs_t* req)
213
{
214
struct rrdengine_worker_config* wc = req->loop->data;
@@ -99,6 +277,33 @@ after_crc_check:
277
debug(D_RRDENGINE, "LZ4 decompressed %u bytes to %d bytes.", payload_length, ret);
278
/* care, we don't hold the descriptor mutex */
279
}
280
+ {
281
+ uint8_t xt_is_cached = 0;
282
+ unsigned xt_idx;
283
+ struct extent_info *extent = xt_io_descr->descr_array[0]->extent;
284
+
285
+ xt_is_cached = !lookup_in_xt_cache(wc, extent, &xt_idx);
286
+ if (xt_is_cached && check_bit(wc->xt_cache.inflight_bitmap, xt_idx)) {
287
+ struct extent_cache *xt_cache = &wc->xt_cache;
288
+ struct extent_cache_element *xt_cache_elem = &xt_cache->extent_array[xt_idx];
289
+ struct extent_io_descriptor *curr, *next;
290
+
291
+ if (have_read_error) {
292
+ memset(xt_cache_elem->pages, 0, sizeof(xt_cache_elem->pages));
293
+ } else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
294
+ (void)memcpy(xt_cache_elem->pages, xt_io_descr->buf + payload_offset, payload_length);
295
+ } else {
296
+ (void)memcpy(xt_cache_elem->pages, uncompressed_buf, uncompressed_payload_length);
297
+ }
298
+ /* complete all connected in-flight read requests */
299
+ for (curr = xt_cache_elem->inflight_io_descr->next ; curr ; curr = next) {
300
+ next = curr->next;
301
+ read_cached_extent_cb(wc, xt_idx, curr);
302
+ }
303
+ xt_cache_elem->inflight_io_descr = NULL;
304
+ modify_bit(&xt_cache->inflight_bitmap, xt_idx, 0); /* not in-flight anymore */
305
+ }
306
+ }
307
308
for (i = 0 ; i < xt_io_descr->descr_count; ++i) {
309
page = mallocz(RRDENG_BLOCK_SIZE);
@@ -159,18 +364,15 @@ static void do_read_extent(struct rrdengine_worker_config* wc,
364
// uint32_t payload_length;
365
struct extent_io_descriptor *xt_io_descr;
366
struct rrdengine_datafile *datafile;
367
+ struct extent_info *extent = descr[0]->extent;
368
+ uint8_t xt_is_cached = 0, xt_is_inflight = 0;
369
+ unsigned xt_idx;
370
163
- datafile = descr[0]->extent->datafile;
164
- pos = descr[0]->extent->offset;
165
- size_bytes = descr[0]->extent->size;
371
+ datafile = extent->datafile;
372
+ pos = extent->offset;
373
+ size_bytes = extent->size;
374
167
- xt_io_descr = mallocz(sizeof(*xt_io_descr));
168
- ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));
169
- if (unlikely(ret)) {
170
- fatal("posix_memalign:%s", strerror(ret));
171
- /* freez(xt_io_descr);
172
- return;*/
173
- }
375
+ xt_io_descr = callocz(1, sizeof(*xt_io_descr));
376
for (i = 0 ; i < count; ++i) {
377
rrdeng_page_descr_mutex_lock(ctx, descr[i]);
378
pg_cache_descr = descr[i]->pg_cache_descr;
@@ -188,6 +390,30 @@ static void do_read_extent(struct rrdengine_worker_config* wc,
390
/* xt_io_descr->descr_commit_idx_array[0] */
391
xt_io_descr->release_descr = release_descr;
392
393
+ xt_is_cached = !lookup_in_xt_cache(wc, extent, &xt_idx);
394
+ if (xt_is_cached) {
395
+ xt_cache_replaceQ_set_hot(wc, &wc->xt_cache.extent_array[xt_idx]);
396
+ xt_is_inflight = check_bit(wc->xt_cache.inflight_bitmap, xt_idx);
397
+ if (xt_is_inflight) {
398
+ enqueue_inflight_read_to_xt_cache(wc, xt_idx, xt_io_descr);
399
+ return;
400
+ }
401
+ return read_cached_extent_cb(wc, xt_idx, xt_io_descr);
402
+ } else {
403
+ ret = try_insert_into_xt_cache(wc, extent);
404
+ if (-1 != ret) {
405
+ xt_idx = (unsigned)ret;
406
+ modify_bit(&wc->xt_cache.inflight_bitmap, xt_idx, 1);
407
+ wc->xt_cache.extent_array[xt_idx].inflight_io_descr = xt_io_descr;
408
+ }
409
+ }
410
+
411
+ ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));
412
+ if (unlikely(ret)) {
413
+ fatal("posix_memalign:%s", strerror(ret));
414
+ /* freez(xt_io_descr);
415
+ return;*/
416
+ }
417
real_io_size = ALIGN_BYTES_CEILING(size_bytes);
418
xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);
419
ret = uv_fs_read(wc->loop, &xt_io_descr->req, datafile->file, &xt_io_descr->iov, 1, pos, read_extent_cb);
database/engine/rrdengine.h
+24
@@ -88,6 +88,7 @@ struct extent_io_descriptor {
88
int release_descr;
89
struct rrdeng_page_descr *descr_array[MAX_PAGES_PER_EXTENT];
90
Word_t descr_commit_idx_array[MAX_PAGES_PER_EXTENT];
91
+ struct extent_io_descriptor *next; /* multiple requests to be served by the same cached extent */
92
};
93
94
struct generic_io_descriptor {
@@ -99,6 +100,27 @@ struct generic_io_descriptor {
100
struct completion *completion;
101
};
102
103
+struct extent_cache_element {
104
+ struct extent_info *extent; /* The ABA problem is avoided with the help of fileno below */
105
+ unsigned fileno;
106
+ struct extent_cache_element *prev; /* LRU */
107
+ struct extent_cache_element *next; /* LRU */
108
+ struct extent_io_descriptor *inflight_io_descr; /* I/O descriptor for in-flight extent */
109
+ uint8_t pages[MAX_PAGES_PER_EXTENT * RRDENG_BLOCK_SIZE];
110
+};
111
+
112
+#define MAX_CACHED_EXTENTS 16 /* cannot be over 32 to fit in 32-bit architectures */
113
+
114
+/* Initialize by setting the structure to zero */
115
+struct extent_cache {
116
+ struct extent_cache_element extent_array[MAX_CACHED_EXTENTS];
117
+ unsigned allocation_bitmap; /* 1 if the corresponding position in the extent_array is allocated */
118
+ unsigned inflight_bitmap; /* 1 if the corresponding position in the extent_array is waiting for I/O */
119
+
120
+ struct extent_cache_element *replaceQ_head; /* LRU */
121
+ struct extent_cache_element *replaceQ_tail; /* MRU */
122
+};
123
+
124
struct rrdengine_worker_config {
125
struct rrdengine_instance *ctx;
126
@@ -122,6 +144,8 @@ struct rrdengine_worker_config {
144
volatile unsigned queue_size;
145
struct rrdeng_cmdqueue cmd_queue;
146
147
+ struct extent_cache xt_cache;
148
+
149
int error;
150
};
151
database/engine/rrdenginelib.h
+28
@@ -36,6 +36,34 @@ typedef uintptr_t rrdeng_stats_t;
36
37
#define rrd_stat_atomic_add(p, n) rrd_atomic_fetch_add(p, n)
38
39
+/* returns -1 if it didn't find the first cleared bit, the position otherwise. Starts from LSB. */
40
+static inline int find_first_zero(unsigned x)
41
+{
42
+ return ffs((int)(~x)) - 1;
43
+}
44
+
45
+/* Starts from LSB. */
46
+static inline uint8_t check_bit(unsigned x, size_t pos)
47
+{
48
+ return !!(x & (1 << pos));
49
+}
50
+
51
+/* Starts from LSB. val is 0 or 1 */
52
+static inline void modify_bit(unsigned *x, unsigned pos, uint8_t val)
53
+{
54
+ switch(val) {
55
+ case 0:
56
+ *x &= ~(1U << pos);
57
+ break;
58
+ case 1:
59
+ *x |= 1U << pos;
60
+ break;
61
+ default:
62
+ error("modify_bit() called with invalid argument.");
63
+ break;
64
+ }
65
+}
66
+
67
#define RRDENG_PATH_MAX (4096)
68
69
/* returns old *ptr value */