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
Costa Tsaousis committed
Jan 14, 2023 at 00:09 UTC
f10f19c8c8bba37e8364fa6138d6ea0f6160c3ea
11 files changed
+91
-34
daemon/main.c
+4
@@ -4,6 +4,10 @@
4
#include "buildinfo.h"
5
#include "static_threads.h"
6
7
+#if defined(ENV32BIT)
8
+#warning COMPILING 32BIT NETDATA
9
+#endif
10
+
11
bool unittest_running = false;
12
int netdata_zero_metrics_enabled;
13
int netdata_anonymous_statistics_enabled;
database/engine/rrdengineapi.c
+11
-2
@@ -10,13 +10,17 @@ struct rrdengine_instance multidb_ctx_storage_tier4;
10
11
#define mrg_metric_ctx(metric) (struct rrdengine_instance *)mrg_metric_section(main_mrg, metric)
12
13
-
13
#if RRD_STORAGE_TIERS != 5
14
#error RRD_STORAGE_TIERS is not 5 - you need to add allocations here
15
#endif
16
struct rrdengine_instance *multidb_ctx[RRD_STORAGE_TIERS];
17
uint8_t tier_page_type[RRD_STORAGE_TIERS] = {PAGE_METRICS, PAGE_TIER, PAGE_TIER, PAGE_TIER, PAGE_TIER};
18
+
19
+#if defined(ENV32BIT)
20
+size_t tier_page_size[RRD_STORAGE_TIERS] = {2048, 1024, 192, 192, 192};
21
+#else
22
size_t tier_page_size[RRD_STORAGE_TIERS] = {4096, 2048, 384, 384, 384};
23
+#endif
24
25
#if PAGE_TYPE_MAX != 1
26
#error PAGE_TYPE_MAX is not 1 - you need to add allocations here
@@ -33,11 +37,16 @@ __attribute__((constructor)) void initialize_multidb_ctx(void) {
37
38
int default_rrdeng_page_fetch_timeout = 3;
39
int default_rrdeng_page_fetch_retries = 3;
36
-int default_rrdeng_page_cache_mb = 32;
40
int db_engine_journal_check = 0;
41
int default_rrdeng_disk_quota_mb = 256;
42
int default_multidb_disk_quota_mb = 256;
43
44
+#if defined(ENV32BIT)
45
+int default_rrdeng_page_cache_mb = 16;
46
+#else
47
+int default_rrdeng_page_cache_mb = 32;
48
+#endif
49
+
50
// ----------------------------------------------------------------------------
51
// metrics groups
52
database/rrd.h
+7
@@ -152,9 +152,16 @@ extern int default_rrd_history_entries;
152
extern int gap_when_lost_iterations_above;
153
extern time_t rrdset_free_obsolete_time_s;
154
155
+#if defined(ENV32BIT)
156
+#define MIN_LIBUV_WORKER_THREADS 8
157
+#define MAX_LIBUV_WORKER_THREADS 64
158
+#define RESERVED_LIBUV_WORKER_THREADS 3
159
+#else
160
#define MIN_LIBUV_WORKER_THREADS 16
161
#define MAX_LIBUV_WORKER_THREADS 128
162
#define RESERVED_LIBUV_WORKER_THREADS 6
163
+#endif
164
+
165
extern int libuv_worker_threads;
166
167
#define RRD_ID_LENGTH_MAX 200
libnetdata/libnetdata.h
+6
@@ -15,6 +15,12 @@ extern "C" {
15
#define NETDATA_INTERNAL_CHECKS 1
16
#endif
17
18
+#if SIZEOF_VOID_P == 4
19
+#define ENV32BIT 1
20
+#else
21
+#define ENV64BIT 1
22
+#endif
23
+
24
// NETDATA_TRACE_ALLOCATIONS does not work under musl libc, so don't enable it
25
//#if defined(NETDATA_INTERNAL_CHECKS) && !defined(NETDATA_TRACE_ALLOCATIONS)
26
//#define NETDATA_TRACE_ALLOCATIONS 1
ml/Host.cc
+3
@@ -142,6 +142,9 @@ void Host::detectOnce() {
142
TSCopy.RemainingUT = 0;
143
}
144
145
+ if(!RH)
146
+ return;
147
+
148
worker_is_busy(WORKER_JOB_DETECTION_DIM_CHART);
149
updateDimensionsChart(RH, MLSCopy);
150
streaming/receiver.c
+4
@@ -457,6 +457,10 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
457
}
458
}
459
460
+// this is a test
461
+// if(rpt->hops <= host->sender->hops)
462
+// rrdpush_sender_thread_stop(host, "HOPS MISMATCH", false);
463
+
464
signal_rrdcontext = true;
465
rrdpush_receiver_replication_reset(host);
466
streaming/replication.c
+38
-21
@@ -3,9 +3,10 @@
3
#include "replication.h"
4
#include "Judy.h"
5
6
-#define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50
7
-#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50
8
-#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10
6
+#define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL
7
+#define MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER 25ULL
8
+#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL
9
+#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10ULL
10
11
#define WORKER_JOB_FIND_NEXT 1
12
#define WORKER_JOB_QUERYING 2
@@ -80,6 +81,7 @@ struct replication_query {
81
82
bool locked_data_collection;
83
bool execute;
84
+ bool interrupted;
85
} query;
86
87
time_t wall_clock_time;
@@ -196,8 +198,7 @@ static struct replication_query *replication_query_prepare(
198
return q;
199
}
200
199
-static time_t replication_query_finalize(struct replication_query *q, bool executed) {
200
- time_t query_before = q->query.before;
201
+static void replication_query_finalize(struct replication_query *q, bool executed) {
202
size_t dimensions = q->dimensions;
203
204
// release all the dictionary items acquired
@@ -231,8 +232,6 @@ static time_t replication_query_finalize(struct replication_query *q, bool execu
232
}
233
234
freez(q);
234
-
235
- return query_before;
235
}
236
237
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 *
257
q->query.before = expanded_before;
258
}
259
261
-static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replication_query *q) {
262
- if(!q->query.execute)
263
- return replication_query_finalize(q, false);
264
-
260
+static void replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) {
261
replication_query_align_to_optimal_before(q);
262
263
time_t after = q->query.after;
@@ -277,6 +273,7 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat
273
#endif
274
275
time_t now = after + 1;
276
+ time_t last_end_time_in_buffer = 0;
277
while(now <= before) {
278
time_t min_start_time = 0, min_end_time = 0;
279
for (size_t i = 0; i < dimensions ;i++) {
@@ -332,6 +329,20 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat
329
actual_before = min_end_time;
330
#endif
331
332
+ if(buffer_strlen(wb) > max_msg_size && last_end_time_in_buffer) {
333
+ internal_error(true, "REPLICATION: buffer size %zu is more than the max message size %zu for chart '%s' of host '%s'."
334
+ "Interrupting replication query at %ld, before the expected %ld.",
335
+ buffer_strlen(wb), max_msg_size, rrdset_id(q->st), rrdhost_hostname(q->st->rrdhost),
336
+ last_end_time_in_buffer, q->query.before);
337
+
338
+ q->query.before = last_end_time_in_buffer;
339
+ q->query.enable_streaming = false;
340
+ q->query.interrupted = true;
341
+
342
+ break;
343
+ }
344
+ last_end_time_in_buffer = min_end_time;
345
+
346
buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN " '' %llu %llu %llu\n",
347
(unsigned long long) min_start_time,
348
(unsigned long long) min_end_time,
@@ -381,7 +392,6 @@ static time_t replication_query_execute_and_finalize(BUFFER *wb, struct replicat
392
393
q->points_read = points_read;
394
q->points_generated = points_generated;
384
- return replication_query_finalize(q, true);
395
}
396
397
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) {
455
456
static bool sender_is_still_connected_for_this_request(struct replication_request *rq);
457
448
-bool replication_response_execute_and_finalize(struct replication_query *q) {
458
+bool replication_response_execute_and_finalize(struct replication_query *q, size_t max_msg_size) {
459
struct replication_request *rq = q->rq;
460
RRDSET *st = q->st;
461
RRDHOST *host = st->rrdhost;
452
- time_t after = q->request.after;
453
- time_t before; // the query will report this
454
- bool enable_streaming = q->query.enable_streaming;
462
463
// we might want to optimize this by filling a temporary buffer
464
// 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) {
470
bool locked_data_collection = q->query.locked_data_collection;
471
q->query.locked_data_collection = false;
472
466
- before = replication_query_execute_and_finalize(wb, q);
473
+ if(q->query.execute)
474
+ replication_query_execute(wb, q, max_msg_size);
475
468
- // IMPORTANT: q is invalid now
469
- q = NULL;
476
+ time_t after = q->request.after;
477
+ time_t before = q->query.before;
478
+ bool enable_streaming = q->query.enable_streaming;
479
+
480
+ replication_query_finalize(q, q->query.execute);
481
+ q = NULL; // IMPORTANT: q is invalid now
482
483
// get again the world clock time
484
if(enable_streaming)
@@ -1204,7 +1216,8 @@ static bool replication_execute_request(struct replication_request *rq, bool wor
1216
1217
// send the replication data
1218
rq->q->rq = rq;
1207
- replication_response_execute_and_finalize(rq->q);
1219
+ replication_response_execute_and_finalize(
1220
+ rq->q, (size_t)((unsigned long long)rq->sender->host->sender->buffer->max_size * MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER / 100ULL));
1221
1222
netdata_thread_enable_cancelability();
1223
@@ -1404,7 +1417,11 @@ static int replication_execute_next_pending_request(void) {
1417
struct replication_request *rq;
1418
1419
if(unlikely(!rqs)) {
1407
- max_requests_ahead = libuv_worker_threads * 2;
1420
+ max_requests_ahead = get_system_cpus() / 2;
1421
+
1422
+ if(max_requests_ahead > libuv_worker_threads * 2)
1423
+ max_requests_ahead = libuv_worker_threads * 2;
1424
+
1425
if(max_requests_ahead < 2)
1426
max_requests_ahead = 2;
1427
streaming/rrdpush.c
+3
-2
@@ -633,9 +633,10 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
633
struct receiver_state *rpt = callocz(1, sizeof(*rpt));
634
rpt->last_msg_t = now_realtime_sec();
635
rpt->capabilities = STREAM_CAP_INVALID;
636
+ rpt->hops = 1;
637
638
rpt->system_info = callocz(1, sizeof(struct rrdhost_system_info));
638
- rpt->system_info->hops = 1;
639
+ rpt->system_info->hops = rpt->hops;
640
641
rpt->fd = w->ifd;
642
rpt->client_ip = strdupz(w->client_ip);
@@ -689,7 +690,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
690
rpt->utc_offset = (int32_t)strtol(value, NULL, 0);
691
692
else if(!strcmp(name, "hops"))
692
- rpt->system_info->hops = (uint16_t) strtoul(value, NULL, 0);
693
+ rpt->hops = rpt->system_info->hops = (uint16_t) strtoul(value, NULL, 0);
694
695
else if(!strcmp(name, "ml_capable"))
696
rpt->system_info->ml_capable = strtoul(value, NULL, 0);
streaming/rrdpush.h
+4
@@ -161,6 +161,8 @@ struct sender_state {
161
int rrdpush_sender_pipe[2]; // collector to sender thread signaling
162
int rrdpush_sender_socket;
163
164
+ uint16_t hops;
165
+
166
#ifdef ENABLE_COMPRESSION
167
struct compressor_state *compressor;
168
#endif
@@ -232,6 +234,8 @@ struct receiver_state {
234
char read_buffer[PLUGINSD_LINE_MAX + 1];
235
int read_len;
236
237
+ uint16_t hops;
238
+
239
struct {
240
bool shutdown; // signal the streaming parser to exit
241
const char *reason; // the reason of disconnection to log
streaming/sender.c
+3
-1
@@ -510,6 +510,8 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
510
stream_encoded_t se;
511
rrdpush_encode_variable(&se, host);
512
513
+ host->sender->hops = host->system_info->hops + 1;
514
+
515
char http[HTTP_HEADER_SIZE + 1];
516
int eol = snprintfz(http, HTTP_HEADER_SIZE,
517
"STREAM "
@@ -568,7 +570,7 @@ static bool rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_p
570
, rrdhost_timezone(host)
571
, rrdhost_abbrev_timezone(host)
572
, host->utc_offset
571
- , host->system_info->hops + 1
573
+ , host->sender->hops
574
, host->system_info->ml_capable
575
, host->system_info->ml_enabled
576
, host->system_info->mc_version
web/api/queries/query.c
+8
-8
@@ -726,15 +726,15 @@ static long query_plan_points_coverage_weight(time_t db_first_time_s, time_t db_
726
db_last_time_s < after_wanted)
727
return -LONG_MAX;
728
729
- time_t common_first_t = MAX(db_first_time_s, after_wanted);
730
- time_t common_last_t = MIN(db_last_time_s, before_wanted);
729
+ long long common_first_t = MAX(db_first_time_s, after_wanted);
730
+ long long common_last_t = MIN(db_last_time_s, before_wanted);
731
732
- long time_coverage = (common_last_t - common_first_t) * 1000000 / (before_wanted - after_wanted);
733
- size_t points_wanted_in_coverage = points_wanted * time_coverage / 1000000;
732
+ long long time_coverage = (common_last_t - common_first_t) * 1000000LL / (before_wanted - after_wanted);
733
+ long long points_wanted_in_coverage = (long long)points_wanted * time_coverage / 1000000LL;
734
735
- long points_available = (common_last_t - common_first_t) / db_update_every_s;
736
- long points_delta = (long)(points_available - points_wanted_in_coverage);
737
- long points_coverage = (points_delta < 0) ? (long)(points_available * time_coverage / points_wanted_in_coverage) : time_coverage;
735
+ long long points_available = (common_last_t - common_first_t) / db_update_every_s;
736
+ long long points_delta = (long)(points_available - points_wanted_in_coverage);
737
+ long long points_coverage = (points_delta < 0) ? (long)(points_available * time_coverage / points_wanted_in_coverage) : time_coverage;
738
739
// a way to benefit higher tiers
740
// points_coverage += (long)tier * 10000;
@@ -742,7 +742,7 @@ static long query_plan_points_coverage_weight(time_t db_first_time_s, time_t db_
742
if(points_available <= 0)
743
return -LONG_MAX;
744
745
- return points_coverage + (long)(25000 * tier); // 2.5% benefit for each higher tier
745
+ return (long)(points_coverage + (25000LL * tier)); // 2.5% benefit for each higher tier
746
}
747
748
static size_t query_metric_best_tier_for_timeframe(QUERY_METRIC *qm, time_t after_wanted, time_t before_wanted, size_t points_wanted) {