26195612357

Committed 20 May 2026 11:19PM UTC coverage: 49.859% (-2.3%) from 52.2%

Build # 26195612357

Build Type

push

github

Committed by

hariharan-devarajan

Commit Message

feat(aggregator): improve system metrics scanning and persistence error handling

Coverage Stats

16041 of 43831 branches covered (36.6%)

Branch coverage included in aggregate %.

6 of 17 new or added lines in 2 files covered. (35.29%)

1072 existing lines in 104 files now uncovered.

21423 of 31309 relevant lines covered (68.42%)

13054.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.09

/src/dftracer/utils/core/common/memory_budget.cpp

#include <dftracer/utils/core/common/memory_budget.h>

#include <algorithm>
#include <cstddef>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <string>
#include <vector>

namespace dftracer::utils {

static constexpr std::size_t FALLBACK_AVAILABLE_BYTES =
    1ULL * 1024 * 1024 * 1024;

static std::size_t read_size_from_file(const char *path) {
    FILE *f = std::fopen(path, "r");
    if (!f) return 0;
    char buf[64];
    std::size_t n = std::fread(buf, 1, sizeof(buf) - 1, f);
    std::fclose(f);
    if (n == 0) return 0;
    buf[n] = '\0';
    if (std::strncmp(buf, "max", 3) == 0) return 0;
    char *end = nullptr;
    unsigned long long val = std::strtoull(buf, &end, 10);
    if (end == buf) return 0;
    return static_cast<std::size_t>(val);
}

static constexpr std::size_t CGROUP_LIMIT_SENTINEL =
    1ULL * 1024 * 1024 * 1024 * 1024;

static void read_self_cgroup_paths(std::string &v2_path, std::string &v1_path) {
    FILE *f = std::fopen("/proc/self/cgroup", "r");
    if (!f) return;
    char line[1024];
    while (std::fgets(line, sizeof(line), f)) {
        std::size_t n = std::strlen(line);
        while (n > 0 && (line[n - 1] == '\n' || line[n - 1] == '\r')) {
            line[--n] = '\0';
        }
        if (n >= 3 && line[0] == '0' && line[1] == ':' && line[2] == ':') {
            v2_path = line + 3;
            continue;
        }
        char *first = std::strchr(line, ':');
        if (!first) continue;
        char *second = std::strchr(first + 1, ':');
        if (!second) continue;
        std::string controllers(first + 1, second - first - 1);
        std::size_t start = 0;
        while (start <= controllers.size()) {
            std::size_t comma = controllers.find(',', start);
            std::size_t end =
                (comma == std::string::npos) ? controllers.size() : comma;
            if (controllers.compare(start, end - start, "memory") == 0) {
                v1_path = second + 1;
                break;
            }
            if (comma == std::string::npos) break;
            start = comma + 1;
        }
    }
    std::fclose(f);
}

static std::size_t cgroup_v2_limit_at(const std::string &cg_path) {
    std::string base = "/sys/fs/cgroup" + cg_path;
    std::string dir = base;
    while (true) {
        std::size_t max_mem =
            read_size_from_file((dir + "/memory.max").c_str());
        if (max_mem > 0 && max_mem < CGROUP_LIMIT_SENTINEL) {
            std::size_t current =
                read_size_from_file((dir + "/memory.current").c_str());
            if (current >= max_mem) return 0;
            return max_mem - current;
        }
        if (dir.size() <= std::strlen("/sys/fs/cgroup")) break;
        std::size_t slash = dir.find_last_of('/');
        if (slash == std::string::npos || slash < std::strlen("/sys/fs/cgroup"))
            break;
        dir.resize(slash);
    }
    return 0;
}

static std::size_t cgroup_v1_limit_at(const std::string &cg_path) {
    std::string base = "/sys/fs/cgroup/memory" + cg_path;
    std::string dir = base;
    while (true) {
        std::size_t limit =
            read_size_from_file((dir + "/memory.limit_in_bytes").c_str());
        if (limit > 0 && limit < CGROUP_LIMIT_SENTINEL) {
            std::size_t usage =
                read_size_from_file((dir + "/memory.usage_in_bytes").c_str());
            if (usage >= limit) return 0;
            return limit - usage;
        }
        if (dir.size() <= std::strlen("/sys/fs/cgroup/memory")) break;
        std::size_t slash = dir.find_last_of('/');
        if (slash == std::string::npos ||
            slash < std::strlen("/sys/fs/cgroup/memory"))
            break;
        dir.resize(slash);
    }
    return 0;
}

static std::size_t try_cgroups_v2() {
    std::string v2_path, v1_path;
    read_self_cgroup_paths(v2_path, v1_path);
    if (v2_path.empty()) v2_path = "/";
    return cgroup_v2_limit_at(v2_path);
}

static std::size_t try_cgroups_v1() {
    std::string v2_path, v1_path;
    read_self_cgroup_paths(v2_path, v1_path);
    if (v1_path.empty()) v1_path = "/";
    return cgroup_v1_limit_at(v1_path);
}

static std::size_t try_proc_meminfo() {
    FILE *f = std::fopen("/proc/meminfo", "r");
    if (!f) return 0;
    char line[256];
    while (std::fgets(line, sizeof(line), f)) {
        if (std::strncmp(line, "MemAvailable:", 13) == 0) {
            char *p = line + 13;
            while (*p == ' ') ++p;
            char *end = nullptr;
            unsigned long long val = std::strtoull(p, &end, 10);
            std::fclose(f);
            return static_cast<std::size_t>(val) * 1024;
        }
    }
    std::fclose(f);
    return 0;
}

std::size_t detect_available_memory() {
    std::size_t avail = try_cgroups_v2();
    if (avail > 0) return avail;
    avail = try_cgroups_v1();
    if (avail > 0) return avail;
    avail = try_proc_meminfo();
    if (avail > 0) return avail;
    return FALLBACK_AVAILABLE_BYTES;
}

std::size_t compute_memory_budget(std::size_t user_override_bytes) {
    if (user_override_bytes > 0) return user_override_bytes;
    std::size_t avail = detect_available_memory();
    std::size_t budget = avail * DEFAULT_MEMORY_BUDGET_FRACTION_PERCENT / 100;
    return std::max(budget, MIN_MEMORY_BUDGET_BYTES);
}

std::size_t compute_channel_capacity(std::size_t memory_budget_bytes,
                                     std::size_t estimated_batch_bytes,
                                     std::size_t num_workers) {
    std::size_t from_budget =
        memory_budget_bytes / std::max(estimated_batch_bytes, std::size_t(1));
    std::size_t minimum = std::max(num_workers * 2, std::size_t(4));
    return std::max(from_budget, minimum);
}

std::size_t compute_file_batch_size(std::size_t memory_budget_bytes,
                                    std::size_t estimated_file_bytes,
                                    std::size_t min_files) {
    std::size_t from_budget =
        memory_budget_bytes / std::max(estimated_file_bytes, std::size_t(1));
    return std::max(from_budget, std::max(min_files, std::size_t(1)));
}

std::size_t estimate_per_file_bytes(const std::vector<std::size_t> &file_sizes,
                                    std::size_t user_override_bytes) {
    if (user_override_bytes > 0) return user_override_bytes;
    if (file_sizes.empty()) return MIN_PER_FILE_PEAK_BYTES;

    const std::size_t total = file_sizes.size();
    const std::size_t sample_count = std::min(total, PER_FILE_SAMPLE_LIMIT);
    const std::size_t stride = std::max(total / sample_count, std::size_t(1));

    std::vector<std::size_t> sizes;
    sizes.reserve(sample_count);
    for (std::size_t i = 0; i < total && sizes.size() < sample_count;
         i += stride) {
        if (file_sizes[i] > 0) sizes.push_back(file_sizes[i]);
    }

    if (sizes.empty()) return MIN_PER_FILE_PEAK_BYTES;

    std::size_t idx = (sizes.size() * 95) / 100;
    if (idx >= sizes.size()) idx = sizes.size() - 1;
    std::nth_element(sizes.begin(), sizes.begin() + idx, sizes.end());
    const std::size_t p95 = sizes[idx];

    std::size_t estimate = p95 * PER_FILE_EXPANSION_FACTOR;
    estimate = std::max(estimate, MIN_PER_FILE_PEAK_BYTES);
    estimate = std::min(estimate, MAX_PER_FILE_PEAK_BYTES);
    return estimate;
}

}  // namespace dftracer::utils

