e91d1110e · cryptotaxi247/netdata-1

CMakeLists.txt

+2

@@ -373,6 +373,8 @@ set(LIBNETDATA_FILES

373 libnetdata/buffer/buffer.h

374 libnetdata/clocks/clocks.c

375 libnetdata/clocks/clocks.h

376 + libnetdata/completion/completion.c

377 + libnetdata/completion/completion.h

378 libnetdata/dictionary/dictionary.c

379 libnetdata/dictionary/dictionary.h

380 libnetdata/eval/eval.c

Makefile.am

+2

@@ -144,6 +144,8 @@ LIBNETDATA_FILES = \

144 libnetdata/circular_buffer/circular_buffer.h \

145 libnetdata/clocks/clocks.c \

146 libnetdata/clocks/clocks.h \

147 + libnetdata/completion/completion.c \

148 + libnetdata/completion/completion.h \

149 libnetdata/dictionary/dictionary.c \

150 libnetdata/dictionary/dictionary.h \

151 libnetdata/eval/eval.c \

configure.ac

+1

@@ -1711,6 +1711,7 @@ AC_CONFIG_FILES([

1711 libnetdata/avl/Makefile

1712 libnetdata/buffer/Makefile

1713 libnetdata/clocks/Makefile

1714 + libnetdata/completion/Makefile

1715 libnetdata/config/Makefile

1716 libnetdata/dictionary/Makefile

1717 libnetdata/ebpf/Makefile

daemon/commands.c

+5 -6

@@ -1,7 +1,6 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

2

3 #include "common.h"

4 -#include "database/engine/rrdenginelib.h"

4

5 static uv_thread_t thread;

6 static uv_loop_t* loop;

@@ -640,7 +639,7 @@ static void command_thread(void *arg)

639 command_thread_error = 0;

640 command_thread_shutdown = 0;

641 /* wake up initialization thread */

643 - complete(&completion);

642 + completion_mark_complete(&completion);

643

644 while (command_thread_shutdown == 0) {

645 uv_run(loop, UV_RUN_DEFAULT);

@@ -669,7 +668,7 @@ error_after_loop_init:

668 freez(loop);

669

670 /* wake up initialization thread */

672 - complete(&completion);

671 + completion_mark_complete(&completion);

672 }

673

674 static void sanity_check(void)

@@ -693,15 +692,15 @@ void commands_init(void)

692 }

693 fatal_assert(0 == uv_rwlock_init(&exclusive_rwlock));

694

696 - init_completion(&completion);

695 + completion_init(&completion);

696 error = uv_thread_create(&thread, command_thread, NULL);

697 if (error) {

698 error("uv_thread_create(): %s", uv_strerror(error));

699 goto after_error;

700 }

701 /* wait for worker thread to initialize */

703 - wait_for_completion(&completion);

704 - destroy_completion(&completion);

702 + completion_wait_for(&completion);

703 + completion_destroy(&completion);

704 uv_thread_set_name_np(thread, "DAEMON_COMMAND");

705

706 if (command_thread_error) {

daemon/unit_test.c

+7 -7

@@ -1917,7 +1917,7 @@ static void generate_dbengine_chart(void *arg)

1917

1918 thread_info->rd[j] = rd[j] = rrddim_add(st, name, NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);

1919 }

1920 - complete(&thread_info->charts_initialized);

1920 + completion_mark_complete(&thread_info->charts_initialized);

1921

1922 // feed it with the test data

1923 time_current = time_present - history_seconds;

@@ -1986,10 +1986,10 @@ void generate_dbengine_dataset(unsigned history_seconds)

1986 thread_info[i]->time_present = time_present;

1987 thread_info[i]->time_max = 0;

1988 thread_info[i]->done = 0;

1989 - init_completion(&thread_info[i]->charts_initialized);

1989 + completion_init(&thread_info[i]->charts_initialized);

1990 assert(0 == uv_thread_create(&thread_info[i]->thread, generate_dbengine_chart, thread_info[i]));

1991 - wait_for_completion(&thread_info[i]->charts_initialized);

1992 - destroy_completion(&thread_info[i]->charts_initialized);

1991 + completion_wait_for(&thread_info[i]->charts_initialized);

1992 + completion_destroy(&thread_info[i]->charts_initialized);

1993 }

1994 for (i = 0 ; i < DSET_CHARTS ; ++i) {

1995 assert(0 == uv_thread_join(&thread_info[i]->thread));

@@ -2177,13 +2177,13 @@ void dbengine_stress_test(unsigned TEST_DURATION_SEC, unsigned DSET_CHARTS, unsi

2177 chart_threads[i]->time_max = 0;

2178 chart_threads[i]->done = 0;

2179 chart_threads[i]->errors = chart_threads[i]->stored_metrics_nr = 0;

2180 - init_completion(&chart_threads[i]->charts_initialized);

2180 + completion_init(&chart_threads[i]->charts_initialized);

2181 assert(0 == uv_thread_create(&chart_threads[i]->thread, generate_dbengine_chart, chart_threads[i]));

2182 }

2183 /* barrier so that subsequent queries can access valid chart data */

2184 for (i = 0 ; i < DSET_CHARTS ; ++i) {

2185 - wait_for_completion(&chart_threads[i]->charts_initialized);

2186 - destroy_completion(&chart_threads[i]->charts_initialized);

2185 + completion_wait_for(&chart_threads[i]->charts_initialized);

2186 + completion_destroy(&chart_threads[i]->charts_initialized);

2187 }

2188 sleep(RAMP_UP_SECONDS);

2189 /* at this point data have already began being written to the database */

database/engine/pagecache.c

+3 -3

@@ -289,14 +289,14 @@ static void pg_cache_reserve_pages(struct rrdengine_instance *ctx, unsigned numb

289 ++failures;

290 uv_rwlock_wrunlock(&pg_cache->pg_cache_rwlock);

291

292 - init_completion(&compl);

292 + completion_init(&compl);

293 cmd.opcode = RRDENG_FLUSH_PAGES;

294 cmd.completion = &compl;

295 rrdeng_enq_cmd(&ctx->worker_config, &cmd);

296 /* wait for some pages to be flushed */

297 debug(D_RRDENGINE, "%s: waiting for pages to be written to disk before evicting.", __func__);

298 - wait_for_completion(&compl);

299 - destroy_completion(&compl);

298 + completion_wait_for(&compl);

299 + completion_destroy(&compl);

300

301 if (unlikely(failures > 1)) {

302 unsigned long slots, usecs_to_sleep;

database/engine/rrdengine.c

+9 -9

@@ -207,7 +207,7 @@ void read_cached_extent_cb(struct rrdengine_worker_config* wc, unsigned idx, str

207 }

208 }

209 if (xt_io_descr->completion)

210 - complete(xt_io_descr->completion);

210 + completion_mark_complete(xt_io_descr->completion);

211 freez(xt_io_descr);

212 }

213

@@ -360,7 +360,7 @@ after_crc_check:

360 freez(uncompressed_buf);

361 }

362 if (xt_io_descr->completion)

363 - complete(xt_io_descr->completion);

363 + completion_mark_complete(xt_io_descr->completion);

364 uv_fs_req_cleanup(req);

365 free(xt_io_descr->buf);

366 freez(xt_io_descr);

@@ -634,7 +634,7 @@ void flush_pages_cb(uv_fs_t* req)

634 rrdeng_page_descr_mutex_unlock(ctx, descr);

635 }

