Fix flushing error threshold (#8425)

* Fix flushing error threshold to account for inactive producers of dbengine instance

* Disable page invalidation during shutdown of dbengine

* Fix crash during netdata shutdown if command server has failed to initialize

* Add fallback for uv_listen to retry with backlog = 1 on failure.

* Adhere to the API change of libuv v1.35

5 files changed, 23 insertions, 6 deletions

diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index e90ac13fe4..4534921267 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -221,7 +221,7 @@ static void pg_cache_release_pages(struct rrdengine_instance *ctx, unsigned numb
 unsigned long pg_cache_hard_limit(struct rrdengine_instance *ctx)
 {
     /* it's twice the number of producers since we pin 2 pages per producer */
-    return ctx->max_cache_pages + 2 * (unsigned long)ctx->stats.metric_API_producers;
+    return ctx->max_cache_pages + 2 * (unsigned long)ctx->metric_API_max_producers;
 }
 
 /*
@@ -231,7 +231,7 @@ unsigned long pg_cache_hard_limit(struct rrdengine_instance *ctx)
 unsigned long pg_cache_soft_limit(struct rrdengine_instance *ctx)
 {
     /* it's twice the number of producers since we pin 2 pages per producer */
-    return ctx->cache_pages_low_watermark + 2 * (unsigned long)ctx->stats.metric_API_producers;
+    return ctx->cache_pages_low_watermark + 2 * (unsigned long)ctx->metric_API_max_producers;
 }
 
 /*
@@ -240,8 +240,8 @@ unsigned long pg_cache_soft_limit(struct rrdengine_instance *ctx)
  */
 unsigned long pg_cache_committed_hard_limit(struct rrdengine_instance *ctx)
 {
-    /* We remove the active pages of the producers from the calculation and only allow 50% of the extra pinned pages */
-    return ctx->cache_pages_low_watermark + (unsigned long)ctx->stats.metric_API_producers / 2;
+    /* We remove the active pages of the producers from the calculation and only allow the extra pinned pages */
+    return ctx->cache_pages_low_watermark + (unsigned long)ctx->metric_API_max_producers;
 }
 
 /*
diff --git a/database/engine/rrdengine.c b/database/engine/rrdengine.c
index 586f2e0b48..b15239e6c4 100644
--- a/database/engine/rrdengine.c
+++ b/database/engine/rrdengine.c
@@ -814,7 +814,7 @@ void timer_cb(uv_timer_t* handle)
         uv_rwlock_rdlock(&pg_cache->committed_page_index.lock);
         nr_committed_pages = pg_cache->committed_page_index.nr_committed_pages;
         uv_rwlock_rdunlock(&pg_cache->committed_page_index.lock);
-        producers = ctx->stats.metric_API_producers;
+        producers = ctx->metric_API_max_producers;
         /* are flushable pages more than 25% of the maximum page cache size */
         high_watermark = (ctx->max_cache_pages * 25LLU) / 100;
         low_watermark = (ctx->max_cache_pages * 5LLU) / 100; /* 5%, must be smaller than high_watermark */
@@ -920,6 +920,7 @@ void rrdeng_worker(void* arg)
                 break;
             case RRDENG_SHUTDOWN:
                 shutdown = 1;
+                ctx->drop_metrics_under_page_cache_pressure = 0;
                 break;
             case RRDENG_READ_PAGE:
                 do_read_extent(wc, &cmd.read_page.page_cache_descr, 1, 0);
diff --git a/database/engine/rrdengine.h b/database/engine/rrdengine.h
index 22c3d33bef..d97a34aa30 100644
--- a/database/engine/rrdengine.h
+++ b/database/engine/rrdengine.h
@@ -183,6 +183,7 @@ struct rrdengine_instance {
     unsigned last_fileno; /* newest index of datafile and journalfile */
     unsigned long max_cache_pages;
     unsigned long cache_pages_low_watermark;
+    unsigned long metric_API_max_producers;
 
     struct rrdengine_statistics stats;
 };
diff --git a/database/engine/rrdengineapi.c b/database/engine/rrdengineapi.c
index bd6110ce38..f72569c5b3 100755
--- a/database/engine/rrdengineapi.c
+++ b/database/engine/rrdengineapi.c
@@ -189,9 +189,21 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n
     if (perfect_page_alignment)
         rd->rrdset->rrddim_page_alignment = descr->page_length;
     if (unlikely(INVALID_TIME == descr->start_time)) {
+        unsigned long new_metric_API_producers, old_metric_API_max_producers, ret_metric_API_max_producers;
         descr->start_time = point_in_time;
 
-        rrd_stat_atomic_add(&ctx->stats.metric_API_producers, 1);
+        new_metric_API_producers = rrd_atomic_add_fetch(&ctx->stats.metric_API_producers, 1);
+        while (unlikely(new_metric_API_producers > (old_metric_API_max_producers = ctx->metric_API_max_producers))) {
+            /* Increase ctx->metric_API_max_producers */
+            ret_metric_API_max_producers = ulong_compare_and_swap(&ctx->metric_API_max_producers,
+                                                                  old_metric_API_max_producers,
+                                                                  new_metric_API_producers);
+            if (old_metric_API_max_producers == ret_metric_API_max_producers) {
+                /* success */
+                break;
+            }
+        }
+
         pg_cache_insert(ctx, handle->page_index, descr);
     } else {
         pg_cache_add_new_metric_time(handle->page_index, descr);
@@ -791,6 +803,7 @@ int rrdeng_init(struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned p
     strncpyz(ctx->dbfiles_path, dbfiles_path, sizeof(ctx->dbfiles_path) - 1);
     ctx->dbfiles_path[sizeof(ctx->dbfiles_path) - 1] = '\0';
     ctx->drop_metrics_under_page_cache_pressure = rrdeng_drop_metrics_under_page_cache_pressure;
+    ctx->metric_API_max_producers = 0;
 
     memset(&ctx->worker_config, 0, sizeof(ctx->worker_config));
     ctx->worker_config.ctx = ctx;
diff --git a/database/engine/rrdenginelib.h b/database/engine/rrdenginelib.h
index 85836b97fe..91836b63b9 100644
--- a/database/engine/rrdenginelib.h
+++ b/database/engine/rrdenginelib.h
@@ -28,8 +28,10 @@ typedef uintptr_t rrdeng_stats_t;
 
 #ifdef __ATOMIC_RELAXED
 #define rrd_atomic_fetch_add(p, n) __atomic_fetch_add(p, n, __ATOMIC_RELAXED)
+#define rrd_atomic_add_fetch(p, n) __atomic_add_fetch(p, n, __ATOMIC_RELAXED)
 #else
 #define rrd_atomic_fetch_add(p, n) __sync_fetch_and_add(p, n)
+#define rrd_atomic_add_fetch(p, n) __sync_add_and_fetch(p, n)
 #endif
 
 #define rrd_stat_atomic_add(p, n) rrd_atomic_fetch_add(p, n)