29787659139

Committed 20 Jul 2026 11:34PM UTC coverage: 51.578% (-1.1%) from 52.66%

Build # 29787659139

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge ee805adeb into 06bc84ec9

Pull Request Pull Request #99: Support CM time_metric (NS/MS/SEC/US) across reader and viz

Coverage Stats

34832 of 86278 branches covered (40.37%)

Branch coverage included in aggregate %.

1034 of 1319 new or added lines in 30 files covered. (78.39%)

5193 existing lines in 197 files now uncovered.

35349 of 49791 relevant lines covered (70.99%)

9765.54 hits per line

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

48.65

/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) {	628✔
17	FILE *f = std::fopen(path, "r");	628✔
18	if (!f) return 0;	628!
19	char buf[64];
UNCOV 20	std::size_t n = std::fread(buf, 1, sizeof(buf) - 1, f);	×
UNCOV 21	std::fclose(f);	×
UNCOV 22	if (n == 0) return 0;	×
UNCOV 23	buf[n] = '\0';	×
UNCOV 24	if (std::strncmp(buf, "max", 3) == 0) return 0;	×
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	}	628✔
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) {	314✔
35	FILE *f = std::fopen("/proc/self/cgroup", "r");	314✔
36	if (!f) return;	314!
37	char line[1024];
UNCOV 38	while (std::fgets(line, sizeof(line), f)) {	×
UNCOV 39	std::size_t n = std::strlen(line);	×
UNCOV 40	while (n > 0 && (line[n - 1] == '\n' \|\| line[n - 1] == '\r')) {	×
UNCOV 41	line[--n] = '\0';	×
42	}
UNCOV 43	if (n >= 3 && line[0] == '0' && line[1] == ':' && line[2] == ':') {	×
UNCOV 44	v2_path = line + 3;	×
UNCOV 45	continue;	×
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);	×
UNCOV 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	}	×
UNCOV 65	std::fclose(f);	×
66	}	314✔
67
68	static std::size_t cgroup_v2_limit_at(const std::string &cg_path) {	157✔
69	std::string base = "/sys/fs/cgroup" + cg_path;	157✔
70	std::string dir = base;	157!
71	while (true) {	314✔
72	std::size_t max_mem =	314✔
73	read_size_from_file((dir + "/memory.max").c_str());	314!
74	if (max_mem > 0 && max_mem < CGROUP_LIMIT_SENTINEL) {	314!
UNCOV 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;	314✔
81	std::size_t slash = dir.find_last_of('/');	157✔
82	if (slash == std::string::npos \|\| slash < std::strlen("/sys/fs/cgroup"))	157!
UNCOV 83	break;	×
84	dir.resize(slash);	157!
85	}
86	return 0;	157✔
87	}	157✔
88
89	static std::size_t cgroup_v1_limit_at(const std::string &cg_path) {	157✔
90	std::string base = "/sys/fs/cgroup/memory" + cg_path;	157✔
91	std::string dir = base;	157!
92	while (true) {	314✔
93	std::size_t limit =	314✔
94	read_size_from_file((dir + "/memory.limit_in_bytes").c_str());	314!
95	if (limit > 0 && limit < CGROUP_LIMIT_SENTINEL) {	314!
UNCOV 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;	314✔
102	std::size_t slash = dir.find_last_of('/');	157✔
103	if (slash == std::string::npos \|\|	157!
104	slash < std::strlen("/sys/fs/cgroup/memory"))	157✔
UNCOV 105	break;	×
106	dir.resize(slash);	157!
107	}
108	return 0;	157✔
109	}	157✔
110
111	static std::size_t try_cgroups_v2() {	157✔
112	std::string v2_path, v1_path;	157✔
113	read_self_cgroup_paths(v2_path, v1_path);	157!
114	if (v2_path.empty()) v2_path = "/";	157!
115	return cgroup_v2_limit_at(v2_path);	157!
116	}	157✔
117
118	static std::size_t try_cgroups_v1() {	157✔
119	std::string v2_path, v1_path;	157✔
120	read_self_cgroup_paths(v2_path, v1_path);	157!
121	if (v1_path.empty()) v1_path = "/";	157!
122	return cgroup_v1_limit_at(v1_path);	157!
123	}	157✔
124
125	static std::size_t try_proc_meminfo() {	157✔
126	FILE *f = std::fopen("/proc/meminfo", "r");	157✔
127	if (!f) return 0;	157!
128	char line[256];
UNCOV 129	while (std::fgets(line, sizeof(line), f)) {	×
UNCOV 130	if (std::strncmp(line, "MemAvailable:", 13) == 0) {	×
UNCOV 131	char *p = line + 13;	×
UNCOV 132	while (*p == ' ') ++p;	×
UNCOV 133	char *end = nullptr;	×
UNCOV 134	unsigned long long val = std::strtoull(p, &end, 10);	×
UNCOV 135	std::fclose(f);	×
UNCOV 136	return static_cast<std::size_t>(val) * 1024;	×
137	}
138	}
139	std::fclose(f);	×
140	return 0;	×
141	}	157✔
142
143	std::size_t detect_available_memory() {	157✔
144	std::size_t avail = try_cgroups_v2();	157✔
145	if (avail > 0) return avail;	157!
146	avail = try_cgroups_v1();	157✔
147	if (avail > 0) return avail;	157!
148	avail = try_proc_meminfo();	157✔
149	if (avail > 0) return avail;	157!
150	return FALLBACK_AVAILABLE_BYTES;	157✔
151	}	157✔
152
153	std::size_t compute_memory_budget(std::size_t user_override_bytes) {	157✔
154	if (user_override_bytes > 0) return user_override_bytes;	157!
155	std::size_t avail = detect_available_memory();	157✔
156	std::size_t budget = avail * DEFAULT_MEMORY_BUDGET_FRACTION_PERCENT / 100;	157✔
157	return std::max(budget, MIN_MEMORY_BUDGET_BYTES);	157✔
158	}	157✔
159
160	std::size_t compute_channel_capacity(std::size_t memory_budget_bytes,	146✔
161	std::size_t estimated_batch_bytes,
162	std::size_t num_workers) {
163	std::size_t from_budget =	146✔
164	memory_budget_bytes / std::max(estimated_batch_bytes, std::size_t(1));	146✔
165	std::size_t minimum = std::max(num_workers * 2, std::size_t(4));	146✔
166	return std::max(from_budget, minimum);	146✔
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 =	4✔
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	}	4✔
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 / 29787659139

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