@cryptotaxi247 / netdata-1 / commits / f351681b0

Improve agent shutdown (#19775)

Enhance DBENGINE flushing mechanism to prioritize dirty page flushing first during shutdown

Stelios Fragkakis committed Mar 5, 2025 at 23:26 UTC f351681b04c1ce5df31ba8c5fa3f24a2092d91c6

11 files changed +50 -12

src/daemon/daemon-shutdown.c

+8 -7

@@ -100,15 +100,16 @@ static void *rrdeng_exit_background(void *ptr) {

100 }

101

102 #ifdef ENABLE_DBENGINE

103 -static void rrdeng_flush_everything_and_wait(bool wait_flush, bool wait_collectors) {

103 +static void rrdeng_flush_everything_and_wait(bool wait_flush, bool wait_collectors, bool dirty_only)

104 +{

105 static size_t starting_size_to_flush = 0;

106

107 if(!pgc_hot_and_dirty_entries(main_cache))

108 return;

109

109 - nd_log(NDLS_DAEMON, NDLP_INFO, "Flushing DBENGINE hot & dirty pages...");

110 + nd_log(NDLS_DAEMON, NDLP_INFO, "Flushing DBENGINE %s dirty pages...", dirty_only ? "only" : "hot &");

111 for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)

111 - rrdeng_quiesce(multidb_ctx[tier]);

112 + rrdeng_quiesce(multidb_ctx[tier], dirty_only);

113

114 struct pgc_statistics pgc_main_stats = pgc_get_statistics(main_cache);

115 size_t size_to_flush = pgc_main_stats.queues[PGC_QUEUE_HOT].size + pgc_main_stats.queues[PGC_QUEUE_DIRTY].size;

@@ -185,7 +186,7 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

186 #ifdef ENABLE_DBENGINE

187 if(!ret && dbengine_enabled)

188 // flush all dirty pages asap

188 - rrdeng_flush_everything_and_wait(false, false);

189 + rrdeng_flush_everything_and_wait(false, false, true);

190 #endif

191

192 // send the stat from our caller

@@ -219,7 +220,7 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

220 #ifdef ENABLE_DBENGINE

221 if(!ret && dbengine_enabled)

222 // flush all dirty pages now that all collectors and streaming completed

222 - rrdeng_flush_everything_and_wait(false, false);

223 + rrdeng_flush_everything_and_wait(false, false, false);

224 #endif

225

226 service_wait_exit(SERVICE_REPLICATION, 3 * USEC_PER_SEC);

@@ -263,7 +264,7 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

264 #ifdef ENABLE_DBENGINE

265 if(dbengine_enabled) {

266 // flush anything remaining and wait for collectors to finish

266 - rrdeng_flush_everything_and_wait(true, true);

267 + rrdeng_flush_everything_and_wait(true, true, false);

268 watcher_step_complete(WATCHER_STEP_ID_WAIT_FOR_DBENGINE_COLLECTORS_TO_FINISH);

269

270 ND_THREAD *th[nd_profile.storage_tiers];

@@ -271,7 +272,7 @@ void netdata_cleanup_and_exit(EXIT_REASON reason, const char *action, const char

272 th[tier] = nd_thread_create("rrdeng-exit", NETDATA_THREAD_OPTION_JOINABLE, rrdeng_exit_background, multidb_ctx[tier]);

273

274 // flush anything remaining again - just in case

274 - rrdeng_flush_everything_and_wait(true, false);

275 + rrdeng_flush_everything_and_wait(true, false, false);

276

277 for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++)

278 nd_thread_join(th[tier]);

src/daemon/libuv_workers.c

@@ -47,6 +47,7 @@ void register_libuv_worker_jobs() {

47 worker_register_job_name(UV_EVENT_DBENGINE_EVICT_OPEN_CACHE, "evict open");

48 worker_register_job_name(UV_EVENT_DBENGINE_EVICT_EXTENT_CACHE, "evict extent");

49 worker_register_job_name(UV_EVENT_DBENGINE_BUFFERS_CLEANUP, "dbengine buffers cleanup");

50 + worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_DIRTY, "dbengine flush dirty");

51 worker_register_job_name(UV_EVENT_DBENGINE_QUIESCE, "dbengine quiesce");

52 worker_register_job_name(UV_EVENT_DBENGINE_SHUTDOWN, "dbengine shutdown");

src/daemon/libuv_workers.h

@@ -39,6 +39,7 @@ enum event_loop_job {

39 UV_EVENT_DBENGINE_EVICT_OPEN_CACHE,

40 UV_EVENT_DBENGINE_EVICT_EXTENT_CACHE,

41 UV_EVENT_DBENGINE_BUFFERS_CLEANUP,

42 + UV_EVENT_DBENGINE_FLUSH_DIRTY,

43 UV_EVENT_DBENGINE_QUIESCE,

44 UV_EVENT_DBENGINE_SHUTDOWN,

src/database/engine/cache.c

@@ -2075,6 +2075,10 @@ struct aral_statistics *pgc_aral_stats(void) {

2075 return &pgc_aral_statistics;

2076 }

2077

2078 +void pgc_flush_dirty_pages(PGC *cache, Word_t section) {

2079 + flush_pages(cache, 0, section, true, true);

2080 +}

2081 +

2082 void pgc_flush_all_hot_and_dirty_pages(PGC *cache, Word_t section) {

2083 all_hot_pages_to_dirty(cache, section);

2084

src/database/engine/cache.h

@@ -178,6 +178,7 @@ PGC *pgc_create(const char *name,

178 void pgc_destroy(PGC *cache);

179

180 #define PGC_SECTION_ALL ((Word_t)0)

181 +void pgc_flush_dirty_pages(PGC *cache, Word_t section);

182 void pgc_flush_all_hot_and_dirty_pages(PGC *cache, Word_t section);

183

184 // add a page to the cache and return a pointer to it

src/database/engine/dbengine-stresstest.c

+1 -1

@@ -447,7 +447,7 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi

447 }

448 freez(query_threads);

449 rrd_wrlock();

450 - rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

450 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si, false);

451 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

452 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

453 rrd_wrunlock();

src/database/engine/dbengine-unittest.c

+1 -1

@@ -408,7 +408,7 @@ int test_dbengine(void) {

408 }

409

410 rrd_wrlock();

411 - rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si);

411 + rrdeng_quiesce((struct rrdengine_instance *)host->db[0].si, false);

412 rrdeng_exit((struct rrdengine_instance *)host->db[0].si);

413 rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL);

414 rrd_wrunlock();

src/database/engine/rrdengine.c

+25

@@ -1325,6 +1325,21 @@ static void *flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine

1325 return data;

1326 }

1327

1328 +static void after_flush_dirty_pages_of_section(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) {

1329 + ;

1330 +}

1331 +

1332 +static void *flush_dirty_pages_of_section_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {

1333 + worker_is_busy(UV_EVENT_DBENGINE_FLUSH_DIRTY);

1334 + pgc_flush_dirty_pages(main_cache, (Word_t)ctx);

1335 +

1336 + for(size_t i = 0; i < pgc_max_flushers() ; i++)

1337 + rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_MAIN, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1338 +

1339 + return data;

1340 +}

1341 +

1342 +

1343 static void after_populate_mrg(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t* req __maybe_unused, int status __maybe_unused) {

1344 ;

1345 }

@@ -1873,6 +1888,7 @@ void dbengine_event_loop(void* arg) {

1888 worker_register_job_name(RRDENG_OPCODE_FLUSH_MAIN, "flush init");

1889 worker_register_job_name(RRDENG_OPCODE_EVICT_MAIN, "evict init");

1890 worker_register_job_name(RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown");

1891 + worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty");

1892 worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce");

1893 worker_register_job_name(RRDENG_OPCODE_SHUTDOWN_EVLOOP, "dbengine shutdown");

1894

@@ -1886,6 +1902,7 @@ void dbengine_event_loop(void* arg) {

1902 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_FLUSH_MAIN, "flush init cb");

1903 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EVICT_MAIN, "evict init cb");

1904 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown cb");

1905 + worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty cb");

1906 worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce cb");

1907

1908 // special jobs

@@ -2013,6 +2030,14 @@ void dbengine_event_loop(void* arg) {

2030 break;

2031 }

2032

2033 + case RRDENG_OPCODE_CTX_FLUSH_DIRTY: {

2034 + struct rrdengine_instance *ctx = cmd.ctx;

2035 + work_dispatch(ctx, NULL, NULL, opcode,

2036 + flush_dirty_pages_of_section_tp_worker,

2037 + after_flush_dirty_pages_of_section);

2038 + break;

2039 + }

2040 +

2041 case RRDENG_OPCODE_CTX_QUIESCE: {

2042 // a ctx will shutdown shortly

2043 struct rrdengine_instance *ctx = cmd.ctx;

src/database/engine/rrdengine.h

@@ -255,6 +255,7 @@ enum rrdeng_opcode {

255 RRDENG_OPCODE_EVICT_OPEN,

256 RRDENG_OPCODE_EVICT_EXTENT,

257 RRDENG_OPCODE_CTX_SHUTDOWN,

258 + RRDENG_OPCODE_CTX_FLUSH_DIRTY,

259 RRDENG_OPCODE_CTX_QUIESCE,

260 RRDENG_OPCODE_CTX_POPULATE_MRG,

261 RRDENG_OPCODE_SHUTDOWN_EVLOOP,

src/database/engine/rrdengineapi.c

+6 -2

@@ -1294,14 +1294,18 @@ int rrdeng_exit(struct rrdengine_instance *ctx) {

1294 return 0;

1295 }

1296

1297 -void rrdeng_quiesce(struct rrdengine_instance *ctx) {

1297 +void rrdeng_quiesce(struct rrdengine_instance *ctx, bool dirty_only)

1298 +{

1299 if (NULL == ctx)

1300 return;

1301

1302 // FIXME - ktsaou - properly cleanup ctx

1303 // 1. make sure all collectors are stopped

1304

1304 - rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1305 + if (dirty_only)

1306 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_FLUSH_DIRTY, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1307 + else

1308 + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);

1309 }

1310

1311 static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats)

src/database/engine/rrdengineapi.h

+1 -1

@@ -74,7 +74,7 @@ int rrdeng_init(

74 void rrdeng_readiness_wait(struct rrdengine_instance *ctx);

76 int rrdeng_exit(struct rrdengine_instance *ctx);

77 -void rrdeng_quiesce(struct rrdengine_instance *ctx);

77 +void rrdeng_quiesce(struct rrdengine_instance *ctx, bool dirty_only);

79 bool rrdeng_metric_retention_by_id(STORAGE_INSTANCE *si, UUIDMAP_ID id, time_t *first_entry_s, time_t *last_entry_s);

80 bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, nd_uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s);