fix receiver deadlock (#18440)
Costa Tsaousis committed
Aug 29, 2024 at 21:04 UTC
1d1c3a8062ba1e508c433b5af07758951147c2a6
2 files changed
+45
-47
src/streaming/receiver.c
+39
-41
@@ -460,46 +460,50 @@ static bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
460
461
static void rrdhost_clear_receiver(struct receiver_state *rpt) {
462
RRDHOST *host = rpt->host;
463
- if(host) {
464
- bool signal_rrdcontext = false;
465
- spinlock_lock(&host->receiver_lock);
463
+ if(!host) return;
464
465
+ spinlock_lock(&host->receiver_lock);
466
+ {
467
// Make sure that we detach this thread and don't kill a freshly arriving receiver
468
- if(host->receiver == rpt) {
468
+
469
+ if (host->receiver == rpt) {
470
+ spinlock_unlock(&host->receiver_lock);
471
+ {
472
+ // run all these without having the receiver lock
473
+
474
+ stream_path_child_disconnected(host);
475
+ rrdpush_sender_thread_stop(host, STREAM_HANDSHAKE_DISCONNECT_RECEIVER_LEFT, false);
476
+ rrdpush_receiver_replication_reset(host);
477
+ rrdcontext_host_child_disconnected(host);
478
+
479
+ if (rpt->config.health_enabled)
480
+ rrdcalc_child_disconnected(host);
481
+
482
+ rrdpush_reset_destinations_postpone_time(host);
483
+ }
484
+ spinlock_lock(&host->receiver_lock);
485
+
486
+ // now we have the lock again
487
+
488
__atomic_sub_fetch(&localhost->connected_children_count, 1, __ATOMIC_RELAXED);
489
rrdhost_flag_set(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
490
472
- pluginsd_process_cleanup(rpt->parser);
473
- __atomic_store_n(&rpt->parser, NULL, __ATOMIC_RELAXED);
474
-
491
host->trigger_chart_obsoletion_check = 0;
492
host->child_connect_time = 0;
493
host->child_disconnected_time = now_realtime_sec();
478
-
494
host->health.health_enabled = 0;
495
481
- rrdpush_sender_thread_stop(host, STREAM_HANDSHAKE_DISCONNECT_RECEIVER_LEFT, false);
482
-
483
- signal_rrdcontext = true;
484
- rrdpush_receiver_replication_reset(host);
485
-
496
+ host->rrdpush_last_receiver_exit_reason = rpt->exit.reason;
497
rrdhost_flag_set(host, RRDHOST_FLAG_ORPHAN);
498
host->receiver = NULL;
488
- host->rrdpush_last_receiver_exit_reason = rpt->exit.reason;
489
-
490
- if(rpt->config.health_enabled)
491
- rrdcalc_child_disconnected(host);
492
-
493
- stream_path_child_disconnected(host);
499
}
500
+ }
501
496
- spinlock_unlock(&host->receiver_lock);
497
-
498
- if(signal_rrdcontext)
499
- rrdcontext_host_child_disconnected(host);
502
+ // this must be cleared with the receiver lock
503
+ pluginsd_process_cleanup(rpt->parser);
504
+ __atomic_store_n(&rpt->parser, NULL, __ATOMIC_RELAXED);
505
501
- rrdpush_reset_destinations_postpone_time(host);
502
- }
506
+ spinlock_unlock(&host->receiver_lock);
507
}
508
509
bool stop_streaming_receiver(RRDHOST *host, STREAM_HANDSHAKE reason) {
@@ -859,21 +863,6 @@ cleanup:
863
;
864
}
865
862
-static void rrdpush_receiver_thread_cleanup(void *pptr) {
863
- struct receiver_state *rpt = CLEANUP_FUNCTION_GET_PTR(pptr);
864
- if(!rpt) return;
865
-
866
- netdata_log_info("STREAM '%s' [receive from [%s]:%s]: "
867
- "receive thread ended (task id %d)"
868
- , rpt->hostname ? rpt->hostname : "-"
869
- , rpt->client_ip ? rpt->client_ip : "-", rpt->client_port ? rpt->client_port : "-", gettid_cached());
870
-
871
- worker_unregister();
872
- rrdhost_clear_receiver(rpt);
873
- receiver_state_free(rpt);
874
- rrdhost_set_is_parent_label();
875
-}
876
-
866
static bool stream_receiver_log_capabilities(BUFFER *wb, void *ptr) {
867
struct receiver_state *rpt = ptr;
868
if(!rpt)
@@ -893,7 +882,6 @@ static bool stream_receiver_log_transport(BUFFER *wb, void *ptr) {
882
}
883
884
void *rrdpush_receiver_thread(void *ptr) {
896
- CLEANUP_FUNCTION_REGISTER(rrdpush_receiver_thread_cleanup) cleanup_ptr = ptr;
885
worker_register("STREAMRCV");
886
887
worker_register_job_custom_metric(WORKER_RECEIVER_JOB_BYTES_READ,
@@ -925,5 +913,15 @@ void *rrdpush_receiver_thread(void *ptr) {
913
, rpt->client_port);
914
915
rrdpush_receive(rpt);
916
+
917
+ netdata_log_info("STREAM '%s' [receive from [%s]:%s]: "
918
+ "receive thread ended (task id %d)"
919
+ , rpt->hostname ? rpt->hostname : "-"
920
+ , rpt->client_ip ? rpt->client_ip : "-", rpt->client_port ? rpt->client_port : "-", gettid_cached());
921
+
922
+ worker_unregister();
923
+ rrdhost_clear_receiver(rpt);
924
+ receiver_state_free(rpt);
925
+ rrdhost_set_is_parent_label();
926
return NULL;
927
}
src/streaming/sender.c
+6
-6
@@ -731,9 +731,9 @@ void *rrdpush_sender_thread(void *ptr) {
731
732
netdata_log_info("STREAM %s [send]: sending thread exits %s",
733
rrdhost_hostname(s->host),
734
- s->host->sender->exit.reason != STREAM_HANDSHAKE_NEVER ? stream_handshake_error_to_string(s->host->sender->exit.reason) : "");
734
+ s->exit.reason != STREAM_HANDSHAKE_NEVER ? stream_handshake_error_to_string(s->exit.reason) : "");
735
736
- sender_lock(s->host->sender);
736
+ sender_lock(s);
737
{
738
rrdpush_sender_thread_close_socket(s);
739
rrdpush_sender_pipe_close(s->host, s->rrdpush_sender_pipe, false);
@@ -742,13 +742,13 @@ void *rrdpush_sender_thread(void *ptr) {
742
rrdhost_clear_sender___while_having_sender_mutex(s->host);
743
744
#ifdef NETDATA_LOG_STREAM_SENDER
745
- if (s->host->sender->stream_log_fp) {
746
- fclose(s->host->sender->stream_log_fp);
747
- s->host->sender->stream_log_fp = NULL;
745
+ if (s->stream_log_fp) {
746
+ fclose(s->stream_log_fp);
747
+ s->stream_log_fp = NULL;
748
}
749
#endif
750
}
751
- sender_unlock(s->host->sender);
751
+ sender_unlock(s);
752
753
freez(pipe_buffer);
754
freez(s);