24057299873

Committed 07 Apr 2026 12:01AM UTC coverage: 52.076% (+0.8%) from 51.228%

Build # 24057299873

Build Type

push

github

Committed by

rayandrew

Commit Message

feat(rocksdb): migrate SQLite indexing to RocksDB

Replace SQLite-backed indexing and provenance storage with RocksDB-backed stores.

  Key changes:
  - add RocksDB async/database/db-manager/filesystem/key-codec layers
  - migrate index and provenance databases from SQLite to RocksDB
  - update index builder, trace reader, reorganize, view, stats, and comparator paths for
  RocksDB
  - harden transaction atomicity and rollback behavior with TransactionScope
  - add iterator status checking for prefix scans
  - harden gzip/tar indexer cache state and metadata handling
  - capture executor context in RocksDB awaitables
  - clean up failed RocksDB open paths and manager lifecycle behavior
  - vendor CPM 0.42.1 and update CI/build integration
  - refresh docs, Python bindings, and C++/Python test coverage for the new backend

  Validation:
  - full test suite passed
  - Ubuntu 22.04 Docker run passed
  - focused RocksDB/indexer regression tests passed.

Coverage Stats

24097 of 59624 branches covered (40.41%)

Branch coverage included in aggregate %.

2516 of 3144 new or added lines in 75 files covered. (80.03%)

72 existing lines in 15 files now uncovered.

20858 of 26701 relevant lines covered (78.12%)

14113.43 hits per line

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

57.02

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

llnl / dftracer-utils / 24057299873

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