@cryptotaxi247 / netdata-1 / commits / dc9f81ccf

DBENGINE v2 - improvements part 6 (#14299)

* query preparation runs before extent reads * populate mrg in parallel * fix formatting warning * first search for a metric then add it if it does not exist * Revert "first search for a metric then add it if it does not exist" This reverts commit 4afa6461fcce859d03f1c9cf56dd3b5933ee5ebc. * Revert "fix formatting warning" This reverts commit 49473493f7f1c3399b5635a573d3c6ed2b6e46f3. * Revert "populate mrg in parallel" This reverts commit a40166708d4222f6329904f109114c47c44ca666. * merge journalfiles metrics before committing them to MRG * Revert "merge journalfiles metrics before committing them to MRG" This reverts commit 50c8934e23a0a09ea4da80e3f88290e46496ad92. * Revert "Revert "populate mrg in parallel"" This reverts commit f4c149d2ab7a8c9af24a10f95438a0d662a5cf8a. * Revert "Revert "fix formatting warning"" This reverts commit 78298ff9efc49806ded029f5f1e868cc42e8f6eb. * Revert "Revert "first search for a metric then add it if it does not exist"" This reverts commit 997b9c813b290882ba18a8c44bf73f9ee5480adf. * preload first and last journal files v2 * fix formatting warning * parallel loading of tiers; cleanup of ctx structures * use half the cores * add partitions to metrics registry * revert accidental change * parallel processing according to MRG partitions; dont recalculate retention on exit

Costa Tsaousis committed Jan 20, 2023 at 23:56 UTC dc9f81ccfe611410f5a710dafcc14c6c9f030aa2
15 files changed +471 -286
daemon/main.c
+10 -2
@@ -327,6 +327,14 @@ void netdata_cleanup_and_exit(int ret) {
327 snprintfz(agent_incomplete_shutdown_file, FILENAME_MAX, "%s/.agent_incomplete_shutdown", netdata_configured_varlib_dir);
328 (void) rename(agent_crash_file, agent_incomplete_shutdown_file);
329
330 +#ifdef ENABLE_DBENGINE
331 + if(dbengine_enabled) {
332 + delta_shutdown_time("dbengine exit mode");
333 + for (size_t tier = 0; tier < storage_tiers; tier++)
334 + rrdeng_exit_mode(multidb_ctx[tier]);
335 + }
336 +#endif
337 +
338 delta_shutdown_time("disable maintenance, new queries, new web requests, new streaming connections and aclk");
339
340 service_signal_exit(
@@ -604,7 +612,7 @@ static void set_nofile_limit(struct rlimit *rl) {
612 // make the soft/hard limits equal
613 rl->rlim_cur = rl->rlim_max;
614 if (setrlimit(RLIMIT_NOFILE, rl) != 0) {
607 - error("setrlimit(RLIMIT_NOFILE, { %llu, %llu }) failed", rl->rlim_cur, rl->rlim_max);
615 + error("setrlimit(RLIMIT_NOFILE, { %zu, %zu }) failed", (size_t)rl->rlim_cur, (size_t)rl->rlim_max);
616 }
617
618 // sanity check to make sure we have enough file descriptors available to open
@@ -614,7 +622,7 @@ static void set_nofile_limit(struct rlimit *rl) {
622 }
623
624 if (rl->rlim_cur < 1024)
617 - error("Number of open file descriptors allowed for this process is too low (RLIMIT_NOFILE=%zu)", (size_t) rl->rlim_cur);
625 + error("Number of open file descriptors allowed for this process is too low (RLIMIT_NOFILE=%zu)", (size_t)rl->rlim_cur);
626 }
627
628 void cancel_main_threads() {
database/engine/datafile.c
+23 -26
@@ -120,7 +120,7 @@ bool datafile_acquire_for_deletion(struct rrdengine_datafile *df) {
120 "%zu clean and %zu hot open cache pages "
121 "- will be deleted shortly "
122 "(scanned open cache in %llu usecs)",
123 - df->fileno, df->ctx->tier,
123 + df->fileno, df->ctx->config.tier,
124 df->users.lockers,
125 df->users.lockers_by_reason[DATAFILE_ACQUIRE_OPEN_CACHE],
126 df->users.lockers_by_reason[DATAFILE_ACQUIRE_PAGE_DETAILS],
@@ -137,7 +137,7 @@ bool datafile_acquire_for_deletion(struct rrdengine_datafile *df) {
137 "%zu clean and %zu hot open cache pages "
138 "- will be deleted now "
139 "(scanned open cache in %llu usecs)",
140 - df->fileno, df->ctx->tier,
140 + df->fileno, df->ctx->config.tier,
141 df->users.lockers,
142 df->users.lockers_by_reason[DATAFILE_ACQUIRE_OPEN_CACHE],
143 df->users.lockers_by_reason[DATAFILE_ACQUIRE_PAGE_DETAILS],
@@ -151,7 +151,7 @@ bool datafile_acquire_for_deletion(struct rrdengine_datafile *df) {
151 "has %u lockers (oc:%u, pd:%u), "
152 "%zu clean and %zu hot open cache pages "
153 "(scanned open cache in %llu usecs)",
154 - df->fileno, df->ctx->tier,
154 + df->fileno, df->ctx->config.tier,
155 df->users.lockers,
156 df->users.lockers_by_reason[DATAFILE_ACQUIRE_OPEN_CACHE],
157 df->users.lockers_by_reason[DATAFILE_ACQUIRE_PAGE_DETAILS],
@@ -168,7 +168,7 @@ bool datafile_acquire_for_deletion(struct rrdengine_datafile *df) {
168 void generate_datafilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
169 {
170 (void) snprintfz(str, maxlen, "%s/" DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION,
171 - datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
171 + datafile->ctx->config.dbfiles_path, datafile->tier, datafile->fileno);
172 }
173
174 int close_data_file(struct rrdengine_datafile *datafile)
@@ -407,16 +407,16 @@ static int scan_data_files(struct rrdengine_instance *ctx)
407 struct rrdengine_datafile **datafiles, *datafile;
408 struct rrdengine_journalfile *journalfile;
409
410 - ret = uv_fs_scandir(NULL, &req, ctx->dbfiles_path, 0, NULL);
410 + ret = uv_fs_scandir(NULL, &req, ctx->config.dbfiles_path, 0, NULL);
411 if (ret < 0) {
412 fatal_assert(req.result < 0);
413 uv_fs_req_cleanup(&req);
414 - error("DBENGINE: uv_fs_scandir(%s): %s", ctx->dbfiles_path, uv_strerror(ret));
414 + error("DBENGINE: uv_fs_scandir(%s): %s", ctx->config.dbfiles_path, uv_strerror(ret));
415 ++ctx->stats.fs_errors;
416 rrd_stat_atomic_add(&global_fs_errors, 1);
417 return ret;
418 }
419 - info("DBENGINE: found %d files in path %s", ret, ctx->dbfiles_path);
419 + info("DBENGINE: found %d files in path %s", ret, ctx->config.dbfiles_path);
420
421 datafiles = callocz(MIN(ret, MAX_DATAFILES), sizeof(*datafiles));
422 for (matched_files = 0 ; UV_EOF != uv_fs_scandir_next(&req, &dent) && matched_files < MAX_DATAFILES ; ) {
@@ -437,7 +437,7 @@ static int scan_data_files(struct rrdengine_instance *ctx)
437 }
438 qsort(datafiles, matched_files, sizeof(*datafiles), scan_data_files_cmp);
439 /* TODO: change this when tiering is implemented */
440 - ctx->last_fileno = datafiles[matched_files - 1]->fileno;
440 + ctx->atomic.last_fileno = datafiles[matched_files - 1]->fileno;
441
442 for (failed_to_load = 0, i = 0 ; i < matched_files ; ++i) {
443 uint8_t must_delete_pair = 0;
@@ -475,7 +475,7 @@ static int scan_data_files(struct rrdengine_instance *ctx)
475 }
476
477 datafile_list_insert(ctx, datafile);
478 - ctx->disk_space += datafile->pos + journalfile->pos;
478 + ctx_current_disk_space_increase(ctx, datafile->pos + journalfile->pos);
479 }
480 matched_files -= failed_to_load;
481 freez(datafiles);
@@ -490,11 +490,11 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx)
490
491 struct rrdengine_datafile *datafile;
492 struct rrdengine_journalfile *journalfile;
493 - unsigned fileno = __atomic_load_n(&ctx->last_fileno, __ATOMIC_RELAXED) + 1;
493 + unsigned fileno = ctx_last_fileno_get(ctx) + 1;
494 int ret;
495 char path[RRDENG_PATH_MAX];
496
497 - info("DBENGINE: creating new data and journal files in path %s", ctx->dbfiles_path);
497 + info("DBENGINE: creating new data and journal files in path %s", ctx->config.dbfiles_path);
498 datafile = datafile_alloc_and_init(ctx, 1, fileno);
499 ret = create_data_file(datafile);
500 if(ret)
@@ -512,9 +512,8 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx)
512 info("DBENGINE: created journal file \"%s\".", path);
513
514 datafile_list_insert(ctx, datafile);
515 - ctx->disk_space += datafile->pos + journalfile->pos;
516 -
517 - __atomic_add_fetch(&ctx->last_fileno, 1, __ATOMIC_RELAXED);
515 + ctx_current_disk_space_increase(ctx, datafile->pos + journalfile->pos);
516 + ctx_last_fileno_increment(ctx);
517
518 return 0;
519
@@ -535,26 +534,24 @@ int init_data_files(struct rrdengine_instance *ctx)
534 int ret;
535
536 fatal_assert(0 == uv_rwlock_init(&ctx->datafiles.rwlock));
538 - __atomic_store_n(&ctx->journal_initialization, true, __ATOMIC_RELAXED);
537 ret = scan_data_files(ctx);
538 if (ret < 0) {
541 - error("DBENGINE: failed to scan path \"%s\".", ctx->dbfiles_path);
539 + error("DBENGINE: failed to scan path \"%s\".", ctx->config.dbfiles_path);
540 return ret;
541 } else if (0 == ret) {
544 - info("DBENGINE: data files not found, creating in path \"%s\".", ctx->dbfiles_path);
545 - ctx->last_fileno = 0;
542 + info("DBENGINE: data files not found, creating in path \"%s\".", ctx->config.dbfiles_path);
543 + ctx->atomic.last_fileno = 0;
544 ret = create_new_datafile_pair(ctx);
545 if (ret) {
548 - error("DBENGINE: failed to create data and journal files in path \"%s\".", ctx->dbfiles_path);
546 + error("DBENGINE: failed to create data and journal files in path \"%s\".", ctx->config.dbfiles_path);
547 return ret;
548 }
549 }
552 - else if(ctx->create_new_datafile_pair)
550 + else if(ctx->loading.create_new_datafile_pair)
551 create_new_datafile_pair(ctx);
552
553 pgc_reset_hot_max(open_cache);
556 - ctx->create_new_datafile_pair = false;
557 - __atomic_store_n(&ctx->journal_initialization, false, __ATOMIC_RELAXED);
554 + ctx->loading.create_new_datafile_pair = false;
555 return 0;
556 }
557
@@ -569,9 +566,9 @@ void finalize_data_files(struct rrdengine_instance *ctx)
566 logged = false;
567 if(datafile == ctx->datafiles.first->prev) {
568 // this is the last file
572 - while(__atomic_load_n(&ctx->worker_config.atomics.extents_currently_being_flushed, __ATOMIC_RELAXED)) {
569 + while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED)) {
570 if(!logged) {
574 - info("Waiting for inflight flush to finish on tier %d to close last datafile %u...", ctx->tier, datafile->fileno);
571 + info("Waiting for inflight flush to finish on tier %d to close last datafile %u...", ctx->config.tier, datafile->fileno);
572 logged = true;
573 }
574 sleep_usec(100 * USEC_PER_MS);
@@ -581,7 +578,7 @@ void finalize_data_files(struct rrdengine_instance *ctx)
578 logged = false;
579 while(!datafile_acquire_for_deletion(datafile) && datafile != ctx->datafiles.first->prev) {
580 if(!logged) {
584 - info("Waiting to acquire data file %u of tier %d to close it...", datafile->fileno, ctx->tier);
581 + info("Waiting to acquire data file %u of tier %d to close it...", datafile->fileno, ctx->config.tier);
582 logged = true;
583 }
584 sleep_usec(100 * USEC_PER_MS);
@@ -598,7 +595,7 @@ void finalize_data_files(struct rrdengine_instance *ctx)
595 netdata_spinlock_unlock(&datafile->writers.spinlock);
596 uv_rwlock_wrunlock(&ctx->datafiles.rwlock);
597 if(!logged) {
601 - info("Waiting for writers to data file %u of tier %d to finish...", datafile->fileno, ctx->tier);
598 + info("Waiting for writers to data file %u of tier %d to finish...", datafile->fileno, ctx->config.tier);
599 logged = true;
600 }
601 sleep_usec(100 * USEC_PER_MS);
database/engine/datafile.h
+5 -5
@@ -45,6 +45,11 @@ struct rrdengine_datafile {
45 struct rrdengine_datafile *prev;
46 struct rrdengine_datafile *next;
47
48 + struct {
49 + SPINLOCK spinlock;
50 + bool populated;
51 + } populate_mrg;
52 +
53 struct {
54 SPINLOCK spinlock;
55 size_t running;
@@ -70,11 +75,6 @@ bool datafile_acquire(struct rrdengine_datafile *df, DATAFILE_ACQUIRE_REASONS re
75 void datafile_release(struct rrdengine_datafile *df, DATAFILE_ACQUIRE_REASONS reason);
76 bool datafile_acquire_for_deletion(struct rrdengine_datafile *df);
77
73 -struct rrdengine_datafile_list {
74 - uv_rwlock_t rwlock;
75 - struct rrdengine_datafile *first; /* oldest - the newest with ->first->prev */
76 -};
77 -
78 void datafile_list_insert(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile);
79 void datafile_list_delete_unsafe(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile);
80 void generate_datafilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
database/engine/journalfile.c
+72 -58
@@ -17,7 +17,7 @@ static void update_metric_retention_and_granularity_by_uuid(
17 last_time_s = now_s;
18 }
19
20 - if(unlikely(first_time_s > last_time_s)) {
20 + if (unlikely(first_time_s > last_time_s)) {
21 error_limit_static_global_var(erl, 1, 0);
22 error_limit(&erl, "DBENGINE JV2: wrong first time on-disk (%ld - %ld, now %ld), "
23 "fixing first time to last time",
@@ -26,23 +26,25 @@ static void update_metric_retention_and_granularity_by_uuid(
26 first_time_s = last_time_s;
27 }
28
29 - if(unlikely(first_time_s == 0 || last_time_s == 0)) {
29 + if (unlikely(first_time_s == 0 || last_time_s == 0)) {
30 error_limit_static_global_var(erl, 1, 0);
31 error_limit(&erl, "DBENGINE JV2: zero on-disk timestamps (%ld - %ld, now %ld), "
32 "using them as-is",
33 first_time_s, last_time_s, now_s);
34 }
35
36 - MRG_ENTRY entry = {
37 - .section = (Word_t)ctx,
38 - .first_time_s = first_time_s,
39 - .last_time_s = last_time_s,
40 - .latest_update_every_s = update_every_s
41 - };
42 - uuid_copy(entry.uuid, *uuid);
43 -
44 - bool added;
45 - METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
36 + bool added = false;
37 + METRIC *metric = mrg_metric_get_and_acquire(main_mrg, uuid, (Word_t) ctx);
38 + if (!metric) {
39 + MRG_ENTRY entry = {
40 + .section = (Word_t) ctx,
41 + .first_time_s = first_time_s,
42 + .last_time_s = last_time_s,
43 + .latest_update_every_s = update_every_s
44 + };
45 + uuid_copy(entry.uuid, *uuid);
46 + metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
47 + }
48
49 if (likely(!added))
50 mrg_metric_expand_retention(main_mrg, metric, first_time_s, last_time_s, update_every_s);
@@ -70,7 +72,7 @@ static void wal_flush_transaction_buffer_cb(uv_fs_t* req)
72 uv_fs_req_cleanup(req);
73 wal_release(wal);
74
73 - __atomic_sub_fetch(&ctx->worker_config.atomics.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
75 + __atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
76
77 worker_is_idle();
78 }
@@ -100,7 +102,7 @@ void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengi
102 journalfile->pos, wal_flush_transaction_buffer_cb);
103 fatal_assert(-1 != ret);
104 journalfile->pos += wal->buf_size;
103 - ctx->disk_space += wal->buf_size;
105 + ctx_current_disk_space_increase(ctx, wal->buf_size);
106 ctx->stats.io_write_bytes += wal->buf_size;
107 ++ctx->stats.io_write_requests;
108 }
@@ -108,13 +110,13 @@ void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengi
110 void journalfile_v2_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
111 {
112 (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION_V2,
111 - datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
113 + datafile->ctx->config.dbfiles_path, datafile->tier, datafile->fileno);
114 }
115
116 void journalfile_v1_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
117 {
118 (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION,
117 - datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
119 + datafile->ctx->config.dbfiles_path, datafile->tier, datafile->fileno);
120 }
121
122 static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengine_journalfile *journalfile, size_t *data_size) {
@@ -213,10 +215,10 @@ static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *jo
215 return unmounted;
216 }
217
216 -void journalfile_v2_data_unmount_cleanup(time_t now_s, int storage_tiers) {
218 +void journalfile_v2_data_unmount_cleanup(time_t now_s) {
219 // DO NOT WAIT ON ANY LOCK!!!
220
219 - for(size_t tier = 0; tier < storage_tiers ;tier++) {
221 + for(size_t tier = 0; tier < (size_t)storage_tiers ;tier++) {
222 struct rrdengine_instance *ctx = multidb_ctx[tier];
223 if(!ctx) continue;
224
@@ -907,6 +909,46 @@ static int journalfile_v2_validate(void *data_start, size_t journal_v2_file_size
909 return 0;
910 }
911
912 +void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile) {
913 + usec_t started_ut = now_monotonic_usec();
914 +
915 + size_t data_size = 0;
916 + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(journalfile, &data_size, 0, 0);
917 + if(!j2_header)
918 + return;
919 +
920 + uint8_t *data_start = (uint8_t *)j2_header;
921 + uint32_t entries = j2_header->metric_count;
922 +
923 + struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
924 + time_t header_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
925 + time_t now_s = now_realtime_sec();
926 + for (size_t i=0; i < entries; i++) {
927 + time_t start_time_s = header_start_time_s + metric->delta_start_s;
928 + time_t end_time_s = header_start_time_s + metric->delta_end_s;
929 + time_t update_every_s = (metric->entries > 1) ? ((end_time_s - start_time_s) / (entries - 1)) : 0;
930 + update_metric_retention_and_granularity_by_uuid(
931 + ctx, &metric->uuid, start_time_s, end_time_s, update_every_s, now_s);
932 +
933 +#ifdef NETDATA_INTERNAL_CHECKS
934 + struct journal_page_header *metric_list_header = (void *) (data_start + metric->page_offset);
935 + fatal_assert(uuid_compare(metric_list_header->uuid, metric->uuid) == 0);
936 + fatal_assert(metric->entries == metric_list_header->entries);
937 +#endif
938 + metric++;
939 + }
940 +
941 + journalfile_v2_data_release(journalfile);
942 + usec_t ended_ut = now_monotonic_usec();
943 +
944 + info("DBENGINE: journal v2 of tier %d, datafile %u populated, size: %0.2f MiB, metrics: %0.2f k, %0.2f ms"
945 + , ctx->config.tier, journalfile->datafile->fileno
946 + , (double)data_size / 1024 / 1024
947 + , (double)entries / 1000
948 + , ((double)(ended_ut - started_ut) / USEC_PER_MS)
949 + );
950 +}
951 +
952 int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
953 {
954 int ret, fd;
@@ -983,38 +1025,15 @@ int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journal
1025 return 1;
1026 }
1027
986 - madvise_dontfork(data_start, journal_v2_file_size);
987 - madvise_dontdump(data_start, journal_v2_file_size);
988 -
989 - usec_t mrg_start_ut = now_monotonic_usec();
990 - struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
991 - time_t header_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
992 - time_t now_s = now_realtime_sec();
993 - for (size_t i=0; i < entries; i++) {
994 - time_t start_time_s = header_start_time_s + metric->delta_start_s;
995 - time_t end_time_s = header_start_time_s + metric->delta_end_s;
996 - time_t update_every_s = (metric->entries > 1) ? ((end_time_s - start_time_s) / (entries - 1)) : 0;
997 - update_metric_retention_and_granularity_by_uuid(
998 - ctx, &metric->uuid, start_time_s, end_time_s, update_every_s, now_s);
999 -
1000 -#ifdef NETDATA_INTERNAL_CHECKS
1001 - struct journal_page_header *metric_list_header = (void *) (data_start + metric->page_offset);
1002 - fatal_assert(uuid_compare(metric_list_header->uuid, metric->uuid) == 0);
1003 - fatal_assert(metric->entries == metric_list_header->entries);
1004 -#endif
1005 - metric++;
1006 - }
1007 -
1028 usec_t finished_ut = now_monotonic_usec();
1029
1030 info("DBENGINE: journal v2 '%s' loaded, size: %0.2f MiB, metrics: %0.2f k, "
1011 - "mmap: %0.2f ms, validate: %0.2f ms, populate: %0.2f ms"
1031 + "mmap: %0.2f ms, validate: %0.2f ms"
1032 , path_v2
1033 , (double)journal_v2_file_size / 1024 / 1024
1034 , (double)entries / 1000
1035 , ((double)(validation_start_ut - mmap_start_ut) / USEC_PER_MS)
1016 - , ((double)(mrg_start_ut - validation_start_ut) / USEC_PER_MS)
1017 - , ((double)(finished_ut - mrg_start_ut) / USEC_PER_MS)
1036 + , ((double)(finished_ut - validation_start_ut) / USEC_PER_MS)
1037 );
1038
1039 // Initialize the journal file to be able to access the data
@@ -1350,7 +1369,7 @@ void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1369 journalfile_v2_data_set(journalfile, fd_v2, data_start, total_file_size);
1370
1371 internal_error(true, "DBENGINE: ACTIVATING NEW INDEX JNL %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1353 - ctx->disk_space += total_file_size;
1372 + ctx_current_disk_space_increase(ctx, total_file_size);
1373 freez(uuid_list);
1374 return;
1375 }
@@ -1370,13 +1389,13 @@ void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno
1389
1390 int ret = truncate(path, (long) resize_file_to);
1391 if (ret < 0) {
1373 - ctx->disk_space += total_file_size;
1392 + ctx_current_disk_space_increase(ctx, total_file_size);
1393 ++ctx->stats.fs_errors;
1394 rrd_stat_atomic_add(&global_fs_errors, 1);
1395 error("DBENGINE: failed to resize file '%s'", path);
1396 }
1397 else
1379 - ctx->disk_space += sizeof(struct journal_v2_header);
1398 + ctx_current_disk_space_increase(ctx, sizeof(struct journal_v2_header));
1399 }
1400
1401 int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
@@ -1389,7 +1408,7 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil
1408 char path[RRDENG_PATH_MAX];
1409
1410 // Do not try to load the latest file (always rebuild and live migrate)
1392 - if (datafile->fileno != ctx->last_fileno) {
1411 + if (datafile->fileno != ctx_last_fileno_get(ctx)) {
1412 if (!journalfile_v2_load(ctx, journalfile, datafile)) {
1413 // unmap_journal_file(journalfile);
1414 return 0;
@@ -1422,28 +1441,28 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil
1441 journalfile->file = file;
1442 journalfile->pos = file_size;
1443
1425 - journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0, !(datafile->fileno == ctx->last_fileno), NULL);
1444 + journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0, !(datafile->fileno == ctx_last_fileno_get(ctx)), NULL);
1445 info("DBENGINE: loading journal file '%s' using %s.", path, journalfile->data?"MMAP":"uv_fs_read");
1446
1447 max_id = journalfile_iterate_transactions(ctx, journalfile);
1448
1430 - ctx->commit_log.transaction_id = MAX(ctx->commit_log.transaction_id, max_id + 1);
1449 + __atomic_store_n(&ctx->atomic.transaction_id, MAX(__atomic_load_n(&ctx->atomic.transaction_id, __ATOMIC_RELAXED), max_id + 1), __ATOMIC_RELAXED);
1450
1451 info("DBENGINE: journal file '%s' loaded (size:%"PRIu64").", path, file_size);
1452 if (likely(journalfile->data))
1453 netdata_munmap(journalfile->data, file_size);
1454
1436 - bool is_last_file = (ctx->last_fileno == journalfile->datafile->fileno);
1455 + bool is_last_file = (ctx_last_fileno_get(ctx) == journalfile->datafile->fileno);
1456 if (is_last_file && journalfile->datafile->pos <= rrdeng_target_data_file_size(ctx) / 3) {
1438 - ctx->create_new_datafile_pair = false;
1457 + ctx->loading.create_new_datafile_pair = false;
1458 return 0;
1459 }
1460
1442 - pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type,
1461 + pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->config.page_type,
1462 journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
1463
1464 if (is_last_file)
1446 - ctx->create_new_datafile_pair = true;
1465 + ctx->loading.create_new_datafile_pair = true;
1466
1467 return 0;
1468
@@ -1458,8 +1477,3 @@ error:
1477 uv_fs_req_cleanup(&req);
1478 return error;
1479 }
1461 -
1462 -void init_commit_log(struct rrdengine_instance *ctx)
1463 -{
1464 - ctx->commit_log.transaction_id = 1;
1465 -}
database/engine/journalfile.h
+2 -9
@@ -121,13 +121,6 @@ struct journal_v2_header {
121
122 #define JOURNAL_V2_HEADER_PADDING_SZ (RRDENG_BLOCK_SIZE - (sizeof(struct journal_v2_header)))
123
124 -
125 -
126 -/* only one event loop is supported for now */
127 -struct transaction_commit_log {
128 - uint64_t transaction_id;
129 -};
130 -
124 struct wal;
125
126 void journalfile_v1_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
@@ -140,7 +133,7 @@ int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct
133 int journalfile_create(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
134 int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
135 struct rrdengine_datafile *datafile);
143 -void init_commit_log(struct rrdengine_instance *ctx);
136 +void journalfile_v2_populate_retention_to_mrg(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile);
137
138 void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_unused, uint8_t type __maybe_unused,
139 Pvoid_t JudyL_metrics, Pvoid_t JudyL_extents_pos,
@@ -152,6 +145,6 @@ size_t journalfile_v2_data_size_get(struct rrdengine_journalfile *journalfile);
145 void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd, void *journal_data, uint32_t journal_data_size);
146 struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfile *journalfile, size_t *data_size, time_t wanted_first_time_s, time_t wanted_last_time_s);
147 void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile);
155 -void journalfile_v2_data_unmount_cleanup(time_t now_s, int storage_tiers);
148 +void journalfile_v2_data_unmount_cleanup(time_t now_s);
149
150 #endif /* NETDATA_JOURNALFILE_H */
\ No newline at end of file
database/engine/metric.c
+51 -32
@@ -21,25 +21,31 @@ struct mrg {
21 ARAL *aral;
22 netdata_rwlock_t rwlock;
23 Pvoid_t uuid_judy; // each UUID has a JudyL of sections (tiers)
24 - } index;
24 + } index[MRG_PARTITIONS];
25
26 struct mrg_statistics stats;
27 +
28 + size_t entries_per_partition[MRG_PARTITIONS];
29 };
30
31 static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg) {
32 __atomic_add_fetch(&mrg->stats.additions_duplicate, 1, __ATOMIC_RELAXED);
33 }
34
33 -static inline void MRG_STATS_ADDED_METRIC(MRG *mrg) {
35 +static inline void MRG_STATS_ADDED_METRIC(MRG *mrg, size_t partition) {
36 __atomic_add_fetch(&mrg->stats.entries, 1, __ATOMIC_RELAXED);
37 __atomic_add_fetch(&mrg->stats.additions, 1, __ATOMIC_RELAXED);
38 __atomic_add_fetch(&mrg->stats.size, sizeof(METRIC), __ATOMIC_RELAXED);
39 +
40 + __atomic_add_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
41 }
42
39 -static inline void MRG_STATS_DELETED_METRIC(MRG *mrg) {
43 +static inline void MRG_STATS_DELETED_METRIC(MRG *mrg, size_t partition) {
44 __atomic_sub_fetch(&mrg->stats.entries, 1, __ATOMIC_RELAXED);
45 __atomic_sub_fetch(&mrg->stats.size, sizeof(METRIC), __ATOMIC_RELAXED);
46 __atomic_add_fetch(&mrg->stats.deletions, 1, __ATOMIC_RELAXED);
47 +
48 + __atomic_sub_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
49 }
50
51 static inline void MRG_STATS_SEARCH_HIT(MRG *mrg) {
@@ -54,17 +60,17 @@ static inline void MRG_STATS_DELETE_MISS(MRG *mrg) {
60 __atomic_add_fetch(&mrg->stats.delete_misses, 1, __ATOMIC_RELAXED);
61 }
62
57 -static inline void mrg_index_read_lock(MRG *mrg) {
58 - netdata_rwlock_rdlock(&mrg->index.rwlock);
63 +static inline void mrg_index_read_lock(MRG *mrg, size_t partition) {
64 + netdata_rwlock_rdlock(&mrg->index[partition].rwlock);
65 }
60 -static inline void mrg_index_read_unlock(MRG *mrg) {
61 - netdata_rwlock_unlock(&mrg->index.rwlock);
66 +static inline void mrg_index_read_unlock(MRG *mrg, size_t partition) {
67 + netdata_rwlock_unlock(&mrg->index[partition].rwlock);
68 }
63 -static inline void mrg_index_write_lock(MRG *mrg) {
64 - netdata_rwlock_wrlock(&mrg->index.rwlock);
69 +static inline void mrg_index_write_lock(MRG *mrg, size_t partition) {
70 + netdata_rwlock_wrlock(&mrg->index[partition].rwlock);
71 }
66 -static inline void mrg_index_write_unlock(MRG *mrg) {
67 - netdata_rwlock_unlock(&mrg->index.rwlock);
72 +static inline void mrg_index_write_unlock(MRG *mrg, size_t partition) {
73 + netdata_rwlock_unlock(&mrg->index[partition].rwlock);
74 }
75
76 static inline void mrg_stats_size_judyl_change(MRG *mrg, size_t mem_before_judyl, size_t mem_after_judyl) {
@@ -82,12 +88,19 @@ static inline void mrg_stats_size_judyhs_removed_uuid(MRG *mrg) {
88 __atomic_sub_fetch(&mrg->stats.size, JUDYHS_INDEX_SIZE_ESTIMATE(sizeof(uuid_t)), __ATOMIC_RELAXED);
89 }
90
91 +static inline size_t uuid_partition(MRG *mrg __maybe_unused, uuid_t *uuid) {
92 + uint8_t *u = (uint8_t *)uuid;
93 + return u[UUID_SZ - 1] % MRG_PARTITIONS;
94 +}
95 +
96 static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
86 - mrg_index_write_lock(mrg);
97 + size_t partition = uuid_partition(mrg, &entry->uuid);
98 +
99 + mrg_index_write_lock(mrg, partition);
100
101 size_t mem_before_judyl, mem_after_judyl;
102
90 - Pvoid_t *sections_judy_pptr = JudyHSIns(&mrg->index.uuid_judy, &entry->uuid, sizeof(uuid_t), PJE0);
103 + Pvoid_t *sections_judy_pptr = JudyHSIns(&mrg->index[partition].uuid_judy, &entry->uuid, sizeof(uuid_t), PJE0);
104 if(unlikely(!sections_judy_pptr || sections_judy_pptr == PJERR))
105 fatal("DBENGINE METRIC: corrupted UUIDs JudyHS array");
106
@@ -104,7 +117,7 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
117
118 if(*PValue != NULL) {
119 METRIC *metric = *PValue;
107 - mrg_index_write_unlock(mrg);
120 + mrg_index_write_unlock(mrg, partition);
121
122 if(ret)
123 *ret = false;
@@ -113,7 +126,7 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
126 return metric;
127 }
128
116 - METRIC *metric = arrayalloc_mallocz(mrg->index.aral);
129 + METRIC *metric = arrayalloc_mallocz(mrg->index[partition].aral);
130 uuid_copy(metric->uuid, entry->uuid);
131 metric->section = entry->section;
132 metric->first_time_s = entry->first_time_s;
@@ -123,49 +136,53 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
136 netdata_spinlock_init(&metric->timestamps_lock);
137 *PValue = metric;
138
126 - mrg_index_write_unlock(mrg);
139 + mrg_index_write_unlock(mrg, partition);
140
141 if(ret)
142 *ret = true;
143
131 - MRG_STATS_ADDED_METRIC(mrg);
144 + MRG_STATS_ADDED_METRIC(mrg, partition);
145
146 return metric;
147 }
148
149 static METRIC *metric_get(MRG *mrg, uuid_t *uuid, Word_t section) {
137 - mrg_index_read_lock(mrg);
150 + size_t partition = uuid_partition(mrg, uuid);
151 +
152 + mrg_index_read_lock(mrg, partition);
153
139 - Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index.uuid_judy, uuid, sizeof(uuid_t));
154 + Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index[partition].uuid_judy, uuid, sizeof(uuid_t));
155 if(unlikely(!sections_judy_pptr)) {
141 - mrg_index_read_unlock(mrg);
156 + mrg_index_read_unlock(mrg, partition);
157 MRG_STATS_SEARCH_MISS(mrg);
158 return NULL;
159 }
160
161 Pvoid_t *PValue = JudyLGet(*sections_judy_pptr, section, PJE0);
162 if(unlikely(!PValue)) {
148 - mrg_index_read_unlock(mrg);
163 + mrg_index_read_unlock(mrg, partition);
164 MRG_STATS_SEARCH_MISS(mrg);
165 return NULL;
166 }
167
168 METRIC *metric = *PValue;
169
155 - mrg_index_read_unlock(mrg);
170 + mrg_index_read_unlock(mrg, partition);
171
172 MRG_STATS_SEARCH_HIT(mrg);
173 return metric;
174 }
175
176 static bool metric_del(MRG *mrg, METRIC *metric) {
177 + size_t partition = uuid_partition(mrg, &metric->uuid);
178 +
179 size_t mem_before_judyl, mem_after_judyl;
180
164 - mrg_index_write_lock(mrg);
181 + mrg_index_write_lock(mrg, partition);
182
166 - Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index.uuid_judy, &metric->uuid, sizeof(uuid_t));
183 + Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index[partition].uuid_judy, &metric->uuid, sizeof(uuid_t));
184 if(unlikely(!sections_judy_pptr || !*sections_judy_pptr)) {
168 - mrg_index_write_unlock(mrg);
185 + mrg_index_write_unlock(mrg, partition);
186 MRG_STATS_DELETE_MISS(mrg);
187 return false;
188 }
@@ -176,24 +193,24 @@ static bool metric_del(MRG *mrg, METRIC *metric) {
193 mrg_stats_size_judyl_change(mrg, mem_before_judyl, mem_after_judyl);
194
195 if(unlikely(!rc)) {
179 - mrg_index_write_unlock(mrg);
196 + mrg_index_write_unlock(mrg, partition);
197 MRG_STATS_DELETE_MISS(mrg);
198 return false;
199 }
200
201 if(!*sections_judy_pptr) {
185 - rc = JudyHSDel(&mrg->index.uuid_judy, &metric->uuid, sizeof(uuid_t), PJE0);
202 + rc = JudyHSDel(&mrg->index[partition].uuid_judy, &metric->uuid, sizeof(uuid_t), PJE0);
203 if(unlikely(!rc))
204 fatal("DBENGINE METRIC: cannot delete UUID from JudyHS");
205 mrg_stats_size_judyhs_removed_uuid(mrg);
206 }
207
208 // arrayalloc is running lockless here
192 - arrayalloc_freez(mrg->index.aral, metric);
209 + arrayalloc_freez(mrg->index[partition].aral, metric);
210
194 - mrg_index_write_unlock(mrg);
211 + mrg_index_write_unlock(mrg, partition);
212
196 - MRG_STATS_DELETED_METRIC(mrg);
213 + MRG_STATS_DELETED_METRIC(mrg, partition);
214
215 return true;
216 }
@@ -203,8 +220,10 @@ static bool metric_del(MRG *mrg, METRIC *metric) {
220
221 MRG *mrg_create(void) {
222 MRG *mrg = callocz(1, sizeof(MRG));
206 - netdata_rwlock_init(&mrg->index.rwlock);
207 - mrg->index.aral = arrayalloc_create(sizeof(METRIC), 65536 / sizeof(METRIC), NULL, NULL, false, true);
223 + for(size_t i = 0; i < MRG_PARTITIONS ; i++) {
224 + netdata_rwlock_init(&mrg->index[i].rwlock);
225 + mrg->index[i].aral = arrayalloc_create(sizeof(METRIC), 32768 / sizeof(METRIC), NULL, NULL, false, true);
226 + }
227 mrg->stats.size = sizeof(MRG);
228 return mrg;
229 }
database/engine/metric.h
+2
@@ -3,6 +3,8 @@
3
4 #include "../rrd.h"
5
6 +#define MRG_PARTITIONS 10
7 +
8 typedef struct metric METRIC;
9 typedef struct mrg MRG;
10
database/engine/pagecache.c
+6 -6
@@ -18,14 +18,14 @@ static void main_cache_flush_dirty_page_init_callback(PGC *cache __maybe_unused,
18 struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
19
20 // mark ctx as having flushing in progress
21 - __atomic_add_fetch(&ctx->worker_config.atomics.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
21 + __atomic_add_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
22 }
23
24 static void main_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_ENTRY *entries_array __maybe_unused, PGC_PAGE **pages_array __maybe_unused, size_t entries __maybe_unused)
25 {
26 struct rrdengine_instance *ctx = (struct rrdengine_instance *) entries_array[0].section;
27
28 - size_t bytes_per_point = PAGE_POINT_CTX_SIZE_BYTES(ctx);
28 + size_t bytes_per_point = CTX_POINT_SIZE_BYTES(ctx);
29
30 struct page_descr_with_data *base = NULL;
31
@@ -39,7 +39,7 @@ static void main_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_
39 descr->start_time_ut = start_time_s * USEC_PER_SEC;
40 descr->end_time_ut = end_time_s * USEC_PER_SEC;
41 descr->update_every_s = entries_array[Index].update_every_s;
42 - descr->type = ctx->page_type;
42 + descr->type = ctx->config.page_type;
43
44 descr->page_length = (end_time_s - (start_time_s - descr->update_every_s)) / descr->update_every_s * bytes_per_point;
45
@@ -58,7 +58,7 @@ static void main_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_
58
59 struct completion completion;
60 completion_init(&completion);
61 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_FLUSH_PAGES, base, &completion, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
61 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_FLUSH_PAGES, base, &completion, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
62 completion_wait_for(&completion);
63 completion_destroy(&completion);
64 }
@@ -760,7 +760,7 @@ void pg_cache_preload(struct rrdeng_query_handle *handle) {
760 if (unlikely(!handle || !handle->metric))
761 return;
762
763 - __atomic_add_fetch(&handle->ctx->inflight_queries, 1, __ATOMIC_RELAXED);
763 + __atomic_add_fetch(&handle->ctx->atomic.inflight_queries, 1, __ATOMIC_RELAXED);
764 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.currently_running_queries, 1, __ATOMIC_RELAXED);
765 handle->pdc = pdc_get();
766 handle->pdc->metric = mrg_metric_dup(main_mrg, handle->metric);
@@ -873,7 +873,7 @@ struct pgc_page *pg_cache_lookup_next(
873 pd->update_every_s = page_update_every_s = pgc_page_fix_update_every(page, last_update_every_s);
874 }
875
876 - size_t entries_by_size = page_entries_by_size(page_length, PAGE_POINT_CTX_SIZE_BYTES(ctx));
876 + size_t entries_by_size = page_entries_by_size(page_length, CTX_POINT_SIZE_BYTES(ctx));
877 size_t entries_by_time = page_entries_by_time(page_start_time_s, page_end_time_s, page_update_every_s);
878 if(unlikely(entries_by_size < entries_by_time)) {
879 time_t fixed_page_end_time_s = (time_t)(page_start_time_s + (entries_by_size - 1) * page_update_every_s);
database/engine/pdc.c
+1 -1
@@ -585,7 +585,7 @@ static void pdc_destroy(PDC *pdc) {
585 PDCJudyLFreeArray(&pdc->page_list_JudyL, PJE0);
586
587 __atomic_sub_fetch(&rrdeng_cache_efficiency_stats.currently_running_queries, 1, __ATOMIC_RELAXED);
588 - __atomic_sub_fetch(&pdc->ctx->inflight_queries, 1, __ATOMIC_RELAXED);
588 + __atomic_sub_fetch(&pdc->ctx->atomic.inflight_queries, 1, __ATOMIC_RELAXED);
589 pdc_release(pdc);
590
591 if(unroutable)
database/engine/rrdengine.c
+112 -58
@@ -496,7 +496,7 @@ WAL *wal_get(struct rrdengine_instance *ctx, unsigned size) {
496 wal_globals.protected.available--;
497 }
498
499 - uint64_t transaction_id = ctx->commit_log.transaction_id++;
499 + uint64_t transaction_id = __atomic_fetch_add(&ctx->atomic.transaction_id, 1, __ATOMIC_RELAXED);
500 netdata_spinlock_unlock(&wal_globals.protected.spinlock);
501
502 if(unlikely(!wal)) {
@@ -566,8 +566,8 @@ static struct {
566 struct {
567 SPINLOCK spinlock;
568 size_t waiting;
569 - struct rrdeng_cmd *waiting_items_by_priority[STORAGE_PRIO_MAX_DONT_USE];
570 - size_t executed_by_priority[STORAGE_PRIO_MAX_DONT_USE];
569 + struct rrdeng_cmd *waiting_items_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
570 + size_t executed_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
571 } queue;
572
573
@@ -605,6 +605,22 @@ static void rrdeng_cmd_cleanup1(void) {
605 }
606 }
607
608 +static inline STORAGE_PRIORITY rrdeng_enq_cmd_map_opcode_to_priority(enum rrdeng_opcode opcode, STORAGE_PRIORITY priority) {
609 + if(unlikely(priority >= STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE))
610 + priority = STORAGE_PRIORITY_BEST_EFFORT;
611 +
612 + switch(opcode) {
613 + case RRDENG_OPCODE_PREP_QUERY:
614 + priority = STORAGE_PRIORITY_INTERNAL_QUERY_PREP;
615 + break;
616 +
617 + default:
618 + break;
619 + }
620 +
621 + return priority;
622 +}
623 +
624 void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd) {
625 epdl_cmd_queued(cmd->data, cmd);
626 }
@@ -617,10 +633,14 @@ void rrdeng_req_cmd(requeue_callback_t get_cmd_cb, void *data, STORAGE_PRIORITY
633 netdata_spinlock_lock(&rrdeng_cmd_globals.queue.spinlock);
634
635 struct rrdeng_cmd *cmd = get_cmd_cb(data);
620 - if(cmd && cmd->priority > priority) {
621 - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[cmd->priority], cmd, cache.prev, cache.next);
622 - DOUBLE_LINKED_LIST_APPEND_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[priority], cmd, cache.prev, cache.next);
623 - cmd->priority = priority;
636 + if(cmd) {
637 + priority = rrdeng_enq_cmd_map_opcode_to_priority(cmd->opcode, priority);
638 +
639 + if (cmd->priority > priority) {
640 + DOUBLE_LINKED_LIST_REMOVE_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[cmd->priority], cmd, cache.prev, cache.next);
641 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[priority], cmd, cache.prev, cache.next);
642 + cmd->priority = priority;
643 + }
644 }
645
646 netdata_spinlock_unlock(&rrdeng_cmd_globals.queue.spinlock);
@@ -630,8 +650,7 @@ void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, v
650 enum storage_priority priority, enqueue_callback_t enqueue_cb, dequeue_callback_t dequeue_cb) {
651 struct rrdeng_cmd *cmd = NULL;
652
633 - if(unlikely(priority >= STORAGE_PRIO_MAX_DONT_USE))
634 - priority = STORAGE_PRIORITY_NORMAL;
653 + priority = rrdeng_enq_cmd_map_opcode_to_priority(opcode, priority);
654
655 netdata_spinlock_lock(&rrdeng_cmd_globals.cache.spinlock);
656 if(likely(rrdeng_cmd_globals.cache.available_items)) {
@@ -675,16 +694,16 @@ static inline bool rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(STORAGE_PR
694 static inline struct rrdeng_cmd rrdeng_deq_cmd(void) {
695 struct rrdeng_cmd *cmd = NULL;
696
678 - STORAGE_PRIORITY max_priority = work_request_full() ? STORAGE_PRIORITY_CRITICAL : STORAGE_PRIORITY_BEST_EFFORT;
697 + STORAGE_PRIORITY max_priority = work_request_full() ? STORAGE_PRIORITY_INTERNAL_DBENGINE : STORAGE_PRIORITY_BEST_EFFORT;
698
699 // find an opcode to execute from the queue
700 netdata_spinlock_lock(&rrdeng_cmd_globals.queue.spinlock);
682 - for(STORAGE_PRIORITY priority = STORAGE_PRIORITY_CRITICAL; priority <= max_priority ; priority++) {
701 + for(STORAGE_PRIORITY priority = STORAGE_PRIORITY_INTERNAL_DBENGINE; priority <= max_priority ; priority++) {
702 cmd = rrdeng_cmd_globals.queue.waiting_items_by_priority[priority];
703 if(cmd) {
704
705 // avoid starvation of lower priorities
687 - if(unlikely(priority > STORAGE_PRIORITY_CRITICAL &&
706 + if(unlikely(priority >= STORAGE_PRIORITY_HIGH &&
707 priority < STORAGE_PRIORITY_BEST_EFFORT &&
708 ++rrdeng_cmd_globals.queue.executed_by_priority[priority] % 50 == 0 &&
709 rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(priority + 1, max_priority))) {
@@ -793,7 +812,7 @@ static void after_extent_flushed_to_open(struct rrdengine_instance *ctx __maybe_
812 completion_mark_complete(completion);
813
814 if(ctx_is_available_for_queries(ctx))
796 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
815 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
816 }
817
818 static void extent_flushed_to_open_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
@@ -837,9 +856,9 @@ static void extent_flushed_to_open_tp_worker(struct rrdengine_instance *ctx __ma
856 datafile->writers.flushed_to_open_running--;
857 netdata_spinlock_unlock(&datafile->writers.spinlock);
858
840 - if(datafile->fileno != __atomic_load_n(&ctx->last_fileno, __ATOMIC_RELAXED) && still_running)
859 + if(datafile->fileno != ctx_last_fileno_get(ctx) && still_running)
860 // we just finished a flushing on a datafile that is not the active one
842 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
861 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
862 }
863
864 // Main event loop callback
@@ -856,7 +875,7 @@ static void extent_flush_io_callback(uv_fs_t *uv_fs_request) {
875
876 datafile->writers.flushed_to_open_running++;
877 rrdeng_enq_cmd(xt_io_descr->ctx, RRDENG_OPCODE_FLUSHED_TO_OPEN, uv_fs_request, xt_io_descr->completion,
859 - STORAGE_PRIORITY_CRITICAL, NULL, NULL);
878 + STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
879
880 netdata_spinlock_unlock(&datafile->writers.spinlock);
881
@@ -876,7 +895,7 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
895 struct extent_buffer *eb = NULL;
896 void *compressed_buf = NULL;
897 Word_t Index;
879 - uint8_t compression_algorithm = ctx->global_compress_alg;
898 + uint8_t compression_algorithm = ctx->config.global_compress_alg;
899 struct rrdengine_datafile *datafile;
900 /* persistent structures */
901 struct rrdeng_df_extent_header *header;
@@ -896,7 +915,7 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
915 if (completion)
916 completion_mark_complete(completion);
917
899 - __atomic_sub_fetch(&ctx->worker_config.atomics.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
918 + __atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
919 return 0;
920 }
921
@@ -974,7 +993,7 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
993 static SPINLOCK sp = NETDATA_SPINLOCK_INITIALIZER;
994 netdata_spinlock_lock(&sp);
995 if(create_new_datafile_pair(ctx) == 0)
977 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL, NULL,
996 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL,
997 NULL);
998 netdata_spinlock_unlock(&sp);
999
@@ -1010,8 +1029,8 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
1029 ++ctx->stats.io_write_extents;
1030 commit_data_extent(ctx, xt_io_descr);
1031 datafile->pos += real_io_size;
1013 - ctx->disk_space += real_io_size;
1014 - ctx->last_flush_fileno = datafile->fileno;
1032 + ctx_current_disk_space_increase(ctx, real_io_size);
1033 + ctx_last_flush_fileno_set(ctx, datafile->fileno);
1034
1035 ret = uv_fs_write(&rrdeng_main.loop, &xt_io_descr->uv_fs_request, datafile->file, &xt_io_descr->iov,
1036 1, xt_io_descr->pos, extent_flush_io_callback);
@@ -1024,7 +1043,7 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
1043 }
1044
1045 static void after_database_rotate(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
1027 - ctx->worker_config.now_deleting_files = false;
1046 + __atomic_store_n(&ctx->atomic.now_deleting_files, false, __ATOMIC_RELAXED);
1047 }
1048
1049 struct uuid_first_time_s {
@@ -1209,7 +1228,7 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1228 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
1229 "' to be available for deletion, "
1230 "it is in use currently by %u users.",
1212 - ctx->dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno, datafile->users.lockers);
1231 + ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno, datafile->users.lockers);
1232
1233 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_spin, 1, __ATOMIC_RELAXED);
1234 sleep_usec(1 * USEC_PER_SEC);
@@ -1220,7 +1239,7 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1239 info("DBENGINE: deleting data file '%s/"
1240 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
1241 "'.",
1223 - ctx->dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
1242 + ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);
1243
1244 if(worker)
1245 worker_is_busy(UV_EVENT_DATAFILE_DELETE);
@@ -1257,11 +1276,11 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1276 freez(journal_file);
1277 freez(datafile);
1278
1260 - ctx->disk_space -= deleted_bytes;
1279 + ctx_current_disk_space_decrease(ctx, deleted_bytes);
1280 info("DBENGINE: reclaimed %u bytes of disk space.", deleted_bytes);
1281
1282 if (rrdeng_ctx_exceeded_disk_quota(ctx))
1264 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1283 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1284
1285 rrdcontext_db_rotation();
1286 }
@@ -1276,7 +1295,43 @@ static void after_flush_all_hot_and_dirty_pages_of_section(struct rrdengine_inst
1295
1296 static void flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1297 pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
1279 - completion_mark_complete(&ctx->quiesce_completion);
1298 + completion_mark_complete(&ctx->quiesce.completion);
1299 +}
1300 +
1301 +static void after_populate_mrg(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
1302 + ;
1303 +}
1304 +
1305 +static void populate_mrg_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1306 + do {
1307 + struct rrdengine_datafile *datafile = NULL;
1308 +
1309 + // find a datafile to work
1310 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1311 + for(datafile = ctx->datafiles.first; datafile ; datafile = datafile->next) {
1312 + if(!netdata_spinlock_trylock(&datafile->populate_mrg.spinlock))
1313 + continue;
1314 +
1315 + if(datafile->populate_mrg.populated) {
1316 + netdata_spinlock_unlock(&datafile->populate_mrg.spinlock);
1317 + continue;
1318 + }
1319 +
1320 + // we have the spinlock and it is not populated
1321 + break;
1322 + }
1323 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1324 +
1325 + if(!datafile)
1326 + break;
1327 +
1328 + journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
1329 + datafile->populate_mrg.populated = true;
1330 + netdata_spinlock_unlock(&datafile->populate_mrg.spinlock);
1331 +
1332 + } while(1);
1333 +
1334 + completion_mark_complete(completion);
1335 }
1336
1337 static void after_ctx_shutdown(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
@@ -1284,11 +1339,11 @@ static void after_ctx_shutdown(struct rrdengine_instance *ctx __maybe_unused, vo
1339 }
1340
1341 static void ctx_shutdown_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1287 - completion_wait_for(&ctx->quiesce_completion);
1288 - completion_destroy(&ctx->quiesce_completion);
1342 + completion_wait_for(&ctx->quiesce.completion);
1343 + completion_destroy(&ctx->quiesce.completion);
1344
1290 - while(__atomic_load_n(&ctx->worker_config.atomics.extents_currently_being_flushed, __ATOMIC_RELAXED) ||
1291 - __atomic_load_n(&ctx->inflight_queries, __ATOMIC_RELAXED))
1345 + while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED) ||
1346 + __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED))
1347 sleep_usec(1 * USEC_PER_MS);
1348
1349 completion_mark_complete(completion);
@@ -1321,7 +1376,7 @@ static void query_prep_tp_worker(struct rrdengine_instance *ctx __maybe_unused,
1376 }
1377
1378 unsigned rrdeng_target_data_file_size(struct rrdengine_instance *ctx) {
1324 - unsigned target_size = ctx->max_disk_space / TARGET_DATAFILES;
1379 + unsigned target_size = ctx->config.max_disk_space / TARGET_DATAFILES;
1380 target_size = MIN(target_size, MAX_DATAFILE_SIZE);
1381 target_size = MAX(target_size, MIN_DATAFILE_SIZE);
1382 return target_size;
@@ -1329,10 +1384,10 @@ unsigned rrdeng_target_data_file_size(struct rrdengine_instance *ctx) {
1384
1385 bool rrdeng_ctx_exceeded_disk_quota(struct rrdengine_instance *ctx)
1386 {
1332 - uint64_t estimated_disk_space = ctx->disk_space + rrdeng_target_data_file_size(ctx) -
1387 + uint64_t estimated_disk_space = ctx_current_disk_space_get(ctx) + rrdeng_target_data_file_size(ctx) -
1388 (ctx->datafiles.first->prev ? ctx->datafiles.first->prev->pos : 0);
1389
1335 - return estimated_disk_space > ctx->max_disk_space;
1390 + return estimated_disk_space > ctx->config.max_disk_space;
1391 }
1392
1393 /* return 0 on success */
@@ -1383,7 +1438,7 @@ static void journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe
1438 unsigned count = 0;
1439 worker_is_busy(UV_EVENT_JOURNAL_INDEX_WAIT);
1440
1386 - while (ctx->worker_config.now_deleting_files && count++ < MAX_RETRIES_TO_START_INDEX)
1441 + while (__atomic_load_n(&ctx->atomic.now_deleting_files, __ATOMIC_RELAXED) && count++ < MAX_RETRIES_TO_START_INDEX)
1442 sleep_usec(100 * USEC_PER_MS);
1443
1444 if (count == MAX_RETRIES_TO_START_INDEX) {
@@ -1394,7 +1449,7 @@ static void journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe
1449 struct rrdengine_datafile *datafile = ctx->datafiles.first;
1450 worker_is_busy(UV_EVENT_JOURNAL_INDEX);
1451 count = 0;
1397 - while (datafile && datafile->fileno != ctx->last_fileno && datafile->fileno != ctx->last_flush_fileno) {
1452 + while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {
1453
1454 netdata_spinlock_lock(&datafile->writers.spinlock);
1455 bool available = (datafile->writers.running || datafile->writers.flushed_to_open_running) ? false : true;
@@ -1405,7 +1460,7 @@ static void journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe
1460
1461 if (unlikely(!journalfile_v2_data_available(datafile->journalfile))) {
1462 info("DBENGINE: journal file %u is ready to be indexed", datafile->fileno);
1408 - pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type,
1463 + pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->config.page_type,
1464 journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
1465 count++;
1466 }
@@ -1435,8 +1490,8 @@ static void after_extent_read(struct rrdengine_instance *ctx __maybe_unused, voi
1490 }
1491
1492 static void after_journal_v2_indexing(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {
1438 - ctx->worker_config.migration_to_v2_running = false;
1439 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1493 + __atomic_store_n(&ctx->atomic.migration_to_v2_running, false, __ATOMIC_RELAXED);
1494 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1495 }
1496
1497 struct rrdeng_buffer_sizes rrdeng_get_buffer_sizes(void) {
@@ -1467,8 +1522,8 @@ void timer_cb(uv_timer_t* handle) {
1522 worker_set_metric(RRDENG_WORKS_DISPATCHED, (NETDATA_DOUBLE)__atomic_load_n(&work_request_globals.atomics.dispatched, __ATOMIC_RELAXED));
1523 worker_set_metric(RRDENG_WORKS_EXECUTING, (NETDATA_DOUBLE)__atomic_load_n(&work_request_globals.atomics.executing, __ATOMIC_RELAXED));
1524
1470 - rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1471 - rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1525 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1526 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1527
1528 rrdeng_cmd_cleanup1();
1529 work_request_cleanup1();
@@ -1487,7 +1542,7 @@ void timer_cb(uv_timer_t* handle) {
1542 time_t now_s = now_monotonic_sec();
1543 if(now_s - last_run_s >= 10) {
1544 last_run_s = now_s;
1490 - journalfile_v2_data_unmount_cleanup(now_s, storage_tiers);
1545 + journalfile_v2_data_unmount_cleanup(now_s);
1546 }
1547 }
1548
@@ -1498,7 +1553,7 @@ void timer_cb(uv_timer_t* handle) {
1553 worker_is_idle();
1554 }
1555
1501 -bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx) {
1556 +bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx __maybe_unused) {
1557 static bool spawned = false;
1558
1559 if(!spawned) {
@@ -1532,10 +1587,6 @@ bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx) {
1587 spawned = true;
1588 }
1589
1535 - ctx->worker_config.now_deleting_files = false;
1536 - ctx->worker_config.migration_to_v2_running = false;
1537 - ctx->worker_config.atomics.extents_currently_being_flushed = 0;
1538 -
1590 return true;
1591 }
1592
@@ -1667,35 +1718,38 @@ void dbengine_event_loop(void* arg) {
1718 case RRDENG_OPCODE_JOURNAL_FILE_INDEX: {
1719 struct rrdengine_instance *ctx = cmd.ctx;
1720 struct rrdengine_datafile *datafile = cmd.data;
1670 - if(!ctx->worker_config.migration_to_v2_running) {
1671 -
1672 - ctx->worker_config.migration_to_v2_running = true;
1673 - if (!work_dispatch(ctx, datafile, NULL, opcode, journal_v2_indexing_tp_worker, after_journal_v2_indexing))
1674 - ctx->worker_config.migration_to_v2_running = false;
1721 + if(!__atomic_load_n(&ctx->atomic.migration_to_v2_running, __ATOMIC_RELAXED)) {
1722
1723 + __atomic_store_n(&ctx->atomic.migration_to_v2_running, true, __ATOMIC_RELAXED);
1724 + work_dispatch(ctx, datafile, NULL, opcode, journal_v2_indexing_tp_worker, after_journal_v2_indexing);
1725 }
1726 break;
1727 }
1728
1729 case RRDENG_OPCODE_DATABASE_ROTATE: {
1730 struct rrdengine_instance *ctx = cmd.ctx;
1682 - if (!ctx->worker_config.now_deleting_files &&
1731 + if (!__atomic_load_n(&ctx->atomic.now_deleting_files, __ATOMIC_RELAXED) &&
1732 ctx->datafiles.first->next != NULL &&
1733 ctx->datafiles.first->next->next != NULL &&
1734 rrdeng_ctx_exceeded_disk_quota(ctx)) {
1735
1687 - ctx->worker_config.now_deleting_files = true;
1688 - if(!work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate))
1689 - ctx->worker_config.now_deleting_files = false;
1690 -
1736 + __atomic_store_n(&ctx->atomic.now_deleting_files, true, __ATOMIC_RELAXED);
1737 + work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate);
1738 }
1739 break;
1740 }
1741
1742 + case RRDENG_OPCODE_CTX_POPULATE_MRG: {
1743 + struct rrdengine_instance *ctx = cmd.ctx;
1744 + struct completion *completion = cmd.completion;
1745 + work_dispatch(ctx, NULL, completion, opcode, populate_mrg_tp_worker, after_populate_mrg);
1746 + break;
1747 + }
1748 +
1749 case RRDENG_OPCODE_CTX_QUIESCE: {
1750 // a ctx will shutdown shortly
1751 struct rrdengine_instance *ctx = cmd.ctx;
1698 - __atomic_store_n(&ctx->quiesce, SET_QUIESCE, __ATOMIC_RELEASE);
1752 + __atomic_store_n(&ctx->quiesce.enabled, true, __ATOMIC_RELEASE);
1753 work_dispatch(ctx, NULL, NULL, opcode,
1754 flush_all_hot_and_dirty_pages_of_section_tp_worker,
1755 after_flush_all_hot_and_dirty_pages_of_section);
database/engine/rrdengine.h
+64 -37
@@ -233,6 +233,7 @@ enum rrdeng_opcode {
233 RRDENG_OPCODE_DATABASE_ROTATE,
234 RRDENG_OPCODE_CTX_SHUTDOWN,
235 RRDENG_OPCODE_CTX_QUIESCE,
236 + RRDENG_OPCODE_CTX_POPULATE_MRG,
237
238 RRDENG_OPCODE_MAX
239 };
@@ -303,16 +304,6 @@ typedef struct wal {
304 WAL *wal_get(struct rrdengine_instance *ctx, unsigned size);
305 void wal_release(WAL *wal);
306
306 -struct rrdengine_worker_config {
307 - bool now_deleting_files;
308 - bool migration_to_v2_running;
309 -
310 - struct {
311 - // non-zero until we commit data to disk (both datafile and journal file)
312 - unsigned extents_currently_being_flushed;
313 - } atomics;
314 -};
315 -
307 /*
308 * Debug statistics not used by code logic.
309 * They only describe operations since DB engine instance load time.
@@ -359,37 +350,73 @@ extern rrdeng_stats_t rrdeng_reserved_file_descriptors;
350 extern rrdeng_stats_t global_pg_cache_over_half_dirty_events;
351 extern rrdeng_stats_t global_flushing_pressure_page_deletions; /* number of deleted pages */
352
362 -#define NO_QUIESCE (0) /* initial state when all operations function normally */
363 -#define SET_QUIESCE (1) /* set it before shutting down the instance, quiesce long running operations */
364 -#define QUIESCED (2) /* is set after all threads have finished running */
365 -
353 struct rrdengine_instance {
367 - struct rrdengine_worker_config worker_config;
368 - struct completion rrdengine_completion;
369 - bool journal_initialization;
370 - uint8_t global_compress_alg;
371 - struct transaction_commit_log commit_log;
372 - struct rrdengine_datafile_list datafiles;
373 - RRDHOST *host; /* the legacy host, or NULL for multi-host DB */
374 - char dbfiles_path[FILENAME_MAX + 1];
375 - char machine_guid[GUID_LEN + 1]; /* the unique ID of the corresponding host, or localhost for multihost DB */
376 - uint64_t disk_space;
377 - uint64_t max_disk_space;
378 - int tier;
379 - unsigned last_fileno; /* newest index of datafile and journalfile */
380 - unsigned last_flush_fileno;
381 -
382 - bool create_new_datafile_pair;
383 - uint8_t quiesce; /* set to SET_QUIESCE before shutdown of the engine */
384 - uint8_t page_type; /* Default page type for this context */
385 -
386 - struct completion quiesce_completion;
387 -
388 - size_t inflight_queries;
354 + struct {
355 + int tier; // the tier of this ctx
356 + uint8_t page_type; // default page type for this context
357 +
358 + uint64_t max_disk_space; // the max disk space this ctx is allowed to use
359 + uint8_t global_compress_alg; // the wanted compression algorithm
360 +
361 + char dbfiles_path[FILENAME_MAX + 1];
362 + } config;
363 +
364 + struct {
365 + uv_rwlock_t rwlock; // the linked list of datafiles is protected by this lock
366 + struct rrdengine_datafile *first; // oldest - the newest with ->first->prev
367 + } datafiles;
368 +
369 + struct {
370 + unsigned last_fileno; // newest index of datafile and journalfile
371 + unsigned last_flush_fileno; // newest index of datafile received data
372 +
373 + size_t inflight_queries; // the number of queries currently running
374 + uint64_t current_disk_space; // the current disk space size used
375 +
376 + uint64_t transaction_id; // the transaction id of the next extent flushing
377 +
378 + bool migration_to_v2_running;
379 + bool now_deleting_files;
380 + unsigned extents_currently_being_flushed; // non-zero until we commit data to disk (both datafile and journal file)
381 + } atomic;
382 +
383 + struct {
384 + bool exit_mode;
385 + bool enabled; // when set (before shutdown), queries are prohibited
386 + struct completion completion;
387 + } quiesce;
388 +
389 + struct {
390 + struct {
391 + size_t size;
392 + struct completion *array;
393 + } populate_mrg;
394 +
395 + bool create_new_datafile_pair;
396 + } loading;
397 +
398 struct rrdengine_statistics stats;
399 };
400
392 -#define ctx_is_available_for_queries(ctx) (__atomic_load_n(&(ctx)->quiesce, __ATOMIC_RELAXED) == NO_QUIESCE)
401 +#define ctx_current_disk_space_get(ctx) __atomic_load_n(&(ctx)->atomic.current_disk_space, __ATOMIC_RELAXED)
402 +#define ctx_current_disk_space_increase(ctx, size) __atomic_add_fetch(&(ctx)->atomic.current_disk_space, size, __ATOMIC_RELAXED)
403 +#define ctx_current_disk_space_decrease(ctx, size) __atomic_sub_fetch(&(ctx)->atomic.current_disk_space, size, __ATOMIC_RELAXED)
404 +
405 +#define ctx_last_fileno_get(ctx) __atomic_load_n(&(ctx)->atomic.last_fileno, __ATOMIC_RELAXED)
406 +#define ctx_last_fileno_increment(ctx) __atomic_add_fetch(&(ctx)->atomic.last_fileno, 1, __ATOMIC_RELAXED)
407 +
408 +#define ctx_last_flush_fileno_get(ctx) __atomic_load_n(&(ctx)->atomic.last_flush_fileno, __ATOMIC_RELAXED)
409 +static inline void ctx_last_flush_fileno_set(struct rrdengine_instance *ctx, unsigned fileno) {
410 + unsigned old_fileno = ctx_last_flush_fileno_get(ctx);
411 +
412 + do {
413 + if(old_fileno >= fileno)
414 + return;
415 +
416 + } while(!__atomic_compare_exchange_n(&ctx->atomic.last_flush_fileno, &old_fileno, fileno, false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
417 +}
418 +
419 +#define ctx_is_available_for_queries(ctx) (__atomic_load_n(&(ctx)->quiesce.enabled, __ATOMIC_RELAXED) == false && __atomic_load_n(&(ctx)->quiesce.exit_mode, __ATOMIC_RELAXED) == false)
420
421 void *dbengine_page_alloc(size_t size);
422 void dbengine_page_free(void *page, size_t size);
database/engine/rrdengineapi.c
+105 -44
@@ -227,7 +227,7 @@ static bool page_has_only_empty_metrics(struct rrdeng_collect_handle *handle) {
227 default: {
228 static bool logged = false;
229 if(!logged) {
230 - error("DBENGINE: cannot check page for nulls on unknown page type id %d", (mrg_metric_ctx(handle->metric))->page_type);
230 + error("DBENGINE: cannot check page for nulls on unknown page type id %d", (mrg_metric_ctx(handle->metric))->config.page_type);
231 logged = true;
232 }
233 return false;
@@ -283,19 +283,19 @@ time_t point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC);
283 "DBENGINE CACHE: requested to add a hot page to the main cache, "
284 "but the page returned is not hot");
285
286 - if(unlikely(pgc_page_data_size(main_cache, page) < PAGE_POINT_CTX_SIZE_BYTES(ctx)))
286 + if(unlikely(pgc_page_data_size(main_cache, page) < CTX_POINT_SIZE_BYTES(ctx)))
287 fatal("DBENGINE: hot page returned from main cache does not have the size for storing 1 point");
288
289 // copy the point in data
290 - memcpy(pgc_page_data(page), data, PAGE_POINT_CTX_SIZE_BYTES(ctx));
290 + memcpy(pgc_page_data(page), data, CTX_POINT_SIZE_BYTES(ctx));
291
292 // free data
293 dbengine_page_free(page_entry.data, data_size);
294
295 - handle->page_entries_max = pgc_page_data_size(main_cache, page) / PAGE_POINT_CTX_SIZE_BYTES(ctx);
295 + handle->page_entries_max = pgc_page_data_size(main_cache, page) / CTX_POINT_SIZE_BYTES(ctx);
296 }
297 else
298 - handle->page_entries_max = data_size / PAGE_POINT_CTX_SIZE_BYTES(ctx);
298 + handle->page_entries_max = data_size / CTX_POINT_SIZE_BYTES(ctx);
299
300 handle->page_end_time_ut = point_in_time_ut;
301 handle->page_position = 1; // zero is already in our data
@@ -308,13 +308,13 @@ static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle,
308
309 if(handle->options & RRDENG_FIRST_PAGE_ALLOCATED) {
310 // any page except the first
311 - size = tier_page_size[ctx->tier];
311 + size = tier_page_size[ctx->config.tier];
312 }
313 else {
314 // the first page
315 handle->options |= RRDENG_FIRST_PAGE_ALLOCATED;
316 - size_t max_size = tier_page_size[ctx->tier];
317 - size_t max_slots = max_size / PAGE_POINT_CTX_SIZE_BYTES(ctx);
316 + size_t max_size = tier_page_size[ctx->config.tier];
317 + size_t max_slots = max_size / CTX_POINT_SIZE_BYTES(ctx);
318 size_t min_slots = max_slots / 5;
319 size_t distribution = max_slots - min_slots;
320 size_t this_page_end_slot = indexing_partition((Word_t)handle->alignment, distribution);
@@ -334,7 +334,7 @@ static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle,
334 if(final_slots < min_slots)
335 final_slots = min_slots;
336
337 - size = final_slots * PAGE_POINT_CTX_SIZE_BYTES(ctx);
337 + size = final_slots * CTX_POINT_SIZE_BYTES(ctx);
338 }
339
340 *data_size = size;
@@ -385,7 +385,7 @@ static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection
385 else
386 data = rrdeng_alloc_new_metric_data(handle, &data_size);
387
388 - switch (ctx->page_type) {
388 + switch (ctx->config.page_type) {
389 case PAGE_METRICS: {
390 storage_number *tier0_metric_data = data;
391 tier0_metric_data[handle->page_position] = pack_storage_number(n, flags);
@@ -407,7 +407,7 @@ static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection
407 default: {
408 static bool logged = false;
409 if(!logged) {
410 - error("DBENGINE: cannot store metric on unknown page type id %d", ctx->page_type);
410 + error("DBENGINE: cannot store metric on unknown page type id %d", ctx->config.page_type);
411 logged = true;
412 }
413 }
@@ -580,9 +580,10 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct sto
580 handle = rrdeng_query_handle_get();
581 register_query_handle(handle);
582
583 - if(unlikely(priority == STORAGE_PRIORITY_CRITICAL))
584 - // critical is reserved for dbengine internal use
583 + if(unlikely(priority < STORAGE_PRIORITY_HIGH))
584 priority = STORAGE_PRIORITY_HIGH;
585 + else if(unlikely(priority > STORAGE_PRIORITY_BEST_EFFORT))
586 + priority = STORAGE_PRIORITY_BEST_EFFORT;
587
588 handle->ctx = ctx;
589 handle->metric = metric;
@@ -678,7 +679,7 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim
679 sp.start_time_s = handle->now_s - handle->dt_s;
680 sp.end_time_s = handle->now_s;
681
681 - switch(handle->ctx->page_type) {
682 + switch(handle->ctx->config.page_type) {
683 case PAGE_METRICS: {
684 storage_number n = handle->metric_data[handle->position];
685 sp.min = sp.max = sp.sum = unpack_storage_number(n);
@@ -703,7 +704,7 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim
704 default: {
705 static bool logged = false;
706 if(!logged) {
706 - error("DBENGINE: unknown page type %d found. Cannot decode it. Ignoring its metrics.", handle->ctx->page_type);
707 + error("DBENGINE: unknown page type %d found. Cannot decode it. Ignoring its metrics.", handle->ctx->config.page_type);
708 logged = true;
709 }
710 storage_point_empty(sp, sp.start_time_s, sp.end_time_s);
@@ -850,13 +851,79 @@ void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long
851 fatal_assert(RRDENG_NR_STATS == 38);
852 }
853
854 +static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {
855 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);
856 + size_t datafiles = 0;
857 + for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next)
858 + datafiles++;
859 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
860 +
861 + size_t cpus = get_system_cpus() / 2;
862 + if(cpus > datafiles)
863 + cpus = datafiles;
864 +
865 + if(cpus < 2)
866 + cpus = 2;
867 +
868 + if(cpus > (size_t)libuv_worker_threads)
869 + cpus = (size_t)libuv_worker_threads;
870 +
871 + if(cpus > MRG_PARTITIONS)
872 + cpus = MRG_PARTITIONS;
873 +
874 + info("DBENGINE: populating retention to MRG from %zu journal files of tier %d, using %zu threads...", datafiles, ctx->config.tier, cpus);
875 +
876 + if(datafiles > 2) {
877 + struct rrdengine_datafile *datafile;
878 +
879 + datafile = ctx->datafiles.first->prev;
880 + if(!(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE))
881 + datafile = datafile->prev;
882 +
883 + if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) {
884 + journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
885 + datafile->populate_mrg.populated = true;
886 + }
887 +
888 + datafile = ctx->datafiles.first;
889 + if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) {
890 + journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile);
891 + datafile->populate_mrg.populated = true;
892 + }
893 + }
894 +
895 + ctx->loading.populate_mrg.size = cpus;
896 + ctx->loading.populate_mrg.array = callocz(ctx->loading.populate_mrg.size, sizeof(struct completion));
897 +
898 + for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) {
899 + completion_init(&ctx->loading.populate_mrg.array[i]);
900 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_POPULATE_MRG, NULL, &ctx->loading.populate_mrg.array[i],
901 + STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
902 + }
903 +}
904 +
905 +void rrdeng_readiness_wait(struct rrdengine_instance *ctx) {
906 + for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) {
907 + completion_wait_for(&ctx->loading.populate_mrg.array[i]);
908 + completion_destroy(&ctx->loading.populate_mrg.array[i]);
909 + }
910 +
911 + freez(ctx->loading.populate_mrg.array);
912 + ctx->loading.populate_mrg.array = NULL;
913 + ctx->loading.populate_mrg.size = 0;
914 +
915 + info("DBENGINE: tier %d is ready for data collection and queries", ctx->config.tier);
916 +}
917 +
918 +void rrdeng_exit_mode(struct rrdengine_instance *ctx) {
919 + __atomic_store_n(&ctx->quiesce.exit_mode, true, __ATOMIC_RELAXED);
920 +}
921 /*
922 * Returns 0 on success, negative on error
923 */
856 -int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
924 +int rrdeng_init(struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
925 unsigned disk_space_mb, size_t tier) {
926 struct rrdengine_instance *ctx;
859 - int error;
927 uint32_t max_open_files;
928
929 max_open_files = rlimit_nofile.rlim_cur / 4;
@@ -881,33 +948,27 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
948 else {
949 *ctxp = ctx = callocz(1, sizeof(*ctx));
950 }
884 - ctx->tier = tier;
885 - ctx->page_type = tier_page_type[tier];
886 - ctx->global_compress_alg = RRD_LZ4;
951 + ctx->config.tier = (int)tier;
952 + ctx->config.page_type = tier_page_type[tier];
953 + ctx->config.global_compress_alg = RRD_LZ4;
954 if (page_cache_mb < RRDENG_MIN_PAGE_CACHE_SIZE_MB)
955 page_cache_mb = RRDENG_MIN_PAGE_CACHE_SIZE_MB;
956 if (disk_space_mb < RRDENG_MIN_DISK_SPACE_MB)
957 disk_space_mb = RRDENG_MIN_DISK_SPACE_MB;
891 - ctx->max_disk_space = disk_space_mb * 1048576LLU;
892 - strncpyz(ctx->dbfiles_path, dbfiles_path, sizeof(ctx->dbfiles_path) - 1);
893 - ctx->dbfiles_path[sizeof(ctx->dbfiles_path) - 1] = '\0';
894 - if (NULL == host)
895 - strncpyz(ctx->machine_guid, registry_get_this_machine_guid(), GUID_LEN);
896 - else
897 - strncpyz(ctx->machine_guid, host->machine_guid, GUID_LEN);
958 + ctx->config.max_disk_space = disk_space_mb * 1048576LLU;
959 + strncpyz(ctx->config.dbfiles_path, dbfiles_path, sizeof(ctx->config.dbfiles_path) - 1);
960 + ctx->config.dbfiles_path[sizeof(ctx->config.dbfiles_path) - 1] = '\0';
961
899 - ctx->quiesce = NO_QUIESCE;
900 - ctx->host = host;
962 + ctx->atomic.transaction_id = 1;
963 + ctx->quiesce.enabled = false;
964
902 - memset(&ctx->worker_config, 0, sizeof(ctx->worker_config));
965 init_page_cache();
904 - init_commit_log(ctx);
905 - error = init_rrd_files(ctx);
906 - if (!error) {
907 -
908 - if(rrdeng_dbengine_spawn(ctx))
966 + if (!init_rrd_files(ctx)) {
967 + if(rrdeng_dbengine_spawn(ctx)) {
968 // success - we run this ctx too
969 + rrdeng_populate_mrg(ctx);
970 return 0;
971 + }
972
973 finalize_rrd_files(ctx);
974 }
@@ -957,8 +1018,8 @@ void rrdeng_prepare_exit(struct rrdengine_instance *ctx) {
1018 // FIXME - ktsaou - properly cleanup ctx
1019 // 1. make sure all collectors are stopped
1020
960 - completion_init(&ctx->quiesce_completion);
961 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1021 + completion_init(&ctx->quiesce.completion);
1022 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1023 }
1024
1025 static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats)
@@ -992,7 +1053,7 @@ static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_S
1053
1054 time_t update_every_s;
1055
995 - size_t points = descr->page_length / PAGE_POINT_CTX_SIZE_BYTES(datafile->ctx);
1056 + size_t points = descr->page_length / CTX_POINT_SIZE_BYTES(datafile->ctx);
1057
1058 time_t start_time_s = journal_start_time_s + descr->delta_start_s;
1059 time_t end_time_s = journal_start_time_s + descr->delta_end_s;
@@ -1000,7 +1061,7 @@ static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_S
1061 if(likely(points > 1))
1062 update_every_s = (time_t) ((end_time_s - start_time_s) / (points - 1));
1063 else {
1003 - update_every_s = (time_t) (default_rrd_update_every * get_tier_grouping(datafile->ctx->tier));
1064 + update_every_s = (time_t) (default_rrd_update_every * get_tier_grouping(datafile->ctx->config.tier));
1065 stats->single_point_pages++;
1066 }
1067
@@ -1045,8 +1106,8 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {
1106 "DBENGINE: metrics pages is %zu, but extents pages is %zu and API consumers is %zu",
1107 stats.metrics_pages, stats.extents_pages, stats.currently_collected_metrics);
1108
1048 - stats.disk_space = ctx->disk_space;
1049 - stats.max_disk_space = ctx->max_disk_space;
1109 + stats.disk_space = ctx_current_disk_space_get(ctx);
1110 + stats.max_disk_space = ctx->config.max_disk_space;
1111
1112 stats.database_retention_secs = (time_t)(stats.last_time_s - stats.first_time_s);
1113
@@ -1075,14 +1136,14 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {
1136 // stats.sizeof_metric = 0;
1137 stats.sizeof_datafile = struct_natural_alignment(sizeof(struct rrdengine_datafile)) + struct_natural_alignment(sizeof(struct rrdengine_journalfile));
1138 stats.sizeof_page_in_cache = 0; // struct_natural_alignment(sizeof(struct page_cache_descr));
1078 - stats.sizeof_point_data = page_type_size[ctx->page_type];
1079 - stats.sizeof_page_data = tier_page_size[ctx->tier];
1139 + stats.sizeof_point_data = page_type_size[ctx->config.page_type];
1140 + stats.sizeof_page_data = tier_page_size[ctx->config.tier];
1141 stats.pages_per_extent = rrdeng_pages_per_extent;
1142
1143 // stats.sizeof_metric_in_index = 40;
1144 // stats.sizeof_page_in_index = 24;
1145
1085 - stats.default_granularity_secs = (size_t)default_rrd_update_every * get_tier_grouping(ctx->tier);
1146 + stats.default_granularity_secs = (size_t)default_rrd_update_every * get_tier_grouping(ctx->config.tier);
1147
1148 return stats;
1149 }
database/engine/rrdengineapi.h
+5 -3
@@ -24,8 +24,7 @@ extern struct rrdengine_instance *multidb_ctx[RRD_STORAGE_TIERS];
24 extern size_t page_type_size[];
25 extern size_t tier_page_size[];
26
27 -#define PAGE_POINT_SIZE_BYTES(x) page_type_size[(x)->type]
28 -#define PAGE_POINT_CTX_SIZE_BYTES(x) page_type_size[(x)->page_type]
27 +#define CTX_POINT_SIZE_BYTES(ctx) page_type_size[(ctx)->config.page_type]
28
29 void rrdeng_generate_legacy_uuid(const char *dim_id, const char *chart_id, uuid_t *ret_uuid);
30 void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uuid_t *legacy_uuid,
@@ -62,9 +61,12 @@ time_t rrdeng_load_align_to_optimal_before(struct storage_engine_query_handle *r
61 void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array);
62
63 /* must call once before using anything */
65 -int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
64 +int rrdeng_init(struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
65 unsigned disk_space_mb, size_t tier);
66
67 +void rrdeng_readiness_wait(struct rrdengine_instance *ctx);
68 +void rrdeng_exit_mode(struct rrdengine_instance *ctx);
69 +
70 int rrdeng_exit(struct rrdengine_instance *ctx);
71 void rrdeng_prepare_exit(struct rrdengine_instance *ctx);
72 bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *db_instance, uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s);
database/rrd.h
+5 -2
@@ -45,13 +45,16 @@ typedef enum __attribute__ ((__packed__)) {
45 } QUERY_SOURCE;
46
47 typedef enum __attribute__ ((__packed__)) storage_priority {
48 - STORAGE_PRIORITY_CRITICAL = 0,
48 + STORAGE_PRIORITY_INTERNAL_DBENGINE = 0,
49 + STORAGE_PRIORITY_INTERNAL_QUERY_PREP,
50 +
51 + // query priorities
52 STORAGE_PRIORITY_HIGH,
53 STORAGE_PRIORITY_NORMAL,
54 STORAGE_PRIORITY_LOW,
55 STORAGE_PRIORITY_BEST_EFFORT,
56
54 - STORAGE_PRIO_MAX_DONT_USE,
57 + STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE,
58 } STORAGE_PRIORITY;
59
60 // forward declarations
database/rrdhost.c
+8 -3
@@ -390,7 +390,6 @@ int is_legacy = 1;
390 host->db[0].tier_grouping = get_tier_grouping(0);
391
392 ret = rrdeng_init(
393 - host,
393 (struct rrdengine_instance **)&host->db[0].instance,
394 dbenginepath,
395 default_rrdeng_page_cache_mb,
@@ -398,8 +397,11 @@ int is_legacy = 1;
397 0); // may fail here for legacy dbengine initialization
398
399 if(ret == 0) {
400 + rrdeng_readiness_wait((struct rrdengine_instance *)host->db[0].instance);
401 +
402 // assign the rest of the shared storage instances to it
403 // to allow them collect its metrics too
404 +
405 for(size_t tier = 1; tier < storage_tiers ; tier++) {
406 host->db[tier].mode = RRD_MEMORY_MODE_DBENGINE;
407 host->db[tier].eng = storage_engine_get(host->db[tier].mode);
@@ -867,7 +869,7 @@ void dbengine_init(char *hostname) {
869 }
870
871 internal_error(true, "DBENGINE tier %zu grouping iterations is set to %zu", tier, storage_tiers_grouping_iterations[tier]);
870 - ret = rrdeng_init(NULL, NULL, dbenginepath, page_cache_mb, disk_space_mb, tier);
872 + ret = rrdeng_init(NULL, dbenginepath, page_cache_mb, disk_space_mb, tier);
873 if(ret != 0) {
874 error("DBENGINE on '%s': Failed to initialize multi-host database tier %zu on path '%s'",
875 hostname, tier, dbenginepath);
@@ -875,7 +877,7 @@ void dbengine_init(char *hostname) {
877 }
878 else {
879 if (rrdeng_ctx_exceeded_disk_quota(multidb_ctx[created_tiers]))
878 - rrdeng_enq_cmd( multidb_ctx[created_tiers], RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
880 + rrdeng_enq_cmd(multidb_ctx[created_tiers], RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
881 created_tiers++;
882 }
883 }
@@ -888,6 +890,9 @@ void dbengine_init(char *hostname) {
890 else if(!created_tiers)
891 fatal("DBENGINE on '%s', failed to initialize databases at '%s'.", hostname, netdata_configured_cache_dir);
892
893 + for(size_t tier = 0; tier < storage_tiers ;tier++)
894 + rrdeng_readiness_wait(multidb_ctx[tier]);
895 +
896 dbengine_enabled = true;
897 #else
898 storage_tiers = config_get_number(CONFIG_SECTION_DB, "storage tiers", 1);