@cryptotaxi247 / netdata-1 / commits / c2075ca19

Streaming improvements No 12 (#19299)

* do not log when a socket is closed on accept() * waiting queues use futex under linux; use waiting-queues for pgc-queue-locks * do not use waiting queue for pgc-queue-lock * multiplex releasing to ensure better ordering; disable futex implementation until further testing * initialize pulse counters to zero * fixed web connected clients count in pulse * coverity fixes

Costa Tsaousis committed Dec 30, 2024 at 21:07 UTC c2075ca19f74c7cf05a56544901f9fa81a76265e

20 files changed +250 -151

src/collectors/debugfs.plugin/module-libsensors.c

@@ -924,6 +924,10 @@ static SENSOR *sensor_get_or_create(DICTIONARY *dict, const sensors_chip_name *c

924 string2str(s->feature.name));

925 }

926

927 + // we have to free this, because it is malloced from libsensors

928 + if(label)

929 + free((void *)label); // do not use freez() here - libsensors uses malloc()

930 +

931 netdata_fix_chart_id(buf);

932 s->id = string_strdupz(buf);

933

src/collectors/proc.plugin/proc_diskstats.c

+4 -7

@@ -340,7 +340,6 @@ static inline int is_major_enabled(int major) {

340

341 static inline int get_disk_name_from_path(const char *path, char *result, size_t result_size, unsigned long major, unsigned long minor, char *disk, char *prefix, int depth) {

342 //collector_info("DEVICE-MAPPER ('%s', %lu:%lu): examining directory '%s' (allowed depth %d).", disk, major, minor, path, depth);

343 -

343 int found = 0, preferred = 0;

344

345 char *first_result = mallocz(result_size + 1);

@@ -401,9 +400,8 @@ static inline int get_disk_name_from_path(const char *path, char *result, size_t

400 else

401 strncpyz(filename, result, FILENAME_MAX);

402 }

404 - else {

403 + else

404 snprintfz(filename, FILENAME_MAX, "%s/%s", path, de->d_name);

406 - }

405

406 struct stat sb;

407 if(stat(filename, &sb) == -1) {

@@ -426,10 +424,11 @@ static inline int get_disk_name_from_path(const char *path, char *result, size_t

424 snprintfz(result, result_size - 1, "%s%s%s", (prefix)?prefix:"", (prefix)?"_":"", de->d_name);

425

426 if(!found) {

429 - strncpyz(first_result, result, result_size);

427 + strncpyz(first_result, result, result_size - 1);

428 found = 1;

429 }

430

431 + result[result_size - 1] = '\0';

432 if(simple_pattern_matches(preferred_ids, result)) {

433 preferred = 1;

434 break;

@@ -438,13 +437,11 @@ static inline int get_disk_name_from_path(const char *path, char *result, size_t

437 }

438 closedir(dir);

439

441 -

440 failed:

443 -

441 if(!found)

442 result[0] = '\0';

443 else if(!preferred)

447 - strncpyz(result, first_result, result_size);

444 + strncpyz(result, first_result, result_size - 1);

445

446 freez(first_result);

447

src/daemon/pulse/pulse-daemon-memory.c

+1 -1

@@ -7,7 +7,7 @@

7 #define dictionary_stats_memory_total(stats) \

8 ((stats).memory.dict + (stats).memory.values + (stats).memory.index)

10 -struct netdata_buffers_statistics netdata_buffers_statistics = {};

10 +struct netdata_buffers_statistics netdata_buffers_statistics = { 0 };

12 void pulse_daemon_memory_do(bool extended) {

13 {

src/daemon/pulse/pulse-http-api.c

+23 -26

@@ -8,8 +8,7 @@

8 static struct web_statistics {

9 bool extended;

11 - PAD64(uint16_t) connected_clients;

12 - PAD64(uint64_t) web_client_count; // oops! this is used for giving unique IDs to web_clients!

11 + PAD64(int32_t) connected_clients;

13 PAD64(uint64_t) web_requests;

14 PAD64(uint64_t) web_usec;

@@ -19,15 +18,14 @@ static struct web_statistics {

19 PAD64(uint64_t) content_size_uncompressed;

20 PAD64(uint64_t) content_size_compressed;

22 -} web_statistics;

21 +} live_stats = { 0 };

24 -uint64_t pulse_web_client_connected(void) {

25 - __atomic_fetch_add(&web_statistics.connected_clients, 1, __ATOMIC_RELAXED);

26 - return __atomic_fetch_add(&web_statistics.web_client_count, 1, __ATOMIC_RELAXED);

23 +void pulse_web_client_connected(void) {

24 + __atomic_fetch_add(&live_stats.connected_clients, 1, __ATOMIC_RELAXED);

25 }

27 void pulse_web_client_disconnected(void) {

30 - __atomic_fetch_sub(&web_statistics.connected_clients, 1, __ATOMIC_RELAXED);

28 + __atomic_fetch_sub(&live_stats.connected_clients, 1, __ATOMIC_RELAXED);

29 }

31 void pulse_web_request_completed(uint64_t dt,

@@ -35,32 +33,31 @@ void pulse_web_request_completed(uint64_t dt,

33 uint64_t bytes_sent,

34 uint64_t content_size,

35 uint64_t compressed_content_size) {

38 - uint64_t old_web_usec_max = web_statistics.web_usec_max;

36 + uint64_t old_web_usec_max = live_stats.web_usec_max;

37 while(dt > old_web_usec_max)

40 - __atomic_compare_exchange(&web_statistics.web_usec_max, &old_web_usec_max, &dt, 1, __ATOMIC_RELAXED, __ATOMIC_RELAXED);

41 -

42 - __atomic_fetch_add(&web_statistics.web_requests, 1, __ATOMIC_RELAXED);

43 - __atomic_fetch_add(&web_statistics.web_usec, dt, __ATOMIC_RELAXED);

44 - __atomic_fetch_add(&web_statistics.bytes_received, bytes_received, __ATOMIC_RELAXED);

45 - __atomic_fetch_add(&web_statistics.bytes_sent, bytes_sent, __ATOMIC_RELAXED);

46 - __atomic_fetch_add(&web_statistics.content_size_uncompressed, content_size, __ATOMIC_RELAXED);

47 - __atomic_fetch_add(&web_statistics.content_size_compressed, compressed_content_size, __ATOMIC_RELAXED);

38 + __atomic_compare_exchange(&live_stats.web_usec_max, &old_web_usec_max, &dt, 1, __ATOMIC_RELAXED, __ATOMIC_RELAXED);

39 +

40 + __atomic_fetch_add(&live_stats.web_requests, 1, __ATOMIC_RELAXED);

41 + __atomic_fetch_add(&live_stats.web_usec, dt, __ATOMIC_RELAXED);

42 + __atomic_fetch_add(&live_stats.bytes_received, bytes_received, __ATOMIC_RELAXED);

43 + __atomic_fetch_add(&live_stats.bytes_sent, bytes_sent, __ATOMIC_RELAXED);

44 + __atomic_fetch_add(&live_stats.content_size_uncompressed, content_size, __ATOMIC_RELAXED);

45 + __atomic_fetch_add(&live_stats.content_size_compressed, compressed_content_size, __ATOMIC_RELAXED);

46 }

48 static inline void pulse_web_copy(struct web_statistics *gs, uint8_t options) {

51 - gs->connected_clients = __atomic_load_n(&web_statistics.connected_clients, __ATOMIC_RELAXED);

52 - gs->web_requests = __atomic_load_n(&web_statistics.web_requests, __ATOMIC_RELAXED);

53 - gs->web_usec = __atomic_load_n(&web_statistics.web_usec, __ATOMIC_RELAXED);

54 - gs->web_usec_max = __atomic_load_n(&web_statistics.web_usec_max, __ATOMIC_RELAXED);

55 - gs->bytes_received = __atomic_load_n(&web_statistics.bytes_received, __ATOMIC_RELAXED);

56 - gs->bytes_sent = __atomic_load_n(&web_statistics.bytes_sent, __ATOMIC_RELAXED);

57 - gs->content_size_uncompressed = __atomic_load_n(&web_statistics.content_size_uncompressed, __ATOMIC_RELAXED);

58 - gs->content_size_compressed = __atomic_load_n(&web_statistics.content_size_compressed, __ATOMIC_RELAXED);

59 - gs->web_client_count = __atomic_load_n(&web_statistics.web_client_count, __ATOMIC_RELAXED);

49 + gs->connected_clients = __atomic_load_n(&live_stats.connected_clients, __ATOMIC_RELAXED);

50 + gs->web_requests = __atomic_load_n(&live_stats.web_requests, __ATOMIC_RELAXED);

51 + gs->web_usec = __atomic_load_n(&live_stats.web_usec, __ATOMIC_RELAXED);

52 + gs->web_usec_max = __atomic_load_n(&live_stats.web_usec_max, __ATOMIC_RELAXED);

53 + gs->bytes_received = __atomic_load_n(&live_stats.bytes_received, __ATOMIC_RELAXED);

54 + gs->bytes_sent = __atomic_load_n(&live_stats.bytes_sent, __ATOMIC_RELAXED);

55 + gs->content_size_uncompressed = __atomic_load_n(&live_stats.content_size_uncompressed, __ATOMIC_RELAXED);

56 + gs->content_size_compressed = __atomic_load_n(&live_stats.content_size_compressed, __ATOMIC_RELAXED);

58 if(options & GLOBAL_STATS_RESET_WEB_USEC_MAX) {

59 uint64_t n = 0;

63 - __atomic_compare_exchange(&web_statistics.web_usec_max, (uint64_t *) &gs->web_usec_max, &n, 1, __ATOMIC_RELAXED, __ATOMIC_RELAXED);

60 + __atomic_compare_exchange(&live_stats.web_usec_max, (uint64_t *) &gs->web_usec_max, &n, 1, __ATOMIC_RELAXED, __ATOMIC_RELAXED);

61 }

62 }

src/daemon/pulse/pulse-http-api.h

+1 -1

@@ -5,7 +5,7 @@

6 #include "daemon/common.h"

8 -uint64_t pulse_web_client_connected(void);

8 +void pulse_web_client_connected(void);

9 void pulse_web_client_disconnected(void);

11 void pulse_web_request_completed(uint64_t dt,

src/daemon/pulse/pulse-ingestion.c

+1 -1

@@ -5,7 +5,7 @@

6 static struct ingest_statistics {

7 uint64_t db_points_stored_per_tier[RRD_STORAGE_TIERS];

8 -} ingest_statistics;

8 +} ingest_statistics = { 0 };

10 void pulse_queries_rrdset_collection_completed(size_t *points_read_per_tier_array) {

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

src/daemon/pulse/pulse-ml.c

+1 -1

@@ -12,7 +12,7 @@ static struct ml_statistics {

12 PAD64(uint64_t) ml_memory_consumption;

13 PAD64(uint64_t) ml_memory_new;

14 PAD64(uint64_t) ml_memory_delete;

15 -} ml_statistics = {0};

15 +} ml_statistics = { 0 };

17 void pulse_ml_models_received()

18 {

src/daemon/pulse/pulse-queries.c

+1 -1

@@ -30,7 +30,7 @@ static struct query_statistics {

31 PAD64(uint64_t) exporters_queries_made;

32 PAD64(uint64_t) exporters_db_points_read;

33 -} query_statistics;

33 +} query_statistics = { 0 };

35 void pulse_queries_ml_query_completed(size_t points_read) {

36 __atomic_fetch_add(&query_statistics.ml_queries_made, 1, __ATOMIC_RELAXED);

src/daemon/pulse/pulse-sqlite3.c

+1 -1

@@ -20,7 +20,7 @@ static struct sqlite3_statistics {

20 PAD64(uint64_t) sqlite3_context_cache_spill;

21 PAD64(uint64_t) sqlite3_metadata_cache_write;

22 PAD64(uint64_t) sqlite3_context_cache_write;

23 -} sqlite3_statistics = { };

23 +} sqlite3_statistics = { 0 };

25 void pulse_sqlite3_query_completed(bool success, bool busy, bool locked) {

26 if(!sqlite3_statistics.enabled) return;

src/database/engine/cache.c

+71 -43

@@ -22,6 +22,9 @@ typedef int32_t REFCOUNT;

22 #define PGC_WITH_ARAL 1

23 #endif

25 +// unfortunately waiting queue is significantly slower than spinlock

26 +// #define PGC_QUEUE_LOCK_AS_WAITING_QUEUE 1

27 +

28 typedef enum __attribute__ ((__packed__)) {

29 // mutually exclusive flags

30 PGC_PAGE_CLEAN = (1 << 0), // none of the following

@@ -71,7 +74,11 @@ struct pgc_page {

74 };

76 struct pgc_queue {

77 +#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)

78 + WAITING_QUEUE *wq;

79 +#else

80 SPINLOCK spinlock;

81 +#endif

82 union {

83 PGC_PAGE *base;

84 Pvoid_t sections_judy;

@@ -239,9 +246,20 @@ static inline size_t pgc_indexing_partition(PGC *cache, Word_t metric_id) {

246 _result; \

247 })

248

242 -#define pgc_queue_trylock(cache, ll) spinlock_trylock(&(ll)->spinlock)

243 -#define pgc_queue_lock(cache, ll) spinlock_lock(&(ll)->spinlock)

244 -#define pgc_queue_unlock(cache, ll) spinlock_unlock(&(ll)->spinlock)

249 +#define PGC_QUEUE_LOCK_PRIO_COLLECTORS WAITING_QUEUE_PRIO_URGENT

250 +#define PGC_QUEUE_LOCK_PRIO_EVICTORS WAITING_QUEUE_PRIO_HIGH

251 +#define PGC_QUEUE_LOCK_PRIO_FLUSHERS WAITING_QUEUE_PRIO_NORMAL

252 +#define PGC_QUEUE_LOCK_PRIO_OTHERS WAITING_QUEUE_PRIO_LOW

253 +

254 +#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)

255 +#define pgc_queue_trylock(cache, ll) waiting_queue_try_acquire((ll)->wq)

256 +#define pgc_queue_lock(cache, ll, prio) waiting_queue_acquire((ll)->wq, prio)

257 +#define pgc_queue_unlock(cache, ll) waiting_queue_release((ll)->wq)

258 +#else

259 +#define pgc_queue_trylock(cache, ll) spinlock_trylock(&((ll)->spinlock))

260 +#define pgc_queue_lock(cache, ll, prio) spinlock_lock(&((ll)->spinlock))

261 +#define pgc_queue_unlock(cache, ll) spinlock_unlock(&((ll)->spinlock))

262 +#endif

263

264 #define page_transition_trylock(cache, page) spinlock_trylock(&(page)->transition_spinlock)

265 #define page_transition_lock(cache, page) spinlock_lock(&(page)->transition_spinlock)

@@ -590,9 +608,9 @@ static inline void pgc_stats_index_judy_change(PGC *cache, size_t mem_before_jud

608 }

609 }

610

593 -static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock) {

611 +static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITING_QUEUE_PRIORITY prio __maybe_unused) {

612 if(!having_lock)

595 - pgc_queue_lock(cache, q);

613 + pgc_queue_lock(cache, q, prio);

614

615 internal_fatal(page_get_status_flags(page) != 0,

616 "DBENGINE CACHE: invalid page flags, the page has %d, but it is should be %d",

@@ -659,7 +677,7 @@ static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA

677 pgc_size_histogram_add(cache, &q->stats->size_histogram, page);

678 }

679

662 -static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock) {

680 +static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITING_QUEUE_PRIORITY prio __maybe_unused) {

681 if(cache->config.stats)

682 pgc_size_histogram_del(cache, &q->stats->size_histogram, page);

683

@@ -669,7 +687,7 @@ static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA

687 __atomic_add_fetch(&q->stats->removed_size, page->assumed_size, __ATOMIC_RELAXED);

688

689 if(!having_lock)

672 - pgc_queue_lock(cache, q);

690 + pgc_queue_lock(cache, q, prio);

691

692 internal_fatal(page_get_status_flags(page) != q->flags,

693 "DBENGINE CACHE: invalid page flags, the page has %d, but it is should be %d",

@@ -688,7 +706,7 @@ static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA

706 DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(sp->base, page, link.prev, link.next);

707

708 if(!sp->base) {

691 - size_t mem_before_judyl, mem_after_judyl;

709 + ssize_t mem_before_judyl, mem_after_judyl;

710

711 mem_before_judyl = JudyLMemUsed(q->sections_judy);

712 int rc = JudyLDel(&q->sections_judy, page->section, PJE0);

@@ -699,7 +717,6 @@ static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA

717

718 // freez(sp);

719 aral_freez(pgc_sections_aral, sp);

702 - mem_after_judyl -= sizeof(struct section_pages);

720 pgc_stats_queue_judy_change(cache, q, mem_before_judyl, mem_after_judyl);

721 }

722 }

@@ -735,7 +752,7 @@ static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {

752 // ----------------------------------------------------------------------------

753 // state transitions

754

738 -static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock) {

755 +static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock, WAITING_QUEUE_PRIORITY prio) {

756 if(!having_transition_lock)

757 page_transition_lock(cache, page);

758

@@ -748,23 +765,23 @@ static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transi

765 }

766

767 if(flags & PGC_PAGE_HOT)

751 - pgc_queue_del(cache, &cache->hot, page, false);

768 + pgc_queue_del(cache, &cache->hot, page, false, prio);

769

770 if(flags & PGC_PAGE_DIRTY)

754 - pgc_queue_del(cache, &cache->dirty, page, false);

771 + pgc_queue_del(cache, &cache->dirty, page, false, prio);

772

773 // first add to linked list, the set the flag (required for move_page_last())

757 - pgc_queue_add(cache, &cache->clean, page, having_clean_lock);

774 + pgc_queue_add(cache, &cache->clean, page, having_clean_lock, prio);

775

776 if(!having_transition_lock)

777 page_transition_unlock(cache, page);

778 }

779

763 -static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock) {

780 +static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock, WAITING_QUEUE_PRIORITY prio) {

781 if(!having_hot_lock)

782 // to avoid deadlocks, we have to get the hot lock before the page transition

783 // since this is what all_hot_to_dirty() does

767 - pgc_queue_lock(cache, &cache->hot);

784 + pgc_queue_lock(cache, &cache->hot, prio);

785

786 page_transition_lock(cache, page);

787

@@ -784,17 +801,17 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo

801 __atomic_add_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);

802

803 if(likely(flags & PGC_PAGE_HOT))

787 - pgc_queue_del(cache, &cache->hot, page, true);

804 + pgc_queue_del(cache, &cache->hot, page, true, prio);

805

806 if(!having_hot_lock)

807 // we don't need the hot lock anymore

808 pgc_queue_unlock(cache, &cache->hot);

809

810 if(unlikely(flags & PGC_PAGE_CLEAN))

794 - pgc_queue_del(cache, &cache->clean, page, false);

811 + pgc_queue_del(cache, &cache->clean, page, false, prio);

812

813 // first add to linked list, the set the flag (required for move_page_last())

797 - pgc_queue_add(cache, &cache->dirty, page, false);

814 + pgc_queue_add(cache, &cache->dirty, page, false, prio);

815

816 __atomic_sub_fetch(&cache->stats.hot2dirty_entries, 1, __ATOMIC_RELAXED);

817 __atomic_sub_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);

@@ -802,7 +819,7 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo

819 page_transition_unlock(cache, page);

820 }

821

805 -static inline void page_set_hot(PGC *cache, PGC_PAGE *page) {

822 +static inline void page_set_hot(PGC *cache, PGC_PAGE *page, WAITING_QUEUE_PRIORITY prio) {

823 page_transition_lock(cache, page);

824

825 PGC_PAGE_FLAGS flags = page_get_status_flags(page);

@@ -813,13 +830,13 @@ static inline void page_set_hot(PGC *cache, PGC_PAGE *page) {

830 }

831

832 if(flags & PGC_PAGE_DIRTY)

816 - pgc_queue_del(cache, &cache->dirty, page, false);

833 + pgc_queue_del(cache, &cache->dirty, page, false, prio);

834

835 if(flags & PGC_PAGE_CLEAN)

819 - pgc_queue_del(cache, &cache->clean, page, false);

836 + pgc_queue_del(cache, &cache->clean, page, false, prio);

837

838 // first add to linked list, the set the flag (required for move_page_last())

822 - pgc_queue_add(cache, &cache->hot, page, false);

839 + pgc_queue_add(cache, &cache->hot, page, false, prio);

840

841 page_transition_unlock(cache, page);

842 }

@@ -1098,10 +1115,10 @@ static inline bool make_acquired_page_clean_and_evict_or_page_release(PGC *cache

1115 pointer_check(cache, page);

1116

1117 page_transition_lock(cache, page);

1101 - pgc_queue_lock(cache, &cache->clean);

1118 + pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1119

1120 // make it clean - it does not have any accesses, so it will be prepended

1104 - page_set_clean(cache, page, true, true);

1121 + page_set_clean(cache, page, true, true, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1122

1123 if(!acquired_page_get_for_deletion_or_release_it(cache, page)) {

1124 pgc_queue_unlock(cache, &cache->clean);

@@ -1110,7 +1127,7 @@ static inline bool make_acquired_page_clean_and_evict_or_page_release(PGC *cache

1127 }

1128

1129 // remove it from the linked list

1113 - pgc_queue_del(cache, &cache->clean, page, true);

1130 + pgc_queue_del(cache, &cache->clean, page, true, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1131 pgc_queue_unlock(cache, &cache->clean);

1132 page_transition_unlock(cache, page);

1133

@@ -1213,7 +1230,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic

1230 // at this point we have the clean lock

1231 }

1232 else

1216 - pgc_queue_lock(cache, &cache->clean);

1233 + pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1234

1235 timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_LOCK);

1236

@@ -1242,7 +1259,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic

1259 // we can delete this page

1260

1261 // remove it from the clean list

1245 - pgc_queue_del(cache, &cache->clean, page, true);

1262 + pgc_queue_del(cache, &cache->clean, page, true, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1263

1264 __atomic_add_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED);

1265 __atomic_add_fetch(&cache->stats.evicting_size, page->assumed_size, __ATOMIC_RELAXED);

@@ -1388,7 +1405,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic

1405 } while(all_of_them || (total_pages_evicted < max_evict && total_pages_relocated < max_skip));

1406

1407 if(all_of_them && !filter) {

1391 - pgc_queue_lock(cache, &cache->clean);

1408 + pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1409 size_t entries = __atomic_load_n(&cache->clean.stats->entries, __ATOMIC_RELAXED);

1410 if(entries) {

1411 nd_log_limit_static_global_var(erl, 1, 0);

@@ -1491,9 +1508,9 @@ static PGC_PAGE *page_add(PGC *cache, PGC_ENTRY *entry, bool *added) {

1508 pgc_index_write_unlock(cache, partition);

1509

1510 if (entry->hot)

1494 - page_set_hot(cache, page);

1511 + page_set_hot(cache, page, PGC_QUEUE_LOCK_PRIO_COLLECTORS);

1512 else

1496 - page_set_clean(cache, page, false, false);

1513 + page_set_clean(cache, page, false, false, PGC_QUEUE_LOCK_PRIO_EVICTORS);

1514

1515 PGC_REFERENCED_PAGES_PLUS1(cache, page);

1516

@@ -1706,7 +1723,7 @@ cleanup:

1723 }

1724

1725 static void all_hot_pages_to_dirty(PGC *cache, Word_t section) {

1709 - pgc_queue_lock(cache, &cache->hot);

1726 + pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_COLLECTORS);

1727

1728 bool first = true;

1729 Word_t last_section = (section == PGC_SECTION_ALL) ? 0 : section;

@@ -1722,7 +1739,7 @@ static void all_hot_pages_to_dirty(PGC *cache, Word_t section) {

1739 PGC_PAGE *next = page->link.next;

1740

1741 if(page_acquire(cache, page)) {

1725 - page_set_dirty(cache, page, true);

1742 + page_set_dirty(cache, page, true, PGC_QUEUE_LOCK_PRIO_COLLECTORS);

1743 page_release(cache, page, false);

1744 // page ptr may be invalid now

1745 }

@@ -1750,7 +1767,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai

1767 // we got the lock at this point

1768 }

1769 else

1753 - pgc_queue_lock(cache, &cache->dirty);

1770 + pgc_queue_lock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS);

1771

1772 size_t optimal_flush_size = cache->config.max_dirty_pages_per_call;

1773 size_t dirty_version_at_entry = cache->dirty.version;

@@ -1847,7 +1864,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai

1864 __atomic_add_fetch(&cache->stats.flushing_size, tpg->assumed_size, __ATOMIC_RELAXED);

1865

1866 // remove it from the dirty list

1850 - pgc_queue_del(cache, &cache->dirty, tpg, true);

1867 + pgc_queue_del(cache, &cache->dirty, tpg, true, PGC_QUEUE_LOCK_PRIO_FLUSHERS);

1868

1869 pages_removed_dirty_size += tpg->assumed_size;

1870 pages_removed_dirty++;

@@ -1914,7 +1931,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai

1931 if(!tpg->accesses)

1932 pages_to_evict++;

1933

1917 - page_set_clean(cache, tpg, true, false);

1934 + page_set_clean(cache, tpg, true, false, PGC_QUEUE_LOCK_PRIO_FLUSHERS);

1935 page_transition_unlock(cache, tpg);

1936 page_release(cache, tpg, false);

1937 // tpg ptr may be invalid now

@@ -1934,7 +1951,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai

1951 }

1952 }

1953 else {

1937 - pgc_queue_lock(cache, &cache->dirty);

1954 + pgc_queue_lock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS);

1955 have_dirty_lock = true;

1956 }

1957 }

@@ -1974,7 +1991,7 @@ static void *pgc_evict_thread(void *ptr) {

1991 job_id = new_job_id;

1992

1993 if (nd_thread_signaled_to_cancel())

1977 - return NULL;

1994 + break;

1995

1996 evict_pages(cache, 0, 0, true, false);

1997

@@ -2081,9 +2098,15 @@ PGC *pgc_create(const char *name,

2098 #endif

2099

2100

2101 +#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)

2102 + cache->hot.wq = waiting_queue_create();

2103 + cache->dirty.wq = waiting_queue_create();

2104 + cache->clean.wq = waiting_queue_create();

2105 +#else

2106 spinlock_init(&cache->hot.spinlock);

2107 spinlock_init(&cache->dirty.spinlock);

2108 spinlock_init(&cache->clean.spinlock);

2109 +#endif

2110

2111 cache->hot.flags = PGC_PAGE_HOT;

2112 cache->hot.linked_list_in_sections_judy = true;

@@ -2150,6 +2173,11 @@ void pgc_destroy(PGC *cache) {

2173 #endif

2174 }

2175

2176 +#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)

2177 + waiting_queue_destroy(cache->hot.wq);

2178 + waiting_queue_destroy(cache->dirty.wq);

2179 + waiting_queue_destroy(cache->clean.wq);

2180 +#endif

2181 freez(cache->index);

2182 freez(cache);

2183 }

@@ -2180,7 +2208,7 @@ void pgc_page_hot_to_dirty_and_release(PGC *cache, PGC_PAGE *page, bool never_fl

2208 //#endif

2209

2210 // make page dirty

2183 - page_set_dirty(cache, page, false);

2211 + page_set_dirty(cache, page, false, PGC_QUEUE_LOCK_PRIO_COLLECTORS);

2212

2213 // release the page

2214 page_release(cache, page, true);

@@ -2425,7 +2453,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_

2453 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_indexing_started, 1, __ATOMIC_RELAXED);

2454 p2_add_fetch(&cache->stats.p2_workers_jv2_flush, 1);

2455

2428 - pgc_queue_lock(cache, &cache->hot);

2456 + pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);

2457

2458 Pvoid_t JudyL_metrics = NULL;

2459 Pvoid_t JudyL_extents_pos = NULL;

@@ -2556,7 +2584,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_

2584 }

2585

2586 yield_the_processor(); // do not lock too aggressively

2559 - pgc_queue_lock(cache, &cache->hot);

2587 + pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);

2588 }

2589

2590 spinlock_unlock(&sp->migration_to_v2_spinlock);

@@ -2580,7 +2608,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_

2608

2609 // balance-parents: transition from hot to clean directly

2610 yield_the_processor(); // do not lock too aggressively

2583 - page_set_clean(cache, pi->page, true, false);

2611 + page_set_clean(cache, pi->page, true, false, PGC_QUEUE_LOCK_PRIO_OTHERS);

2612 page_transition_unlock(cache, pi->page);

2613 page_release(cache, pi->page, true);

2614

@@ -2631,7 +2659,7 @@ void pgc_open_evict_clean_pages_of_datafile(PGC *cache, struct rrdengine_datafil

2659 size_t pgc_count_clean_pages_having_data_ptr(PGC *cache, Word_t section, void *ptr) {

2660 size_t found = 0;

2661

2634 - pgc_queue_lock(cache, &cache->clean);

2662 + pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_OTHERS);

2663 for(PGC_PAGE *page = cache->clean.base; page ;page = page->link.next)

2664 found += (page->data == ptr && page->section == section) ? 1 : 0;

2665 pgc_queue_unlock(cache, &cache->clean);

@@ -2642,7 +2670,7 @@ size_t pgc_count_clean_pages_having_data_ptr(PGC *cache, Word_t section, void *p

2670 size_t pgc_count_hot_pages_having_data_ptr(PGC *cache, Word_t section, void *ptr) {

2671 size_t found = 0;

2672

2645 - pgc_queue_lock(cache, &cache->hot);

2673 + pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);

2674 Pvoid_t *section_pages_pptr = JudyLGet(cache->hot.sections_judy, section, PJE0);

2675 if(section_pages_pptr) {

2676 struct section_pages *sp = *section_pages_pptr;

src/database/rrdlabels.c

+2 -5

@@ -168,14 +168,11 @@ static void delete_label(RRDLABEL *label)

168 if (refcount == 0) {

169 JudyAllocThreadPulseReset();

170

171 - int ret = JudyHSDel(&global_labels.JudyHS, (void *)label, sizeof(*label), PJE0);

171 + JudyHSDel(&global_labels.JudyHS, (void *)label, sizeof(*label), PJE0);

172

173 int64_t judy_mem = JudyAllocThreadPulseGetAndReset();

174

175 - if (unlikely(ret == JERR))

176 - STATS_MINUS_MEMORY(&dictionary_stats_category_rrdlabels, judy_mem, sizeof(*rrdlabel), 0);

177 - else

178 - STATS_MINUS_MEMORY(&dictionary_stats_category_rrdlabels, judy_mem, sizeof(*rrdlabel), 0);

175 + STATS_MINUS_MEMORY(&dictionary_stats_category_rrdlabels, judy_mem, sizeof(*rrdlabel), 0);

176 string_freez(label->index.key);

177 string_freez(label->index.value);

178 freez(rrdlabel);

src/libnetdata/local-sockets/local-sockets.h

+2 -2

@@ -472,8 +472,8 @@ static inline const char *local_sockets_protocol_name(LOCAL_SOCKET *n) {

472 static inline void local_listeners_print_socket(LS_STATE *ls __maybe_unused, const LOCAL_SOCKET *nn, void *data __maybe_unused) {

473 LOCAL_SOCKET *n = (LOCAL_SOCKET *)nn;

474

475 - char local_address[INET6_ADDRSTRLEN];

476 - char remote_address[INET6_ADDRSTRLEN];

475 + char local_address[INET6_ADDRSTRLEN] = "";

476 + char remote_address[INET6_ADDRSTRLEN] = "";

477

478 if(n->local.family == AF_INET) {

479 ipv4_address_to_txt(n->local.ip.ipv4, local_address);

src/libnetdata/log/systemd-cat-native.c

+5 -5

@@ -231,13 +231,13 @@ static CURLcode journal_remote_send_buffer(CURL* curl, BUFFER *msg) {

231 if(verbose)

232 log_message_to_stderr(msg, "REMOTE");

233

234 - struct upload_data upload = {0};

235 -

234 if (!curl || !buffer_strlen(msg))

235 return CURLE_FAILED_INIT;

236

239 - upload.data = (char *) buffer_tostring(msg);

240 - upload.length = buffer_strlen(msg);

237 + struct upload_data upload = {

238 + .data = (char *) buffer_tostring(msg),

239 + .length = buffer_strlen(msg),

240 + };

241

242 curl_easy_setopt(curl, CURLOPT_READDATA, &upload);

243 curl_easy_setopt(curl, CURLOPT_INFILESIZE_LARGE, (curl_off_t)upload.length);

@@ -290,7 +290,7 @@ static log_to_journal_remote_ret_t log_input_to_journal_remote(const char *url,

290 fprintf(stderr, "WARNING: cannot read '%s'. Will generate a random _MACHINE_ID.\n", MACHINE_ID_PATH);

291 nd_uuid_t uuid;

292 uuid_generate_random(uuid);

293 - uuid_unparse_lower_compact(uuid, global_boot_id);

293 + uuid_unparse_lower_compact(uuid, global_machine_id);

294 }

295

296 if(global_stream_id[0] == '\0') {

src/libnetdata/socket/poll-events.c

+3 -6

@@ -280,12 +280,9 @@ static int poll_process_new_tcp_connection(POLLINFO *pi, time_t now) {

280 "POLLFD: LISTENER: accept() failed.");

281

282 }

283 - else if(is_socket_closed(nfd)) {

284 - nd_log_limit_static_global_var(erl, 10, 1000);

285 - nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,

286 - "POLLFD: LISTENER: received client socket %d is closed on accept(), dropping connection", nfd);

287 - close(nfd);

288 - }

283 + else if(is_socket_closed(nfd))

284 + close(nfd);

285 +

286 else {

287 // accept ok

288

src/libnetdata/socket/socket.c

+8 -4

@@ -159,12 +159,14 @@ int sock_enlarge_rcv_buf(int fd) {

159 if (getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &current_bs, &optlen) == 0) {

160 // Set the buffer size only if it's smaller than the desired size

161 if (current_bs < bs) {

162 - setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bs, sizeof(bs));

162 + if(setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bs, sizeof(bs)) != 0)

163 + return -1;

164

165 // Re-check the buffer size after attempting to set it

166 if (getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &current_bs, &optlen) == 0)

167 ret = current_bs;

167 - } else {

168 + }

169 + else {

170 // Current buffer size is already large enough

171 ret = current_bs;

172 }

@@ -184,12 +186,14 @@ int sock_enlarge_snd_buf(int fd) {

186 if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &current_bs, &optlen) == 0) {

187 // Set the buffer size only if it's smaller than the desired size

188 if (current_bs < bs) {

187 - setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bs, sizeof(bs));

189 + if(setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bs, sizeof(bs)) != 0)

190 + return -1;

191

192 // Re-check the buffer size after attempting to set it

193 if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &current_bs, &optlen) == 0)

194 ret = current_bs;

192 - } else {

195 + }

196 + else {

197 // Current buffer size is already large enough

198 ret = current_bs;

199 }

src/libnetdata/waiting-queue/waiting-queue.c

+101 -34

@@ -2,19 +2,75 @@

3 #include "waiting-queue.h"

5 +// #define WAITING_QUEUE_USE_FUTEX 1

6 +

7 +#if defined(WAITING_QUEUE_USE_FUTEX)

8 +#include <linux/futex.h>

9 +#include <sys/syscall.h>

10 +

11 +typedef int WQ_COND;

12 +typedef SPINLOCK WQ_MUTEX;

13 +

14 +static inline int futex_wait(int *uaddr, int val) {

15 + return syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);

16 +}

17 +

18 +static inline int futex_wake(int *uaddr, int n) {

19 + return syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, n, NULL, NULL, 0);

20 +}

21 +

22 +static int WQ_COND_init(WQ_COND *cond) { *(cond) = 0; return 0; }

23 +#define WQ_COND_destroy(cond) debug_dummy()

24 +

25 +static inline int WQ_COND_wait(WQ_COND *cond, WQ_MUTEX *mutex) {

26 + int value = *cond;

27 + spinlock_unlock(mutex);

28 + futex_wait(cond, value);

29 + spinlock_lock(mutex);

30 + return 0;

31 +}

32 +

33 +static inline int WQ_COND_signal(WQ_COND *cond) {

34 + __atomic_add_fetch(cond, 1, __ATOMIC_SEQ_CST);

35 + futex_wake(cond, 1);

36 + return 0;

37 +}

38 +

39 +#define WQ_MUTEX_init(mutex) ({ spinlock_init(mutex) ; 0; })

40 +#define WQ_MUTEX_destroy(mutex) debug_dummy()

41 +#define WQ_MUTEX_lock(mutex) spinlock_lock(mutex)

42 +#define WQ_MUTEX_unlock(mutex) spinlock_unlock(mutex)

43 +

44 +#else // !USE_FUTEX

45 +

46 +typedef uv_cond_t WQ_COND;

47 +typedef uv_mutex_t WQ_MUTEX;

48 +

49 +#define WQ_COND_init(cond) uv_cond_init(cond)

50 +#define WQ_COND_destroy(cond) uv_cond_destroy(cond)

51 +#define WQ_COND_wait(cond, mutex) uv_cond_wait(cond, mutex)

52 +#define WQ_COND_signal(cond) uv_cond_signal(cond)

53 +

54 +#define WQ_MUTEX_init(mutex) uv_mutex_init(mutex)

55 +#define WQ_MUTEX_destroy(mutex) uv_mutex_destroy(mutex)

56 +#define WQ_MUTEX_lock(mutex) uv_mutex_lock(mutex)

57 +#define WQ_MUTEX_unlock(mutex) uv_mutex_unlock(mutex)

58 +

59 +#endif // USE_FUTEX

60 +

61 typedef struct waiting_thread {

6 - uv_cond_t cond; // condition variable for this thread

62 + WQ_COND cond; // condition variable for this thread

63 usec_t waiting_since_ut; // when we started waiting

8 - Word_t order;

64 + Word_t priority;

65 struct waiting_thread *prev, *next;

66 } WAITING_THREAD;

68 struct waiting_queue {

13 - uv_mutex_t mutex; // protect the queue structure

69 + WQ_MUTEX mutex; // ensure acquirers and releasers are synchronized

70 Word_t last_seqno; // incrementing sequence counter

15 - WAITING_THREAD *list;

16 - size_t running; // number of threads currently running/waiting

17 - SPINLOCK spinlock;

71 + SPINLOCK spinlock; // ensures there is only 1 runner at a time

72 + REFCOUNT running; // number of threads, including the one holding the lock

73 + WAITING_THREAD *list; // the list of threads waiting, not including the 1 holding the lock

74 };

76 // Determine available bits based on system word size

@@ -41,7 +97,7 @@ static inline Word_t key_get_seqno(Word_t key) {

97 WAITING_QUEUE *waiting_queue_create(void) {

98 WAITING_QUEUE *wq = callocz(1, sizeof(WAITING_QUEUE));

44 - int ret = uv_mutex_init(&wq->mutex);

100 + int ret = WQ_MUTEX_init(&wq->mutex);

101 if(ret != 0) {

102 freez(wq);

103 return NULL;

@@ -57,15 +113,15 @@ void waiting_queue_destroy(WAITING_QUEUE *wq) {

113 if(!wq) return;

114

115 if(wq->running)

60 - fatal("WAITING_QUEUE: destroying waiting queue that still has %zu threads running/waiting", wq->running);

116 + fatal("WAITING_QUEUE: destroying waiting queue that still has %d threads running/waiting", wq->running);

117

62 - uv_mutex_destroy(&wq->mutex);

118 + WQ_MUTEX_destroy(&wq->mutex);

119 freez(wq);

120 }

121

122 static inline void WAITERS_SET(WAITING_QUEUE *wq, WAITING_THREAD *wt) {

123 for(WAITING_THREAD *t = wq->list ; t ;t = t->next) {

68 - if(wt->order < t->order) {

124 + if(wt->priority < t->priority) {

125 DOUBLE_LINKED_LIST_INSERT_ITEM_BEFORE_UNSAFE(wq->list, t, wt, prev, next);

126 return;

127 }

@@ -83,24 +139,34 @@ static inline WAITING_THREAD *WAITERS_FIRST(WAITING_QUEUE *wq) {

139

140 static inline void WAITING_THREAD_init(WAITING_QUEUE *wq, WAITING_THREAD *wt, WAITING_QUEUE_PRIORITY priority) {

141 Word_t seqno = __atomic_add_fetch(&wq->last_seqno, 1, __ATOMIC_RELAXED);

86 - wt->order = make_key(priority, seqno);

142 + wt->priority = make_key(priority, seqno);

143 wt->waiting_since_ut = now_monotonic_usec();

144 wt->prev = wt->next = NULL;

145

90 - int ret = uv_cond_init(&wt->cond);

146 + int ret = WQ_COND_init(&wt->cond);

147 if(ret != 0)

148 fatal("WAITING_QUEUE: cannot initialize condition variable");

149 }

150

95 -static inline void WAITING_THREAD_cleanup(WAITING_QUEUE *wq __maybe_unused, WAITING_THREAD *wt) {

96 - uv_cond_destroy(&wt->cond);

151 +static inline void WAITING_THREAD_cleanup(WAITING_QUEUE *wq __maybe_unused, WAITING_THREAD *wt __maybe_unused) {

152 + WQ_COND_destroy(&wt->cond);

153 }

154

99 -usec_t waiting_queue_wait(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority) {

155 +bool waiting_queue_try_acquire(WAITING_QUEUE *wq) {

156 + if(__atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 1 &&

157 + spinlock_trylock(&wq->spinlock)) {

158 + return true;

159 + }

160 +

161 + __atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED);

162 + return false;

163 +}

164 +

165 +usec_t waiting_queue_acquire(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority) {

166 // Try fast path first - if we're the only one, just go

101 - if(__atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 1) {

102 - if(spinlock_trylock(&wq->spinlock))

103 - return 0;

167 + if(__atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 1 &&

168 + spinlock_trylock(&wq->spinlock)) {

169 + return 0;

170 }

171

172 // Slow path - need to wait

@@ -108,7 +174,7 @@ usec_t waiting_queue_wait(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority) {

174 WAITING_THREAD wt;

175 WAITING_THREAD_init(wq, &wt, priority);

176

111 - uv_mutex_lock(&wq->mutex);

177 + WQ_MUTEX_lock(&wq->mutex);

178 WAITERS_SET(wq, &wt);

179

180 // Wait for our turn

@@ -116,31 +182,32 @@ usec_t waiting_queue_wait(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority) {

182 if (WAITERS_FIRST(wq) == &wt && spinlock_trylock(&wq->spinlock))

183 break;

184 else

119 - uv_cond_wait(&wt.cond, &wq->mutex);

185 + WQ_COND_wait(&wt.cond, &wq->mutex);

186 } while(true);

187

188 WAITERS_DEL(wq, &wt);

123 - uv_mutex_unlock(&wq->mutex);

189 + WQ_MUTEX_unlock(&wq->mutex);

190 WAITING_THREAD_cleanup(wq, &wt);

191

192 return now_monotonic_usec() - wt.waiting_since_ut;

193 }

194

129 -void waiting_queue_done(WAITING_QUEUE *wq) {

130 - spinlock_unlock(&wq->spinlock);

131 -

195 +void waiting_queue_release(WAITING_QUEUE *wq) {

196 // Fast path if we're alone

133 - if(__atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 0)

197 + if(__atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 0) {

198 + spinlock_unlock(&wq->spinlock);

199 return;

200 + }

201

202 // Slow path - need to signal next in line

137 - uv_mutex_lock(&wq->mutex);

203 + WQ_MUTEX_lock(&wq->mutex);

204

205 // Wake up next in line if any

206 if(wq->list)

141 - uv_cond_signal(&wq->list->cond);

207 + WQ_COND_signal(&wq->list->cond);

208

143 - uv_mutex_unlock(&wq->mutex);

209 + spinlock_unlock(&wq->spinlock);

210 + WQ_MUTEX_unlock(&wq->mutex);

211 }

212

213 size_t waiting_queue_waiting(WAITING_QUEUE *wq) {

@@ -183,8 +250,8 @@ static int unittest_functional(void) {

250

251 // Test 1: Fast path should work with no contention

252 fprintf(stderr, " Test 1: Fast path - no contention: ");

186 - usec_t wait_time = waiting_queue_wait(wq, WAITING_QUEUE_PRIO_NORMAL);

187 - waiting_queue_done(wq);

253 + usec_t wait_time = waiting_queue_acquire(wq, WAITING_QUEUE_PRIO_NORMAL);

254 + waiting_queue_release(wq);

255 if(wait_time != 0) {

256 fprintf(stderr, "FAILED (waited %"PRIu64" usec)\n", wait_time);

257 errors++;

@@ -210,8 +277,8 @@ static int unittest_functional(void) {

277 WAITERS_DEL(wq, wt);

278 __atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED);

279

213 - WAITING_QUEUE_PRIORITY prio = key_get_priority(wt->order);

214 - Word_t seqno = key_get_seqno(wt->order);

280 + WAITING_QUEUE_PRIORITY prio = key_get_priority(wt->priority);

281 + Word_t seqno = key_get_seqno(wt->priority);

282

283 prio_counts[prio]++;

284 if(prio < last_prio) {

@@ -265,7 +332,7 @@ static void *stress_thread(void *arg) {

332 bool *stop_flag = args->stop_flag;

333

334 while(!__atomic_load_n(stop_flag, __ATOMIC_ACQUIRE)) {

268 - usec_t wait_time = waiting_queue_wait(wq, stats->priority);

335 + usec_t wait_time = waiting_queue_acquire(wq, stats->priority);

336 stats->executions++;

337 stats->total_wait_time += wait_time;

338 if(wait_time > stats->max_wait_time)

@@ -274,7 +341,7 @@ static void *stress_thread(void *arg) {

341 if(with_sleep)

342 tinysleep();

343

277 - waiting_queue_done(wq);

344 + waiting_queue_release(wq);

345 }

346

347 return NULL;

src/libnetdata/waiting-queue/waiting-queue.h

+6 -4

@@ -4,7 +4,6 @@

4 #define NETDATA_WAITING_QUEUE_H

6 #include "libnetdata/libnetdata.h"

7 -#include <uv.h>

8 /*

9 * WAITING QUEUE

@@ -43,16 +42,19 @@ WAITING_QUEUE *waiting_queue_create(void);

42 // Destroy a waiting queue - must be empty

43 void waiting_queue_destroy(WAITING_QUEUE *wq);

45 +// Returns true when the queue is acquired

46 +bool waiting_queue_try_acquire(WAITING_QUEUE *wq);

47 +

48 // Returns when it is our turn to run

49 // Returns time spent waiting in microseconds

48 -usec_t waiting_queue_wait(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority);

50 +usec_t waiting_queue_acquire(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority);

52 // Mark that we are done - wakes up the next in line

51 -void waiting_queue_done(WAITING_QUEUE *wq);

53 +void waiting_queue_release(WAITING_QUEUE *wq);

55 // Return the number of threads currently waiting

56 size_t waiting_queue_waiting(WAITING_QUEUE *wq);

58 int unittest_waiting_queue(void);

58 -#endif // NETDATA_WAITING_QUEUE_H

\ No newline at end of file

60 +#endif // NETDATA_WAITING_QUEUE_H

src/streaming/stream-sender-commit.c

+9 -7

@@ -68,9 +68,11 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

68 if (unlikely(!src || !src_len))

69 return;

71 - waiting_queue_wait(s->wait_queue, (s->host->stream.rcv.status.tid == gettid_cached()

72 - || s->host->stream.snd.status.tid == gettid_cached()) ?

73 - WAITING_QUEUE_PRIO_HIGH : WAITING_QUEUE_PRIO_NORMAL);

71 + waiting_queue_acquire(

72 + s->wait_queue,

73 + (s->host->stream.rcv.status.tid == gettid_cached() || s->host->stream.snd.status.tid == gettid_cached()) ?

74 + WAITING_QUEUE_PRIO_HIGH :

75 + WAITING_QUEUE_PRIO_NORMAL);

76 stream_sender_lock(s);

78 // copy the sequence number of sender buffer recreates, while having our lock

@@ -85,7 +87,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

87 sender_buffer_destroy(commit);

89 stream_sender_unlock(s);

88 - waiting_queue_done(s->wait_queue);

90 + waiting_queue_release(s->wait_queue);

91 return;

92 }

@@ -181,7 +183,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

183 msg = s->thread.msg;

184

185 stream_sender_unlock(s);

184 - waiting_queue_done(s->wait_queue);

186 + waiting_queue_release(s->wait_queue);

187

188 if (enable_sending) {

189 msg.opcode = STREAM_OPCODE_SENDER_POLLOUT;

@@ -193,7 +195,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

195 overflow_with_lock: {

196 msg = s->thread.msg;

197 stream_sender_unlock(s);

196 - waiting_queue_done(s->wait_queue);

198 + waiting_queue_release(s->wait_queue);

199 msg.opcode = STREAM_OPCODE_SENDER_BUFFER_OVERFLOW;

200 stream_sender_send_opcode(s, msg);

201 nd_log_limit_static_global_var(erl, 1, 0);

@@ -209,7 +211,7 @@ compression_failed_with_lock: {

211 stream_compression_deactivate(s);

212 msg = s->thread.msg;

213 stream_sender_unlock(s);

212 - waiting_queue_done(s->wait_queue);

214 + waiting_queue_release(s->wait_queue);

215 msg.opcode = STREAM_OPCODE_SENDER_RECONNECT_WITHOUT_COMPRESSION;

216 stream_sender_send_opcode(s, msg);

217 nd_log_limit_static_global_var(erl, 1, 0);

src/streaming/stream-sender.c

+1 -1

@@ -371,7 +371,7 @@ static void stream_sender_log_disconnection(struct stream_thread *sth, struct se

371 ND_LOG_STACK_PUSH(lgs);

372

373 nd_log(NDLS_DAEMON, NDLP_NOTICE,

374 - "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s (replication in: %u, out: %u, pending: %u)",

374 + "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s (replication in: %u, out: %u, pending: %zu)",

375 sth->id, rrdhost_hostname(s->host), s->remote_ip, stream_handshake_error_to_string(reason),

376 s->host->stream.snd.status.replication.counter_in, s->host->stream.snd.status.replication.counter_out,

377 dictionary_entries(s->replication.requests));

src/web/server/web_client_cache.c

+5 -1

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

17 // the number of currently connected clients.

19 static struct clients_cache {

20 + unsigned long long client_id;

21 +

22 struct {

23 SPINLOCK spinlock;

24 struct web_client *head; // the structures of the currently connected clients

@@ -103,10 +105,12 @@ struct web_client *web_client_get_from_cache(void) {

105 w = web_client_create(&netdata_buffers_statistics.buffers_web);

106 spinlock_lock(&web_clients_cache.used.spinlock);

107

106 - w->id = pulse_web_client_connected();

108 + w->id = __atomic_add_fetch(&web_clients_cache.client_id, 1, __ATOMIC_RELAXED);

109 web_clients_cache.used.allocated++;

110 }

111

112 + pulse_web_client_connected();

113 +

114 // link it to used web clients

115 DOUBLE_LINKED_LIST_PREPEND_ITEM_UNSAFE(web_clients_cache.used.head, w, cache.prev, cache.next);

116 web_clients_cache.used.count++;