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

55.61

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

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

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