Stream receiver/sender compress BEGIN-SET-END performance (#19352)
* reorganization for begin-set-end functions - no functional changes * the stream receiver/sender thread compresses once every 10 begin-set-end * compress once every 100 jobs or when the size of bigger than 2/3 of COMPRESSION_MAX_MSG_SIZE * fix comments
Costa Tsaousis committed
Jan 9, 2025 at 09:05 UTC
2c273898a6670a85284d2fb97acea46f3cb5e536
7 files changed
+169
-135
CMakeLists.txt
+3
-1
@@ -1556,7 +1556,7 @@ set(STREAMING_PLUGIN_FILES
1556
src/streaming/protocol/command-function.c
1557
src/streaming/protocol/command-host-labels.c
1558
src/streaming/protocol/command-chart-definition.c
1559
- src/streaming/protocol/command-begin-set-end.c
1559
+ src/streaming/protocol/command-begin-set-end-v2.c
1560
src/streaming/protocol/command-host-variables.c
1561
src/streaming/stream-conf.c
1562
src/streaming/stream-conf.h
@@ -1582,6 +1582,8 @@ set(STREAMING_PLUGIN_FILES
1582
src/streaming/stream-replication-receiver.h
1583
src/streaming/stream-replication-tracking.c
1584
src/streaming/stream-replication-tracking.h
1585
+ src/streaming/protocol/command-begin-set-end-v1.c
1586
+ src/streaming/protocol/command-begin-set-end-init.c
1587
)
1588
1589
set(WEB_PLUGIN_FILES
src/streaming/protocol/command-begin-set-end-init.c
new
+78
@@ -0,0 +1,78 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "commands.h"
4
+#include "../stream-sender-internals.h"
5
+#include "plugins.d/pluginsd_internals.h"
6
+
7
+static BUFFER *preferred_sender_buffer(RRDHOST *host) {
8
+ if(host->stream.snd.commit.receiver_tid == gettid_cached())
9
+ return sender_host_buffer(host);
10
+ else
11
+ return sender_thread_buffer(host->sender);
12
+}
13
+
14
+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
18
+ RRDHOST_FLAGS host_flags = __atomic_load_n(&host->flags, __ATOMIC_SEQ_CST);
19
+
20
+ // check if we are not connected
21
+ if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS))) {
22
+
23
+ if(unlikely((host_flags & RRDHOST_FLAG_COLLECTOR_ONLINE) &&
24
+ !(host_flags & RRDHOST_FLAG_STREAM_SENDER_ADDED)))
25
+ stream_sender_start_host(host);
26
+
27
+ if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS))) {
28
+ rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS);
29
+
30
+ // this message is logged in 2 cases:
31
+ // - the parent is connected, but not yet available for streaming data
32
+ // - the parent just disconnected, so local data are not streamed to parent
33
+
34
+ nd_log(NDLS_DAEMON, NDLP_INFO,
35
+ "STREAM SND '%s': streaming is not ready, not sending data to a parent...",
36
+ rrdhost_hostname(host));
37
+ }
38
+
39
+ return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
40
+ }
41
+ else if(unlikely(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS)) {
42
+ nd_log(NDLS_DAEMON, NDLP_INFO,
43
+ "STREAM SND '%s': streaming is ready, sending metrics to parent...",
44
+ rrdhost_hostname(host));
45
+ rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS);
46
+ }
47
+
48
+ if(unlikely(host_flags & RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED)) {
49
+ BUFFER *wb = preferred_sender_buffer(host);
50
+ stream_sender_send_global_rrdhost_functions(host, wb, stream_has_capability(host->sender, STREAM_CAP_DYNCFG));
51
+ sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
52
+ }
53
+
54
+ bool exposed_upstream = rrdset_check_upstream_exposed(st);
55
+ RRDSET_FLAGS rrdset_flags = rrdset_flag_get(st);
56
+ bool replication_in_progress = !(rrdset_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
57
+
58
+ if(unlikely((exposed_upstream && replication_in_progress) ||
59
+ !should_send_rrdset_matching(st, rrdset_flags)))
60
+ return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
61
+
62
+ if(unlikely(!exposed_upstream)) {
63
+ BUFFER *wb = preferred_sender_buffer(host);
64
+ replication_in_progress = stream_sender_send_rrdset_definition(wb, st);
65
+ sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
66
+ }
67
+
68
+ if(unlikely(replication_in_progress))
69
+ return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
70
+
71
+ return (RRDSET_STREAM_BUFFER) {
72
+ .capabilities = host->sender->capabilities,
73
+ .v2 = stream_has_capability(host->sender, STREAM_CAP_INTERPOLATED),
74
+ .rrdset_flags = rrdset_flags,
75
+ .wb = preferred_sender_buffer(host),
76
+ .wall_clock_time = wall_clock_time,
77
+ };
78
+}
src/streaming/protocol/command-begin-set-end-v1.c
new
+48
@@ -0,0 +1,48 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "commands.h"
4
+#include "../stream-sender-internals.h"
5
+
6
+void stream_send_rrdset_metrics_v1(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
7
+ RRDHOST *host = st->rrdhost; (void)host;
8
+ BUFFER *wb = rsb->wb;
9
+ struct sender_state *s = host->sender; (void)s;
10
+ RRDSET_FLAGS flags = rsb->rrdset_flags;
11
+
12
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN " \"", 7);
13
+ buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id));
14
+ buffer_fast_strcat(wb, "\" ", 2);
15
+
16
+ if(st->last_collected_time.tv_sec > st->stream.snd.resync_time_s)
17
+ buffer_print_uint64(wb, st->usec_since_last_update);
18
+ else
19
+ buffer_fast_strcat(wb, "0", 1);
20
+
21
+ buffer_fast_strcat(wb, "\n", 1);
22
+
23
+ RRDDIM *rd;
24
+ rrddim_foreach_read(rd, st) {
25
+ if(unlikely(!rrddim_check_updated(rd)))
26
+ continue;
27
+
28
+ if(likely(rrddim_check_upstream_exposed_collector(rd))) {
29
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET " \"", 5);
30
+ buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id));
31
+ buffer_fast_strcat(wb, "\" = ", 4);
32
+ buffer_print_int64(wb, rd->collector.collected_value);
33
+ buffer_fast_strcat(wb, "\n", 1);
34
+ }
35
+ else {
36
+ internal_error(true, "STREAM SND '%s': 'chart:%s/dim:%s' flag 'exposed' is updated but not exposed",
37
+ rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
38
+ // we will include it in the next iteration
39
+ rrddim_metadata_updated(rd);
40
+ }
41
+ }
42
+ rrddim_foreach_done(rd);
43
+
44
+ if(unlikely(flags & RRDSET_FLAG_UPSTREAM_SEND_VARIABLES))
45
+ rrdvar_print_to_streaming_custom_chart_variables(st, wb);
46
+
47
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END "\n", 4);
48
+}
src/streaming/protocol/command-begin-set-end-v2.c
renamed
-45
@@ -4,51 +4,6 @@
4
#include "../stream-sender-internals.h"
5
#include "plugins.d/pluginsd_internals.h"
6
7
-static void
8
-stream_send_rrdset_metrics_v1_internal(BUFFER *wb, RRDSET *st, struct sender_state *s __maybe_unused, RRDSET_FLAGS flags) {
9
- buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN " \"", 7);
10
- buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id));
11
- buffer_fast_strcat(wb, "\" ", 2);
12
-
13
- if(st->last_collected_time.tv_sec > st->stream.snd.resync_time_s)
14
- buffer_print_uint64(wb, st->usec_since_last_update);
15
- else
16
- buffer_fast_strcat(wb, "0", 1);
17
-
18
- buffer_fast_strcat(wb, "\n", 1);
19
-
20
- RRDDIM *rd;
21
- rrddim_foreach_read(rd, st) {
22
- if(unlikely(!rrddim_check_updated(rd)))
23
- continue;
24
-
25
- if(likely(rrddim_check_upstream_exposed_collector(rd))) {
26
- buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET " \"", 5);
27
- buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id));
28
- buffer_fast_strcat(wb, "\" = ", 4);
29
- buffer_print_int64(wb, rd->collector.collected_value);
30
- buffer_fast_strcat(wb, "\n", 1);
31
- }
32
- else {
33
- internal_error(true, "STREAM SND '%s': 'chart:%s/dim:%s' flag 'exposed' is updated but not exposed",
34
- rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
35
- // we will include it in the next iteration
36
- rrddim_metadata_updated(rd);
37
- }
38
- }
39
- rrddim_foreach_done(rd);
40
-
41
- if(unlikely(flags & RRDSET_FLAG_UPSTREAM_SEND_VARIABLES))
42
- rrdvar_print_to_streaming_custom_chart_variables(st, wb);
43
-
44
- buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END "\n", 4);
45
-}
46
-
47
-void stream_send_rrdset_metrics_v1(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
48
- RRDHOST *host = st->rrdhost;
49
- stream_send_rrdset_metrics_v1_internal(rsb->wb, st, host->sender, rsb->rrdset_flags);
50
-}
51
-
7
void stream_send_rrddim_metrics_v2(RRDSET_STREAM_BUFFER *rsb, RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags) {
8
if(!rsb->wb || !rsb->v2 || !netdata_double_isnumber(n) || !does_storage_number_exist(flags))
9
return;
src/streaming/protocol/commands.c
-74
@@ -1,77 +1,3 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "commands.h"
4
-#include "../stream-sender-internals.h"
5
-
6
-static BUFFER *preferred_sender_buffer(RRDHOST *host) {
7
- if(host->stream.snd.commit.receiver_tid == gettid_cached())
8
- return sender_host_buffer(host);
9
- else
10
- return sender_thread_buffer(localhost->sender);
11
-}
12
-
13
-RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time) {
14
- RRDHOST *host = st->rrdhost;
15
-
16
- // fetch the flags we need to check with one atomic operation
17
- RRDHOST_FLAGS host_flags = __atomic_load_n(&host->flags, __ATOMIC_SEQ_CST);
18
-
19
- // check if we are not connected
20
- if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS))) {
21
-
22
- if(unlikely((host_flags & RRDHOST_FLAG_COLLECTOR_ONLINE) &&
23
- !(host_flags & RRDHOST_FLAG_STREAM_SENDER_ADDED)))
24
- stream_sender_start_host(host);
25
-
26
- if(unlikely(!(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS))) {
27
- rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS);
28
-
29
- // this message is logged in 2 cases:
30
- // - the parent is connected, but not yet available for streaming data
31
- // - the parent just disconnected, so local data are not streamed to parent
32
-
33
- nd_log(NDLS_DAEMON, NDLP_INFO,
34
- "STREAM SND '%s': streaming is not ready, not sending data to a parent...",
35
- rrdhost_hostname(host));
36
- }
37
-
38
- return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
39
- }
40
- else if(unlikely(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS)) {
41
- nd_log(NDLS_DAEMON, NDLP_INFO,
42
- "STREAM SND '%s': streaming is ready, sending metrics to parent...",
43
- rrdhost_hostname(host));
44
- rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS);
45
- }
46
-
47
- if(unlikely(host_flags & RRDHOST_FLAG_GLOBAL_FUNCTIONS_UPDATED)) {
48
- BUFFER *wb = preferred_sender_buffer(host);
49
- stream_sender_send_global_rrdhost_functions(host, wb, stream_has_capability(host->sender, STREAM_CAP_DYNCFG));
50
- sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
51
- }
52
-
53
- bool exposed_upstream = rrdset_check_upstream_exposed(st);
54
- RRDSET_FLAGS rrdset_flags = rrdset_flag_get(st);
55
- bool replication_in_progress = !(rrdset_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
56
-
57
- if(unlikely((exposed_upstream && replication_in_progress) ||
58
- !should_send_rrdset_matching(st, rrdset_flags)))
59
- return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
60
-
61
- if(unlikely(!exposed_upstream)) {
62
- BUFFER *wb = preferred_sender_buffer(host);
63
- replication_in_progress = stream_sender_send_rrdset_definition(wb, st);
64
- sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA);
65
- }
66
-
67
- if(replication_in_progress)
68
- return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
69
-
70
- return (RRDSET_STREAM_BUFFER) {
71
- .capabilities = host->sender->capabilities,
72
- .v2 = stream_has_capability(host->sender, STREAM_CAP_INTERPOLATED),
73
- .rrdset_flags = rrdset_flags,
74
- .wb = preferred_sender_buffer(host),
75
- .wall_clock_time = wall_clock_time,
76
- };
77
-}
src/streaming/stream-sender-commit.c
+28
-6
@@ -9,6 +9,7 @@ void sender_buffer_destroy(struct sender_buffer *commit) {
9
buffer_free(commit->wb);
10
commit->wb = NULL;
11
commit->used = false;
12
+ commit->reused = 0;
13
commit->our_recreates = 0;
14
commit->sender_recreates = 0;
15
commit->last_function = NULL;
@@ -19,7 +20,7 @@ void sender_thread_buffer_free(void) {
20
}
21
22
// Collector thread starting a transmission
22
-BUFFER *sender_commit_start_with_trace(struct sender_state *s __maybe_unused, struct sender_buffer *commit, const char *func) {
23
+BUFFER *sender_commit_start_with_trace(struct sender_state *s, struct sender_buffer *commit, const char *func) {
24
if(unlikely(commit->used))
25
fatal("STREAM SND '%s' [to %s]: thread buffer is used multiple times concurrently (%u). "
26
"It is already being used by '%s()', and now is called by '%s()'",
@@ -46,11 +47,14 @@ BUFFER *sender_commit_start_with_trace(struct sender_state *s __maybe_unused, st
47
}
48
49
commit->used = true;
49
- buffer_flush(commit->wb);
50
+
51
+ if(!commit->reused)
52
+ buffer_flush(commit->wb);
53
+
54
return commit->wb;
55
}
56
53
-BUFFER *sender_thread_buffer_with_trace(struct sender_state *s __maybe_unused, const char *func) {
57
+BUFFER *sender_thread_buffer_with_trace(struct sender_state *s, const char *func) {
58
return sender_commit_start_with_trace(s, &commit___thread, func);
59
}
60
@@ -223,7 +227,17 @@ compression_failed_with_lock: {
227
}
228
229
void sender_thread_commit_with_trace(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type, const char *func) {
226
- struct sender_buffer *commit = (wb == commit___thread.wb) ? &commit___thread : &s->host->stream.snd.commit;
230
+ struct sender_buffer *commit;
231
+ bool is_receiver;
232
+
233
+ if(unlikely(wb == commit___thread.wb)) {
234
+ commit = &commit___thread;
235
+ is_receiver = false;
236
+ }
237
+ else {
238
+ commit = &s->host->stream.snd.commit;
239
+ is_receiver = commit->receiver_tid == gettid_cached();
240
+ }
241
242
if (unlikely(wb != commit->wb))
243
fatal("STREAM SND '%s' [to %s]: function '%s()' is trying to commit an unknown commit buffer.",
@@ -233,8 +247,16 @@ void sender_thread_commit_with_trace(struct sender_state *s, BUFFER *wb, STREAM_
247
fatal("STREAM SND '%s' [to %s]: function '%s()' is committing a sender buffer twice.",
248
rrdhost_hostname(s->host), s->remote_ip, func);
249
250
+ if(!is_receiver ||
251
+ type != STREAM_TRAFFIC_TYPE_DATA ||
252
+ commit->reused >= 100 ||
253
+ buffer_strlen(wb) >= COMPRESSION_MAX_MSG_SIZE * 2 / 3) {
254
+ sender_buffer_commit(s, wb, commit, type);
255
+ commit->reused = 0;
256
+ }
257
+ else
258
+ commit->reused++;
259
+
260
commit->used = false;
261
commit->last_function = NULL;
238
-
239
- sender_buffer_commit(s, wb, commit, type);
262
}
src/streaming/stream-sender-commit.h
+12
-9
@@ -11,35 +11,38 @@ struct sender_state;
11
struct receiver_state;
12
13
struct sender_buffer {
14
- pid_t receiver_tid;
14
+ const char *last_function;
15
BUFFER *wb;
16
+ pid_t receiver_tid;
17
bool used;
17
- size_t our_recreates;
18
- size_t sender_recreates;
19
- const char *last_function;
18
+ uint16_t reused;
19
+ uint32_t our_recreates;
20
+ uint32_t sender_recreates;
21
};
22
void sender_buffer_destroy(struct sender_buffer *commit);
23
24
// thread buffer for sending data upstream (to a parent)
25
26
void sender_thread_buffer_free(void);
27
+
28
+// get the thread buffer
29
+// this is the preferred buffer for dedicated workers sending a lot of messages (like replication)
30
+// these threads need to maintain enough allocation for repeated use of the buffer
31
BUFFER *sender_thread_buffer_with_trace(struct sender_state *s, const char *func);
32
#define sender_thread_buffer(s) sender_thread_buffer_with_trace(s, __FUNCTION__)
33
29
-// commit the global host buffer
34
+// get the global host buffer
35
// this is the preferred buffer for stream threads (unified receiver / sender threads)
36
// these threads require a buffer that can remain intact while switching hosts
37
BUFFER *sender_host_buffer_with_trace(struct rrdhost *host, const char *func);
38
#define sender_host_buffer(host) sender_host_buffer_with_trace(host, __FUNCTION__)
39
35
-// commit a buffer acquired with sender_thread_buffer()
36
-// this is the preferred buffer for dedicated workers sending a lot of messages (like replication)
37
-// these threads need to maintain enough allocation for repeated use of the buffer
40
+// commit a buffer acquired with sender_thread_buffer() or sender_host_buffer()
41
void sender_thread_commit_with_trace(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type, const char *func);
42
#define sender_commit(s, wb, type) sender_thread_commit_with_trace(s, wb, type, __FUNCTION__)
43
44
// commit any buffer
42
-// this is the preferred buffer for occasional senders, as it avoids constant buffer allocations
45
+// this is the preferred buffer for occasional senders, as it avoids a permanently allocated buffer
46
void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buffer *commit, STREAM_TRAFFIC_TYPE type);
47
#define sender_commit_clean_buffer(s, wb, type) sender_buffer_commit(s, wb, NULL, type)
48