summaryrefslogtreecommitdiffstats
path: root/database/engine/journalfile.c
diff options
context:
space:
mode:
Diffstat (limited to 'database/engine/journalfile.c')
-rw-r--r--database/engine/journalfile.c113
1 files changed, 80 insertions, 33 deletions
diff --git a/database/engine/journalfile.c b/database/engine/journalfile.c
index dc61f569d..500dd7880 100644
--- a/database/engine/journalfile.c
+++ b/database/engine/journalfile.c
@@ -17,7 +17,7 @@ static void flush_transaction_buffer_cb(uv_fs_t* req)
}
uv_fs_req_cleanup(req);
- free(io_descr->buf);
+ posix_memfree(io_descr->buf);
freez(io_descr);
}
@@ -225,7 +225,7 @@ int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdeng
rrd_stat_atomic_add(&global_io_errors, 1);
}
uv_fs_req_cleanup(&req);
- free(superblock);
+ posix_memfree(superblock);
if (ret < 0) {
destroy_journal_file(journalfile, datafile);
return ret;
@@ -268,7 +268,7 @@ static int check_journal_file_superblock(uv_file file)
ret = 0;
}
error:
- free(superblock);
+ posix_memfree(superblock);
return ret;
}
@@ -311,20 +311,46 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
}
continue;
}
- uint64_t start_time = jf_metric_data->descr[i].start_time;
- uint64_t end_time = jf_metric_data->descr[i].end_time;
+ uint64_t start_time_ut = jf_metric_data->descr[i].start_time_ut;
+ uint64_t end_time_ut = jf_metric_data->descr[i].end_time_ut;
+ size_t entries = jf_metric_data->descr[i].page_length / page_type_size[page_type];
+ time_t update_every_s = (entries > 1) ? ((end_time_ut - start_time_ut) / USEC_PER_SEC / (entries - 1)) : 0;
+
+ if (unlikely(start_time_ut > end_time_ut)) {
+ ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].counter++;
+ if(ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].latest_end_time_ut < end_time_ut)
+ ctx->load_errors[LOAD_ERRORS_PAGE_FLIPPED_TIME].latest_end_time_ut = end_time_ut;
+ continue;
+ }
- if (unlikely(start_time > end_time)) {
- error("Invalid page encountered, start time %lu > end time %lu", start_time , end_time );
+ if (unlikely(start_time_ut == end_time_ut && entries != 1)) {
+ ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].counter++;
+ if(ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].latest_end_time_ut < end_time_ut)
+ ctx->load_errors[LOAD_ERRORS_PAGE_EQUAL_TIME].latest_end_time_ut = end_time_ut;
continue;
}
- if (unlikely(start_time == end_time)) {
- size_t entries = jf_metric_data->descr[i].page_length / page_type_size[page_type];
- if (unlikely(entries > 1)) {
- error("Invalid page encountered, start time %lu = end time but %zu entries were found", start_time, entries);
- continue;
- }
+ if (unlikely(!entries)) {
+ ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].counter++;
+ if(ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].latest_end_time_ut < end_time_ut)
+ ctx->load_errors[LOAD_ERRORS_PAGE_ZERO_ENTRIES].latest_end_time_ut = end_time_ut;
+ continue;
+ }
+
+ if(entries > 1 && update_every_s == 0) {
+ ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].counter++;
+ if(ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].latest_end_time_ut < end_time_ut)
+ ctx->load_errors[LOAD_ERRORS_PAGE_UPDATE_ZERO].latest_end_time_ut = end_time_ut;
+ continue;
+ }
+
+ if(start_time_ut + update_every_s * USEC_PER_SEC * (entries - 1) != end_time_ut) {
+ ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].counter++;
+ if(ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].latest_end_time_ut < end_time_ut)
+ ctx->load_errors[LOAD_ERRORS_PAGE_FLEXY_TIME].latest_end_time_ut = end_time_ut;
+
+ // let this be
+ // end_time_ut = start_time_ut + update_every_s * USEC_PER_SEC * (entries - 1);
}
temp_id = (uuid_t *)jf_metric_data->descr[i].uuid;
@@ -340,7 +366,7 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
uv_rwlock_wrlock(&pg_cache->metrics_index.lock);
PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, temp_id, sizeof(uuid_t), PJE0);
fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */
- *PValue = page_index = create_page_index(temp_id);
+ *PValue = page_index = create_page_index(temp_id, ctx);
page_index->prev = pg_cache->metrics_index.last_page_index;
pg_cache->metrics_index.last_page_index = page_index;
uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
@@ -348,21 +374,32 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
descr = pg_cache_create_descr();
descr->page_length = jf_metric_data->descr[i].page_length;
- descr->start_time = start_time;
- descr->end_time = end_time;
+ descr->start_time_ut = start_time_ut;
+ descr->end_time_ut = end_time_ut;
+ descr->update_every_s = (update_every_s > 0) ? (uint32_t)update_every_s : (page_index->latest_update_every_s);
descr->id = &page_index->id;
descr->extent = extent;
descr->type = page_type;
extent->pages[valid_pages++] = descr;
pg_cache_insert(ctx, page_index, descr);
+
+ if(page_index->latest_time_ut == descr->end_time_ut)
+ page_index->latest_update_every_s = descr->update_every_s;
+
+ if(descr->update_every_s == 0)
+ fatal(
+ "DBENGINE: page descriptor update every is zero, end_time_ut = %llu, start_time_ut = %llu, entries = %zu",
+ (unsigned long long)end_time_ut, (unsigned long long)start_time_ut, entries);
}
extent->number_of_pages = valid_pages;
if (likely(valid_pages))
df_extent_insert(extent);
- else
+ else {
freez(extent);
+ ctx->load_errors[LOAD_ERRORS_DROPPED_EXTENT].counter++;
+ }
}
/*
@@ -442,27 +479,30 @@ static uint64_t iterate_transactions(struct rrdengine_instance *ctx, struct rrde
//data_file_size = journalfile->datafile->pos; TODO: utilize this?
max_id = 1;
- ret = posix_memalign((void *)&buf, RRDFILE_ALIGNMENT, READAHEAD_BYTES);
- if (unlikely(ret)) {
- fatal("posix_memalign:%s", strerror(ret));
+ bool journal_is_mmapped = (journalfile->data != NULL);
+ if (unlikely(!journal_is_mmapped)) {
+ ret = posix_memalign((void *)&buf, RRDFILE_ALIGNMENT, READAHEAD_BYTES);
+ if (unlikely(ret))
+ fatal("posix_memalign:%s", strerror(ret));
}
-
+ else
+ buf = journalfile->data + sizeof(struct rrdeng_jf_sb);
for (pos = sizeof(struct rrdeng_jf_sb) ; pos < file_size ; pos += READAHEAD_BYTES) {
size_bytes = MIN(READAHEAD_BYTES, file_size - pos);
- iov = uv_buf_init(buf, size_bytes);
- ret = uv_fs_read(NULL, &req, file, &iov, 1, pos, NULL);
- if (ret < 0) {
- error("uv_fs_read: pos=%"PRIu64", %s", pos, uv_strerror(ret));
+ if (unlikely(!journal_is_mmapped)) {
+ iov = uv_buf_init(buf, size_bytes);
+ ret = uv_fs_read(NULL, &req, file, &iov, 1, pos, NULL);
+ if (ret < 0) {
+ error("uv_fs_read: pos=%" PRIu64 ", %s", pos, uv_strerror(ret));
+ uv_fs_req_cleanup(&req);
+ goto skip_file;
+ }
+ fatal_assert(req.result >= 0);
uv_fs_req_cleanup(&req);
- goto skip_file;
+ ++ctx->stats.io_read_requests;
+ ctx->stats.io_read_bytes += size_bytes;
}
- fatal_assert(req.result >= 0);
- uv_fs_req_cleanup(&req);
- ctx->stats.io_read_bytes += size_bytes;
- ++ctx->stats.io_read_requests;
- //pos_i = pos;
- //while (pos_i < pos + size_bytes) {
for (pos_i = 0 ; pos_i < size_bytes ; ) {
unsigned max_size;
@@ -475,9 +515,12 @@ static uint64_t iterate_transactions(struct rrdengine_instance *ctx, struct rrde
pos_i += ret;
max_id = MAX(max_id, id);
}
+ if (likely(journal_is_mmapped))
+ buf += size_bytes;
}
skip_file:
- free(buf);
+ if (unlikely(!journal_is_mmapped))
+ posix_memfree(buf);
return max_id;
}
@@ -512,12 +555,16 @@ int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfi
journalfile->file = file;
journalfile->pos = file_size;
+ journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0);
+ info("Loading journal file \"%s\" using %s.", path, journalfile->data?"MMAP":"uv_fs_read");
max_id = iterate_transactions(ctx, journalfile);
ctx->commit_log.transaction_id = MAX(ctx->commit_log.transaction_id, max_id + 1);
info("Journal file \"%s\" loaded (size:%"PRIu64").", path, file_size);
+ if (likely(journalfile->data))
+ netdata_munmap(journalfile->data, file_size);
return 0;
error:
generated by cgit v1.2.3 (git 2.39.1) at 2024-05-06 06:28:30 +0000