636 if (xt_io_descr->completion)

637 - complete(xt_io_descr->completion);

637 + completion_mark_complete(xt_io_descr->completion);

638 uv_fs_req_cleanup(req);

639 free(xt_io_descr->buf);

640 freez(xt_io_descr);

@@ -712,7 +712,7 @@ static int do_flush_pages(struct rrdengine_worker_config* wc, int force, struct

712 if (!count) {

713 debug(D_RRDENGINE, "%s: no pages eligible for flushing.", __func__);

714 if (completion)

715 - complete(completion);

715 + completion_mark_complete(completion);

716 return 0;

717 }

718 wc->inflight_dirty_pages += count;

@@ -975,7 +975,7 @@ static void rrdeng_cleanup_finished_threads(struct rrdengine_worker_config* wc)

975 }

976 if (unlikely(SET_QUIESCE == ctx->quiesce && !rrdeng_threads_alive(wc))) {

977 ctx->quiesce = QUIESCED;

978 - complete(&ctx->rrdengine_completion);

978 + completion_mark_complete(&ctx->rrdengine_completion);

979 }

980 }

981

@@ -1171,7 +1171,7 @@ void rrdeng_worker(void* arg)

1171

1172 wc->error = 0;

1173 /* wake up initialization thread */

1174 - complete(&ctx->rrdengine_completion);

