fixed bug in streaming sender read (#19136)
Costa Tsaousis committed
Dec 6, 2024 at 02:53 UTC
cb731f303fd469934ce3f49fc34f5317b120f4a8
6 files changed
+60
-51
src/database/engine/rrdengineapi.c
+1
-1
@@ -455,7 +455,7 @@ static PGD *rrdeng_alloc_new_page_data(struct rrdeng_collect_handle *handle, use
455
if(slots < 3)
456
slots = 3;
457
458
- size_t size = slots * CTX_POINT_SIZE_BYTES(ctx);
458
+ size_t size = slots * CTX_POINT_SIZE_BYTES(ctx); (void)size;
459
460
// internal_error(true, "PAGE ALLOC %zu bytes (%zu max)", size, max_size);
461
src/streaming/stream-receiver.c
+2
-2
@@ -417,7 +417,7 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st
417
}
418
419
// process poll() events for streaming receivers
420
-void stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events __maybe_unused, time_t now_s) {
420
+void stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events __maybe_unused, usec_t now_ut) {
421
PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);
422
ND_LOG_STACK lgs[] = {
423
ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->client_ip),
@@ -439,7 +439,7 @@ void stream_receive_process_poll_events(struct stream_thread *sth, struct receiv
439
return;
440
}
441
442
- rpt->last_msg_t = now_s;
442
+ rpt->last_msg_t = (time_t)(now_ut / USEC_PER_SEC);
443
444
if(rpt->thread.compressed.enabled) {
445
worker_is_busy(WORKER_STREAM_JOB_SOCKET_RECEIVE);
src/streaming/stream-sender-internals.h
+4
-2
@@ -55,7 +55,8 @@ struct sender_state {
55
size_t bytes_uncompressed;
56
57
// the current buffer statistics
58
- // these SHOULD ALWAYS BE CALCULATED ON EVERY sender_unlock() IF THE BUFFER WAS MODIFIED
58
+ // these SHOULD ALWAYS BE CALCULATED ON EVERY stream_sender_unlock() IF THE BUFFER WAS MODIFIED
59
+ // stream_sender_lock() IS REQUIRED TO READ/WRITE THESE
60
size_t bytes_outstanding;
61
size_t bytes_available;
62
NETDATA_DOUBLE buffer_ratio;
@@ -65,6 +66,8 @@ struct sender_state {
66
size_t bytes_sent;
67
size_t bytes_sent_by_type[STREAM_TRAFFIC_TYPE_MAX];
68
69
+ usec_t last_traffic_ut;
70
+
71
struct pollfd_meta meta;
72
} thread;
73
@@ -73,7 +76,6 @@ struct sender_state {
76
} connector;
77
78
char connected_to[CONNECTED_TO_SIZE + 1]; // We don't know which proxy we connect to, passed back from socket.c
76
- time_t last_traffic_seen_t;
79
time_t last_state_since_t; // the timestamp of the last state (online/offline) change
80
81
struct {
src/streaming/stream-sender.c
+44
-36
@@ -7,13 +7,13 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
7
8
// --------------------------------------------------------------------------------------------------------------------
9
10
-static void stream_sender_cbuffer_recreate_timed_unsafe(struct sender_state *s, time_t now_s, bool force) {
11
- static __thread time_t last_reset_time_s = 0;
10
+static void stream_sender_cbuffer_recreate_timed_unsafe(struct sender_state *s, usec_t now_ut, bool force) {
11
+ static __thread usec_t last_reset_time_ut = 0;
12
13
- if(!force && now_s - last_reset_time_s < 300)
13
+ if(!force && now_ut - last_reset_time_ut < 300 * USEC_PER_SEC)
14
return;
15
16
- last_reset_time_s = now_s;
16
+ last_reset_time_ut = now_ut;
17
18
s->sbuf.recreates++; // we increase even if we don't do it, to have sender_start() recreate its buffers
19
@@ -30,7 +30,7 @@ static void rrdpush_sender_cbuffer_flush(RRDHOST *host) {
30
31
// flush the output buffer from any data it may have
32
cbuffer_flush(host->sender->sbuf.cb);
33
- stream_sender_cbuffer_recreate_timed_unsafe(host->sender, now_monotonic_sec(), true);
33
+ stream_sender_cbuffer_recreate_timed_unsafe(host->sender, now_monotonic_usec(), true);
34
35
stream_sender_unlock(host->sender);
36
}
@@ -76,7 +76,7 @@ void stream_sender_on_connect(struct sender_state *s) {
76
rrdpush_sender_charts_and_replication_reset(s);
77
rrdpush_sender_cbuffer_flush(s->host);
78
79
- s->last_traffic_seen_t = now_monotonic_sec();
79
+ s->thread.last_traffic_ut = now_monotonic_usec();
80
s->rbuf.read_len = 0;
81
s->sbuf.cb->read = 0;
82
s->sbuf.cb->write = 0;
@@ -370,23 +370,27 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
370
stream_connector_requeue(s);
371
}
372
373
-void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth) {
373
+void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut) {
374
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
375
376
- usec_t now_ut = now_monotonic_usec();
377
- time_t now_s = (time_t)(now_ut / USEC_PER_SEC);
378
-
376
size_t bytes_uncompressed = 0;
377
size_t bytes_compressed = 0;
381
- NETDATA_DOUBLE buffer_ratio = 0.0;
378
+ NETDATA_DOUBLE overall_buffer_ratio = 0.0;
379
380
Word_t idx = 0;
381
for(struct sender_state *s = SENDERS_FIRST(&sth->snd.senders, &idx);
382
s;
383
s = SENDERS_NEXT(&sth->snd.senders, &idx)) {
384
388
- // If the TCP window never opened, then something is wrong, restart connection
389
- if(unlikely(now_s - s->last_traffic_seen_t > stream_send.parents.timeout_s &&
385
+ stream_sender_lock(s);
386
+ size_t outstanding = cbuffer_next_unsafe(s->sbuf.cb, NULL);
387
+ NETDATA_DOUBLE buffer_ratio = s->thread.buffer_ratio;
388
+ stream_sender_unlock(s);
389
+
390
+ if (buffer_ratio > overall_buffer_ratio)
391
+ overall_buffer_ratio = buffer_ratio;
392
+
393
+ if(unlikely(s->thread.last_traffic_ut + stream_send.parents.timeout_s * USEC_PER_SEC < now_ut &&
394
!stream_sender_pending_replication_requests(s) &&
395
!stream_sender_replicating_charts(s)
396
)) {
@@ -404,28 +408,30 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth) {
408
409
worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
410
411
+ char since[RFC3339_MAX_LENGTH];
412
+ rfc3339_datetime_ut(since, sizeof(since), s->thread.last_traffic_ut, 2, false);
413
+
414
+ char pending[64];
415
+ size_snprintf(pending, sizeof(pending), outstanding, "B", false);
416
+
417
nd_log(NDLS_DAEMON, NDLP_ERR,
408
- "STREAM SEND[%zu] %s [send to %s]: could not send metrics for %ld seconds - closing connection - "
409
- "we have sent %zu bytes on this connection via %zu send attempts.",
418
+ "STREAM SEND[%zu] %s [send to %s]: could not send data for %ld seconds - closing connection - "
419
+ "we have sent %zu bytes in %zu operations, it is idle since: %s, and we have %s pending to send "
420
+ "(buffer is used %.2f%%).",
421
sth->id, rrdhost_hostname(s->host), s->connected_to, stream_send.parents.timeout_s,
411
- s->thread.bytes_sent, s->thread.sends);
422
+ s->thread.bytes_sent, s->thread.sends, since, pending, buffer_ratio);
423
424
stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_SOCKET_TIMEOUT, true);
425
continue;
426
}
427
417
- stream_sender_lock(s);
418
- {
419
- bytes_compressed += s->thread.bytes_compressed;
420
- bytes_uncompressed += s->thread.bytes_uncompressed;
421
- uint64_t outstanding = s->thread.bytes_outstanding;
422
- if (s->thread.buffer_ratio > buffer_ratio)
423
- buffer_ratio = s->thread.buffer_ratio;
424
-
425
- if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, ND_POLL_READ | (outstanding ? ND_POLL_WRITE : 0), &s->thread.meta))
426
- internal_fatal(true, "Failed to update sender socket in nd_poll()");
427
- }
428
- stream_sender_unlock(s);
428
+ bytes_compressed += s->thread.bytes_compressed;
429
+ bytes_uncompressed += s->thread.bytes_uncompressed;
430
+
431
+ if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, ND_POLL_READ | (outstanding ? ND_POLL_WRITE : 0), &s->thread.meta))
432
+ nd_log(NDLS_DAEMON, NDLP_ERR,
433
+ "STREAM SEND[%zu] %s [send to %s]: failed to update nd_poll().",
434
+ sth->id, rrdhost_hostname(s->host), s->connected_to);
435
}
436
437
if (bytes_compressed && bytes_uncompressed) {
@@ -435,10 +441,10 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth) {
441
442
worker_set_metric(WORKER_SENDER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_uncompressed);
443
worker_set_metric(WORKER_SENDER_JOB_BYTES_COMPRESSED, (NETDATA_DOUBLE)bytes_compressed);
438
- worker_set_metric(WORKER_SENDER_JOB_BUFFER_RATIO, buffer_ratio);
444
+ worker_set_metric(WORKER_SENDER_JOB_BUFFER_RATIO, overall_buffer_ratio);
445
}
446
441
-void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, time_t now_s) {
447
+void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, usec_t now_ut) {
448
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
449
450
ND_LOG_STACK lgs[] = {
@@ -490,16 +496,18 @@ void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_
496
if (likely(bytes > 0)) {
497
cbuffer_remove_unsafe(s->sbuf.cb, bytes);
498
stream_sender_thread_data_sent_data_unsafe(s, bytes);
493
- s->last_traffic_seen_t = now_s;
499
+ s->thread.last_traffic_ut = now_ut;
500
sth->snd.bytes_sent += bytes;
501
502
if(!s->thread.bytes_outstanding) {
497
- // we sent them all - remove POLLOUT
503
+ // we sent them all - remove ND_POLL_WRITE
504
if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, ND_POLL_READ, &s->thread.meta))
499
- internal_fatal(true, "Failed to update sender socket in nd_poll()");
505
+ nd_log(NDLS_DAEMON, NDLP_ERR,
506
+ "STREAM SEND[%zu] %s [send to %s]: failed to update nd_poll().",
507
+ sth->id, rrdhost_hostname(s->host), s->connected_to);
508
509
// recreate the circular buffer if we have to
502
- stream_sender_cbuffer_recreate_timed_unsafe(s, now_s, false);
510
+ stream_sender_cbuffer_recreate_timed_unsafe(s, now_ut, false);
511
}
512
}
513
else if (bytes < 0 && errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR)
@@ -519,14 +527,14 @@ void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_
527
}
528
}
529
522
- if(events & POLLIN) {
530
+ if(events & ND_POLL_READ) {
531
// we can receive data from this socket
532
533
worker_is_busy(WORKER_STREAM_JOB_SOCKET_RECEIVE);
534
ssize_t bytes = nd_sock_revc_nowait(&s->sock, s->rbuf.b + s->rbuf.read_len, sizeof(s->rbuf.b) - s->rbuf.read_len - 1);
535
if (bytes > 0) {
536
s->rbuf.read_len += bytes;
529
- s->last_traffic_seen_t = now_s;
537
+ s->thread.last_traffic_ut = now_ut;
538
sth->snd.bytes_received += bytes;
539
}
540
else if (bytes == 0 || errno == ECONNRESET) {
src/streaming/stream-thread.c
+6
-7
@@ -177,7 +177,7 @@ static void stream_thread_messages_resize_unsafe(struct stream_thread *sth) {
177
178
// --------------------------------------------------------------------------------------------------------------------
179
180
-static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_result_t *ev, time_t now_s, size_t *replay_entries) {
180
+static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_result_t *ev, usec_t now_ut, size_t *replay_entries) {
181
struct pollfd_meta *m = ev->data;
182
internal_fatal(!m, "Failed to get meta from event");
183
@@ -185,7 +185,7 @@ static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_r
185
case POLLFD_TYPE_SENDER: {
186
struct sender_state *s = m->s;
187
internal_fatal(SENDERS_GET(&sth->snd.senders, (Word_t)s) == NULL, "Sender is not found in the senders list");
188
- stream_sender_process_poll_events(sth, s, ev->events, now_s);
188
+ stream_sender_process_poll_events(sth, s, ev->events, now_ut);
189
*replay_entries += dictionary_entries(s->replication.requests);
190
break;
191
}
@@ -193,7 +193,7 @@ static bool stream_thread_process_poll_slot(struct stream_thread *sth, nd_poll_r
193
case POLLFD_TYPE_RECEIVER: {
194
struct receiver_state *rpt = m->rpt;
195
internal_fatal(RECEIVERS_GET(&sth->rcv.receivers, (Word_t)rpt) == NULL, "Receiver is not found in the receiver list");
196
- stream_receive_process_poll_events(sth, rpt, ev->events, now_s);
196
+ stream_receive_process_poll_events(sth, rpt, ev->events, now_ut);
197
break;
198
}
199
@@ -356,7 +356,7 @@ void *stream_thread(void *ptr) {
356
357
// periodically check the entire list of nodes
358
// this detects unresponsive parents too (timeout)
359
- stream_sender_check_all_nodes_from_poll(sth);
359
+ stream_sender_check_all_nodes_from_poll(sth, now_ut);
360
worker_set_metric(WORKER_SENDER_JOB_MESSAGES, (NETDATA_DOUBLE)(sth->messages.processed));
361
worker_set_metric(WORKER_STREAM_METRIC_NODES, (NETDATA_DOUBLE)sth->nodes_count);
362
@@ -390,12 +390,11 @@ void *stream_thread(void *ptr) {
390
continue;
391
}
392
393
- time_t now_s = now_monotonic_sec();
394
-
393
if(nd_thread_signaled_to_cancel() || !service_running(SERVICE_STREAMING))
394
break;
395
398
- exit_thread = stream_thread_process_poll_slot(sth, &ev, now_s, &replay_entries);
396
+ now_ut = now_monotonic_usec();
397
+ exit_thread = stream_thread_process_poll_slot(sth, &ev, now_ut, &replay_entries);
398
}
399
400
// dequeue
src/streaming/stream-thread.h
+3
-3
@@ -173,13 +173,13 @@ extern struct stream_thread_globals stream_thread_globals;
173
174
void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth);
175
void stream_receiver_move_queue_to_running_unsafe(struct stream_thread *sth);
176
-void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth);
176
+void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut);
177
178
void stream_receiver_add_to_queue(struct receiver_state *rpt);
179
void stream_sender_add_to_connector_queue(struct rrdhost *host);
180
181
-void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, time_t now_s);
182
-void stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events, time_t now_s);
181
+void stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, usec_t now_ut);
182
+void stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events, usec_t now_ut);
183
184
void stream_sender_cleanup(struct stream_thread *sth);
185
void stream_receiver_cleanup(struct stream_thread *sth);