diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2022-08-12 07:26:11 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2022-08-12 07:26:11 +0000 |
commit | 3c315f0fff93aa072472abc10815963ac0035268 (patch) | |
tree | a95f6a96e0e7bd139c010f8dc60b40e5b3062a99 /streaming/sender.c | |
parent | Adding upstream version 1.35.1. (diff) | |
download | netdata-3c315f0fff93aa072472abc10815963ac0035268.tar.xz netdata-3c315f0fff93aa072472abc10815963ac0035268.zip |
Adding upstream version 1.36.0.upstream/1.36.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | streaming/sender.c | 42 |
1 files changed, 23 insertions, 19 deletions
diff --git a/streaming/sender.c b/streaming/sender.c index a95cc8673..c4836aeaf 100644 --- a/streaming/sender.c +++ b/streaming/sender.c @@ -80,7 +80,7 @@ void sender_commit(struct sender_state *s) { static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) { - host->rrdpush_sender_connected = 0; + __atomic_clear(&host->rrdpush_sender_connected, __ATOMIC_SEQ_CST); if(host->rrdpush_sender_socket != -1) { close(host->rrdpush_sender_socket); @@ -89,20 +89,20 @@ static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) { } static inline void rrdpush_sender_add_host_variable_to_buffer_nolock(RRDHOST *host, RRDVAR *rv) { - calculated_number *value = (calculated_number *)rv->value; + NETDATA_DOUBLE *value = (NETDATA_DOUBLE *)rv->value; buffer_sprintf( host->sender->build - , "VARIABLE HOST %s = " CALCULATED_NUMBER_FORMAT "\n" + , "VARIABLE HOST %s = " NETDATA_DOUBLE_FORMAT "\n" , rv->name , *value ); - debug(D_STREAM, "RRDVAR pushed HOST VARIABLE %s = " CALCULATED_NUMBER_FORMAT, rv->name, *value); + debug(D_STREAM, "RRDVAR pushed HOST VARIABLE %s = " NETDATA_DOUBLE_FORMAT, rv->name, *value); } void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, RRDVAR *rv) { - if(host->rrdpush_send_enabled && host->rrdpush_sender_spawn && host->rrdpush_sender_connected) { + if(host->rrdpush_send_enabled && host->rrdpush_sender_spawn && __atomic_load_n(&host->rrdpush_sender_connected, __ATOMIC_SEQ_CST)) { sender_start(host->sender); rrdpush_sender_add_host_variable_to_buffer_nolock(host, rv); sender_commit(host->sender); @@ -172,8 +172,8 @@ static inline void rrdpush_sender_thread_data_flush(RRDHOST *host) { } static inline void rrdpush_set_flags_to_newest_stream(RRDHOST *host) { - host->labels.labels_flag |= LABEL_FLAG_UPDATE_STREAM; - host->labels.labels_flag &= ~LABEL_FLAG_STOP_STREAM; + rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_LABELS_UPDATE); + rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_LABELS_STOP); } void rrdpush_encode_variable(stream_encoded_t *se, RRDHOST *host) @@ -236,8 +236,8 @@ static inline long int parse_stream_version(RRDHOST *host, char *http) answer = memcmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT)); if (!answer) { stream_version = 0; - host->labels.labels_flag |= LABEL_FLAG_STOP_STREAM; - host->labels.labels_flag &= ~LABEL_FLAG_UPDATE_STREAM; + rrdhost_flag_set(host, RRDHOST_FLAG_STREAM_LABELS_STOP); + rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_LABELS_UPDATE); } else { stream_version = parse_stream_version_for_errors(http); @@ -563,7 +563,7 @@ static void attempt_to_connect(struct sender_state *state) state->sent_bytes_on_this_connection = 0; // let the data collection threads know we are ready - state->host->rrdpush_sender_connected = 1; + __atomic_test_and_set(&state->host->rrdpush_sender_connected, __ATOMIC_SEQ_CST); } else { // increase the failed connections counter @@ -724,17 +724,21 @@ static void rrdpush_sender_thread_cleanup_callback(void *ptr) { netdata_mutex_unlock(&host->sender->mutex); } -void sender_init(struct sender_state *s, RRDHOST *parent) { - memset(s, 0, sizeof(*s)); - s->host = parent; - s->buffer = cbuffer_new(1024, 1024*1024); - s->build = buffer_create(1); +void sender_init(RRDHOST *parent) +{ + if (parent->sender) + return; + + parent->sender = callocz(1, sizeof(*parent->sender)); + parent->sender->host = parent; + parent->sender->buffer = cbuffer_new(1024, 1024*1024); + parent->sender->build = buffer_create(1); #ifdef ENABLE_COMPRESSION - s->rrdpush_compression = default_compression_enabled; + parent->sender->rrdpush_compression = default_compression_enabled; if (default_compression_enabled) - s->compressor = create_compressor(); + parent->sender->compressor = create_compressor(); #endif - netdata_mutex_init(&s->mutex); + netdata_mutex_init(&parent->sender->mutex); } void *rrdpush_sender_thread(void *ptr) { @@ -770,7 +774,7 @@ void *rrdpush_sender_thread(void *ptr) { remote_clock_resync_iterations); // TODO: REMOVE FOR SLEW / GAPFILLING // initialize rrdpush globals - s->host->rrdpush_sender_connected = 0; + __atomic_clear(&s->host->rrdpush_sender_connected, __ATOMIC_SEQ_CST); if(pipe(s->host->rrdpush_sender_pipe) == -1) { error("STREAM %s [send]: cannot create required pipe. DISABLING STREAMING THREAD", s->host->hostname); return NULL; |