@cryptotaxi247 / netdata / commits / 8883dd39e

Replace uv mutex and condition variables with netdata equivalents (#20941)

* replace uv mutex and condition variables with netdata equivalents * replace uv_mutex_unlock with netdata_mutex_unlock in exporting_engine.h * replace pthread_rwlock_t with uv_rwlock_t in netdata_rwlock_t * replace ETIMEDOUT with UV_ETIMEDOUT in rrdfunctions-inflight.c

Stelios Fragkakis committed Sep 9, 2025 at 17:57 UTC 8883dd39e34e19bd1aea36cc42a0d145e85484c2

30 files changed +235 -284

src/collectors/cgroups.plugin/cgroup-discovery.c

+7 -7

@@ -1271,7 +1271,7 @@ static inline void discovery_find_all_cgroups() {

1271 discovery_update_filenames_all_cgroups();

1272

1273 worker_is_busy(WORKER_DISCOVERY_LOCK);

1274 - uv_mutex_lock(&cgroup_root_mutex);

1274 + netdata_mutex_lock(&cgroup_root_mutex);

1275

1276 worker_is_busy(WORKER_DISCOVERY_CLEANUP);

1277 discovery_cleanup_all_cgroups();

@@ -1279,7 +1279,7 @@ static inline void discovery_find_all_cgroups() {

1279 worker_is_busy(WORKER_DISCOVERY_COPY);

1280 discovery_copy_discovered_cgroups_to_reader();

1281

1282 - uv_mutex_unlock(&cgroup_root_mutex);

1282 + netdata_mutex_unlock(&cgroup_root_mutex);

1283

1284 worker_is_busy(WORKER_DISCOVERY_SHARE);

1285 discovery_share_cgroups_with_ebpf();

@@ -1318,9 +1318,9 @@ void cgroup_discovery_worker(void *ptr)

1318 while (service_running(SERVICE_COLLECTORS)) {

1319 worker_is_idle();

1320

1321 - uv_mutex_lock(&discovery_thread.mutex);

1322 - uv_cond_wait(&discovery_thread.cond_var, &discovery_thread.mutex);

1323 - uv_mutex_unlock(&discovery_thread.mutex);

1321 + netdata_mutex_lock(&discovery_thread.mutex);

1322 + netdata_cond_wait(&discovery_thread.cond_var, &discovery_thread.mutex);

1323 + netdata_mutex_unlock(&discovery_thread.mutex);

1324

1325 if (unlikely(!service_running(SERVICE_COLLECTORS)))

1326 break;

@@ -1329,13 +1329,13 @@ void cgroup_discovery_worker(void *ptr)

1329 }

1330

1331 // free all cgroups

1332 - uv_mutex_lock(&cgroup_root_mutex);

1332 + netdata_mutex_lock(&cgroup_root_mutex);

1333 while(cgroup_root) {

1334 struct cgroup *cg = cgroup_root;

1335 cgroup_root = cg->next;

1336 cgroup_free(cg);

1337 }

1338 - uv_mutex_unlock(&cgroup_root_mutex);

1338 + netdata_mutex_unlock(&cgroup_root_mutex);

1339

1340 collector_info("discovery thread stopped");

1341 cgroup_cleanup_ebpf_integration();

src/collectors/cgroups.plugin/cgroup-internals.h

+3 -3

@@ -262,8 +262,8 @@ struct cgroup {

262

263 struct discovery_thread {

264 ND_THREAD *thread;

265 - uv_mutex_t mutex;

266 - uv_cond_t cond_var;

265 + netdata_mutex_t mutex;

266 + netdata_cond_t cond_var;

267 int exited;

268 };

269

@@ -272,7 +272,7 @@ extern struct discovery_thread discovery_thread;

272 extern const char *cgroups_rename_script;

273 extern char cgroup_chart_id_prefix[];

274 extern char services_chart_id_prefix[];

275 -extern uv_mutex_t cgroup_root_mutex;

275 +extern netdata_mutex_t cgroup_root_mutex;

276

277 void cgroup_discovery_worker(void *ptr);

278

src/collectors/cgroups.plugin/cgroup-top.c

+4 -4

@@ -131,7 +131,7 @@ int cgroup_function_cgroup_top(BUFFER *wb, const char *function __maybe_unused,

131

132 RRDDIM *rd = NULL;

133

134 - uv_mutex_lock(&cgroup_root_mutex);

134 + netdata_mutex_lock(&cgroup_root_mutex);

135

136 for(struct cgroup *cg = cgroup_root; cg ; cg = cg->next) {

137 if(unlikely(!cg->enabled || cg->pending_renames || !cg->function_ready || is_cgroup_systemd_service(cg)))

@@ -185,7 +185,7 @@ int cgroup_function_cgroup_top(BUFFER *wb, const char *function __maybe_unused,

185 buffer_json_array_close(wb);

186 }

187

188 - uv_mutex_unlock(&cgroup_root_mutex);

188 + netdata_mutex_unlock(&cgroup_root_mutex);

189

190 buffer_json_array_close(wb); // data

191 buffer_json_member_add_object(wb, "columns");

@@ -371,7 +371,7 @@ int cgroup_function_systemd_top(BUFFER *wb, const char *function __maybe_unused,

371

372 RRDDIM *rd = NULL;

373

374 - uv_mutex_lock(&cgroup_root_mutex);

374 + netdata_mutex_lock(&cgroup_root_mutex);

375

376 for(struct cgroup *cg = cgroup_root; cg ; cg = cg->next) {

377 if(unlikely(!cg->enabled || cg->pending_renames || !cg->function_ready || !is_cgroup_systemd_service(cg)))

@@ -405,7 +405,7 @@ int cgroup_function_systemd_top(BUFFER *wb, const char *function __maybe_unused,

405 buffer_json_array_close(wb);

406 }

407

408 - uv_mutex_unlock(&cgroup_root_mutex);

408 + netdata_mutex_unlock(&cgroup_root_mutex);

409

410 buffer_json_array_close(wb); // data

411 buffer_json_member_add_object(wb, "columns");

src/collectors/cgroups.plugin/sys_fs_cgroup.c

+14 -14

@@ -54,7 +54,7 @@ uint32_t throttled_usec_hash = 0;

55 // *** WARNING *** The fields are not thread safe. Take care of safe usage.

56 struct cgroup *cgroup_root = NULL;

57 -uv_mutex_t cgroup_root_mutex;

57 +netdata_mutex_t cgroup_root_mutex;

59 struct cgroups_systemd_config_setting cgroups_systemd_options[] = {

60 { .name = "legacy", .setting = SYSTEMD_CGROUP_LEGACY },

@@ -1335,9 +1335,9 @@ static void cgroup_main_cleanup(void *pptr) {

1335 if (!__atomic_load_n(&discovery_thread.exited, __ATOMIC_RELAXED)) {

1336 collector_info("waiting for discovery thread to finish...");

1337 while (!__atomic_load_n(&discovery_thread.exited, __ATOMIC_RELAXED) && max > 0) {

1338 - uv_mutex_lock(&discovery_thread.mutex);

1339 - uv_cond_signal(&discovery_thread.cond_var);

1340 - uv_mutex_unlock(&discovery_thread.mutex);

1338 + netdata_mutex_lock(&discovery_thread.mutex);

1339 + netdata_cond_signal(&discovery_thread.cond_var);

1340 + netdata_mutex_unlock(&discovery_thread.mutex);

1341 max -= step;

1342 sleep_usec(step);

1343 }

@@ -1381,7 +1381,7 @@ void cgroups_main(void *ptr) {

1381

1382 cgroup_read_host_total_ram();

1383

1384 - if (uv_mutex_init(&cgroup_root_mutex)) {

1384 + if (netdata_mutex_init(&cgroup_root_mutex)) {

1385 collector_error("CGROUP: cannot initialize mutex for the main cgroup list");

1386 return;

1387 }

@@ -1392,11 +1392,11 @@ void cgroups_main(void *ptr) {

1392

1393 discovery_thread.exited = 0;

1394

1395 - if (uv_mutex_init(&discovery_thread.mutex)) {

1395 + if (netdata_mutex_init(&discovery_thread.mutex)) {

1396 collector_error("CGROUP: cannot initialize mutex for discovery thread");

1397 return;

1398 }

1399 - if (uv_cond_init(&discovery_thread.cond_var)) {

1399 + if (netdata_cond_init(&discovery_thread.cond_var)) {

1400 collector_error("CGROUP: cannot initialize conditional variable for discovery thread");

1401 return;

1402 }

@@ -1434,21 +1434,21 @@ void cgroups_main(void *ptr) {

1434

1435 find_dt += hb_dt;

1436 if (unlikely(find_dt >= find_every || (!is_inside_k8s && cgroups_check))) {

1437 - uv_mutex_lock(&discovery_thread.mutex);

1438 - uv_cond_signal(&discovery_thread.cond_var);

1439 - uv_mutex_unlock(&discovery_thread.mutex);

1437 + netdata_mutex_lock(&discovery_thread.mutex);

1438 + netdata_cond_signal(&discovery_thread.cond_var);

1439 + netdata_mutex_unlock(&discovery_thread.mutex);

1440 find_dt = 0;

1441 cgroups_check = 0;

1442 }

1443

1444 worker_is_busy(WORKER_CGROUPS_LOCK);

1445 - uv_mutex_lock(&cgroup_root_mutex);

1445 + netdata_mutex_lock(&cgroup_root_mutex);

1446

1447 worker_is_busy(WORKER_CGROUPS_READ);

1448 read_all_discovered_cgroups(cgroup_root);

1449

1450 if (unlikely(!service_running(SERVICE_COLLECTORS))) {

1451 - uv_mutex_unlock(&cgroup_root_mutex);

1451 + netdata_mutex_unlock(&cgroup_root_mutex);

1452 break;

1453 }

1454

@@ -1458,11 +1458,11 @@ void cgroups_main(void *ptr) {

1458 update_cgroup_systemd_services_charts();

1459

1460 if (unlikely(!service_running(SERVICE_COLLECTORS))) {

1461 - uv_mutex_unlock(&cgroup_root_mutex);

1461 + netdata_mutex_unlock(&cgroup_root_mutex);

1462 break;

1463 }

1464

1465 worker_is_idle();

1466 - uv_mutex_unlock(&cgroup_root_mutex);

1466 + netdata_mutex_unlock(&cgroup_root_mutex);

1467 }

1468 }

src/daemon/commands.c

+14 -14

@@ -73,9 +73,9 @@ static command_info_t command_info_array[] = {

73 };

75 /* Mutexes for commands of type CMD_TYPE_ORTHOGONAL */

76 -static uv_mutex_t command_lock_array[CMD_TOTAL_COMMANDS];

76 +static netdata_mutex_t command_lock_array[CMD_TOTAL_COMMANDS];

77 /* Commands of type CMD_TYPE_EXCLUSIVE are writers */

78 -static uv_rwlock_t exclusive_rwlock;

78 +static netdata_rwlock_t exclusive_rwlock;

79 /*

80 * Locking order:

81 * 1. exclusive_rwlock

@@ -485,20 +485,20 @@ static void cmd_lock_exclusive(unsigned index)

485 {

486 (void)index;

487

488 - uv_rwlock_wrlock(&exclusive_rwlock);

488 + netdata_rwlock_wrlock(&exclusive_rwlock);

489 }

490

491 static void cmd_lock_orthogonal(unsigned index)

492 {

493 - uv_rwlock_rdlock(&exclusive_rwlock);

494 - uv_mutex_lock(&command_lock_array[index]);

493 + netdata_rwlock_rdlock(&exclusive_rwlock);

494 + netdata_mutex_lock(&command_lock_array[index]);

495 }

496

497 static void cmd_lock_idempotent(unsigned index)

498 {

499 (void)index;

500

501 - uv_rwlock_rdlock(&exclusive_rwlock);

501 + netdata_rwlock_rdlock(&exclusive_rwlock);

502 }

503

504 static void cmd_lock_high_priority(unsigned index)

@@ -510,20 +510,20 @@ static void cmd_unlock_exclusive(unsigned index)

510 {

511 (void)index;

512

513 - uv_rwlock_wrunlock(&exclusive_rwlock);

513 + netdata_rwlock_wrunlock(&exclusive_rwlock);

514 }

515

516 static void cmd_unlock_orthogonal(unsigned index)

517 {

518 - uv_rwlock_rdunlock(&exclusive_rwlock);

519 - uv_mutex_unlock(&command_lock_array[index]);

518 + netdata_rwlock_rdunlock(&exclusive_rwlock);

519 + netdata_mutex_unlock(&command_lock_array[index]);

520 }

521

522 static void cmd_unlock_idempotent(unsigned index)

523 {

524 (void)index;

525

526 - uv_rwlock_rdunlock(&exclusive_rwlock);

526 + netdata_rwlock_rdunlock(&exclusive_rwlock);

527 }

528

529 static void cmd_unlock_high_priority(unsigned index)

@@ -856,9 +856,9 @@ void commands_init(void)

856

857 netdata_log_info("Initializing command server.");

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

859 - fatal_assert(0 == uv_mutex_init(&command_lock_array[i]));

859 + fatal_assert(0 == netdata_mutex_init(&command_lock_array[i]));

860 }

861 - fatal_assert(0 == uv_rwlock_init(&exclusive_rwlock));

861 + fatal_assert(0 == netdata_rwlock_init(&exclusive_rwlock));

862

863 completion_init(&completion);

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

@@ -899,9 +899,9 @@ void commands_exit(void)

899 fatal_assert(0 == uv_thread_join(&thread));

900

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

902 - uv_mutex_destroy(&command_lock_array[i]);

902 + netdata_mutex_destroy(&command_lock_array[i]);

903 }

904 - uv_rwlock_destroy(&exclusive_rwlock);

904 + netdata_rwlock_destroy(&exclusive_rwlock);

905 netdata_log_info("Command server has stopped.");

906 command_server_initialized = 0;

907 }

src/daemon/libuv_workers.c

+12 -12

@@ -167,19 +167,19 @@ void init_cmd_pool(CmdPool *pool, int size) {

167 pool->tail = 0;

168 pool->count = 0;

169

170 - fatal_assert(0 == uv_mutex_init(&pool->lock));

171 - fatal_assert(0 == uv_cond_init(&pool->not_full));

170 + fatal_assert(0 == netdata_mutex_init(&pool->lock));

171 + fatal_assert(0 == netdata_cond_init(&pool->not_full));

172 }

173

174 bool push_cmd(CmdPool *pool, const cmd_data_t *cmd, bool wait_on_full)

175 {

176 - uv_mutex_lock(&pool->lock);

176 + netdata_mutex_lock(&pool->lock);

177

178 while (pool->count == pool->size) {

179 if (wait_on_full)

180 - uv_cond_wait(&pool->not_full, &pool->lock);

180 + netdata_cond_wait(&pool->not_full, &pool->lock);

181 else {

182 - uv_mutex_unlock(&pool->lock); // No space, return

182 + netdata_mutex_unlock(&pool->lock); // No space, return

183 return false;

184 }

185 }

@@ -188,22 +188,22 @@ bool push_cmd(CmdPool *pool, const cmd_data_t *cmd, bool wait_on_full)

188 pool->tail = (pool->tail + 1) % pool->size;

189 pool->count++;

190

191 - uv_mutex_unlock(&pool->lock);

191 + netdata_mutex_unlock(&pool->lock);

192 return true;

193 }

194

195 bool pop_cmd(CmdPool *pool, cmd_data_t *out_cmd) {

196 - uv_mutex_lock(&pool->lock);

196 + netdata_mutex_lock(&pool->lock);

197 if (pool->count == 0) {

198 - uv_mutex_unlock(&pool->lock); // No commands to pop

198 + netdata_mutex_unlock(&pool->lock); // No commands to pop

199 return false;

200 }

201 *out_cmd = pool->buffer[pool->head];

202 pool->head = (pool->head + 1) % pool->size;

203 pool->count--;

204

205 - uv_cond_signal(&pool->not_full);

206 - uv_mutex_unlock(&pool->lock);

205 + netdata_cond_signal(&pool->not_full);

206 + netdata_mutex_unlock(&pool->lock);

207 return true;

208 }

209

@@ -212,8 +212,8 @@ void release_cmd_pool(CmdPool *pool) {

212 free(pool->buffer);

213 pool->buffer = NULL;

214 }

215 - uv_mutex_destroy(&pool->lock);

216 - uv_cond_destroy(&pool->not_full);

215 + netdata_mutex_destroy(&pool->lock);

216 + netdata_cond_destroy(&pool->not_full);

217 }

218

219 /// Test

src/daemon/libuv_workers.h

+2 -2

@@ -124,8 +124,8 @@ typedef struct {

124 int tail;

125 int count;

126

127 - uv_mutex_t lock;

128 - uv_cond_t not_full;

127 + netdata_mutex_t lock;

128 + netdata_cond_t not_full;

129 } CmdPool;

130

131

src/database/engine/datafile.c

+6 -6

@@ -3,13 +3,13 @@

4 void datafile_list_insert(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile)

5 {

6 - uv_rwlock_wrlock(&ctx->datafiles.rwlock);

6 + netdata_rwlock_wrlock(&ctx->datafiles.rwlock);

7 Pvoid_t *Pvalue = JudyLIns(&ctx->datafiles.JudyL, (Word_t ) datafile->fileno, PJE0);

8 if(!Pvalue || Pvalue == PJERR)

9 fatal("DBENGINE: cannot insert datafile %u of tier %d into the datafiles list",

10 datafile->fileno, ctx->config.tier);

11 *Pvalue = datafile;

12 - uv_rwlock_wrunlock(&ctx->datafiles.rwlock);

12 + netdata_rwlock_wrunlock(&ctx->datafiles.rwlock);

13 }

15 void datafile_list_delete_unsafe(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile)

@@ -26,7 +26,7 @@ static struct rrdengine_datafile *datafile_alloc_and_init(struct rrdengine_insta

27 datafile->tier = tier;

28 datafile->fileno = fileno;

29 - fatal_assert(0 == uv_rwlock_init(&datafile->extent_rwlock));

29 + fatal_assert(0 == netdata_rwlock_init(&datafile->extent_rwlock));

30 datafile->ctx = ctx;

31 datafile->magic1 = datafile->magic2 = DATAFILE_MAGIC;

@@ -656,13 +656,13 @@ void finalize_data_files(struct rrdengine_instance *ctx)

656 logged = false;

657 bool available = false;

658 do {

659 - uv_rwlock_wrlock(&ctx->datafiles.rwlock);

659 + netdata_rwlock_wrlock(&ctx->datafiles.rwlock);

660 spinlock_lock(&datafile->writers.spinlock);

661 available = (datafile->writers.running || datafile->writers.flushed_to_open_running) ? false : true;

662

663 if(!available) {

664 spinlock_unlock(&datafile->writers.spinlock);

665 - uv_rwlock_wrunlock(&ctx->datafiles.rwlock);

665 + netdata_rwlock_wrunlock(&ctx->datafiles.rwlock);

666 if(!logged) {

667 netdata_log_info("Waiting for writers to data file %u of tier %d to finish...", datafile->fileno, ctx->config.tier);

668 logged = true;

@@ -675,7 +675,7 @@ void finalize_data_files(struct rrdengine_instance *ctx)

675 close_data_file(datafile);

676 datafile_list_delete_unsafe(ctx, datafile);

677 spinlock_unlock(&datafile->writers.spinlock);

678 - uv_rwlock_wrunlock(&ctx->datafiles.rwlock);

678 + netdata_rwlock_wrunlock(&ctx->datafiles.rwlock);

679

680 // Clean up EPDL_EXTENT structures

681 cleanup_datafile_epdl_structures(datafile);

src/database/engine/datafile.h

+1 -1

@@ -56,7 +56,7 @@ struct rrdengine_datafile {

56 unsigned fileno;

57 uv_file file;

58 uint64_t pos;

59 - uv_rwlock_t extent_rwlock;

59 + netdata_rwlock_t extent_rwlock;

60 struct rrdengine_instance *ctx;

61 struct rrdengine_journalfile *journalfile;

src/database/engine/journalfile.c

+2 -2

@@ -293,7 +293,7 @@ void journalfile_v2_data_unmount_cleanup(time_t now_s) {

293 if(!ctx) continue;

294

295 struct rrdengine_datafile *datafile;

296 - if(uv_rwlock_tryrdlock(&ctx->datafiles.rwlock) != 0)

296 + if(netdata_rwlock_tryrdlock(&ctx->datafiles.rwlock) != 0)

297 continue;

298

299 bool first_then_next = true;

@@ -328,7 +328,7 @@ void journalfile_v2_data_unmount_cleanup(time_t now_s) {

328 if (unmount)

329 journalfile_v2_mounted_data_unmount(journalfile, false, false);

330 }

331 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

331 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

332 }

333 }

334

src/database/engine/rrdengine.c

+30 -30

@@ -703,12 +703,12 @@ size_t datafile_count(struct rrdengine_instance *ctx, bool with_lock)

703 size_t count = 0;

704

705 if (!with_lock)

706 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

706 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

707

708 count = JudyLCount(ctx->datafiles.JudyL, 0, -1, PJE0);

709

710 if (!with_lock)

711 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

711 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

712

713 return count;

714 }

@@ -723,7 +723,7 @@ get_next_datafile(struct rrdengine_datafile *this_datafile, struct rrdengine_ins

723 return NULL;

724

725 if (!with_lock)

726 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

726 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

727

728 Word_t Index = this_datafile ? this_datafile->fileno : 0;

729 Pvoid_t *Pvalue;

@@ -734,7 +734,7 @@ get_next_datafile(struct rrdengine_datafile *this_datafile, struct rrdengine_ins

734 datafile = *Pvalue;

735

736 if (!with_lock)

737 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

737 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

738

739 return datafile;

740 }

@@ -744,7 +744,7 @@ static struct rrdengine_datafile *get_ctx_datafile_first_or_last(struct rrdengin

744 struct rrdengine_datafile *datafile = NULL;

745

746 if (!with_lock)

747 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

747 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

748

749 Word_t Index = 0;

750 Pvoid_t *Pvalue;

@@ -760,7 +760,7 @@ static struct rrdengine_datafile *get_ctx_datafile_first_or_last(struct rrdengin

760 datafile = *Pvalue;

761

762 if (!with_lock)

763 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

763 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

764

765 return datafile;

766 }

@@ -787,14 +787,14 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_

787 struct rrdengine_datafile *datafile;

788

789 // get the latest datafile

790 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

790 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

791

792 datafile = get_last_ctx_datafile(ctx, true);

793 // become a writer on this datafile, to prevent it from vanishing

794 spinlock_lock(&datafile->writers.spinlock);

795 datafile->writers.running++;

796 spinlock_unlock(&datafile->writers.spinlock);

797 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

797 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

798

799 if(datafile_is_full(ctx, datafile)) {

800 // remember the datafile we have become writers to

@@ -813,13 +813,13 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_

813 netdata_mutex_unlock(&mutex);

814

815 // get the new latest datafile again, like above

816 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

816 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

817 datafile = get_last_ctx_datafile(ctx, true);

818 // become a writer on this datafile, to prevent it from vanishing

819 spinlock_lock(&datafile->writers.spinlock);

820 datafile->writers.running++;

821 spinlock_unlock(&datafile->writers.spinlock);

822 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

822 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

823

824 // release the writers on the old datafile

825 spinlock_lock(&old_datafile->writers.spinlock);

@@ -1038,14 +1038,14 @@ struct uuid_first_time_s {

1038

1039 struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile) {

1040

1041 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1041 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1042

1043 struct rrdengine_datafile *next_datafile = get_next_datafile(datafile, NULL, true);

1044

1045 while(next_datafile && !datafile_acquire(next_datafile, DATAFILE_ACQUIRE_RETENTION))

1046 next_datafile = get_next_datafile(next_datafile, NULL, true);

1047

1048 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1048 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1049

1050 datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);

1051

@@ -1061,11 +1061,11 @@ static time_t find_uuid_first_time(

1061 time_t global_first_time_s = LONG_MAX;

1062

1063 // acquire the datafile to work with it

1064 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1064 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1065 while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))

1066 datafile = get_next_datafile(datafile, NULL, true);

1067

1068 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1068 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1069

1070 if (unlikely(!datafile))

1071 return global_first_time_s;

@@ -1432,9 +1432,9 @@ void datafile_delete(

1432 int ret;

1433 char path[RRDENG_PATH_MAX];

1434

1435 - uv_rwlock_wrlock(&ctx->datafiles.rwlock);

1435 + netdata_rwlock_wrlock(&ctx->datafiles.rwlock);

1436 datafile_list_delete_unsafe(ctx, datafile);

1437 - uv_rwlock_wrunlock(&ctx->datafiles.rwlock);

1437 + netdata_rwlock_wrunlock(&ctx->datafiles.rwlock);

1438

1439 journal_file = datafile->journalfile;

1440 datafile_bytes = datafile->pos;

@@ -1570,7 +1570,7 @@ static void *populate_mrg_tp_worker(

1570 size_t thread_index = 0;

1571 int rc;

1572

1573 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1573 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1574

1575 size_t total_datafiles = 0;

1576 size_t populated_datafiles = 0;

@@ -1580,7 +1580,7 @@ static void *populate_mrg_tp_worker(

1580 if (df->populate_mrg.populated)

1581 populated_datafiles++;

1582 }

1583 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1583 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1584

1585 if (total_datafiles == 0) {

1586 nd_log_daemon(NDLP_WARNING, "DBENGINE: No datafiles to populate MRG");

@@ -1592,7 +1592,7 @@ static void *populate_mrg_tp_worker(

1592 struct rrdengine_datafile *datafile = NULL;

1593

1594 // find a datafile to work on

1595 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1595 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1596 bool first_then_next = true;

1597 Pvoid_t *Pvalue = NULL;

1598 Word_t Index = 0;

@@ -1610,7 +1610,7 @@ static void *populate_mrg_tp_worker(

1610 }

1611 break;

1612 }

1613 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1613 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1614

1615 if(!datafile)

1616 break;

@@ -1794,7 +1794,7 @@ time_t get_datafile_end_time(struct rrdengine_instance *ctx)

1794 {

1795 time_t last_time_s = 0;

1796

1797 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1797 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1798 struct rrdengine_datafile *datafile = get_last_ctx_datafile(ctx, true);

1799

1800 if (datafile) {

@@ -1803,7 +1803,7 @@ time_t get_datafile_end_time(struct rrdengine_instance *ctx)

1803 last_time_s = datafile->journalfile->v2.first_time_s;

1804 }

1805

1806 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1806 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1807 return last_time_s;

1808 }

1809

@@ -1859,7 +1859,7 @@ static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(

1859 {

1860 struct rrdengine_datafile *datafile = NULL;

1861

1862 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1862 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1863 if (release_datafile) {

1864 datafile = get_next_datafile(release_datafile, NULL, true);

1865 datafile_release(release_datafile, DATAFILE_ACQUIRE_INDEXING);

@@ -1882,13 +1882,13 @@ static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(

1882 sleep_usec(200 * USEC_PER_MS);

1883 }

1884 if (locked) {

1885 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1885 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1886 return datafile;

1887 }

1888 nd_log_daemon(NDLP_INFO, "DBENGINE: Datafile %u CANNOT be locked for indexing after retries; skipping", datafile->fileno);

1889 datafile = get_next_datafile(datafile, NULL, true);

1890 }

1891 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1891 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1892 return NULL;

1893 }

1894

@@ -2031,7 +2031,7 @@ uint64_t rrdeng_get_used_disk_space(struct rrdengine_instance *ctx, bool having_

2031 uint64_t active_space = 0;

2032

2033 if (!having_lock)

2034 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

2034 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

2035

2036 struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);

2037 struct rrdengine_datafile *last_datafile = get_last_ctx_datafile(ctx, true);

@@ -2040,7 +2040,7 @@ uint64_t rrdeng_get_used_disk_space(struct rrdengine_instance *ctx, bool having_

2040 active_space = last_datafile->pos;

2041

2042 if (!having_lock)

2043 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

2043 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

2044

2045 // calculate the estimated disk space based on the expected final size of the datafile

2046 // We cant know the final v1/v2 journal size -- we let the current v1 size be part of the calculation by not

@@ -2071,17 +2071,17 @@ static time_t get_tier_retention(struct rrdengine_instance *ctx)

2071 // Check if disk or retention time cap reached

2072 bool rrdeng_ctx_tier_cap_exceeded(struct rrdengine_instance *ctx)

2073 {

2074 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

2074 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

2075 struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);

2076

2077 if (!first_datafile || datafile_count(ctx, true) < 2) {

2078 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

2078 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

2079 return false;

2080 }

2081

2082 uint64_t estimated_disk_space = rrdeng_get_used_disk_space(ctx, true);

2083

2084 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

2084 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

2085

2086 if (ctx->config.max_retention_s) {

2087 time_t retention = get_tier_retention(ctx);

src/database/engine/rrdengine.h

+1 -1

@@ -397,7 +397,7 @@ struct rrdengine_instance {

397 TIER_CONFIG_PROTOTYPE config;

398

399 struct {

400 - uv_rwlock_t rwlock; // the JudyL of datafiles is protected by this lock

400 + netdata_rwlock_t rwlock; // the JudyL of datafiles is protected by this lock

401 bool disk_time; // true: delete for disk quota, false: delete for retention

402 bool pending_rotate; // Change from event loop

403 bool pending_index; // Change from event loop

src/database/engine/rrdengineapi.c

+3 -3

@@ -44,7 +44,7 @@ size_t page_type_size[256] = {

45 static inline void initialize_single_ctx(struct rrdengine_instance *ctx) {

46 memset(ctx, 0, sizeof(*ctx));

47 - uv_rwlock_init(&ctx->datafiles.rwlock);

47 + netdata_rwlock_init(&ctx->datafiles.rwlock);

48 rw_spinlock_init(&ctx->njfv2idx.spinlock);

49 }

@@ -1348,14 +1348,14 @@ static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_S

1348 RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {

1349 RRDENG_SIZE_STATS stats = { 0 };

1350

1351 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1351 + netdata_rwlock_rdlock(&ctx->datafiles.rwlock);

1352 struct rrdengine_datafile *df = NULL;

1353

1354 while ((df = get_next_datafile(df, ctx, true))) {

1355 stats.datafiles++;

1356 populate_v2_statistics(df, &stats);

1357 }

1358 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1358 + netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);

1359

1360 stats.currently_collected_metrics = __atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED);

1361

src/database/rrdfunctions-inflight.c

+4 -12

@@ -290,24 +290,16 @@ static int rrd_call_function_async_and_wait(struct rrd_function_inflight *r) {

290 usec_t now_mono_ut = now_monotonic_usec();

291 usec_t stop_mono_ut = __atomic_load_n(&r->stop_monotonic_ut, __ATOMIC_RELAXED) + RRDFUNCTIONS_TIMEOUT_EXTENSION_UT;

292 if(now_mono_ut > stop_mono_ut) {

293 - rc = ETIMEDOUT;

293 + rc = UV_ETIMEDOUT;

294 break;

295 }

296

297 // wait for 10ms, and loop again...

298 - struct timespec tp;

299 - clock_gettime(CLOCK_REALTIME, &tp);

300 - tp.tv_nsec += 10 * NSEC_PER_MSEC;

301 - if(tp.tv_nsec > (long)(1 * NSEC_PER_SEC)) {

302 - tp.tv_sec++;

303 - tp.tv_nsec -= 1 * NSEC_PER_SEC;

304 - }

305 -

298 // the mutex is unlocked within cond_timedwait()

307 - rc = netdata_cond_timedwait(&tmp->cond, &tmp->mutex, &tp);

299 + rc = netdata_cond_timedwait(&tmp->cond, &tmp->mutex, 10 * NSEC_PER_MSEC);

300 // the mutex is again ours

301

310 - if(rc == ETIMEDOUT) {

302 + if(rc == UV_ETIMEDOUT) {

303 // 10ms have passed

304

305 rc = 0;

@@ -339,7 +331,7 @@ static int rrd_call_function_async_and_wait(struct rrd_function_inflight *r) {

331 tmp->free_with_signal = false;

332 we_should_free = true;

333 }

342 - else if (rc == ETIMEDOUT || cancelled) {

334 + else if (rc == UV_ETIMEDOUT || cancelled) {

335 // timeout

336 // we will go away and let the callback free the structure

337

src/exporting/aws_kinesis/aws_kinesis.c

+6 -6

@@ -58,9 +58,9 @@ int init_aws_kinesis_instance(struct instance *instance)

58 instance->config.name);

59 return 1;

60 }

61 - if (uv_mutex_init(&instance->mutex))

61 + if (netdata_mutex_init(&instance->mutex))

62 return 1;

63 - if (uv_cond_init(&instance->cond_var))

63 + if (netdata_cond_init(&instance->cond_var))

64 return 1;

66 if (!instance->engine->aws_sdk_initialized) {

@@ -104,13 +104,13 @@ void *aws_kinesis_connector_worker(void *instance_p)

104 unsigned long long partition_key_seq = 0;

105 struct stats *stats = &instance->stats;

106

107 - uv_mutex_lock(&instance->mutex);

107 + netdata_mutex_lock(&instance->mutex);

108 while (!instance->data_is_ready)

109 - uv_cond_wait(&instance->cond_var, &instance->mutex);

109 + netdata_cond_wait(&instance->cond_var, &instance->mutex);

110 instance->data_is_ready = 0;

111

112 if (unlikely(instance->engine->exit)) {

113 - uv_mutex_unlock(&instance->mutex);

113 + netdata_mutex_unlock(&instance->mutex);

114 break;

115 }

116

@@ -208,7 +208,7 @@ void *aws_kinesis_connector_worker(void *instance_p)

208

209 stats->buffered_metrics = 0;

210

211 - uv_mutex_unlock(&instance->mutex);

211 + netdata_mutex_unlock(&instance->mutex);

212

213 #ifdef UNIT_TESTING

214 return;

src/exporting/clean_connectors.c

+1 -2

@@ -35,8 +35,7 @@ void clean_instance(struct instance *instance)

35 clean_instance_config(&instance->config);

36 buffer_free(instance->labels_buffer);

38 - uv_cond_destroy(&instance->cond_var);

39 - // uv_mutex_destroy(&instance->mutex);

38 + netdata_cond_destroy(&instance->cond_var);

39 }

41 /**

src/exporting/exporting_engine.c

+3 -3

@@ -140,10 +140,10 @@ static void exporting_main_cleanup(void *pptr)

140 if (!instance->exited) {

141 netdata_log_info("EXPORTING: signaling worker '%s' to stop...", instance->config.name);

142 // Lock the mutex before signaling the condition variable

143 - uv_mutex_lock(&instance->mutex);

143 + netdata_mutex_lock(&instance->mutex);

144 instance->data_is_ready = 1;

145 - uv_cond_signal(&instance->cond_var);

146 - uv_mutex_unlock(&instance->mutex);

145 + netdata_cond_signal(&instance->cond_var);

146 + netdata_mutex_unlock(&instance->mutex);

147 }

148 else

149 netdata_log_info("EXPORTING: found worker '%s' already stopped", instance->config.name);

src/exporting/exporting_engine.h

+3 -3

@@ -210,8 +210,8 @@ struct instance {

210 time_t before;

211

212 ND_THREAD *thread;

213 - uv_mutex_t mutex;

214 - uv_cond_t cond_var;

213 + netdata_mutex_t mutex;

214 + netdata_cond_t cond_var;

215 int data_is_ready;

216

217 int (*start_batch_formatting)(struct instance *instance);

@@ -312,7 +312,7 @@ static inline void disable_instance(struct instance *instance)

312 {

313 instance->disabled = 1;

314 instance->scheduled = 0;

315 - uv_mutex_unlock(&instance->mutex);

315 + netdata_mutex_unlock(&instance->mutex);

316 netdata_log_error("EXPORTING: Instance %s disabled", instance->config.name);

317 }

318

src/exporting/graphite/graphite.c

+2 -2

@@ -53,9 +53,9 @@ int init_graphite_instance(struct instance *instance)

54 simple_connector_init(instance);

56 - if (uv_mutex_init(&instance->mutex))

56 + if (netdata_mutex_init(&instance->mutex))

57 return 1;

58 - if (uv_cond_init(&instance->cond_var))

58 + if (netdata_cond_init(&instance->cond_var))

59 return 1;

61 return 0;

src/exporting/json/json.c

+4 -4

@@ -45,9 +45,9 @@ int init_json_instance(struct instance *instance)

46 simple_connector_init(instance);

48 - if (uv_mutex_init(&instance->mutex))

48 + if (netdata_mutex_init(&instance->mutex))

49 return 1;

50 - if (uv_cond_init(&instance->cond_var))

50 + if (netdata_cond_init(&instance->cond_var))

51 return 1;

53 return 0;

@@ -97,9 +97,9 @@ int init_json_http_instance(struct instance *instance)

98 simple_connector_init(instance);

100 - if (uv_mutex_init(&instance->mutex))

100 + if (netdata_mutex_init(&instance->mutex))

101 return 1;

102 - if (uv_cond_init(&instance->cond_var))

102 + if (netdata_cond_init(&instance->cond_var))

103 return 1;

104

105 return 0;

src/exporting/mongodb/mongodb.c

+8 -8

@@ -113,9 +113,9 @@ int init_mongodb_instance(struct instance *instance)

113 instance->config.name);

114 return 1;

115 }

116 - if (uv_mutex_init(&instance->mutex))

116 + if (netdata_mutex_init(&instance->mutex))

117 return 1;

118 - if (uv_cond_init(&instance->cond_var))

118 + if (netdata_cond_init(&instance->cond_var))

119 return 1;

120

121 struct mongodb_specific_data *connector_specific_data = callocz(1, sizeof(struct mongodb_specific_data));

@@ -288,16 +288,16 @@ void mongodb_connector_worker(void *instance_p)

288 while (!instance->engine->exit) {

289 struct stats *stats = &instance->stats;

290

291 - uv_mutex_lock(&instance->mutex);

291 + netdata_mutex_lock(&instance->mutex);

292 if (!connector_specific_data->first_buffer->insert ||

293 !connector_specific_data->first_buffer->documents_inserted) {

294 while (!instance->data_is_ready)

295 - uv_cond_wait(&instance->cond_var, &instance->mutex);

295 + netdata_cond_wait(&instance->cond_var, &instance->mutex);

296 instance->data_is_ready = 0;

297 }

298

299 if (unlikely(instance->engine->exit)) {

300 - uv_mutex_unlock(&instance->mutex);

300 + netdata_mutex_unlock(&instance->mutex);

301 break;

302 }

303

@@ -322,7 +322,7 @@ void mongodb_connector_worker(void *instance_p)

322 connector_specific_data->first_buffer->buffered_bytes = 0;

323 connector_specific_data->first_buffer = connector_specific_data->first_buffer->next;

324

325 - uv_mutex_unlock(&instance->mutex);

325 + netdata_mutex_unlock(&instance->mutex);

326

327 size_t data_size = 0;

328 for (size_t i = 0; i < documents_inserted; i++) {

@@ -370,7 +370,7 @@ void mongodb_connector_worker(void *instance_p)

370 if (unlikely(instance->engine->exit))

371 break;

372

373 - uv_mutex_lock(&instance->mutex);

373 + netdata_mutex_lock(&instance->mutex);

374

375 stats->buffered_metrics = connector_specific_data->total_documents_inserted;

376

@@ -381,7 +381,7 @@ void mongodb_connector_worker(void *instance_p)

381 stats->buffered_metrics = 0;

382 stats->buffered_bytes -= buffered_bytes;

383

384 - uv_mutex_unlock(&instance->mutex);

384 + netdata_mutex_unlock(&instance->mutex);

385

386 #ifdef UNIT_TESTING

387 return;

src/exporting/opentsdb/opentsdb.c

+4 -4

@@ -50,9 +50,9 @@ int init_opentsdb_telnet_instance(struct instance *instance)

51 simple_connector_init(instance);

53 - if (uv_mutex_init(&instance->mutex))

53 + if (netdata_mutex_init(&instance->mutex))

54 return 1;

55 - if (uv_cond_init(&instance->cond_var))

55 + if (netdata_cond_init(&instance->cond_var))

56 return 1;

58 return 0;

@@ -104,9 +104,9 @@ int init_opentsdb_http_instance(struct instance *instance)

104

105 simple_connector_init(instance);

106

107 - if (uv_mutex_init(&instance->mutex))

107 + if (netdata_mutex_init(&instance->mutex))

108 return 1;

109 - if (uv_cond_init(&instance->cond_var))

109 + if (netdata_cond_init(&instance->cond_var))

110 return 1;

111

112 return 0;

src/exporting/process_data.c

+3 -3

@@ -168,7 +168,7 @@ void start_batch_formatting(struct engine *engine)

168 {

169 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {

170 if (instance->scheduled) {

171 - uv_mutex_lock(&instance->mutex);

171 + netdata_mutex_lock(&instance->mutex);

172 if (instance->start_batch_formatting && instance->start_batch_formatting(instance) != 0) {

173 netdata_log_error("EXPORTING: cannot start batch formatting for %s", instance->config.name);

174 disable_instance(instance);

@@ -317,8 +317,8 @@ void end_batch_formatting(struct engine *engine)

317 continue;

318 }

319 instance->data_is_ready = 1;

320 - uv_cond_signal(&instance->cond_var);

321 - uv_mutex_unlock(&instance->mutex);

320 + netdata_cond_signal(&instance->cond_var);

321 + netdata_mutex_unlock(&instance->mutex);

322

323 instance->scheduled = 0;

324 instance->after = instance->before;

src/exporting/prometheus/remote_write/remote_write.c

+2 -2

@@ -106,9 +106,9 @@ int init_prometheus_remote_write_instance(struct instance *instance)

106

107 instance->buffer = (void *)buffer_create(0, &netdata_buffers_statistics.buffers_exporters);

108

109 - if (uv_mutex_init(&instance->mutex))

109 + if (netdata_mutex_init(&instance->mutex))

110 return 1;

111 - if (uv_cond_init(&instance->cond_var))

111 + if (netdata_cond_init(&instance->cond_var))

112 return 1;

113

114 struct simple_connector_data *simple_connector_data = callocz(1, sizeof(struct simple_connector_data));

src/exporting/pubsub/pubsub.c

+6 -6

@@ -35,8 +35,8 @@ int init_pubsub_instance(struct instance *instance)

35 netdata_log_error("EXPORTING: cannot create buffer for Pub/Sub exporting connector instance %s", instance->config.name);

36 return 1;

37 }

38 - uv_mutex_init(&instance->mutex);

39 - uv_cond_init(&instance->cond_var);

38 + netdata_mutex_init(&instance->mutex);

39 + netdata_cond_init(&instance->cond_var);

41 struct pubsub_specific_data *connector_specific_data = callocz(1, sizeof(struct pubsub_specific_data));

42 instance->connector_specific_data = (void *)connector_specific_data;

@@ -103,14 +103,14 @@ void *pubsub_connector_worker(void *instance_p)

103 struct stats *stats = &instance->stats;

104 char error_message[ERROR_LINE_MAX + 1] = "";

105

106 - uv_mutex_lock(&instance->mutex);

106 + netdata_mutex_lock(&instance->mutex);

107 while (!instance->data_is_ready)

108 - uv_cond_wait(&instance->cond_var, &instance->mutex);

108 + netdata_cond_wait(&instance->cond_var, &instance->mutex);

109 instance->data_is_ready = 0;

110

111

112 if (unlikely(instance->engine->exit)) {

113 - uv_mutex_unlock(&instance->mutex);

113 + netdata_mutex_unlock(&instance->mutex);

114 break;

115 }

116

@@ -184,7 +184,7 @@ void *pubsub_connector_worker(void *instance_p)

184 buffer_flush(buffer);

185 stats->buffered_metrics = 0;

186

187 - uv_mutex_unlock(&instance->mutex);

187 + netdata_mutex_unlock(&instance->mutex);

188

189 #ifdef UNIT_TESTING

190 return;

src/exporting/send_data.c

+6 -6

@@ -231,16 +231,16 @@ void simple_connector_worker(void *instance_p)

231 if (instance->data_is_ready)

232 send_stats = 1;

233

234 - uv_mutex_lock(&instance->mutex);

234 + netdata_mutex_lock(&instance->mutex);

235 if (!connector_specific_data->first_buffer->used || failures) {

236 while (!instance->data_is_ready)

237 - uv_cond_wait(&instance->cond_var, &instance->mutex);

237 + netdata_cond_wait(&instance->cond_var, &instance->mutex);

238 instance->data_is_ready = 0;

239 send_stats = 1;

240 }

241

242 if (unlikely(instance->engine->exit)) {

243 - uv_mutex_unlock(&instance->mutex);

243 + netdata_mutex_unlock(&instance->mutex);

244 break;

245 }

246

@@ -272,7 +272,7 @@ void simple_connector_worker(void *instance_p)

272 buffered_metrics = connector_specific_data->buffered_metrics;

273 }

274

275 - uv_mutex_unlock(&instance->mutex);

275 + netdata_mutex_unlock(&instance->mutex);

276

277 // ------------------------------------------------------------------------

278 // if we are connected, receive a response, without blocking

@@ -354,7 +354,7 @@ void simple_connector_worker(void *instance_p)

354 break;

355

356 if (send_stats) {

357 - uv_mutex_lock(&instance->mutex);

357 + netdata_mutex_lock(&instance->mutex);

358

359 stats->buffered_metrics = connector_specific_data->total_buffered_metrics;

360

@@ -376,7 +376,7 @@ void simple_connector_worker(void *instance_p)

376 stats->lost_metrics =

377 stats->lost_bytes = 0;

378

379 - uv_mutex_unlock(&instance->mutex);

379 + netdata_mutex_unlock(&instance->mutex);

380 }

381

382 #ifdef UNIT_TESTING

src/libnetdata/completion/completion.c

+24 -24

@@ -14,24 +14,24 @@ ALWAYS_INLINE void completion_init(struct completion *p)

14 {

15 p->completed = 0;

16 p->completed_jobs = 0;

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

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

17 + fatal_assert(0 == netdata_cond_init(&p->cond));

18 + fatal_assert(0 == netdata_mutex_init(&p->mutex));

19 }

21 ALWAYS_INLINE void completion_destroy(struct completion *p)

22 {

23 - uv_cond_destroy(&p->cond);

24 - uv_mutex_destroy(&p->mutex);

23 + netdata_cond_destroy(&p->cond);

24 + netdata_mutex_destroy(&p->mutex);

25 }

27 ALWAYS_INLINE void completion_wait_for(struct completion *p)

28 {

29 - uv_mutex_lock(&p->mutex);

29 + netdata_mutex_lock(&p->mutex);

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

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

31 + netdata_cond_wait(&p->cond, &p->mutex);

32 }

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

34 - uv_mutex_unlock(&p->mutex);

34 + netdata_mutex_unlock(&p->mutex);

35 }

37 ALWAYS_INLINE bool completion_timedwait_for(struct completion *p, uint64_t timeout_s)

@@ -41,9 +41,9 @@ ALWAYS_INLINE bool completion_timedwait_for(struct completion *p, uint64_t timeo

41 uint64_t start_time = uv_hrtime();

42 bool result = true;

44 - uv_mutex_lock(&p->mutex);

44 + netdata_mutex_lock(&p->mutex);

45 while (!p->completed) {

46 - int rc = uv_cond_timedwait(&p->cond, &p->mutex, timeout_s);

46 + int rc = netdata_cond_timedwait(&p->cond, &p->mutex, timeout_s);

48 if (rc == 0) {

49 result = true;

@@ -64,27 +64,27 @@ ALWAYS_INLINE bool completion_timedwait_for(struct completion *p, uint64_t timeo

64 }

65 timeout_s -= elapsed;

66 }

67 - uv_mutex_unlock(&p->mutex);

67 + netdata_mutex_unlock(&p->mutex);

69 return result;

70 }

72 ALWAYS_INLINE void completion_mark_complete(struct completion *p)

73 {

74 - uv_mutex_lock(&p->mutex);

74 + netdata_mutex_lock(&p->mutex);

75 p->completed = 1;

76 - uv_cond_broadcast(&p->cond);

77 - uv_mutex_unlock(&p->mutex);

76 + netdata_cond_broadcast(&p->cond);

77 + netdata_mutex_unlock(&p->mutex);

78 }

80 ALWAYS_INLINE unsigned completion_wait_for_a_job(struct completion *p, unsigned completed_jobs)

81 {

82 - uv_mutex_lock(&p->mutex);

82 + netdata_mutex_lock(&p->mutex);

83 while (0 == p->completed && p->completed_jobs <= completed_jobs) {

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

84 + netdata_cond_wait(&p->cond, &p->mutex);

85 }

86 completed_jobs = p->completed_jobs;

87 - uv_mutex_unlock(&p->mutex);

87 + netdata_mutex_unlock(&p->mutex);

89 return completed_jobs;

90 }

@@ -96,7 +96,7 @@ ALWAYS_INLINE unsigned completion_wait_for_a_job_with_timeout(struct completion

97 uint64_t deadline_ns = uv_hrtime() + timeout_ns;

99 - uv_mutex_lock(&p->mutex);

99 + netdata_mutex_lock(&p->mutex);

100

101 while (p->completed == 0 && p->completed_jobs <= completed_jobs) {

102 uint64_t current_time_ns = uv_hrtime();

@@ -108,30 +108,30 @@ ALWAYS_INLINE unsigned completion_wait_for_a_job_with_timeout(struct completion

108

109 uint64_t remaining_timeout_ns = deadline_ns - current_time_ns;

110

111 - int rc = uv_cond_timedwait(&p->cond, &p->mutex, remaining_timeout_ns);

111 + int rc = netdata_cond_timedwait(&p->cond, &p->mutex, remaining_timeout_ns);

112 if (rc == UV_ETIMEDOUT)

113 break;

114 }

115

116 completed_jobs = p->completed_jobs;

117 - uv_mutex_unlock(&p->mutex);

117 + netdata_mutex_unlock(&p->mutex);

118

119 return completed_jobs;

120 }

121

122 ALWAYS_INLINE void completion_mark_complete_a_job(struct completion *p)

123 {

124 - uv_mutex_lock(&p->mutex);

124 + netdata_mutex_lock(&p->mutex);

125 p->completed_jobs++;

126 - uv_cond_broadcast(&p->cond);

127 - uv_mutex_unlock(&p->mutex);

126 + netdata_cond_broadcast(&p->cond);

127 + netdata_mutex_unlock(&p->mutex);

128 }

129

130 ALWAYS_INLINE bool completion_is_done(struct completion *p)

131 {

132 bool ret;

133 - uv_mutex_lock(&p->mutex);

133 + netdata_mutex_lock(&p->mutex);

134 ret = p->completed;

135 - uv_mutex_unlock(&p->mutex);

135 + netdata_mutex_unlock(&p->mutex);

136 return ret;

137 }

src/libnetdata/completion/completion.h

+2 -2

@@ -6,8 +6,8 @@

6 #include "../libnetdata.h"

8 struct completion {

9 - uv_mutex_t mutex;

10 - uv_cond_t cond;

9 + netdata_mutex_t mutex;

10 + netdata_cond_t cond;

11 volatile unsigned completed;

12 volatile unsigned completed_jobs;

13 };

src/libnetdata/locks/locks.c

+40 -82

@@ -22,67 +22,53 @@

22 // mutex

24 ALWAYS_INLINE int __netdata_cond_init(netdata_cond_t *cond) {

25 - int ret = pthread_cond_init(cond, NULL);

25 + int ret = uv_cond_init(cond);

26 if(unlikely(ret != 0))

27 netdata_log_error("COND: failed to initialize (code %d).", ret);

28 return ret;

29 }

31 -ALWAYS_INLINE int __netdata_cond_destroy(netdata_cond_t *cond) {

32 - int ret = pthread_cond_destroy(cond);

33 - if(unlikely(ret != 0))

34 - netdata_log_error("COND: failed to destroy (code %d).", ret);

35 - return ret;

31 +ALWAYS_INLINE void __netdata_cond_destroy(netdata_cond_t *cond) {

32 + uv_cond_destroy(cond);

33 }

38 -ALWAYS_INLINE int __netdata_cond_signal(netdata_cond_t *cond) {

39 - int ret = pthread_cond_signal(cond);

40 - if(unlikely(ret != 0))

41 - netdata_log_error("COND: failed to signal (code %d).", ret);

42 - return ret;

35 +ALWAYS_INLINE void __netdata_cond_signal(netdata_cond_t *cond) {

36 + uv_cond_signal(cond);

37 +}

38 +

39 +ALWAYS_INLINE void __netdata_cond_broadcast(netdata_cond_t *cond) {

40 + uv_cond_broadcast(cond);

41 }

45 -ALWAYS_INLINE int __netdata_cond_wait(netdata_cond_t *cond, netdata_mutex_t *mutex)

43 +ALWAYS_INLINE void __netdata_cond_wait(netdata_cond_t *cond, netdata_mutex_t *mutex)

44 {

47 - int ret = pthread_cond_wait(cond, mutex);

48 - if (unlikely(ret != 0))

49 - netdata_log_error("COND: failed to signal (code %d).", ret);

50 - return ret;

45 + uv_cond_wait(cond, mutex);

46 }

53 -ALWAYS_INLINE int __netdata_cond_timedwait(netdata_cond_t *cond, netdata_mutex_t *mutex, struct timespec *tp)

48 +ALWAYS_INLINE int __netdata_cond_timedwait(netdata_cond_t *cond, netdata_mutex_t *mutex, uint64_t timeout_ns)

49 {

55 - int ret = pthread_cond_timedwait(cond, mutex, tp);

50 + int ret = uv_cond_timedwait(cond, mutex, timeout_ns);

51 return ret;

52 }

54 ALWAYS_INLINE int __netdata_mutex_init(netdata_mutex_t *mutex) {

60 - int ret = pthread_mutex_init(mutex, NULL);

55 + int ret = uv_mutex_init(mutex);

56 if(unlikely(ret != 0))

57 netdata_log_error("MUTEX_LOCK: failed to initialize (code %d).", ret);

58 return ret;

59 }

66 -ALWAYS_INLINE int __netdata_mutex_destroy(netdata_mutex_t *mutex) {

67 - int ret = pthread_mutex_destroy(mutex);

68 - if(unlikely(ret != 0))

69 - netdata_log_error("MUTEX_LOCK: failed to destroy (code %d).", ret);

70 - return ret;

61 +ALWAYS_INLINE void __netdata_mutex_destroy(netdata_mutex_t *mutex) {

62 + uv_mutex_destroy(mutex);

63 }

73 -ALWAYS_INLINE int __netdata_mutex_lock(netdata_mutex_t *mutex) {

74 - int ret = pthread_mutex_lock(mutex);

75 - if(unlikely(ret != 0)) {

76 - netdata_log_error("MUTEX_LOCK: failed to get lock (code %d)", ret);

77 - }

78 - else

79 - nd_thread_mutex_locked();

80 -

81 - return ret;

65 +ALWAYS_INLINE void __netdata_mutex_lock(netdata_mutex_t *mutex) {

66 + uv_mutex_lock(mutex);

67 + nd_thread_mutex_locked();

68 }

70 ALWAYS_INLINE int __netdata_mutex_trylock(netdata_mutex_t *mutex) {

85 - int ret = pthread_mutex_trylock(mutex);

71 + int ret = uv_mutex_trylock(mutex);

72 if(ret != 0)

73 ;

74 else

@@ -91,14 +77,9 @@ ALWAYS_INLINE int __netdata_mutex_trylock(netdata_mutex_t *mutex) {

77 return ret;

78 }

94 -ALWAYS_INLINE int __netdata_mutex_unlock(netdata_mutex_t *mutex) {

95 - int ret = pthread_mutex_unlock(mutex);

96 - if(unlikely(ret != 0))

97 - netdata_log_error("MUTEX_LOCK: failed to unlock (code %d).", ret);

98 - else

99 - nd_thread_mutex_unlocked();

100 -

101 - return ret;

80 +ALWAYS_INLINE void __netdata_mutex_unlock(netdata_mutex_t *mutex) {

81 + uv_mutex_unlock(mutex);

82 + nd_thread_mutex_unlocked();

83 }

85 #ifdef NETDATA_TRACE_RWLOCKS

@@ -181,62 +162,39 @@ int netdata_mutex_unlock_debug(const char *file __maybe_unused, const char *func

162 // ----------------------------------------------------------------------------

163 // rwlock

164

184 -ALWAYS_INLINE int __netdata_rwlock_destroy(netdata_rwlock_t *rwlock) {

185 - int ret = pthread_rwlock_destroy(&rwlock->rwlock_t);

186 - if(unlikely(ret != 0))

187 - netdata_log_error("RW_LOCK: failed to destroy lock (code %d)", ret);

188 - return ret;

165 +ALWAYS_INLINE void __netdata_rwlock_destroy(netdata_rwlock_t *rwlock) {

166 + uv_rwlock_destroy(&rwlock->rwlock_t);

167 }

168

169 ALWAYS_INLINE int __netdata_rwlock_init(netdata_rwlock_t *rwlock) {

192 - int ret = pthread_rwlock_init(&rwlock->rwlock_t, NULL);

170 + int ret = uv_rwlock_init(&rwlock->rwlock_t);

171 if(unlikely(ret != 0))

172 netdata_log_error("RW_LOCK: failed to initialize lock (code %d)", ret);

173 return ret;

174 }

175

198 -ALWAYS_INLINE int __netdata_rwlock_rdlock(netdata_rwlock_t *rwlock) {

199 - int ret = pthread_rwlock_rdlock(&rwlock->rwlock_t);

200 - if(unlikely(ret != 0))

201 - netdata_log_error("RW_LOCK: failed to obtain read lock (code %d)", ret);

202 - else

203 - nd_thread_rwlock_read_locked();

204 -

205 - return ret;

176 +ALWAYS_INLINE void __netdata_rwlock_rdlock(netdata_rwlock_t *rwlock) {

177 + uv_rwlock_rdlock(&rwlock->rwlock_t);

178 + nd_thread_rwlock_read_locked();

179 }

180

208 -ALWAYS_INLINE int __netdata_rwlock_wrlock(netdata_rwlock_t *rwlock) {

209 - int ret = pthread_rwlock_wrlock(&rwlock->rwlock_t);

210 - if(unlikely(ret != 0))

211 - netdata_log_error("RW_LOCK: failed to obtain write lock (code %d)", ret);

212 - else

213 - nd_thread_rwlock_write_locked();

214 -

215 - return ret;

181 +ALWAYS_INLINE void __netdata_rwlock_wrlock(netdata_rwlock_t *rwlock) {

182 + uv_rwlock_wrlock(&rwlock->rwlock_t);

183 + nd_thread_rwlock_write_locked();

184 }

185

218 -ALWAYS_INLINE int __netdata_rwlock_rdunlock(netdata_rwlock_t *rwlock) {

219 - int ret = pthread_rwlock_unlock(&rwlock->rwlock_t);

220 - if(unlikely(ret != 0))

221 - netdata_log_error("RW_LOCK: failed to release lock (code %d)", ret);

222 - else

223 - nd_thread_rwlock_read_unlocked();

224 -

225 - return ret;

186 +ALWAYS_INLINE void __netdata_rwlock_rdunlock(netdata_rwlock_t *rwlock) {

187 + uv_rwlock_rdunlock(&rwlock->rwlock_t);

188 + nd_thread_rwlock_read_unlocked();

189 }

190

228 -ALWAYS_INLINE int __netdata_rwlock_wrunlock(netdata_rwlock_t *rwlock) {

229 - int ret = pthread_rwlock_unlock(&rwlock->rwlock_t);

230 - if(unlikely(ret != 0))

231 - netdata_log_error("RW_LOCK: failed to release lock (code %d)", ret);

232 - else

233 - nd_thread_rwlock_write_unlocked();

234 -

235 - return ret;

191 +ALWAYS_INLINE void __netdata_rwlock_wrunlock(netdata_rwlock_t *rwlock) {

192 + uv_rwlock_wrunlock(&rwlock->rwlock_t);

193 + nd_thread_rwlock_write_unlocked();

194 }

195

196 ALWAYS_INLINE int __netdata_rwlock_tryrdlock(netdata_rwlock_t *rwlock) {

239 - int ret = pthread_rwlock_tryrdlock(&rwlock->rwlock_t);

197 + int ret = uv_rwlock_tryrdlock(&rwlock->rwlock_t);

198 if(ret != 0)

199 ;

200 else

@@ -246,7 +204,7 @@ ALWAYS_INLINE int __netdata_rwlock_tryrdlock(netdata_rwlock_t *rwlock) {

204 }

205

206 ALWAYS_INLINE int __netdata_rwlock_trywrlock(netdata_rwlock_t *rwlock) {

249 - int ret = pthread_rwlock_trywrlock(&rwlock->rwlock_t);

207 + int ret = uv_rwlock_trywrlock(&rwlock->rwlock_t);

208 if(ret != 0)

209 ;

210 else

src/libnetdata/locks/locks.h

+18 -16

@@ -10,8 +10,8 @@

10 // #define SPINLOCK_IMPL_WITH_MUTEX

11 // #endif

13 -typedef pthread_mutex_t netdata_mutex_t;

14 -typedef pthread_cond_t netdata_cond_t;

13 +typedef uv_mutex_t netdata_mutex_t;

14 +typedef uv_cond_t netdata_cond_t;

16 #ifdef NETDATA_TRACE_RWLOCKS

@@ -35,7 +35,7 @@ typedef struct netdata_rwlock_locker {

35 } netdata_rwlock_locker;

37 typedef struct netdata_rwlock_t {

38 - pthread_rwlock_t rwlock_t; // the lock

38 + uv_rwlock_t rwlock_t; // the lock

39 size_t readers; // the number of reader on the lock

40 size_t writers; // the number of writers on the lock

41 netdata_mutex_t lockers_mutex; // a mutex to protect the linked list of the lock holding threads

@@ -55,31 +55,32 @@ typedef struct netdata_rwlock_t {

55 #else // NETDATA_TRACE_RWLOCKS

57 typedef struct netdata_rwlock_t {

58 - pthread_rwlock_t rwlock_t;

58 + uv_rwlock_t rwlock_t;

59 } netdata_rwlock_t;

61 #endif // NETDATA_TRACE_RWLOCKS

63 int __netdata_cond_init(netdata_cond_t *cond);

64 -int __netdata_cond_destroy(netdata_cond_t *cond);

65 -int __netdata_cond_signal(netdata_cond_t *cond);

66 -int __netdata_cond_wait(netdata_cond_t *cond, netdata_mutex_t *mutex);

67 -int __netdata_cond_timedwait(netdata_cond_t *cond, netdata_mutex_t *mutex, struct timespec *tp);

64 +void __netdata_cond_destroy(netdata_cond_t *cond);

65 +void __netdata_cond_signal(netdata_cond_t *cond);

66 +void __netdata_cond_wait(netdata_cond_t *cond, netdata_mutex_t *mutex);

67 +int __netdata_cond_timedwait(netdata_cond_t *cond, netdata_mutex_t *mutex, uint64_t timeout_ns);

69 int __netdata_mutex_init(netdata_mutex_t *mutex);

70 -int __netdata_mutex_destroy(netdata_mutex_t *mutex);

71 -int __netdata_mutex_lock(netdata_mutex_t *mutex);

70 +void __netdata_mutex_destroy(netdata_mutex_t *mutex);

71 +void __netdata_mutex_lock(netdata_mutex_t *mutex);

72 int __netdata_mutex_trylock(netdata_mutex_t *mutex);

73 -int __netdata_mutex_unlock(netdata_mutex_t *mutex);

73 +void __netdata_mutex_unlock(netdata_mutex_t *mutex);

75 -int __netdata_rwlock_destroy(netdata_rwlock_t *rwlock);

75 +void __netdata_rwlock_destroy(netdata_rwlock_t *rwlock);

76 int __netdata_rwlock_init(netdata_rwlock_t *rwlock);

77 -int __netdata_rwlock_rdlock(netdata_rwlock_t *rwlock);

78 -int __netdata_rwlock_wrlock(netdata_rwlock_t *rwlock);

79 -int __netdata_rwlock_rdunlock(netdata_rwlock_t *rwlock);

80 -int __netdata_rwlock_wrunlock(netdata_rwlock_t *rwlock);

77 +void __netdata_rwlock_rdlock(netdata_rwlock_t *rwlock);

78 +void __netdata_rwlock_wrlock(netdata_rwlock_t *rwlock);

79 +void __netdata_rwlock_rdunlock(netdata_rwlock_t *rwlock);

80 +void __netdata_rwlock_wrunlock(netdata_rwlock_t *rwlock);

81 int __netdata_rwlock_tryrdlock(netdata_rwlock_t *rwlock);

82 int __netdata_rwlock_trywrlock(netdata_rwlock_t *rwlock);

83 +void __netdata_cond_broadcast(netdata_cond_t *cond);

85 #ifdef NETDATA_TRACE_RWLOCKS

@@ -135,6 +136,7 @@ int netdata_rwlock_trywrlock_debug( const char *file, const char *function, cons

136 #define netdata_cond_init(cond) __netdata_cond_init(cond)

137 #define netdata_cond_destroy(cond) __netdata_cond_destroy(cond)

138 #define netdata_cond_signal(cond) __netdata_cond_signal(cond)

139 +#define netdata_cond_broadcast(cond) __netdata_cond_broadcast(cond)

140 #define netdata_cond_wait(cond, mutex) __netdata_cond_wait(cond, mutex)

141 #define netdata_cond_timedwait(cond, mutex, tp) __netdata_cond_timedwait(cond, mutex, tp)

142