diff options
Diffstat (limited to 'database')
-rw-r--r-- | database/engine/cache.c | 85 | ||||
-rw-r--r-- | database/engine/cache.h | 4 | ||||
-rw-r--r-- | database/engine/datafile.c | 4 | ||||
-rw-r--r-- | database/engine/journalfile.c | 4 | ||||
-rw-r--r-- | database/engine/metric.c | 265 | ||||
-rw-r--r-- | database/engine/metric.h | 22 | ||||
-rw-r--r-- | database/engine/pagecache.c | 10 | ||||
-rw-r--r-- | database/engine/pagecache.h | 2 | ||||
-rw-r--r-- | database/engine/pdc.c | 77 | ||||
-rw-r--r-- | database/engine/rrdengine.c | 91 | ||||
-rwxr-xr-x | database/engine/rrdengineapi.c | 8 | ||||
-rw-r--r-- | database/rrd.h | 4 | ||||
-rw-r--r-- | database/rrdcalc.c | 4 | ||||
-rw-r--r-- | database/rrdcontext.h | 2 | ||||
-rw-r--r-- | database/rrdhost.c | 6 | ||||
-rw-r--r-- | database/rrdset.c | 2 |
16 files changed, 421 insertions, 169 deletions
diff --git a/database/engine/cache.c b/database/engine/cache.c index 92bf33c429..9292bc5354 100644 --- a/database/engine/cache.c +++ b/database/engine/cache.c @@ -16,7 +16,7 @@ typedef int32_t REFCOUNT; #define REFCOUNT_DELETING (-100) -// to use arrayalloc uncomment the following line: +// to use ARAL uncomment the following line: #define PGC_WITH_ARAL 1 typedef enum __attribute__ ((__packed__)) { @@ -82,6 +82,8 @@ struct pgc_linked_list { struct pgc { struct { + char name[PGC_NAME_MAX + 1]; + size_t partitions; size_t clean_size; size_t max_dirty_pages_per_call; @@ -415,13 +417,25 @@ struct section_pages { PGC_PAGE *base; }; -static ARAL section_pages_aral = { - .filename = NULL, - .cache_dir = NULL, - .use_mmap = false, - .initial_elements = 16384 / sizeof(struct section_pages), - .requested_element_size = sizeof(struct section_pages), -}; +static ARAL *pgc_section_pages_aral = NULL; +static void pgc_section_pages_static_aral_init(void) { + static SPINLOCK spinlock = NETDATA_SPINLOCK_INITIALIZER; + + if(unlikely(!pgc_section_pages_aral)) { + netdata_spinlock_lock(&spinlock); + + // we have to check again + if(!pgc_section_pages_aral) + pgc_section_pages_aral = aral_create( + "pgc_section", + sizeof(struct section_pages), + 0, + 4096, + NULL, NULL, false, false); + + netdata_spinlock_unlock(&spinlock); + } +} static inline void pgc_stats_ll_judy_change(PGC *cache, struct pgc_linked_list *ll, size_t mem_before_judyl, size_t mem_after_judyl) { if(mem_after_judyl > mem_before_judyl) { @@ -462,7 +476,7 @@ static void pgc_ll_add(PGC *cache __maybe_unused, struct pgc_linked_list *ll, PG struct section_pages *sp = *section_pages_pptr; if(!sp) { // sp = callocz(1, sizeof(struct section_pages)); - sp = arrayalloc_mallocz(§ion_pages_aral); + sp = aral_mallocz(pgc_section_pages_aral); memset(sp, 0, sizeof(struct section_pages)); *section_pages_pptr = sp; @@ -473,7 +487,7 @@ static void pgc_ll_add(PGC *cache __maybe_unused, struct pgc_linked_list *ll, PG sp->entries++; sp->size += page->assumed_size; - DOUBLE_LINKED_LIST_APPEND_UNSAFE(sp->base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(sp->base, page, link.prev, link.next); if((sp->entries % cache->config.max_dirty_pages_per_call) == 0) ll->version++; @@ -484,11 +498,11 @@ static void pgc_ll_add(PGC *cache __maybe_unused, struct pgc_linked_list *ll, PG // - DIRTY pages made CLEAN, depending on their accesses may be appended (accesses > 0) or prepended (accesses = 0). if(page->accesses || page_flag_check(page, PGC_PAGE_HAS_BEEN_ACCESSED | PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES) == PGC_PAGE_HAS_BEEN_ACCESSED) { - DOUBLE_LINKED_LIST_APPEND_UNSAFE(ll->base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(ll->base, page, link.prev, link.next); page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED); } else - DOUBLE_LINKED_LIST_PREPEND_UNSAFE(ll->base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_PREPEND_ITEM_UNSAFE(ll->base, page, link.prev, link.next); ll->version++; } @@ -530,7 +544,7 @@ static void pgc_ll_del(PGC *cache __maybe_unused, struct pgc_linked_list *ll, PG struct section_pages *sp = *section_pages_pptr; sp->entries--; sp->size -= page->assumed_size; - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(sp->base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(sp->base, page, link.prev, link.next); if(!sp->base) { size_t mem_before_judyl, mem_after_judyl; @@ -543,13 +557,13 @@ static void pgc_ll_del(PGC *cache __maybe_unused, struct pgc_linked_list *ll, PG fatal("DBENGINE CACHE: cannot delete section from Judy LL"); // freez(sp); - arrayalloc_freez(§ion_pages_aral, sp); + aral_freez(pgc_section_pages_aral, sp); mem_after_judyl -= sizeof(struct section_pages); pgc_stats_ll_judy_change(cache, ll, mem_before_judyl, mem_after_judyl); } } else { - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(ll->base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(ll->base, page, link.prev, link.next); ll->version++; } @@ -565,8 +579,8 @@ static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) { if (flags & PGC_PAGE_CLEAN) { if(pgc_ll_trylock(cache, &cache->clean)) { - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(cache->clean.base, page, link.prev, link.next); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); pgc_ll_unlock(cache, &cache->clean); page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED); } @@ -860,7 +874,7 @@ static inline void free_this_page(PGC *cache, PGC_PAGE *page) { // free our memory #ifdef PGC_WITH_ARAL - arrayalloc_freez(cache->aral, page); + aral_freez(cache->aral, page); #else freez(page); #endif @@ -1038,8 +1052,8 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic break; if(unlikely(page_flag_check(page, PGC_PAGE_HAS_BEEN_ACCESSED | PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES) == PGC_PAGE_HAS_BEEN_ACCESSED)) { - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(cache->clean.base, page, link.prev, link.next); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED); continue; } @@ -1056,7 +1070,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic __atomic_add_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED); __atomic_add_fetch(&cache->stats.evicting_size, page->assumed_size, __ATOMIC_RELAXED); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(pages_to_evict, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(pages_to_evict, page, link.prev, link.next); pages_to_evict_size += page->assumed_size; @@ -1073,8 +1087,8 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic if(!first_page_we_relocated) first_page_we_relocated = page; - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(cache->clean.base, page, link.prev, link.next); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next); // check if we have to stop if(unlikely(++total_pages_skipped >= max_skip && !all_of_them)) { @@ -1099,8 +1113,8 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic next = page->link.next; size_t partition = pgc_indexing_partition(cache, page->metric_id); - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(pages_to_evict, page, link.prev, link.next); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(pages_per_partition[partition], page, link.prev, link.next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(pages_to_evict, page, link.prev, link.next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(pages_per_partition[partition], page, link.prev, link.next); } // remove them from the index @@ -1178,7 +1192,7 @@ static PGC_PAGE *page_add(PGC *cache, PGC_ENTRY *entry, bool *added) { __atomic_add_fetch(&cache->stats.workers_add, 1, __ATOMIC_RELAXED); #ifdef PGC_WITH_ARAL - PGC_PAGE *allocation = arrayalloc_mallocz(cache->aral); + PGC_PAGE *allocation = aral_mallocz(cache->aral); #endif PGC_PAGE *page; size_t spins = 0; @@ -1285,7 +1299,7 @@ static PGC_PAGE *page_add(PGC *cache, PGC_ENTRY *entry, bool *added) { #ifdef PGC_WITH_ARAL if(allocation) - arrayalloc_freez(cache->aral, allocation); + aral_freez(cache->aral, allocation); #endif __atomic_sub_fetch(&cache->stats.workers_add, 1, __ATOMIC_RELAXED); @@ -1713,7 +1727,8 @@ void free_all_unreferenced_clean_pages(PGC *cache) { // ---------------------------------------------------------------------------- // public API -PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb, +PGC *pgc_create(const char *name, + size_t clean_size_bytes, free_clean_page_callback pgc_free_cb, size_t max_dirty_pages_per_flush, save_dirty_init_callback pgc_save_init_cb, save_dirty_page_callback pgc_save_dirty_cb, @@ -1732,6 +1747,7 @@ PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb, max_flushes_inline = 2; PGC *cache = callocz(1, sizeof(PGC)); + strncpyz(cache->config.name, name, PGC_NAME_MAX); cache->config.options = options; cache->config.clean_size = (clean_size_bytes < 1 * 1024 * 1024) ? 1 * 1024 * 1024 : clean_size_bytes; cache->config.pgc_free_clean_cb = pgc_free_cb; @@ -1772,10 +1788,14 @@ PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb, cache->clean.stats = &cache->stats.queues.clean; #ifdef PGC_WITH_ARAL - cache->aral = arrayalloc_create(sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page, 65536 / sizeof(PGC_PAGE), - NULL, NULL, false, false); + cache->aral = aral_create(name, + sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page, + 0, + 4096, + NULL, NULL, false, false); #endif + pgc_section_pages_static_aral_init(); pointer_index_init(cache); return cache; @@ -1803,7 +1823,7 @@ void pgc_destroy(PGC *cache) { else { pointer_destroy_index(cache); #ifdef PGC_WITH_ARAL - arrayalloc_destroy(cache->aral); + aral_destroy(cache->aral); #endif freez(cache); } @@ -2602,7 +2622,8 @@ void unittest_stress_test(void) { #endif int pgc_unittest(void) { - PGC *cache = pgc_create(32 * 1024 * 1024, unittest_free_clean_page_callback, + PGC *cache = pgc_create("test", + 32 * 1024 * 1024, unittest_free_clean_page_callback, 64, NULL, unittest_save_dirty_page_callback, 10, 10, 1000, 10, PGC_OPTIONS_DEFAULT, 1, 11); diff --git a/database/engine/cache.h b/database/engine/cache.h index 9a0bfbfe91..740e94c065 100644 --- a/database/engine/cache.h +++ b/database/engine/cache.h @@ -8,6 +8,7 @@ typedef struct pgc PGC; typedef struct pgc_page PGC_PAGE; +#define PGC_NAME_MAX 23 typedef enum __attribute__ ((__packed__)) { PGC_OPTIONS_NONE = 0, @@ -165,7 +166,8 @@ typedef void (*free_clean_page_callback)(PGC *cache, PGC_ENTRY entry); typedef void (*save_dirty_page_callback)(PGC *cache, PGC_ENTRY *entries_array, PGC_PAGE **pages_array, size_t entries); typedef void (*save_dirty_init_callback)(PGC *cache, Word_t section); // create a cache -PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_clean_cb, +PGC *pgc_create(const char *name, + size_t clean_size_bytes, free_clean_page_callback pgc_free_clean_cb, size_t max_dirty_pages_per_flush, save_dirty_init_callback pgc_save_init_cb, save_dirty_page_callback pgc_save_dirty_cb, size_t max_pages_per_inline_eviction, size_t max_inline_evictors, size_t max_skip_pages_per_inline_eviction, diff --git a/database/engine/datafile.c b/database/engine/datafile.c index 61d4b34b1a..cc0ed1354d 100644 --- a/database/engine/datafile.c +++ b/database/engine/datafile.c @@ -4,13 +4,13 @@ void datafile_list_insert(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) { uv_rwlock_wrlock(&ctx->datafiles.rwlock); - DOUBLE_LINKED_LIST_APPEND_UNSAFE(ctx->datafiles.first, datafile, prev, next); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(ctx->datafiles.first, datafile, prev, next); uv_rwlock_wrunlock(&ctx->datafiles.rwlock); } void datafile_list_delete_unsafe(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) { - DOUBLE_LINKED_LIST_REMOVE_UNSAFE(ctx->datafiles.first, datafile, prev, next); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(ctx->datafiles.first, datafile, prev, next); } diff --git a/database/engine/journalfile.c b/database/engine/journalfile.c index 8b6472a775..3f9c9e6fb2 100644 --- a/database/engine/journalfile.c +++ b/database/engine/journalfile.c @@ -605,7 +605,9 @@ static void journalfile_restore_extent_metadata(struct rrdengine_instance *ctx, false); if(!vd.is_valid) { - mrg_metric_release(main_mrg, metric); + if(metric) + mrg_metric_release(main_mrg, metric); + continue; } diff --git a/database/engine/metric.c b/database/engine/metric.c index d16bc063d9..bad15b50c2 100644 --- a/database/engine/metric.c +++ b/database/engine/metric.c @@ -3,6 +3,10 @@ typedef int32_t REFCOUNT; #define REFCOUNT_DELETING (-100) +typedef enum __attribute__ ((__packed__)) { + METRIC_FLAG_HAS_RETENTION = (1 << 0), +} METRIC_FLAGS; + struct metric { uuid_t uuid; // never changes Word_t section; // never changes @@ -12,6 +16,8 @@ struct metric { time_t latest_time_s_hot; // latest time of the currently collected page uint32_t latest_update_every_s; // pid_t writer; + METRIC_FLAGS flags; + REFCOUNT refcount; SPINLOCK spinlock; // protects all variable members // THIS IS allocated with malloc() @@ -19,8 +25,9 @@ struct metric { }; struct mrg { + ARAL *aral[MRG_PARTITIONS]; + struct pgc_index { - ARAL *aral; netdata_rwlock_t rwlock; Pvoid_t uuid_judy; // each UUID has a JudyL of sections (tiers) } index[MRG_PARTITIONS]; @@ -95,9 +102,75 @@ static inline size_t uuid_partition(MRG *mrg __maybe_unused, uuid_t *uuid) { return u[UUID_SZ - 1] % MRG_PARTITIONS; } -static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) { +static inline bool metric_has_retention_unsafe(MRG *mrg __maybe_unused, METRIC *metric) { + bool has_retention = (metric->first_time_s || metric->latest_time_s_clean || metric->latest_time_s_hot); + + if(has_retention && !(metric->flags & METRIC_FLAG_HAS_RETENTION)) { + metric->flags |= METRIC_FLAG_HAS_RETENTION; + __atomic_add_fetch(&mrg->stats.entries_with_retention, 1, __ATOMIC_RELAXED); + } + else if(!has_retention && (metric->flags & METRIC_FLAG_HAS_RETENTION)) { + metric->flags &= ~METRIC_FLAG_HAS_RETENTION; + __atomic_sub_fetch(&mrg->stats.entries_with_retention, 1, __ATOMIC_RELAXED); + } + + return has_retention; +} + +static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric, bool having_spinlock) { + REFCOUNT refcount; + + if(!having_spinlock) + netdata_spinlock_lock(&metric->spinlock); + + if(unlikely(metric->refcount < 0)) + fatal("METRIC: refcount is %d (negative) during acquire", metric->refcount); + + refcount = ++metric->refcount; + + // update its retention flags + metric_has_retention_unsafe(mrg, metric); + + if(!having_spinlock) + netdata_spinlock_unlock(&metric->spinlock); + + if(refcount == 1) + __atomic_add_fetch(&mrg->stats.entries_referenced, 1, __ATOMIC_RELAXED); + + __atomic_add_fetch(&mrg->stats.current_references, 1, __ATOMIC_RELAXED); + + return refcount; +} + +static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, METRIC *metric) { + bool ret = true; + REFCOUNT refcount; + + netdata_spinlock_lock(&metric->spinlock); + + if(unlikely(metric->refcount <= 0)) + fatal("METRIC: refcount is %d (zero or negative) during release", metric->refcount); + + refcount = --metric->refcount; + + if(likely(metric_has_retention_unsafe(mrg, metric) || refcount != 0)) + ret = false; + + netdata_spinlock_unlock(&metric->spinlock); + + if(unlikely(!refcount)) + __atomic_sub_fetch(&mrg->stats.entries_referenced, 1, __ATOMIC_RELAXED); + + __atomic_sub_fetch(&mrg->stats.current_references, 1, __ATOMIC_RELAXED); + + return ret; +} + +static METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) { size_t partition = uuid_partition(mrg, &entry->uuid); + METRIC *allocation = aral_mallocz(mrg->aral[partition]); + mrg_index_write_lock(mrg, partition); size_t mem_before_judyl, mem_after_judyl; @@ -117,18 +190,22 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) { if(unlikely(!PValue || PValue == PJERR)) fatal("DBENGINE METRIC: corrupted section JudyL array"); - if(*PValue != NULL) { + if(unlikely(*PValue != NULL)) { METRIC *metric = *PValue; + + metric_acquire(mrg, metric, false); mrg_index_write_unlock(mrg, partition); if(ret) *ret = false; + aral_freez(mrg->aral[partition], allocation); + MRG_STATS_DUPLICATE_ADD(mrg); return metric; } - METRIC *metric = arrayalloc_mallocz(mrg->index[partition].aral); + METRIC *metric = allocation; uuid_copy(metric->uuid, entry->uuid); metric->section = entry->section; metric->first_time_s = entry->first_time_s; @@ -136,7 +213,10 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) { metric->latest_time_s_hot = 0; metric->latest_update_every_s = entry->latest_update_every_s; metric->writer = 0; + metric->refcount = 0; + metric->flags = 0; netdata_spinlock_init(&metric->spinlock); + metric_acquire(mrg, metric, true); // no spinlock use required here *PValue = metric; mrg_index_write_unlock(mrg, partition); @@ -149,7 +229,7 @@ static METRIC *metric_add(MRG *mrg, MRG_ENTRY *entry, bool *ret) { return metric; } -static METRIC *metric_get(MRG *mrg, uuid_t *uuid, Word_t section) { +static METRIC *metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) { size_t partition = uuid_partition(mrg, uuid); mrg_index_read_lock(mrg, partition); @@ -170,19 +250,27 @@ static METRIC *metric_get(MRG *mrg, uuid_t *uuid, Word_t section) { METRIC *metric = *PValue; + metric_acquire(mrg, metric, false); + mrg_index_read_unlock(mrg, partition); MRG_STATS_SEARCH_HIT(mrg); return metric; } -static bool metric_del(MRG *mrg, METRIC *metric) { +static bool acquired_metric_del(MRG *mrg, METRIC *metric) { size_t partition = uuid_partition(mrg, &metric->uuid); size_t mem_before_judyl, mem_after_judyl; mrg_index_write_lock(mrg, partition); + if(!metric_release_and_can_be_deleted(mrg, metric)) { + mrg_index_write_unlock(mrg, partition); + __atomic_add_fetch(&mrg->stats.delete_having_retention_or_referenced, 1, __ATOMIC_RELAXED); + return false; + } + Pvoid_t *sections_judy_pptr = JudyHSGet(mrg->index[partition].uuid_judy, &metric->uuid, sizeof(uuid_t)); if(unlikely(!sections_judy_pptr || !*sections_judy_pptr)) { mrg_index_write_unlock(mrg, partition); @@ -208,11 +296,10 @@ static bool metric_del(MRG *mrg, METRIC *metric) { mrg_stats_size_judyhs_removed_uuid(mrg); } - // arrayalloc is running lockless here - arrayalloc_freez(mrg->index[partition].aral, metric); - mrg_index_write_unlock(mrg, partition); + aral_freez(mrg->aral[partition], metric); + MRG_STATS_DELETED_METRIC(mrg, partition); return true; @@ -223,11 +310,22 @@ static bool metric_del(MRG *mrg, METRIC *metric) { MRG *mrg_create(void) { MRG *mrg = callocz(1, sizeof(MRG)); + for(size_t i = 0; i < MRG_PARTITIONS ; i++) { + char buf[ARAL_MAX_NAME + 1]; + snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i); netdata_rwlock_init(&mrg->index[i].rwlock); - mrg->index[i].aral = arrayalloc_create(sizeof(METRIC), 32768 / sizeof(METRIC), NULL, NULL, false, true); + + mrg->aral[i] = aral_create("mrg", + sizeof(METRIC), + 0, + 512, + NULL, NULL, false, + false); } + mrg->stats.size = sizeof(MRG); + return mrg; } @@ -242,32 +340,27 @@ void mrg_destroy(MRG *mrg __maybe_unused) { } METRIC *mrg_metric_add_and_acquire(MRG *mrg, MRG_ENTRY entry, bool *ret) { - // FIXME - support refcount - // internal_fatal(entry.latest_time_s > max_acceptable_collected_time(), // "DBENGINE METRIC: metric latest time is in the future"); - return metric_add(mrg, &entry, ret); + return metric_add_and_acquire(mrg, &entry, ret); } METRIC *mrg_metric_get_and_acquire(MRG *mrg, uuid_t *uuid, Word_t section) { - // FIXME - support refcount - return metric_get(mrg, uuid, section); + return metric_get_and_acquire(mrg, uuid, section); } bool mrg_metric_release_and_delete(MRG *mrg, METRIC *metric) { - // FIXME - support refcount - return metric_del(mrg, metric); + return acquired_metric_del(mrg, metric); } -METRIC *mrg_metric_dup(MRG *mrg __maybe_unused, METRIC *metric) { - // FIXME - duplicate refcount +METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric) { + metric_acquire(mrg, metric, false); return metric; } -void mrg_metric_release(MRG *mrg __maybe_unused, METRIC *metric __maybe_unused) { - // FIXME - release refcount - +bool mrg_metric_release(MRG *mrg, METRIC *metric) { + return metric_release_and_can_be_deleted(mrg, metric); } Word_t mrg_metric_id(MRG *mrg __maybe_unused, METRIC *metric) { @@ -285,6 +378,7 @@ Word_t mrg_metric_section(MRG *mrg __maybe_unused, METRIC *metric) { bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) { netdata_spinlock_lock(&metric->spinlock); metric->first_time_s = first_time_s; + metric_has_retention_unsafe(mrg, metric); netdata_spinlock_unlock(&metric->spinlock); return true; @@ -311,6 +405,7 @@ void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t else if(unlikely(!metric->latest_update_every_s && update_every_s)) metric->latest_update_every_s = update_every_s; + metric_has_retention_unsafe(mrg, metric); netdata_spinlock_unlock(&metric->spinlock); } @@ -322,6 +417,7 @@ bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metr metric->first_time_s = first_time_s; ret = true; } + metric_has_retention_unsafe(mrg, metric); netdata_spinlock_unlock(&metric->spinlock); return ret; @@ -382,10 +478,63 @@ bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, // if(unlikely(metric->first_time_s > latest_time_s)) // metric->first_time_s = latest_time_s; + metric_has_retention_unsafe(mrg, metric); netdata_spinlock_unlock(&metric->spinlock); return true; } +// returns true when metric still has retention +bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metric) { + Word_t section = mrg_metric_section(mrg, metric); + bool do_again = false; + size_t countdown = 5; + bool ret = true; + + do { + time_t min_first_time_s = LONG_MAX; + time_t max_end_time_s = 0; + PGC_PAGE *page; + PGC_SEARCH method = PGC_SEARCH_FIRST; + time_t page_first_time_s = 0; + time_t page_end_time_s = 0; + while ((page = pgc_page_get_and_acquire(main_cache, section, (Word_t)metric, page_first_time_s, method))) { + method = PGC_SEARCH_NEXT; + + bool is_hot = pgc_is_page_hot(page); + bool is_dirty = pgc_is_page_dirty(page); + page_first_time_s = pgc_page_start_time_s(page); + page_end_time_s = pgc_page_end_time_s(page); + + if ((is_hot || is_dirty) && page_first_time_s < min_first_time_s) + min_first_time_s = page_first_time_s; + + if (is_dirty && page_end_time_s > max_end_time_s) + max_end_time_s = page_end_time_s; + + pgc_page_release(main_cache, page); + } + + if (min_first_time_s == LONG_MAX) + min_first_time_s = 0; + + netdata_spinlock_lock(&metric->spinlock); + if (--countdown && !min_first_time_s && metric->latest_time_s_hot) + do_again = true; + else { + internal_error(!countdown, "METRIC: giving up on updating the retention of metric without disk retention"); + + do_again = false; + metric->first_time_s = min_first_time_s; + metric->latest_time_s_clean = max_end_time_s; + + ret = metric_has_retention_unsafe(mrg, metric); + } + netdata_spinlock_unlock(&metric->spinlock); + } while(do_again); + + return ret; +} + bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) { // internal_fatal(latest_time_s > max_acceptable_collected_time(), // "DBENGINE METRIC: metric latest time is in the future"); @@ -399,6 +548,7 @@ bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, t // if(unlikely(metric->first_time_s > latest_time_s)) // metric->first_time_s = latest_time_s; + metric_has_retention_unsafe(mrg, metric); netdata_spinlock_unlock(&metric->spinlock); return true; } @@ -444,7 +594,7 @@ time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) { return update_every_s; } -bool mrg_metric_writer_acquire(MRG *mrg, METRIC *metric) { +bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) { bool done = false; netdata_spinlock_lock(&metric->spinlock); if(!metric->writer) { @@ -452,11 +602,13 @@ bool mrg_metric_writer_acquire(MRG *mrg, METRIC *metric) { __atomic_add_fetch(&mrg->stats.writers, 1, __ATOMIC_RELAXED); done = true; } + else + __atomic_add_fetch(&mrg->stats.writers_conflicts, 1, __ATOMIC_RELAXED); netdata_spinlock_unlock(&metric->spinlock); return done; } -bool mrg_metric_writer_release(MRG *mrg, METRIC *metric) { +bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) { bool done = false; netdata_spinlock_lock(&metric->spinlock); if(metric->writer) { @@ -584,73 +736,80 @@ static void *mrg_stress_test_thread3(void *ptr) { int mrg_unittest(void) { MRG *mrg = mrg_create(); - METRIC *metric1, *metric2; + METRIC *m1_t0, *m2_t0, *m3_t0, *m4_t0; + METRIC *m1_t1, *m2_t1, *m3_t1, *m4_t1; bool ret; MRG_ENTRY entry = { - .section = 1, + .section = 0, .first_time_s = 2, .last_time_s = 3, .latest_update_every_s = 4, }; uuid_generate(entry.uuid); - metric1 = mrg_metric_add_and_acquire(mrg, entry, &ret); + m1_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret); if(!ret) fatal("DBENGINE METRIC: failed to add metric"); // add the same metric again - if(mrg_metric_add_and_acquire(mrg, entry, &ret) != metric1) + m2_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret); + if(m2_t0 != m1_t0) fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer"); if(ret) fatal("DBENGINE METRIC: managed to add the same metric twice"); - if(mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section) != metric1) + m3_t0 = mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section); + if(m3_t0 != m1_t0) fatal("DBENGINE METRIC: cannot find the metric added"); // add the same metric again - if(mrg_metric_add_and_acquire(mrg, entry, &ret) != metric1) + m4_t0 = mrg_metric_add_and_acquire(mrg, entry, &ret); + if(m4_t0 != m1_t0) fatal("DBENGINE METRIC: adding the same metric twice, does not return the same pointer"); if(ret) fatal("DBENGINE METRIC: managed to add the same metric twice"); // add the same metric in another section - entry.section = 0; - metric2 = mrg_metric_add_and_acquire(mrg, entry, &ret); + entry.section = 1; + m1_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret); if(!ret) - fatal("DBENGINE METRIC: failed to add metric in different section"); + fatal("DBENGINE METRIC: failed to add metric in section %zu", (size_t)entry.section); // add the same metric again - if(mrg_metric_add_and_acquire(mrg, entry, &ret) != metric2) - fatal("DBENGINE METRIC: adding the same metric twice (section 0), does not return the same pointer"); + m2_t1 = mrg_metric_add_and_acquire(mrg, entry, &ret); + if(m2_t1 != m1_t1) + fatal("DBENGINE METRIC: adding the same metric twice (section %zu), does not return the same pointer", (size_t)entry.section); if(ret) fatal("DBENGINE METRIC: managed to add the same metric twice in (section 0)"); - if(mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section) != metric2) - fatal("DBENGINE METRIC: cannot find the metric added (section 0)"); + m3_t1 = mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section); + if(m3_t1 != m1_t1) + fatal("DBENGINE METRIC: cannot find the metric added (section %zu)", (size_t)entry.section); // delete the first metric - if(!mrg_metric_release_and_delete(mrg, metric1)) + mrg_metric_release(mrg, m2_t0); + mrg_metric_release(mrg, m3_t0); + mrg_metric_release(mrg, m4_t0); + mrg_metric_set_first_time_s(mrg, m1_t0, 0); + mrg_metric_set_clean_latest_time_s(mrg, m1_t0, 0); + mrg_metric_set_hot_latest_time_s(mrg, m1_t0, 0); + if(!mrg_metric_release_and_delete(mrg, m1_t0)) fatal("DBENGINE METRIC: cannot delete the first metric"); - if(mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section) != metric2) - fatal("DBENGINE METRIC: cannot find the metric added (section 0), after deleting the first one"); - - // delete the first metric again - metric1 pointer is invalid now - if(mrg_metric_release_and_delete(mrg, metric1)) - fatal("DBENGINE METRIC: deleted again an already deleted metric"); - - // find the section 0 metric again - if(mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section) != metric2) - fatal("DBENGINE METRIC: cannot find the metric added (section 0), after deleting the first one twice"); + m4_t1 = mrg_metric_get_and_acquire(mrg, &entry.uuid, entry.section); + if(m4_t1 != m1_t1) + fatal("DBENGINE METRIC: cannot find the metric added (section %zu), after deleting the first one", (size_t)entry.section); // delete the second metric - if(!mrg_metric_release_and_delete(mrg, metric2)) + mrg_metric_release(mrg, m2_t1); + mrg_metric_release(mrg, m3_t1); + mrg_metric_release(mrg, m4_t1); + mrg_metric_set_first_time_s(mrg, m1_t1, 0); + mrg_metric_set_clean_latest_time_s(mrg, m1_t1, 0); + mrg_metric_set_hot_latest_time_s(mrg, m1_t1, 0); + if(!mrg_metric_release_and_delete(mrg, m1_t1)) fatal("DBENGINE METRIC: cannot delete the second metric"); - // delete the second metric again - if(mrg |