@cryptotaxi247 / netdata-1 / commits / 04e674fcb

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

Markos Fountoulakis committed Mar 23, 2020 at 18:50 UTC 04e674fcba1d1fab1647203734f918f2cfc3904e

6 files changed +41 -8

daemon/commands.c

+18 -2

@@ -15,6 +15,7 @@ char cmd_prefix_by_status[] = {

15 CMD_PREFIX_ERROR

16 };

18 +static int command_server_initialized = 0;

19 static int command_thread_error;

20 static int command_thread_shutdown;

21 static unsigned clients = 0;

@@ -496,7 +497,7 @@ static void command_thread(void *arg)

497 }

498 async.data = NULL;

499

499 - ret = uv_pipe_init(loop, &server_pipe, 1);

500 + ret = uv_pipe_init(loop, &server_pipe, 0);

501 if (ret) {

502 error("uv_pipe_init(): %s", uv_strerror(ret));

503 command_thread_error = ret;

@@ -510,7 +511,13 @@ static void command_thread(void *arg)

511 command_thread_error = ret;

512 goto error_after_pipe_bind;

513 }

513 - if ((ret = uv_listen((uv_stream_t *)&server_pipe, SOMAXCONN, connection_cb))) {

514 + ret = uv_listen((uv_stream_t *)&server_pipe, SOMAXCONN, connection_cb);

515 + if (ret) {

516 + /* Fallback to backlog of 1 */

517 + info("uv_listen() failed with backlog = %d, falling back to backlog = 1.", SOMAXCONN);

518 + ret = uv_listen((uv_stream_t *)&server_pipe, 1, connection_cb);

519 + }

520 + if (ret) {

521 error("uv_listen(): %s", uv_strerror(ret));

522 command_thread_error = ret;

523 goto error_after_uv_listen;

@@ -563,6 +570,9 @@ void commands_init(void)

570 int error;

571

572 sanity_check();

573 + if (command_server_initialized)

574 + return;

575 +

576 info("Initializing command server.");

577 for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {

578 uv_mutex_init(&command_lock_array[i]);

@@ -587,6 +597,8 @@ void commands_init(void)

597 }

598 goto after_error;

599 }

600 +

601 + command_server_initialized = 1;

602 return;

603

604 after_error:

@@ -597,6 +609,9 @@ void commands_exit(void)

609 {

610 cmd_t i;

611

612 + if (!command_server_initialized)

613 + return;

614 +

615 command_thread_shutdown = 1;

616 info("Shutting down command server.");

617 /* wake up event loop */

@@ -608,4 +623,5 @@ void commands_exit(void)

623 }

624 uv_rwlock_destroy(&exclusive_rwlock);

625 info("Command server has stopped.");

626 + command_server_initialized = 0;

627 }

database/engine/pagecache.c

+4 -4

@@ -221,7 +221,7 @@ static void pg_cache_release_pages(struct rrdengine_instance *ctx, unsigned numb

221 unsigned long pg_cache_hard_limit(struct rrdengine_instance *ctx)

222 {

223 /* it's twice the number of producers since we pin 2 pages per producer */

224 - return ctx->max_cache_pages + 2 * (unsigned long)ctx->stats.metric_API_producers;

224 + return ctx->max_cache_pages + 2 * (unsigned long)ctx->metric_API_max_producers;

225 }

226

227 /*

@@ -231,7 +231,7 @@ unsigned long pg_cache_hard_limit(struct rrdengine_instance *ctx)

231 unsigned long pg_cache_soft_limit(struct rrdengine_instance *ctx)

232 {

233 /* it's twice the number of producers since we pin 2 pages per producer */

234 - return ctx->cache_pages_low_watermark + 2 * (unsigned long)ctx->stats.metric_API_producers;

234 + return ctx->cache_pages_low_watermark + 2 * (unsigned long)ctx->metric_API_max_producers;

235 }

236

237 /*

@@ -240,8 +240,8 @@ unsigned long pg_cache_soft_limit(struct rrdengine_instance *ctx)

240 */

241 unsigned long pg_cache_committed_hard_limit(struct rrdengine_instance *ctx)