1	#include <dftracer/utils/core/common/memory_budget.h>
2
3	#include <algorithm>
4	#include <cstddef>
5	#include <cstdio>
6	#include <cstdlib>
7	#include <cstring>
8	#include <string>
9	#include <vector>
10
11	namespace dftracer::utils {
12
13	static constexpr std::size_t FALLBACK_AVAILABLE_BYTES =
14	1ULL * 1024 * 1024 * 1024;
15
16	static std::size_t read_size_from_file(const char *path) {	715✔
17	FILE *f = std::fopen(path, "r");	715!
18	if (!f) return 0;	715✔
19	char buf[64];
20	std::size_t n = std::fread(buf, 1, sizeof(buf) - 1, f);	286!
21	std::fclose(f);	286!
22	if (n == 0) return 0;	286!
23	buf[n] = '\0';	286✔
24	if (std::strncmp(buf, "max", 3) == 0) return 0;	286!
25	char *end = nullptr;	×
26	unsigned long long val = std::strtoull(buf, &end, 10);	×
27	if (end == buf) return 0;	×
28	return static_cast<std::size_t>(val);	×
29	}
30
31	static constexpr std::size_t CGROUP_LIMIT_SENTINEL =
32	1ULL * 1024 * 1024 * 1024 * 1024;
33
34	static void read_self_cgroup_paths(std::string &v2_path, std::string &v1_path) {	286✔
35	FILE *f = std::fopen("/proc/self/cgroup", "r");	286!
36	if (!f) return;	286!
37	char line[1024];
38	while (std::fgets(line, sizeof(line), f)) {	572!
39	std::size_t n = std::strlen(line);	286✔
40	while (n > 0 && (line[n - 1] == '\n' \|\| line[n - 1] == '\r')) {	572!
41	line[--n] = '\0';	286✔
42	}
43	if (n >= 3 && line[0] == '0' && line[1] == ':' && line[2] == ':') {	286!
44	v2_path = line + 3;	286!
45	continue;	286✔
46	}
47	char *first = std::strchr(line, ':');	×
48	if (!first) continue;	×
49	char *second = std::strchr(first + 1, ':');	×
50	if (!second) continue;	×
51	std::string controllers(first + 1, second - first - 1);	×
52	std::size_t start = 0;	×
53	while (start <= controllers.size()) {	×
54	std::size_t comma = controllers.find(',', start);	×
55	std::size_t end =
56	(comma == std::string::npos) ? controllers.size() : comma;	×
57	if (controllers.compare(start, end - start, "memory") == 0) {	×
58	v1_path = second + 1;	×
59	break;	×
60	}
61	if (comma == std::string::npos) break;	×
62	start = comma + 1;	×
63	}
64	}	×
65	std::fclose(f);	286!
66	}
67
68	static std::size_t cgroup_v2_limit_at(const std::string &cg_path) {	143✔
69	std::string base = "/sys/fs/cgroup" + cg_path;	143!
70	std::string dir = base;	143!
71	while (true) {
72	std::size_t max_mem =
73	read_size_from_file((dir + "/memory.max").c_str());	429!
74	if (max_mem > 0 && max_mem < CGROUP_LIMIT_SENTINEL) {	429!
75	std::size_t current =
76	read_size_from_file((dir + "/memory.current").c_str());	×
77	if (current >= max_mem) return 0;	×
78	return max_mem - current;	×
79	}
80	if (dir.size() <= std::strlen("/sys/fs/cgroup")) break;	429✔
81	std::size_t slash = dir.find_last_of('/');	286✔
82	if (slash == std::string::npos \|\| slash < std::strlen("/sys/fs/cgroup"))	286!
83	break;
84	dir.resize(slash);	286!
85	}	286✔
86	return 0;	143✔
87	}	143✔
88
89	static std::size_t cgroup_v1_limit_at(const std::string &cg_path) {	143✔
90	std::string base = "/sys/fs/cgroup/memory" + cg_path;	143!
91	std::string dir = base;	143!
92	while (true) {
93	std::size_t limit =
94	read_size_from_file((dir + "/memory.limit_in_bytes").c_str());	286!
95	if (limit > 0 && limit < CGROUP_LIMIT_SENTINEL) {	286!
96	std::size_t usage =
97	read_size_from_file((dir + "/memory.usage_in_bytes").c_str());	×
98	if (usage >= limit) return 0;	×
99	return limit - usage;	×
100	}
101	if (dir.size() <= std::strlen("/sys/fs/cgroup/memory")) break;	286✔
102	std::size_t slash = dir.find_last_of('/');	143✔
103	if (slash == std::string::npos \|\|	143!
104	slash < std::strlen("/sys/fs/cgroup/memory"))
105	break;
106	dir.resize(slash);	143!
107	}	143✔
108	return 0;	143✔
109	}	143✔
110
111	static std::size_t try_cgroups_v2() {	143✔
112	std::string v2_path, v1_path;	143✔
113	read_self_cgroup_paths(v2_path, v1_path);	143!
114	if (v2_path.empty()) v2_path = "/";	143!
115	return cgroup_v2_limit_at(v2_path);	286!
116	}	143✔
117
118	static std::size_t try_cgroups_v1() {	143✔
119	std::string v2_path, v1_path;	143✔
120	read_self_cgroup_paths(v2_path, v1_path);	143!
121	if (v1_path.empty()) v1_path = "/";	143!
122	return cgroup_v1_limit_at(v1_path);	286!
123	}	143✔
124
125	static std::size_t try_proc_meminfo() {	143✔
126	FILE *f = std::fopen("/proc/meminfo", "r");	143!
127	if (!f) return 0;	143!
128	char line[256];
129	while (std::fgets(line, sizeof(line), f)) {	429!
130	if (std::strncmp(line, "MemAvailable:", 13) == 0) {	429✔
131	char *p = line + 13;	143✔
132	while (*p == ' ') ++p;	572✔
133	char *end = nullptr;	143✔
134	unsigned long long val = std::strtoull(p, &end, 10);	143✔
135	std::fclose(f);	143!
136	return static_cast<std::size_t>(val) * 1024;	143✔
137	}
138	}
139	std::fclose(f);	×
140	return 0;	×
141	}
142
143	std::size_t detect_available_memory() {	143✔
144	std::size_t avail = try_cgroups_v2();	143✔
145	if (avail > 0) return avail;	143!
146	avail = try_cgroups_v1();	143✔
147	if (avail > 0) return avail;	143!
148	avail = try_proc_meminfo();	143✔
149	if (avail > 0) return avail;	143!
UNCOV 150	return FALLBACK_AVAILABLE_BYTES;	×
151	}
152
153	std::size_t compute_memory_budget(std::size_t user_override_bytes) {	143✔
154	if (user_override_bytes > 0) return user_override_bytes;	143!
155	std::size_t avail = detect_available_memory();	143!
156	std::size_t budget = avail * DEFAULT_MEMORY_BUDGET_FRACTION_PERCENT / 100;	143✔
157	return std::max(budget, MIN_MEMORY_BUDGET_BYTES);	143✔
158	}
159
160	std::size_t compute_channel_capacity(std::size_t memory_budget_bytes,	132✔
161	std::size_t estimated_batch_bytes,
162	std::size_t num_workers) {
163	std::size_t from_budget =
164	memory_budget_bytes / std::max(estimated_batch_bytes, std::size_t(1));	132✔
165	std::size_t minimum = std::max(num_workers * 2, std::size_t(4));	132✔
166	return std::max(from_budget, minimum);	132✔
167	}
168
169	std::size_t compute_file_batch_size(std::size_t memory_budget_bytes,	4✔
170	std::size_t estimated_file_bytes,
171	std::size_t min_files) {
172	std::size_t from_budget =
173	memory_budget_bytes / std::max(estimated_file_bytes, std::size_t(1));	4✔
174	return std::max(from_budget, std::max(min_files, std::size_t(1)));	4✔
175	}
176
177	std::size_t estimate_per_file_bytes(const std::vector<std::size_t> &file_sizes,	4✔
178	std::size_t user_override_bytes) {
179	if (user_override_bytes > 0) return user_override_bytes;	4!
180	if (file_sizes.empty()) return MIN_PER_FILE_PEAK_BYTES;	4!
181
182	const std::size_t total = file_sizes.size();	4✔
183	const std::size_t sample_count = std::min(total, PER_FILE_SAMPLE_LIMIT);	4✔
184	const std::size_t stride = std::max(total / sample_count, std::size_t(1));	4✔
185
186	std::vector<std::size_t> sizes;	4✔
187	sizes.reserve(sample_count);	4!
188	for (std::size_t i = 0; i < total && sizes.size() < sample_count;	8!
189	i += stride) {	4✔
190	if (file_sizes[i] > 0) sizes.push_back(file_sizes[i]);	4!
191	}
192
193	if (sizes.empty()) return MIN_PER_FILE_PEAK_BYTES;	4!
194
195	std::size_t idx = (sizes.size() * 95) / 100;	4✔
196	if (idx >= sizes.size()) idx = sizes.size() - 1;	4!
197	std::nth_element(sizes.begin(), sizes.begin() + idx, sizes.end());	4!
198	const std::size_t p95 = sizes[idx];	4✔
199
200	std::size_t estimate = p95 * PER_FILE_EXPANSION_FACTOR;	4✔
201	estimate = std::max(estimate, MIN_PER_FILE_PEAK_BYTES);	4✔
202	estimate = std::min(estimate, MAX_PER_FILE_PEAK_BYTES);	4✔
203	return estimate;	4✔
204	}	4✔
205
206	} // namespace dftracer::utils

llnl / dftracer-utils / 26195612357

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous