diff options
author | Costa Tsaousis <costa@netdata.cloud> | 2023-01-14 00:09:02 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-01-14 00:09:02 +0200 |
commit | f10f19c8c8bba37e8364fa6138d6ea0f6160c3ea (patch) | |
tree | 6d71bc40d56362dd4de892239a633bdcfa2183be /streaming | |
parent | 3e2924fd466dbc05338518412f426d34b215ef0b (diff) |
More 32bit fixes (#14264)
* query planer weight calculation using long long
* adjust replication query ahead pipeline for smaller systems
* do not generate huge replication messages
* add message to indicate replication message was interrupted
* improved message
* max replication size 25% of sender buffer
* fix for last commit
* use less cache and smaller page sizes and fewer threads on 32-bits
* fix reserved libuv workers for 32bits
* fix detection of 32/64 bit
Diffstat (limited to 'streaming')
-rw-r--r-- | streaming/receiver.c | 4 | ||||
-rw-r--r-- | streaming/replication.c | 59 | ||||
-rw-r--r-- | streaming/rrdpush.c | 5 | ||||
-rw-r--r-- | streaming/rrdpush.h | 4 | ||||
-rw-r--r-- | streaming/sender.c | 4 |
5 files changed, 52 insertions, 24 deletions
diff --git a/streaming/receiver.c b/streaming/receiver.c index 012124edd9..53f8b7977d 100644 --- a/streaming/receiver.c +++ b/streaming/receiver.c @@ -457,6 +457,10 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) { } } +// this is a test +// if(rpt->hops <= host->sender->hops) +// rrdpush_sender_thread_stop(host, "HOPS MISMATCH", false); + signal_rrdcontext = true; rrdpush_receiver_replication_reset(host); diff --git a/streaming/replication.c b/streaming/replication.c index 5bfcb7540d..96859723a9 100644 --- a/streaming/replication.c +++ b/streaming/replication.c @@ -3,9 +3,10 @@ #include "replication.h" #include "Judy.h" -#define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50 -#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50 -#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10 +#define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL +#define MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER 25ULL +#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL +#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10ULL #define WORKER_JOB_FIND_NEXT 1 #define WORKER_JOB_QUERYING 2 @@ -80,6 +81,7 @@ struct replication_query { bool locked_data_collection; bool execute; + bool interrupted; } query; time_t wall_clock_time; @@ -196,8 +198,7 @@ static struct replication_query *replication_query_prepare( return q; } -static time_t replication_query_finalize(struct replication_query *q, bool executed) { - time_t query_before = q->query.before; +static void replication_query_finalize(struct replication_query *q, bool executed) { size_t dimensions = q->dimensions; // release all the dictionary items acquired @@ -231,8 +232,6 @@ static time_t replication_query_finalize(struct replication_query *q, bool execu } freez(q); - - return query_before; } static void replication_query_align_to_optimal_before(struct replication_query *q) { @@ -258,10 +257,7 @@ static void replication_query_align_to_optimal_before(struct replication_query * q->query.before = expanded_before; } -static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replication_query *q) { - if(!q->query.execute) - return replication_query_finalize(q, false); - +static void replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) { replication_query_align_to_optimal_before(q); time_t after = q->query.after; @@ -277,6 +273,7 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat #endif time_t now = after + 1; + time_t last_end_time_in_buffer = 0; while(now <= before) { time_t min_start_time = 0, min_end_time = 0; for (size_t i = 0; i < dimensions ;i++) { @@ -332,6 +329,20 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat actual_before = min_end_time; #endif + if(buffer_strlen(wb) > max_msg_size && last_end_time_in_buffer) { + internal_error(true, "REPLICATION: buffer size %zu is more than the max message size %zu for chart '%s' of host '%s'." + "Interrupting replication query at %ld, before the expected %ld.", + buffer_strlen(wb), max_msg_size, rrdset_id(q->st), rrdhost_hostname(q->st->rrdhost), + last_end_time_in_buffer, q->query.before); + + q->query.before = last_end_time_in_buffer; + q->query.enable_streaming = false; + q->query.interrupted = true; + + break; + } + last_end_time_in_buffer = min_end_time; + buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN " '' %llu %llu %llu\n", (unsigned long long) min_start_time, (unsigned long long) min_end_time, @@ -381,7 +392,6 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat q->points_read = points_read; q->points_generated = points_generated; - return replication_query_finalize(q, true); } static void replication_send_chart_collection_state(BUFFER *wb, RRDSET *st) { @@ -445,13 +455,10 @@ void replication_response_cancel_and_finalize(struct replication_query *q) { static bool sender_is_still_connected_for_this_request(struct replication_request *rq); -bool replication_response_execute_and_finalize(struct replication_query *q) { +bool replication_response_execute_and_finalize(struct replication_query *q, size_t max_msg_size) { struct replication_request *rq = q->rq; RRDSET *st = q->st; RRDHOST *host = st->rrdhost; - time_t after = q->request.after; - time_t before; // the query will report this - bool enable_streaming = q->query.enable_streaming; // we might want to optimize this by filling a temporary buffer // and copying the result to the host's buffer in order to avoid @@ -463,10 +470,15 @@ bool replication_response_execute_and_finalize(struct replication_query *q) { bool locked_data_collection = q->query.locked_data_collection; q->query.locked_data_collection = false; - before = replication_query_execute_and_finalize(wb, q); + if(q->query.execute) + replication_query_execute(wb, q, max_msg_size); - // IMPORTANT: q is invalid now - q = NULL; + time_t after = q->request.after; + time_t before = q->query.before; + bool enable_streaming = q->query.enable_streaming; + + replication_query_finalize(q, q->query.execute); + q = NULL; // IMPORTANT: q is invalid now // get again the world clock time if(enable_streaming) @@ -1204,7 +1216,8 @@ static bool replication_execute_request(struct replication_request *rq, bool wor // send the replication data rq->q->rq = rq; - replication_response_execute_and_finalize(rq->q); + replication_response_execute_and_finalize( + rq->q, (size_t)((unsigned long long)rq->sender->host->sender->buffer->max_size * MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER / 100ULL)); netdata_thread_enable_cancelability(); @@ -1404,7 +1417,11 @@ static int replication_execute_next_pending_request(void) { struct replication_request *rq; if(unlikely(!rqs)) { - max_requests_ahead = libuv_worker_threads * 2; + max_requests_ahead = get_system_cpus() / 2; + + if(max_requests_ahead > libuv_worker_threads * 2) + max_requests_ahead = libuv_worker_threads * 2; + if(max_requests_ahead < 2) max_requests_ahead = 2; diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c index 27c0b24f4e..9fa32b9df5 100644 --- a/streaming/rrdpush.c +++ b/streaming/rrdpush.c @@ -633,9 +633,10 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) { struct receiver_state *rpt = callocz(1, sizeof(*rpt)); rpt->last_msg_t = now_realtime_sec(); rpt->capabilities = STREAM_CAP_INVALID; + rpt->hops = 1; rpt->system_info = callocz(1, sizeof(struct rrdhost_system_info)); - rpt->system_info->hops = 1; + rpt->system_info->hops = rpt->hops; rpt->fd = w->ifd; rpt->client_ip = strdupz(w->client_ip); @@ -689,7 +690,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) { rpt->utc_offset = (int32_t)strtol(value, NULL, 0); else if(!strcmp(name, "hops")) - rpt->system_info->hops = (uint16_t) strtoul(value, NULL, 0); + rpt->hops = rpt->system_info->hops = (uint16_t) strtoul(value, NULL, 0); else if(!strcmp(name, "ml_capable")) rpt->system_info->ml_capable = strtoul(value, NULL, 0); diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h index 222a23bb07..406fdcdcb2 100644 --- a/streaming/rrdpush.h +++ b/streaming/rrdpush.h @@ -161,6 +161,8 @@ struct sender_state { int rrdpush_sender_pipe[2]; // collector to sender thread signaling int rrdpush_sender_socket; + uint16_t hops; + #ifdef ENABLE_COMPRESSION struct compressor_state *compressor; #endif @@ -232,6 +234,8 @@ struct receiver_state { char read_buffer[PLUGINSD_LINE_MAX + 1]; int read_len; + uint16_t hops; + struct { bool shutdown; // signal the streaming parser to exit const char *reason; // the reason of disconnection to log diff --git a/streaming/sender.c b/streaming/sender.c index dd76389462..4a33990558 100644 --- a/streaming/sender.c +++ b/streaming/sender.c @@ -510,6 +510,8 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p stream_encoded_t se; rrdpush_encode_variable(&se, host); + host->sender->hops = host->system_info->hops + 1; + char http[HTTP_HEADER_SIZE + 1]; int eol = snprintfz(http, HTTP_HEADER_SIZE, "STREAM " @@ -568,7 +570,7 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p , rrdhost_timezone(host) , rrdhost_abbrev_timezone(host) , host->utc_offset - , host->system_info->hops + 1 + , host->sender->hops , host->system_info->ml_capable , host->system_info->ml_enabled , host->system_info->mc_version |