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

57.35

/src/dftracer/utils/core/rocksdb/db_manager.cpp

#include <dftracer/utils/core/rocksdb/db_manager.h>

#include <stdexcept>

namespace dftracer::utils::rocksdb {

RocksDBManager& RocksDBManager::instance() {
    static RocksDBManager manager;
    return manager;
}

std::shared_ptr<RocksDatabase> RocksDBManager::get_or_open(
    const std::string& db_path, RocksDatabase::OpenMode open_mode,
    RocksDatabase::CfOptionsOverride cf_override) {
    for (;;) {
        bool needs_upgrade = false;
        bool do_open = false;

        {
            std::unique_lock<std::mutex> lock(mutex_);

            for (;;) {
                if (auto it = databases_.find(db_path);
                    it != databases_.end()) {
                    auto current = it->second.lock();
                    if (!current) {
                        databases_.erase(it);
                        continue;
                    }
                    if (!(current->is_read_only() &&
                          open_mode == RocksDatabase::OpenMode::ReadWrite)) {
                        return current;
                    }

                    if (opening_.contains(db_path)) {
                        cv_.wait(lock,
                                 [&] { return !opening_.contains(db_path); });
                        continue;
                    }

                    if (current.use_count() != 1) {
                        throw std::runtime_error(
                            "Cannot upgrade RocksDB instance at '" + db_path +
                            "' from read-only to read-write while it is still "
                            "in use");
                    }

                    needs_upgrade = true;
                    opening_.insert(db_path);
                    do_open = true;
                    break;
                }

                if (opening_.contains(db_path)) {
                    cv_.wait(lock, [&] { return !opening_.contains(db_path); });
                    continue;
                }

                opening_.insert(db_path);
                do_open = true;
                break;
            }
        }

        if (!do_open) {
            continue;
        }

        std::shared_ptr<RocksDatabase> database;
        try {
            database = std::make_shared<RocksDatabase>();
            if (cf_override) {
                database->set_cf_options_override(std::move(cf_override));
            }
            database->open(db_path, needs_upgrade
                                        ? RocksDatabase::OpenMode::ReadWrite
                                        : open_mode);
        } catch (...) {
            std::lock_guard<std::mutex> lock(mutex_);
            opening_.erase(db_path);
            cv_.notify_all();
            throw;
        }

        {
            std::lock_guard<std::mutex> lock(mutex_);
            auto it = databases_.find(db_path);

            if (it == databases_.end()) {
                databases_[db_path] = database;
                opening_.erase(db_path);
                cv_.notify_all();
                return database;
            }

            auto current = it->second.lock();
            if (!current) {
                databases_[db_path] = database;
                opening_.erase(db_path);
                cv_.notify_all();
                return database;
            }

            if (!(current->is_read_only() &&
                  open_mode == RocksDatabase::OpenMode::ReadWrite)) {
                opening_.erase(db_path);
                cv_.notify_all();
                return current;
            }

            if (current.use_count() != 1) {
                opening_.erase(db_path);
                cv_.notify_all();
                throw std::runtime_error(
                    "Cannot upgrade RocksDB instance at '" + db_path +
                    "' from read-only to read-write while it is still in use");
            }

            databases_[db_path] = database;
            opening_.erase(db_path);
            cv_.notify_all();
            return database;
        }
    }
}

void RocksDBManager::reset(const std::string& db_path) {
    std::unique_lock<std::mutex> lock(mutex_);

    cv_.wait(lock, [&] { return !opening_.contains(db_path); });

    auto it = databases_.find(db_path);
    if (it == databases_.end()) {
        return;
    }

    databases_.erase(it);
}

void RocksDBManager::shutdown() {
    {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [&] { return opening_.empty(); });
        databases_.clear();
    }
}

}  // namespace dftracer::utils::rocksdb

