#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #if defined(__clang__) #pragma clang diagnostic push #pragma clang diagnostic ignored "-Wdeprecated" #endif #include #include #include #include #include // IWYU pragma: keep #include namespace xrpl { static auto gCheckpointPageCount = 1000; namespace detail { std::string getSociSqliteInit(std::string const& name, std::string const& dir, std::string const& ext) { if (name.empty()) { Throw( "Sqlite databases must specify a dir and a name. Name: " + name + " Dir: " + dir); } boost::filesystem::path file(dir); if (is_directory(file)) file /= name + ext; return file.string(); } std::string getSociInit(BasicConfig const& config, std::string const& dbName) { auto const& section = config.section(Sections::kSqdb); auto const backendName = get(section, Keys::kBackend, "sqlite"); if (backendName != "sqlite") Throw("Unsupported soci backend: " + backendName); auto const path = config.legacy(Sections::kDatabasePath); auto const ext = dbName == "validators" || dbName == "peerfinder" ? ".sqlite" : ".db"; return detail::getSociSqliteInit(dbName, path, ext); } } // namespace detail DBConfig::DBConfig(std::string dbPath) : connectionString_(std::move(dbPath)) { } DBConfig::DBConfig(BasicConfig const& config, std::string const& dbName) : DBConfig(detail::getSociInit(config, dbName)) { } std::string DBConfig::connectionString() const { return connectionString_; } void DBConfig::open(soci::session& s) const { s.open(soci::sqlite3, connectionString()); } void open(soci::session& s, BasicConfig const& config, std::string const& dbName) { DBConfig(config, dbName).open(s); } void open(soci::session& s, std::string const& beName, std::string const& connectionString) { if (beName == "sqlite") { s.open(soci::sqlite3, connectionString); } else { Throw("Unsupported soci backend: " + beName); } } static sqlite_api::sqlite3* getConnection(soci::session& s) { sqlite_api::sqlite3* result = nullptr; // NOLINT(misc-const-correctness) auto be = s.get_backend(); if (auto b = dynamic_cast(be)) result = b->conn_; if (result == nullptr) Throw("Didn't get a database connection."); return result; } std::uint32_t getKBUsedAll(soci::session& s) { if (getConnection(s) == nullptr) Throw("No connection found."); return static_cast(sqlite_api::sqlite3_memory_used() / kilobytes(1)); } std::uint32_t getKBUsedDB(soci::session& s) { // This function will have to be customized when other backends are added if (auto conn = getConnection(s)) { int cur = 0, hiw = 0; sqlite_api::sqlite3_db_status(conn, SQLITE_DBSTATUS_CACHE_USED, &cur, &hiw, 0); return cur / kilobytes(1); } Throw(""); return 0; // Silence compiler warning. } void convert(soci::blob& from, std::vector& to) { to.resize(from.get_len()); if (to.empty()) return; from.read(0, reinterpret_cast(&to[0]), from.get_len()); } void convert(soci::blob& from, std::string& to) { std::vector tmp; convert(from, tmp); to.assign(tmp.begin(), tmp.end()); } void convert(std::vector const& from, soci::blob& to) { if (!from.empty()) { to.write(0, reinterpret_cast(&from[0]), from.size()); } else { to.trim(0); } } void convert(std::string const& from, soci::blob& to) { if (!from.empty()) { to.write(0, from.data(), from.size()); } else { to.trim(0); } } namespace { /** Run a thread to checkpoint the write ahead log (wal) for the given soci::session every 1000 pages. This is only implemented for sqlite databases. Note: According to: https://www.sqlite.org/wal.html#ckpt this is the default behavior of sqlite. We may be able to remove this class. */ class WALCheckpointer : public Checkpointer { public: WALCheckpointer( std::uintptr_t id, std::weak_ptr session, JobQueue& q, ServiceRegistry& registry) : id_(id) , session_(std::move(session)) , jobQueue_(q) , j_(registry.getJournal("WALCheckpointer")) { if (auto [conn, keepAlive] = getConnection(); conn) { (void)keepAlive; sqlite_api::sqlite3_wal_hook(conn, &sqliteWALHook, reinterpret_cast(id_)); } } std::pair> getConnection() const { if (auto p = session_.lock()) { return {xrpl::getConnection(*p), p}; } return {nullptr, std::shared_ptr{}}; } std::uintptr_t id() const override { return id_; } ~WALCheckpointer() override = default; void schedule() override { { std::scoped_lock const lock(mutex_); if (running_) return; running_ = true; } // If the Job is not added to the JobQueue then we're not running_. if (!jobQueue_.addJob( JtWal, "WAL", // If the owning DatabaseCon is destroyed, no need to checkpoint // or keep the checkpointer alive so use a weak_ptr to this. // There is a separate check in `checkpoint` for a valid // connection in the rare case when the DatabaseCon is destroyed // after locking this weak_ptr [wp = std::weak_ptr{shared_from_this()}]() { if (auto self = wp.lock()) self->checkpoint(); })) { std::scoped_lock const lock(mutex_); running_ = false; } } void checkpoint() override { auto [conn, keepAlive] = getConnection(); (void)keepAlive; if (conn == nullptr) return; int log = 0, ckpt = 0; int const ret = sqlite_api::sqlite3_wal_checkpoint_v2( conn, nullptr, SQLITE_CHECKPOINT_PASSIVE, &log, &ckpt); auto fname = sqlite_api::sqlite3_db_filename(conn, "main"); if (ret != SQLITE_OK) { auto jm = (ret == SQLITE_LOCKED) ? j_.trace() : j_.warn(); JLOG(jm) << "WAL(" << fname << "): error " << ret; } else { JLOG(j_.trace()) << "WAL(" << fname << "): frames=" << log << ", written=" << ckpt; } std::scoped_lock const lock(mutex_); running_ = false; } protected: std::uintptr_t const id_; // session is owned by the DatabaseCon parent that holds the checkpointer. // It is possible (though rare) for the DatabaseCon class to be destroyed // before the checkpointer. std::weak_ptr session_; std::mutex mutex_; JobQueue& jobQueue_; bool running_ = false; beast::Journal const j_; static int sqliteWALHook(void* cpId, sqlite_api::sqlite3* conn, char const* dbName, int walSize) { if (walSize >= gCheckpointPageCount) { if (auto checkpointer = checkpointerFromId(reinterpret_cast(cpId))) { checkpointer->schedule(); } else { sqlite_api::sqlite3_wal_hook(conn, nullptr, nullptr); } } return SQLITE_OK; } }; } // namespace std::shared_ptr makeCheckpointer( std::uintptr_t id, std::weak_ptr session, JobQueue& queue, ServiceRegistry& registry) { return std::make_shared(id, std::move(session), queue, registry); } } // namespace xrpl #if defined(__clang__) #pragma clang diagnostic pop #endif