| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #define STREAM_INTERNALS |
| 4 | #include "stream-thread.h" |
| 5 | #include "stream-waiting-list.h" |
| 6 | |
| 7 | struct stream_thread_globals stream_thread_globals = { |
| 8 | .assign = { |
| 9 | .spinlock = SPINLOCK_INITIALIZER, |
| 10 | } |
| 11 | }; |
| 12 | |
| 13 | // -------------------------------------------------------------------------------------------------------------------- |
| 14 | // pipe messages |
| 15 | |
| 16 | static void stream_thread_handle_op(struct stream_thread *sth, struct stream_opcode *msg) { |
| 17 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 18 | |
| 19 | sth->messages.processed++; |
| 20 | |
| 21 | struct pollfd_meta *m = META_GET(&sth->run.meta, (Word_t)msg->meta); |
| 22 | if (m && // there is a meta |
| 23 | m == msg->meta && // the meta are equal |
| 24 | msg->session && // there is a session |
| 25 | (size_t)msg->thread_slot == sth->id && // the right thread |
| 26 | (m->type == POLLFD_TYPE_SENDER || m->type == POLLFD_TYPE_RECEIVER) && // it is either sender or receiver |
| 27 | ((m->type == POLLFD_TYPE_SENDER && m == &m->s->thread.meta) || // sender matches |
| 28 | (m->type == POLLFD_TYPE_RECEIVER && m == &m->rpt->thread.meta))) // receiver matches |
| 29 | { |
| 30 | if(m->type == POLLFD_TYPE_SENDER) { |
| 31 | if(msg->opcode & STREAM_OPCODE_SENDER_POLLOUT) { |
| 32 | m->s->thread.wanted = ND_POLL_READ | ND_POLL_WRITE; |
| 33 | if(!nd_poll_upd(sth->run.ndpl, m->s->sock.fd, m->s->thread.wanted)) { |
| 34 | nd_log_limit_static_global_var(erl, 1, 0); |
| 35 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_DEBUG, |
| 36 | "STREAM SND[%zu] '%s' [to %s]: cannot enable output on sender socket %d.", |
| 37 | sth->id, rrdhost_hostname(m->s->host), m->s->remote_ip, m->s->sock.fd); |
| 38 | } |
| 39 | |
| 40 | if(!stream_sender_send_data(sth, m->s, now_monotonic_usec(), false)) |
| 41 | // sender has been removed |
| 42 | return; |
| 43 | |
| 44 | msg->opcode &= ~(STREAM_OPCODE_SENDER_POLLOUT); |
| 45 | } |
| 46 | |
| 47 | if(msg->opcode) |
| 48 | stream_sender_handle_op(sth, m->s, msg); |
| 49 | } |
| 50 | else if(m->type == POLLFD_TYPE_RECEIVER) { |
| 51 | if (msg->opcode & STREAM_OPCODE_RECEIVER_POLLOUT) { |
| 52 | m->rpt->thread.wanted = ND_POLL_READ | ND_POLL_WRITE; |
| 53 | if (!nd_poll_upd(sth->run.ndpl, m->rpt->sock.fd, m->rpt->thread.wanted)) { |
| 54 | nd_log_limit_static_global_var(erl, 1, 0); |
| 55 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 56 | "STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot enable output on receiver socket %d.", |
| 57 | sth->id, rrdhost_hostname(m->rpt->host), m->rpt->remote_ip, m->rpt->remote_port, m->rpt->sock.fd); |
| 58 | } |
| 59 | |
| 60 | if(!stream_receiver_send_data(sth, m->rpt, now_monotonic_usec(), false)) |
| 61 | // receiver has been removed |
| 62 | return; |
| 63 | |
| 64 | msg->opcode &= ~(STREAM_OPCODE_RECEIVER_POLLOUT); |
| 65 | } |
| 66 | |
| 67 | if (msg->opcode) |
| 68 | stream_receiver_handle_op(sth, m->rpt, msg); |
| 69 | } |
| 70 | } |
| 71 | else { |
| 72 | // this may happen if we receive a POLLOUT opcode, but the sender has been disconnected |
| 73 | nd_log_limit_static_global_var(erl, 1, 0); |
| 74 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_DEBUG, "STREAM THREAD[%zu]: OPCODE %u ignored.", sth->id, (unsigned)msg->opcode); |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | static void stream_thread_send_pipe_signal(struct stream_thread *sth) { |
| 79 | if(sth->tid == gettid_cached()) |
| 80 | // no need for this if we are the same thread |
| 81 | // we will process all the events shortly |
| 82 | return; |
| 83 | |
| 84 | if(sth->pipe.fds[PIPE_WRITE] != -1 && |
| 85 | write(sth->pipe.fds[PIPE_WRITE], " ", 1) != 1) { |
| 86 | nd_log_limit_static_global_var(erl, 1, 1 * USEC_PER_MS); |
| 87 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, |
| 88 | "STREAM THREAD[%zu]: cannot write to signal pipe", sth->id); |
| 89 | } |
| 90 | } |
| 91 | |
| 92 | void stream_receiver_send_opcode(struct receiver_state *rpt, struct stream_opcode msg) { |
| 93 | if (!msg.session || !msg.meta || !rpt) |
| 94 | return; |
| 95 | |
| 96 | if(msg.meta != &rpt->thread.meta) { |
| 97 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 98 | "STREAM RCV '%s' [from [%s]:%s]: the receiver in the opcode the message does not match this receiver. " |
| 99 | "Ignoring opcode.", rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port); |
| 100 | return; |
| 101 | } |
| 102 | struct stream_thread *sth = stream_thread_by_slot_id(msg.thread_slot); |
| 103 | if(!sth) { |
| 104 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 105 | "STREAM RCV '%s' [from [%s]:%s]: the opcode (%u) message cannot be verified. Ignoring it.", |
| 106 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, msg.opcode); |
| 107 | return; |
| 108 | } |
| 109 | |
| 110 | // check if we can execute the message now |
| 111 | if(sth->tid == gettid_cached() && msg.opcode == STREAM_OPCODE_RECEIVER_POLLOUT) { |
| 112 | // we are running at the stream thread, and the request is about enabling POLLOUT, |
| 113 | // we can do this synchronously. |
| 114 | // IMPORTANT: DO NOT HANDLE FAILURES THAT REMOVE THE RECEIVER OR THE SENDER THIS WAY |
| 115 | // THE EVENT LOOP DRAINS THE INPUT SOCKET (BOTH RECEIVER AND SENDER) |
| 116 | // AND THE LOOP WILL CRASH IF THE RECEIVER OR THE SENDER VANISH WHILE IT |
| 117 | // WORKS WITH THEM! |
| 118 | sth->messages.bypassed++; |
| 119 | stream_thread_handle_op(sth, &msg); |
| 120 | return; |
| 121 | } |
| 122 | |
| 123 | bool send_pipe_msg = false; |
| 124 | |
| 125 | // add it to the message queue of the thread |
| 126 | spinlock_lock(&sth->messages.spinlock); |
| 127 | { |
| 128 | sth->messages.added++; |
| 129 | if (rpt->thread.send_to_child.msg_slot >= sth->messages.used || sth->messages.array[rpt->thread.send_to_child.msg_slot].meta != &rpt->thread.meta) { |
| 130 | if (unlikely(sth->messages.used >= sth->messages.size)) { |
| 131 | // this should never happen, but let's find the root cause |
| 132 | |
| 133 | if (!sth->messages.size) { |
| 134 | // we are exiting |
| 135 | spinlock_unlock(&sth->messages.spinlock); |
| 136 | return; |
| 137 | } |
| 138 | |
| 139 | #ifdef NETDATA_INTERNAL_CHECKS |
| 140 | // try to find us in the list |
| 141 | for (size_t i = 0; i < sth->messages.size; i++) { |
| 142 | if (sth->messages.array[i].meta == &rpt->thread.meta) { |
| 143 | rpt->thread.send_to_child.msg_slot = i; |
| 144 | sth->messages.array[rpt->thread.send_to_child.msg_slot].opcode |= msg.opcode; |
| 145 | if(msg.reason) |
| 146 | sth->messages.array[rpt->thread.send_to_child.msg_slot].reason = msg.reason; |
| 147 | spinlock_unlock(&sth->messages.spinlock); |
| 148 | internal_fatal(true, "the stream opcode queue is full, but this receiver is already on slot %zu", i); |
| 149 | return; |
| 150 | } |
| 151 | } |
| 152 | #endif |
| 153 | |
| 154 | fatal("STREAM RCV '%s' [from [%s]:%s]: The streaming opcode queue is full, but this should never happen...", |
| 155 | rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port); |
| 156 | } |
| 157 | |
| 158 | // let's use a new slot |
| 159 | send_pipe_msg = !sth->messages.used; // write to the pipe, only when the queue was empty before this msg |
| 160 | rpt->thread.send_to_child.msg_slot = sth->messages.used++; |
| 161 | sth->messages.array[rpt->thread.send_to_child.msg_slot] = msg; |
| 162 | } |
| 163 | else { |
| 164 | // the existing slot is good |
| 165 | sth->messages.array[rpt->thread.send_to_child.msg_slot].opcode |= msg.opcode; |
| 166 | if(msg.reason) |
| 167 | sth->messages.array[rpt->thread.send_to_child.msg_slot].reason = msg.reason; |
| 168 | } |
| 169 | } |
| 170 | spinlock_unlock(&sth->messages.spinlock); |
| 171 | |
| 172 | // signal the streaming thread to wake up and process messages |
| 173 | if(send_pipe_msg) |
| 174 | stream_thread_send_pipe_signal(sth); |
| 175 | } |
| 176 | |
| 177 | void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg) { |
| 178 | if (!msg.session || !msg.meta || !s) |
| 179 | return; |
| 180 | |
| 181 | if(msg.meta != &s->thread.meta) { |
| 182 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 183 | "STREAM SND '%s' [to %s]: the opcode message does not match this sender. " |
| 184 | "Ignoring opcode.", rrdhost_hostname(s->host), s->remote_ip); |
| 185 | return; |
| 186 | } |
| 187 | |
| 188 | struct stream_thread *sth = stream_thread_by_slot_id(msg.thread_slot); |
| 189 | if(!sth) { |
| 190 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 191 | "STREAM SND[x] '%s' [to %s] the opcode (%u) message cannot be verified. Ignoring it.", |
| 192 | rrdhost_hostname(s->host), s->remote_ip, msg.opcode); |
| 193 | return; |
| 194 | } |
| 195 | |
| 196 | // check if we can execute the message now |
| 197 | if(sth->tid == gettid_cached() && msg.opcode == STREAM_OPCODE_SENDER_POLLOUT) { |
| 198 | // we are running at the stream thread, and the request is about enabling POLLOUT, |
| 199 | // we can do this synchronously. |
| 200 | // IMPORTANT: DO NOT HANDLE FAILURES THAT REMOVE THE RECEIVER OR THE SENDER THIS WAY |
| 201 | // THE EVENT LOOP DRAINS THE INPUT SOCKET (BOTH RECEIVER AND SENDER) |
| 202 | // AND THE LOOP WILL CRASH IF THE RECEIVER OR THE SENDER VANISH WHILE IT |
| 203 | // WORKS WITH THEM! |
| 204 | sth->messages.bypassed++; |
| 205 | stream_thread_handle_op(sth, &msg); |
| 206 | return; |
| 207 | } |
| 208 | |
| 209 | bool send_pipe_msg = false; |
| 210 | |
| 211 | // add it to the message queue of the thread |
| 212 | spinlock_lock(&sth->messages.spinlock); |
| 213 | { |
| 214 | sth->messages.added++; |
| 215 | if (s->thread.msg_slot >= sth->messages.used || sth->messages.array[s->thread.msg_slot].meta != &s->thread.meta) { |
| 216 | if (unlikely(sth->messages.used >= sth->messages.size)) { |
| 217 | // this should never happen, but let's find the root cause |
| 218 | |
| 219 | if (!sth->messages.size) { |
| 220 | // we are exiting |
| 221 | spinlock_unlock(&sth->messages.spinlock); |
| 222 | return; |
| 223 | } |
| 224 | |
| 225 | #ifdef NETDATA_INTERNAL_CHECKS |
| 226 | // try to find us in the list |
| 227 | for (size_t i = 0; i < sth->messages.size; i++) { |
| 228 | if (sth->messages.array[i].meta == &s->thread.meta) { |
| 229 | s->thread.msg_slot = i; |
| 230 | sth->messages.array[s->thread.msg_slot].opcode |= msg.opcode; |
| 231 | if(msg.reason) |
| 232 | sth->messages.array[s->thread.msg_slot].reason = msg.reason; |
| 233 | spinlock_unlock(&sth->messages.spinlock); |
| 234 | internal_fatal(true, "the dispatcher message queue is full, but this sender is already on slot %zu", i); |
| 235 | return; |
| 236 | } |
| 237 | } |
| 238 | #endif |
| 239 | |
| 240 | fatal("STREAM SND '%s' [to %s]: The streaming opcode queue is full, but this should never happen...", |
| 241 | rrdhost_hostname(s->host), s->remote_ip); |
| 242 | } |
| 243 | |
| 244 | // let's use a new slot |
| 245 | send_pipe_msg = !sth->messages.used; // write to the pipe, only when the queue was empty before this msg |
| 246 | s->thread.msg_slot = sth->messages.used++; |
| 247 | sth->messages.array[s->thread.msg_slot] = msg; |
| 248 | } |
| 249 | else { |
| 250 | // the existing slot is good |
| 251 | sth->messages.array[s->thread.msg_slot].opcode |= msg.opcode; |
| 252 | if(msg.reason) |
| 253 | sth->messages.array[s->thread.msg_slot].reason = msg.reason; |
| 254 | } |
| 255 | } |
| 256 | spinlock_unlock(&sth->messages.spinlock); |
| 257 | |
| 258 | // signal the streaming thread to wake up and process messages |
| 259 | if(send_pipe_msg) |
| 260 | stream_thread_send_pipe_signal(sth); |
| 261 | } |
| 262 | |
| 263 | bool stream_thread_process_opcodes(struct stream_thread *sth, struct pollfd_meta *my_meta) { |
| 264 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 265 | |
| 266 | size_t used = 0; |
| 267 | spinlock_lock(&sth->messages.spinlock); |
| 268 | if(sth->messages.used) { |
| 269 | used = sth->messages.used; |
| 270 | memcpy(sth->messages.copy, sth->messages.array, used * sizeof(*sth->messages.copy)); |
| 271 | sth->messages.used = 0; |
| 272 | } |
| 273 | spinlock_unlock(&sth->messages.spinlock); |
| 274 | |
| 275 | bool rc = false; |
| 276 | for(size_t i = 0; i < used ;i++) { |
| 277 | struct stream_opcode *msg = &sth->messages.copy[i]; |
| 278 | if(msg->meta == my_meta) rc = true; |
| 279 | stream_thread_handle_op(sth, msg); |
| 280 | } |
| 281 | |
| 282 | return rc; |
| 283 | } |
| 284 | |
| 285 | static void stream_thread_read_pipe_messages(struct stream_thread *sth) { |
| 286 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 287 | |
| 288 | if(read(sth->pipe.fds[PIPE_READ], sth->pipe.buffer, sth->pipe.size * sizeof(*sth->pipe.buffer)) <= 0) |
| 289 | nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: signal pipe read error", sth->id); |
| 290 | |
| 291 | stream_thread_process_opcodes(sth, NULL); |
| 292 | } |
| 293 | |
| 294 | // -------------------------------------------------------------------------------------------------------------------- |
| 295 | |
| 296 | static int set_pipe_size(int pipe_fd __maybe_unused, int new_size) { |
| 297 | int default_size = new_size; (void)default_size; |
| 298 | int result = new_size; |
| 299 | |
| 300 | #ifdef F_GETPIPE_SZ |
| 301 | // get the current size of the pipe |
| 302 | result = fcntl(pipe_fd, F_GETPIPE_SZ); |
| 303 | if(result > 0) |
| 304 | default_size = result; |
| 305 | #endif |
| 306 | |
| 307 | #ifdef F_SETPIPE_SZ |
| 308 | // set the new size to the pipe |
| 309 | if(result <= new_size) { |
| 310 | result = fcntl(pipe_fd, F_SETPIPE_SZ, new_size); |
| 311 | if (result <= 0) |
| 312 | return default_size; |
| 313 | } |
| 314 | #endif |
| 315 | |
| 316 | // we return either: |
| 317 | // 1. the new_size (after setting it) |
| 318 | // 2. the current size (if we can't set it, but we can read it) |
| 319 | // 3. the new_size (without setting it when we can't read the current size) |
| 320 | return result; // Returns the new pipe size |
| 321 | } |
| 322 | |
| 323 | // -------------------------------------------------------------------------------------------------------------------- |
| 324 | |
| 325 | static void stream_thread_messages_resize(struct stream_thread *sth) { |
| 326 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 327 | |
| 328 | if(sth->nodes_count * 2 >= sth->messages.size) { |
| 329 | spinlock_lock(&sth->messages.spinlock); |
| 330 | size_t new_size = MAX(sth->messages.size * 2, sth->nodes_count * 2); |
| 331 | sth->messages.array = reallocz(sth->messages.array, new_size * sizeof(*sth->messages.array)); |
| 332 | sth->messages.copy = reallocz(sth->messages.copy, new_size * sizeof(*sth->messages.copy)); |
| 333 | sth->messages.size = new_size; |
| 334 | spinlock_unlock(&sth->messages.spinlock); |
| 335 | } |
| 336 | } |
| 337 | |
| 338 | // -------------------------------------------------------------------------------------------------------------------- |
| 339 | |
| 340 | ALWAYS_INLINE_HOT_FLATTEN |
| 341 | static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_result_t *ev, usec_t now_ut, size_t *replay_entries) { |
| 342 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 343 | |
| 344 | struct pollfd_meta *m = (struct pollfd_meta *)ev->data; |
| 345 | if(!m) { |
| 346 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 347 | "STREAM THREAD[%zu]: cannot get meta from nd_poll() event. Ignoring event.", sth->id); |
| 348 | return false; |
| 349 | } |
| 350 | |
| 351 | switch(m->type) { |
| 352 | case POLLFD_TYPE_SENDER: { |
| 353 | struct sender_state *s = m->s; |
| 354 | if(stream_sender_process_poll_events(sth, s, ev->events, now_ut)) { |
| 355 | // the sender is still there |
| 356 | *replay_entries += dictionary_entries(s->replication.requests); |
| 357 | } |
| 358 | break; |
| 359 | } |
| 360 | |
| 361 | case POLLFD_TYPE_RECEIVER: { |
| 362 | struct receiver_state *rpt = m->rpt; |
| 363 | if(stream_receive_process_poll_events(sth, rpt, ev->events, now_ut)) { |
| 364 | // the receiver is still there |
| 365 | ; |
| 366 | } |
| 367 | break; |
| 368 | } |
| 369 | |
| 370 | case POLLFD_TYPE_PIPE: |
| 371 | if (likely(ev->events & ND_POLL_READ)) { |
| 372 | worker_is_busy(WORKER_SENDER_JOB_PIPE_READ); |
| 373 | stream_thread_read_pipe_messages(sth); |
| 374 | } |
| 375 | else if(unlikely(ev->events & ND_POLL_ERROR)) { |
| 376 | // we have errors on this pipe |
| 377 | nd_log(NDLS_DAEMON, NDLP_ERR, |
| 378 | "STREAM THREAD[%zu]: got errors on pipe - exiting to be restarted.", sth->id); |
| 379 | return true; |
| 380 | } |
| 381 | break; |
| 382 | |
| 383 | case POLLFD_TYPE_EMPTY: |
| 384 | // should never happen - but make sure it never happens again |
| 385 | internal_fatal(true, "What is this?"); |
| 386 | break; |
| 387 | } |
| 388 | |
| 389 | return false; |
| 390 | } |
| 391 | |
| 392 | void stream_thread(void *ptr) { |
| 393 | struct stream_thread *sth = ptr; |
| 394 | |
| 395 | nd_thread_can_run_sql(false); |
| 396 | |
| 397 | worker_register("STREAM"); |
| 398 | |
| 399 | // stream thread main event loop |
| 400 | worker_register_job_name(WORKER_STREAM_JOB_LIST, "list"); |
| 401 | worker_register_job_name(WORKER_STREAM_JOB_DEQUEUE, "dequeue"); |
| 402 | worker_register_job_name(WORKER_STREAM_JOB_PREP, "prep"); |
| 403 | worker_register_job_name(WORKER_STREAM_JOB_POLL_ERROR, "poll error"); |
| 404 | worker_register_job_name(WORKER_SENDER_JOB_PIPE_READ, "pipe read"); |
| 405 | |
| 406 | // both sender and receiver |
| 407 | worker_register_job_name(WORKER_STREAM_JOB_SOCKET_RECEIVE, "receive"); |
| 408 | worker_register_job_name(WORKER_STREAM_JOB_SOCKET_SEND, "send"); |
| 409 | |
| 410 | // receiver |
| 411 | worker_register_job_name(WORKER_STREAM_JOB_COMPRESS, "compress"); |
| 412 | worker_register_job_name(WORKER_STREAM_JOB_DECOMPRESS, "decompress"); |
| 413 | |
| 414 | // sender |
| 415 | worker_register_job_name(WORKER_SENDER_JOB_EXECUTE, "execute"); |
| 416 | worker_register_job_name(WORKER_SENDER_JOB_EXECUTE_REPLAY, "replay"); |
| 417 | worker_register_job_name(WORKER_SENDER_JOB_EXECUTE_FUNCTION, "function"); |
| 418 | worker_register_job_name(WORKER_SENDER_JOB_EXECUTE_META, "meta"); |
| 419 | |
| 420 | // disconnection reasons |
| 421 | worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_OVERFLOW, "disconnect overflow"); |
| 422 | worker_register_job_name(WORKER_STREAM_JOB_DISCONNECT_TIMEOUT, "disconnect timeout"); |
| 423 | worker_register_job_name(WORKER_STREAM_JOB_DISCONNECT_SOCKET_ERROR, "disconnect socket error"); |
| 424 | worker_register_job_name(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED, "disconnect remote closed"); |
| 425 | worker_register_job_name(WORKER_STREAM_JOB_DISCONNECT_RECEIVE_ERROR, "disconnect receive error"); |
| 426 | worker_register_job_name(WORKER_STREAM_JOB_DISCONNECT_SEND_ERROR, "disconnect send error"); |
| 427 | worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR, "disconnect compression error"); |
| 428 | worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_RECEIVER_LEFT, "disconnect receiver left"); |
| 429 | worker_register_job_name(WORKER_SENDER_JOB_DISCONNECT_HOST_CLEANUP, "disconnect host cleanup"); |
| 430 | |
| 431 | // metrics |
| 432 | worker_register_job_custom_metric(WORKER_STREAM_METRIC_NODES, |
| 433 | "nodes", "nodes", |
| 434 | WORKER_METRIC_ABSOLUTE); |
| 435 | |
| 436 | worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_READ, |
| 437 | "receiver received bytes", "bytes/s", |
| 438 | WORKER_METRIC_INCREMENT); |
| 439 | |
| 440 | worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, |
| 441 | "receiver received uncompressed bytes", "bytes/s", |
| 442 | WORKER_METRIC_INCREMENT); |
| 443 | |
| 444 | worker_register_job_custom_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, |
| 445 | "receiver replication completion", "%", |
| 446 | WORKER_METRIC_ABSOLUTE); |
| 447 | |
| 448 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BUFFER_RATIO, |
| 449 | "sender used buffer ratio", "%", |
| 450 | WORKER_METRIC_ABSOLUTE); |
| 451 | |
| 452 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, |
| 453 | "sender bytes received", "bytes/s", |
| 454 | WORKER_METRIC_INCREMENT); |
| 455 | |
| 456 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_SENT, |
| 457 | "sender bytes sent", "bytes/s", |
| 458 | WORKER_METRIC_INCREMENT); |
| 459 | |
| 460 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_COMPRESSED, |
| 461 | "sender bytes compressed", "bytes/s", |
| 462 | WORKER_METRIC_INCREMENTAL_TOTAL); |
| 463 | |
| 464 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_UNCOMPRESSED, |
| 465 | "sender bytes uncompressed", "bytes/s", |
| 466 | WORKER_METRIC_INCREMENTAL_TOTAL); |
| 467 | |
| 468 | worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO, |
| 469 | "sender cumulative compression savings ratio", "%", |
| 470 | WORKER_METRIC_ABSOLUTE); |
| 471 | |
| 472 | worker_register_job_custom_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, |
| 473 | "sender replication dict entries", "entries", |
| 474 | WORKER_METRIC_ABSOLUTE); |
| 475 | |
| 476 | worker_register_job_custom_metric(WORKER_SENDER_JOB_MESSAGES, |
| 477 | "ops processed", "messages", |
| 478 | WORKER_METRIC_INCREMENTAL_TOTAL); |
| 479 | |
| 480 | worker_register_job_custom_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE, |
| 481 | "receivers waiting to be added", "nodes", |
| 482 | WORKER_METRIC_ABSOLUTE); |
| 483 | |
| 484 | worker_register_job_custom_metric(WORKER_STREAM_JOB_SEND_MISSES, |
| 485 | "send misses", "misses", |
| 486 | WORKER_METRIC_INCREMENTAL_TOTAL); |
| 487 | |
| 488 | |
| 489 | if(pipe(sth->pipe.fds) != 0) { |
| 490 | nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: cannot create required pipe.", sth->id); |
| 491 | sth->pipe.fds[PIPE_READ] = -1; |
| 492 | sth->pipe.fds[PIPE_WRITE] = -1; |
| 493 | return; |
| 494 | } |
| 495 | |
| 496 | sth->tid = gettid_cached(); |
| 497 | |
| 498 | sth->pipe.size = set_pipe_size(sth->pipe.fds[PIPE_READ], 65536 * sizeof(*sth->pipe.buffer)) / sizeof(*sth->pipe.buffer); |
| 499 | sth->pipe.buffer = mallocz(sth->pipe.size * sizeof(*sth->pipe.buffer)); |
| 500 | |
| 501 | usec_t last_check_replication_ut, last_check_all_nodes_ut, last_dequeue_ut; |
| 502 | last_check_replication_ut = last_check_all_nodes_ut = last_dequeue_ut = now_monotonic_usec(); |
| 503 | |
| 504 | sth->run.pipe = (struct pollfd_meta){ |
| 505 | .type = POLLFD_TYPE_PIPE, |
| 506 | }; |
| 507 | sth->run.ndpl = nd_poll_create(); |
| 508 | if(!sth->run.ndpl) |
| 509 | fatal("Cannot create nd_poll()"); |
| 510 | |
| 511 | META_SET(&sth->run.meta, (Word_t)&sth->run.pipe, &sth->run.pipe); |
| 512 | |
| 513 | if(!nd_poll_add(sth->run.ndpl, sth->pipe.fds[PIPE_READ], ND_POLL_READ, &sth->run.pipe)) |
| 514 | nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: failed to add pipe to nd_poll()", sth->id); |
| 515 | |
| 516 | bool exit_thread = false; |
| 517 | size_t replay_entries = 0; |
| 518 | size_t receivers_waiting = 0; |
| 519 | sth->snd.bytes_received = 0; |
| 520 | sth->snd.bytes_sent = 0; |
| 521 | |
| 522 | rrd_collector_started(); |
| 523 | |
| 524 | usec_t now_ut = now_monotonic_usec(); |
| 525 | while(!exit_thread && !nd_thread_signaled_to_cancel() && service_running(SERVICE_STREAMING)) { |
| 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); |
| 532 | |
| 533 | // move any pending hosts in the inbound queue, to the running list |
| 534 | spinlock_lock(&sth->queue.spinlock); |
| 535 | |
| 536 | stream_thread_process_waiting_list_unsafe(sth, now_ut); |
| 537 | // stream_receiver_move_entire_queue_to_running_unsafe(sth); |
| 538 | |
| 539 | stream_sender_move_queue_to_running_unsafe(sth); |
| 540 | |
| 541 | receivers_waiting = sth->queue.receivers_waiting; |
| 542 | spinlock_unlock(&sth->queue.spinlock); |
| 543 | |
| 544 | // process any opcodes waiting |
| 545 | stream_thread_process_opcodes(sth, NULL); |
| 546 | |
| 547 | if(now_ut - last_check_all_nodes_ut >= nd_profile.update_every * USEC_PER_SEC) { |
| 548 | last_check_all_nodes_ut = now_ut; |
| 549 | |
| 550 | worker_is_busy(WORKER_STREAM_JOB_LIST); |
| 551 | |
| 552 | // periodically check the entire list of nodes |
| 553 | // this detects unresponsive parents too (timeout) |
| 554 | stream_sender_check_all_nodes_from_poll(sth, now_ut); |
| 555 | stream_receiver_check_all_nodes_from_poll(sth, now_ut); |
| 556 | |
| 557 | worker_set_metric(WORKER_SENDER_JOB_MESSAGES, (NETDATA_DOUBLE)(sth->messages.processed)); |
| 558 | worker_set_metric(WORKER_STREAM_METRIC_NODES, (NETDATA_DOUBLE)sth->nodes_count); |
| 559 | |
| 560 | worker_set_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, (NETDATA_DOUBLE)sth->snd.bytes_received); |
| 561 | worker_set_metric(WORKER_SENDER_JOB_BYTES_SENT, (NETDATA_DOUBLE)sth->snd.bytes_sent); |
| 562 | worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE)replay_entries); |
| 563 | |
| 564 | worker_set_metric(WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting); |
| 565 | worker_set_metric(WORKER_STREAM_JOB_SEND_MISSES, (NETDATA_DOUBLE)sth->snd.send_misses); |
| 566 | replay_entries = 0; |
| 567 | sth->snd.bytes_received = 0; |
| 568 | sth->snd.bytes_sent = 0; |
| 569 | |
| 570 | if(now_ut - last_check_replication_ut >= 10 * 60 * USEC_PER_SEC) { |
| 571 | last_check_replication_ut = now_ut; |
| 572 | |
| 573 | worker_is_busy(WORKER_STREAM_JOB_LIST); |
| 574 | |
| 575 | stream_sender_replication_check_from_poll(sth, now_ut); |
| 576 | stream_receiver_replication_check_from_poll(sth, now_ut); |
| 577 | } |
| 578 | } |
| 579 | } |
| 580 | |
| 581 | worker_is_idle(); |
| 582 | |
| 583 | nd_poll_result_t ev; |
| 584 | int poll_rc = nd_poll_wait(sth->run.ndpl, 100, &ev); |
| 585 | |
| 586 | worker_is_busy(WORKER_STREAM_JOB_PREP); |
| 587 | |
| 588 | if (unlikely(poll_rc == 0)) { |
| 589 | // nd_poll() timed out - just loop again |
| 590 | now_ut = now_monotonic_usec(); |
| 591 | continue; |
| 592 | } |
| 593 | |
| 594 | if(unlikely(poll_rc == -1)) { |
| 595 | // nd_poll() returned an error |
| 596 | internal_fatal(true, "nd_poll() failed"); |
| 597 | worker_is_busy(WORKER_STREAM_JOB_POLL_ERROR); |
| 598 | nd_log_limit_static_thread_var(erl, 1, 1 * USEC_PER_MS); |
| 599 | nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu] nd_poll() returned error", sth->id); |
| 600 | now_ut = now_monotonic_usec(); |
| 601 | continue; |
| 602 | } |
| 603 | |
| 604 | if(unlikely(nd_thread_signaled_to_cancel() || !service_running(SERVICE_STREAMING))) |
| 605 | break; |
| 606 | |
| 607 | // nd_poll() may have received events for a socket we have already removed |
| 608 | // so, if we don't find it in our meta index, do not access it - it has been removed |
| 609 | if(unlikely(META_GET(&sth->run.meta, (Word_t)ev.data) != ev.data)) { |
| 610 | now_ut = now_monotonic_usec(); |
| 611 | continue; |
| 612 | } |
| 613 | |
| 614 | now_ut = now_monotonic_usec(); |
| 615 | exit_thread = stream_thread_process_poll_slot(sth, &ev, now_ut, &replay_entries); |
| 616 | now_ut = now_monotonic_usec(); |
| 617 | } |
| 618 | |
| 619 | // dequeue |
| 620 | spinlock_lock(&sth->queue.spinlock); |
| 621 | stream_sender_move_queue_to_running_unsafe(sth); |
| 622 | stream_receiver_move_entire_queue_to_running_unsafe(sth); |
| 623 | spinlock_unlock(&sth->queue.spinlock); |
| 624 | |
| 625 | // cleanup receiver and dispatcher |
| 626 | stream_sender_cleanup(sth); |
| 627 | stream_receiver_cleanup(sth); |
| 628 | META_FREE(&sth->run.meta, NULL, NULL); |
| 629 | |
| 630 | // cleanup the thread structures |
| 631 | spinlock_lock(&sth->messages.spinlock); |
| 632 | freez(sth->messages.array); |
| 633 | sth->messages.array = NULL; |
| 634 | sth->messages.size = 0; |
| 635 | sth->messages.used = 0; |
| 636 | spinlock_unlock(&sth->messages.spinlock); |
| 637 | |
| 638 | freez(sth->pipe.buffer); |
| 639 | sth->pipe.buffer = NULL; |
| 640 | sth->pipe.size = 0; |
| 641 | |
| 642 | nd_poll_destroy(sth->run.ndpl); |
| 643 | sth->run.ndpl = NULL; |
| 644 | |
| 645 | close(sth->pipe.fds[PIPE_READ]); |
| 646 | close(sth->pipe.fds[PIPE_WRITE]); |
| 647 | sth->pipe.fds[PIPE_READ] = -1; |
| 648 | sth->pipe.fds[PIPE_WRITE] = -1; |
| 649 | |
| 650 | sth->thread = NULL; |
| 651 | sth->tid = 0; |
| 652 | |
| 653 | worker_unregister(); |
| 654 | |
| 655 | rrd_collector_finished(); |
| 656 | } |
| 657 | |
| 658 | // -------------------------------------------------------------------------------------------------------------------- |
| 659 | |
| 660 | void stream_thread_node_queued(RRDHOST *host) { |
| 661 | spinlock_lock(&stream_thread_globals.assign.spinlock); |
| 662 | host->stream.refcount++; |
| 663 | internal_fatal(host->stream.refcount > 2, "invalid stream refcount %u (while adding node)", host->stream.refcount); |
| 664 | spinlock_unlock(&stream_thread_globals.assign.spinlock); |
| 665 | } |
| 666 | |
| 667 | void stream_thread_node_removed(RRDHOST *host) { |
| 668 | spinlock_lock(&stream_thread_globals.assign.spinlock); |
| 669 | internal_fatal(!host->stream.refcount, "invalid stream refcount %u (while stopping node)", host->stream.refcount); |
| 670 | |
| 671 | if(--host->stream.refcount == 0) { |
| 672 | struct stream_thread *sth = host->stream.thread; |
| 673 | sth->nodes_count--; |
| 674 | host->stream.thread = NULL; |
| 675 | } |
| 676 | |
| 677 | spinlock_unlock(&stream_thread_globals.assign.spinlock); |
| 678 | } |
| 679 | |
| 680 | static struct stream_thread *stream_thread_get_unsafe(RRDHOST *host) { |
| 681 | if(host->stream.thread) |
| 682 | return host->stream.thread; |
| 683 | |
| 684 | if(!stream_thread_globals.assign.cores) { |
| 685 | stream_thread_globals.assign.cores = netdata_conf_cpus() - 1; |
| 686 | if(stream_thread_globals.assign.cores < 4) |
| 687 | stream_thread_globals.assign.cores = 4; |
| 688 | else if(stream_thread_globals.assign.cores > STREAM_MAX_THREADS) |
| 689 | stream_thread_globals.assign.cores = STREAM_MAX_THREADS; |
| 690 | } |
| 691 | |
| 692 | size_t selected_thread_slot = 0; |
| 693 | size_t min_nodes = stream_thread_globals.threads[0].nodes_count; |
| 694 | for(size_t i = 1; i < stream_thread_globals.assign.cores ; i++) { |
| 695 | if(stream_thread_globals.threads[i].nodes_count < min_nodes) { |
| 696 | selected_thread_slot = i; |
| 697 | min_nodes = stream_thread_globals.threads[i].nodes_count; |
| 698 | } |
| 699 | } |
| 700 | |
| 701 | struct stream_thread *sth = host->stream.thread = &stream_thread_globals.threads[selected_thread_slot]; |
| 702 | host->stream.refcount = 0; |
| 703 | sth->nodes_count++; |
| 704 | |
| 705 | return host->stream.thread; |
| 706 | } |
| 707 | |
| 708 | static struct stream_thread * stream_thread_assign_and_start(RRDHOST *host) { |
| 709 | spinlock_lock(&stream_thread_globals.assign.spinlock); |
| 710 | |
| 711 | struct stream_thread *sth = stream_thread_get_unsafe(host); |
| 712 | |
| 713 | if(!sth->thread) { |
| 714 | sth->id = (sth - stream_thread_globals.threads); // find the slot number |
| 715 | if(&stream_thread_globals.threads[sth->id] != sth) |
| 716 | fatal("STREAM THREAD[x] [%s]: thread and slot owner do not match!", |
| 717 | rrdhost_hostname(host)); |
| 718 | |
| 719 | sth->pipe.fds[PIPE_READ] = -1; |
| 720 | sth->pipe.fds[PIPE_WRITE] = -1; |
| 721 | spinlock_init(&sth->pipe.spinlock); |
| 722 | spinlock_init(&sth->queue.spinlock); |
| 723 | spinlock_init(&sth->messages.spinlock); |
| 724 | sth->messages.used = 0; |
| 725 | |
| 726 | char tag[NETDATA_THREAD_TAG_MAX + 1]; |
| 727 | snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM "[%zu]", sth->id); |
| 728 | |
| 729 | sth->thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT, stream_thread, sth); |
| 730 | if (!sth->thread) |
| 731 | nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: failed to create new thread for client.", sth->id); |
| 732 | } |
| 733 | |
| 734 | spinlock_unlock(&stream_thread_globals.assign.spinlock); |
| 735 | |
| 736 | return sth; |
| 737 | } |
| 738 | |
| 739 | void stream_sender_add_to_connector_queue(RRDHOST *host) { |
| 740 | ND_LOG_STACK lgs[] = { |
| 741 | ND_LOG_FIELD_STR(NDF_NIDL_NODE, host->hostname), |
| 742 | ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid), |
| 743 | ND_LOG_FIELD_END(), |
| 744 | }; |
| 745 | ND_LOG_STACK_PUSH(lgs); |
| 746 | |
| 747 | stream_connector_init(host->sender); |
| 748 | rrdhost_stream_parent_ssl_init(host->sender); |
| 749 | stream_connector_add(host->sender); |
| 750 | } |
| 751 | |
| 752 | void stream_receiver_add_to_queue(struct receiver_state *rpt) { |
| 753 | struct stream_thread *sth = stream_thread_assign_and_start(rpt->host); |
| 754 | |
| 755 | stream_thread_node_queued(rpt->host); |
| 756 | |
| 757 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 758 | "STREAM RCV[%zu] '%s': moving host to receiver queue...", |
| 759 | sth->id, rrdhost_hostname(rpt->host)); |
| 760 | |
| 761 | spinlock_lock(&sth->queue.spinlock); |
| 762 | RECEIVERS_SET(&sth->queue.receivers, ++sth->queue.id, rpt); |
| 763 | sth->queue.receivers_waiting++; |
| 764 | spinlock_unlock(&sth->queue.spinlock); |
| 765 | |
| 766 | pulse_host_status(rpt->host, PULSE_HOST_STATUS_RCV_WAITING, 0); |
| 767 | } |
| 768 | |
| 769 | void stream_sender_add_to_queue(struct sender_state *s) { |
| 770 | struct stream_thread *sth = stream_thread_assign_and_start(s->host); |
| 771 | |
| 772 | stream_thread_node_queued(s->host); |
| 773 | |
| 774 | nd_log(NDLS_DAEMON, NDLP_DEBUG, |
| 775 | "STREAM THREAD[%zu] '%s': moving host to sender queue...", |
| 776 | sth->id, rrdhost_hostname(s->host)); |
| 777 | |
| 778 | spinlock_lock(&sth->queue.spinlock); |
| 779 | SENDERS_SET(&sth->queue.senders, ++sth->queue.id, s); |
| 780 | spinlock_unlock(&sth->queue.spinlock); |
| 781 | |
| 782 | pulse_host_status(s->host, PULSE_HOST_STATUS_SND_WAITING, 0); |
| 783 | } |
| 784 | |
| 785 | void stream_threads_cancel(void) { |
| 786 | stream_connector_cancel_threads(); |
| 787 | for(size_t i = 0; i < STREAM_MAX_THREADS ;i++) |
| 788 | nd_thread_signal_cancel(stream_thread_globals.threads[i].thread); |
| 789 | } |
| 790 | |
| 791 | struct stream_thread *stream_thread_by_slot_id(size_t thread_slot) { |
| 792 | if(thread_slot < STREAM_MAX_THREADS && stream_thread_globals.threads[thread_slot].thread) |
| 793 | return &stream_thread_globals.threads[thread_slot]; |
| 794 | |
| 795 | return NULL; |
| 796 | } |