| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "stream-thread.h" |
| 4 | #include "stream-replication-sender.h" |
| 5 | |
| 6 | static __thread struct sender_buffer commit___thread = { 0 }; |
| 7 | |
| 8 | 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; |
| 16 | } |
| 17 | |
| 18 | void sender_thread_buffer_free(void) { |
| 19 | sender_buffer_destroy(&commit___thread); |
| 20 | } |
| 21 | |
| 22 | void sender_host_buffer_free(RRDHOST *host) { |
| 23 | sender_buffer_destroy(&host->stream.snd.commit); |
| 24 | } |
| 25 | |
| 26 | // Collector thread starting a transmission |
| 27 | static BUFFER *sender_commit_start_with_trace(struct sender_state *s, struct sender_buffer *commit, size_t default_size, const char *func) { |
| 28 | if(unlikely(commit->used)) |
| 29 | fatal("STREAM SND '%s' [to %s]: thread buffer is used multiple times concurrently (%u). " |
| 30 | "It is already being used by '%s()', and now is called by '%s()'", |
| 31 | rrdhost_hostname(s->host), s->remote_ip, |
| 32 | (unsigned)commit->used, |
| 33 | commit->last_function ? commit->last_function : "(null)", |
| 34 | func ? func : "(null)"); |
| 35 | |
| 36 | if(unlikely(commit->receiver_tid && commit->receiver_tid != gettid_cached())) |
| 37 | fatal("STREAM SND '%s' [to %s]: thread buffer is reserved for tid %d, but it used by thread %d function '%s()'.", |
| 38 | rrdhost_hostname(s->host), s->remote_ip, |
| 39 | commit->receiver_tid, gettid_cached(), func ? func : "(null)"); |
| 40 | |
| 41 | if(unlikely(commit->wb && |
| 42 | commit->wb->size > default_size && |
| 43 | commit->our_recreates != commit->sender_recreates)) { |
| 44 | buffer_free(commit->wb); |
| 45 | commit->wb = NULL; |
| 46 | } |
| 47 | |
| 48 | if(unlikely(!commit->wb)) { |
| 49 | commit->wb = buffer_create(default_size, &netdata_buffers_statistics.buffers_streaming); |
| 50 | commit->our_recreates = commit->sender_recreates; |
| 51 | } |
| 52 | |
| 53 | commit->used = true; |
| 54 | |
| 55 | if(!commit->reused) |
| 56 | buffer_flush(commit->wb); |
| 57 | |
| 58 | return commit->wb; |
| 59 | } |
| 60 | |
| 61 | BUFFER *sender_thread_buffer_with_trace(struct sender_state *s, size_t default_size, const char *func) { |
| 62 | return sender_commit_start_with_trace(s, &commit___thread, default_size, func); |
| 63 | } |
| 64 | |
| 65 | BUFFER *sender_host_buffer_with_trace(struct rrdhost *host, const char *func) { |
| 66 | return sender_commit_start_with_trace(host->sender, &host->stream.snd.commit, HOST_THREAD_BUFFER_INITIAL_SIZE, func); |
| 67 | } |
| 68 | |
| 69 | // Collector thread finishing a transmission |
| 70 | void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buffer *commit, STREAM_TRAFFIC_TYPE type) { |
| 71 | struct stream_opcode msg; |
| 72 | |
| 73 | char *src = (char *)buffer_tostring(wb); |
| 74 | size_t src_len = buffer_strlen(wb); |
| 75 | |
| 76 | if (unlikely(!src || !src_len)) |
| 77 | return; |
| 78 | |
| 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 |
| 83 | STREAM_CIRCULAR_BUFFER_STATS *stats = stream_circular_buffer_stats_unsafe(s->scb); |
| 84 | if(commit) |
| 85 | commit->sender_recreates = stats->recreates; |
| 86 | |
| 87 | if (!s->thread.msg.session) { |
| 88 | // the dispatcher is not there anymore - ignore these data |
| 89 | |
| 90 | if(commit) |
| 91 | sender_buffer_destroy(commit); |
| 92 | |
| 93 | stream_sender_unlock(s); |
| 94 | waitq_release(&s->waitq); |
| 95 | return; |
| 96 | } |
| 97 | |
| 98 | if (unlikely(stream_circular_buffer_set_max_size_unsafe( |
| 99 | s->scb, src_len * STREAM_CIRCULAR_BUFFER_ADAPT_TO_TIMES_MAX_SIZE, false))) { |
| 100 | // adaptive sizing of the circular buffer |
| 101 | nd_log(NDLS_DAEMON, NDLP_NOTICE, |
| 102 | "STREAM SND '%s' [to %s]: Increased max buffer size to %u (message size %zu).", |
| 103 | rrdhost_hostname(s->host), s->remote_ip, stats->bytes_max_size, src_len + 1); |
| 104 | } |
| 105 | |
| 106 | stream_sender_log_payload(s, wb, type, false); |
| 107 | |
| 108 | // if there are data already in the buffer, we don't need to send an opcode |
| 109 | bool enable_sending = stats->bytes_outstanding == 0; |
| 110 | |
| 111 | if (s->thread.compressor.initialized) { |
| 112 | // compressed traffic |
| 113 | if(rrdhost_is_this_a_stream_thread(s->host)) |
| 114 | worker_is_busy(WORKER_STREAM_JOB_COMPRESS); |
| 115 | |
| 116 | while (src_len) { |
| 117 | size_t size_to_compress = src_len; |
| 118 | |
| 119 | if (unlikely(size_to_compress > COMPRESSION_MAX_MSG_SIZE)) { |
| 120 | if (stream_has_capability(s, STREAM_CAP_BINARY)) |
| 121 | size_to_compress = COMPRESSION_MAX_MSG_SIZE; |
| 122 | else { |
| 123 | if (size_to_compress > COMPRESSION_MAX_MSG_SIZE) { |
| 124 | // we need to find the last newline |
| 125 | // so that the decompressor will have a whole line to work with |
| 126 | |
| 127 | const char *t = &src[COMPRESSION_MAX_MSG_SIZE]; |
| 128 | while (--t >= src) |
| 129 | if (unlikely(*t == '\n')) |
| 130 | break; |
| 131 | |
| 132 | if (t <= src) |
| 133 | size_to_compress = COMPRESSION_MAX_MSG_SIZE; |
| 134 | else |
| 135 | size_to_compress = t - src + 1; |
| 136 | } |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | const char *dst; |
| 141 | size_t dst_len = stream_compress(&s->thread.compressor, src, size_to_compress, &dst); |
| 142 | if (!dst_len) { |
| 143 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 144 | "STREAM SND '%s' [to %s]: COMPRESSION failed. Resetting compressor and re-trying", |
| 145 | rrdhost_hostname(s->host), s->remote_ip); |
| 146 | |
| 147 | stream_compression_initialize(s); |
| 148 | dst_len = stream_compress(&s->thread.compressor, src, size_to_compress, &dst); |
| 149 | if (!dst_len) |
| 150 | goto compression_failed_with_lock; |
| 151 | } |
| 152 | |
| 153 | stream_compression_signature_t signature = stream_compress_encode_signature(dst_len); |
| 154 | |
| 155 | #ifdef NETDATA_INTERNAL_CHECKS |
| 156 | // check if reversing the signature provides the same length |
| 157 | size_t decoded_dst_len = stream_decompress_decode_signature((const char *)&signature, sizeof(signature)); |
| 158 | if (decoded_dst_len != dst_len) |
| 159 | fatal( |
| 160 | "STREAM SND '%s' [to %s]: invalid signature, original payload %zu bytes, " |
| 161 | "compressed payload length %zu bytes, but signature says payload is %zu bytes", |
| 162 | rrdhost_hostname(s->host), s->remote_ip, |
| 163 | size_to_compress, dst_len, decoded_dst_len); |
| 164 | #endif |
| 165 | |
| 166 | if (!stream_circular_buffer_add_unsafe(s->scb, (const char *)&signature, sizeof(signature), |
| 167 | sizeof(signature), type, false) || |
| 168 | !stream_circular_buffer_add_unsafe(s->scb, dst, dst_len, |
| 169 | size_to_compress, type, false)) |
| 170 | goto overflow_with_lock; |
| 171 | |
| 172 | src = src + size_to_compress; |
| 173 | src_len -= size_to_compress; |
| 174 | } |
| 175 | } |
| 176 | else { |
| 177 | // uncompressed traffic |
| 178 | |
| 179 | if (!stream_circular_buffer_add_unsafe(s->scb, src, src_len, |
| 180 | src_len, type, false)) |
| 181 | goto overflow_with_lock; |
| 182 | } |
| 183 | |
| 184 | replication_sender_recalculate_buffer_used_ratio_unsafe(s); |
| 185 | |
| 186 | if (enable_sending) |
| 187 | msg = s->thread.msg; |
| 188 | |
| 189 | stream_sender_unlock(s); |
| 190 | waitq_release(&s->waitq); |
| 191 | |
| 192 | if (enable_sending) { |
| 193 | msg.opcode = STREAM_OPCODE_SENDER_POLLOUT; |
| 194 | msg.reason = 0; |
| 195 | stream_sender_send_opcode(s, msg); |
| 196 | } |
| 197 | |
| 198 | return; |
| 199 | |
| 200 | overflow_with_lock: { |
| 201 | msg = s->thread.msg; |
| 202 | stream_sender_unlock(s); |
| 203 | waitq_release(&s->waitq); |
| 204 | msg.opcode = STREAM_OPCODE_SENDER_BUFFER_OVERFLOW; |
| 205 | msg.reason = STREAM_HANDSHAKE_DISCONNECT_BUFFER_OVERFLOW; |
| 206 | stream_sender_send_opcode(s, msg); |
| 207 | nd_log_limit_static_global_var(erl, 1, 0); |
| 208 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 209 | "STREAM SND '%s' [to %s]: buffer overflow (buffer size %u, max size %u, available %u). " |
| 210 | "Restarting connection.", |
| 211 | rrdhost_hostname(s->host), s->remote_ip, |
| 212 | stats->bytes_size, stats->bytes_max_size, stats->bytes_available); |
| 213 | return; |
| 214 | } |
| 215 | |
| 216 | compression_failed_with_lock: { |
| 217 | stream_compression_deactivate(s); |
| 218 | msg = s->thread.msg; |
| 219 | stream_sender_unlock(s); |
| 220 | waitq_release(&s->waitq); |
| 221 | msg.opcode = STREAM_OPCODE_SENDER_RECONNECT_WITHOUT_COMPRESSION; |
| 222 | msg.reason = STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED; |
| 223 | stream_sender_send_opcode(s, msg); |
| 224 | nd_log_limit_static_global_var(erl, 1, 0); |
| 225 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 226 | "STREAM SND '%s' [to %s]: COMPRESSION failed (twice). " |
| 227 | "Deactivating compression and restarting connection.", |
| 228 | rrdhost_hostname(s->host), s->remote_ip); |
| 229 | } |
| 230 | } |
| 231 | |
| 232 | void sender_thread_commit_with_trace(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type, const char *func) { |
| 233 | struct sender_buffer *commit; |
| 234 | bool is_receiver; |
| 235 | |
| 236 | if(unlikely(wb == commit___thread.wb)) { |
| 237 | commit = &commit___thread; |
| 238 | is_receiver = false; |
| 239 | } |
| 240 | else { |
| 241 | commit = &s->host->stream.snd.commit; |
| 242 | is_receiver = commit->receiver_tid == gettid_cached(); |
| 243 | } |
| 244 | |
| 245 | if (unlikely(wb != commit->wb)) |
| 246 | fatal("STREAM SND '%s' [to %s]: function '%s()' is trying to commit an unknown commit buffer.", |
| 247 | rrdhost_hostname(s->host), s->remote_ip, func); |
| 248 | |
| 249 | if (unlikely(!commit->used)) |
| 250 | fatal("STREAM SND '%s' [to %s]: function '%s()' is committing a sender buffer twice.", |
| 251 | rrdhost_hostname(s->host), s->remote_ip, func); |
| 252 | |
| 253 | if(!is_receiver || |
| 254 | type != STREAM_TRAFFIC_TYPE_DATA || |
| 255 | commit->reused >= 100 || |
| 256 | buffer_strlen(wb) >= COMPRESSION_MAX_MSG_SIZE * 2 / 3) { |
| 257 | sender_buffer_commit(s, wb, commit, type); |
| 258 | commit->reused = 0; |
| 259 | } |
| 260 | else |
| 261 | commit->reused++; |
| 262 | |
| 263 | commit->used = false; |
| 264 | commit->last_function = NULL; |
| 265 | } |