@cryptotaxi247 / netdata-1 / commits / c83abcfb9

DBENGINE v2 - improvements part 1 (#14251)

* allow running multiple evictors and flushers * flipped aggressive and critical evictions * dont run more than 1 evictor * switch to batch evictions when the size of the cache is critical * remove batching of evictions * dedup extent load pending requests * accounting for merged extents * always use double linked list * add extent merging to the overall cache hit ratio * support requeuing merged extents to higher priorities * fix function name * query planner now prefers higher tiers even when they miss some data at the end, which it fills from lower tiers; adding the option "plan" to jsonwrap now renders the query plan * update statistics after every dimension completes * use the retention of all tiers to calculate coverage per tier * use the original window of the query for the planner * give 2.5% befenit for each higher tier * update cmd->priority so that it be requeued multiple times * merged extent pages is a cache hit * fixed dbegnine cache hit stats

Costa Tsaousis committed Jan 12, 2023 at 23:10 UTC c83abcfb9d194a0168e54d2684e3090912c78095
17 files changed +461 -272
daemon/event_loop.c
-1
@@ -18,7 +18,6 @@ void register_libuv_worker_jobs() {
18 worker_register_job_name(UV_EVENT_COMMIT_PAGE_CB, "commit cb");
19 worker_register_job_name(UV_EVENT_FLUSH_PAGES_CB, "flush cb");
20 worker_register_job_name(UV_EVENT_PAGE_LOOKUP, "page lookup");
21 - worker_register_job_name(UV_EVENT_METRIC_LOOKUP, "metric lookup");
21 worker_register_job_name(UV_EVENT_PAGE_POPULATION, "populate page");
22 worker_register_job_name(UV_EVENT_EXT_DECOMPRESSION, "extent decompression");
23 worker_register_job_name(UV_EVENT_READ_MMAP_EXTENT, "read extent (mmap)");
daemon/event_loop.h
-1
@@ -11,7 +11,6 @@ enum event_loop_job {
11 UV_EVENT_FLUSH_PAGES_CB,
12 UV_EVENT_EXT_DECOMPRESSION,
13 UV_EVENT_PAGE_LOOKUP,
14 - UV_EVENT_METRIC_LOOKUP,
14 UV_EVENT_PAGE_POPULATION,
15 UV_EVENT_READ_MMAP_EXTENT,
16 UV_EVENT_EXTENT_PROCESSING,
daemon/global_statistics.c
+43 -53
@@ -1373,8 +1373,8 @@ static void dbengine2_cache_statistics_charts(struct dbengine2_cache_pointers *p
1373 localhost->rrd_update_every,
1374 RRDSET_TYPE_AREA);
1375
1376 - ptrs->rd_pgc_memory_evictions_aggressive = rrddim_add(ptrs->st_pgc_memory_events, "evictions critical", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1377 - ptrs->rd_pgc_memory_evictions_critical = rrddim_add(ptrs->st_pgc_memory_events, "evictions aggressive", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1376 + ptrs->rd_pgc_memory_evictions_aggressive = rrddim_add(ptrs->st_pgc_memory_events, "evictions aggressive", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1377 + ptrs->rd_pgc_memory_evictions_critical = rrddim_add(ptrs->st_pgc_memory_events, "evictions critical", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1378 ptrs->rd_pgc_memory_flushes_critical = rrddim_add(ptrs->st_pgc_memory_events, "flushes critical", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1379
1380 buffer_free(id);
@@ -1671,11 +1671,9 @@ static void dbengine2_statistics_charts(void) {
1671 {
1672 static RRDSET *st_cache_hit_ratio = NULL;
1673 static RRDDIM *rd_hit_ratio = NULL;
1674 - static RRDDIM *rd_preloaded_ratio = NULL;
1675 - static RRDDIM *rd_extent_ratio = NULL;
1676 - static RRDDIM *rd_parallel_load_ratio = NULL;
1677 - static RRDDIM *rd_before_allocation_ratio = NULL;
1678 - static RRDDIM *rd_insert_conflict_ratio = NULL;
1674 + static RRDDIM *rd_main_cache_hit_ratio = NULL;
1675 + static RRDDIM *rd_extent_cache_hit_ratio = NULL;
1676 + static RRDDIM *rd_extent_merge_hit_ratio = NULL;
1677
1678 if (unlikely(!st_cache_hit_ratio)) {
1679 st_cache_hit_ratio = rrdset_create_localhost(
@@ -1693,60 +1691,55 @@ static void dbengine2_statistics_charts(void) {
1691 RRDSET_TYPE_LINE);
1692
1693 rd_hit_ratio = rrddim_add(st_cache_hit_ratio, "overall", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1696 - rd_preloaded_ratio = rrddim_add(st_cache_hit_ratio, "main cache", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1697 - rd_extent_ratio = rrddim_add(st_cache_hit_ratio, "extent cache", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1698 - rd_parallel_load_ratio = rrddim_add(st_cache_hit_ratio, "parallel load", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1699 - rd_before_allocation_ratio = rrddim_add(st_cache_hit_ratio, "before allocation", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1700 - rd_insert_conflict_ratio = rrddim_add(st_cache_hit_ratio, "insert conflict", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1694 + rd_main_cache_hit_ratio = rrddim_add(st_cache_hit_ratio, "main cache", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1695 + rd_extent_cache_hit_ratio = rrddim_add(st_cache_hit_ratio, "extent cache", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1696 + rd_extent_merge_hit_ratio = rrddim_add(st_cache_hit_ratio, "extent merge", NULL, 1, 10000, RRD_ALGORITHM_ABSOLUTE);
1697 }
1698 priority++;
1699
1704 - size_t pages_to_load = cache_efficiency_stats.pages_to_load_from_disk - cache_efficiency_stats_old.pages_to_load_from_disk;
1705 - size_t pages_hit_parallel_load = cache_efficiency_stats.pages_load_ok_preloaded - cache_efficiency_stats_old.pages_load_ok_preloaded;
1706 - size_t pages_hit_before_allocation = cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_before_allocation - cache_efficiency_stats_old.pages_load_ok_loaded_but_cache_hit_before_allocation;
1707 - size_t pages_hit_insert_conflict = cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_while_inserting - cache_efficiency_stats_old.pages_load_ok_loaded_but_cache_hit_while_inserting;
1708 - size_t pages_hit_cached_extent = cache_efficiency_stats.pages_data_source_extent_cache - cache_efficiency_stats_old.pages_data_source_extent_cache;
1709 - size_t pages_hit_at_query_plan = cache_efficiency_stats.pages_meta_source_main_cache - cache_efficiency_stats_old.pages_meta_source_main_cache;
1700 + size_t delta_pages_total = cache_efficiency_stats.pages_total - cache_efficiency_stats_old.pages_total;
1701 + size_t delta_pages_to_load_from_disk = cache_efficiency_stats.pages_to_load_from_disk - cache_efficiency_stats_old.pages_to_load_from_disk;
1702 + size_t delta_extents_loaded_from_disk = cache_efficiency_stats.extents_loaded_from_disk - cache_efficiency_stats_old.extents_loaded_from_disk;
1703
1711 - size_t pages_total_hit = pages_hit_at_query_plan + pages_hit_before_allocation + pages_hit_insert_conflict + pages_hit_cached_extent;
1712 - size_t pages_total = cache_efficiency_stats.pages_total - cache_efficiency_stats_old.pages_total;
1704 + size_t delta_pages_data_source_main_cache = cache_efficiency_stats.pages_data_source_main_cache - cache_efficiency_stats_old.pages_data_source_main_cache;
1705 + size_t delta_pages_pending_found_in_cache_at_pass4 = cache_efficiency_stats.pages_data_source_main_cache_at_pass4 - cache_efficiency_stats_old.pages_data_source_main_cache_at_pass4;
1706 +
1707 + size_t delta_pages_data_source_extent_cache = cache_efficiency_stats.pages_data_source_extent_cache - cache_efficiency_stats_old.pages_data_source_extent_cache;
1708 + size_t delta_pages_load_extent_merged = cache_efficiency_stats.pages_load_extent_merged - cache_efficiency_stats_old.pages_load_extent_merged;
1709 +
1710 + size_t pages_total_hit = delta_pages_total - delta_extents_loaded_from_disk;
1711
1712 static size_t overall_hit_ratio = 100;
1715 - size_t preloaded_hit_ratio = 0, extent_hit_ratio = 0;
1716 - size_t before_allocation_hit_ratio = 0, insert_conflict_hit_ratio = 0, parallel_load_hit_ratio = 0;
1717 - if(pages_total) {
1718 - if(pages_total_hit > pages_total)
1719 - pages_total_hit = pages_total;
1713 + size_t main_cache_hit_ratio = 0, extent_cache_hit_ratio = 0, extent_merge_hit_ratio = 0;
1714 + if(delta_pages_total) {
1715 + if(pages_total_hit > delta_pages_total)
1716 + pages_total_hit = delta_pages_total;
1717 +
1718 + overall_hit_ratio = pages_total_hit * 100 * 10000 / delta_pages_total;
1719
1721 - overall_hit_ratio = pages_total_hit * 100 * 10000 / pages_total;
1720 + size_t delta_pages_main_cache = delta_pages_data_source_main_cache + delta_pages_pending_found_in_cache_at_pass4;
1721 + if(delta_pages_main_cache > delta_pages_total)
1722 + delta_pages_main_cache = delta_pages_total;
1723
1723 - preloaded_hit_ratio = pages_hit_at_query_plan * 100 * 10000 / pages_total;
1724 + main_cache_hit_ratio = delta_pages_main_cache * 100 * 10000 / delta_pages_total;
1725 }
1726
1726 - if(pages_to_load) {
1727 - extent_hit_ratio = pages_hit_cached_extent * 100 * 10000 / pages_to_load;
1728 - if(extent_hit_ratio > 100 * 10000)
1729 - extent_hit_ratio = 100 * 10000;
1727 + if(delta_pages_to_load_from_disk) {
1728 + if(delta_pages_data_source_extent_cache > delta_pages_to_load_from_disk)
1729 + delta_pages_data_source_extent_cache = delta_pages_to_load_from_disk;
1730
1731 - parallel_load_hit_ratio = pages_hit_parallel_load * 100 * 10000 / pages_to_load;
1732 - if(parallel_load_hit_ratio > 100 * 10000)
1733 - parallel_load_hit_ratio = 100 * 10000;
1731 + extent_cache_hit_ratio = delta_pages_data_source_extent_cache * 100 * 10000 / delta_pages_to_load_from_disk;
1732
1735 - before_allocation_hit_ratio = pages_hit_before_allocation * 100 * 10000 / pages_to_load;
1736 - if(before_allocation_hit_ratio > 100 * 10000)
1737 - before_allocation_hit_ratio = 100 * 10000;
1733 + if(delta_pages_load_extent_merged > delta_pages_to_load_from_disk)
1734 + delta_pages_load_extent_merged = delta_pages_to_load_from_disk;
1735
1739 - insert_conflict_hit_ratio = pages_hit_insert_conflict * 100 * 10000 / pages_to_load;
1740 - if(insert_conflict_hit_ratio > 100 * 10000)
1741 - insert_conflict_hit_ratio = 100 * 10000;
1736 + extent_merge_hit_ratio = delta_pages_load_extent_merged * 100 * 10000 / delta_pages_to_load_from_disk;
1737 }
1738
1739 rrddim_set_by_pointer(st_cache_hit_ratio, rd_hit_ratio, (collected_number)overall_hit_ratio);
1745 - rrddim_set_by_pointer(st_cache_hit_ratio, rd_preloaded_ratio, (collected_number)preloaded_hit_ratio);
1746 - rrddim_set_by_pointer(st_cache_hit_ratio, rd_extent_ratio, (collected_number)extent_hit_ratio);
1747 - rrddim_set_by_pointer(st_cache_hit_ratio, rd_parallel_load_ratio, (collected_number)parallel_load_hit_ratio);
1748 - rrddim_set_by_pointer(st_cache_hit_ratio, rd_before_allocation_ratio, (collected_number)before_allocation_hit_ratio);
1749 - rrddim_set_by_pointer(st_cache_hit_ratio, rd_insert_conflict_ratio, (collected_number)insert_conflict_hit_ratio);
1740 + rrddim_set_by_pointer(st_cache_hit_ratio, rd_main_cache_hit_ratio, (collected_number)main_cache_hit_ratio);
1741 + rrddim_set_by_pointer(st_cache_hit_ratio, rd_extent_cache_hit_ratio, (collected_number)extent_cache_hit_ratio);
1742 + rrddim_set_by_pointer(st_cache_hit_ratio, rd_extent_merge_hit_ratio, (collected_number)extent_merge_hit_ratio);
1743
1744 rrdset_done(st_cache_hit_ratio);
1745 }
@@ -1880,7 +1873,7 @@ static void dbengine2_statistics_charts(void) {
1873 }
1874 priority++;
1875
1883 - rrddim_set_by_pointer(st_query_pages_data_source, rd_pages_main_cache, (collected_number)cache_efficiency_stats.pages_data_source_main_cache);
1876 + rrddim_set_by_pointer(st_query_pages_data_source, rd_pages_main_cache, (collected_number)cache_efficiency_stats.pages_data_source_main_cache + (collected_number)cache_efficiency_stats.pages_data_source_main_cache_at_pass4);
1877 rrddim_set_by_pointer(st_query_pages_data_source, rd_pages_disk, (collected_number)cache_efficiency_stats.pages_to_load_from_disk);
1878 rrddim_set_by_pointer(st_query_pages_data_source, rd_pages_extent_cache, (collected_number)cache_efficiency_stats.pages_data_source_extent_cache);
1879
@@ -1918,7 +1911,7 @@ static void dbengine2_statistics_charts(void) {
1911 }
1912 priority++;
1913
1921 - rrddim_set_by_pointer(st_query_next_page, rd_pass4, (collected_number)cache_efficiency_stats.pages_pending_found_in_cache_at_pass4);
1914 + rrddim_set_by_pointer(st_query_next_page, rd_pass4, (collected_number)cache_efficiency_stats.pages_data_source_main_cache_at_pass4);
1915 rrddim_set_by_pointer(st_query_next_page, rd_wait_failed, (collected_number)cache_efficiency_stats.page_next_wait_failed);
1916 rrddim_set_by_pointer(st_query_next_page, rd_nowait_failed, (collected_number)cache_efficiency_stats.page_next_nowait_failed);
1917 rrddim_set_by_pointer(st_query_next_page, rd_wait_loaded, (collected_number)cache_efficiency_stats.page_next_wait_loaded);
@@ -1977,11 +1970,10 @@ static void dbengine2_statistics_charts(void) {
1970 static RRDDIM *rd_uncompressed = NULL;
1971 static RRDDIM *rd_mmap_failed = NULL;
1972 static RRDDIM *rd_unavailable = NULL;
1980 - static RRDDIM *rd_already_loaded = NULL;
1981 - static RRDDIM *rd_preloaded = NULL;
1973 static RRDDIM *rd_unroutable = NULL;
1974 static RRDDIM *rd_not_found = NULL;
1975 static RRDDIM *rd_invalid_extent = NULL;
1976 + static RRDDIM *rd_extent_merged = NULL;
1977
1978 if (unlikely(!st_query_pages_from_disk)) {
1979 st_query_pages_from_disk = rrdset_create_localhost(
@@ -2002,12 +1994,11 @@ static void dbengine2_statistics_charts(void) {
1994 rd_invalid = rrddim_add(st_query_pages_from_disk, "fail invalid page", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
1995 rd_uncompressed = rrddim_add(st_query_pages_from_disk, "ok uncompressed", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1996 rd_mmap_failed = rrddim_add(st_query_pages_from_disk, "fail cant mmap", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2005 - rd_already_loaded = rrddim_add(st_query_pages_from_disk, "ok but preloaded", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1997 rd_unavailable = rrddim_add(st_query_pages_from_disk, "fail unavailable", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2007 - rd_preloaded = rrddim_add(st_query_pages_from_disk, "ok preloaded", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1998 rd_unroutable = rrddim_add(st_query_pages_from_disk, "fail unroutable", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
1999 rd_not_found = rrddim_add(st_query_pages_from_disk, "fail uuid not found", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2000 rd_invalid_extent = rrddim_add(st_query_pages_from_disk, "fail invalid extent", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
2001 + rd_extent_merged = rrddim_add(st_query_pages_from_disk, "extent merged", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2002 }
2003 priority++;
2004
@@ -2015,12 +2006,11 @@ static void dbengine2_statistics_charts(void) {
2006 rrddim_set_by_pointer(st_query_pages_from_disk, rd_invalid, (collected_number)cache_efficiency_stats.pages_load_fail_invalid_page_in_extent);
2007 rrddim_set_by_pointer(st_query_pages_from_disk, rd_uncompressed, (collected_number)cache_efficiency_stats.pages_load_ok_uncompressed);
2008 rrddim_set_by_pointer(st_query_pages_from_disk, rd_mmap_failed, (collected_number)cache_efficiency_stats.pages_load_fail_cant_mmap_extent);
2018 - rrddim_set_by_pointer(st_query_pages_from_disk, rd_already_loaded, (collected_number)cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_before_allocation);
2009 rrddim_set_by_pointer(st_query_pages_from_disk, rd_unavailable, (collected_number)cache_efficiency_stats.pages_load_fail_datafile_not_available);
2020 - rrddim_set_by_pointer(st_query_pages_from_disk, rd_preloaded, (collected_number)cache_efficiency_stats.pages_load_ok_preloaded);
2010 rrddim_set_by_pointer(st_query_pages_from_disk, rd_unroutable, (collected_number)cache_efficiency_stats.pages_load_fail_unroutable);
2011 rrddim_set_by_pointer(st_query_pages_from_disk, rd_not_found, (collected_number)cache_efficiency_stats.pages_load_fail_uuid_not_found);
2012 rrddim_set_by_pointer(st_query_pages_from_disk, rd_invalid_extent, (collected_number)cache_efficiency_stats.pages_load_fail_invalid_extent);
2013 + rrddim_set_by_pointer(st_query_pages_from_disk, rd_extent_merged, (collected_number)cache_efficiency_stats.pages_load_extent_merged);
2014
2015 rrdset_done(st_query_pages_from_disk);
2016 }
database/engine/cache.c
+56 -8
@@ -353,7 +353,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
353 #define evict_pages(cache, max_skip, max_evict, wait, all_of_them) evict_pages_with_filter(cache, max_skip, max_evict, wait, all_of_them, NULL, NULL)
354
355 static inline void evict_on_clean_page_added(PGC *cache __maybe_unused) {
356 - if((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_needs_space_aggressively(cache)) {
356 + if((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_under_severe_pressure(cache)) {
357 evict_pages(cache,
358 cache->config.max_skip_pages_per_inline_eviction,
359 cache->config.max_pages_per_inline_eviction,
@@ -362,7 +362,7 @@ static inline void evict_on_clean_page_added(PGC *cache __maybe_unused) {
362 }
363
364 static inline void evict_on_page_release_when_permitted(PGC *cache __maybe_unused) {
365 - if (unlikely((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_needs_space_aggressively(cache))) {
365 + if (unlikely((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_under_severe_pressure(cache))) {
366 evict_pages(cache,
367 cache->config.max_skip_pages_per_inline_eviction,
368 cache->config.max_pages_per_inline_eviction,
@@ -1046,8 +1046,11 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
1046
1047 DOUBLE_LINKED_LIST_APPEND_UNSAFE(pages_to_evict, page, link.prev, link.next);
1048
1049 - if(!all_of_them)
1050 - // we do it one-by-one to avoid locking clean queue for way too long
1049 + if(unlikely(all_of_them))
1050 + // get more pages
1051 + ;
1052 + else
1053 + // one page at a time
1054 break;
1055 }
1056 else {
@@ -1071,8 +1074,53 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
1074 if(likely(pages_to_evict)) {
1075 // remove them from the index
1076
1074 - for(PGC_PAGE *page = pages_to_evict, *next = NULL; page ; page = next) {
1075 - next = page->link.next;
1077 + if(unlikely(pages_to_evict->link.next)) {
1078 + // we have many pages, let's minimize the index locks we are going to get
1079 +
1080 + PGC_PAGE *pages_per_partition[cache->config.partitions];
1081 + memset(pages_per_partition, 0, sizeof(PGC_PAGE *) * cache->config.partitions);
1082 +
1083 + // sort them by partition
1084 + for (PGC_PAGE *page = pages_to_evict, *next = NULL; page; page = next) {
1085 + next = page->link.next;
1086 +
1087 + size_t partition = pgc_indexing_partition(cache, page->metric_id);
1088 + DOUBLE_LINKED_LIST_REMOVE_UNSAFE(pages_to_evict, page, link.prev, link.next);
1089 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(pages_per_partition[partition], page, link.prev, link.next);
1090 + }
1091 +
1092 + // remove them from the index
1093 + for (size_t partition = 0; partition < cache->config.partitions; partition++) {
1094 + if (!pages_per_partition[partition]) continue;
1095 +
1096 + pgc_index_write_lock(cache, partition);
1097 +
1098 + for (PGC_PAGE *page = pages_per_partition[partition]; page; page = page->link.next)
1099 + remove_this_page_from_index_unsafe(cache, page, partition);
1100 +
1101 + pgc_index_write_unlock(cache, partition);
1102 + }
1103 +
1104 + // free them
1105 + for (size_t partition = 0; partition < cache->config.partitions; partition++) {
1106 + if (!pages_per_partition[partition]) continue;
1107 +
1108 + for (PGC_PAGE *page = pages_per_partition[partition], *next = NULL; page; page = next) {
1109 + next = page->link.next;
1110 +
1111 + size_t page_size = page->assumed_size;
1112 + free_this_page(cache, page);
1113 +
1114 + __atomic_sub_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED);
1115 + __atomic_sub_fetch(&cache->stats.evicting_size, page_size, __ATOMIC_RELAXED);
1116 +
1117 + total_pages_evicted++;
1118 + }
1119 + }
1120 + }
1121 + else {
1122 + // just one page to be evicted
1123 + PGC_PAGE *page = pages_to_evict;
1124
1125 size_t page_size = page->assumed_size;
1126
@@ -1676,7 +1724,7 @@ PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb,
1724 cache->config.additional_bytes_per_page = additional_bytes_per_page;
1725
1726 cache->config.max_workers_evict_inline = 10;
1679 - cache->config.severe_pressure_per1000 = 1000;
1727 + cache->config.severe_pressure_per1000 = 1010;
1728 cache->config.aggressive_evict_per1000 = 990;
1729 cache->config.healthy_size_per1000 = 980;
1730 cache->config.evict_low_threshold_per1000 = 970;
@@ -1896,7 +1944,7 @@ size_t pgc_get_wanted_cache_size(PGC *cache) {
1944 }
1945
1946 bool pgc_evict_pages(PGC *cache, size_t max_skip, size_t max_evict) {
1899 - bool under_pressure = cache_under_severe_pressure(cache);
1947 + bool under_pressure = cache_needs_space_aggressively(cache);
1948 return evict_pages(cache,
1949 under_pressure ? 0 : max_skip,
1950 under_pressure ? 0 : max_evict,
database/engine/datafile.c
+1 -1
@@ -28,8 +28,8 @@ static struct rrdengine_datafile *datafile_alloc_and_init(struct rrdengine_insta
28 datafile->users.available = true;
29
30 netdata_spinlock_init(&datafile->users.spinlock);
31 - netdata_spinlock_init(&datafile->extent_exclusive_access.spinlock);
31 netdata_spinlock_init(&datafile->writers.spinlock);
32 + netdata_spinlock_init(&datafile->extent_queries.spinlock);
33
34 return datafile;
35 }
database/engine/datafile.h
+5 -7
@@ -44,13 +44,6 @@ struct rrdengine_datafile {
44 size_t flushed_to_open_running;
45 } writers;
46
47 - // exclusive access to extents
48 - struct {
49 - SPINLOCK spinlock;
50 - unsigned lockers;
51 - Pvoid_t extents_JudyL;
52 - } extent_exclusive_access;
53 -
47 struct {
48 SPINLOCK spinlock;
49 unsigned lockers;
@@ -58,6 +51,11 @@ struct rrdengine_datafile {
51 bool available;
52 time_t time_to_evict;
53 } users;
54 +
55 + struct {
56 + SPINLOCK spinlock;
57 + Pvoid_t pending_epdl_by_extent_offset_judyL;
58 + } extent_queries;
59 };
60
61 typedef enum __attribute__ ((__packed__)) {
database/engine/pagecache.c
+3 -4
@@ -55,7 +55,7 @@ static void main_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_
55
56 struct completion completion;
57 completion_init(&completion);
58 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_FLUSH_PAGES, base, &completion, STORAGE_PRIORITY_CRITICAL);
58 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_FLUSH_PAGES, base, &completion, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
59 completion_wait_for(&completion);
60 completion_destroy(&completion);
61 }
@@ -713,7 +713,7 @@ we_are_done:
713 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_meta_source_open_cache, pages_found_in_open_cache, __ATOMIC_RELAXED);
714 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_meta_source_journal_v2, pages_found_in_journals_v2, __ATOMIC_RELAXED);
715 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_main_cache, pages_found_in_main_cache, __ATOMIC_RELAXED);
716 - __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_pending_found_in_cache_at_pass4, pages_found_pass4, __ATOMIC_RELAXED);
716 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_main_cache_at_pass4, pages_found_pass4, __ATOMIC_RELAXED);
717 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_to_load_from_disk, pages_pending, __ATOMIC_RELAXED);
718 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_overlapping_skipped, pages_overlapping, __ATOMIC_RELAXED);
719
@@ -739,7 +739,6 @@ void rrdeng_prep_query(PDC *pdc) {
739
740 if (pages_to_load && pdc->page_list_JudyL) {
741 pdc_acquire(pdc); // we get 1 for the 1st worker in the chain: do_read_page_list_work()
742 - pdc->preload_all_extent_pages = false;
742 usec_t start_ut = now_monotonic_usec();
743 // if(likely(priority == STORAGE_PRIORITY_BEST_EFFORT))
744 // dbengine_load_page_list_directly(ctx, handle->pdc);
@@ -783,7 +782,7 @@ void pg_cache_preload(struct rrdeng_query_handle *handle) {
782
783 if(ctx_is_available_for_queries(handle->ctx)) {
784 handle->pdc->refcount++; // we get 1 for the query thread and 1 for the prep thread
786 - rrdeng_enq_cmd(handle->ctx, RRDENG_OPCODE_PREP_QUERY, handle->pdc, NULL, handle->priority);
785 + rrdeng_enq_cmd(handle->ctx, RRDENG_OPCODE_PREP_QUERY, handle->pdc, NULL, handle->priority, NULL, NULL);
786 }
787 else {
788 completion_mark_complete(&handle->pdc->prep_completion);
database/engine/pdc.c
+177 -155
@@ -11,6 +11,13 @@ struct extent_page_details_list {
11 struct page_details_control *pdc;
12 struct rrdengine_datafile *datafile;
13
14 + struct rrdeng_cmd *cmd;
15 +
16 + struct {
17 + struct extent_page_details_list *prev;
18 + struct extent_page_details_list *next;
19 + } query;
20 +
21 struct {
22 struct extent_page_details_list *prev;
23 struct extent_page_details_list *next;
@@ -465,7 +472,7 @@ static void epdl_mark_all_not_loaded_pages_as_failed(EPDL *epdl, PDC_PAGE_STATUS
472 if(pages_matched && statistics_counter)
473 __atomic_add_fetch(statistics_counter, pages_matched, __ATOMIC_RELAXED);
474 }
468 -
475 +/*
476 static bool epdl_check_if_pages_are_already_in_cache(struct rrdengine_instance *ctx, EPDL *epdl, PDC_PAGE_STATUS tags)
477 {
478 size_t count_remaining = 0;
@@ -501,6 +508,7 @@ static bool epdl_check_if_pages_are_already_in_cache(struct rrdengine_instance *
508
509 return count_remaining == 0;
510 }
511 +*/
512
513 // ----------------------------------------------------------------------------
514 // PDC logic
@@ -587,6 +595,56 @@ bool pdc_release_and_destroy_if_unreferenced(PDC *pdc, bool worker, bool router
595 return false;
596 }
597
598 +void epdl_cmd_queued(void *epdl_ptr, struct rrdeng_cmd *cmd) {
599 + EPDL *epdl = epdl_ptr;
600 + epdl->cmd = cmd;
601 +}
602 +
603 +void epdl_cmd_dequeued(void *epdl_ptr) {
604 + EPDL *epdl = epdl_ptr;
605 + epdl->cmd = NULL;
606 +}
607 +
608 +static struct rrdeng_cmd *epdl_get_cmd(void *epdl_ptr) {
609 + EPDL *epdl = epdl_ptr;
610 + return epdl->cmd;
611 +}
612 +
613 +static bool epdl_pending_add(EPDL *epdl) {
614 + bool added_new;
615 +
616 + netdata_spinlock_lock(&epdl->datafile->extent_queries.spinlock);
617 + Pvoid_t *PValue = JudyLIns(&epdl->datafile->extent_queries.pending_epdl_by_extent_offset_judyL, epdl->extent_offset, PJE0);
618 + internal_fatal(!PValue || PValue == PJERR, "DBENGINE: corrupted pending extent judy");
619 +
620 + EPDL *base = *PValue;
621 +
622 + if(!base)
623 + added_new = true;
624 + else {
625 + added_new = false;
626 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_extent_merged, 1, __ATOMIC_RELAXED);
627 +
628 + if(base->pdc->priority > epdl->pdc->priority)
629 + rrdeng_req_cmd(epdl_get_cmd, base, epdl->pdc->priority);
630 + }
631 +
632 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(base, epdl, query.prev, query.next);
633 + *PValue = base;
634 +
635 + netdata_spinlock_unlock(&epdl->datafile->extent_queries.spinlock);
636 +
637 + return added_new;
638 +}
639 +
640 +static void epdl_pending_del(EPDL *epdl) {
641 + netdata_spinlock_lock(&epdl->datafile->extent_queries.spinlock);
642 + int rc = JudyLDel(&epdl->datafile->extent_queries.pending_epdl_by_extent_offset_judyL, epdl->extent_offset, PJE0);
643 + (void)rc;
644 + internal_fatal(!rc, "DBENGINE: epdl not found in pending list");
645 + netdata_spinlock_unlock(&epdl->datafile->extent_queries.spinlock);
646 +}
647 +
648 void pdc_to_epdl_router(struct rrdengine_instance *ctx, PDC *pdc, execute_extent_page_details_list_t exec_first_extent_list, execute_extent_page_details_list_t exec_rest_extent_list)
649 {
650 Pvoid_t *PValue;
@@ -674,10 +732,12 @@ void pdc_to_epdl_router(struct rrdengine_instance *ctx, PDC *pdc, execute_extent
732 pdc_acquire(pdc); // we do this for the next worker: do_read_extent_work()
733 epdl->pdc = pdc;
734
677 - if(extent_list_no++ == 0)
678 - exec_first_extent_list(ctx, epdl, pdc->priority);
679 - else
680 - exec_rest_extent_list(ctx, epdl, pdc->priority);
735 + if(epdl_pending_add(epdl)) {
736 + if (extent_list_no++ == 0)
737 + exec_first_extent_list(ctx, epdl, pdc->priority);
738 + else
739 + exec_rest_extent_list(ctx, epdl, pdc->priority);
740 + }
741 }
742 PDCJudyLFreeArray(&deol->extent_pd_list_by_extent_offset_JudyL, PJE0);
743 deol_release(deol);
@@ -688,51 +748,6 @@ void pdc_to_epdl_router(struct rrdengine_instance *ctx, PDC *pdc, execute_extent
748 pdc_release_and_destroy_if_unreferenced(pdc, true, true);
749 }
750
691 -static bool datafile_get_exclusive_access_to_extent(EPDL *epdl) {
692 - struct rrdengine_datafile *df = epdl->datafile;
693 - bool is_it_mine = false;
694 -
695 - while(!is_it_mine) {
696 - netdata_spinlock_lock(&df->extent_exclusive_access.spinlock);
697 - if(!df->users.available) {
698 - netdata_spinlock_unlock(&df->extent_exclusive_access.spinlock);
699 - return false;
700 - }
701 - Pvoid_t *PValue = JudyLIns(&df->extent_exclusive_access.extents_JudyL, epdl->extent_offset, PJE0);
702 - if (!*PValue) {
703 - *(Word_t *) PValue = gettid();
704 - df->extent_exclusive_access.lockers++;
705 - is_it_mine = true;
706 - }
707 - netdata_spinlock_unlock(&df->extent_exclusive_access.spinlock);
708 -
709 - if(!is_it_mine) {
710 - static const struct timespec ns = { .tv_sec = 0, .tv_nsec = 1 };
711 - nanosleep(&ns, NULL);
712 - }
713 - }
714 - return true;
715 -}
716 -
717 -static void datafile_release_exclusive_access_to_extent(EPDL *epdl) {
718 - struct rrdengine_datafile *df = epdl->datafile;
719 -
720 - netdata_spinlock_lock(&df->extent_exclusive_access.spinlock);
721 -
722 -#ifdef NETDATA_INTERNAL_CHECKS
723 - Pvoid_t *PValue = JudyLGet(df->extent_exclusive_access.extents_JudyL, epdl->extent_offset, PJE0);
724 - if (*(Word_t *) PValue != (Word_t)gettid())
725 - fatal("DBENGINE: exclusive extent access is not mine");
726 -#endif
727 -
728 - int rc = JudyLDel(&df->extent_exclusive_access.extents_JudyL, epdl->extent_offset, PJE0);
729 - if (!rc)
730 - fatal("DBENGINE: cannot find my exclusive access");
731 -
732 - df->extent_exclusive_access.lockers--;
733 - netdata_spinlock_unlock(&df->extent_exclusive_access.spinlock);
734 -}
735 -
751 static void fill_page_with_nulls(void *page, uint32_t page_length, uint8_t type) {
752 switch(type) {
753 case PAGE_METRICS: {
@@ -843,12 +858,34 @@ inline VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_
858 return vd;
859 }
860
861 +static struct page_details *epdl_to_pd_load_list(EPDL *epdl, Word_t metric_id, time_t start_time_s) {
862 + struct page_details *pd_list = NULL;
863 +
864 + for(EPDL *ep = epdl; ep ;ep = ep->query.next) {
865 + Pvoid_t *pd_by_start_time_s_judyL = PDCJudyLGet(ep->page_details_by_metric_id_JudyL, metric_id, PJE0);
866 + internal_fatal(pd_by_start_time_s_judyL == PJERR, "DBENGINE: corrupted extent metrics JudyL");
867 +
868 + if (pd_by_start_time_s_judyL && *pd_by_start_time_s_judyL) {
869 + Pvoid_t *pd_pptr = PDCJudyLGet(*pd_by_start_time_s_judyL, start_time_s, PJE0);
870 + internal_fatal(pd_pptr == PJERR, "DBENGINE: corrupted metric page details JudyHS");
871 +
872 + if (pd_pptr && *pd_pptr) {
873 + struct page_details *pd = *pd_pptr;
874 + internal_fatal(metric_id != pd->metric_id, "DBENGINE: metric ids do not match");
875 +
876 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(pd_list, pd, load.prev, load.next);
877 + }
878 + }
879 + }
880 +
881 + return pd_list;
882 +}
883 +
884 static bool epdl_populate_pages_from_extent_data(
885 struct rrdengine_instance *ctx,
886 void *data,
887 size_t data_length,
888 EPDL *epdl,
851 - bool preload_all_pages,
889 bool worker,
890 PDC_PAGE_STATUS tags,
891 bool cached_extent)
@@ -936,13 +973,15 @@ static bool epdl_populate_pages_from_extent_data(
973 }
974 }
975
976 + if(worker)
977 + worker_is_busy(UV_EVENT_PAGE_LOOKUP);
978 +
979 size_t stats_data_from_main_cache = 0;
980 size_t stats_data_from_extent = 0;
981 size_t stats_load_compressed = 0;
982 size_t stats_load_uncompressed = 0;
983 size_t stats_load_invalid_page = 0;
984 size_t stats_cache_hit_while_inserting = 0;
945 - size_t stats_cache_hit_before_allocation = 0;
985
986 uint32_t page_offset = 0, page_length;
987 time_t now_s = now_realtime_sec();
@@ -957,9 +996,6 @@ static bool epdl_populate_pages_from_extent_data(
996 continue;
997 }
998
960 - if(worker)
961 - worker_is_busy(UV_EVENT_METRIC_LOOKUP);
962 -
999 METRIC *metric = mrg_metric_get_and_acquire(main_mrg, &header->descr[i].uuid, (Word_t)ctx);
1000 Word_t metric_id = (Word_t)metric;
1001 if(!metric) {
@@ -970,97 +1006,81 @@ static bool epdl_populate_pages_from_extent_data(
1006 }
1007 mrg_metric_release(main_mrg, metric);
1008
973 - if(worker)
974 - worker_is_busy(UV_EVENT_PAGE_LOOKUP);
975 -
976 - struct page_details *pd = NULL;
977 - Pvoid_t *pd_by_start_time_s_judyL = PDCJudyLGet(epdl->page_details_by_metric_id_JudyL, metric_id, PJE0);
978 - internal_fatal(pd_by_start_time_s_judyL == PJERR, "DBENGINE: corrupted extent metrics JudyL");
979 -
980 - if(pd_by_start_time_s_judyL && *pd_by_start_time_s_judyL) {
981 - Pvoid_t *pd_pptr = PDCJudyLGet(*pd_by_start_time_s_judyL, start_time_s, PJE0);
982 - internal_fatal(pd_pptr == PJERR, "DBENGINE: corrupted metric page details JudyHS");
983 -
984 - if(pd_pptr && *pd_pptr) {
985 - pd = *pd_pptr;
986 - internal_fatal(metric_id != pd->metric_id, "DBENGINE: metric ids do not match");
987 - }
988 - }
989 -
990 - if(!pd && !preload_all_pages)
1009 + struct page_details *pd_list = epdl_to_pd_load_list(epdl, metric_id, start_time_s);
1010 + if(likely(!pd_list))
1011 continue;
1012
1013 VALIDATED_PAGE_DESCRIPTOR vd = validate_extent_page_descr(
1014 &header->descr[i], now_s,
995 - (pd) ? pd->update_every_s : 0,
1015 + (pd_list) ? pd_list->update_every_s : 0,
1016 have_read_error);
1017
1018 if(worker)
1019 worker_is_busy(UV_EVENT_PAGE_POPULATION);
1020
1001 - PGC_PAGE *page = pgc_page_get_and_acquire(main_cache, (Word_t)ctx, metric_id, start_time_s, PGC_SEARCH_EXACT);
1002 - if (!page) {
1003 - void *page_data = dbengine_page_alloc(ctx, vd.page_length);
1021 + void *page_data = dbengine_page_alloc(ctx, vd.page_length);
1022 +
1023 + if (unlikely(!vd.data_on_disk_valid)) {
1024 + fill_page_with_nulls(page_data, vd.page_length, vd.type);
1025 + stats_load_invalid_page++;
1026 + }
1027 +
1028 + else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
1029 + memcpy(page_data, data + payload_offset + page_offset, (size_t) vd.page_length);
1030 + stats_load_uncompressed++;
1031 + }
1032 +
1033 + else {
1034 + if(unlikely(page_offset + vd.page_length > uncompressed_payload_length)) {
1035 + error_limit_static_global_var(erl, 10, 0);
1036 + error_limit(&erl,
1037 + "DBENGINE: page %u offset %u + page length %zu exceeds the uncompressed buffer size %u",
1038 + i, page_offset, vd.page_length, uncompressed_payload_length);
1039
1005 - if (unlikely(!vd.data_on_disk_valid)) {
1040 fill_page_with_nulls(page_data, vd.page_length, vd.type);
1041 stats_load_invalid_page++;
1042 }
1009 -
1010 - else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
1011 - memcpy(page_data, data + payload_offset + page_offset, (size_t) vd.page_length);
1012 - stats_load_uncompressed++;
1013 - }
1014 -
1043 else {
1016 - if(unlikely(page_offset + vd.page_length > uncompressed_payload_length)) {
1017 - error_limit_static_global_var(erl, 10, 0);
1018 - error_limit(&erl,
1019 - "DBENGINE: page %u offset %u + page length %zu exceeds the uncompressed buffer size %u",
1020 - i, page_offset, vd.page_length, uncompressed_payload_length);
1021 -
1022 - fill_page_with_nulls(page_data, vd.page_length, vd.type);
1023 - stats_load_invalid_page++;
1024 - }
1025 - else {
1026 - memcpy(page_data, uncompressed_buf + page_offset, vd.page_length);
1027 - stats_load_compressed++;
1028 - }
1044 + memcpy(page_data, uncompressed_buf + page_offset, vd.page_length);
1045 + stats_load_compressed++;
1046 }
1030 -
1031 - PGC_ENTRY page_entry = {
1032 - .hot = false,
1033 - .section = (Word_t)ctx,
1034 - .metric_id = metric_id,
1035 - .start_time_s = vd.start_time_s,
1036 - .end_time_s = vd.end_time_s,
1037 - .update_every_s = vd.update_every_s,
1038 - .size = (size_t) vd.page_length,
1039 - .data = page_data
1040 - };
1041 -
1042 - bool added = true;
1043 - page = pgc_page_add_and_acquire(main_cache, page_entry, &added);
1044 - if (false == added) {
1045 - dbengine_page_free(page_data);
1046 - stats_cache_hit_while_inserting++;
1047 - stats_data_from_main_cache++;
1048 - }
1049 - else
1050 - stats_data_from_extent++;
1047 }
1052 - else {
1053 - stats_cache_hit_before_allocation++;
1048 +
1049 + PGC_ENTRY page_entry = {
1050 + .hot = false,
1051 + .section = (Word_t)ctx,
1052 + .metric_id = metric_id,
1053 + .start_time_s = vd.start_time_s,
1054 + .end_time_s = vd.end_time_s,
1055 + .update_every_s = vd.update_every_s,
1056 + .size = (size_t) vd.page_length,
1057 + .data = page_data
1058 + };
1059 +
1060 + bool added = true;
1061 + PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, &added);
1062 + if (false == added) {
1063 + dbengine_page_free(page_data);
1064 + stats_cache_hit_while_inserting++;
1065 stats_data_from_main_cache++;
1066 }
1067 + else
1068 + stats_data_from_extent++;
1069 +
1070 + struct page_details *pd = pd_list;
1071 + do {
1072 + if(pd != pd_list)
1073 + pgc_page_dup(main_cache, page);
1074
1057 - if (pd) {
1075 pd->page = page;
1076 pd->page_length = pgc_page_data_size(main_cache, page);
1077 pdc_page_status_set(pd, PDC_PAGE_READY | tags);
1061 - }
1062 - else
1063 - pgc_page_release(main_cache, page);
1078 +
1079 + pd = pd->load.next;
1080 + } while(pd);
1081 +
1082 + if(worker)
1083 + worker_is_busy(UV_EVENT_PAGE_LOOKUP);
1084 }
1085
1086 if(stats_data_from_main_cache)
@@ -1068,11 +1088,10 @@ static bool epdl_populate_pages_from_extent_data(
1088
1089 if(cached_extent)
1090 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_extent_cache, stats_data_from_extent, __ATOMIC_RELAXED);
1071 - else
1091 + else {
1092 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_data_source_disk, stats_data_from_extent, __ATOMIC_RELAXED);
1073 -
1074 - if(stats_cache_hit_before_allocation)
1075 - __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_before_allocation, stats_cache_hit_before_allocation, __ATOMIC_RELAXED);
1093 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.extents_loaded_from_disk, 1, __ATOMIC_RELAXED);
1094 + }
1095
1096 if(stats_cache_hit_while_inserting)
1097 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_ok_loaded_but_cache_hit_while_inserting, stats_cache_hit_while_inserting, __ATOMIC_RELAXED);
@@ -1095,22 +1114,23 @@ static bool epdl_populate_pages_from_extent_data(
1114 }
1115
1116 void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *epdl, bool worker) {
1098 - struct page_details_control *pdc = epdl->pdc;
1099 -
1100 - bool extent_exclusive = false;
1101 -
1102 - if(pdc->preload_all_extent_pages) {
1103 - if (!datafile_get_exclusive_access_to_extent(epdl)) {
1104 - __atomic_add_fetch(&rrdeng_cache_efficiency_stats.pages_load_fail_datafile_not_available, 1, __ATOMIC_RELAXED);
1105 - goto cleanup;
1117 + epdl_pending_del(epdl);
1118 +
1119 + bool should_stop = __atomic_load_n(&epdl->pdc->workers_should_stop, __ATOMIC_RELAXED);
1120 + for(EPDL *ep = epdl->query.next; ep ;ep = ep->query.next) {
1121 + internal_fatal(ep->datafile != epdl->datafile, "DBENGINE: datafiles do not match");
1122 + internal_fatal(ep->extent_offset != epdl->extent_offset, "DBENGINE: extent offsets do not match");
1123 + internal_fatal(ep->extent_size != epdl->extent_size, "DBENGINE: extent sizes do not match");
1124 + internal_fatal(ep->file != epdl->file, "DBENGINE: files do not match");
1125 +
1126 + PDC *pdc = ep->pdc;
1127 + if(!__atomic_load_n(&pdc->workers_should_stop, __ATOMIC_RELAXED)) {
1128 + should_stop = false;
1129 + break;
1130 }
1107 - extent_exclusive = true;
1131 }
1132
1110 - if (epdl_check_if_pages_are_already_in_cache(ctx, epdl, PDC_PAGE_PRELOADED_WORKER))
1111 - goto cleanup;
1112 -
1113 - if(__atomic_load_n(&pdc->workers_should_stop, __ATOMIC_RELAXED))
1133 + if(should_stop)
1134 goto cleanup;
1135
1136 if(worker)
@@ -1183,8 +1203,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1203 // Need to decompress and then process the pagelist
1204 bool extent_used = epdl_populate_pages_from_extent_data(
1205 ctx, extent_compressed_data, epdl->extent_size,
1186 - epdl, pdc->preload_all_extent_pages,
1187 - worker, loaded_pages_tag, extent_found_in_cache);
1206 + epdl, worker, loaded_pages_tag, extent_found_in_cache);
1207
1208 if(extent_used) {
1209 // since the extent was used, all the pages that are not
@@ -1198,23 +1217,26 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1217 not_loaded_pages_tag |= PDC_PAGE_FAILED_TO_MAP_EXTENT;
1218
1219
1201 - // mark all pending pages as failed
1202 - epdl_mark_all_not_loaded_pages_as_failed(
1203 - epdl, not_loaded_pages_tag,
1204 - &rrdeng_cache_efficiency_stats.pages_load_fail_cant_mmap_extent);
1205 -
1220 if(extent_cache_page)
1221 pgc_page_release(extent_cache, extent_cache_page);
1222
1209 - cleanup:
1210 - if(extent_exclusive)
1211 - datafile_release_exclusive_access_to_extent(epdl);
1223 + // mark all pending pages as failed
1224 + for(EPDL *ep = epdl; ep ;ep = ep->query.next) {
1225 + epdl_mark_all_not_loaded_pages_as_failed(
1226 + ep, not_loaded_pages_tag,
1227 + &rrdeng_cache_efficiency_stats.pages_load_fail_cant_mmap_extent);
1228 + }
1229 +
1230 +cleanup:
1231 + for(EPDL *ep = epdl, *next = NULL; ep ; ep = next) {
1232 + next = ep->query.next;
1233
1213 - completion_mark_complete_a_job(&epdl->pdc->page_completion);
1214 - pdc_release_and_destroy_if_unreferenced(pdc, true, false);
1234 + completion_mark_complete_a_job(&ep->pdc->page_completion);
1235 + pdc_release_and_destroy_if_unreferenced(ep->pdc, true, false);
1236
1216 - // Free the Judy that holds the requested pagelist and the extents
1217 - epdl_destroy(epdl);
1237 + // Free the Judy that holds the requested pagelist and the extents
1238 + epdl_destroy(ep);
1239 + }
1240
1241 if(worker)
1242 worker_is_idle();
database/engine/pdc.h
+5
@@ -5,6 +5,8 @@
5
6 #include "../engine/rrdengine.h"
7
8 +struct rrdeng_cmd;
9 +
10 #ifdef PDC_USE_JULYL
11 #define PDCJudyLIns JulyLIns
12 #define PDCJudyLGet JulyLGet
@@ -44,6 +46,9 @@ void epdl_cleanup(void);
46 void deol_cleanup(void);
47 void extent_buffer_cleanup(void);
48
49 +void epdl_cmd_dequeued(void *epdl_ptr);
50 +void epdl_cmd_queued(void *epdl_ptr, struct rrdeng_cmd *cmd);
51 +
52 struct extent_buffer {
53 size_t bytes;
54
database/engine/rrdengine.c
+56 -27
@@ -25,16 +25,16 @@ struct rrdeng_main {
25
26 time_t last_buffers_cleanup_s;
27
28 - bool flush_running;
29 - bool evict_running;
28 + size_t flushes_running;
29 + size_t evictions_running;
30 } rrdeng_main = {
31 .thread = 0,
32 .loop = {},
33 .async = {},
34 .timer = {},
35 .last_buffers_cleanup_s = 0,
36 - .flush_running = false,
37 - .evict_running = false,
36 + .flushes_running = 0,
37 + .evictions_running = 0,
38 };
39
40 static void sanity_check(void)
@@ -515,6 +515,7 @@ struct rrdeng_cmd {
515 void *data;
516 struct completion *completion;
517 enum storage_priority priority;
518 + dequeue_callback_t dequeue_cb;
519
520 struct {
521 struct rrdeng_cmd *prev;
@@ -568,7 +569,29 @@ static void rrdeng_cmd_cleanup(void) {
569 netdata_spinlock_unlock(&rrdeng_cmd_globals.cache.spinlock);
570 }
571
571 -void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, void *data, struct completion *completion, STORAGE_PRIORITY priority) {
572 +void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd) {
573 + epdl_cmd_queued(cmd->data, cmd);
574 +}
575 +
576 +void rrdeng_dequeue_epdl_cmd(struct rrdeng_cmd *cmd) {
577 + epdl_cmd_dequeued(cmd->data);
578 +}
579 +
580 +void rrdeng_req_cmd(requeue_callback_t get_cmd_cb, void *data, STORAGE_PRIORITY priority) {
581 + netdata_spinlock_lock(&rrdeng_cmd_globals.queue.spinlock);
582 +
583 + struct rrdeng_cmd *cmd = get_cmd_cb(data);
584 + if(cmd && cmd->priority > priority) {
585 + DOUBLE_LINKED_LIST_REMOVE_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[cmd->priority], cmd, cache.prev, cache.next);
586 + DOUBLE_LINKED_LIST_APPEND_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[priority], cmd, cache.prev, cache.next);
587 + cmd->priority = priority;
588 + }
589 +
590 + netdata_spinlock_unlock(&rrdeng_cmd_globals.queue.spinlock);
591 +}
592 +
593 +void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, void *data, struct completion *completion,
594 + enum storage_priority priority, enqueue_callback_t enqueue_cb, dequeue_callback_t dequeue_cb) {
595 struct rrdeng_cmd *cmd = NULL;
596
597 if(unlikely(priority >= STORAGE_PRIO_MAX_DONT_USE))
@@ -593,10 +616,13 @@ void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, v
616 cmd->data = data;
617 cmd->completion = completion;
618 cmd->priority = priority;
619 + cmd->dequeue_cb = dequeue_cb;
620
621 netdata_spinlock_lock(&rrdeng_cmd_globals.queue.spinlock);
622 DOUBLE_LINKED_LIST_APPEND_UNSAFE(rrdeng_cmd_globals.queue.waiting_items_by_priority[priority], cmd, cache.prev, cache.next);
623 rrdeng_cmd_globals.queue.waiting++;
624 + if(enqueue_cb)
625 + enqueue_cb(cmd);
626 netdata_spinlock_unlock(&rrdeng_cmd_globals.queue.spinlock);
627
628 fatal_assert(0 == uv_async_send(&rrdeng_main.async));
@@ -637,6 +663,12 @@ static inline struct rrdeng_cmd rrdeng_deq_cmd(void) {
663 break;
664 }
665 }
666 +
667 + if(cmd && cmd->dequeue_cb) {
668 + cmd->dequeue_cb(cmd);
669 + cmd->dequeue_cb = NULL;
670 + }
671 +
672 netdata_spinlock_unlock(&rrdeng_cmd_globals.queue.spinlock);
673
674 struct rrdeng_cmd ret;
@@ -716,7 +748,7 @@ static void after_extent_flushed_to_open(struct rrdengine_instance *ctx __maybe_
748 completion_mark_complete(completion);
749
750 if(ctx_is_available_for_queries(ctx))
719 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL);
751 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
752 }
753
754 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) {
@@ -762,7 +794,7 @@ static void extent_flushed_to_open_tp_worker(struct rrdengine_instance *ctx __ma
794
795 if(datafile->fileno != __atomic_load_n(&ctx->last_fileno, __ATOMIC_RELAXED) && still_running)
796 // we just finished a flushing on a datafile that is not the active one
765 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL);
797 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
798 }
799
800 // Main event loop callback
@@ -778,7 +810,8 @@ static void extent_flush_io_callback(uv_fs_t *uv_fs_request) {
810 datafile->writers.running--;
811
812 datafile->writers.flushed_to_open_running++;
781 - rrdeng_enq_cmd(xt_io_descr->ctx, RRDENG_OPCODE_FLUSHED_TO_OPEN, uv_fs_request, xt_io_descr->completion, STORAGE_PRIORITY_CRITICAL);
813 + rrdeng_enq_cmd(xt_io_descr->ctx, RRDENG_OPCODE_FLUSHED_TO_OPEN, uv_fs_request, xt_io_descr->completion,
814 + STORAGE_PRIORITY_CRITICAL, NULL, NULL);
815
816 netdata_spinlock_unlock(&datafile->writers.spinlock);
817
@@ -896,7 +929,8 @@ static unsigned do_flush_extent(struct rrdengine_instance *ctx, struct page_desc
929 static SPINLOCK sp = NETDATA_SPINLOCK_INITIALIZER;
930 netdata_spinlock_lock(&sp);
931 if(create_new_datafile_pair(ctx) == 0)
899 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL);
932 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_FILE_INDEX, datafile, NULL, STORAGE_PRIORITY_CRITICAL, NULL,
933 + NULL);
934 netdata_spinlock_unlock(&sp);
935
936 // unlock the old datafile
@@ -1247,7 +1281,8 @@ static void extent_read_tp_worker(struct rrdengine_instance *ctx __maybe_unused,
1281 }
1282
1283 static void epdl_populate_pages_asynchronously(struct rrdengine_instance *ctx, EPDL *epdl, STORAGE_PRIORITY priority) {
1250 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_EXTENT_READ, epdl, NULL, priority);
1284 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_EXTENT_READ, epdl, NULL, priority,
1285 + rrdeng_enqueue_epdl_cmd, rrdeng_dequeue_epdl_cmd);
1286 }
1287
1288 void pdc_route_asynchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc) {
@@ -1306,11 +1341,11 @@ static void journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe
1341 }
1342
1343 static void after_do_cache_flush(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) {
1309 - rrdeng_main.flush_running = false;
1344 + rrdeng_main.flushes_running--;
1345 }
1346
1347 static void after_do_cache_evict(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) {
1313 - rrdeng_main.evict_running = false;
1348 + rrdeng_main.evictions_running--;
1349 }
1350
1351 static void after_extent_read(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) {
@@ -1319,7 +1354,7 @@ static void after_extent_read(struct rrdengine_instance *ctx __maybe_unused, voi
1354
1355 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) {
1356 ctx->worker_config.migration_to_v2_running = false;
1322 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL);
1357 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1358 }
1359
1360 struct rrdeng_buffer_sizes rrdeng_get_buffer_sizes(void) {
@@ -1350,8 +1385,8 @@ void timer_cb(uv_timer_t* handle) {
1385 worker_set_metric(RRDENG_WORKS_DISPATCHED, (NETDATA_DOUBLE)__atomic_load_n(&work_request_globals.atomics.dispatched, __ATOMIC_RELAXED));
1386 worker_set_metric(RRDENG_WORKS_EXECUTING, (NETDATA_DOUBLE)__atomic_load_n(&work_request_globals.atomics.executing, __ATOMIC_RELAXED));
1387
1353 - rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL);
1354 - rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL);
1388 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1389 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_EVICT_INIT, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
1390
1391 time_t now_s = now_monotonic_sec();
1392 if(now_s - rrdeng_main.last_buffers_cleanup_s > 600) {
@@ -1516,23 +1551,17 @@ void dbengine_event_loop(void* arg) {
1551 }
1552
1553 case RRDENG_OPCODE_FLUSH_INIT: {
1519 - if(!rrdeng_main.flush_running) {
1520 -
1521 - rrdeng_main.flush_running = true;
1522 - if(!work_dispatch(NULL, NULL, NULL, opcode, cache_flush_tp_worker, after_do_cache_flush))
1523 - rrdeng_main.flush_running = false;
1524 -
1554 + if(rrdeng_main.flushes_running < (size_t)(libuv_worker_threads / 4)) {
1555 + rrdeng_main.flushes_running++;
1556 + work_dispatch(NULL, NULL, NULL, opcode, cache_flush_tp_worker, after_do_cache_flush);
1557 }
1558 break;
1559 }
1560
1561 case RRDENG_OPCODE_EVICT_INIT: {
1530 - if(!rrdeng_main.evict_running) {
1531 -
1532 - rrdeng_main.evict_running = true;
1533 - if (!work_dispatch(NULL, NULL, NULL, opcode, cache_evict_tp_worker, after_do_cache_evict))
1534 - rrdeng_main.evict_running = false;
1535 -
1562 + if(!rrdeng_main.evictions_running) {
1563 + rrdeng_main.evictions_running++;
1564 + work_dispatch(NULL, NULL, NULL, opcode, cache_evict_tp_worker, after_do_cache_evict);
1565 }
1566 break;
1567 }
database/engine/rrdengine.h
+18 -2
@@ -27,6 +27,7 @@ extern unsigned rrdeng_pages_per_extent;
27
28 /* Forward declarations */
29 struct rrdengine_instance;
30 +struct rrdeng_cmd;
31
32 #define MAX_PAGES_PER_EXTENT (64) /* TODO: can go higher only when journal supports bigger than 4KiB transactions */
33
@@ -47,7 +48,6 @@ typedef struct page_details_control {
48
49 Pvoid_t page_list_JudyL; // the list of page details
50 unsigned completed_jobs; // the number of jobs completed last time the query thread checked
50 - bool preload_all_extent_pages; // true to preload all the pages on each extent involved in the query
51 bool workers_should_stop; // true when the query thread left and the workers should stop
52 bool prep_done;
53
@@ -127,6 +127,11 @@ struct page_details {
127 uint16_t page_length;
128 PDC_PAGE_STATUS status;
129
130 + struct {
131 + struct page_details *prev;
132 + struct page_details *next;
133 + } load;
134 +
135 struct {
136 struct page_details *prev;
137 struct page_details *next;
@@ -400,7 +405,18 @@ int init_rrd_files(struct rrdengine_instance *ctx);
405 void finalize_rrd_files(struct rrdengine_instance *ctx);
406 bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx);
407 void dbengine_event_loop(void *arg);
403 -void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, void *data, struct completion *completion, enum storage_priority priority);
408 +typedef void (*enqueue_callback_t)(struct rrdeng_cmd *cmd);
409 +typedef void (*dequeue_callback_t)(struct rrdeng_cmd *cmd);
410 +
411 +void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd);
412 +void rrdeng_dequeue_epdl_cmd(struct rrdeng_cmd *cmd);
413 +
414 +typedef struct rrdeng_cmd *(*requeue_callback_t)(void *data);
415 +void rrdeng_req_cmd(requeue_callback_t get_cmd_cb, void *data, STORAGE_PRIORITY priority);
416 +
417 +void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_opcode opcode, void *data,
418 + struct completion *completion, enum storage_priority priority,
419 + enqueue_callback_t enqueue_cb, dequeue_callback_t dequeue_cb);
420
421 void pdc_route_asynchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc);
422 void pdc_route_synchronously(struct rrdengine_instance *ctx, struct page_details_control *pdc);
database/engine/rrdengineapi.c
+2 -2
@@ -930,7 +930,7 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {
930
931 struct completion completion = {};
932 completion_init(&completion);
933 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT);
933 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);
934 completion_wait_for(&completion);
935 completion_destroy(&completion);
936
@@ -951,7 +951,7 @@ void rrdeng_prepare_exit(struct rrdengine_instance *ctx) {
951 // 1. make sure all collectors are stopped
952
953 completion_init(&ctx->quiesce_completion);
954 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_CRITICAL);
954 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_CRITICAL, NULL, NULL);
955 }
956
957 static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats)
database/engine/rrdengineapi.h
+3 -3
@@ -134,6 +134,7 @@ struct rrdeng_cache_efficiency_stats {
134 // query planner output of the queries
135 size_t pages_total;
136 size_t pages_to_load_from_disk;
137 + size_t extents_loaded_from_disk;
138
139 // pages metadata sources
140 size_t pages_meta_source_main_cache;
@@ -141,7 +142,6 @@ struct rrdeng_cache_efficiency_stats {
142 size_t pages_meta_source_journal_v2;
143
144 // preloading
144 - size_t pages_pending_found_in_cache_at_pass4;
145 size_t page_next_wait_failed;
146 size_t page_next_wait_loaded;
147 size_t page_next_nowait_failed;
@@ -149,17 +149,17 @@ struct rrdeng_cache_efficiency_stats {
149
150 // pages data sources
151 size_t pages_data_source_main_cache;
152 + size_t pages_data_source_main_cache_at_pass4;
153 size_t pages_data_source_disk;
154 size_t pages_data_source_extent_cache; // loaded by a cached extent
155
156 // cache hits at different points
156 - size_t pages_load_ok_loaded_but_cache_hit_before_allocation; // found in cache after loading, before allocating
157 size_t pages_load_ok_loaded_but_cache_hit_while_inserting; // found in cache while inserting it (conflict)
158
159 // loading
160 + size_t pages_load_extent_merged;
161 size_t pages_load_ok_uncompressed;
162 size_t pages_load_ok_compressed;
162 - size_t pages_load_ok_preloaded;
163 size_t pages_load_fail_invalid_page_in_extent;
164 size_t pages_load_fail_cant_mmap_extent;
165 size_t pages_load_fail_datafile_not_available;
web/api/formatters/json_wrapper.c
+52 -2
@@ -32,6 +32,53 @@ static int fill_formatted_callback(const char *name, const char *value, RRDLABEL
32 return 1;
33 }
34
35 +void rrdr_show_plan(RRDR *r, BUFFER *wb, const char *kq, const char *sq __maybe_unused) {
36 + QUERY_TARGET *qt = r->internal.qt;
37 +
38 + buffer_sprintf(wb, "\n\t%squery_plan%s: {", kq, kq);
39 +
40 + for(size_t m = 0; m < qt->query.used; m++) {
41 + QUERY_METRIC *qm = &qt->query.array[m];
42 +
43 + if(m)
44 + buffer_strcat(wb, ",");
45 +
46 + buffer_sprintf(wb, "\n\t\t%s%s%s: {", kq, string2str(qm->dimension.id), kq);
47 +
48 + buffer_sprintf(wb, "\n\t\t\t%splans%s: [", kq, kq);
49 + for(size_t p = 0; p < qm->plan.used ;p++) {
50 + QUERY_PLAN_ENTRY *qp = &qm->plan.array[p];
51 + if(p)
52 + buffer_strcat(wb, ",");
53 +
54 + buffer_strcat(wb, "\n\t\t\t\t{");
55 + buffer_sprintf(wb, "\n\t\t\t\t\t%stier%s: %zu,", kq, kq, qp->tier);
56 + buffer_sprintf(wb, "\n\t\t\t\t\t%safter%s: %ld,", kq, kq, qp->after);
57 + buffer_sprintf(wb, "\n\t\t\t\t\t%sbefore%s: %ld", kq, kq, qp->before);
58 + buffer_strcat(wb, "\n\t\t\t\t}");
59 + }
60 + buffer_strcat(wb, "\n\t\t\t],");
61 +
62 + buffer_sprintf(wb, "\n\t\t\t%stiers%s: [", kq, kq);
63 + for(size_t tier = 0; tier < storage_tiers ;tier++) {
64 + if(tier)
65 + buffer_strcat(wb, ",");
66 +
67 + buffer_strcat(wb, "\n\t\t\t\t{");
68 + buffer_sprintf(wb, "\n\t\t\t\t\t%stier%s: %zu,", kq, kq, tier);
69 + buffer_sprintf(wb, "\n\t\t\t\t\t%sdb_first_time%s: %ld,", kq, kq, qm->tiers[tier].db_first_time_s);
70 + buffer_sprintf(wb, "\n\t\t\t\t\t%sdb_last_time%s: %ld,", kq, kq, qm->tiers[tier].db_last_time_s);
71 + buffer_sprintf(wb, "\n\t\t\t\t\t%sweight%s: %ld", kq, kq, qm->tiers[tier].weight);
72 + buffer_strcat(wb, "\n\t\t\t\t}");
73 + }
74 + buffer_strcat(wb, "\n\t\t\t]");
75 +
76 + buffer_strcat(wb, "\n\t\t}");
77 + }
78 +
79 + buffer_strcat(wb, "\n\t},");
80 +}
81 +
82 void rrdr_json_wrapper_begin(RRDR *r, BUFFER *wb, uint32_t format, RRDR_OPTIONS options, int string_value,
83 RRDR_GROUPING group_method)
84 {
@@ -369,9 +416,12 @@ void rrdr_json_wrapper_begin(RRDR *r, BUFFER *wb, uint32_t format, RRDR_OPTIONS
416 for(size_t tier = 0; tier < storage_tiers ; tier++)
417 buffer_sprintf(wb, "%s%zu", tier>0?", ":"", r->internal.tier_points_read[tier]);
418
372 - buffer_strcat(wb, " ]");
419 + buffer_strcat(wb, " ],");
420 +
421 + if(options & RRDR_OPTION_SHOW_PLAN)
422 + rrdr_show_plan(r, wb, kq, sq);
423
374 - buffer_sprintf(wb, ",\n %sresult%s: ", kq, kq);
424 + buffer_sprintf(wb, "\n %sresult%s: ", kq, kq);
425
426 if(string_value) buffer_strcat(wb, sq);
427 //info("JSONWRAPPER(): %s: END", r->st->id);
web/api/queries/query.c
+37 -5
@@ -738,7 +738,7 @@ static long query_plan_points_coverage_weight(time_t db_first_time_s, time_t db_
738 if(points_available <= 0)
739 return -LONG_MAX;
740
741 - return points_coverage;
741 + return points_coverage + (long)(25000 * tier); // 2.5% benefit for each higher tier
742 }
743
744 static size_t query_metric_best_tier_for_timeframe(QUERY_METRIC *qm, time_t after_wanted, time_t before_wanted, size_t points_wanted) {
@@ -748,6 +748,20 @@ static size_t query_metric_best_tier_for_timeframe(QUERY_METRIC *qm, time_t afte
748 if(unlikely(after_wanted == before_wanted || points_wanted <= 0))
749 return query_metric_first_working_tier(qm);
750
751 + time_t min_first_time_s = 0;
752 + time_t max_last_time_s = 0;
753 +
754 + for(size_t tier = 0; tier < storage_tiers ; tier++) {
755 + time_t first_time_s = qm->tiers[tier].db_first_time_s;
756 + time_t last_time_s = qm->tiers[tier].db_last_time_s;
757 +
758 + if(!min_first_time_s || (first_time_s && first_time_s < min_first_time_s))
759 + min_first_time_s = first_time_s;
760 +
761 + if(!max_last_time_s || (last_time_s && last_time_s > max_last_time_s))
762 + max_last_time_s = last_time_s;
763 + }
764 +
765 for(size_t tier = 0; tier < storage_tiers ; tier++) {
766
767 // find the db time-range for this tier for all metrics
@@ -756,12 +770,20 @@ static size_t query_metric_best_tier_for_timeframe(QUERY_METRIC *qm, time_t afte
770 time_t last_time_s = qm->tiers[tier].db_last_time_s;
771 time_t update_every_s = qm->tiers[tier].db_update_every_s;
772
759 - if(!db_metric_handle || !first_time_s || !last_time_s || !update_every_s) {
773 + if( !db_metric_handle ||
774 + !first_time_s ||
775 + !last_time_s ||
776 + !update_every_s ||
777 + first_time_s > max_last_time_s ||
778 + last_time_s < min_first_time_s
779 + ) {
780 qm->tiers[tier].weight = -LONG_MAX;
781 continue;
782 }
783
764 - qm->tiers[tier].weight = query_plan_points_coverage_weight(first_time_s, last_time_s, update_every_s, after_wanted, before_wanted, points_wanted, tier);
784 + qm->tiers[tier].weight = query_plan_points_coverage_weight(
785 + min_first_time_s, max_last_time_s, update_every_s,
786 + after_wanted, before_wanted, points_wanted, tier);
787 }
788
789 size_t best_tier = 0;
@@ -2093,6 +2115,9 @@ RRDR *rrd2rrdr(ONEWAYALLOC *owa, QUERY_TARGET *qt) {
2115 if (qt->request.timeout)
2116 now_realtime_timeval(&query_start_time);
2117
2118 + size_t last_db_points_read = 0;
2119 + size_t last_result_points_generated = 0;
2120 +
2121 QUERY_ENGINE_OPS **ops = onewayalloc_callocz(r->internal.owa, qt->query.used, sizeof(QUERY_ENGINE_OPS *));
2122
2123 size_t capacity = libuv_worker_threads * 2;
@@ -2123,6 +2148,15 @@ RRDR *rrd2rrdr(ONEWAYALLOC *owa, QUERY_TARGET *qt) {
2148 rrd2rrdr_query_execute(r, c, ops[c]);
2149 }
2150
2151 + global_statistics_rrdr_query_completed(
2152 + 1,
2153 + r->internal.db_points_read - last_db_points_read,
2154 + r->internal.result_points_generated - last_result_points_generated,
2155 + qt->request.query_source);
2156 +
2157 + last_db_points_read = r->internal.db_points_read;
2158 + last_result_points_generated = r->internal.result_points_generated;
2159 +
2160 if (qt->request.timeout)
2161 now_realtime_timeval(&query_current_time);
2162
@@ -2232,7 +2266,5 @@ RRDR *rrd2rrdr(ONEWAYALLOC *owa, QUERY_TARGET *qt) {
2266 }
2267 }
2268
2235 - global_statistics_rrdr_query_completed(dimensions_used, r->internal.db_points_read,
2236 - r->internal.result_points_generated, qt->request.query_source);
2269 return r;
2270 }
web/api/queries/rrdr.h
+2 -1
@@ -40,7 +40,8 @@ typedef enum rrdr_options {
40 RRDR_OPTION_RETURN_RAW = 0x00100000, // Return raw data for aggregating across multiple nodes
41 RRDR_OPTION_RETURN_JWAR = 0x00200000, // Return anomaly rates in jsonwrap
42 RRDR_OPTION_SELECTED_TIER = 0x00400000, // Use the selected tier for the query
43 - RRDR_OPTION_ALL_DIMENSIONS = 0x00800000, // Return the full dimensions list
43 + RRDR_OPTION_ALL_DIMENSIONS = 0x00800000, // Return the full dimensions list
44 + RRDR_OPTION_SHOW_PLAN = 0x01000000, // Return the query plan in jsonwrap
45
46 // internal ones - not to be exposed to the API
47 RRDR_OPTION_INTERNAL_AR = 0x10000000, // internal use only, to let the formatters we want to render the anomaly rate
web/api/web_api_v1.c
+1
@@ -41,6 +41,7 @@ static struct {
41 , {"natural-points" , 0 , RRDR_OPTION_NATURAL_POINTS}
42 , {"virtual-points" , 0 , RRDR_OPTION_VIRTUAL_POINTS}
43 , {"all-dimensions" , 0 , RRDR_OPTION_ALL_DIMENSIONS}
44 + , {"plan" , 0 , RRDR_OPTION_SHOW_PLAN}
45 , {NULL , 0 , 0}
46 };
47