@cryptotaxi247 / netdata-1 / commits / cf03e531d

inline functions related to metrics ingestion (#19524)

* inline functions related to metrics ingestion * move pulse outside the lock * cleanup senders in connector thread - untested * receiver and sender cleanup * buffer functions inlining * more inlining * do not disconnect receivers when the sender fails * heavy parts of sender connect/disconnect run with connector unlocked

Costa Tsaousis committed Jan 30, 2025 at 13:27 UTC cf03e531d10cc1f55f47fe866762a6a7e0a03775
23 files changed +257 -204
src/daemon/pulse/pulse-ingestion.c
+1 -1
@@ -7,7 +7,7 @@ static struct ingest_statistics {
7 uint64_t db_points_stored_per_tier[RRD_STORAGE_TIERS];
8 } ingest_statistics = { 0 };
9
10 -void pulse_queries_rrdset_collection_completed(size_t *points_read_per_tier_array) {
10 +ALWAYS_INLINE 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++) {
12 __atomic_fetch_add(&ingest_statistics.db_points_stored_per_tier[tier], points_read_per_tier_array[tier], __ATOMIC_RELAXED);
13 points_read_per_tier_array[tier] = 0;
src/database/contexts/instance.c
+1 -1
@@ -512,7 +512,7 @@ inline void rrdinstance_updated_rrdset_flags(RRDSET *st) {
512 rrdinstance_trigger_updates(ri, __FUNCTION__ );
513 }
514
515 -inline void rrdinstance_collected_rrdset(RRDSET *st) {
515 +ALWAYS_INLINE void rrdinstance_collected_rrdset(RRDSET *st) {
516 if(st->rrdcontexts.collected)
517 return;
518
src/database/contexts/rrdcontext.c
+1 -1
@@ -88,7 +88,7 @@ void rrdcontext_updated_rrdset_flags(RRDSET *st) {
88 rrdinstance_updated_rrdset_flags(st);
89 }
90
91 -void rrdcontext_collected_rrdset(RRDSET *st) {
91 +ALWAYS_INLINE void rrdcontext_collected_rrdset(RRDSET *st) {
92 rrdinstance_collected_rrdset(st);
93 }
94
src/database/engine/cache.c
+22 -22
@@ -591,7 +591,7 @@ static void pgc_section_pages_static_aral_init(void) {
591 spinlock_unlock(&spinlock);
592 }
593
594 -static inline void
594 +static ALWAYS_INLINE void
595 pgc_stats_queue_judy_change(PGC *cache, struct pgc_queue *ll, size_t mem_before_judyl, size_t mem_after_judyl) {
596 if(mem_after_judyl > mem_before_judyl) {
597 __atomic_add_fetch(&ll->stats->size, mem_after_judyl - mem_before_judyl, __ATOMIC_RELAXED);
@@ -603,7 +603,7 @@ pgc_stats_queue_judy_change(PGC *cache, struct pgc_queue *ll, size_t mem_before_
603 }
604 }
605
606 -static inline void pgc_stats_index_judy_change(PGC *cache, size_t mem_before_judyl, size_t mem_after_judyl) {
606 +static ALWAYS_INLINE void pgc_stats_index_judy_change(PGC *cache, size_t mem_before_judyl, size_t mem_after_judyl) {
607 if(mem_after_judyl > mem_before_judyl) {
608 __atomic_add_fetch(&cache->stats.size, mem_after_judyl - mem_before_judyl, __ATOMIC_RELAXED);
609 }
@@ -612,7 +612,7 @@ static inline void pgc_stats_index_judy_change(PGC *cache, size_t mem_before_jud
612 }
613 }
614
615 -static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITQ_PRIORITY prio __maybe_unused) {
615 +static ALWAYS_INLINE void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITQ_PRIORITY prio __maybe_unused) {
616 if(!having_lock)
617 pgc_queue_lock(cache, q, prio);
618
@@ -681,7 +681,7 @@ static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA
681 pgc_size_histogram_add(cache, &q->stats->size_histogram, page);
682 }
683
684 -static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock,
684 +static ALWAYS_INLINE void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock,
685 WAITQ_PRIORITY prio __maybe_unused) {
686 if(cache->config.stats)
687 pgc_size_histogram_del(cache, &q->stats->size_histogram, page);
@@ -735,7 +735,7 @@ static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA
735 pgc_queue_unlock(cache, q);
736 }
737
738 -static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {
738 +static ALWAYS_INLINE void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {
739 PGC_PAGE_FLAGS flags = page_flag_check(page, PGC_PAGE_CLEAN | PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES);
740
741 if (!(flags & PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES)) {
@@ -758,7 +758,7 @@ static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {
758 // ----------------------------------------------------------------------------
759 // state transitions
760
761 -static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock, WAITQ_PRIORITY prio) {
761 +static ALWAYS_INLINE void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock, WAITQ_PRIORITY prio) {
762 if(!having_transition_lock)
763 page_transition_lock(cache, page);
764
@@ -783,7 +783,7 @@ static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transi
783 page_transition_unlock(cache, page);
784 }
785
786 -static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock, WAITQ_PRIORITY prio) {
786 +static ALWAYS_INLINE void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock, WAITQ_PRIORITY prio) {
787 if(!having_hot_lock)
788 // to avoid deadlocks, we have to get the hot lock before the page transition
789 // since this is what all_hot_to_dirty() does
@@ -825,7 +825,7 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo
825 page_transition_unlock(cache, page);
826 }
827
828 -static inline void page_set_hot(PGC *cache, PGC_PAGE *page, WAITQ_PRIORITY prio) {
828 +static ALWAYS_INLINE void page_set_hot(PGC *cache, PGC_PAGE *page, WAITQ_PRIORITY prio) {
829 page_transition_lock(cache, page);
830
831 PGC_PAGE_FLAGS flags = page_get_status_flags(page);
@@ -851,16 +851,16 @@ static inline void page_set_hot(PGC *cache, PGC_PAGE *page, WAITQ_PRIORITY prio)
851 // ----------------------------------------------------------------------------
852 // Referencing
853
854 -static inline size_t PGC_REFERENCED_PAGES(PGC *cache) {
854 +static ALWAYS_INLINE size_t PGC_REFERENCED_PAGES(PGC *cache) {
855 return __atomic_load_n(&cache->stats.referenced_entries, __ATOMIC_RELAXED);
856 }
857
858 -static inline void PGC_REFERENCED_PAGES_PLUS1(PGC *cache, PGC_PAGE *page) {
858 +static ALWAYS_INLINE void PGC_REFERENCED_PAGES_PLUS1(PGC *cache, PGC_PAGE *page) {
859 __atomic_add_fetch(&cache->stats.referenced_entries, 1, __ATOMIC_RELAXED);
860 __atomic_add_fetch(&cache->stats.referenced_size, page->assumed_size, __ATOMIC_RELAXED);
861 }
862
863 -static inline void PGC_REFERENCED_PAGES_MINUS1(PGC *cache, size_t assumed_size) {
863 +static ALWAYS_INLINE void PGC_REFERENCED_PAGES_MINUS1(PGC *cache, size_t assumed_size) {
864 __atomic_sub_fetch(&cache->stats.referenced_entries, 1, __ATOMIC_RELAXED);
865 __atomic_sub_fetch(&cache->stats.referenced_size, assumed_size, __ATOMIC_RELAXED);
866 }
@@ -868,7 +868,7 @@ static inline void PGC_REFERENCED_PAGES_MINUS1(PGC *cache, size_t assumed_size)
868 // If the page is not already acquired,
869 // YOU HAVE TO HAVE THE QUEUE (hot, dirty, clean - the page is in), LOCKED!
870 // If you don't have it locked, NOTHING PREVENTS THIS PAGE FROM VANISHING WHILE THIS IS CALLED!
871 -static inline bool page_acquire(PGC *cache, PGC_PAGE *page) {
871 +static ALWAYS_INLINE bool page_acquire(PGC *cache, PGC_PAGE *page) {
872 __atomic_add_fetch(&cache->stats.acquires, 1, __ATOMIC_RELAXED);
873
874 REFCOUNT rc = refcount_acquire_advanced(&page->refcount);
@@ -882,7 +882,7 @@ static inline bool page_acquire(PGC *cache, PGC_PAGE *page) {
882 return false;
883 }
884
885 -static inline void page_release(PGC *cache, PGC_PAGE *page, bool evict_if_necessary) {
885 +static ALWAYS_INLINE void page_release(PGC *cache, PGC_PAGE *page, bool evict_if_necessary) {
886 __atomic_add_fetch(&cache->stats.releases, 1, __ATOMIC_RELAXED);
887
888 size_t assumed_size = page->assumed_size; // take the size before we release it
@@ -895,7 +895,7 @@ static inline void page_release(PGC *cache, PGC_PAGE *page, bool evict_if_necess
895 }
896 }
897
898 -static inline bool non_acquired_page_get_for_deletion___while_having_clean_locked(PGC *cache __maybe_unused, PGC_PAGE *page) {
898 +static ALWAYS_INLINE bool non_acquired_page_get_for_deletion___while_having_clean_locked(PGC *cache __maybe_unused, PGC_PAGE *page) {
899 __atomic_add_fetch(&cache->stats.acquires_for_deletion, 1, __ATOMIC_RELAXED);
900
901 internal_fatal(!is_page_clean(page),
@@ -914,7 +914,7 @@ static inline bool non_acquired_page_get_for_deletion___while_having_clean_locke
914 return false;
915 }
916
917 -static inline bool acquired_page_get_for_deletion_or_release_it(PGC *cache __maybe_unused, PGC_PAGE *page) {
917 +static ALWAYS_INLINE bool acquired_page_get_for_deletion_or_release_it(PGC *cache __maybe_unused, PGC_PAGE *page) {
918 __atomic_add_fetch(&cache->stats.acquires_for_deletion, 1, __ATOMIC_RELAXED);
919
920 size_t assumed_size = page->assumed_size; // take the size before we release it
@@ -1489,7 +1489,7 @@ static PGC_PAGE *page_add(PGC *cache, PGC_ENTRY *entry, bool *added) {
1489 return page;
1490 }
1491
1492 -static inline PGC_PAGE *page_find_and_acquire_exact_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1492 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_exact_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1493 Pvoid_t *page_ptr = JudyLGet(*pages_judy_pptr, start_time_s, PJE0);
1494 if(!page_ptr)
1495 return NULL;
@@ -1505,7 +1505,7 @@ static inline PGC_PAGE *page_find_and_acquire_exact_unsafe(PGC *cache, Pvoid_t *
1505 return NULL;
1506 }
1507
1508 -static inline PGC_PAGE *page_find_and_acquire_first_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1508 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_first_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1509 Word_t time = start_time_s;
1510 for(Pvoid_t *page_ptr = JudyLFirst(*pages_judy_pptr, &time, PJE0);
1511 page_ptr ;
@@ -1523,7 +1523,7 @@ static inline PGC_PAGE *page_find_and_acquire_first_unsafe(PGC *cache, Pvoid_t *
1523 return NULL;
1524 }
1525
1526 -static inline PGC_PAGE *page_find_and_acquire_next_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1526 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_next_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1527 Word_t time = start_time_s;
1528 for(Pvoid_t *page_ptr = JudyLNext(*pages_judy_pptr, &time, PJE0);
1529 page_ptr ;
@@ -1541,7 +1541,7 @@ static inline PGC_PAGE *page_find_and_acquire_next_unsafe(PGC *cache, Pvoid_t *p
1541 return NULL;
1542 }
1543
1544 -static inline PGC_PAGE *page_find_and_acquire_last_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1544 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_last_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1545 Word_t time = start_time_s;
1546 for(Pvoid_t *page_ptr = JudyLLast(*pages_judy_pptr, &time, PJE0);
1547 page_ptr ;
@@ -1559,7 +1559,7 @@ static inline PGC_PAGE *page_find_and_acquire_last_unsafe(PGC *cache, Pvoid_t *p
1559 return NULL;
1560 }
1561
1562 -static inline PGC_PAGE *page_find_and_acquire_prev_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1562 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_prev_unsafe(PGC *cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1563 Word_t time = start_time_s;
1564 for(Pvoid_t *page_ptr = JudyLPrev(*pages_judy_pptr, &time, PJE0);
1565 page_ptr ;
@@ -1577,7 +1577,7 @@ static inline PGC_PAGE *page_find_and_acquire_prev_unsafe(PGC *cache, Pvoid_t *p
1577 return NULL;
1578 }
1579
1580 -static PGC_PAGE *page_find_and_acquire_once(PGC *cache, Word_t section, Word_t metric_id, time_t start_time_s, PGC_SEARCH method) {
1580 +static ALWAYS_INLINE PGC_PAGE *page_find_and_acquire_once(PGC *cache, Word_t section, Word_t metric_id, time_t start_time_s, PGC_SEARCH method) {
1581 PGC_PAGE *page = NULL;
1582 size_t partition = pgc_indexing_partition(cache, metric_id);
1583
@@ -2126,7 +2126,7 @@ PGC_PAGE *pgc_page_dup(PGC *cache, PGC_PAGE *page) {
2126 return page;
2127 }
2128
2129 -void pgc_page_release(PGC *cache, PGC_PAGE *page) {
2129 +ALWAYS_INLINE void pgc_page_release(PGC *cache, PGC_PAGE *page) {
2130 page_release(cache, page, is_page_clean(page));
2131 }
2132
src/database/engine/rrdengineapi.c
+1 -1
@@ -803,7 +803,7 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *smh,
803 }
804 }
805
806 -static inline bool rrdeng_load_page_next(struct storage_engine_query_handle *seqh, bool debug_this __maybe_unused) {
806 +static ALWAYS_INLINE bool rrdeng_load_page_next(struct storage_engine_query_handle *seqh, bool debug_this __maybe_unused) {
807 struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle;
808 struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric);
809
src/database/rrddim-collection.c
+1 -1
@@ -2,7 +2,7 @@
2
3 #include "rrddim-collection.h"
4
5 -void store_metric_collection_completed() {
5 +ALWAYS_INLINE void store_metric_collection_completed() {
6 pulse_queries_rrdset_collection_completed(rrdset_done_statistics_points_stored_per_tier);
7 }
8
src/libnetdata/atomics/refcount.h
+6 -6
@@ -40,7 +40,7 @@ typedef int32_t REFCOUNT;
40 ((refcount) >= REFCOUNT_DELETED && (refcount) <= -REFCOUNT_MAX))
41
42 // returns the non-usable refcount found when it fails, the final refcount when it succeeds
43 -static inline REFCOUNT WARNUNUSED refcount_acquire_advanced_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
43 +static ALWAYS_INLINE REFCOUNT WARNUNUSED refcount_acquire_advanced_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
44 REFCOUNT expected = refcount_references(refcount);
45 REFCOUNT desired;
46
@@ -60,12 +60,12 @@ static inline REFCOUNT WARNUNUSED refcount_acquire_advanced_with_trace(REFCOUNT
60 return desired;
61 }
62
63 -static inline bool WARNUNUSED refcount_acquire_with_trace(REFCOUNT *refcount, const char *func) {
63 +static ALWAYS_INLINE bool WARNUNUSED refcount_acquire_with_trace(REFCOUNT *refcount, const char *func) {
64 return REFCOUNT_ACQUIRED(refcount_acquire_advanced_with_trace(refcount, func));
65 }
66
67 // returns the number of references remaining
68 -static inline REFCOUNT refcount_release_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
68 +static ALWAYS_INLINE REFCOUNT refcount_release_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
69 REFCOUNT expected, desired;
70
71 do {
@@ -84,7 +84,7 @@ static inline REFCOUNT refcount_release_with_trace(REFCOUNT *refcount, const cha
84 }
85
86 // returns true when the item can be deleted, false when the item is currently referenced
87 -static inline bool WARNUNUSED refcount_acquire_for_deletion_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
87 +static ALWAYS_INLINE bool WARNUNUSED refcount_acquire_for_deletion_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
88 REFCOUNT expected = 0;
89 REFCOUNT desired = REFCOUNT_DELETED;
90
@@ -97,7 +97,7 @@ static inline bool WARNUNUSED refcount_acquire_for_deletion_with_trace(REFCOUNT
97 return false;
98 }
99
100 -static inline bool WARNUNUSED refcount_release_and_acquire_for_deletion_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
100 +static ALWAYS_INLINE bool WARNUNUSED refcount_release_and_acquire_for_deletion_with_trace(REFCOUNT *refcount, const char *func __maybe_unused) {
101 REFCOUNT expected, desired;
102
103 do {
@@ -123,7 +123,7 @@ static inline bool WARNUNUSED refcount_release_and_acquire_for_deletion_with_tra
123 // this sleeps for 1 nanosecond (posix systems), or Sleep(0) on Windows
124 void tinysleep(void);
125
126 -static inline bool refcount_acquire_for_deletion_and_wait_with_trace(REFCOUNT *refcount, const char *func) {
126 +static ALWAYS_INLINE bool refcount_acquire_for_deletion_and_wait_with_trace(REFCOUNT *refcount, const char *func) {
127 REFCOUNT expected = refcount_references(refcount);
128 REFCOUNT desired;
129
src/libnetdata/buffer/buffer.c
+4 -4
@@ -2,13 +2,13 @@
2
3 #include "../libnetdata.h"
4
5 -static inline void buffer_overflow_init(BUFFER *b)
5 +static ALWAYS_INLINE void buffer_overflow_init(BUFFER *b)
6 {
7 b->buffer[b->size] = '\0';
8 strcpy(&b->buffer[b->size + 1], BUFFER_OVERFLOW_EOF);
9 }
10
11 -void buffer_reset(BUFFER *wb) {
11 +ALWAYS_INLINE void buffer_reset(BUFFER *wb) {
12 buffer_flush(wb);
13
14 wb->content_type = CT_TEXT_PLAIN;
@@ -31,7 +31,7 @@ void buffer_char_replace(BUFFER *wb, char from, char to) {
31 buffer_overflow_check(wb);
32 }
33
34 -void buffer_print_sn_flags(BUFFER *wb, SN_FLAGS flags, bool send_anomaly_bit) {
34 +ALWAYS_INLINE void buffer_print_sn_flags(BUFFER *wb, SN_FLAGS flags, bool send_anomaly_bit) {
35 if(unlikely(flags == SN_EMPTY_SLOT)) {
36 buffer_fast_strcat(wb, "E", 1);
37 return;
@@ -90,7 +90,7 @@ void buffer_snprintf(BUFFER *wb, size_t len, const char *fmt, ...)
90 // the buffer is \0 terminated by vsnprintfz
91 }
92
93 -inline void buffer_vsprintf(BUFFER *wb, const char *fmt, va_list args) {
93 +void buffer_vsprintf(BUFFER *wb, const char *fmt, va_list args) {
94 if(unlikely(!fmt || !*fmt)) return;
95
96 size_t full_size_bytes = 0, need = 2, space_remaining = 0;
src/libnetdata/buffer/buffer.h
+45 -45
@@ -84,7 +84,7 @@ static inline void _buffer_overflow_check(BUFFER *b __maybe_unused) {
84 "BUFFER: detected overflow.");
85 }
86
87 -static inline void buffer_flush(BUFFER *wb) {
87 +static ALWAYS_INLINE void buffer_flush(BUFFER *wb) {
88 wb->len = 0;
89
90 wb->json.depth = 0;
@@ -117,7 +117,7 @@ void buffer_char_replace(BUFFER *wb, char from, char to);
117
118 void buffer_print_sn_flags(BUFFER *wb, SN_FLAGS flags, bool send_anomaly_bit);
119
120 -static inline void buffer_need_bytes(BUFFER *buffer, size_t needed_free_size) {
120 +static ALWAYS_INLINE void buffer_need_bytes(BUFFER *buffer, size_t needed_free_size) {
121 if(unlikely(buffer->len + needed_free_size >= buffer->size))
122 buffer_increase(buffer, needed_free_size + 1);
123 }
@@ -127,7 +127,7 @@ void buffer_json_initialize(BUFFER *wb, const char *key_quote, const char *value
127
128 void buffer_json_finalize(BUFFER *wb);
129
130 -static const char *buffer_tostring(BUFFER *wb)
130 +static ALWAYS_INLINE const char *buffer_tostring(BUFFER *wb)
131 {
132 if(unlikely(!wb))
133 return NULL;
@@ -140,7 +140,7 @@ static const char *buffer_tostring(BUFFER *wb)
140 return(wb->buffer);
141 }
142
143 -static inline void _buffer_json_depth_push(BUFFER *wb, BUFFER_JSON_NODE_TYPE type) {
143 +static ALWAYS_INLINE void _buffer_json_depth_push(BUFFER *wb, BUFFER_JSON_NODE_TYPE type) {
144 #ifdef NETDATA_INTERNAL_CHECKS
145 assert(wb->json.depth <= BUFFER_JSON_MAX_DEPTH && "BUFFER JSON: max nesting reached");
146 #endif
@@ -152,18 +152,18 @@ static inline void _buffer_json_depth_push(BUFFER *wb, BUFFER_JSON_NODE_TYPE typ
152 wb->json.stack[wb->json.depth].type = type;
153 }
154
155 -static inline void _buffer_json_depth_pop(BUFFER *wb) {
155 +static ALWAYS_INLINE void _buffer_json_depth_pop(BUFFER *wb) {
156 wb->json.depth--;
157 }
158
159 -static inline void buffer_putc(BUFFER *wb, char c) {
159 +static ALWAYS_INLINE void buffer_putc(BUFFER *wb, char c) {
160 buffer_need_bytes(wb, 2);
161 wb->buffer[wb->len++] = c;
162 wb->buffer[wb->len] = '\0';
163 buffer_overflow_check(wb);
164 }
165
166 -static inline void buffer_fast_rawcat(BUFFER *wb, const char *txt, size_t len) {
166 +static ALWAYS_INLINE void buffer_fast_rawcat(BUFFER *wb, const char *txt, size_t len) {
167 if(unlikely(!txt || !*txt || !len)) return;
168
169 buffer_need_bytes(wb, len + 1);
@@ -182,7 +182,7 @@ static inline void buffer_fast_rawcat(BUFFER *wb, const char *txt, size_t len) {
182 buffer_overflow_check(wb);
183 }
184
185 -static inline void buffer_fast_strcat(BUFFER *wb, const char *txt, size_t len) {
185 +static ALWAYS_INLINE void buffer_fast_strcat(BUFFER *wb, const char *txt, size_t len) {
186 if(unlikely(!txt || !*txt || !len)) return;
187
188 buffer_need_bytes(wb, len + 1);
@@ -209,7 +209,7 @@ static inline void buffer_fast_strcat(BUFFER *wb, const char *txt, size_t len) {
209 buffer_overflow_check(wb);
210 }
211
212 -static inline void buffer_strcat(BUFFER *wb, const char *txt) {
212 +static ALWAYS_INLINE void buffer_strcat(BUFFER *wb, const char *txt) {
213 if(unlikely(!txt || !*txt)) return;
214
215 const char *t = txt;
@@ -231,7 +231,7 @@ static inline void buffer_strcat(BUFFER *wb, const char *txt) {
231 buffer_overflow_check(wb);
232 }
233
234 -static inline void buffer_contents_replace(BUFFER *wb, const char *txt, size_t len) {
234 +static ALWAYS_INLINE void buffer_contents_replace(BUFFER *wb, const char *txt, size_t len) {
235 wb->len = 0;
236 buffer_need_bytes(wb, len + 1);
237
@@ -242,7 +242,7 @@ static inline void buffer_contents_replace(BUFFER *wb, const char *txt, size_t l
242 buffer_overflow_check(wb);
243 }
244
245 -static inline void buffer_strncat(BUFFER *wb, const char *txt, size_t len) {
245 +static ALWAYS_INLINE void buffer_strncat(BUFFER *wb, const char *txt, size_t len) {
246 if(unlikely(!txt || !*txt)) return;
247
248 buffer_need_bytes(wb, len + 1);
@@ -255,7 +255,7 @@ static inline void buffer_strncat(BUFFER *wb, const char *txt, size_t len) {
255 buffer_overflow_check(wb);
256 }
257
258 -static inline void buffer_memcat(BUFFER *wb, const void *mem, size_t bytes) {
258 +static ALWAYS_INLINE void buffer_memcat(BUFFER *wb, const void *mem, size_t bytes) {
259 if(unlikely(!mem)) return;
260
261 buffer_need_bytes(wb, bytes + 1);
@@ -268,7 +268,7 @@ static inline void buffer_memcat(BUFFER *wb, const void *mem, size_t bytes) {
268 buffer_overflow_check(wb);
269 }
270
271 -static inline void buffer_json_strcat(BUFFER *wb, const char *txt)
271 +static ALWAYS_INLINE void buffer_json_strcat(BUFFER *wb, const char *txt)
272 {
273 if(unlikely(!txt || !*txt)) return;
274
@@ -333,7 +333,7 @@ static inline void buffer_json_strcat(BUFFER *wb, const char *txt)
333 buffer_overflow_check(wb);
334 }
335
336 -static inline void buffer_json_quoted_strcat(BUFFER *wb, const char *txt) {
336 +static ALWAYS_INLINE void buffer_json_quoted_strcat(BUFFER *wb, const char *txt) {
337 if(unlikely(!txt || !*txt)) return;
338
339 if(*txt == '"')
@@ -372,13 +372,13 @@ static inline void buffer_json_quoted_strcat(BUFFER *wb, const char *txt) {
372 // point the remaining value fits in 32 bits, and then calls
373 // print_number_lu_r() to print the rest with 32 bit arithmetic.
374
375 -static inline char *print_uint32_reversed(char *dst, uint32_t value) {
375 +static ALWAYS_INLINE char *print_uint32_reversed(char *dst, uint32_t value) {
376 char *d = dst;
377 do *d++ = (char)('0' + (value % 10)); while((value /= 10));
378 return d;
379 }
380
381 -static inline char *print_uint64_reversed(char *dst, uint64_t value) {
381 +static ALWAYS_INLINE char *print_uint64_reversed(char *dst, uint64_t value) {
382 #ifdef ENV32BIT
383 if(value <= (uint64_t)0xffffffff)
384 return print_uint32_reversed(dst, value);
@@ -394,14 +394,14 @@ static inline char *print_uint64_reversed(char *dst, uint64_t value) {
394 #endif
395 }
396
397 -static inline char *print_uint32_hex_reversed(char *dst, uint32_t value) {
397 +static ALWAYS_INLINE char *print_uint32_hex_reversed(char *dst, uint32_t value) {
398 static const char *digits = "0123456789ABCDEF";
399 char *d = dst;
400 do *d++ = digits[value & 0xf]; while((value >>= 4));
401 return d;
402 }
403
404 -static inline char *print_uint64_hex_reversed(char *dst, uint64_t value) {
404 +static ALWAYS_INLINE char *print_uint64_hex_reversed(char *dst, uint64_t value) {
405 #ifdef ENV32BIT
406 if(value <= (uint64_t)0xffffffff)
407 return print_uint32_hex_reversed(dst, value);
@@ -417,7 +417,7 @@ static inline char *print_uint64_hex_reversed(char *dst, uint64_t value) {
417 #endif
418 }
419
420 -static inline char *print_uint64_hex_reversed_full(char *dst, uint64_t value) {
420 +static ALWAYS_INLINE char *print_uint64_hex_reversed_full(char *dst, uint64_t value) {
421 char *d = dst;
422 for(size_t c = 0; c < sizeof(uint64_t) * 2; c++) {
423 *d++ = hex_digits[value & 0xf];
@@ -427,19 +427,19 @@ static inline char *print_uint64_hex_reversed_full(char *dst, uint64_t value) {
427 return d;
428 }
429
430 -static inline char *print_uint64_base64_reversed(char *dst, uint64_t value) {
430 +static ALWAYS_INLINE char *print_uint64_base64_reversed(char *dst, uint64_t value) {
431 char *d = dst;
432 do *d++ = base64_digits[value & 63]; while ((value >>= 6));
433 return d;
434 }
435
436 -static inline void char_array_reverse(char *from, char *to) {
436 +static ALWAYS_INLINE void char_array_reverse(char *from, char *to) {
437 // from and to are inclusive
438 char *begin = from, *end = to, aux;
439 while (end > begin) aux = *end, *end-- = *begin, *begin++ = aux;
440 }
441
442 -static inline int print_netdata_double(char *dst, NETDATA_DOUBLE value) {
442 +static ALWAYS_INLINE int print_netdata_double(char *dst, NETDATA_DOUBLE value) {
443 char *s = dst;
444
445 if(unlikely(value < 0)) {
@@ -502,7 +502,7 @@ static inline int print_netdata_double(char *dst, NETDATA_DOUBLE value) {
502 return (int)(d - dst);
503 }
504
505 -static inline size_t print_uint64(char *dst, uint64_t value) {
505 +static ALWAYS_INLINE size_t print_uint64(char *dst, uint64_t value) {
506 char *s = dst;
507 char *d = print_uint64_reversed(s, value);
508 char_array_reverse(s, d - 1);
@@ -510,7 +510,7 @@ static inline size_t print_uint64(char *dst, uint64_t value) {
510 return d - s;
511 }
512
513 -static inline size_t print_int64(char *dst, int64_t value) {
513 +static ALWAYS_INLINE size_t print_int64(char *dst, int64_t value) {
514 size_t len = 0;
515
516 if(value < 0) {
@@ -523,20 +523,20 @@ static inline size_t print_int64(char *dst, int64_t value) {
523 }
524
525 #define UINT64_MAX_LENGTH (24) // 21 should be enough
526 -static inline void buffer_print_uint64(BUFFER *wb, uint64_t value) {
526 +static ALWAYS_INLINE void buffer_print_uint64(BUFFER *wb, uint64_t value) {
527 buffer_need_bytes(wb, UINT64_MAX_LENGTH);
528 wb->len += print_uint64(&wb->buffer[wb->len], value);
529 buffer_overflow_check(wb);
530 }
531
532 -static inline void buffer_print_int64(BUFFER *wb, int64_t value) {
532 +static ALWAYS_INLINE void buffer_print_int64(BUFFER *wb, int64_t value) {
533 buffer_need_bytes(wb, UINT64_MAX_LENGTH);
534 wb->len += print_int64(&wb->buffer[wb->len], value);
535 buffer_overflow_check(wb);
536 }
537
538 #define UINT64_HEX_MAX_LENGTH ((sizeof(HEX_PREFIX) - 1) + (sizeof(uint64_t) * 2) + 1)
539 -static inline size_t print_uint64_hex(char *dst, uint64_t value) {
539 +static ALWAYS_INLINE size_t print_uint64_hex(char *dst, uint64_t value) {
540 char *d = dst;
541
542 const char *s = HEX_PREFIX;
@@ -548,7 +548,7 @@ static inline size_t print_uint64_hex(char *dst, uint64_t value) {
548 return e - dst;
549 }
550
551 -static inline size_t print_uint64_hex_full(char *dst, uint64_t value) {
551 +static ALWAYS_INLINE size_t print_uint64_hex_full(char *dst, uint64_t value) {
552 char *d = dst;
553
554 const char *s = HEX_PREFIX;
@@ -560,20 +560,20 @@ static inline size_t print_uint64_hex_full(char *dst, uint64_t value) {
560 return e - dst;
561 }
562
563 -static inline void buffer_print_uint64_hex(BUFFER *wb, uint64_t value) {
563 +static ALWAYS_INLINE void buffer_print_uint64_hex(BUFFER *wb, uint64_t value) {
564 buffer_need_bytes(wb, UINT64_HEX_MAX_LENGTH);
565 wb->len += print_uint64_hex(&wb->buffer[wb->len], value);
566 buffer_overflow_check(wb);
567 }
568
569 -static inline void buffer_print_uint64_hex_full(BUFFER *wb, uint64_t value) {
569 +static ALWAYS_INLINE void buffer_print_uint64_hex_full(BUFFER *wb, uint64_t value) {
570 buffer_need_bytes(wb, UINT64_HEX_MAX_LENGTH);
571 wb->len += print_uint64_hex_full(&wb->buffer[wb->len], value);
572 buffer_overflow_check(wb);
573 }
574
575 #define UINT64_B64_MAX_LENGTH ((sizeof(IEEE754_UINT64_B64_PREFIX) - 1) + (sizeof(uint64_t) * 2) + 1)
576 -static inline void buffer_print_uint64_base64(BUFFER *wb, uint64_t value) {
576 +static ALWAYS_INLINE void buffer_print_uint64_base64(BUFFER *wb, uint64_t value) {
577 buffer_need_bytes(wb, UINT64_B64_MAX_LENGTH);
578
579 buffer_fast_strcat(wb, IEEE754_UINT64_B64_PREFIX, sizeof(IEEE754_UINT64_B64_PREFIX) - 1);
@@ -587,7 +587,7 @@ static inline void buffer_print_uint64_base64(BUFFER *wb, uint64_t value) {
587 buffer_overflow_check(wb);
588 }
589
590 -static inline void buffer_print_int64_hex(BUFFER *wb, int64_t value) {
590 +static ALWAYS_INLINE void buffer_print_int64_hex(BUFFER *wb, int64_t value) {
591 buffer_need_bytes(wb, 2);
592
593 if(value < 0) {
@@ -600,7 +600,7 @@ static inline void buffer_print_int64_hex(BUFFER *wb, int64_t value) {
600 buffer_overflow_check(wb);
601 }
602
603 -static inline void buffer_print_int64_base64(BUFFER *wb, int64_t value) {
603 +static ALWAYS_INLINE void buffer_print_int64_base64(BUFFER *wb, int64_t value) {
604 buffer_need_bytes(wb, 2);
605
606 if(value < 0) {
@@ -614,7 +614,7 @@ static inline void buffer_print_int64_base64(BUFFER *wb, int64_t value) {
614 }
615
616 #define DOUBLE_MAX_LENGTH (512) // 318 should be enough, including null
617 -static inline void buffer_print_netdata_double(BUFFER *wb, NETDATA_DOUBLE value) {
617 +static ALWAYS_INLINE void buffer_print_netdata_double(BUFFER *wb, NETDATA_DOUBLE value) {
618 buffer_need_bytes(wb, DOUBLE_MAX_LENGTH);
619
620 if(isnan(value) || isinf(value)) {
@@ -632,7 +632,7 @@ static inline void buffer_print_netdata_double(BUFFER *wb, NETDATA_DOUBLE value)
632 }
633
634 #define DOUBLE_HEX_MAX_LENGTH ((sizeof(IEEE754_DOUBLE_HEX_PREFIX) - 1) + (sizeof(uint64_t) * 2) + 1)
635 -static inline void buffer_print_netdata_double_hex(BUFFER *wb, NETDATA_DOUBLE value) {
635 +static ALWAYS_INLINE void buffer_print_netdata_double_hex(BUFFER *wb, NETDATA_DOUBLE value) {
636 buffer_need_bytes(wb, DOUBLE_HEX_MAX_LENGTH);
637
638 uint64_t *ptr = (uint64_t *) (&value);
@@ -648,7 +648,7 @@ static inline void buffer_print_netdata_double_hex(BUFFER *wb, NETDATA_DOUBLE va
648 }
649
650 #define DOUBLE_B64_MAX_LENGTH ((sizeof(IEEE754_DOUBLE_B64_PREFIX) - 1) + (sizeof(uint64_t) * 2) + 1)
651 -static inline void buffer_print_netdata_double_base64(BUFFER *wb, NETDATA_DOUBLE value) {
651 +static ALWAYS_INLINE void buffer_print_netdata_double_base64(BUFFER *wb, NETDATA_DOUBLE value) {
652 buffer_need_bytes(wb, DOUBLE_B64_MAX_LENGTH);
653
654 uint64_t *ptr = (uint64_t *) (&value);
@@ -669,7 +669,7 @@ typedef enum {
669 NUMBER_ENCODING_BASE64,
670 } NUMBER_ENCODING;
671
672 -static inline void buffer_print_int64_encoded(BUFFER *wb, NUMBER_ENCODING encoding, int64_t value) {
672 +static ALWAYS_INLINE void buffer_print_int64_encoded(BUFFER *wb, NUMBER_ENCODING encoding, int64_t value) {
673 if(encoding == NUMBER_ENCODING_BASE64)
674 return buffer_print_int64_base64(wb, value);
675
@@ -679,7 +679,7 @@ static inline void buffer_print_int64_encoded(BUFFER *wb, NUMBER_ENCODING encodi
679 return buffer_print_int64(wb, value);
680 }
681
682 -static inline void buffer_print_uint64_encoded(BUFFER *wb, NUMBER_ENCODING encoding, uint64_t value) {
682 +static ALWAYS_INLINE void buffer_print_uint64_encoded(BUFFER *wb, NUMBER_ENCODING encoding, uint64_t value) {
683 if(encoding == NUMBER_ENCODING_BASE64)
684 return buffer_print_uint64_base64(wb, value);
685
@@ -689,7 +689,7 @@ static inline void buffer_print_uint64_encoded(BUFFER *wb, NUMBER_ENCODING encod
689 return buffer_print_uint64(wb, value);
690 }
691
692 -static inline void buffer_print_netdata_double_encoded(BUFFER *wb, NUMBER_ENCODING encoding, NETDATA_DOUBLE value) {
692 +static ALWAYS_INLINE void buffer_print_netdata_double_encoded(BUFFER *wb, NUMBER_ENCODING encoding, NETDATA_DOUBLE value) {
693 if(encoding == NUMBER_ENCODING_BASE64)
694 return buffer_print_netdata_double_base64(wb, value);
695
@@ -699,7 +699,7 @@ static inline void buffer_print_netdata_double_encoded(BUFFER *wb, NUMBER_ENCODI
699 return buffer_print_netdata_double(wb, value);
700 }
701
702 -static inline void buffer_print_spaces(BUFFER *wb, size_t spaces) {
702 +static ALWAYS_INLINE void buffer_print_spaces(BUFFER *wb, size_t spaces) {
703 buffer_need_bytes(wb, spaces * 4 + 1);
704
705 char *d = &wb->buffer[wb->len];
@@ -716,12 +716,12 @@ static inline void buffer_print_spaces(BUFFER *wb, size_t spaces) {
716 buffer_overflow_check(wb);
717 }
718
719 -static inline void buffer_print_json_comma(BUFFER *wb) {
719 +static ALWAYS_INLINE void buffer_print_json_comma(BUFFER *wb) {
720 if(wb->json.stack[wb->json.depth].count)
721 buffer_fast_strcat(wb, ",", 1);
722 }
723
724 -static inline void buffer_print_json_comma_newline_spacing(BUFFER *wb) {
724 +static ALWAYS_INLINE void buffer_print_json_comma_newline_spacing(BUFFER *wb) {
725 buffer_print_json_comma(wb);
726
727 if((wb->json.options & BUFFER_JSON_OPTIONS_MINIFY) ||
@@ -732,13 +732,13 @@ static inline void buffer_print_json_comma_newline_spacing(BUFFER *wb) {
732 buffer_print_spaces(wb, wb->json.depth + 1);
733 }
734
735 -static inline void buffer_print_json_key(BUFFER *wb, const char *key) {
735 +static ALWAYS_INLINE void buffer_print_json_key(BUFFER *wb, const char *key) {
736 buffer_strcat(wb, wb->json.key_quote);
737 buffer_json_strcat(wb, key);
738 buffer_strcat(wb, wb->json.key_quote);
739 }
740
741 -static inline void buffer_json_add_string_value(BUFFER *wb, const char *value) {
741 +static ALWAYS_INLINE void buffer_json_add_string_value(BUFFER *wb, const char *value) {
742 if(value) {
743 buffer_strcat(wb, wb->json.value_quote);
744 buffer_json_strcat(wb, value);
@@ -748,7 +748,7 @@ static inline void buffer_json_add_string_value(BUFFER *wb, const char *value) {
748 buffer_fast_strcat(wb, "null", 4);
749 }
750
751 -static inline void buffer_json_add_quoted_string_value(BUFFER *wb, const char *value) {
751 +static ALWAYS_INLINE void buffer_json_add_quoted_string_value(BUFFER *wb, const char *value) {
752 if(value) {
753 buffer_strcat(wb, wb->json.value_quote);
754 buffer_json_quoted_strcat(wb, value);
src/libnetdata/inlined.h
+3 -3
@@ -351,7 +351,7 @@ static inline NETDATA_DOUBLE str2ndd(const char *src, char **endptr) {
351 return sign * result;
352 }
353
354 -static inline unsigned long long str2ull_encoded(const char *s) {
354 +static ALWAYS_INLINE unsigned long long str2ull_encoded(const char *s) {
355 if(*s == IEEE754_UINT64_B64_PREFIX[0])
356 return str2uint64_base64(s + sizeof(IEEE754_UINT64_B64_PREFIX) - 1, NULL);
357
@@ -361,14 +361,14 @@ static inline unsigned long long str2ull_encoded(const char *s) {
361 return str2uint64_t(s, NULL);
362 }
363
364 -static inline long long str2ll_encoded(const char *s) {
364 +static ALWAYS_INLINE long long str2ll_encoded(const char *s) {
365 if(*s == '-')
366 return -(long long) str2ull_encoded(&s[1]);
367 else
368 return (long long) str2ull_encoded(s);
369 }
370
371 -static inline NETDATA_DOUBLE str2ndd_encoded(const char *src, char **endptr) {
371 +static ALWAYS_INLINE NETDATA_DOUBLE str2ndd_encoded(const char *src, char **endptr) {
372 if (*src == IEEE754_DOUBLE_B64_PREFIX[0]) {
373 // double parsing from base64
374 uint64_t n = str2uint64_base64(src + sizeof(IEEE754_DOUBLE_B64_PREFIX) - 1, endptr);
src/libnetdata/line_splitter/line_splitter.h
+2 -2
@@ -31,7 +31,7 @@ extern bool isspace_map_config[256];
31 extern bool isspace_map_group_by_label[256];
32 extern bool isspace_dyncfg_id_map[256];
33
34 -static inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words, bool *isspace_map) {
34 +static ALWAYS_INLINE size_t quoted_strings_splitter(char *str, char **words, size_t max_words, bool *isspace_map) {
35 char *s = str, quote = 0;
36 size_t i = 0;
37
@@ -123,7 +123,7 @@ static inline size_t quoted_strings_splitter(char *str, char **words, size_t max
123 #define quoted_strings_splitter_dyncfg_id(str, words, max_words) \
124 quoted_strings_splitter(str, words, max_words, isspace_dyncfg_id_map)
125
126 -static inline char *get_word(char **words, size_t num_words, size_t index) {
126 +static ALWAYS_INLINE char *get_word(char **words, size_t num_words, size_t index) {
127 if (unlikely(index >= num_words))
128 return NULL;
129
src/ml/ml.cc
+1 -1
@@ -761,8 +761,8 @@ ml_dimension_predict(ml_dimension_t *dim, calculated_number_t value, bool exists
761 continue;
762
763 if (anomaly_score < (100 * Cfg.dimension_anomaly_score_threshold)) {
764 - pulse_ml_models_consulted(models_consulted);
764 spinlock_unlock(&dim->slock);
765 + pulse_ml_models_consulted(models_consulted);
766 return false;
767 }
768
src/ml/ml_public.cc
+3 -3
@@ -242,9 +242,9 @@ void ml_chart_delete(RRDSET *rs)
242 rs->ml_chart = NULL;
243 }
244
245 -bool ml_chart_update_begin(RRDSET *rs)
245 +ALWAYS_INLINE_ONLY bool ml_chart_update_begin(RRDSET *rs)
246 {
247 - ml_chart_t *chart = (ml_chart_t *) rs->ml_chart;
247 + ml_chart_t *chart = (ml_chart_t *)rs->ml_chart;
248 if (!chart)
249 return false;
250
@@ -320,7 +320,7 @@ void ml_dimension_delete(RRDDIM *rd)
320 rd->ml_dimension = NULL;
321 }
322
323 -void ml_dimension_received_anomaly(RRDDIM *rd, bool is_anomalous) {
323 +ALWAYS_INLINE_ONLY void ml_dimension_received_anomaly(RRDDIM *rd, bool is_anomalous) {
324 ml_dimension_t *dim = (ml_dimension_t *) rd->ml_dimension;
325 if (!dim)
326 return;
src/plugins.d/pluginsd_internals.h
+7 -7
@@ -15,7 +15,7 @@ PARSER_RC PLUGINSD_DISABLE_PLUGIN(PARSER *parser, const char *keyword, const cha
15
16 ssize_t send_to_plugin(const char *txt, PARSER *parser, STREAM_TRAFFIC_TYPE type);
17
18 -static inline RRDHOST *pluginsd_require_scope_host(PARSER *parser, const char *cmd) {
18 +static ALWAYS_INLINE RRDHOST *pluginsd_require_scope_host(PARSER *parser, const char *cmd) {
19 RRDHOST *host = parser->user.host;
20
21 if(unlikely(!host))
@@ -24,7 +24,7 @@ static inline RRDHOST *pluginsd_require_scope_host(PARSER *parser, const char *c
24 return host;
25 }
26
27 -static inline RRDSET *pluginsd_require_scope_chart(PARSER *parser, const char *cmd, const char *parent_cmd) {
27 +static ALWAYS_INLINE RRDSET *pluginsd_require_scope_chart(PARSER *parser, const char *cmd, const char *parent_cmd) {
28 RRDSET *st = parser->user.st;
29
30 if(unlikely(!st))
@@ -57,7 +57,7 @@ static inline bool rrdset_data_collection_unlock_with_trace(PARSER *parser, cons
57 #define rrdset_data_collection_lock(parser) rrdset_data_collection_lock_with_trace(parser, __FUNCTION__)
58 #define rrdset_data_collection_unlock(parser) rrdset_data_collection_unlock_with_trace(parser, __FUNCTION__)
59
60 -static inline void rrdset_previous_scope_chart_unlock(PARSER *parser, const char *keyword, bool stale) {
60 +static ALWAYS_INLINE void rrdset_previous_scope_chart_unlock(PARSER *parser, const char *keyword, bool stale) {
61 if(unlikely(rrdset_data_collection_unlock(parser))) {
62 if(stale)
63 netdata_log_error("PLUGINSD: 'host:%s/chart:%s/' stale data collection lock found during %s; it has been unlocked",
@@ -88,7 +88,7 @@ static inline void pluginsd_clear_scope_chart(PARSER *parser, const char *keywor
88 parser->user.cleanup_slots = false;
89 }
90
91 -static inline bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, const char *keyword) {
91 +static ALWAYS_INLINE bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, const char *keyword) {
92 RRDSET *old_st = parser->user.st;
93 pid_t old_collector_tid = (old_st) ? old_st->pluginsd.collector_tid : 0;
94 pid_t my_collector_tid = gettid_cached();
@@ -159,7 +159,7 @@ static inline void pluginsd_rrddim_put_to_slot(PARSER *parser, RRDSET *st, RRDDI
159 }
160 }
161
162 -static inline RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, const char *dimension, ssize_t slot, const char *cmd) {
162 +static ALWAYS_INLINE RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, const char *dimension, ssize_t slot, const char *cmd) {
163 if (unlikely(!dimension || !*dimension)) {
164 netdata_log_error("PLUGINSD: 'host:%s/chart:%s' got a %s, without a dimension.",
165 rrdhost_hostname(host), rrdset_id(st), cmd);
@@ -265,7 +265,7 @@ static inline RRDSET *pluginsd_find_chart(RRDHOST *host, const char *chart, cons
265 return st;
266 }
267
268 -static inline ssize_t pluginsd_parse_rrd_slot(char **words, size_t num_words) {
268 +static ALWAYS_INLINE ssize_t pluginsd_parse_rrd_slot(char **words, size_t num_words) {
269 ssize_t slot = -1;
270 char *id = get_word(words, num_words, 1);
271 if(id && id[0] == PLUGINSD_KEYWORD_SLOT[0] && id[1] == PLUGINSD_KEYWORD_SLOT[1] &&
@@ -311,7 +311,7 @@ static inline void pluginsd_rrdset_cache_put_to_slot(PARSER *parser, RRDSET *st,
311 parser->user.cleanup_slots = obsolete;
312 }
313
314 -static inline RRDSET *pluginsd_rrdset_cache_get_from_slot(PARSER *parser, RRDHOST *host, const char *id, ssize_t slot, const char *keyword) {
314 +static ALWAYS_INLINE RRDSET *pluginsd_rrdset_cache_get_from_slot(PARSER *parser, RRDHOST *host, const char *id, ssize_t slot, const char *keyword) {
315 if(unlikely(slot < 1 || (size_t)slot > host->stream.rcv.pluginsd_chart_slots.size))
316 return pluginsd_find_chart(host, id, keyword);
317
src/plugins.d/pluginsd_parser.c
+4 -4
@@ -698,7 +698,7 @@ static inline PARSER_RC pluginsd_clabel_commit(char **words __maybe_unused, size
698 return PARSER_RC_OK;
699 }
700
701 -static inline PARSER_RC pluginsd_begin_v2(char **words, size_t num_words, PARSER *parser) {
701 +static ALWAYS_INLINE PARSER_RC pluginsd_begin_v2(char **words, size_t num_words, PARSER *parser) {
702 timing_init();
703
704 int idx = 1;
@@ -838,7 +838,7 @@ static inline PARSER_RC pluginsd_begin_v2(char **words, size_t num_words, PARSER
838 return PARSER_RC_OK;
839 }
840
841 -static inline PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *parser) {
841 +static ALWAYS_INLINE PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *parser) {
842 timing_init();
843
844 int idx = 1;
@@ -977,7 +977,7 @@ static inline PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *
977 return PARSER_RC_OK;
978 }
979
980 -static inline PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
980 +static ALWAYS_INLINE PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
981 timing_init();
982
983 RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_END_V2);
@@ -1229,7 +1229,7 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, int fd_input,
1229
1230 #include "gperf-hashtable.h"
1231
1232 -PARSER_RC parser_execute(PARSER *parser, const PARSER_KEYWORD *keyword, char **words, size_t num_words) {
1232 +ALWAYS_INLINE PARSER_RC parser_execute(PARSER *parser, const PARSER_KEYWORD *keyword, char **words, size_t num_words) {
1233 // put all the keywords ordered by the frequency they are used
1234
1235 switch(keyword->id) {
src/plugins.d/pluginsd_replication.c
+5 -5
@@ -109,7 +109,7 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, PARSER *
109 return ok ? PARSER_RC_OK : PARSER_RC_ERROR;
110 }
111
112 -PARSER_RC pluginsd_replay_begin(char **words, size_t num_words, PARSER *parser) {
112 +ALWAYS_INLINE PARSER_RC pluginsd_replay_begin(char **words, size_t num_words, PARSER *parser) {
113 int idx = 1;
114 ssize_t slot = pluginsd_parse_rrd_slot(words, num_words);
115 if(slot >= 0) idx++;
@@ -214,7 +214,7 @@ PARSER_RC pluginsd_replay_begin(char **words, size_t num_words, PARSER *parser)
214 return PARSER_RC_OK;
215 }
216
217 -PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARSER *parser) {
217 +ALWAYS_INLINE PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARSER *parser) {
218 int idx = 1;
219 ssize_t slot = pluginsd_parse_rrd_slot(words, num_words);
220 if(slot >= 0) idx++;
@@ -290,7 +290,7 @@ PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARSER *parser) {
290 return PARSER_RC_OK;
291 }
292
293 -PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words, PARSER *parser) {
293 +ALWAYS_INLINE PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words, PARSER *parser) {
294 if(parser->user.replay.rset_enabled == false)
295 return PARSER_RC_OK;
296
@@ -333,7 +333,7 @@ PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words
333 return PARSER_RC_OK;
334 }
335
336 -PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words, PARSER *parser) {
336 +ALWAYS_INLINE PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words, PARSER *parser) {
337 if(parser->user.replay.rset_enabled == false)
338 return PARSER_RC_OK;
339
@@ -367,7 +367,7 @@ PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words
367 return PARSER_RC_OK;
368 }
369
370 -PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARSER *parser) {
370 +ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARSER *parser) {
371 if (num_words < 7) { // accepts 7, but the 7th is optional
372 nd_log(NDLS_DAEMON, NDLP_ERR, "REPLAY: malformed " PLUGINSD_KEYWORD_REPLAY_END " command");
373 return PARSER_RC_ERROR;
src/streaming/protocol/command-begin-set-end-init.c
+1 -1
@@ -11,7 +11,7 @@ static BUFFER *preferred_sender_buffer(RRDHOST *host) {
11 return sender_thread_buffer(host->sender);
12 }
13
14 -RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time) {
14 +ALWAYS_INLINE RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time) {
15 RRDHOST *host = st->rrdhost;
16
17 // fetch the flags we need to check with one atomic operation
src/streaming/protocol/command-begin-set-end-v2.c
+1 -1
@@ -65,7 +65,7 @@ void stream_send_rrddim_metrics_v2(RRDSET_STREAM_BUFFER *rsb, RRDDIM *rd, usec_t
65 buffer_fast_strcat(wb, "\n", 1);
66 }
67
68 -void stream_send_rrdset_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
68 +ALWAYS_INLINE void stream_send_rrdset_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
69 if(!rsb->wb)
70 return;
71
src/streaming/protocol/command-nodeid.c
+1 -1
@@ -12,7 +12,7 @@ void stream_sender_clear_parent_claim_id(RRDHOST *host) {
12
13 // the parent sends to the child its claim id, node id and cloud url
14 void stream_receiver_send_node_and_claim_id_to_child(RRDHOST *host) {
15 - if(host == localhost || UUIDiszero(host->node_id)) return;
15 + if(rrdhost_is_local(host) || UUIDiszero(host->node_id)) return;
16
17 rrdhost_receiver_lock(host);
18 if(stream_has_capability(host->receiver, STREAM_CAP_NODE_ID)) {
src/streaming/stream-connector.c
+73 -25
@@ -394,6 +394,8 @@ struct connector {
394 ND_THREAD *thread;
395 struct completion completion;
396
397 + Word_t idx;
398 +
399 size_t nodes;
400
401 struct {
@@ -404,6 +406,12 @@ struct connector {
406 } queue;
407 };
408
409 +static inline Word_t get_unique_idx(struct connector *cn, STRCNT_CMD cmd) {
410 + Word_t t = STRCNT_CMD_MAX - 1;
411 + Word_t reserved_bits = (sizeof(Word_t) * 8) - __builtin_clz(t);
412 + return (__atomic_add_fetch(&cn->idx, 1, __ATOMIC_RELAXED) << reserved_bits) | cmd;
413 +}
414 +
415 static struct {
416 int id;
417 struct connector connectors[MAX_CONNECTORS];
@@ -441,20 +449,29 @@ struct connector *stream_connector_get(struct sender_state *s) {
449 return sc;
450 }
451
444 -void stream_connector_requeue(struct sender_state *s) {
452 +void stream_connector_requeue(struct sender_state *s, STRCNT_CMD cmd) {
453 struct connector *sc = stream_connector_get(s);
454
447 - nd_log(NDLS_DAEMON, NDLP_DEBUG,
448 - "STREAM CONNECT '%s' [to parent]: adding host in connector queue...",
449 - rrdhost_hostname(s->host));
455 + switch(cmd) {
456 + case STRCNT_CMD_CONNECT:
457 + nd_log(NDLS_DAEMON, NDLP_DEBUG,
458 + "STREAM CONNECT '%s' [to parent]: adding host in connector queue...",
459 + rrdhost_hostname(s->host));
460 +
461 + pulse_host_status(s->host, PULSE_HOST_STATUS_SND_PENDING, 0);
462 + break;
463 +
464 + case STRCNT_CMD_REMOVE:
465 + break;
466 +
467 + default:
468 + fatal("STREAM CONNECT '%s': invalid cmd %d", rrdhost_hostname(s->host), cmd);
469 + }
470
471 spinlock_lock(&sc->queue.spinlock);
452 - internal_fatal(SENDERS_GET(&sc->queue.senders, (Word_t)s) != NULL, "Sender is already in the connector queue");
453 - SENDERS_SET(&sc->queue.senders, (Word_t)s, s);
472 + SENDERS_SET(&sc->queue.senders, get_unique_idx(sc, cmd), s);
473 spinlock_unlock(&sc->queue.spinlock);
474
456 - pulse_host_status(s->host, PULSE_HOST_STATUS_SND_PENDING, 0);
457 -
475 // signal the connector to catch the job
476 completion_mark_complete_a_job(&sc->completion);
477 }
@@ -482,7 +499,7 @@ void stream_connector_add(struct sender_state *s) {
499 s->parent_using_h2o = stream_send.parents.h2o;
500
501 // do not call this with any locks held
485 - stream_connector_requeue(s);
502 + stream_connector_requeue(s, STRCNT_CMD_CONNECT);
503 }
504
505 static void stream_connector_remove(struct sender_state *s) {
@@ -507,6 +524,7 @@ static void *stream_connector_thread(void *ptr) {
524 worker_register("STREAMCNT");
525 worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_CONNECTING, "connect");
526 worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_CONNECTED, "connected");
527 + worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_REMOVED, "removed");
528 worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_BAD_HANDSHAKE, "bad handshake");
529 worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_TIMEOUT, "timeout");
530 worker_register_job_name(WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_CANT_UPGRADE_CONNECTION, "cant upgrade");
@@ -539,31 +557,61 @@ static void *stream_connector_thread(void *ptr) {
557
558 if(stream_connector_is_signaled_to_stop(s)) {
559 cancelled_nodes++;
542 - SENDERS_DEL(&sc->queue.senders, (Word_t)s);
560 + SENDERS_DEL(&sc->queue.senders, idx);
561 + spinlock_unlock(&sc->queue.spinlock);
562 +
563 + // do not have the connector lock when calling these
564 stream_connector_remove(s);
565 +
566 + spinlock_lock(&sc->queue.spinlock);
567 continue;
568 }
569
547 - spinlock_unlock(&sc->queue.spinlock);
548 - worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_CONNECTING);
549 - bool move_to_sender = stream_connect(s, stream_send.parents.default_port, stream_send.parents.timeout_s);
550 - spinlock_lock(&sc->queue.spinlock);
570 + STRCNT_CMD cmd = idx & (STRCNT_CMD_CONNECT| STRCNT_CMD_REMOVE);
571 + switch(cmd) {
572 + case STRCNT_CMD_CONNECT:
573 + spinlock_unlock(&sc->queue.spinlock);
574 + worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_CONNECTING);
575
552 - if(move_to_sender) {
553 - connected_nodes++;
554 - stream_sender_on_connect(s);
576 + // do not have the connector lock when calling these
577 + bool move_to_sender =
578 + stream_connect(s, stream_send.parents.default_port, stream_send.parents.timeout_s);
579
556 - worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_CONNECTED);
557 - SENDERS_DEL(&sc->queue.senders, (Word_t)s);
558 - spinlock_unlock(&sc->queue.spinlock);
580 + spinlock_lock(&sc->queue.spinlock);
581
560 - // do not have the connector lock when calling this
561 - stream_sender_add_to_queue(s);
582 + if (move_to_sender) {
583 + connected_nodes++;
584
563 - spinlock_lock(&sc->queue.spinlock);
585 + worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_CONNECTED);
586 + SENDERS_DEL(&sc->queue.senders, idx);
587 + spinlock_unlock(&sc->queue.spinlock);
588 +
589 + // do not have the connector lock when calling these
590 + stream_sender_on_connect(s);
591 + stream_sender_add_to_queue(s);
592 +
593 + spinlock_lock(&sc->queue.spinlock);
594 + }
595 + else
596 + failed_nodes++;
597 +
598 + break;
599 +
600 + case STRCNT_CMD_REMOVE:
601 + worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_REMOVED);
602 + SENDERS_DEL(&sc->queue.senders, idx);
603 + spinlock_unlock(&sc->queue.spinlock);
604 +
605 + // do not have the connector lock when calling these
606 + stream_sender_on_disconnect(s);
607 + stream_sender_remove(s, s->exit.reason);
608 +
609 + spinlock_lock(&sc->queue.spinlock);
610 + break;
611 +
612 + default:
613 + fatal("STREAM CONNECT '%s': invalid cmd %d", rrdhost_hostname(s->host), cmd);
614 }
565 - else
566 - failed_nodes++;
615
616 worker_is_idle();
617 }
src/streaming/stream-receiver.c
+27 -33
@@ -291,9 +291,11 @@ static inline ssize_t receiver_read_compressed(struct receiver_state *r) {
291
292 // --------------------------------------------------------------------------------------------------------------------
293
294 -static void receiver_set_exit_reason(struct receiver_state *rpt, STREAM_HANDSHAKE reason, bool force) {
294 +static STREAM_HANDSHAKE receiver_set_exit_reason(struct receiver_state *rpt, STREAM_HANDSHAKE reason, bool force) {
295 if(force || !rpt->exit.reason)
296 rpt->exit.reason = reason;
297 +
298 + return rpt->exit.reason;
299 }
300
301 static inline bool receiver_should_stop(struct receiver_state *rpt) {
@@ -516,6 +518,8 @@ void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *s
518 static void stream_receiver_remove(struct stream_thread *sth, struct receiver_state *rpt, STREAM_HANDSHAKE reason) {
519 internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
520
521 + receiver_set_exit_reason(rpt, reason, false);
522 +
523 ND_LOG_STACK lgs[] = {
524 ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),
525 ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),
@@ -593,9 +597,11 @@ static bool stream_receiver_dequeue_senders(struct stream_thread *sth, struct re
597
598 if(rpt->host->sender && // the host has a sender
599 rpt->host->stream.snd.status.tid == gettid_cached() && // the sender is mine
596 - (rpt->host->sender->thread.wanted & ND_POLL_WRITE)) // the sender needs to send data
597 - if(!stream_sender_send_data(sth, rpt->host->sender, now_ut, false))
598 - return false;
600 + (rpt->host->sender->thread.wanted & ND_POLL_WRITE)) { // the sender needs to send data
601 + // we return true even if this fais,
602 + // so that we will not disconnect the receiver because the sender failed
603 + stream_sender_send_data(sth, rpt->host->sender, now_ut, false);
604 + }
605
606 return true;
607 }
@@ -627,7 +633,6 @@ stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt
633
634 while (buffered_reader_next_line(&rpt->thread.uncompressed, rpt->thread.line_buffer)) {
635 if (unlikely(parser_action(parser, rpt->thread.line_buffer->buffer))) {
630 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED, false);
636 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED);
637 *removed = true;
638 return -1;
@@ -641,7 +646,6 @@ stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt
646 break;
647
648 else {
644 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED, false);
649 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_RCV_DECOMPRESSION_FAILED);
650 *removed = true;
651 return -1;
@@ -651,7 +655,6 @@ stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt
655 else if (feed_rc == DECOMPRESS_NEED_MORE_DATA)
656 break;
657 else {
654 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED, false);
658 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_RCV_DECOMPRESSION_FAILED);
659 *removed = true;
660 return -1;
@@ -659,9 +662,7 @@ stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt
662 }
663
664 if(receiver_should_stop(rpt)) {
662 - STREAM_HANDSHAKE reason = rpt->exit.reason ? rpt->exit.reason : STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP;
663 - receiver_set_exit_reason(rpt, reason, false);
664 - stream_receiver_remove(sth, rpt, reason);
665 + stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP);
666 *removed = true;
667 return -1;
668 }
@@ -673,7 +674,6 @@ stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt
674
675 while(buffered_reader_next_line(&rpt->thread.uncompressed, rpt->thread.line_buffer)) {
676 if(unlikely(parser_action(parser, rpt->thread.line_buffer->buffer))) {
676 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED, false);
677 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED);
678 *removed = true;
679 return -1;
@@ -755,27 +755,27 @@ bool stream_receiver_send_data(struct stream_thread *sth, struct receiver_state
755 sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port,
756 stream_handshake_error_to_string(reason), rc, rpt->sock.fd, stats->bytes_sent, stats->sends);
757
758 - receiver_set_exit_reason(rpt, reason, false);
759 -
758 if(process_opcodes_and_enable_removal) {
759 // this is not executed from the opcode handling mechanism
760 // so we can safely remove the receiver.
761 stream_receiver_remove(sth, rpt, reason);
762 }
763 else {
766 - // protection against this case:
767 - //
768 - // 1. receiver gets a replication request
769 - // 2. parser processes the request
770 - // 3. parser decides to send back a message to the child (REPLAY_CHART)
771 - // 4. send_to_child appends the data to the sending circular buffer
772 - // 5. send_to_child sends opcode to enable sending
773 - // 6. opcode bypasses the signal and runs this function inline to dispatch immediately
774 - // 7. sending fails (child disconnected)
775 - // 8. receiver is removed
776 - //
777 - // Point 2 above crashes. The parser is no longer there (freed at point 7)
778 - // and there is no way for point 2 to know...
764 + receiver_set_exit_reason(rpt, reason, false);
765 +
766 + // protection against this case:
767 + //
768 + // 1. receiver gets a replication request
769 + // 2. parser processes the request
770 + // 3. parser decides to send back a message to the child (REPLAY_CHART)
771 + // 4. send_to_child appends the data to the sending circular buffer
772 + // 5. send_to_child sends opcode to enable sending
773 + // 6. opcode bypasses the signal and runs this function inline to dispatch immediately
774 + // 7. sending fails (child disconnected)
775 + // 8. receiver is removed
776 + //
777 + // Point 2 above crashes. The parser is no longer there (freed at point 7)
778 + // and there is no way for point 2 to know...
779 }
780 }
781 else if(process_opcodes_and_enable_removal &&
@@ -840,7 +840,6 @@ bool stream_receiver_receive_data(struct stream_thread *sth, struct receiver_sta
840 sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port,
841 stream_handshake_error_to_string(reason), rpt->sock.fd);
842
843 - receiver_set_exit_reason(rpt, reason, false);
843 stream_receiver_remove(sth, rpt, reason);
844 }
845 else if(status == EVLOOP_STATUS_CONTINUE && process_opcodes && stream_thread_process_opcodes(sth, &rpt->thread.meta))
@@ -866,9 +865,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv
865 ND_LOG_STACK_PUSH(lgs);
866
867 if (receiver_should_stop(rpt)) {
869 - STREAM_HANDSHAKE reason = rpt->exit.reason ? rpt->exit.reason : STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP;
870 - receiver_set_exit_reason(rpt, reason, false);
871 - stream_receiver_remove(sth, rpt, reason);
868 + stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP);
869 return false;
870 }
871
@@ -884,7 +881,6 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv
881 sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port,
882 stream_handshake_error_to_string(reason));
883
887 - receiver_set_exit_reason(rpt, reason, false);
884 stream_receiver_remove(sth, rpt, reason);
885 return false;
886 }
@@ -953,7 +949,6 @@ void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t
949 sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, timeout_s,
950 stats.bytes_sent, stats.sends, duration, pending, stats.buffer_ratio);
951
956 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_TIMEOUT, false);
952 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_TIMEOUT);
953 continue;
954 }
@@ -1044,7 +1039,6 @@ void stream_receiver_replication_check_from_poll(struct stream_thread *sth, usec
1039 __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED),
1040 __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED));
1041
1047 - receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_REPLICATION_STALLED, false);
1042 stream_receiver_remove(sth, rpt, STREAM_HANDSHAKE_DISCONNECT_REPLICATION_STALLED);
1043 }
1044
src/streaming/stream-sender-internals.h
+19 -8
@@ -13,13 +13,14 @@
13 // connector thread
14 #define WORKER_SENDER_CONNECTOR_JOB_CONNECTING 0
15 #define WORKER_SENDER_CONNECTOR_JOB_CONNECTED 1
16 -#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_BAD_HANDSHAKE 2
17 -#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_TIMEOUT 3
18 -#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_CANT_UPGRADE_CONNECTION 4
19 -#define WORKER_SENDER_CONNECTOR_JOB_QUEUED_NODES 5
20 -#define WORKER_SENDER_CONNECTOR_JOB_CONNECTED_NODES 6
21 -#define WORKER_SENDER_CONNECTOR_JOB_FAILED_NODES 7
22 -#define WORKER_SENDER_CONNECTOR_JOB_CANCELLED_NODES 8
16 +#define WORKER_SENDER_CONNECTOR_JOB_REMOVED 2
17 +#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_BAD_HANDSHAKE 3
18 +#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_TIMEOUT 4
19 +#define WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_CANT_UPGRADE_CONNECTION 5
20 +#define WORKER_SENDER_CONNECTOR_JOB_QUEUED_NODES 6
21 +#define WORKER_SENDER_CONNECTOR_JOB_CONNECTED_NODES 7
22 +#define WORKER_SENDER_CONNECTOR_JOB_FAILED_NODES 8
23 +#define WORKER_SENDER_CONNECTOR_JOB_CANCELLED_NODES 9
24
25 #define CONNECTED_TO_SIZE 100
26
@@ -143,13 +144,23 @@ void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg)
144 void stream_sender_add_to_queue(struct sender_state *s);
145
146 // stream connector
147 +typedef enum __attribute__((packed)) {
148 + STRCNT_CMD_NONE = 0,
149 + STRCNT_CMD_CONNECT,
150 + STRCNT_CMD_REMOVE,
151 +
152 + // terminator
153 + STRCNT_CMD_MAX,
154 +} STRCNT_CMD;
155 +
156 bool stream_connector_init(struct sender_state *s);
157 void stream_connector_cancel_threads(void);
158 void stream_connector_add(struct sender_state *s);
149 -void stream_connector_requeue(struct sender_state *s);
159 +void stream_connector_requeue(struct sender_state *s, STRCNT_CMD cmd);
160 bool stream_connector_is_signaled_to_stop(struct sender_state *s);
161
162 void stream_sender_on_connect(struct sender_state *s);
163 +void stream_sender_on_disconnect(struct sender_state *s);
164
165 void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason);
166
src/streaming/stream-sender.c
+28 -28
@@ -68,7 +68,7 @@ void stream_sender_log_payload(struct sender_state *s, BUFFER *payload, STREAM_T
68
69 // --------------------------------------------------------------------------------------------------------------------
70
71 -static void stream_sender_charts_and_replication_reset(struct sender_state *s) {
71 +void stream_sender_charts_and_replication_reset(struct sender_state *s) {
72 // stop all replication commands inflight
73 replication_sender_delete_pending_requests(s);
74
@@ -111,6 +111,15 @@ static void stream_sender_charts_and_replication_reset(struct sender_state *s) {
111
112 // --------------------------------------------------------------------------------------------------------------------
113
114 +static void stream_sender_on_connect_and_disconnect(struct sender_state *s) {
115 + stream_sender_execute_commands_cleanup(s);
116 + stream_sender_charts_and_replication_reset(s);
117 +
118 + stream_sender_lock(s);
119 + stream_circular_buffer_flush_unsafe(s->scb, stream_send.buffer_max_size);
120 + stream_sender_unlock(s);
121 +}
122 +
123 void stream_sender_on_connect(struct sender_state *s) {
124 nd_log(NDLS_DAEMON, NDLP_DEBUG,
125 "STREAM SND [%s]: running on-connect hooks...",
@@ -118,11 +127,7 @@ void stream_sender_on_connect(struct sender_state *s) {
127
128 rrdhost_flag_set(s->host, RRDHOST_FLAG_STREAM_SENDER_CONNECTED);
129
121 - stream_sender_charts_and_replication_reset(s);
122 -
123 - stream_sender_lock(s);
124 - stream_circular_buffer_flush_unsafe(s->scb, stream_send.buffer_max_size);
125 - stream_sender_unlock(s);
130 + stream_sender_on_connect_and_disconnect(s);
131
132 s->thread.last_traffic_ut = now_monotonic_usec();
133 s->rbuf.read_len = 0;
@@ -136,7 +141,7 @@ static void stream_sender_on_ready_to_dispatch(struct sender_state *s) {
141 // set this flag before sending any data, or the data will not be sent
142 rrdhost_flag_set(s->host, RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS);
143
139 - stream_sender_execute_commands_cleanup(s);
144 + // send our global metadata to the parent
145 stream_sender_send_custom_host_variables(s->host);
146 stream_path_send_to_parent(s->host);
147 stream_sender_send_claimed_id(s->host);
@@ -144,21 +149,16 @@ static void stream_sender_on_ready_to_dispatch(struct sender_state *s) {
149 stream_send_global_functions(s->host);
150 }
151
147 -static void stream_sender_on_disconnect(struct sender_state *s) {
152 +void stream_sender_on_disconnect(struct sender_state *s) {
153 nd_log(NDLS_DAEMON, NDLP_DEBUG,
154 "STREAM SND '%s': running on-disconnect hooks...",
155 rrdhost_hostname(s->host));
156
152 - stream_sender_lock(s);
153 - stream_circular_buffer_flush_unsafe(s->scb, stream_send.buffer_max_size);
154 - stream_sender_unlock(s);
157 + stream_sender_on_connect_and_disconnect(s);
158
156 - stream_sender_execute_commands_cleanup(s);
157 - stream_sender_charts_and_replication_reset(s);
158 - stream_sender_clear_parent_claim_id(s->host);
159 - stream_receiver_send_node_and_claim_id_to_child(s->host);
159 + // update the child (the receiver side) for this parent
160 stream_path_parent_disconnected(s->host);
161 - sender_host_buffer_free(s->host);
161 + stream_receiver_send_node_and_claim_id_to_child(s->host);
162 }
163
164 // --------------------------------------------------------------------------------------------------------------------
@@ -344,10 +344,10 @@ void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason) {
344
345 stream_sender_lock(s);
346
347 - if(reason == STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP && s->exit.reason) {
347 + if(reason == STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP && s->exit.reason)
348 reason = s->exit.reason;
349 - s->exit.reason = 0;
350 - }
349 +
350 + s->exit.reason = 0;
351
352 __atomic_store_n(&s->exit.shutdown, false, __ATOMIC_RELAXED);
353 rrdhost_flag_clear(s->host,
@@ -357,7 +357,6 @@ void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason) {
357 s->last_state_since_t = now_realtime_sec();
358 stream_parent_set_disconnect_reason(s->host->stream.snd.parents.current, reason, s->last_state_since_t);
359 s->connector.id = -1;
360 - s->exit.reason = 0;
360
361 stream_sender_unlock(s);
362
@@ -422,6 +421,10 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
421 // clear these asap, to make sender_commit() stop processing data for this host
422 stream_sender_lock(s);
423
424 + if(reason == STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP && s->exit.reason)
425 + reason = s->exit.reason;
426 +
427 + s->exit.reason = reason;
428 s->thread.msg.session = 0;
429 s->thread.msg.meta = NULL;
430
@@ -433,18 +436,15 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
436 nd_sock_close(&s->sock);
437
438 stream_parent_set_disconnect_reason(s->host->stream.snd.parents.current, reason, now_realtime_sec());
436 - stream_sender_on_disconnect(s);
439 + stream_sender_clear_parent_claim_id(s->host);
440 + sender_host_buffer_free(s->host);
441
438 - bool should_remove = !reconnect || stream_connector_is_signaled_to_stop(s);
442 + pulse_host_status(s->host, PULSE_HOST_STATUS_SND_OFFLINE, reason);
443
444 stream_thread_node_removed(s->host);
445
442 - pulse_host_status(s->host, PULSE_HOST_STATUS_SND_OFFLINE, reason);
443 -
444 - if (should_remove)
445 - stream_sender_remove(s, reason);
446 - else
447 - stream_connector_requeue(s);
446 + stream_connector_requeue(
447 + s, reconnect && !stream_connector_is_signaled_to_stop(s) ? STRCNT_CMD_CONNECT : STRCNT_CMD_REMOVE);
448 }
449
450 void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut) {