Add empty page detection in DB engine (#6173)
Markos Fountoulakis committed
May 30, 2019 at 15:27 UTC
2f72bfafba57cd30a7b5bcf1b5ccc3281d713759
5 files changed
+69
-29
database/engine/pagecache.c
+14
-13
@@ -221,7 +221,7 @@ static void pg_cache_reserve_pages(struct rrdengine_instance *ctx, unsigned numb
221
222
uv_rwlock_wrlock(&pg_cache->pg_cache_rwlock);
223
if (pg_cache->populated_pages + number >= ctx->max_cache_pages + 1)
224
- debug(D_RRDENGINE, "=================================\nPage cache full. Reserving %u pages.\n=================================",
224
+ debug(D_RRDENGINE, "==Page cache full. Reserving %u pages.==",
225
number);
226
while (pg_cache->populated_pages + number >= ctx->max_cache_pages + 1) {
227
if (!pg_cache_try_evict_one_page_unsafe(ctx)) {
@@ -263,7 +263,7 @@ static int pg_cache_try_reserve_pages(struct rrdengine_instance *ctx, unsigned n
263
uv_rwlock_wrlock(&pg_cache->pg_cache_rwlock);
264
if (pg_cache->populated_pages + number >= ctx->cache_pages_low_watermark + 1) {
265
debug(D_RRDENGINE,
266
- "=================================\nPage cache full. Trying to reserve %u pages.\n=================================",
266
+ "==Page cache full. Trying to reserve %u pages.==",
267
number);
268
do {
269
if (!pg_cache_try_evict_one_page_unsafe(ctx))
@@ -337,10 +337,7 @@ static int pg_cache_try_evict_one_page_unsafe(struct rrdengine_instance *ctx)
337
return 0;
338
}
339
340
-/*
341
- * TODO: last waiter frees descriptor ?
342
- */
343
-void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr)
340
+void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr, uint8_t remove_dirty)
341
{
342
struct page_cache *pg_cache = &ctx->pg_cache;
343
struct page_cache_descr *pg_cache_descr = NULL;
@@ -382,12 +379,14 @@ void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_desc
379
print_page_cache_descr(descr);
380
pg_cache_wait_event_unsafe(descr);
381
}
385
- /* even a locked page could be dirty */
386
- while (unlikely(pg_cache_descr->flags & RRD_PAGE_DIRTY)) {
387
- debug(D_RRDENGINE, "%s: Found dirty page, waiting for it to be flushed:", __func__);
388
- if (unlikely(debug_flags & D_RRDENGINE))
389
- print_page_cache_descr(descr);
390
- pg_cache_wait_event_unsafe(descr);
382
+ if (!remove_dirty) {
383
+ /* even a locked page could be dirty */
384
+ while (unlikely(pg_cache_descr->flags & RRD_PAGE_DIRTY)) {
385
+ debug(D_RRDENGINE, "%s: Found dirty page, waiting for it to be flushed:", __func__);
386
+ if (unlikely(debug_flags & D_RRDENGINE))
387
+ print_page_cache_descr(descr);
388
+ pg_cache_wait_event_unsafe(descr);
389
+ }
390
}
391
if (pg_cache_descr->flags & RRD_PAGE_POPULATED) {
392
/* only after locking can it be safely deleted from LRU */
@@ -403,7 +402,9 @@ void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_desc
402
rrdeng_destroy_pg_cache_descr(ctx, pg_cache_descr);
403
}
404
destroy:
406
- assert(0 == descr->pg_cache_descr_state);
405
+ if (!remove_dirty) {
406
+ assert(0 == descr->pg_cache_descr_state);
407
+ }
408
freez(descr);
409
pg_cache_update_metric_times(page_index);
410
}
database/engine/pagecache.h
+1
-1
@@ -145,7 +145,7 @@ extern void pg_cache_put_unsafe(struct rrdeng_page_descr *descr);
145
extern void pg_cache_put(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr);
146
extern void pg_cache_insert(struct rrdengine_instance *ctx, struct pg_cache_page_index *index,
147
struct rrdeng_page_descr *descr);
148
-extern void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr);
148
+extern void pg_cache_punch_hole(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr, uint8_t remove_dirty);
149
extern struct pg_cache_page_index *
150
pg_cache_preload(struct rrdengine_instance *ctx, uuid_t *id, usec_t start_time, usec_t end_time);
151
extern struct rrdeng_page_descr *
database/engine/rrdengine.c
+1
-1
@@ -482,7 +482,7 @@ static void delete_old_data(uv_work_t *req)
482
count = extent->number_of_pages;
483
for (i = 0 ; i < count ; ++i) {
484
descr = extent->pages[i];
485
- pg_cache_punch_hole(ctx, descr);
485
+ pg_cache_punch_hole(ctx, descr, 0);
486
}
487
next = extent->next;
488
free(extent);
database/engine/rrdengineapi.c
+51
-13
@@ -60,6 +60,24 @@ void rrdeng_store_metric_init(RRDDIM *rd)
60
handle->page_index = page_index;
61
}
62
63
+/* The page must be populated and referenced */
64
+static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr)
65
+{
66
+ unsigned i;
67
+ uint8_t has_only_empty_metrics = 1;
68
+ storage_number *page;
69
+
70
+ page = descr->pg_cache_descr->page;
71
+ for (i = 0 ; i < descr->page_length / sizeof(storage_number); ++i) {
72
+ if (SN_EMPTY_SLOT != page[i]) {
73
+ has_only_empty_metrics = 0;
74
+ break;
75
+ }
76
+ }
77
+ return has_only_empty_metrics;
78
+}
79
+
80
+
81
void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number number)
82
{
83
struct rrdeng_collect_handle *handle;
@@ -75,23 +93,32 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n
93
if (unlikely(NULL == descr || descr->page_length + sizeof(number) > RRDENG_BLOCK_SIZE)) {
94
if (descr) {
95
if (descr->page_length) {
78
- int ret;
96
+ int ret, page_is_empty;
97
98
#ifdef NETDATA_INTERNAL_CHECKS
99
rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1);
100
#endif
83
- /* added 1 extra reference to keep 2 dirty pages pinned per metric, expected refcnt = 2 */
84
- rrdeng_page_descr_mutex_lock(ctx, descr);
85
- ret = pg_cache_try_get_unsafe(descr, 0);
86
- rrdeng_page_descr_mutex_unlock(ctx, descr);
87
- assert (1 == ret);
88
-
89
- rrdeng_commit_page(ctx, descr, handle->page_correlation_id);
101
+ page_is_empty = page_has_only_empty_metrics(descr);
102
+ if (page_is_empty) {
103
+ debug(D_RRDENGINE, "Page has empty metrics only, deleting:");
104
+ if(unlikely(debug_flags & D_RRDENGINE))
105
+ print_page_cache_descr(descr);
106
+ pg_cache_put(ctx, descr);
107
+ pg_cache_punch_hole(ctx, descr, 1);
108
+ handle->descr = NULL;
109
+ } else {
110
+ /* added 1 extra reference to keep 2 dirty pages pinned per metric, expected refcnt = 2 */
111
+ rrdeng_page_descr_mutex_lock(ctx, descr);
112
+ ret = pg_cache_try_get_unsafe(descr, 0);
113
+ rrdeng_page_descr_mutex_unlock(ctx, descr);
114
+ assert (1 == ret);
115
+
116
+ rrdeng_commit_page(ctx, descr, handle->page_correlation_id);
117
+ }
118
if (handle->prev_descr) {
119
/* unpin old second page */
120
pg_cache_put(ctx, handle->prev_descr);
121
}
94
- handle->prev_descr = descr;
122
} else {
123
free(descr->pg_cache_descr->page);
124
rrdeng_destroy_pg_cache_descr(ctx, descr->pg_cache_descr);
@@ -138,10 +165,21 @@ void rrdeng_store_metric_finalize(RRDDIM *rd)
165
descr = handle->descr;
166
if (descr) {
167
if (descr->page_length) {
168
+ int page_is_empty;
169
+
170
#ifdef NETDATA_INTERNAL_CHECKS
171
rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1);
172
#endif
144
- rrdeng_commit_page(ctx, descr, handle->page_correlation_id);
173
+ page_is_empty = page_has_only_empty_metrics(descr);
174
+ if (page_is_empty) {
175
+ debug(D_RRDENGINE, "Page has empty metrics only, deleting:");
176
+ if(unlikely(debug_flags & D_RRDENGINE))
177
+ print_page_cache_descr(descr);
178
+ pg_cache_put(ctx, descr);
179
+ pg_cache_punch_hole(ctx, descr, 1);
180
+ } else {
181
+ rrdeng_commit_page(ctx, descr, handle->page_correlation_id);
182
+ }
183
if (handle->prev_descr) {
184
/* unpin old second page */
185
pg_cache_put(ctx, handle->prev_descr);
@@ -304,7 +342,7 @@ void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrde
342
pg_cache_descr->flags = RRD_PAGE_DIRTY /*| RRD_PAGE_LOCKED */ | RRD_PAGE_POPULATED /* | BEING_COLLECTED */;
343
pg_cache_descr->refcnt = 1;
344
307
- debug(D_RRDENGINE, "-----------------\nCreated new page:\n-----------------");
345
+ debug(D_RRDENGINE, "Created new page:");
346
if(unlikely(debug_flags & D_RRDENGINE))
347
print_page_cache_descr(descr);
348
rrdeng_page_descr_mutex_unlock(ctx, descr);
@@ -340,7 +378,7 @@ void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **
378
struct rrdeng_page_descr *descr;
379
struct page_cache_descr *pg_cache_descr;
380
343
- debug(D_RRDENGINE, "----------------------\nReading existing page:\n----------------------");
381
+ debug(D_RRDENGINE, "Reading existing page:");
382
descr = pg_cache_lookup(ctx, NULL, id, INVALID_TIME);
383
if (NULL == descr) {
384
*handle = NULL;
@@ -359,7 +397,7 @@ void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_i
397
struct rrdeng_page_descr *descr;
398
struct page_cache_descr *pg_cache_descr;
399
362
- debug(D_RRDENGINE, "----------------------\nReading existing page:\n----------------------");
400
+ debug(D_RRDENGINE, "Reading existing page:");
401
descr = pg_cache_lookup(ctx, NULL, id, point_in_time);
402
if (NULL == descr) {
403
*handle = NULL;
database/engine/rrdenginelib.c
+2
-1
@@ -23,8 +23,9 @@ void print_page_cache_descr(struct rrdeng_page_descr *descr)
23
} else {
24
pos += snprintfz(str + pos, BUFSIZE - pos, "%"PRIu64, descr->extent->offset);
25
}
26
+
27
snprintfz(str + pos, BUFSIZE - pos, " flags:0x%2.2lX refcnt:%u\n\n", pg_cache_descr->flags, pg_cache_descr->refcnt);
27
- fputs(str, stderr);
28
+ debug(D_RRDENGINE, "%s", str);
29
}
30
31
void print_page_descr(struct rrdeng_page_descr *descr)