Fill chart gaps efficiently. (#6216)
Markos Fountoulakis committed
Jun 4, 2019 at 18:20 UTC
118534bd8b6c900c6b72144aedc253deeff7f06b
3 files changed
+76
-63
database/engine/rrdengineapi.c
+54
-63
@@ -77,6 +77,53 @@ static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr)
77
return has_only_empty_metrics;
78
}
79
80
+void rrdeng_store_metric_flush_current_page(RRDDIM *rd)
81
+{
82
+ struct rrdeng_collect_handle *handle;
83
+ struct rrdengine_instance *ctx;
84
+ struct rrdeng_page_descr *descr;
85
+
86
+ handle = &rd->state->handle.rrdeng;
87
+ ctx = handle->ctx;
88
+ descr = handle->descr;
89
+ if (unlikely(NULL == descr)) {
90
+ return;
91
+ }
92
+ if (likely(descr->page_length)) {
93
+ int ret, page_is_empty;
94
+
95
+#ifdef NETDATA_INTERNAL_CHECKS
96
+ rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1);
97
+#endif
98
+ if (handle->prev_descr) {
99
+ /* unpin old second page */
100
+ pg_cache_put(ctx, handle->prev_descr);
101
+ }
102
+ page_is_empty = page_has_only_empty_metrics(descr);
103
+ if (page_is_empty) {
104
+ debug(D_RRDENGINE, "Page has empty metrics only, deleting:");
105
+ if(unlikely(debug_flags & D_RRDENGINE))
106
+ print_page_cache_descr(descr);
107
+ pg_cache_put(ctx, descr);
108
+ pg_cache_punch_hole(ctx, descr, 1);
109
+ handle->prev_descr = NULL;
110
+ } else {
111
+ /* added 1 extra reference to keep 2 dirty pages pinned per metric, expected refcnt = 2 */
112
+ rrdeng_page_descr_mutex_lock(ctx, descr);
113
+ ret = pg_cache_try_get_unsafe(descr, 0);
114
+ rrdeng_page_descr_mutex_unlock(ctx, descr);
115
+ assert (1 == ret);
116
+
117
+ rrdeng_commit_page(ctx, descr, handle->page_correlation_id);
118
+ }
119
+ handle->prev_descr = descr;
120
+ } else {
121
+ free(descr->pg_cache_descr->page);
122
+ rrdeng_destroy_pg_cache_descr(ctx, descr->pg_cache_descr);
123
+ free(descr);
124
+ }
125
+ handle->descr = NULL;
126
+}
127
128
void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number number)
129
{
@@ -91,45 +138,13 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n
138
pg_cache = &ctx->pg_cache;
139
descr = handle->descr;
140
if (unlikely(NULL == descr || descr->page_length + sizeof(number) > RRDENG_BLOCK_SIZE)) {
94
- if (descr) {
95
- if (descr->page_length) {
96
- int ret, page_is_empty;
141
+ rrdeng_store_metric_flush_current_page(rd);
142
98
-#ifdef NETDATA_INTERNAL_CHECKS
99
- rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1);
100
-#endif
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
- }
122
- } else {
123
- free(descr->pg_cache_descr->page);
124
- rrdeng_destroy_pg_cache_descr(ctx, descr->pg_cache_descr);
125
- free(descr);
126
- handle->descr = NULL;
127
- }
128
- }
143
page = rrdeng_create_page(ctx, &handle->page_index->id, &descr);
144
assert(page);
131
- handle->prev_descr = handle->descr;
145
+
146
handle->descr = descr;
147
+
148
uv_rwlock_wrlock(&pg_cache->commited_page_index.lock);
149
handle->page_correlation_id = pg_cache->commited_page_index.latest_corr_id++;
150
uv_rwlock_wrunlock(&pg_cache->commited_page_index.lock);
@@ -158,37 +173,13 @@ void rrdeng_store_metric_finalize(RRDDIM *rd)
173
{
174
struct rrdeng_collect_handle *handle;
175
struct rrdengine_instance *ctx;
161
- struct rrdeng_page_descr *descr;
176
177
handle = &rd->state->handle.rrdeng;
178
ctx = handle->ctx;
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
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);
186
- }
187
- } else {
188
- free(descr->pg_cache_descr->page);
189
- rrdeng_destroy_pg_cache_descr(ctx, descr->pg_cache_descr);
190
- free(descr);
191
- }
179
+ rrdeng_store_metric_flush_current_page(rd);
180
+ if (handle->prev_descr) {
181
+ /* unpin old second page */
182
+ pg_cache_put(ctx, handle->prev_descr);
183
}
184
}
185
database/engine/rrdengineapi.h
+1
@@ -20,6 +20,7 @@ extern void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id,
20
extern void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle);
21
extern void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle);
22
extern void rrdeng_store_metric_init(RRDDIM *rd);
23
+extern void rrdeng_store_metric_flush_current_page(RRDDIM *rd);
24
extern void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number number);
25
extern void rrdeng_store_metric_finalize(RRDDIM *rd);
26
extern void rrdeng_load_metric_init(RRDDIM *rd, struct rrddim_query_handle *rrdimm_handle,
database/rrdset.c
+21
@@ -258,6 +258,11 @@ void rrdset_reset(RRDSET *st) {
258
rd->last_collected_time.tv_usec = 0;
259
rd->collections_counter = 0;
260
// memset(rd->values, 0, rd->entries * sizeof(storage_number));
261
+#ifdef ENABLE_DBENGINE
262
+ if (RRD_MEMORY_MODE_DBENGINE == st->rrd_memory_mode) {
263
+ rrdeng_store_metric_flush_current_page(rd);
264
+ }
265
+#endif
266
}
267
}
268
@@ -1273,6 +1278,22 @@ void rrdset_done(RRDSET *st) {
1278
first_entry = 1;
1279
}
1280
1281
+#ifdef ENABLE_DBENGINE
1282
+ // check if we will re-write the entire page
1283
+ if(unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE &&
1284
+ dt_usec(&st->last_collected_time, &st->last_updated) > (RRDENG_BLOCK_SIZE / sizeof(storage_number)) * update_every_ut)) {
1285
+ info("%s: too old data (last updated at %ld.%ld, last collected at %ld.%ld). Resetting it. Will not store the next entry.", st->name, st->last_updated.tv_sec, st->last_updated.tv_usec, st->last_collected_time.tv_sec, st->last_collected_time.tv_usec);
1286
+ rrdset_reset(st);
1287
+ rrdset_init_last_updated_time(st);
1288
+
1289
+ st->usec_since_last_update = update_every_ut;
1290
+
1291
+ // the first entry should not be stored
1292
+ store_this_entry = 0;
1293
+ first_entry = 1;
1294
+ }
1295
+#endif
1296
+
1297
// these are the 3 variables that will help us in interpolation
1298
// last_stored_ut = the last time we added a value to the storage
1299
// now_collect_ut = the time the current value has been collected