@cryptotaxi247 / netdata-1 / commits / 63900bcf1

Streaming improvements No 8 (#19206)

* add sqlite3 memory to total netdata memory * recheck stream_conf_is_parent() when stream.conf is loaded * cleanup about getting the number of cores; we now also use the official posix way * fix the log message when a node connects for the first time * better alternative for freebsd * allow empty flags * do not expect a filename on non-linux machines * enable deleting dyncfg alerts, even if they are on disk

Costa Tsaousis committed Dec 15, 2024 at 21:40 UTC 63900bcf129b16354738d9ed653ab0e8e15956cd
28 files changed +215 -200
src/daemon/common.h
-2
@@ -96,8 +96,6 @@ extern int netdata_anonymous_statistics_enabled;
96 extern bool netdata_ready;
97 extern time_t netdata_start_time;
98
99 -long get_netdata_cpus(void);
100 -
99 void set_environment_for_plugins_and_scripts(void);
100
101 #ifdef __cplusplus
src/daemon/config/netdata-conf-cloud.c
+3 -3
@@ -3,9 +3,9 @@
3 #include "netdata-conf-cloud.h"
4 #include "../common.h"
5
6 -int netdata_conf_cloud_query_threads(void) {
7 - int cpus = MIN(get_netdata_cpus(), 256); // max 256 cores
8 - int threads = MIN(cpus * (stream_conf_is_parent(false) ? 2 : 1), libuv_worker_threads / 2);
6 +size_t netdata_conf_cloud_query_threads(void) {
7 + size_t cpus = MIN(netdata_conf_cpus(), 256); // max 256 cores
8 + size_t threads = MIN(cpus * (stream_conf_is_parent(false) ? 2 : 1), (size_t)libuv_worker_threads / 2);
9 threads = MAX(threads, 6);
10
11 threads = config_get_number(CONFIG_SECTION_CLOUD, "query threads", threads);
src/daemon/config/netdata-conf-cloud.h
+1 -1
@@ -5,6 +5,6 @@
5
6 #include "libnetdata/libnetdata.h"
7
8 -int netdata_conf_cloud_query_threads(void);
8 +size_t netdata_conf_cloud_query_threads(void);
9
10 #endif //NETDATA_NETDATA_CONF_CLOUD_H
src/daemon/config/netdata-conf-db.c
+1 -1
@@ -250,7 +250,7 @@ void netdata_conf_dbengine_init(const char *hostname) {
250 // fails on Windows.
251 bool parallel_initialization = false;
252 #else
253 - bool parallel_initialization = (storage_tiers <= (size_t)get_netdata_cpus()) ? true : false;
253 + bool parallel_initialization = (storage_tiers <= netdata_conf_cpus()) ? true : false;
254 #endif
255
256 struct dbengine_initialization tiers_init[RRD_STORAGE_TIERS] = {};
src/daemon/config/netdata-conf-global.c
+39 -1
@@ -2,6 +2,43 @@
2
3 #include "netdata-conf-global.h"
4
5 +size_t netdata_conf_cpus(void) {
6 + static size_t processors = 0;
7 +
8 + if(processors)
9 + return processors;
10 +
11 + SPINLOCK spinlock = SPINLOCK_INITIALIZER;
12 + spinlock_lock(&spinlock);
13 + size_t p = 0;
14 +
15 + if(processors)
16 + goto skip;
17 +
18 +#if defined(OS_LINUX)
19 + p = os_read_cpuset_cpus("/sys/fs/cgroup/cpuset.cpus", p);
20 + if(!p)
21 + p = os_read_cpuset_cpus("/sys/fs/cgroup/cpuset/cpuset.cpus", p);
22 +#endif
23 +
24 + if(!p)
25 + p = os_get_system_cpus_uncached();
26 +
27 + p = config_get_number(CONFIG_SECTION_GLOBAL, "cpu cores", p);
28 + if(p < 1)
29 + p = 1;
30 +
31 + processors = p;
32 +
33 + char buf[24];
34 + snprintfz(buf, sizeof(buf), "%zu", processors);
35 + nd_setenv("NETDATA_CONF_CPUS", buf, 1);
36 +
37 +skip:
38 + spinlock_unlock(&spinlock);
39 + return processors;
40 +}
41 +
42 static int get_hostname(char *buf, size_t buf_size) {
43 if (netdata_configured_host_prefix && *netdata_configured_host_prefix) {
44 char filename[FILENAME_MAX + 1];
@@ -36,7 +73,7 @@ static void glibc_initialize(void) {
73 }
74
75 static void libuv_initialize(void) {
39 - libuv_worker_threads = (int)get_netdata_cpus() * 6;
76 + libuv_worker_threads = (int)netdata_conf_cpus() * 6;
77
78 if(libuv_worker_threads < MIN_LIBUV_WORKER_THREADS)
79 libuv_worker_threads = MIN_LIBUV_WORKER_THREADS;
@@ -96,3 +133,4 @@ void netdata_conf_section_global_run_as_user(const char **user) {
133 *user = config_get(CONFIG_SECTION_GLOBAL, "run as user", (passwd && passwd->pw_name)?passwd->pw_name:"");
134 }
135 }
136 +
src/daemon/config/netdata-conf-global.h
+2
@@ -8,6 +8,8 @@
8 void netdata_conf_section_global(void);
9 void netdata_conf_section_global_run_as_user(const char **user);
10
11 +size_t netdata_conf_cpus(void);
12 +
13 #include "netdata-conf.h"
14
15 #endif //NETDATA_NETDATA_CONF_GLOBAL_H
src/daemon/config/netdata-conf-web.c
+3 -3
@@ -3,7 +3,7 @@
3 #include "netdata-conf-web.h"
4 #include "daemon/static_threads.h"
5
6 -int netdata_conf_web_query_threads(void) {
6 +size_t netdata_conf_web_query_threads(void) {
7 // See https://github.com/netdata/netdata/issues/11081#issuecomment-831998240 for more details
8 if (OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110) {
9 config_set_number(CONFIG_SECTION_WEB, "web server threads", 1);
@@ -11,8 +11,8 @@ int netdata_conf_web_query_threads(void) {
11 return 1;
12 }
13
14 - int cpus = MIN(get_netdata_cpus(), 256); // max 256 cores
15 - int threads = cpus * (stream_conf_is_parent(false) ? 2 : 1);
14 + size_t cpus = MIN(netdata_conf_cpus(), 256); // max 256 cores
15 + size_t threads = cpus * (stream_conf_is_parent(false) ? 2 : 1);
16 threads = MAX(threads, 6);
17
18 threads = config_get_number(CONFIG_SECTION_WEB, "web server threads", threads);
src/daemon/config/netdata-conf-web.h
+1 -1
@@ -9,6 +9,6 @@ void netdata_conf_section_web(void);
9 void web_server_threading_selection(void);
10 void netdata_conf_web_security_init(void);
11
12 -int netdata_conf_web_query_threads(void);
12 +size_t netdata_conf_web_query_threads(void);
13
14 #endif //NETDATA_NETDATA_CONF_WEB_H
src/daemon/daemon.c
-39
@@ -6,45 +6,6 @@
6 char *pidfile = NULL;
7 char *netdata_exe_path = NULL;
8
9 -long get_netdata_cpus(void) {
10 - static long processors = 0;
11 -
12 - if(processors)
13 - return processors;
14 -
15 - long cores_proc_stat = os_get_system_cpus_cached(false, true);
16 - long cores_cpuset_v1 = (long)os_read_cpuset_cpus("/sys/fs/cgroup/cpuset/cpuset.cpus", cores_proc_stat);
17 - long cores_cpuset_v2 = (long)os_read_cpuset_cpus("/sys/fs/cgroup/cpuset.cpus", cores_proc_stat);
18 -
19 - if(cores_cpuset_v2)
20 - processors = cores_cpuset_v2;
21 - else if(cores_cpuset_v1)
22 - processors = cores_cpuset_v1;
23 - else
24 - processors = cores_proc_stat;
25 -
26 - long cores_user_configured = config_get_number(CONFIG_SECTION_GLOBAL, "cpu cores", processors);
27 -
28 - errno_clear();
29 - internal_error(true,
30 - "System CPUs: %ld, ("
31 - "system: %ld, cgroups cpuset v1: %ld, cgroups cpuset v2: %ld, netdata.conf: %ld"
32 - ")"
33 - , processors
34 - , cores_proc_stat
35 - , cores_cpuset_v1
36 - , cores_cpuset_v2
37 - , cores_user_configured
38 - );
39 -
40 - processors = cores_user_configured;
41 -
42 - if(processors < 1)
43 - processors = 1;
44 -
45 - return processors;
46 -}
47 -
9 void get_netdata_execution_path(void) {
10 struct passwd *passwd = getpwuid(getuid());
11 char *user = (passwd && passwd->pw_name) ? passwd->pw_name : "";
src/daemon/pulse/pulse-daemon-memory.c
+6
@@ -14,6 +14,7 @@ void pulse_daemon_memory_do(bool extended) {
14 static RRDSET *st_memory = NULL;
15 static RRDDIM *rd_db_dbengine = NULL;
16 static RRDDIM *rd_db_rrd = NULL;
17 + static RRDDIM *rd_db_sqlite3 = NULL;
18
19 #ifdef DICT_WITH_STATS
20 static RRDDIM *rd_collectors = NULL;
@@ -54,6 +55,7 @@ void pulse_daemon_memory_do(bool extended) {
55
56 rd_db_dbengine = rrddim_add(st_memory, "dbengine", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
57 rd_db_rrd = rrddim_add(st_memory, "rrd", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
58 + rd_db_sqlite3 = rrddim_add(st_memory, "sqlite3", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
59
60 #ifdef DICT_WITH_STATS
61 rd_collectors = rrddim_add(st_memory, "collectors", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
@@ -93,11 +95,15 @@ void pulse_daemon_memory_do(bool extended) {
95 netdata_buffers_statistics.buffers_web +
96 replication_allocated_buffers() + aral_by_size_free_bytes() + judy_aral_free_bytes();
97
98 + int sqlite3_memory_used_current = 0, sqlite3_memory_used_highwater = 0;
99 + sqlite3_status(SQLITE_STATUS_MEMORY_USED, &sqlite3_memory_used_current, &sqlite3_memory_used_highwater, 1);
100 +
101 size_t strings = 0;
102 string_statistics(NULL, NULL, NULL, NULL, NULL, &strings, NULL, NULL);
103
104 rrddim_set_by_pointer(st_memory, rd_db_dbengine, (collected_number)pulse_dbengine_total_memory);
105 rrddim_set_by_pointer(st_memory, rd_db_rrd, (collected_number)pulse_rrd_memory_size);
106 + rrddim_set_by_pointer(st_memory, rd_db_sqlite3, (collected_number)sqlite3_memory_used_highwater);
107
108 #ifdef DICT_WITH_STATS
109 rrddim_set_by_pointer(st_memory, rd_collectors,
src/database/engine/cache.c
+4 -4
@@ -2046,11 +2046,11 @@ PGC *pgc_create(const char *name,
2046 cache->config.out_of_memory_protection_bytes = dbengine_out_of_memory_protection;
2047
2048 // partitions
2049 - if(partitions == 0) partitions = get_netdata_cpus();
2050 - if(partitions <= 4) partitions = 4;
2049 + if(partitions == 0) partitions = netdata_conf_cpus();
2050 + if(partitions <= 4) partitions = 4;
2051 if(partitions > 256) partitions = 256;
2052 - cache->config.partitions = partitions;
2053 - cache->index = callocz(cache->config.partitions, sizeof(struct pgc_index));
2052 + cache->config.partitions = partitions;
2053 + cache->index = callocz(cache->config.partitions, sizeof(struct pgc_index));
2054
2055 pgc_section_pages_static_aral_init();
2056
src/database/engine/cache.h
+2 -4
@@ -320,14 +320,12 @@ static inline size_t indexing_partition(Word_t ptr, Word_t modulo) {
320 return hash % modulo;
321 }
322
323 -long get_netdata_cpus(void);
324 -
323 static inline size_t pgc_max_evictors(void) {
326 - return 1 + get_netdata_cpus() / 2;
324 + return 1 + netdata_conf_cpus() / 2;
325 }
326
327 static inline size_t pgc_max_flushers(void) {
330 - return get_netdata_cpus();
328 + return netdata_conf_cpus();
329 }
330
331 #endif // DBENGINE_CACHE_H
src/database/engine/metric.c
+1 -1
@@ -367,7 +367,7 @@ static inline METRIC *metric_get_and_acquire(MRG *mrg, nd_uuid_t *uuid, Word_t s
367
368 inline MRG *mrg_create(ssize_t partitions) {
369 if(partitions < 1)
370 - partitions = get_netdata_cpus();
370 + partitions = (ssize_t)netdata_conf_cpus();
371
372 MRG *mrg = callocz(1, sizeof(MRG) + sizeof(struct mrg_partition) * partitions);
373 mrg->partitions = partitions;
src/database/engine/page.c
+1 -1
@@ -121,7 +121,7 @@ int aral_size_sort_compare(const void *a, const void *b) {
121 }
122
123 void pgd_init_arals(void) {
124 - size_t partitions = get_netdata_cpus();
124 + size_t partitions = netdata_conf_cpus();
125 if(partitions < 4) partitions = 4;
126 if(partitions > PGD_ARAL_PARTITIONS_MAX) partitions = PGD_ARAL_PARTITIONS_MAX;
127 pgd_alloc_globals.partitions = partitions;
src/database/engine/rrdengineapi.c
+3 -3
@@ -1065,15 +1065,15 @@ static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) {
1065 datafiles++;
1066 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1067
1068 - ssize_t cpus = (ssize_t)get_netdata_cpus() / (ssize_t)storage_tiers;
1068 + ssize_t cpus = (ssize_t)netdata_conf_cpus() / (ssize_t)storage_tiers;
1069 if(cpus > (ssize_t)datafiles)
1070 cpus = (ssize_t)datafiles;
1071
1072 if(cpus > (ssize_t)libuv_worker_threads)
1073 cpus = (ssize_t)libuv_worker_threads;
1074
1075 - if(cpus >= (ssize_t)get_netdata_cpus() / 2)
1076 - cpus = (ssize_t)(get_netdata_cpus() / 2 - 1);
1075 + if(cpus >= (ssize_t)netdata_conf_cpus() / 2)
1076 + cpus = (ssize_t)(netdata_conf_cpus() / 2 - 1);
1077
1078 if(cpus < 1)
1079 cpus = 1;
src/database/sqlite/sqlite_metadata.c
+1 -1
@@ -1699,7 +1699,7 @@ static void start_all_host_load_context(uv_work_t *req __maybe_unused)
1699
1700 RRDHOST *host;
1701
1702 - size_t max_threads = MIN(get_netdata_cpus() / 2, 6);
1702 + size_t max_threads = MIN(netdata_conf_cpus() / 2, 6);
1703 if (max_threads < 1)
1704 max_threads = 1;
1705
src/health/health_dyncfg.c
+1 -1
@@ -802,7 +802,7 @@ static void health_dyncfg_register_prototype(RRD_ALERT_PROTOTYPE *ap) {
802 ap->config.source_type, string2str(ap->config.source),
803 DYNCFG_CMD_SCHEMA | DYNCFG_CMD_GET | DYNCFG_CMD_ENABLE | DYNCFG_CMD_DISABLE |
804 DYNCFG_CMD_UPDATE | DYNCFG_CMD_USERCONFIG |
805 - (ap->config.source_type == DYNCFG_SOURCE_TYPE_DYNCFG && !ap->_internal.is_on_disk ? DYNCFG_CMD_REMOVE : 0),
805 + (ap->config.source_type == DYNCFG_SOURCE_TYPE_DYNCFG /* && !ap->_internal.is_on_disk */ ? DYNCFG_CMD_REMOVE : 0),
806 HTTP_ACCESS_NONE,
807 HTTP_ACCESS_NONE,
808 dyncfg_health_cb, NULL);
src/libnetdata/os/get_system_cpus.c
+104 -58
@@ -2,95 +2,141 @@
2
3 #include "../libnetdata.h"
4
5 -#define CPUS_FOR_COLLECTORS 0
6 -#define CPUS_FOR_NETDATA 1
5 +size_t os_get_system_cpus_cached(bool cache) {
6 + static size_t processors = 0;
7
8 -long os_get_system_cpus_cached(bool cache, bool for_netdata) {
9 - static long processors[2] = { 0, 0 };
8 + if(likely(cache && processors > 0))
9 + return processors;
10
11 - int index = for_netdata ? CPUS_FOR_NETDATA : CPUS_FOR_COLLECTORS;
11 + SPINLOCK spinlock = SPINLOCK_INITIALIZER;
12 + spinlock_lock(&spinlock);
13
13 - if(likely(cache && processors[index] > 0))
14 - return processors[index];
14 + long p = 0;
15
16 -#if defined(OS_FREEBSD) || defined(OS_MACOS)
17 -#if defined(OS_MACOS)
18 -#define HW_CPU_NAME "hw.logicalcpu"
19 -#else
20 -#define HW_CPU_NAME "hw.ncpu"
16 +#if defined(_SC_NPROCESSORS_ONLN)
17 + // currently online processors
18 + p = sysconf(_SC_NPROCESSORS_ONLN);
19 + if(p > 1) goto done; // if it is 1, we will try harder below
20 #endif
21
23 - int32_t tmp_processors;
24 - bool error = false;
25 -
26 - if (unlikely(GETSYSCTL_BY_NAME(HW_CPU_NAME, tmp_processors)))
27 - error = true;
28 - else
29 - processors[index] = tmp_processors;
22 +#if defined(_SC_NPROCESSORS_CONF)
23 + // all configured processors (online and offline)
24 + p = sysconf(_SC_NPROCESSORS_CONF);
25 + if(p > 1) goto done; // if it is 1, we will try harder below
26 +#endif
27
31 - if(processors[index] < 1) {
32 - processors[index] = 1;
28 +#if defined(OS_FREEBSD) || defined(OS_MACOS)
29 + #if defined(OS_MACOS)
30 + #define HW_CPU_NAME "hw.logicalcpu"
31 + #else
32 + #define HW_CPU_NAME "kern.smp.cpus"
33 + #endif
34
34 - if(error)
35 - netdata_log_error("Assuming system has %ld processors.", processors[index]);
36 - }
35 + if (unlikely(GETSYSCTL_BY_NAME(HW_CPU_NAME, p) || p < 1))
36 + goto error;
37
38 - return processors[index];
38 #elif defined(OS_LINUX)
39 + // we will count the number of cpus in /proc/stat
40
41 char filename[FILENAME_MAX + 1];
42 snprintfz(filename, FILENAME_MAX, "%s/proc/stat",
43 - (!for_netdata && netdata_configured_host_prefix) ? netdata_configured_host_prefix : "");
43 + (netdata_configured_host_prefix) ? netdata_configured_host_prefix : "");
44
45 procfile *ff = procfile_open(filename, NULL, PROCFILE_FLAG_DEFAULT);
46 - if(!ff) {
47 - processors[index] = 1;
48 - netdata_log_error("Cannot open file '%s'. Assuming system has %ld processors.", filename, processors[index]);
49 - return processors[index];
50 - }
46 + if(!ff || !(ff = procfile_readall(ff))) goto error;
47
52 - ff = procfile_readall(ff);
53 - if(!ff) {
54 - processors[index] = 1;
55 - netdata_log_error("Cannot open file '%s'. Assuming system has %ld processors.", filename, processors[index]);
56 - return processors[index];
57 - }
58 -
59 - long tmp_processors = 0;
48 + p = 0;
49 unsigned int i;
50 for(i = 0; i < procfile_lines(ff); i++) {
51 if(!procfile_linewords(ff, i)) continue;
52
64 - if(strncmp(procfile_lineword(ff, i, 0), "cpu", 3) == 0)
65 - tmp_processors++;
53 + const char *starting = procfile_lineword(ff, i, 0);
54 + if(strncmp(starting, "cpu", 3) == 0 && isdigit((uint8_t)starting[3]))
55 + p++;
56 }
57 procfile_close(ff);
68 -
69 - processors[index] = --tmp_processors;
70 -
71 - if(processors[index] < 1)
72 - processors[index] = 1;
73 -
74 - netdata_log_debug(D_SYSTEM, "System has %ld processors.", processors[index]);
75 - return processors[index];
58 + if(p < 1) goto error;
59
60 #elif defined(OS_WINDOWS)
61
62 SYSTEM_INFO sysInfo;
63 GetSystemInfo(&sysInfo);
81 - processors[index] = sysInfo.dwNumberOfProcessors;
64 + p = sysInfo.dwNumberOfProcessors;
65 + if(p < 1) goto error;
66 +
67 +#else
68 +
69 + p = 1;
70 +
71 +#endif
72 +
73 +done:
74 + processors = (size_t)p;
75 + spinlock_unlock(&spinlock);
76 + return processors;
77 +
78 +error:
79 + spinlock_unlock(&spinlock);
80 + processors = 1;
81 + netdata_log_error("Cannot detect number of CPU cores. Assuming the system has %zu processors.", processors);
82 + return processors;
83 +}
84
83 - if(processors[index] < 1) {
84 - processors[index] = 1;
85 - netdata_log_error("Assuming system has %ld processors.", processors[index]);
85 +// --------------------------------------------------------------------------------------------------------------------
86 +// cpuset cpus
87 +
88 +static inline unsigned long cpuset_str2ul(char **s) {
89 + unsigned long n = 0;
90 + char c;
91 + for(c = **s; c >= '0' && c <= '9' ; c = *(++*s)) {
92 + n *= 10;
93 + n += c - '0';
94 }
95 + return n;
96 +}
97
88 - return processors[index];
98 +#if defined(OS_LINUX)
99 +size_t os_read_cpuset_cpus(const char *filename, size_t system_cpus) {
100 + static char *buf = NULL;
101 + static size_t buf_size = 0;
102
90 -#else
103 + if(!buf) {
104 + buf_size = 100U + 6 * system_cpus + 1; // taken from kernel/cgroup/cpuset.c
105 + buf = mallocz(buf_size);
106 + }
107
92 - processors[index] = 1;
93 - return processors[index];
108 + int ret = read_txt_file(filename, buf, buf_size);
109 +
110 + if(!ret) {
111 + char *s = buf;
112 + unsigned long ncpus = 0;
113 +
114 + // parse the cpuset string and calculate the number of cpus the cgroup is allowed to use
115 + while (*s) {
116 + if (isspace((uint8_t)*s)) {
117 + s++;
118 + continue;
119 + }
120 + unsigned long n = cpuset_str2ul(&s);
121 + ncpus++;
122 + if(*s == ',') {
123 + s++;
124 + continue;
125 + }
126 + if(*s == '-') {
127 + s++;
128 + unsigned long m = cpuset_str2ul(&s);
129 + ncpus += m - n; // calculate the number of cpus in the region
130 + }
131 + s++;
132 + }
133 +
134 + if(!ncpus)
135 + return 0;
136 +
137 + return ncpus;
138 + }
139
95 -#endif
140 + return 0;
141 }
142 +#endif
\ No newline at end of file
src/libnetdata/os/get_system_cpus.h
+17 -1
@@ -5,6 +5,22 @@
5
6 #include "../libnetdata.h"
7
8 -long os_get_system_cpus_cached(bool cache, bool for_netdata);
8 +/*
9 + * The functions return the actual cpu cores of the system.
10 + * For configuring netdata workers, please use netdata_conf_cpus()
11 + * which is based on these settings, but it allows users to override it.
12 + *
13 + * External plugins can use the environment variable NETDATA_CONF_CPUS
14 + * to get the user configured setting.
15 + *
16 + */
17 +
18 +#define os_get_system_cpus() os_get_system_cpus_cached(true)
19 +#define os_get_system_cpus_uncached() os_get_system_cpus_cached(false)
20 +size_t os_get_system_cpus_cached(bool cache);
21 +
22 +#if defined(OS_LINUX)
23 +size_t os_read_cpuset_cpus(const char *filename, size_t system_cpus);
24 +#endif
25
26 #endif //NETDATA_GET_SYSTEM_CPUS_H
src/libnetdata/os/os.c
-54
@@ -18,60 +18,6 @@ void os_get_system_HZ(void) {
18 system_hz = (unsigned int) ticks;
19 }
20
21 -static inline unsigned long cpuset_str2ul(char **s) {
22 - unsigned long n = 0;
23 - char c;
24 - for(c = **s; c >= '0' && c <= '9' ; c = *(++*s)) {
25 - n *= 10;
26 - n += c - '0';
27 - }
28 - return n;
29 -}
30 -
31 -unsigned long os_read_cpuset_cpus(const char *filename, long system_cpus) {
32 - static char *buf = NULL;
33 - static size_t buf_size = 0;
34 -
35 - if(!buf) {
36 - buf_size = 100U + 6 * system_cpus + 1; // taken from kernel/cgroup/cpuset.c
37 - buf = mallocz(buf_size);
38 - }
39 -
40 - int ret = read_txt_file(filename, buf, buf_size);
41 -
42 - if(!ret) {
43 - char *s = buf;
44 - unsigned long ncpus = 0;
45 -
46 - // parse the cpuset string and calculate the number of cpus the cgroup is allowed to use
47 - while (*s) {
48 - if (isspace((uint8_t)*s)) {
49 - s++;
50 - continue;
51 - }
52 - unsigned long n = cpuset_str2ul(&s);
53 - ncpus++;
54 - if(*s == ',') {
55 - s++;
56 - continue;
57 - }
58 - if(*s == '-') {
59 - s++;
60 - unsigned long m = cpuset_str2ul(&s);
61 - ncpus += m - n; // calculate the number of cpus in the region
62 - }
63 - s++;
64 - }
65 -
66 - if(!ncpus)
67 - return 0;
68 -
69 - return ncpus;
70 - }
71 -
72 - return 0;
73 -}
74 -
21 // =====================================================================================================================
22 // os_type
23
src/libnetdata/os/os.h
-5
@@ -40,11 +40,6 @@
40
41 extern const char *os_type;
42
43 -#define os_get_system_cpus() os_get_system_cpus_cached(true, false)
44 -#define os_get_system_cpus_uncached() os_get_system_cpus_cached(false, false)
45 -long os_get_system_cpus_cached(bool cache, bool for_netdata);
46 -unsigned long os_read_cpuset_cpus(const char *filename, long system_cpus);
47 -
43 extern unsigned int system_hz;
44 void os_get_system_HZ(void);
45
src/ml/ml_config.cc
+2 -2
@@ -45,7 +45,7 @@ void ml_config_load(ml_config_t *cfg) {
45 std::string anomaly_detection_grouping_method = config_get(config_section_ml, "anomaly detection grouping method", "average");
46 time_t anomaly_detection_query_duration = config_get_duration_seconds(config_section_ml, "anomaly detection grouping duration", 5 * 60);
47
48 - size_t num_worker_threads = stream_conf_is_parent(false) ? get_netdata_cpus() / 4 : 1;
48 + size_t num_worker_threads = stream_conf_is_parent(false) ? netdata_conf_cpus() / 4 : 1;
49 if (num_worker_threads < 1) num_worker_threads = 1;
50 else if (num_worker_threads > 256) num_worker_threads = 256;
51 num_worker_threads = config_get_number(config_section_ml, "num training threads", num_worker_threads);
@@ -83,7 +83,7 @@ void ml_config_load(ml_config_t *cfg) {
83 host_anomaly_rate_threshold = clamp(host_anomaly_rate_threshold, 0.1, 10.0);
84 anomaly_detection_query_duration = clamp<time_t>(anomaly_detection_query_duration, 60, 15 * 60);
85
86 - num_worker_threads = clamp<size_t>(num_worker_threads, 4, os_get_system_cpus());
86 + num_worker_threads = clamp<size_t>(num_worker_threads, 4, netdata_conf_cpus());
87 flush_models_batch_size = clamp<size_t>(flush_models_batch_size, 8, 512);
88
89 suppression_window = clamp<size_t>(suppression_window, 1, max_train_samples);
src/streaming/replication.c
+5 -5
@@ -1889,18 +1889,18 @@ void *replication_thread_main(void *ptr) {
1889
1890 replication_initialize_workers(true);
1891
1892 - int threads = stream_conf_is_parent(false) ? (int)(get_netdata_cpus() / 2) : 1;
1892 + size_t threads = stream_conf_is_parent(false) ? (netdata_conf_cpus() / 2) : 1;
1893 if (threads < 1) threads = 1;
1894 else if (threads > MAX_REPLICATION_THREADS) threads = MAX_REPLICATION_THREADS;
1895
1896 threads = config_get_number(CONFIG_SECTION_DB, "replication threads", threads);
1897 if(threads < 1) {
1898 - netdata_log_error("replication threads given %d is invalid, resetting to 1", threads);
1898 + netdata_log_error("replication threads given %zu is invalid, resetting to 1", threads);
1899 threads = 1;
1900 config_set_number(CONFIG_SECTION_DB, "replication threads", threads);
1901 }
1902 else if(threads > MAX_REPLICATION_THREADS) {
1903 - netdata_log_error("replication threads given %d is invalid, resetting to %d", threads, (int)MAX_REPLICATION_THREADS);
1903 + netdata_log_error("replication threads given %zu is invalid, resetting to %d", threads, (int)MAX_REPLICATION_THREADS);
1904 threads = MAX_REPLICATION_THREADS;
1905 config_set_number(CONFIG_SECTION_DB, "replication threads", threads);
1906 }
@@ -1910,9 +1910,9 @@ void *replication_thread_main(void *ptr) {
1910 replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(ND_THREAD *));
1911 __atomic_add_fetch(&replication_buffers_allocated, threads * sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1912
1913 - for(int i = 0; i < threads ;i++) {
1913 + for(size_t i = 0; i < threads ;i++) {
1914 char tag[NETDATA_THREAD_TAG_MAX + 1];
1915 - snprintfz(tag, NETDATA_THREAD_TAG_MAX, "REPLAY[%d]", i + 2);
1915 + snprintfz(tag, NETDATA_THREAD_TAG_MAX, "REPLAY[%zu]", i + 2);
1916 replication_globals.main_thread.threads_ptrs[i] = mallocz(sizeof(ND_THREAD *));
1917 __atomic_add_fetch(&replication_buffers_allocated, sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1918 replication_globals.main_thread.threads_ptrs[i] = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE,
src/streaming/stream-conf.c
+2
@@ -184,6 +184,8 @@ void stream_conf_load() {
184 nd_log_daemon(NDLP_ERR, "STREAM [send]: cannot enable sending thread - information is missing.");
185 stream_send.enabled = false;
186 }
187 +
188 + stream_conf_is_parent(true);
189 }
190
191 bool stream_conf_is_parent(bool recheck) {
src/streaming/stream-path.c
+2 -2
@@ -303,8 +303,8 @@ static bool parse_single_path(json_object *jobj, const char *path, STREAM_PATH *
303 JSONC_PARSE_UINT64_OR_ERROR_AND_RETURN(jobj, path, "first_time_t", p->first_time_t, error, true);
304 JSONC_PARSE_INT64_OR_ERROR_AND_RETURN(jobj, path, "start_time", p->start_time_ms, error, true);
305 JSONC_PARSE_INT64_OR_ERROR_AND_RETURN(jobj, path, "shutdown_time", p->shutdown_time_ms, error, true);
306 - JSONC_PARSE_ARRAY_OF_TXT2BITMAP_OR_ERROR_AND_RETURN(jobj, path, "flags", STREAM_PATH_FLAGS_2id_one, p->flags, error, true);
307 - JSONC_PARSE_ARRAY_OF_TXT2BITMAP_OR_ERROR_AND_RETURN(jobj, path, "capabilities", stream_capabilities_parse_one, p->capabilities, error, true);
306 + JSONC_PARSE_ARRAY_OF_TXT2BITMAP_OR_ERROR_AND_RETURN(jobj, path, "flags", STREAM_PATH_FLAGS_2id_one, p->flags, error, false);
307 + JSONC_PARSE_ARRAY_OF_TXT2BITMAP_OR_ERROR_AND_RETURN(jobj, path, "capabilities", stream_capabilities_parse_one, p->capabilities, error, false);
308
309 if(!p->hostname) {
310 buffer_strcat(error, "hostname cannot be empty");
src/streaming/stream-receiver.c
+12 -5
@@ -395,11 +395,18 @@ static void stream_receive_log_database_gap(struct receiver_state *rpt) {
395 if(now < last_db_entry)
396 last_db_entry = now;
397
398 - char buf[128];
399 - duration_snprintf(buf, sizeof(buf), now - last_db_entry, "s", true);
400 - nd_log(NDLS_DAEMON, NDLP_NOTICE,
401 - "STREAM RCV '%s' [from [%s]:%s]: node connected; last sample in the database %s ago",
402 - rrdhost_hostname(host), rpt->client_ip, rpt->client_port, buf);
398 + if(!last_db_entry) {
399 + nd_log(NDLS_DAEMON, NDLP_NOTICE,
400 + "STREAM RCV '%s' [from [%s]:%s]: node connected; for the first time!",
401 + rrdhost_hostname(host), rpt->client_ip, rpt->client_port);
402 + }
403 + else {
404 + char buf[128];
405 + duration_snprintf(buf, sizeof(buf), now - last_db_entry, "s", true);
406 + nd_log(NDLS_DAEMON, NDLP_NOTICE,
407 + "STREAM RCV '%s' [from [%s]:%s]: node connected; last sample in the database %s ago",
408 + rrdhost_hostname(host), rpt->client_ip, rpt->client_port, buf);
409 + }
410 }
411
412 void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct receiver_state *rpt) {
src/streaming/stream-thread.c
+1 -1
@@ -637,7 +637,7 @@ static struct stream_thread *stream_thread_get_unsafe(RRDHOST *host) {
637 return host->stream.thread;
638
639 if(!stream_thread_globals.assign.cores) {
640 - stream_thread_globals.assign.cores = get_netdata_cpus() - 1;
640 + stream_thread_globals.assign.cores = netdata_conf_cpus() - 1;
641 if(stream_thread_globals.assign.cores < 4)
642 stream_thread_globals.assign.cores = 4;
643 else if(stream_thread_globals.assign.cores > STREAM_MAX_THREADS)
src/web/api/queries/backfill.c
+1 -1
@@ -188,7 +188,7 @@ void *backfill_thread(void *ptr) {
188 backfill_globals.running = true;
189 spinlock_unlock(&backfill_globals.spinlock);
190
191 - size_t threads = get_netdata_cpus() / 2;
191 + size_t threads = netdata_conf_cpus() / 2;
192 if(threads < 2) threads = 2;
193 if(threads > 16) threads = 16;
194 ND_THREAD *th[threads - 1];