@cryptotaxi247 / netdata-1 / commits / c62dcb2a9

Optimizations part 2 (#15280)

* make all pluginsd functions inline, instead of function pointers * dynamic MRG partitions based on the number of CPUs * report the right size of the MRG * prevent invalid read on pluginsd exit * faster service_running() check; fix compiler warnings; shutdown replication after streaming to prevent crash on shutdown * sender is now using a spinlock * rrdcontext uses spinlock * replace select() with poll() * signed calculation of threads * disable read-ahead on jnfv2 files during scan

Costa Tsaousis committed Jun 29, 2023 at 15:02 UTC c62dcb2a9bd8ca6ec0c483bb3506c733d96648c6
22 files changed +385 -282
collectors/plugins.d/gperf-config.txt
+38 -29
@@ -1,43 +1,52 @@
1 +%struct-type
2 +%omit-struct-type
3 +%define hash-function-name gperf_keyword_hash_function
4 +%define lookup-function-name gperf_lookup_keyword
5 +%define word-array-name gperf_keywords
6 +%define constants-prefix GPERF_PARSER_
7 +%define slot-name keyword
8 +%global-table
9 +%null-strings
10 PARSER_KEYWORD;
11 %%
12 #
13 # Plugins Only Keywords
14 #
6 -FLUSH, pluginsd_flush, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 1
7 -DISABLE, pluginsd_disable, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 2
8 -EXIT, pluginsd_exit, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 3
9 -HOST, pluginsd_host, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 4
10 -HOST_DEFINE, pluginsd_host_define, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 5
11 -HOST_DEFINE_END, pluginsd_host_define_end, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 6
12 -HOST_LABEL, pluginsd_host_labels, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 7
15 +FLUSH, 97, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 1
16 +DISABLE, 98, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 2
17 +EXIT, 99, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 3
18 +HOST, 71, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 4
19 +HOST_DEFINE, 72, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 5
20 +HOST_DEFINE_END, 73, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 6
21 +HOST_LABEL, 74, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 7
22 #
23 # Common keywords
24 #
16 -BEGIN, pluginsd_begin, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 8
17 -CHART, pluginsd_chart, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 9
18 -CLABEL, pluginsd_clabel, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 10
19 -CLABEL_COMMIT, pluginsd_clabel_commit, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 11
20 -DIMENSION, pluginsd_dimension, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 12
21 -END, pluginsd_end, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 13
22 -FUNCTION, pluginsd_function, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 14
23 -FUNCTION_RESULT_BEGIN, pluginsd_function_result_begin, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 15
24 -LABEL, pluginsd_label, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 16
25 -OVERWRITE, pluginsd_overwrite, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 17
26 -SET, pluginsd_set, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 18
27 -VARIABLE, pluginsd_variable, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 19
25 +BEGIN, 12, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 8
26 +CHART, 32, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 9
27 +CLABEL, 34, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 10
28 +CLABEL_COMMIT, 35, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 11
29 +DIMENSION, 31, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 12
30 +END, 13, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 13
31 +FUNCTION, 41, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 14
32 +FUNCTION_RESULT_BEGIN, 42, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 15
33 +LABEL, 51, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 16
34 +OVERWRITE, 52, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 17
35 +SET, 11, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 18
36 +VARIABLE, 53, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 19
37 #
38 # Streaming only keywords
39 #
31 -CLAIMED_ID, streaming_claimed_id, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 20
32 -BEGIN2, pluginsd_begin_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 21
33 -SET2, pluginsd_set_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 22
34 -END2, pluginsd_end_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 23
40 +CLAIMED_ID, 61, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 20
41 +BEGIN2, 2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 21
42 +SET2, 1, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 22
43 +END2, 3, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 23
44 #
45 # Streaming Replication keywords
46 #
38 -CHART_DEFINITION_END, pluginsd_chart_definition_end, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 24
39 -RBEGIN, pluginsd_replay_begin, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 25
40 -RDSTATE, pluginsd_replay_rrddim_collection_state, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 26
41 -REND, pluginsd_replay_end, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 27
42 -RSET, pluginsd_replay_set, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 28
43 -RSSTATE, pluginsd_replay_rrdset_collection_state, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 29
47 +CHART_DEFINITION_END, 33, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 24
48 +RBEGIN, 22, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 25
49 +RDSTATE, 23, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 26
50 +REND, 25, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 27
51 +RSET, 21, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 28
52 +RSSTATE, 24, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 29
collectors/plugins.d/gperf-hashtable.h
+55 -56
@@ -1,6 +1,5 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
1 /* ANSI-C code produced by gperf version 3.1 */
3 -/* Command-line: gperf --multiple-iterations=1000 --hash-function-name=gperf_keyword_hash_function --lookup-function-name=gperf_lookup_keyword --word-array-name=gperf_keywords --constants-prefix=GPERF_PARSER_ --struct-type --slot-name=keyword --global-table --null-strings --omit-struct-type --output-file=gperf-hashtable.h gperf-config.txt */
2 +/* Command-line: gperf --multiple-iterations=1000 --output-file=gperf-hashtable.h gperf-config.txt */
3 /* Computed positions: -k'1-2' */
4
5 #if !((' ' == 32) && ('!' == 33) && ('"' == 34) && ('#' == 35) \
@@ -83,66 +82,66 @@ gperf_keyword_hash_function (register const char *str, register size_t len)
82 static PARSER_KEYWORD gperf_keywords[] =
83 {
84 {(char*)0}, {(char*)0}, {(char*)0}, {(char*)0},
86 -#line 9 "gperf-config.txt"
87 - {"HOST", pluginsd_host, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 4},
88 -#line 42 "gperf-config.txt"
89 - {"RSET", pluginsd_replay_set, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 28},
90 -#line 17 "gperf-config.txt"
91 - {"CHART", pluginsd_chart, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 9},
92 - {(char*)0},
93 -#line 43 "gperf-config.txt"
94 - {"RSSTATE", pluginsd_replay_rrdset_collection_state, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 29},
95 -#line 40 "gperf-config.txt"
96 - {"RDSTATE", pluginsd_replay_rrddim_collection_state, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 26},
97 -#line 12 "gperf-config.txt"
98 - {"HOST_LABEL", pluginsd_host_labels, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 7},
99 -#line 10 "gperf-config.txt"
100 - {"HOST_DEFINE", pluginsd_host_define, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 5},
101 -#line 26 "gperf-config.txt"
102 - {"SET", pluginsd_set, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 18},
103 -#line 33 "gperf-config.txt"
104 - {"SET2", pluginsd_set_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 22},
105 -#line 41 "gperf-config.txt"
106 - {"REND", pluginsd_replay_end, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 27},
107 -#line 11 "gperf-config.txt"
108 - {"HOST_DEFINE_END", pluginsd_host_define_end, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 6},
85 #line 18 "gperf-config.txt"
110 - {"CLABEL", pluginsd_clabel, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 10},
111 -#line 39 "gperf-config.txt"
112 - {"RBEGIN", pluginsd_replay_begin, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 25},
113 -#line 6 "gperf-config.txt"
114 - {"FLUSH", pluginsd_flush, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 1},
115 -#line 22 "gperf-config.txt"
116 - {"FUNCTION", pluginsd_function, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 14},
86 + {"HOST", 71, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 4},
87 +#line 51 "gperf-config.txt"
88 + {"RSET", 21, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 28},
89 +#line 26 "gperf-config.txt"
90 + {"CHART", 32, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 9},
91 + {(char*)0},
92 +#line 52 "gperf-config.txt"
93 + {"RSSTATE", 24, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 29},
94 +#line 49 "gperf-config.txt"
95 + {"RDSTATE", 23, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 26},
96 +#line 21 "gperf-config.txt"
97 + {"HOST_LABEL", 74, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 7},
98 +#line 19 "gperf-config.txt"
99 + {"HOST_DEFINE", 72, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 5},
100 +#line 35 "gperf-config.txt"
101 + {"SET", 11, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 18},
102 +#line 42 "gperf-config.txt"
103 + {"SET2", 1, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 22},
104 +#line 50 "gperf-config.txt"
105 + {"REND", 25, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 27},
106 +#line 20 "gperf-config.txt"
107 + {"HOST_DEFINE_END", 73, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 6},
108 +#line 27 "gperf-config.txt"
109 + {"CLABEL", 34, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 10},
110 +#line 48 "gperf-config.txt"
111 + {"RBEGIN", 22, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 25},
112 +#line 15 "gperf-config.txt"
113 + {"FLUSH", 97, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 1},
114 #line 31 "gperf-config.txt"
118 - {"CLAIMED_ID", streaming_claimed_id, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 20},
119 -#line 38 "gperf-config.txt"
120 - {"CHART_DEFINITION_END", pluginsd_chart_definition_end, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 24},
115 + {"FUNCTION", 41, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 14},
116 +#line 40 "gperf-config.txt"
117 + {"CLAIMED_ID", 61, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 20},
118 +#line 47 "gperf-config.txt"
119 + {"CHART_DEFINITION_END", 33, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 24},
120 +#line 34 "gperf-config.txt"
121 + {"OVERWRITE", 52, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 17},
122 +#line 28 "gperf-config.txt"
123 + {"CLABEL_COMMIT", 35, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 11},
124 #line 25 "gperf-config.txt"
122 - {"OVERWRITE", pluginsd_overwrite, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 17},
123 -#line 19 "gperf-config.txt"
124 - {"CLABEL_COMMIT", pluginsd_clabel_commit, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 11},
125 + {"BEGIN", 12, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 8},
126 +#line 41 "gperf-config.txt"
127 + {"BEGIN2", 2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 21},
128 +#line 30 "gperf-config.txt"
129 + {"END", 13, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 13},
130 +#line 43 "gperf-config.txt"
131 + {"END2", 3, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 23},
132 #line 16 "gperf-config.txt"
126 - {"BEGIN", pluginsd_begin, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 8},
133 + {"DISABLE", 98, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 2},
134 +#line 33 "gperf-config.txt"
135 + {"LABEL", 51, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 16},
136 +#line 29 "gperf-config.txt"
137 + {"DIMENSION", 31, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 12},
138 +#line 17 "gperf-config.txt"
139 + {"EXIT", 99, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 3},
140 #line 32 "gperf-config.txt"
128 - {"BEGIN2", pluginsd_begin_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 21},
129 -#line 21 "gperf-config.txt"
130 - {"END", pluginsd_end, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 13},
131 -#line 34 "gperf-config.txt"
132 - {"END2", pluginsd_end_v2, PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 23},
133 -#line 7 "gperf-config.txt"
134 - {"DISABLE", pluginsd_disable, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 2},
135 -#line 24 "gperf-config.txt"
136 - {"LABEL", pluginsd_label, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 16},
137 -#line 20 "gperf-config.txt"
138 - {"DIMENSION", pluginsd_dimension, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 12},
139 -#line 8 "gperf-config.txt"
140 - {"EXIT", pluginsd_exit, PARSER_INIT_PLUGINSD, WORKER_PARSER_FIRST_JOB + 3},
141 -#line 23 "gperf-config.txt"
142 - {"FUNCTION_RESULT_BEGIN", pluginsd_function_result_begin, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 15},
141 + {"FUNCTION_RESULT_BEGIN", 42, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 15},
142 {(char*)0}, {(char*)0}, {(char*)0},
144 -#line 27 "gperf-config.txt"
145 - {"VARIABLE", pluginsd_variable, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 19}
143 +#line 36 "gperf-config.txt"
144 + {"VARIABLE", 53, PARSER_INIT_PLUGINSD|PARSER_INIT_STREAMING, WORKER_PARSER_FIRST_JOB + 19}
145 };
146
147 PARSER_KEYWORD *
collectors/plugins.d/pluginsd_parser.c
+100 -4
@@ -1973,6 +1973,8 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1973
1974 rrd_collector_started();
1975
1976 + size_t count = 0;
1977 +
1978 // this keeps the parser with its current value
1979 // so, parser needs to be allocated before pushing it
1980 netdata_thread_cleanup_push(pluginsd_process_thread_cleanup, parser);
@@ -1984,11 +1986,8 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1986 break;
1987 }
1988
1987 - // free parser with the pop function
1988 - netdata_thread_cleanup_pop(1);
1989 -
1989 cd->unsafe.enabled = parser->user.enabled;
1991 - size_t count = parser->user.data_collections_count;
1990 + count = parser->user.data_collections_count;
1991
1992 if (likely(count)) {
1993 cd->successful_collections += count;
@@ -1997,6 +1996,9 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1996 else
1997 cd->serial_failures++;
1998
1999 + // free parser with the pop function
2000 + netdata_thread_cleanup_pop(1);
2001 +
2002 return count;
2003 }
2004
@@ -2026,6 +2028,100 @@ PARSER *parser_init(struct parser_user_object *user, FILE *fp_input, FILE *fp_ou
2028 return parser;
2029 }
2030
2031 +PARSER_RC parser_execute(PARSER *parser, PARSER_KEYWORD *keyword, char **words, size_t num_words) {
2032 + switch(keyword->id) {
2033 + case 1:
2034 + return pluginsd_set_v2(words, num_words, parser);
2035 +
2036 + case 2:
2037 + return pluginsd_begin_v2(words, num_words, parser);
2038 +
2039 + case 3:
2040 + return pluginsd_end_v2(words, num_words, parser);
2041 +
2042 + case 11:
2043 + return pluginsd_set(words, num_words, parser);
2044 +
2045 + case 12:
2046 + return pluginsd_begin(words, num_words, parser);
2047 +
2048 + case 13:
2049 + return pluginsd_end(words, num_words, parser);
2050 +
2051 + case 21:
2052 + return pluginsd_replay_set(words, num_words, parser);
2053 +
2054 + case 22:
2055 + return pluginsd_replay_begin(words, num_words, parser);
2056 +
2057 + case 23:
2058 + return pluginsd_replay_rrddim_collection_state(words, num_words, parser);
2059 +
2060 + case 24:
2061 + return pluginsd_replay_rrdset_collection_state(words, num_words, parser);
2062 +
2063 + case 25:
2064 + return pluginsd_replay_end(words, num_words, parser);
2065 +
2066 + case 31:
2067 + return pluginsd_dimension(words, num_words, parser);
2068 +
2069 + case 32:
2070 + return pluginsd_chart(words, num_words, parser);
2071 +
2072 + case 33:
2073 + return pluginsd_chart_definition_end(words, num_words, parser);
2074 +
2075 + case 34:
2076 + return pluginsd_clabel(words, num_words, parser);
2077 +
2078 + case 35:
2079 + return pluginsd_clabel_commit(words, num_words, parser);
2080 +
2081 + case 41:
2082 + return pluginsd_function(words, num_words, parser);
2083 +
2084 + case 42:
2085 + return pluginsd_function_result_begin(words, num_words, parser);
2086 +
2087 + case 51:
2088 + return pluginsd_label(words, num_words, parser);
2089 +
2090 + case 52:
2091 + return pluginsd_overwrite(words, num_words, parser);
2092 +
2093 + case 53:
2094 + return pluginsd_variable(words, num_words, parser);
2095 +
2096 + case 61:
2097 + return streaming_claimed_id(words, num_words, parser);
2098 +
2099 + case 71:
2100 + return pluginsd_host(words, num_words, parser);
2101 +
2102 + case 72:
2103 + return pluginsd_host_define(words, num_words, parser);
2104 +
2105 + case 73:
2106 + return pluginsd_host_define_end(words, num_words, parser);
2107 +
2108 + case 74:
2109 + return pluginsd_host_labels(words, num_words, parser);
2110 +
2111 + case 97:
2112 + return pluginsd_flush(words, num_words, parser);
2113 +
2114 + case 98:
2115 + return pluginsd_disable(words, num_words, parser);
2116 +
2117 + case 99:
2118 + return pluginsd_exit(words, num_words, parser);
2119 +
2120 + default:
2121 + fatal("Unknown keyword '%s' with id %zu", keyword->keyword, keyword->id);
2122 + }
2123 +}
2124 +
2125 #include "gperf-hashtable.h"
2126
2127 void parser_init_repertoire(PARSER *parser, PARSER_REPERTOIRE repertoire) {
collectors/plugins.d/pluginsd_parser.h
+11 -9
@@ -32,7 +32,7 @@ typedef PARSER_RC (*keyword_function)(char **words, size_t num_words, struct par
32
33 typedef struct parser_keyword {
34 char *keyword;
35 - keyword_function func;
35 + size_t id;
36 PARSER_REPERTOIRE repertoire;
37 size_t worker_job_id;
38 } PARSER_KEYWORD;
@@ -113,6 +113,14 @@ typedef struct parser {
113
114 } PARSER;
115
116 +PARSER *parser_init(struct parser_user_object *user, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
117 +void parser_init_repertoire(PARSER *parser, PARSER_REPERTOIRE repertoire);
118 +void parser_destroy(PARSER *working_parser);
119 +void pluginsd_cleanup_v2(PARSER *parser);
120 +void inflight_functions_init(PARSER *parser);
121 +void pluginsd_keywords_init(PARSER *parser, PARSER_REPERTOIRE repertoire);
122 +PARSER_RC parser_execute(PARSER *parser, PARSER_KEYWORD *keyword, char **words, size_t num_words);
123 +
124 static inline int find_first_keyword(const char *src, char *dst, int dst_size, bool *isspace_map) {
125 const char *s = src, *keyword_start;
126
@@ -181,7 +189,8 @@ static inline int parser_action(PARSER *parser, char *input) {
189 PARSER_KEYWORD *t = parser_find_keyword(parser, command);
190 if(likely(t)) {
191 worker_is_busy(t->worker_job_id);
184 - rc = (*t->func)(words, num_words, parser);
192 + rc = parser_execute(parser, t, words, num_words);
193 + // rc = (*t->func)(words, num_words, parser);
194 worker_is_idle();
195 }
196 else
@@ -207,11 +216,4 @@ static inline int parser_action(PARSER *parser, char *input) {
216 return (rc == PARSER_RC_ERROR || rc == PARSER_RC_STOP);
217 }
218
210 -PARSER *parser_init(struct parser_user_object *user, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
211 -void parser_init_repertoire(PARSER *parser, PARSER_REPERTOIRE repertoire);
212 -void parser_destroy(PARSER *working_parser);
213 -void pluginsd_cleanup_v2(PARSER *parser);
214 -void inflight_functions_init(PARSER *parser);
215 -void pluginsd_keywords_init(PARSER *parser, PARSER_REPERTOIRE repertoire);
216 -
219 #endif //NETDATA_PLUGINSD_PARSER_H
daemon/common.c
+3
@@ -58,6 +58,9 @@ long get_netdata_cpus(void) {
58
59 processors = cores_user_configured;
60
61 + if(processors < 1)
62 + processors = 1;
63 +
64 return processors;
65 }
66
daemon/main.c
+17 -18
@@ -35,6 +35,7 @@ typedef struct service_thread {
35 SERVICE_THREAD_TYPE type;
36 SERVICE_TYPE services;
37 char name[NETDATA_THREAD_NAME_MAX + 1];
38 + bool stop_immediately;
39 bool cancelled;
40
41 union {
@@ -48,11 +49,9 @@ typedef struct service_thread {
49 } SERVICE_THREAD;
50
51 struct service_globals {
51 - SERVICE_TYPE running;
52 SPINLOCK lock;
53 Pvoid_t pid_judy;
54 } service_globals = {
55 - .running = ~0,
55 .pid_judy = NULL,
56 };
57
@@ -110,20 +109,12 @@ bool service_running(SERVICE_TYPE service) {
109 if(unlikely(!sth))
110 sth = service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, NULL, NULL, false);
111
113 - if(unlikely(netdata_exit))
114 - __atomic_store_n(&service_globals.running, 0, __ATOMIC_RELAXED);
115 -
116 - if(unlikely(service == 0))
117 - service = sth->services;
118 -
112 sth->services |= service;
113
121 - return ((__atomic_load_n(&service_globals.running, __ATOMIC_RELAXED) & service) == service);
114 + return !(sth->stop_immediately || netdata_exit);
115 }
116
117 void service_signal_exit(SERVICE_TYPE service) {
125 - __atomic_and_fetch(&service_globals.running, ~(service), __ATOMIC_RELAXED);
126 -
118 spinlock_lock(&service_globals.lock);
119
120 Pvoid_t *PValue;
@@ -132,11 +123,14 @@ void service_signal_exit(SERVICE_TYPE service) {
123 while((PValue = JudyLFirstThenNext(service_globals.pid_judy, &tid, &first))) {
124 SERVICE_THREAD *sth = *PValue;
125
135 - if((sth->services & service) && sth->request_quit_callback) {
136 - spinlock_unlock(&service_globals.lock);
137 - sth->request_quit_callback(sth->data);
138 - spinlock_lock(&service_globals.lock);
139 - continue;
126 + if((sth->services & service)) {
127 + sth->stop_immediately = true;
128 +
129 + if(sth->request_quit_callback) {
130 + spinlock_unlock(&service_globals.lock);
131 + sth->request_quit_callback(sth->data);
132 + spinlock_lock(&service_globals.lock);
133 + }
134 }
135 }
136
@@ -358,8 +352,7 @@ void netdata_cleanup_and_exit(int ret) {
352 delta_shutdown_time("stop replication, exporters, health and web servers threads");
353
354 timeout = !service_wait_exit(
361 - SERVICE_REPLICATION
362 - | SERVICE_EXPORTERS
355 + SERVICE_EXPORTERS
356 | SERVICE_HEALTH
357 | SERVICE_WEB_SERVER
358 | SERVICE_HTTPD
@@ -372,6 +365,12 @@ void netdata_cleanup_and_exit(int ret) {
365 | SERVICE_STREAMING
366 , 3 * USEC_PER_SEC);
367
368 + delta_shutdown_time("stop replication threads");
369 +
370 + timeout = !service_wait_exit(
371 + SERVICE_REPLICATION // replication has to be stopped after STREAMING, because it cleans up ARAL
372 + , 3 * USEC_PER_SEC);
373 +
374 delta_shutdown_time("disable ML detection and training threads");
375
376 ml_stop_threads();
database/contexts/context.c
+1 -2
@@ -80,7 +80,7 @@ static void rrdcontext_insert_callback(const DICTIONARY_ITEM *item __maybe_unuse
80 }
81
82 rrdinstances_create_in_rrdcontext(rc);
83 - netdata_mutex_init(&rc->mutex);
83 + spinlock_init(&rc->spinlock);
84
85 // signal the react callback to do the job
86 rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_NEW_OBJECT);
@@ -91,7 +91,6 @@ static void rrdcontext_delete_callback(const DICTIONARY_ITEM *item __maybe_unuse
91 RRDCONTEXT *rc = (RRDCONTEXT *)value;
92
93 rrdinstances_destroy_from_rrdcontext(rc);
94 - netdata_mutex_destroy(&rc->mutex);
94 rrdcontext_freez(rc);
95 }
96
database/contexts/internal.h
+4 -4
@@ -250,6 +250,8 @@ typedef struct rrdcontext {
250 uint32_t priority;
251 RRDSET_TYPE chart_type;
252
253 + SPINLOCK spinlock;
254 +
255 RRD_FLAGS flags;
256 time_t first_time_s;
257 time_t last_time_s;
@@ -278,8 +280,6 @@ typedef struct rrdcontext {
280 struct {
281 uint32_t metrics; // the number of metrics in this context
282 } stats;
281 -
282 - netdata_mutex_t mutex;
283 } RRDCONTEXT;
284
285
@@ -356,8 +356,8 @@ static inline void rrdcontext_release(RRDCONTEXT_ACQUIRED *rca) {
356 void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, bool worker_jobs);
357 void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, bool worker_jobs);
358
359 -#define rrdcontext_lock(rc) netdata_mutex_lock(&((rc)->mutex))
360 -#define rrdcontext_unlock(rc) netdata_mutex_unlock(&((rc)->mutex))
359 +#define rrdcontext_lock(rc) spinlock_lock(&((rc)->spinlock))
360 +#define rrdcontext_unlock(rc) spinlock_unlock(&((rc)->spinlock))
361
362 void rrdinstance_trigger_updates(RRDINSTANCE *ri, const char *function);
363 void rrdcontext_trigger_updates(RRDCONTEXT *rc, const char *function);
database/engine/journalfile.c
+5 -1
@@ -180,6 +180,7 @@ static void njfv2idx_remove(struct rrdengine_datafile *datafile) {
180 rw_spinlock_write_lock(&datafile->ctx->njfv2idx.spinlock);
181
182 int rc = JudyLDel(&datafile->ctx->njfv2idx.JudyL, datafile->journalfile->njfv2idx.indexed_as, PJE0);
183 + (void)rc;
184 internal_fatal(!rc, "DBENGINE: NJFV2IDX cannot remove entry");
185
186 datafile->journalfile->njfv2idx.indexed_as = 0;
@@ -214,8 +215,11 @@ static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengin
215
216 madvise_dontfork(journalfile->mmap.data, journalfile->mmap.size);
217 madvise_dontdump(journalfile->mmap.data, journalfile->mmap.size);
218 +
219 + // let the kernel know that we don't want read-ahead on this file
220 + madvise_random(journalfile->mmap.data, journalfile->mmap.size);
221 +
222 // madvise_willneed(journalfile->mmap.data, journalfile->v2.size_of_directory);
218 -// madvise_random(journalfile->mmap.data, journalfile->mmap.size);
223 // madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
224
225 spinlock_lock(&journalfile->v2.spinlock);
database/engine/metric.c
+61 -86
@@ -28,27 +28,16 @@ struct metric {
28 static struct aral_statistics mrg_aral_statistics;
29
30 struct mrg {
31 - ARAL *aral[MRG_PARTITIONS];
31 + size_t partitions;
32
33 - struct pgc_index {
34 - MRG_CACHE_LINE_PADDING(0);
33 + struct mrg_partition {
34 + ARAL *aral; // not protected by our spinlock - it has its own
35
36 RW_SPINLOCK rw_spinlock;
37 -
38 - MRG_CACHE_LINE_PADDING(1);
39 -
40 - Pvoid_t uuid_judy; // each UUID has a JudyL of sections (tiers)
41 -
42 - MRG_CACHE_LINE_PADDING(2);
37 + Pvoid_t uuid_judy; // JudyHS: each UUID has a JudyL of sections (tiers)
38
39 struct mrg_statistics stats;
45 -
46 - MRG_CACHE_LINE_PADDING(3);
47 - } index[MRG_PARTITIONS];
48 -
49 -#ifdef NETDATA_INTERNAL_CHECKS
50 - size_t entries_per_partition[MRG_PARTITIONS];
51 -#endif
40 + } index[];
41 };
42
43 static inline void MRG_STATS_DUPLICATE_ADD(MRG *mrg, size_t partition) {
@@ -59,20 +48,12 @@ static inline void MRG_STATS_ADDED_METRIC(MRG *mrg, size_t partition) {
48 mrg->index[partition].stats.entries++;
49 mrg->index[partition].stats.additions++;
50 mrg->index[partition].stats.size += sizeof(METRIC);
62 -
63 -#ifdef NETDATA_INTERNAL_CHECKS
64 - __atomic_add_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
65 -#endif
51 }
52
53 static inline void MRG_STATS_DELETED_METRIC(MRG *mrg, size_t partition) {
54 mrg->index[partition].stats.entries--;
55 mrg->index[partition].stats.size -= sizeof(METRIC);
56 mrg->index[partition].stats.deletions++;
72 -
73 -#ifdef NETDATA_INTERNAL_CHECKS
74 - __atomic_sub_fetch(&mrg->entries_per_partition[partition], 1, __ATOMIC_RELAXED);
75 -#endif
57 }
58
59 static inline void MRG_STATS_SEARCH_HIT(MRG *mrg, size_t partition) {
@@ -87,18 +68,13 @@ static inline void MRG_STATS_DELETE_MISS(MRG *mrg, size_t partition) {
68 mrg->index[partition].stats.delete_misses++;
69 }
70
90 -static inline void mrg_index_read_lock(MRG *mrg, size_t partition) {
91 - rw_spinlock_read_lock(&mrg->index[partition].rw_spinlock);
92 -}
93 -static inline void mrg_index_read_unlock(MRG *mrg, size_t partition) {
94 - rw_spinlock_read_unlock(&mrg->index[partition].rw_spinlock);
95 -}
96 -static inline void mrg_index_write_lock(MRG *mrg, size_t partition) {
97 - rw_spinlock_write_lock(&mrg->index[partition].rw_spinlock);
98 -}
99 -static inline void mrg_index_write_unlock(MRG *mrg, size_t partition) {
100 - rw_spinlock_write_unlock(&mrg->index[partition].rw_spinlock);
101 -}
71 +#define mrg_index_read_lock(mrg, partition) rw_spinlock_read_lock(&(mrg)->index[partition].rw_spinlock)
72 +#define mrg_index_read_unlock(mrg, partition) rw_spinlock_read_unlock(&(mrg)->index[partition].rw_spinlock)
73 +#define mrg_index_write_lock(mrg, partition) rw_spinlock_write_lock(&(mrg)->index[partition].rw_spinlock)
74 +#define mrg_index_write_unlock(mrg, partition) rw_spinlock_write_unlock(&(mrg)->index[partition].rw_spinlock)
75 +
76 +#define metric_lock(metric) spinlock_lock(&(metric)->spinlock)
77 +#define metric_unlock(metric) spinlock_unlock(&(metric)->spinlock)
78
79 static inline void mrg_stats_size_judyl_change(MRG *mrg, size_t mem_before_judyl, size_t mem_after_judyl, size_t partition) {
80 if(mem_after_judyl > mem_before_judyl)
@@ -117,7 +93,8 @@ static inline void mrg_stats_size_judyhs_removed_uuid(MRG *mrg, size_t partition
93
94 static inline size_t uuid_partition(MRG *mrg __maybe_unused, uuid_t *uuid) {
95 uint8_t *u = (uint8_t *)uuid;
120 - return u[UUID_SZ - 1] % MRG_PARTITIONS;
96 + size_t *n = (size_t *)&u[UUID_SZ - sizeof(size_t)];
97 + return *n % mrg->partitions;
98 }
99
100 static inline bool metric_has_retention_unsafe(MRG *mrg __maybe_unused, METRIC *metric) {
@@ -142,7 +119,7 @@ static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric, b
119 REFCOUNT refcount;
120
121 if(!having_spinlock)
145 - spinlock_lock(&metric->spinlock);
122 + metric_lock(metric);
123
124 if(unlikely(metric->refcount < 0))
125 fatal("METRIC: refcount is %d (negative) during acquire", metric->refcount);
@@ -153,7 +130,7 @@ static inline REFCOUNT metric_acquire(MRG *mrg __maybe_unused, METRIC *metric, b
130 metric_has_retention_unsafe(mrg, metric);
131
132 if(!having_spinlock)
156 - spinlock_unlock(&metric->spinlock);
133 + metric_unlock(metric);
134
135 if(refcount == 1)
136 __atomic_add_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
@@ -168,7 +145,7 @@ static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, ME
145 size_t partition = metric->partition;
146 REFCOUNT refcount;
147
171 - spinlock_lock(&metric->spinlock);
148 + metric_lock(metric);
149
150 if(unlikely(metric->refcount <= 0))
151 fatal("METRIC: refcount is %d (zero or negative) during release", metric->refcount);
@@ -178,7 +155,7 @@ static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, ME
155 if(likely(metric_has_retention_unsafe(mrg, metric) || refcount != 0))
156 ret = false;
157
181 - spinlock_unlock(&metric->spinlock);
158 + metric_unlock(metric);
159
160 if(unlikely(!refcount))
161 __atomic_sub_fetch(&mrg->index[partition].stats.entries_referenced, 1, __ATOMIC_RELAXED);
@@ -191,7 +168,7 @@ static inline bool metric_release_and_can_be_deleted(MRG *mrg __maybe_unused, ME
168 static inline METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *ret) {
169 size_t partition = uuid_partition(mrg, &entry->uuid);
170
194 - METRIC *allocation = aral_mallocz(mrg->aral[partition]);
171 + METRIC *allocation = aral_mallocz(mrg->index[partition].aral);
172
173 mrg_index_write_lock(mrg, partition);
174
@@ -224,7 +201,7 @@ static inline METRIC *metric_add_and_acquire(MRG *mrg, MRG_ENTRY *entry, bool *r
201 if(ret)
202 *ret = false;
203
227 - aral_freez(mrg->aral[partition], allocation);
204 + aral_freez(mrg->index[partition].aral, allocation);
205
206 return metric;
207 }
@@ -326,7 +303,7 @@ static inline bool acquired_metric_del(MRG *mrg, METRIC *metric) {
303
304 mrg_index_write_unlock(mrg, partition);
305
329 - aral_freez(mrg->aral[partition], metric);
306 + aral_freez(mrg->index[partition].aral, metric);
307
308 return true;
309 }
@@ -334,22 +311,20 @@ static inline bool acquired_metric_del(MRG *mrg, METRIC *metric) {
311 // ----------------------------------------------------------------------------
312 // public API
313
337 -inline MRG *mrg_create(void) {
338 - MRG *mrg = callocz(1, sizeof(MRG));
314 +inline MRG *mrg_create(ssize_t partitions) {
315 + if(partitions < 1)
316 + partitions = get_netdata_cpus();
317 +
318 + MRG *mrg = callocz(1, sizeof(MRG) + sizeof(struct mrg_partition) * partitions);
319 + mrg->partitions = partitions;
320
340 - for(size_t i = 0; i < MRG_PARTITIONS ; i++) {
321 + for(size_t i = 0; i < mrg->partitions ; i++) {
322 rw_spinlock_init(&mrg->index[i].rw_spinlock);
323
324 char buf[ARAL_MAX_NAME + 1];
325 snprintfz(buf, ARAL_MAX_NAME, "mrg[%zu]", i);
326
346 - mrg->aral[i] = aral_create(buf,
347 - sizeof(METRIC),
348 - 0,
349 - 16384,
350 - &mrg_aral_statistics,
351 - NULL, NULL, false,
352 - false);
327 + mrg->index[i].aral = aral_create(buf, sizeof(METRIC), 0, 16384, &mrg_aral_statistics, NULL, NULL, false, false);
328 }
329
330 return mrg;
@@ -415,10 +390,10 @@ inline bool mrg_metric_set_first_time_s(MRG *mrg __maybe_unused, METRIC *metric,
390 if(unlikely(first_time_s < 0))
391 return false;
392
418 - spinlock_lock(&metric->spinlock);
393 + metric_lock(metric);
394 metric->first_time_s = first_time_s;
395 metric_has_retention_unsafe(mrg, metric);
421 - spinlock_unlock(&metric->spinlock);
396 + metric_unlock(metric);
397
398 return true;
399 }
@@ -443,7 +418,7 @@ inline void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric,
418 if(unlikely(!first_time_s && !last_time_s && !update_every_s))
419 return;
420
446 - spinlock_lock(&metric->spinlock);
421 + metric_lock(metric);
422
423 if(unlikely(first_time_s && (!metric->first_time_s || first_time_s < metric->first_time_s)))
424 metric->first_time_s = first_time_s;
@@ -458,7 +433,7 @@ inline void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric,
433 metric->latest_update_every_s = (uint32_t) update_every_s;
434
435 metric_has_retention_unsafe(mrg, metric);
461 - spinlock_unlock(&metric->spinlock);
436 + metric_unlock(metric);
437 }
438
439 inline bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
@@ -466,13 +441,13 @@ inline bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRI
441
442 bool ret = false;
443
469 - spinlock_lock(&metric->spinlock);
444 + metric_lock(metric);
445 if(first_time_s > metric->first_time_s) {
446 metric->first_time_s = first_time_s;
447 ret = true;
448 }
449 metric_has_retention_unsafe(mrg, metric);
475 - spinlock_unlock(&metric->spinlock);
450 + metric_unlock(metric);
451
452 return ret;
453 }
@@ -480,7 +455,7 @@ inline bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRI
455 inline time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
456 time_t first_time_s;
457
483 - spinlock_lock(&metric->spinlock);
458 + metric_lock(metric);
459
460 if(unlikely(!metric->first_time_s)) {
461 if(metric->latest_time_s_clean)
@@ -492,13 +467,13 @@ inline time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metri
467
468 first_time_s = metric->first_time_s;
469
495 - spinlock_unlock(&metric->spinlock);
470 + metric_unlock(metric);
471
472 return first_time_s;
473 }
474
475 inline void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, time_t *update_every_s) {
501 - spinlock_lock(&metric->spinlock);
476 + metric_lock(metric);
477
478 if(unlikely(!metric->first_time_s)) {
479 if(metric->latest_time_s_clean)
@@ -512,7 +487,7 @@ inline void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, ti
487 *last_time_s = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
488 *update_every_s = metric->latest_update_every_s;
489
515 - spinlock_unlock(&metric->spinlock);
490 + metric_unlock(metric);
491 }
492
493 inline bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
@@ -521,7 +496,7 @@ inline bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *
496 if(unlikely(latest_time_s < 0))
497 return false;
498
524 - spinlock_lock(&metric->spinlock);
499 + metric_lock(metric);
500
501 // internal_fatal(latest_time_s > max_acceptable_collected_time(),
502 // "DBENGINE METRIC: metric latest time is in the future");
@@ -535,7 +510,7 @@ inline bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *
510 metric->first_time_s = latest_time_s;
511
512 metric_has_retention_unsafe(mrg, metric);
538 - spinlock_unlock(&metric->spinlock);
513 + metric_unlock(metric);
514 return true;
515 }
516
@@ -573,7 +548,7 @@ inline bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metr
548 if (min_first_time_s == LONG_MAX)
549 min_first_time_s = 0;
550
576 - spinlock_lock(&metric->spinlock);
551 + metric_lock(metric);
552 if (--countdown && !min_first_time_s && metric->latest_time_s_hot)
553 do_again = true;
554 else {
@@ -585,7 +560,7 @@ inline bool mrg_metric_zero_disk_retention(MRG *mrg __maybe_unused, METRIC *metr
560
561 ret = metric_has_retention_unsafe(mrg, metric);
562 }
588 - spinlock_unlock(&metric->spinlock);
563 + metric_unlock(metric);
564 } while(do_again);
565
566 return ret;
@@ -600,22 +575,22 @@ inline bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *me
575 if(unlikely(latest_time_s < 0))
576 return false;
577
603 - spinlock_lock(&metric->spinlock);
578 + metric_lock(metric);
579 metric->latest_time_s_hot = latest_time_s;
580
581 if(unlikely(!metric->first_time_s))
582 metric->first_time_s = latest_time_s;
583
584 metric_has_retention_unsafe(mrg, metric);
610 - spinlock_unlock(&metric->spinlock);
585 + metric_unlock(metric);
586 return true;
587 }
588
589 inline time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
590 time_t max;
616 - spinlock_lock(&metric->spinlock);
591 + metric_lock(metric);
592 max = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
618 - spinlock_unlock(&metric->spinlock);
593 + metric_unlock(metric);
594 return max;
595 }
596
@@ -625,9 +600,9 @@ inline bool mrg_metric_set_update_every(MRG *mrg __maybe_unused, METRIC *metric,
600 if(update_every_s <= 0)
601 return false;
602
628 - spinlock_lock(&metric->spinlock);
603 + metric_lock(metric);
604 metric->latest_update_every_s = (uint32_t) update_every_s;
630 - spinlock_unlock(&metric->spinlock);
605 + metric_unlock(metric);
606
607 return true;
608 }
@@ -638,10 +613,10 @@ inline bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRI
613 if(update_every_s <= 0)
614 return false;
615
641 - spinlock_lock(&metric->spinlock);
616 + metric_lock(metric);
617 if(!metric->latest_update_every_s)
618 metric->latest_update_every_s = (uint32_t) update_every_s;
644 - spinlock_unlock(&metric->spinlock);
619 + metric_unlock(metric);
620
621 return true;
622 }
@@ -649,16 +624,16 @@ inline bool mrg_metric_set_update_every_s_if_zero(MRG *mrg __maybe_unused, METRI
624 inline time_t mrg_metric_get_update_every_s(MRG *mrg __maybe_unused, METRIC *metric) {
625 time_t update_every_s;
626
652 - spinlock_lock(&metric->spinlock);
627 + metric_lock(metric);
628 update_every_s = metric->latest_update_every_s;
654 - spinlock_unlock(&metric->spinlock);
629 + metric_unlock(metric);
630
631 return update_every_s;
632 }
633
634 inline bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
635 bool done = false;
661 - spinlock_lock(&metric->spinlock);
636 + metric_lock(metric);
637 if(!metric->writer) {
638 metric->writer = gettid();
639 __atomic_add_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
@@ -666,19 +641,19 @@ inline bool mrg_metric_set_writer(MRG *mrg, METRIC *metric) {
641 }
642 else
643 __atomic_add_fetch(&mrg->index[metric->partition].stats.writers_conflicts, 1, __ATOMIC_RELAXED);
669 - spinlock_unlock(&metric->spinlock);
644 + metric_unlock(metric);
645 return done;
646 }
647
648 inline bool mrg_metric_clear_writer(MRG *mrg, METRIC *metric) {
649 bool done = false;
675 - spinlock_lock(&metric->spinlock);
650 + metric_lock(metric);
651 if(metric->writer) {
652 metric->writer = 0;
653 __atomic_sub_fetch(&mrg->index[metric->partition].stats.writers, 1, __ATOMIC_RELAXED);
654 done = true;
655 }
681 - spinlock_unlock(&metric->spinlock);
656 + metric_unlock(metric);
657 return done;
658 }
659
@@ -734,7 +709,7 @@ inline void mrg_update_metric_retention_and_granularity_by_uuid(
709 inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
710 memset(s, 0, sizeof(struct mrg_statistics));
711
737 - for(int i = 0; i < MRG_PARTITIONS ;i++) {
712 + for(size_t i = 0; i < mrg->partitions ;i++) {
713 s->entries += __atomic_load_n(&mrg->index[i].stats.entries, __ATOMIC_RELAXED);
714 s->entries_referenced += __atomic_load_n(&mrg->index[i].stats.entries_referenced, __ATOMIC_RELAXED);
715 s->entries_with_retention += __atomic_load_n(&mrg->index[i].stats.entries_with_retention, __ATOMIC_RELAXED);
@@ -751,7 +726,7 @@ inline void mrg_get_statistics(MRG *mrg, struct mrg_statistics *s) {
726 s->writers_conflicts += __atomic_load_n(&mrg->index[i].stats.writers_conflicts, __ATOMIC_RELAXED);
727 }
728
754 - s->size += sizeof(MRG);
729 + s->size += sizeof(MRG) + sizeof(struct mrg_partition) * mrg->partitions;
730 }
731
732 // ----------------------------------------------------------------------------
@@ -808,7 +783,7 @@ static void *mrg_stress(void *ptr) {
783 }
784
785 int mrg_unittest(void) {
811 - MRG *mrg = mrg_create();
786 + MRG *mrg = mrg_create(0);
787 METRIC *m1_t0, *m2_t0, *m3_t0, *m4_t0;
788 METRIC *m1_t1, *m2_t1, *m3_t1, *m4_t1;
789 bool ret;
@@ -889,7 +864,7 @@ int mrg_unittest(void) {
864 fatal("DBENGINE METRIC: invalid entries counter");
865
866 size_t entries = 1000000;
892 - size_t threads = MRG_PARTITIONS / 3 + 1;
867 + size_t threads = mrg->partitions / 3 + 1;
868 size_t tiers = 3;
869 size_t run_for_secs = 5;
870 info("preparing stress test of %zu entries...", entries);
database/engine/metric.h
+7 -7
@@ -3,8 +3,6 @@
3
4 #include "../rrd.h"
5
6 -#define MRG_PARTITIONS 10
7 -
6 #define MRG_CACHE_LINE_PADDING(x) uint8_t padding##x[64]
7
8 typedef struct metric METRIC;
@@ -19,9 +17,10 @@ typedef struct mrg_entry {
17 } MRG_ENTRY;
18
19 struct mrg_statistics {
22 - // non-atomic - under a write lock
20 + // --- non-atomic --- under a write lock
21 +
22 size_t entries;
24 - size_t size; // total memory used, with indexing
23 + size_t size; // total memory used, with indexing
24
25 size_t additions;
26 size_t additions_duplicate;
@@ -30,9 +29,10 @@ struct mrg_statistics {
29 size_t delete_having_retention_or_referenced;
30 size_t delete_misses;
31
33 - // atomic - multiple readers / writers
34 -
32 MRG_CACHE_LINE_PADDING(0);
33 +
34 + // --- atomic --- multiple readers / writers
35 +
36 size_t entries_referenced;
37
38 MRG_CACHE_LINE_PADDING(1);
@@ -50,7 +50,7 @@ struct mrg_statistics {
50 size_t writers_conflicts;
51 };
52
53 -MRG *mrg_create(void);
53 +MRG *mrg_create(ssize_t partitions);
54 void mrg_destroy(MRG *mrg);
55
56 METRIC *mrg_metric_dup(MRG *mrg, METRIC *metric);
database/engine/pagecache.c
+1 -1
@@ -1083,7 +1083,7 @@ size_t dynamic_extent_cache_size(void) {
1083
1084 void pgc_and_mrg_initialize(void)
1085 {
1086 - main_mrg = mrg_create();
1086 + main_mrg = mrg_create(0);
1087
1088 size_t target_cache_size = (size_t)default_rrdeng_page_cache_mb * 1024ULL * 1024ULL;
1089 size_t main_cache_size = (target_cache_size / 100) * 95;
database/engine/rrdengineapi.c
+7 -7
@@ -1072,15 +1072,15 @@ static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {
1072 datafiles++;
1073 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1074
1075 - size_t cpus = get_netdata_cpus() / storage_tiers;
1076 - if(cpus > datafiles)
1077 - cpus = datafiles;
1075 + ssize_t cpus = (ssize_t)get_netdata_cpus() / (ssize_t)storage_tiers;
1076 + if(cpus > (ssize_t)datafiles)
1077 + cpus = (ssize_t)datafiles;
1078
1079 - if(cpus > (size_t)libuv_worker_threads)
1080 - cpus = (size_t)libuv_worker_threads;
1079 + if(cpus > (ssize_t)libuv_worker_threads)
1080 + cpus = (ssize_t)libuv_worker_threads;
1081
1082 - if(cpus >= MRG_PARTITIONS / 2)
1083 - cpus = MRG_PARTITIONS / 2 - 1;
1082 + if(cpus >= (ssize_t)get_netdata_cpus() / 2)
1083 + cpus = (ssize_t)(get_netdata_cpus() / 2 - 1);
1084
1085 if(cpus < 1)
1086 cpus = 1;
database/rrdhost.c
+3 -3
@@ -1124,7 +1124,7 @@ static void rrdhost_streaming_sender_structures_init(RRDHOST *host)
1124 host->sender->flags &= ~SENDER_FLAG_COMPRESSION;
1125 #endif
1126
1127 - netdata_mutex_init(&host->sender->mutex);
1127 + spinlock_init(&host->sender->spinlock);
1128 replication_init_sender(host->sender);
1129 }
1130
@@ -1828,7 +1828,7 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
1828 s->stream.hops = s->ingest.hops + 1;
1829 }
1830 else {
1831 - netdata_mutex_lock(&host->sender->mutex);
1831 + sender_lock(host->sender);
1832
1833 s->stream.since = host->sender->last_state_since_t;
1834 s->stream.peers = socket_peers(host->sender->rrdpush_sender_socket);
@@ -1862,7 +1862,7 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
1862 s->stream.reason = host->sender->exit.reason;
1863 }
1864
1865 - netdata_mutex_unlock(&host->sender->mutex);
1865 + sender_unlock(host->sender);
1866 }
1867
1868 s->stream.id = host->rrdpush_sender_connection_counter;
database/sqlite/sqlite_health.c
+2 -1
@@ -1752,6 +1752,7 @@ void sql_health_alarm_log2json_v3(BUFFER *wb, DICTIONARY *alert_instances, time_
1752 char sql[512];
1753 sqlite3_stmt *res = NULL;
1754 int rc;
1755 + BUFFER *command = NULL;
1756
1757 if (unlikely(!alert_instances))
1758 return;
@@ -1823,7 +1824,7 @@ void sql_health_alarm_log2json_v3(BUFFER *wb, DICTIONARY *alert_instances, time_
1824 error_report("Failed to finalize statement for sql_health_alarm_log2json_v3 temp table population");
1825 }
1826
1826 - BUFFER *command = buffer_create(MAX_HEALTH_SQL_SIZE, NULL);
1827 + command = buffer_create(MAX_HEALTH_SQL_SIZE, NULL);
1828
1829 buffer_sprintf(command, SQL_SEARCH_ALERT_LOG, alert_instances);
1830
libnetdata/socket/socket.c
+25 -10
@@ -812,21 +812,36 @@ int connect_to_this_ip46(int protocol, int socktype, const char *host, uint32_t
812 if(errno == EALREADY || errno == EINPROGRESS) {
813 info("Waiting for connection to ip %s port %s to be established", hostBfr, servBfr);
814
815 - fd_set fds;
816 - FD_ZERO(&fds);
817 - FD_SET(0, &fds);
818 - int rc = select (1, NULL, &fds, NULL, timeout);
819 -
820 - if(rc > 0 && FD_ISSET(fd, &fds)) {
821 - info("connect() to ip %s port %s completed successfully", hostBfr, servBfr);
815 + // Convert 'struct timeval' to milliseconds for poll():
816 + int timeout_milliseconds = timeout->tv_sec * 1000 + timeout->tv_usec / 1000;
817 +
818 + struct pollfd fds[1];
819 + fds[0].fd = fd;
820 + fds[0].events = POLLOUT; // We are looking for the ability to write to the socket
821 +
822 + int ret = poll(fds, 1, timeout_milliseconds);
823 + if (ret > 0) {
824 + // poll() completed normally. We can check the revents to see what happened
825 + if (fds[0].revents & POLLOUT) {
826 + // connect() completed successfully, socket is writable.
827 + info("connect() to ip %s port %s completed successfully", hostBfr, servBfr);
828 + }
829 + else {
830 + // This means that the socket is in error. We will close it and set fd to -1
831 + error("Failed to connect to '%s', port '%s'.", hostBfr, servBfr);
832 + close(fd);
833 + fd = -1;
834 + }
835 }
823 - else if(rc == -1) {
824 - error("Failed to connect to '%s', port '%s'. select() returned %d", hostBfr, servBfr, rc);
836 + else if (ret == 0) {
837 + // poll() timed out, the connection is not established within the specified timeout.
838 + error("Timed out while connecting to '%s', port '%s'.", hostBfr, servBfr);
839 close(fd);
840 fd = -1;
841 }
842 else {
829 - error("Timed out while connecting to '%s', port '%s'. select() returned %d", hostBfr, servBfr, rc);
843 + // poll() returned an error.
844 + error("Failed to connect to '%s', port '%s'. poll() returned %d", hostBfr, servBfr, ret);
845 close(fd);
846 fd = -1;
847 }
streaming/receiver.c
+11 -12
@@ -282,31 +282,30 @@ static void receiver_set_exit_reason(struct receiver_state *rpt, STREAM_HANDSHAK
282 rpt->exit.reason = reason;
283 }
284
285 -static inline bool receiver_should_continue(struct receiver_state *rpt) {
285 +static inline bool receiver_should_stop(struct receiver_state *rpt) {
286 static __thread size_t counter = 0;
287
288 if(unlikely(rpt->exit.shutdown)) {
289 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SHUTDOWN, false);
290 - return false;
290 + return true;
291 }
292
293 - // check every 1000 lines read
294 - if((counter++ % 1000) != 0) return true;
295 -
293 if(unlikely(!service_running(SERVICE_STREAMING))) {
294 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_NETDATA_EXIT, false);
298 - return false;
295 + return true;
296 }
297
301 - netdata_thread_testcancel();
302 -
303 - rpt->last_msg_t = now_monotonic_sec();
298 + if(unlikely((counter++ % 1000) == 0)) {
299 + // check every 1000 lines read
300 + netdata_thread_testcancel();
301 + rpt->last_msg_t = now_monotonic_sec();
302 + }
303
305 - return true;
304 + return false;
305 }
306
307 static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
309 - size_t result;
308 + size_t result = 0;
309
310 PARSER *parser = NULL;
311 {
@@ -346,7 +345,7 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
345
346 size_t read_buffer_start = 0;
347 char buffer[PLUGINSD_LINE_MAX + 2] = "";
349 - while(receiver_should_continue(rpt)) {
348 + while(!receiver_should_stop(rpt)) {
349
350 if(!receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &read_buffer_start)) {
351 bool have_new_data = compressed_connection ? receiver_read_compressed(rpt) : receiver_read_uncompressed(rpt);
streaming/replication.c
+2 -2
@@ -1096,7 +1096,7 @@ void replication_set_next_point_in_time(time_t after, size_t unique_id) {
1096 // ----------------------------------------------------------------------------
1097 // replication sort entry management
1098
1099 -static struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
1099 +static inline struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
1100 struct replication_sort_entry *rse = aral_mallocz(replication_globals.aral_rse);
1101 __atomic_add_fetch(&replication_globals.atomic.memory, sizeof(struct replication_sort_entry), __ATOMIC_RELAXED);
1102
@@ -1120,7 +1120,7 @@ static void replication_sort_entry_destroy(struct replication_sort_entry *rse) {
1120 }
1121
1122 static void replication_sort_entry_add(struct replication_request *rq) {
1123 - if(rrdpush_sender_replication_buffer_full_get(rq->sender)) {
1123 + if(unlikely(rrdpush_sender_replication_buffer_full_get(rq->sender))) {
1124 rq->indexed_in_judy = false;
1125 rq->not_indexed_buffer_full = true;
1126 rq->not_indexed_preprocessing = false;
streaming/rrdpush.c
+8 -9
@@ -687,7 +687,7 @@ void rrdpush_sender_thread_stop(RRDHOST *host, STREAM_HANDSHAKE reason, bool wai
687 if (!host->sender)
688 return;
689
690 - netdata_mutex_lock(&host->sender->mutex);
690 + sender_lock(host->sender);
691
692 if(rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)) {
693
@@ -698,20 +698,19 @@ void rrdpush_sender_thread_stop(RRDHOST *host, STREAM_HANDSHAKE reason, bool wai
698 netdata_thread_cancel(host->rrdpush_sender_thread);
699 }
700
701 - netdata_mutex_unlock(&host->sender->mutex);
701 + sender_unlock(host->sender);
702
703 if(wait) {
704 - netdata_mutex_lock(&host->sender->mutex);
704 + sender_lock(host->sender);
705 while(host->sender->tid) {
706 - netdata_mutex_unlock(&host->sender->mutex);
706 + sender_unlock(host->sender);
707 sleep_usec(10 * USEC_PER_MS);
708 - netdata_mutex_lock(&host->sender->mutex);
708 + sender_lock(host->sender);
709 }
710 - netdata_mutex_unlock(&host->sender->mutex);
710 + sender_unlock(host->sender);
711 }
712 }
713
714 -
714 // ----------------------------------------------------------------------------
715 // rrdpush receiver thread
716
@@ -721,7 +720,7 @@ void log_stream_connection(const char *client_ip, const char *client_port, const
720
721
722 static void rrdpush_sender_thread_spawn(RRDHOST *host) {
724 - netdata_mutex_lock(&host->sender->mutex);
723 + sender_lock(host->sender);
724
725 if(!rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)) {
726 char tag[NETDATA_THREAD_TAG_MAX + 1];
@@ -733,7 +732,7 @@ static void rrdpush_sender_thread_spawn(RRDHOST *host) {
732 rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN);
733 }
734
736 - netdata_mutex_unlock(&host->sender->mutex);
735 + sender_unlock(host->sender);
736 }
737
738 int rrdpush_receiver_permission_denied(struct web_client *w) {
streaming/rrdpush.h
+4 -1
@@ -249,7 +249,7 @@ struct sender_state {
249 size_t not_connected_loops;
250 // Metrics are collected asynchronously by collector threads calling rrdset_done_push(). This can also trigger
251 // the lazy creation of the sender thread - both cases (buffer access and thread creation) are guarded here.
252 - netdata_mutex_t mutex;
252 + SPINLOCK spinlock;
253 struct circular_buffer *buffer;
254 char read_buffer[PLUGINSD_LINE_MAX + 1];
255 ssize_t read_len;
@@ -296,6 +296,9 @@ struct sender_state {
296 } atomic;
297 };
298
299 +#define sender_lock(sender) spinlock_lock(&(sender)->spinlock)
300 +#define sender_unlock(sender) spinlock_unlock(&(sender)->spinlock)
301 +
302 #define rrdpush_sender_pipe_has_pending_data(sender) __atomic_load_n(&(sender)->atomic.pending_data, __ATOMIC_RELAXED)
303 #define rrdpush_sender_pipe_set_pending_data(sender) __atomic_store_n(&(sender)->atomic.pending_data, true, __ATOMIC_RELAXED)
304 #define rrdpush_sender_pipe_clear_pending_data(sender) __atomic_store_n(&(sender)->atomic.pending_data, false, __ATOMIC_RELAXED)
streaming/sender.c
+15 -15
@@ -100,7 +100,7 @@ void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type)
100 if(unlikely(!src || !src_len))
101 return;
102
103 - netdata_mutex_lock(&s->mutex);
103 + sender_lock(s);
104
105 // FILE *fp = fopen("/tmp/stream.txt", "a");
106 // fprintf(fp,
@@ -156,7 +156,7 @@ void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type)
156 rrdhost_hostname(s->host), s->connected_to);
157
158 deactivate_compression(s);
159 - netdata_mutex_unlock(&s->mutex);
159 + sender_unlock(s);
160 return;
161 }
162 }
@@ -189,7 +189,7 @@ void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type)
189 signal_sender = true;
190 }
191
192 - netdata_mutex_unlock(&s->mutex);
192 + sender_unlock(s);
193
194 if(signal_sender)
195 rrdpush_signal_sender_to_wake_up(s);
@@ -273,7 +273,7 @@ static void rrdpush_sender_cbuffer_recreate_timed(struct sender_state *s, time_t
273 return;
274
275 if(!have_mutex)
276 - netdata_mutex_lock(&s->mutex);
276 + sender_lock(s);
277
278 rrdpush_sender_last_buffer_recreate_set(s, now_s);
279 last_reset_time_s = now_s;
@@ -287,20 +287,20 @@ static void rrdpush_sender_cbuffer_recreate_timed(struct sender_state *s, time_t
287 sender_thread_buffer_free();
288
289 if(!have_mutex)
290 - netdata_mutex_unlock(&s->mutex);
290 + sender_unlock(s);
291 }
292
293 static void rrdpush_sender_cbuffer_flush(RRDHOST *host) {
294 rrdpush_sender_set_flush_time(host->sender);
295
296 - netdata_mutex_lock(&host->sender->mutex);
296 + sender_lock(host->sender);
297
298 // flush the output buffer from any data it may have
299 cbuffer_flush(host->sender->buffer);
300 rrdpush_sender_cbuffer_recreate_timed(host->sender, now_monotonic_sec(), true, true);
301 replication_recalculate_buffer_used_ratio_unsafe(host->sender);
302
303 - netdata_mutex_unlock(&host->sender->mutex);
303 + sender_unlock(host->sender);
304 }
305
306 static void rrdpush_sender_charts_and_replication_reset(RRDHOST *host) {
@@ -821,7 +821,7 @@ static ssize_t attempt_to_send(struct sender_state *s) {
821 struct circular_buffer *cb = s->buffer;
822 #endif
823
824 - netdata_mutex_lock(&s->mutex);
824 + sender_lock(s);
825 char *chunk;
826 size_t outstanding = cbuffer_next_unsafe(s->buffer, &chunk);
827 debug(D_STREAM, "STREAM: Sending data. Buffer r=%zu w=%zu s=%zu, next chunk=%zu", cb->read, cb->write, cb->size, outstanding);
@@ -853,7 +853,7 @@ static ssize_t attempt_to_send(struct sender_state *s) {
853 debug(D_STREAM, "STREAM: send() returned 0 -> no error but no transmission");
854
855 replication_recalculate_buffer_used_ratio_unsafe(s);
856 - netdata_mutex_unlock(&s->mutex);
856 + sender_unlock(s);
857
858 return ret;
859 }
@@ -1093,7 +1093,7 @@ static bool rrdhost_set_sender(RRDHOST *host) {
1093 if(unlikely(!host->sender)) return false;
1094
1095 bool ret = false;
1096 - netdata_mutex_lock(&host->sender->mutex);
1096 + sender_lock(host->sender);
1097 if(!host->sender->tid) {
1098 rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED | RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
1099 rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN);
@@ -1103,7 +1103,7 @@ static bool rrdhost_set_sender(RRDHOST *host) {
1103 host->sender->exit.reason = STREAM_HANDSHAKE_NEVER;
1104 ret = true;
1105 }
1106 - netdata_mutex_unlock(&host->sender->mutex);
1106 + sender_unlock(host->sender);
1107
1108 rrdpush_reset_destinations_postpone_time(host);
1109
@@ -1164,7 +1164,7 @@ static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
1164
1165 RRDHOST *host = s->host;
1166
1167 - netdata_mutex_lock(&host->sender->mutex);
1167 + sender_lock(host->sender);
1168 info("STREAM %s [send]: sending thread exits %s",
1169 rrdhost_hostname(host),
1170 host->sender->exit.reason != STREAM_HANDSHAKE_NEVER ? stream_handshake_error_to_string(host->sender->exit.reason) : "");
@@ -1173,7 +1173,7 @@ static void rrdpush_sender_thread_cleanup_callback(void *ptr) {
1173 rrdpush_sender_pipe_close(host, host->sender->rrdpush_sender_pipe, false);
1174
1175 rrdhost_clear_sender___while_having_sender_mutex(host);
1176 - netdata_mutex_unlock(&host->sender->mutex);
1176 + sender_unlock(host->sender);
1177
1178 freez(s->pipe_buffer);
1179 freez(s);
@@ -1343,14 +1343,14 @@ void *rrdpush_sender_thread(void *ptr) {
1343 continue;
1344 }
1345
1346 - netdata_mutex_lock(&s->mutex);
1346 + sender_lock(s);
1347 size_t outstanding = cbuffer_next_unsafe(s->buffer, NULL);
1348 size_t available = cbuffer_available_size_unsafe(s->buffer);
1349 if (unlikely(!outstanding)) {
1350 rrdpush_sender_pipe_clear_pending_data(s);
1351 rrdpush_sender_cbuffer_recreate_timed(s, now_s, true, false);
1352 }
1353 - netdata_mutex_unlock(&s->mutex);
1353 + sender_unlock(s);
1354
1355 worker_set_metric(WORKER_SENDER_JOB_BUFFER_RATIO, (NETDATA_DOUBLE)(s->buffer->max_size - available) * 100.0 / (NETDATA_DOUBLE)s->buffer->max_size);
1356
web/rtc/webrtc.c
+5 -5
@@ -367,7 +367,7 @@ cleanup:
367 // ----------------------------------------------------------------------------
368 // webrtc data channel
369
370 -static void myOpenCallback(int id, void *user_ptr) {
370 +static void myOpenCallback(int id __maybe_unused, void *user_ptr) {
371 webrtc_set_thread_name();
372
373 WEBRTC_DC *chan = user_ptr;
@@ -378,7 +378,7 @@ static void myOpenCallback(int id, void *user_ptr) {
378 chan->open = true;
379 }
380
381 -static void myClosedCallback(int id, void *user_ptr) {
381 +static void myClosedCallback(int id __maybe_unused, void *user_ptr) {
382 webrtc_set_thread_name();
383
384 WEBRTC_DC *chan = user_ptr;
@@ -397,7 +397,7 @@ static void myClosedCallback(int id, void *user_ptr) {
397 freez(chan);
398 }
399
400 -static void myErrorCallback(int id, const char *error, void *user_ptr) {
400 +static void myErrorCallback(int id __maybe_unused, const char *error, void *user_ptr) {
401 webrtc_set_thread_name();
402
403 WEBRTC_DC *chan = user_ptr;
@@ -406,7 +406,7 @@ static void myErrorCallback(int id, const char *error, void *user_ptr) {
406 error("WEBRTC[%d],DC[%d]: ERROR: '%s'", chan->conn->pc, chan->dc, error);
407 }
408
409 -static void myMessageCallback(int id, const char *message, int size, void *user_ptr) {
409 +static void myMessageCallback(int id __maybe_unused, const char *message, int size, void *user_ptr) {
410 webrtc_set_thread_name();
411
412 WEBRTC_DC *chan = user_ptr;
@@ -441,7 +441,7 @@ static void myMessageCallback(int id, const char *message, int size, void *user_
441 // }
442 //}
443
444 -static void myDataChannelCallback(int pc, int dc, void *user_ptr) {
444 +static void myDataChannelCallback(int pc __maybe_unused, int dc, void *user_ptr) {
445 webrtc_set_thread_name();
446
447 WEBRTC_CONN *conn = user_ptr;