optimized rrdhost_status (#19472)
* optimized rrdhost_status * write the status * optimize the conditions on every iteration
Costa Tsaousis committed
Jan 23, 2025 at 22:06 UTC
6c7625ac606f6c85dc6cccae950bd6cbf2ea3ee4
13 files changed
+216
-191
src/daemon/pulse/pulse-parents.c
+1
-1
@@ -59,7 +59,7 @@ struct {
59
60
static PULSE_HOST_STATUS pulse_host_detect_receiver_status(RRDHOST *host) {
61
RRDHOST_STATUS status = { 0 };
62
- rrdhost_status(host, now_realtime_sec(), &status);
62
+ rrdhost_status(host, now_realtime_sec(), &status, RRDHOST_STATUS_BASIC);
63
64
PULSE_HOST_STATUS rc = 0;
65
src/database/contexts/api_v2_contexts.c
+1
-1
@@ -366,7 +366,7 @@ static void rrdcontext_to_json_v2_rrdhost(BUFFER *wb, RRDHOST *host, struct rrdc
366
367
if(ctl->mode & (CONTEXTS_V2_NODES_INFO | CONTEXTS_V2_NODES_STREAM_PATH | CONTEXTS_V2_NODE_INSTANCES)) {
368
RRDHOST_STATUS s;
369
- rrdhost_status(host, ctl->now, &s);
369
+ rrdhost_status(host, ctl->now, &s, RRDHOST_STATUS_ALL);
370
371
if (ctl->mode & (CONTEXTS_V2_NODES_INFO | CONTEXTS_V2_NODES_STREAM_PATH)) {
372
buffer_json_member_add_string(wb, "v", rrdhost_program_version(host));
src/database/rrdhost-status.c
+140
-124
@@ -106,103 +106,124 @@ int16_t rrdhost_ingestion_hops(RRDHOST *host) {
106
return rrdhost_system_info_hops(host->system_info);
107
}
108
109
-void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
110
- memset(s, 0, sizeof(*s));
109
+static inline RRDHOST_DB_STATUS rrdhost_status_db(RRDHOST *host, time_t now, RRDHOST_STATUS *s, RRDHOST_FLAGS flags, bool online) {
110
+ RRDHOST_DB_STATUS status;
111
112
- s->host = host;
113
- s->now = now;
114
-
115
- RRDHOST_FLAGS flags = __atomic_load_n(&host->flags, __ATOMIC_RELAXED);
112
+ uint32_t metrics = UINT32_MAX;
113
+ uint32_t instances = UINT32_MAX;
114
+ uint32_t contexts = UINT32_MAX;
115
117
- // --- dyncfg ---
116
+ time_t first_time_s = 0, last_time_s = 0;
117
+ rrdhost_retention(host, now, online, &first_time_s, &last_time_s);
118
119
- s->dyncfg.status = dyncfg_available_for_rrdhost(host) ? RRDHOST_DYNCFG_STATUS_AVAILABLE : RRDHOST_DYNCFG_STATUS_UNAVAILABLE;
119
+ if (!first_time_s ||
120
+ !last_time_s ||
121
+ (flags & RRDHOST_FLAG_PENDING_CONTEXT_LOAD) ||
122
+ !(metrics = __atomic_load_n(&host->rrdctx.metrics_count, __ATOMIC_RELAXED)) ||
123
+ !(instances = __atomic_load_n(&host->rrdctx.instances_count, __ATOMIC_RELAXED)) ||
124
+ !(contexts = __atomic_load_n(&host->rrdctx.contexts_count, __ATOMIC_RELAXED)))
125
+ status = RRDHOST_DB_STATUS_INITIALIZING;
126
+ else
127
+ status = RRDHOST_DB_STATUS_QUERYABLE;
128
121
- // --- db ---
129
123
- bool online = rrdhost_is_online(host);
130
+ if(s) {
131
+ s->db.status = status;
132
125
- rrdhost_retention(host, now, online, &s->db.first_time_s, &s->db.last_time_s);
126
- s->db.metrics = __atomic_load_n(&host->rrdctx.metrics_count, __ATOMIC_RELAXED);
127
- s->db.instances = __atomic_load_n(&host->rrdctx.instances_count, __ATOMIC_RELAXED);
128
- s->db.contexts = __atomic_load_n(&host->rrdctx.contexts_count, __ATOMIC_RELAXED);
129
- if(!s->db.first_time_s || !s->db.last_time_s || !s->db.metrics || !s->db.instances || !s->db.contexts ||
130
- (flags & (RRDHOST_FLAG_PENDING_CONTEXT_LOAD)))
131
- s->db.status = RRDHOST_DB_STATUS_INITIALIZING;
132
- else
133
- s->db.status = RRDHOST_DB_STATUS_QUERYABLE;
133
+ s->db.first_time_s = first_time_s;
134
+ s->db.last_time_s = last_time_s;
135
+ s->db.status = status;
136
+ s->db.mode = host->rrd_memory_mode;
137
135
- s->db.mode = host->rrd_memory_mode;
138
+ s->db.metrics = (metrics == UINT32_MAX) ? __atomic_load_n(&host->rrdctx.metrics_count, __ATOMIC_RELAXED) : metrics;
139
+ s->db.instances = (instances == UINT32_MAX) ? __atomic_load_n(&host->rrdctx.instances_count, __ATOMIC_RELAXED) : instances;
140
+ s->db.contexts = (contexts == UINT32_MAX) ? __atomic_load_n(&host->rrdctx.contexts_count, __ATOMIC_RELAXED) : contexts;
141
+ }
142
137
- // --- ingest ---
143
+ return status;
144
+}
145
139
- s->ingest.since = MAX(host->stream.rcv.status.last_connected, host->stream.rcv.status.last_disconnected);
140
- s->ingest.reason = (online) ? STREAM_HANDSHAKE_NEVER : host->stream.rcv.status.exit_reason;
146
+static inline RRDHOST_INGEST_STATUS rrdhost_status_ingest(RRDHOST *host, RRDHOST_STATUS *s, RRDHOST_FLAGS flags, RRDHOST_DB_STATUS db_status, bool online) {
147
+ RRDHOST_INGEST_STATUS status;
148
142
- rrdhost_receiver_lock(host);
143
- s->ingest.hops = rrdhost_ingestion_hops(host);
144
- bool has_receiver = false;
145
- if (host->receiver && rrdhost_flag_check(host, RRDHOST_FLAG_COLLECTOR_ONLINE)) {
146
- has_receiver = true;
147
- s->ingest.replication.instances = rrdhost_receiver_replicating_charts(host);
148
- s->ingest.replication.completion = host->stream.rcv.status.replication.percent;
149
- s->ingest.replication.in_progress = s->ingest.replication.instances > 0;
150
-
151
- s->ingest.capabilities = host->receiver->capabilities;
152
- s->ingest.peers = nd_sock_socket_peers(&host->receiver->sock);
153
- s->ingest.ssl = nd_sock_is_ssl(&host->receiver->sock);
154
- }
155
- rrdhost_receiver_unlock(host);
149
+ uint32_t collected_metrics = UINT32_MAX;
150
+ uint32_t replicating_instances = UINT32_MAX;
151
157
- s->ingest.collected.metrics = __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED);
158
- s->ingest.collected.instances = __atomic_load_n(&host->collected.instances_count, __ATOMIC_RELAXED);
159
- s->ingest.collected.contexts = __atomic_load_n(&host->collected.contexts_count, __ATOMIC_RELAXED);
152
+ time_t since = MAX(host->stream.rcv.status.last_connected, host->stream.rcv.status.last_disconnected);
153
+ STREAM_HANDSHAKE reason = (online) ? STREAM_HANDSHAKE_NEVER : host->stream.rcv.status.exit_reason;
154
155
if (online) {
162
- if(s->db.status == RRDHOST_DB_STATUS_INITIALIZING)
163
- s->ingest.status = RRDHOST_INGEST_STATUS_INITIALIZING;
156
+ if (db_status == RRDHOST_DB_STATUS_INITIALIZING)
157
+ status = RRDHOST_INGEST_STATUS_INITIALIZING;
158
159
else if (rrdhost_is_local(host)) {
166
- s->ingest.status = RRDHOST_INGEST_STATUS_ONLINE;
167
- s->ingest.since = netdata_start_time;
160
+ status = RRDHOST_INGEST_STATUS_ONLINE;
161
+ since = netdata_start_time;
162
}
169
-
170
- else if (s->ingest.replication.in_progress || !s->ingest.collected.metrics)
171
- s->ingest.status = RRDHOST_INGEST_STATUS_REPLICATING;
163
+ else if (
164
+ (replicating_instances = rrdhost_receiver_replicating_charts(host)) > 0 ||
165
+ !(collected_metrics = __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED)))
166
+ status = RRDHOST_INGEST_STATUS_REPLICATING;
167
168
else
174
- s->ingest.status = RRDHOST_INGEST_STATUS_ONLINE;
169
+ status = RRDHOST_INGEST_STATUS_ONLINE;
170
}
171
else {
177
- if (!s->ingest.since) {
178
- s->ingest.status = RRDHOST_INGEST_STATUS_ARCHIVED;
179
- s->ingest.since = s->db.last_time_s;
180
- }
181
-
172
+ if(!since)
173
+ status = RRDHOST_INGEST_STATUS_ARCHIVED;
174
else
183
- s->ingest.status = RRDHOST_INGEST_STATUS_OFFLINE;
175
+ status = RRDHOST_INGEST_STATUS_OFFLINE;
176
}
177
186
- if(host == localhost)
187
- s->ingest.type = RRDHOST_INGEST_TYPE_LOCALHOST;
188
- else if(has_receiver)
189
- s->ingest.type = RRDHOST_INGEST_TYPE_CHILD;
190
- else if(rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST))
191
- s->ingest.type = RRDHOST_INGEST_TYPE_VIRTUAL;
192
- else
193
- s->ingest.type = RRDHOST_INGEST_TYPE_ARCHIVED;
178
+ bool has_receiver = false;
179
195
- s->ingest.id = host->stream.rcv.status.connections;
180
+ if(s) {
181
+ if(status == RRDHOST_INGEST_STATUS_ARCHIVED)
182
+ since = s->db.last_time_s;
183
197
- if(!s->ingest.since)
198
- s->ingest.since = netdata_start_time;
184
+ s->ingest.status = status;
185
200
- if(s->ingest.status == RRDHOST_INGEST_STATUS_ONLINE)
201
- s->db.liveness = RRDHOST_DB_LIVENESS_LIVE;
202
- else
203
- s->db.liveness = RRDHOST_DB_LIVENESS_STALE;
186
+ s->ingest.since = since ? since : netdata_start_time;
187
+ s->ingest.reason = reason;
188
+ s->ingest.hops = rrdhost_ingestion_hops(host);
189
205
- // --- stream ---
190
+ s->ingest.collected.metrics = collected_metrics == UINT32_MAX ? __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED) : collected_metrics;
191
+ s->ingest.collected.instances = __atomic_load_n(&host->collected.instances_count, __ATOMIC_RELAXED);
192
+ s->ingest.collected.contexts = __atomic_load_n(&host->collected.contexts_count, __ATOMIC_RELAXED);
193
+
194
+ if(!rrdhost_is_local(host)) {
195
+ rrdhost_receiver_lock(host);
196
+ if (host->receiver && (flags & RRDHOST_FLAG_COLLECTOR_ONLINE)) {
197
+ has_receiver = true;
198
+ s->ingest.replication.instances = replicating_instances == UINT32_MAX ? rrdhost_receiver_replicating_charts(host) : replicating_instances;
199
+ s->ingest.replication.completion = host->stream.rcv.status.replication.percent;
200
+ s->ingest.replication.in_progress = s->ingest.replication.instances > 0;
201
+
202
+ s->ingest.capabilities = host->receiver->capabilities;
203
+ s->ingest.peers = nd_sock_socket_peers(&host->receiver->sock);
204
+ s->ingest.ssl = nd_sock_is_ssl(&host->receiver->sock);
205
+ }
206
+ rrdhost_receiver_unlock(host);
207
+ }
208
+
209
+ if(host == localhost)
210
+ s->ingest.type = RRDHOST_INGEST_TYPE_LOCALHOST;
211
+ else if(has_receiver)
212
+ s->ingest.type = RRDHOST_INGEST_TYPE_CHILD;
213
+ else if(rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST))
214
+ s->ingest.type = RRDHOST_INGEST_TYPE_VIRTUAL;
215
+ else
216
+ s->ingest.type = RRDHOST_INGEST_TYPE_ARCHIVED;
217
+
218
+ s->ingest.id = host->stream.rcv.status.connections;
219
+ }
220
+
221
+ return status;
222
+}
223
+
224
+static void rrdhost_status_stream_internal(RRDHOST_STATUS *s) {
225
+ RRDHOST *host = s->host;
226
+ time_t now = s->now;
227
228
if (!host->sender) {
229
s->stream.status = RRDHOST_STREAM_STATUS_DISABLED;
@@ -252,8 +273,10 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
273
274
if(!s->stream.since)
275
s->stream.since = netdata_start_time;
276
+}
277
256
- // --- ml ---
278
+static void rrdhost_status_ml_internal(RRDHOST_STATUS *s) {
279
+ RRDHOST *host = s->host;
280
281
if(ml_host_get_host_status(host, &s->ml.metrics)) {
282
if(stream_has_capability(&s->ingest, STREAM_CAP_ML_MODELS))
@@ -271,8 +294,10 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
294
s->ml.type = RRDHOST_ML_TYPE_DISABLED;
295
s->ml.status = RRDHOST_ML_STATUS_DISABLED;
296
}
297
+}
298
275
- // --- health ---
299
+static void rrdhost_status_health_internal(RRDHOST_STATUS *s, RRDHOST_FLAGS flags) {
300
+ RRDHOST *host = s->host;
301
302
if(host->health.enabled) {
303
if(flags & RRDHOST_FLAG_PENDING_HEALTH_INITIALIZATION)
@@ -317,67 +342,58 @@ void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s) {
342
s->health.status = RRDHOST_HEALTH_STATUS_DISABLED;
343
}
344
345
+void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s, RRDHOST_STATUS_INFO info) {
346
+ memset(s, 0, sizeof(*s));
347
+
348
+ s->host = host;
349
+ s->now = now;
350
321
-// Minimal function to get the ingest status only
322
-RRDHOST_INGEST_STATUS rrdhost_get_ingest_status(RRDHOST *host, time_t now) {
351
RRDHOST_FLAGS flags = __atomic_load_n(&host->flags, __ATOMIC_RELAXED);
324
- bool online = rrdhost_is_online(host);
352
+ bool online = rrdhost_is_local(host) || rrdhost_is_online_flags(flags);
353
326
- // Initialize ingest status variables
327
- RRDHOST_INGEST_STATUS ingest_status;
328
- time_t ingest_since = MAX(host->stream.rcv.status.last_connected, host->stream.rcv.status.last_disconnected);
329
- time_t db_last_time_s;
354
+ // --- db ---
355
331
- // Database state
332
- time_t first_time_s, last_time_s;
333
- rrdhost_retention(host, now, online, &first_time_s, &last_time_s);
334
- db_last_time_s = last_time_s;
356
+ rrdhost_status_db(host, now, s, flags, online);
357
336
- uint32_t metrics = __atomic_load_n(&host->rrdctx.metrics_count, __ATOMIC_RELAXED);
337
- uint32_t instances = __atomic_load_n(&host->rrdctx.instances_count, __ATOMIC_RELAXED);
338
- uint32_t contexts = __atomic_load_n(&host->rrdctx.contexts_count, __ATOMIC_RELAXED);
358
+ // --- ingest ---
359
340
- bool db_initializing = !first_time_s || !last_time_s || !metrics || !instances || !contexts ||
341
- (flags & RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
360
+ rrdhost_status_ingest(host, s, flags, s->db.status, online);
361
343
- uint32_t collected_metrics = __atomic_load_n(&host->collected.metrics_count, __ATOMIC_RELAXED);
362
+ // --- db (part 2) ---
363
345
- // Replication state, if set in progress due to zero collected metrics, no need to
346
- // get the receiver lock
347
- bool replication_in_progress = (!collected_metrics);
348
- uint32_t replication_instances = 0;
364
+ if(s->ingest.status == RRDHOST_INGEST_STATUS_ONLINE)
365
+ s->db.liveness = RRDHOST_DB_LIVENESS_LIVE;
366
+ else
367
+ s->db.liveness = RRDHOST_DB_LIVENESS_STALE;
368
350
- if (!replication_in_progress) {
351
- rrdhost_receiver_lock(host);
352
- if (host->receiver && rrdhost_flag_check(host, RRDHOST_FLAG_COLLECTOR_ONLINE)) {
353
- replication_instances = rrdhost_receiver_replicating_charts(host);
354
- replication_in_progress = replication_instances > 0;
355
- }
356
- rrdhost_receiver_unlock(host);
357
- }
369
+ // --- stream ---
370
359
- // Compute ingest status
360
- if (online) {
361
- if (db_initializing) {
362
- ingest_status = RRDHOST_INGEST_STATUS_INITIALIZING;
363
- } else if (rrdhost_is_local(host)) {
364
- ingest_status = RRDHOST_INGEST_STATUS_ONLINE;
365
- ingest_since = netdata_start_time;
366
- } else {
367
- if (replication_in_progress || !collected_metrics) {
368
- ingest_status = RRDHOST_INGEST_STATUS_REPLICATING;
369
- } else {
370
- ingest_status = RRDHOST_INGEST_STATUS_ONLINE;
371
- }
372
- }
373
- } else {
374
- if (!ingest_since) {
375
- ingest_status = RRDHOST_INGEST_STATUS_ARCHIVED;
376
- ingest_since = db_last_time_s;
377
- } else {
378
- ingest_status = RRDHOST_INGEST_STATUS_OFFLINE;
379
- }
380
- }
371
+ if(info & (RRDHOST_STATUS_STREAM | RRDHOST_STATUS_ML))
372
+ rrdhost_status_stream_internal(s);
373
+
374
+ // --- ml ---
375
+
376
+ if(info & RRDHOST_STATUS_ML)
377
+ rrdhost_status_ml_internal(s);
378
+
379
+ // --- dyncfg ---
380
+
381
+ if(info & RRDHOST_STATUS_DYNCFG)
382
+ s->dyncfg.status = dyncfg_available_for_rrdhost(host) ? RRDHOST_DYNCFG_STATUS_AVAILABLE : RRDHOST_DYNCFG_STATUS_UNAVAILABLE;
383
+
384
+ // --- health ---
385
+
386
+ if(info & RRDHOST_STATUS_HEALTH)
387
+ rrdhost_status_health_internal(s, flags);
388
382
- return ingest_status;
389
}
390
+
391
+// Minimal function to get the ingest status only
392
+RRDHOST_INGEST_STATUS rrdhost_get_ingest_status(RRDHOST *host, time_t now) {
393
+ RRDHOST_FLAGS flags = __atomic_load_n(&host->flags, __ATOMIC_RELAXED);
394
+ bool online = rrdhost_is_local(host) || rrdhost_is_online_flags(flags);
395
+
396
+ RRDHOST_DB_STATUS db_status = rrdhost_status_db(host, now, NULL, flags, online);
397
+ return rrdhost_status_ingest(host, NULL, flags, db_status, online);
398
+}
399
+
src/database/rrdhost-status.h
+15
-5
@@ -5,6 +5,16 @@
5
6
#include "libnetdata/libnetdata.h"
7
8
+typedef enum __attribute__((packed)) {
9
+ RRDHOST_STATUS_BASIC = 0,
10
+ RRDHOST_STATUS_STREAM = (1 << 0),
11
+ RRDHOST_STATUS_ML = (1 << 1),
12
+ RRDHOST_STATUS_DYNCFG = (1 << 2),
13
+ RRDHOST_STATUS_HEALTH = (1 << 3),
14
+} RRDHOST_STATUS_INFO;
15
+
16
+#define RRDHOST_STATUS_ALL (RRDHOST_STATUS_BASIC|RRDHOST_STATUS_STREAM|RRDHOST_STATUS_ML|RRDHOST_STATUS_DYNCFG|RRDHOST_STATUS_HEALTH)
17
+
18
typedef enum __attribute__((packed)) {
19
RRDHOST_DB_STATUS_INITIALIZING = 0,
20
RRDHOST_DB_STATUS_QUERYABLE,
@@ -121,15 +131,15 @@ typedef struct rrdhost_status_t {
131
STREAM_HANDSHAKE reason;
132
133
struct {
124
- size_t metrics; // currently collected
125
- size_t instances; // currently collected
126
- size_t contexts; // currently collected
134
+ uint32_t metrics; // currently collected
135
+ uint32_t instances; // currently collected
136
+ uint32_t contexts; // currently collected
137
} collected;
138
139
struct {
140
bool in_progress;
141
NETDATA_DOUBLE completion;
132
- size_t instances;
142
+ uint32_t instances;
143
} replication;
144
} ingest;
145
@@ -165,7 +175,7 @@ typedef struct rrdhost_status_t {
175
} health;
176
} RRDHOST_STATUS;
177
168
-void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s);
178
+void rrdhost_status(RRDHOST *host, time_t now, RRDHOST_STATUS *s, RRDHOST_STATUS_INFO info);
179
RRDHOST_INGEST_STATUS rrdhost_get_ingest_status(RRDHOST *host, time_t now);
180
RRDHOST_INGEST_STATUS rrdhost_ingestion_status(RRDHOST *host);
181
int16_t rrdhost_ingestion_hops(RRDHOST *host);
src/database/rrdhost.h
+9
-4
@@ -357,10 +357,15 @@ extern RRDHOST *localhost;
357
rrdhost_option_check(host, RRDHOST_OPTION_VIRTUAL_HOST) \
358
)
359
360
-#define rrdhost_is_online(host) ( \
361
- rrdhost_is_local(host) || \
362
- (rrdhost_flag_check(host, RRDHOST_FLAG_COLLECTOR_ONLINE) && !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)) \
363
- )
360
+#define rrdhost_is_online_flags(flags) ((flags & RRDHOST_FLAG_COLLECTOR_ONLINE) && !(flags & RRDHOST_FLAG_ORPHAN))
361
+
362
+static inline bool rrdhost_is_online(RRDHOST *host) {
363
+ if(rrdhost_is_local(host))
364
+ return true;
365
+
366
+ RRDHOST_FLAGS flags = rrdhost_flag_get(host);
367
+ return rrdhost_is_online_flags(flags);
368
+}
369
370
bool rrdhost_matches_window(RRDHOST *host, time_t after, time_t before, time_t now);
371
src/streaming/stream-parents.c
+1
-1
@@ -298,7 +298,7 @@ int stream_info_to_json_v1(BUFFER *wb, const char *machine_guid) {
298
if(!machine_guid || !*machine_guid || !(host = rrdhost_find_by_guid(machine_guid)))
299
ret = HTTP_RESP_NOT_FOUND;
300
else
301
- rrdhost_status(host, now_realtime_sec(), &status);
301
+ rrdhost_status(host, now_realtime_sec(), &status, RRDHOST_STATUS_BASIC);
302
303
buffer_json_member_add_uint64(wb, "version", 1);
304
buffer_json_member_add_uint64(wb, "status", ret);
src/streaming/stream-receiver-internals.h
+1
-2
@@ -78,8 +78,7 @@ struct receiver_state {
78
} thread;
79
80
struct {
81
- uint32_t last_counter_in; // copy from the host, to detect progress
82
- uint32_t last_counter_out; // copy from the host, to detect progress
81
+ uint32_t last_counter_sum; // copy from the host, to detect progress
82
usec_t last_progress_ut; // last time we found some progress (monotonic)
83
usec_t last_checked_ut; // last time we checked for stalled progress (monotonic)
84
src/streaming/stream-receiver.c
+9
-10
@@ -404,7 +404,7 @@ void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct re
404
"STREAM RCV '%s' [from [%s]:%s]: failed to set non-blocking mode on socket %d",
405
rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);
406
407
- rpt->host->stream.rcv.status.tid = gettid_cached();
407
+ __atomic_store_n(&rpt->host->stream.rcv.status.tid, gettid_cached(), __ATOMIC_RELAXED);
408
rpt->thread.meta.type = POLLFD_TYPE_RECEIVER;
409
rpt->thread.meta.rpt = rpt;
410
@@ -550,7 +550,7 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st
550
if(!nd_poll_del(sth->run.ndpl, rpt->sock.fd))
551
nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to delete receiver socket from nd_poll()");
552
553
- rpt->host->stream.rcv.status.tid = 0;
553
+ __atomic_store_n(&rpt->host->stream.rcv.status.tid, 0, __ATOMIC_RELAXED);
554
555
// make sure send_to_plugin() will not write any data to the socket (or wait for it to finish)
556
if(parser) {
@@ -970,19 +970,18 @@ void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t
970
static bool stream_receiver_did_replication_progress(struct receiver_state *rpt) {
971
RRDHOST *host = rpt->host;
972
973
- size_t my_counter_in = __atomic_load_n(&rpt->replication.last_counter_in, __ATOMIC_RELAXED);
974
- size_t my_counter_out = __atomic_load_n(&rpt->replication.last_counter_out, __ATOMIC_RELAXED);
975
- size_t host_counter_in = __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED);
976
- size_t host_counter_out = __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED);
977
- if(my_counter_in != host_counter_in || my_counter_out != host_counter_out) {
973
+ size_t host_counter_sum =
974
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED) +
975
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED);
976
+
977
+ if(rpt->replication.last_counter_sum != host_counter_sum) {
978
// there has been some progress
979
- __atomic_store_n(&rpt->replication.last_counter_in, __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
980
- __atomic_store_n(&rpt->replication.last_counter_out, __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
979
+ rpt->replication.last_counter_sum = host_counter_sum;
980
rpt->replication.last_progress_ut = now_monotonic_usec();
981
return true;
982
}
983
985
- if(!my_counter_in || !my_counter_out)
984
+ if(!host_counter_sum)
985
// we have not started yet
986
return true;
987
src/streaming/stream-sender-commit.c
+1
-5
@@ -76,11 +76,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff
76
if (unlikely(!src || !src_len))
77
return;
78
79
- waitq_acquire(
80
- &s->waitq,
81
- (s->host->stream.rcv.status.tid == gettid_cached() || s->host->stream.snd.status.tid == gettid_cached()) ?
82
- WAITQ_PRIO_HIGH :
83
- WAITQ_PRIO_NORMAL);
79
+ waitq_acquire(&s->waitq, (rrdhost_is_this_a_stream_thread(s->host)) ? WAITQ_PRIO_HIGH : WAITQ_PRIO_NORMAL);
80
stream_sender_lock(s);
81
82
// copy the sequence number of sender buffer recreates, while having our lock
src/streaming/stream-sender-internals.h
+1
-2
@@ -85,8 +85,7 @@ struct sender_state {
85
} exit;
86
87
struct {
88
- uint32_t last_counter_in; // copy from the host, to detect progress
89
- uint32_t last_counter_out; // copy from the host, to detect progress
88
+ uint32_t last_counter_sum; // copy from the host, to detect progress
89
usec_t last_progress_ut; // last time we found some progress (monotonic)
90
usec_t last_checked_ut; // last time we checked for stalled progress (monotonic)
91
src/streaming/stream-sender.c
+9
-10
@@ -313,7 +313,7 @@ void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth) {
313
s->thread.msg.session = os_random32();
314
s->thread.msg.meta = &s->thread.meta;
315
316
- s->host->stream.snd.status.tid = gettid_cached();
316
+ __atomic_store_n(&s->host->stream.snd.status.tid, gettid_cached(), __ATOMIC_RELAXED);
317
s->host->stream.snd.status.connections++;
318
s->last_state_since_t = now_realtime_sec();
319
@@ -425,7 +425,7 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
425
s->thread.msg.session = 0;
426
s->thread.msg.meta = NULL;
427
428
- s->host->stream.snd.status.tid = 0;
428
+ __atomic_store_n(&s->host->stream.snd.status.tid, 0, __ATOMIC_RELAXED);
429
stream_sender_unlock(s);
430
431
stream_sender_log_disconnection(sth, s, reason, receiver_reason);
@@ -528,19 +528,18 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t n
528
static bool stream_sender_did_replication_progress(struct sender_state *s) {
529
RRDHOST *host = s->host;
530
531
- size_t my_counter_in = __atomic_load_n(&s->replication.last_counter_in, __ATOMIC_RELAXED);
532
- size_t my_counter_out = __atomic_load_n(&s->replication.last_counter_out, __ATOMIC_RELAXED);
533
- size_t host_counter_in = __atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED);
534
- size_t host_counter_out = __atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED);
535
- if(my_counter_in != host_counter_in || my_counter_out != host_counter_out) {
531
+ size_t host_counter_sum =
532
+ __atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED) +
533
+ __atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED);
534
+
535
+ if(s->replication.last_counter_sum != host_counter_sum) {
536
// there has been some progress
537
- __atomic_store_n(&s->replication.last_counter_in, __atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
538
- __atomic_store_n(&s->replication.last_counter_out, __atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
537
+ s->replication.last_counter_sum = host_counter_sum;
538
s->replication.last_progress_ut = now_monotonic_usec();
539
return true;
540
}
541
543
- if(!my_counter_in || !my_counter_out)
542
+ if(!host_counter_sum)
543
// we have not started yet
544
return true;
545
src/streaming/stream-thread.c
+27
-25
@@ -524,6 +524,8 @@ void *stream_thread(void *ptr) {
524
usec_t now_ut = now_monotonic_usec();
525
526
if(now_ut - last_dequeue_ut >= 100 * USEC_PER_MS) {
527
+ last_dequeue_ut = now_ut;
528
+
529
worker_is_busy(WORKER_STREAM_JOB_DEQUEUE);
530
531
stream_thread_messages_resize(sth);
@@ -538,40 +540,40 @@ void *stream_thread(void *ptr) {
540
541
receivers_waiting = sth->queue.receivers_waiting;
542
spinlock_unlock(&sth->queue.spinlock);
541
- last_dequeue_ut = now_ut;
542
- }
543
544
- if(now_ut - last_check_all_nodes_ut >= nd_profile.update_every * USEC_PER_SEC) {
545
- worker_is_busy(WORKER_STREAM_JOB_LIST);
544
547
- // periodically check the entire list of nodes
548
- // this detects unresponsive parents too (timeout)
549
- stream_sender_check_all_nodes_from_poll(sth, now_ut);
550
- stream_receiver_check_all_nodes_from_poll(sth, now_ut);
545
+ if(now_ut - last_check_all_nodes_ut >= nd_profile.update_every * USEC_PER_SEC) {
546
+ last_check_all_nodes_ut = now_ut;
547
552
- worker_set_metric(WORKER_SENDER_JOB_MESSAGES, (NETDATA_DOUBLE)(sth->messages.processed));
553
- worker_set_metric(WORKER_STREAM_METRIC_NODES, (NETDATA_DOUBLE)sth->nodes_count);
548
+ worker_is_busy(WORKER_STREAM_JOB_LIST);
549
555
- worker_set_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, (NETDATA_DOUBLE)sth->snd.bytes_received);
556
- worker_set_metric(WORKER_SENDER_JOB_BYTES_SENT, (NETDATA_DOUBLE)sth->snd.bytes_sent);
557
- worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE)replay_entries);
550
+ // periodically check the entire list of nodes
551
+ // this detects unresponsive parents too (timeout)
552
+ stream_sender_check_all_nodes_from_poll(sth, now_ut);
553
+ stream_receiver_check_all_nodes_from_poll(sth, now_ut);
554
559
- worker_set_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting);
560
- worker_set_metric(WORKER_STREAM_JOB_SEND_MISSES, (NETDATA_DOUBLE)sth->snd.send_misses);
561
- replay_entries = 0;
562
- sth->snd.bytes_received = 0;
563
- sth->snd.bytes_sent = 0;
555
+ worker_set_metric(WORKER_SENDER_JOB_MESSAGES, (NETDATA_DOUBLE)(sth->messages.processed));
556
+ worker_set_metric(WORKER_STREAM_METRIC_NODES, (NETDATA_DOUBLE)sth->nodes_count);
557
565
- last_check_all_nodes_ut = now_ut;
566
- }
558
+ worker_set_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, (NETDATA_DOUBLE)sth->snd.bytes_received);
559
+ worker_set_metric(WORKER_SENDER_JOB_BYTES_SENT, (NETDATA_DOUBLE)sth->snd.bytes_sent);
560
+ worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE)replay_entries);
561
+
562
+ worker_set_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting);
563
+ worker_set_metric(WORKER_STREAM_JOB_SEND_MISSES, (NETDATA_DOUBLE)sth->snd.send_misses);
564
+ replay_entries = 0;
565
+ sth->snd.bytes_received = 0;
566
+ sth->snd.bytes_sent = 0;
567
568
- if(now_ut - last_check_replication_ut >= 10 * 60 * USEC_PER_SEC) {
569
- worker_is_busy(WORKER_STREAM_JOB_LIST);
568
+ if(now_ut - last_check_replication_ut >= 10 * 60 * USEC_PER_SEC) {
569
+ last_check_replication_ut = now_ut;
570
571
- stream_sender_replication_check_from_poll(sth, now_ut);
572
- stream_receiver_replication_check_from_poll(sth, now_ut);
571
+ worker_is_busy(WORKER_STREAM_JOB_LIST);
572
574
- last_check_replication_ut = now_ut;
573
+ stream_sender_replication_check_from_poll(sth, now_ut);
574
+ stream_receiver_replication_check_from_poll(sth, now_ut);
575
+ }
576
+ }
577
}
578
579
worker_is_idle();
src/web/api/functions/function-streaming.c
+1
-1
@@ -50,7 +50,7 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
50
RRDHOST *host;
51
dfe_start_read(rrdhost_root_index, host) {
52
RRDHOST_STATUS s;
53
- rrdhost_status(host, now, &s);
53
+ rrdhost_status(host, now, &s, RRDHOST_STATUS_ALL);
54
buffer_json_add_array_item_array(wb);
55
56
if(s.db.metrics > max_db_metrics)