1174 + completion_mark_complete(&ctx->rrdengine_completion);

1175

1176 fatal_assert(0 == uv_timer_start(&timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));

1177 shutdown = 0;

@@ -1211,7 +1211,7 @@ void rrdeng_worker(void* arg)

1211 wal_flush_transaction_buffer(wc);

1212 if (!rrdeng_threads_alive(wc)) {

1213 ctx->quiesce = QUIESCED;

1214 - complete(&ctx->rrdengine_completion);

1214 + completion_mark_complete(&ctx->rrdengine_completion);

1215 }

1216 break;

1217 case RRDENG_READ_PAGE:

@@ -1226,7 +1226,7 @@ void rrdeng_worker(void* arg)

1226 case RRDENG_FLUSH_PAGES: {

1227 if (wc->now_invalidating_dirty_pages) {

1228 /* Do not flush if the disk cannot keep up */

1229 - complete(cmd.completion);

1229 + completion_mark_complete(cmd.completion);

1230 } else {

1231 (void)do_flush_pages(wc, 1, cmd.completion);

1232 }

@@ -1276,7 +1276,7 @@ error_after_loop_init:

1276

1277 wc->error = UV_EAGAIN;

1278 /* wake up initialization thread */

1279 - complete(&ctx->rrdengine_completion);

1279 + completion_mark_complete(&ctx->rrdengine_completion);

1280 }

1281

1282 /* C entry point for development purposes

database/engine/rrdengineapi.c

+6 -6

@@ -944,11 +944,11 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p

944 goto error_after_init_rrd_files;

945 }

946

947 - init_completion(&ctx->rrdengine_completion);

947 + completion_init(&ctx->rrdengine_completion);

948 fatal_assert(0 == uv_thread_create(&ctx->worker_config.thread, rrdeng_worker, &ctx->worker_config));

949 /* wait for worker thread to initialize */

950 - wait_for_completion(&ctx->rrdengine_completion);

951 - destroy_completion(&ctx->rrdengine_completion);

950 + completion_wait_for(&ctx->rrdengine_completion);

951 + completion_destroy(&ctx->rrdengine_completion);

952 uv_thread_set_name_np(ctx->worker_config.thread, "DBENGINE");

953 if (ctx->worker_config.error) {

954 goto error_after_rrdeng_worker;

@@ -1009,13 +1009,13 @@ void rrdeng_prepare_exit(struct rrdengine_instance *ctx)

1009 return;

1010 }

1011

1012 - init_completion(&ctx->rrdengine_completion);

1012 + completion_init(&ctx->rrdengine_completion);

1013 cmd.opcode = RRDENG_QUIESCE;

1014 rrdeng_enq_cmd(&ctx->worker_config, &cmd);

1015

1016 /* wait for dbengine to quiesce */