1	#include <dftracer/utils/core/rocksdb/db_manager.h>
2
3	#include <stdexcept>
4
5	namespace dftracer::utils::rocksdb {
6
7	RocksDBManager& RocksDBManager::instance() {	6,523✔
8	static RocksDBManager manager;	6,523✔
9	return manager;	6,516✔
10	}
11
12	std::shared_ptr<RocksDatabase> RocksDBManager::get_or_open(	6,519✔
13	const std::string& db_path, RocksDatabase::OpenMode open_mode,
14	RocksDatabase::CfOptionsOverride cf_override) {
15	for (;;) {
16	bool needs_upgrade = false;	6,519✔
17	bool do_open = false;	6,519✔
18
19	{
20	std::unique_lock<std::mutex> lock(mutex_);	6,519!
21
22	for (;;) {
23	if (auto it = databases_.find(db_path);	11,692!
24	it != databases_.end()) {	11,692✔
25	auto current = it->second.lock();	5,940✔
26	if (!current) {	5,940✔
27	databases_.erase(it);	4,929!
28	continue;	4,929✔
29	}
30	if (!(current->is_read_only() &&	1,011✔
31	open_mode == RocksDatabase::OpenMode::ReadWrite)) {
32	return current;	1,009✔
33	}
34
35	if (opening_.contains(db_path)) {	2!
36	cv_.wait(lock,	×
37	[&] { return !opening_.contains(db_path); });	×
38	continue;	×
UNCOV 39	}	×
40
41	if (current.use_count() != 1) {	2!
42	throw std::runtime_error(	2!
43	"Cannot upgrade RocksDB instance at '" + db_path +	4!
44	"' from read-only to read-write while it is still "
45	"in use");	4✔
46	}
47
48	needs_upgrade = true;	×
49	opening_.insert(db_path);	×
50	do_open = true;	×
51	break;	×
52	}	5,940!
53
54	if (opening_.contains(db_path)) {	5,752!
55	cv_.wait(lock, [&] { return !opening_.contains(db_path); });	811!
56	continue;	235✔
57	}	235✔
58
59	opening_.insert(db_path);	5,517!
60	do_open = true;	5,517✔
61	break;	5,517✔
62	}	5,164✔
63	}	6,528✔
64
65	if (!do_open) {	5,517!
66	continue;	×
67	}
68
69	std::shared_ptr<RocksDatabase> database;	5,517✔
70	try {
71	database = std::make_shared<RocksDatabase>();	5,517!
72	if (cf_override) {	5,517✔
73	database->set_cf_options_override(std::move(cf_override));	5,482!
74	}
75	database->open(db_path, needs_upgrade	5,516!
76	? RocksDatabase::OpenMode::ReadWrite
77	: open_mode);
78	} catch (...) {	22✔
79	std::lock_guard<std::mutex> lock(mutex_);	22!
80	opening_.erase(db_path);	22!
81	cv_.notify_all();	22✔
82	throw;	22✔
83	}	44✔
84
85	{
86	std::lock_guard<std::mutex> lock(mutex_);	5,495!
87	auto it = databases_.find(db_path);	5,495!
88
89	if (it == databases_.end()) {	5,495!
90	databases_[db_path] = database;	5,495!
91	opening_.erase(db_path);	5,495!
92	cv_.notify_all();	5,495✔
93	return database;	5,495✔
94	}
95
96	auto current = it->second.lock();	×
97	if (!current) {	×
98	databases_[db_path] = database;	×
99	opening_.erase(db_path);	×
100	cv_.notify_all();	×
101	return database;	×
102	}
103
104	if (!(current->is_read_only() &&	×
105	open_mode == RocksDatabase::OpenMode::ReadWrite)) {
106	opening_.erase(db_path);	×
107	cv_.notify_all();	×
108	return current;	×
109	}
110
111	if (current.use_count() != 1) {	×
112	opening_.erase(db_path);	×
113	cv_.notify_all();	×
114	throw std::runtime_error(	×
115	"Cannot upgrade RocksDB instance at '" + db_path +	×
UNCOV 116	"' from read-only to read-write while it is still in use");	×
117	}
118
119	databases_[db_path] = database;	×
120	opening_.erase(db_path);	×
121	cv_.notify_all();	×
122	return database;	×
123	}	5,495✔
124	}	5,517✔
125	}
126
127	void RocksDBManager::reset(const std::string& db_path) {	65✔
128	std::unique_lock<std::mutex> lock(mutex_);	65!
129
130	cv_.wait(lock, [&] { return !opening_.contains(db_path); });	130!
131
132	auto it = databases_.find(db_path);	65!
133	if (it == databases_.end()) {	65✔
134	return;	47✔
135	}
136
137	databases_.erase(it);	18!
138	}	65✔
139
140	void RocksDBManager::shutdown() {	1✔
141	{
142	std::unique_lock<std::mutex> lock(mutex_);	1!
143	cv_.wait(lock, [&] { return opening_.empty(); });	2✔
144	databases_.clear();	1✔
145	}	1✔
146	}	1✔
147
148	} // namespace dftracer::utils::rocksdb

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