242 {

243 - /* We remove the active pages of the producers from the calculation and only allow 50% of the extra pinned pages */

244 - return ctx->cache_pages_low_watermark + (unsigned long)ctx->stats.metric_API_producers / 2;

243 + /* We remove the active pages of the producers from the calculation and only allow the extra pinned pages */

244 + return ctx->cache_pages_low_watermark + (unsigned long)ctx->metric_API_max_producers;

245 }

246

247 /*

database/engine/rrdengine.c

+2 -1

@@ -814,7 +814,7 @@ void timer_cb(uv_timer_t* handle)

814 uv_rwlock_rdlock(&pg_cache->committed_page_index.lock);

815 nr_committed_pages = pg_cache->committed_page_index.nr_committed_pages;

816 uv_rwlock_rdunlock(&pg_cache->committed_page_index.lock);

817 - producers = ctx->stats.metric_API_producers;

817 + producers = ctx->metric_API_max_producers;

818 /* are flushable pages more than 25% of the maximum page cache size */

819 high_watermark = (ctx->max_cache_pages * 25LLU) / 100;

820 low_watermark = (ctx->max_cache_pages * 5LLU) / 100; /* 5%, must be smaller than high_watermark */

@@ -920,6 +920,7 @@ void rrdeng_worker(void* arg)

920 break;

921 case RRDENG_SHUTDOWN:

922 shutdown = 1;

923 + ctx->drop_metrics_under_page_cache_pressure = 0;

924 break;

925 case RRDENG_READ_PAGE:

926 do_read_extent(wc, &cmd.read_page.page_cache_descr, 1, 0);

database/engine/rrdengine.h

@@ -183,6 +183,7 @@ struct rrdengine_instance {

183 unsigned last_fileno; /* newest index of datafile and journalfile */

184 unsigned long max_cache_pages;

185 unsigned long cache_pages_low_watermark;

186 + unsigned long metric_API_max_producers;

187

188 struct rrdengine_statistics stats;

189 };

database/engine/rrdengineapi.c

+14 -1

@@ -189,9 +189,21 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n

189 if (perfect_page_alignment)

190 rd->rrdset->rrddim_page_alignment = descr->page_length;

191 if (unlikely(INVALID_TIME == descr->start_time)) {

192 + unsigned long new_metric_API_producers, old_metric_API_max_producers, ret_metric_API_max_producers;

193 descr->start_time = point_in_time;

194

194 - rrd_stat_atomic_add(&ctx->stats.metric_API_producers, 1);

195 + new_metric_API_producers = rrd_atomic_add_fetch(&ctx->stats.metric_API_producers, 1);

196 + while (unlikely(new_metric_API_producers > (old_metric_API_max_producers = ctx->metric_API_max_producers))) {

197 + /* Increase ctx->metric_API_max_producers */

198 + ret_metric_API_max_producers = ulong_compare_and_swap(&ctx->metric_API_max_producers,

199 + old_metric_API_max_producers,

200 + new_metric_API_producers);

201 + if (old_metric_API_max_producers == ret_metric_API_max_producers) {

202 + /* success */

203 + break;

204 + }

205 + }

206 +

207 pg_cache_insert(ctx, handle->page_index, descr);

208 } else {

209 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

803 strncpyz(ctx->dbfiles_path, dbfiles_path, sizeof(ctx->dbfiles_path) - 1);

804 ctx->dbfiles_path[sizeof(ctx->dbfiles_path) - 1] = '\0';

805 ctx->drop_metrics_under_page_cache_pressure = rrdeng_drop_metrics_under_page_cache_pressure;

806 + ctx->metric_API_max_producers = 0;

807

808 memset(&ctx->worker_config, 0, sizeof(ctx->worker_config));

809 ctx->worker_config.ctx = ctx;

database/engine/rrdenginelib.h

@@ -28,8 +28,10 @@ typedef uintptr_t rrdeng_stats_t;

29 #ifdef __ATOMIC_RELAXED

30 #define rrd_atomic_fetch_add(p, n) __atomic_fetch_add(p, n, __ATOMIC_RELAXED)

31 +#define rrd_atomic_add_fetch(p, n) __atomic_add_fetch(p, n, __ATOMIC_RELAXED)

32 #else

33 #define rrd_atomic_fetch_add(p, n) __sync_fetch_and_add(p, n)

34 +#define rrd_atomic_add_fetch(p, n) __sync_add_and_fetch(p, n)

35 #endif

37 #define rrd_stat_atomic_add(p, n) rrd_atomic_fetch_add(p, n)