1017 - wait_for_completion(&ctx->rrdengine_completion);

1018 - destroy_completion(&ctx->rrdengine_completion);

1017 + completion_wait_for(&ctx->rrdengine_completion);

1018 + completion_destroy(&ctx->rrdengine_completion);

1019

1020 //metalog_prepare_exit(ctx->metalog_ctx);

1021 }

database/engine/rrdenginelib.h

-40

@@ -14,9 +14,6 @@ struct rrdengine_instance;

14

15 #define BITS_PER_ULONG (sizeof(unsigned long) * 8)

16

17 -/* Taken from linux kernel */

18 -#define BUILD_BUG_ON(condition) ((void)sizeof(char[1 - 2*!!(condition)]))

19 -

17 #define ALIGN_BYTES_FLOOR(x) (((x) / RRDENG_BLOCK_SIZE) * RRDENG_BLOCK_SIZE)

18 #define ALIGN_BYTES_CEILING(x) ((((x) + RRDENG_BLOCK_SIZE - 1) / RRDENG_BLOCK_SIZE) * RRDENG_BLOCK_SIZE)

19

@@ -76,43 +73,6 @@ static inline unsigned long ulong_compare_and_swap(volatile unsigned long *ptr,

73 #define O_DIRECT (0)

74 #endif

75

79 -struct completion {

80 - uv_mutex_t mutex;

81 - uv_cond_t cond;

82 - volatile unsigned completed;

83 -};

84 -

85 -static inline void init_completion(struct completion *p)

86 -{

87 - p->completed = 0;

88 - fatal_assert(0 == uv_cond_init(&p->cond));

89 - fatal_assert(0 == uv_mutex_init(&p->mutex));

90 -}

91 -

92 -static inline void destroy_completion(struct completion *p)

93 -{

94 - uv_cond_destroy(&p->cond);

95 - uv_mutex_destroy(&p->mutex);

96 -}

97 -

98 -static inline void wait_for_completion(struct completion *p)

99 -{

100 - uv_mutex_lock(&p->mutex);

101 - while (0 == p->completed) {

102 - uv_cond_wait(&p->cond, &p->mutex);

103 - }

104 - fatal_assert(1 == p->completed);

105 - uv_mutex_unlock(&p->mutex);

106 -}

107 -

108 -static inline void complete(struct completion *p)

109 -{

110 - uv_mutex_lock(&p->mutex);

111 - p->completed = 1;

112 - uv_mutex_unlock(&p->mutex);

113 - uv_cond_broadcast(&p->cond);

114 -}

115 -

76 static inline int crc32cmp(void *crcp, uLong crc)

77 {

78 return (*(uint32_t *)crcp != crc);

libnetdata/Makefile.am

+1

@@ -8,6 +8,7 @@ SUBDIRS = \

8 avl \

9 buffer \

10 clocks \

11 + completion \

12 config \

13 dictionary \

14 ebpf \

libnetdata/completion/Makefile.am new

+4

@@ -0,0 +1,4 @@

1 +# SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +AUTOMAKE_OPTIONS = subdir-objects

4 +MAINTAINERCLEANFILES = $(srcdir)/Makefile.in

libnetdata/completion/completion.c new

+34

@@ -0,0 +1,34 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "completion.h"

4 +

5 +void completion_init(struct completion *p)

6 +{

7 + p->completed = 0;

8 + fatal_assert(0 == uv_cond_init(&p->cond));

9 + fatal_assert(0 == uv_mutex_init(&p->mutex));

10 +}

11 +

12 +void completion_destroy(struct completion *p)

13 +{

14 + uv_cond_destroy(&p->cond);

15 + uv_mutex_destroy(&p->mutex);

16 +}

17 +

18 +void completion_wait_for(struct completion *p)

19 +{

20 + uv_mutex_lock(&p->mutex);

21 + while (0 == p->completed) {

22 + uv_cond_wait(&p->cond, &p->mutex);

23 + }

24 + fatal_assert(1 == p->completed);

25 + uv_mutex_unlock(&p->mutex);

26 +}

27 +

28 +void completion_mark_complete(struct completion *p)

29 +{

30 + uv_mutex_lock(&p->mutex);

31 + p->completed = 1;

32 + uv_mutex_unlock(&p->mutex);

33 + uv_cond_broadcast(&p->cond);

34 +}

libnetdata/completion/completion.h new

+22

@@ -0,0 +1,22 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_COMPLETION_H

4 +#define NETDATA_COMPLETION_H

5 +

6 +#include "../libnetdata.h"

7 +

8 +struct completion {

9 + uv_mutex_t mutex;

10 + uv_cond_t cond;

11 + volatile unsigned completed;

12 +};

13 +

14 +void completion_init(struct completion *p);

15 +

16 +void completion_destroy(struct completion *p);

17 +

18 +void completion_wait_for(struct completion *p);

19 +

20 +void completion_mark_complete(struct completion *p);

21 +

22 +#endif /* NETDATA_COMPLETION_H */

libnetdata/libnetdata.h

+6

@@ -303,9 +303,14 @@ extern char *find_and_replace(const char *src, const char *find, const char *rep

303 #define KILOBITS_IN_A_MEGABIT 1000

304

305 /* misc. */

306 +

307 #define UNUSED(x) (void)(x)

308 #define error_report(x, args...) do { errno = 0; error(x, ##args); } while(0)

309

310 +// Taken from linux kernel

311 +#define BUILD_BUG_ON(condition) ((void)sizeof(char[1 - 2*!!(condition)]))

312 +

313 +

314 extern void netdata_cleanup_and_exit(int ret) NORETURN;

315 extern void send_statistics(const char *action, const char *action_result, const char *action_data);

316 extern char *netdata_configured_host_prefix;

@@ -318,6 +323,7 @@ extern char *netdata_configured_host_prefix;

323 #include "avl/avl.h"

324 #include "inlined.h"

325 #include "clocks/clocks.h"

326 +#include "completion/completion.h"

327 #include "popen/popen.h"

328 #include "simple_pattern/simple_pattern.h"

329 #ifdef ENABLE_HTTPS

spawn/spawn.c

+3 -4

@@ -1,7 +1,6 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

2

3 #include "spawn.h"

4 -#include "database/engine/rrdenginelib.h"

4

5 static uv_thread_t thread;

6 int spawn_thread_error;

@@ -240,15 +239,15 @@ void spawn_init(void)

239

240 init_spawn_cmd_queue();

241

243 - init_completion(&completion);

242 + completion_init(&completion);

243 error = uv_thread_create(&thread, spawn_client, &completion);

244 if (error) {

245 error("uv_thread_create(): %s", uv_strerror(error));

246 goto after_error;

247 }

248 /* wait for spawn client thread to initialize */

250 - wait_for_completion(&completion);

251 - destroy_completion(&completion);

249 + completion_wait_for(&completion);

250 + completion_destroy(&completion);

251 uv_thread_set_name_np(thread, "DAEMON_SPAWN");

252

253 if (spawn_thread_error) {

spawn/spawn_client.c

+2 -2

@@ -201,7 +201,7 @@ void spawn_client(void *arg)

201 spawn_thread_error = 0;

202 spawn_thread_shutdown = 0;

203 /* wake up initialization thread */

204 - complete(completion);

204 + completion_mark_complete(completion);

205

206 prot_buffer_len = 0;

207 ret = uv_read_start((uv_stream_t *)&spawn_channel, on_read_alloc, on_pipe_read);

@@ -237,5 +237,5 @@ error_after_loop_init:

237 freez(loop);

238

239 /* wake up initialization thread */

240 - complete(completion);

240 + completion_mark_complete(completion);

241 }

Do not use dbengine headers when dbengine is disabled. (#11967)