DatabaseCon.cpp

#include <xrpl/rdb/DatabaseCon.h>
 
#include <xrpl/basics/contract.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/rdb/SociDB.h>
 
#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
 
namespace xrpl {
 
class CheckpointersCollection
{
    std::uintptr_t nextId_{0};
    // Mutex protects the CheckpointersCollection
    std::mutex mutex_;
    // Each checkpointer is given a unique id. All the checkpointers that are
    // part of a DatabaseCon are part of this collection. When the DatabaseCon
    // is destroyed, its checkpointer is removed from the collection
    std::unordered_map<std::uintptr_t, std::shared_ptr<Checkpointer>> checkpointers_;
 
public:
    std::shared_ptr<Checkpointer>
    fromId(std::uintptr_t id)
    {
        std::scoped_lock const l{mutex_};
        auto it = checkpointers_.find(id);
        if (it != checkpointers_.end())
            return it->second;
        return nullptr;
    }
 
    void
    erase(std::uintptr_t id)
    {
        std::scoped_lock const lock{mutex_};
        checkpointers_.erase(id);
    }
 
    std::shared_ptr<Checkpointer>
    create(
        std::shared_ptr<soci::session> const& session,
        JobQueue& jobQueue,
        ServiceRegistry& registry)
    {
        std::scoped_lock const lock{mutex_};
        auto const id = nextId_++;
        auto const r = makeCheckpointer(id, session, jobQueue, registry);
        checkpointers_[id] = r;
        return r;
    }
};
 
CheckpointersCollection gCheckpointers;
 
std::shared_ptr<Checkpointer>
checkpointerFromId(std::uintptr_t id)
{
    return gCheckpointers.fromId(id);
}
 
DatabaseCon::~DatabaseCon()
{
    if (checkpointer_)
    {
        gCheckpointers.erase(checkpointer_->id());
 
        std::weak_ptr<Checkpointer> const wk(checkpointer_);
        checkpointer_.reset();
 
        // The references to our Checkpointer held by 'checkpointer_' and
        // 'checkpointers' have been removed, so if the use count is nonzero, a
        // checkpoint is currently in progress. Wait for it to end, otherwise
        // creating a new DatabaseCon to the same database may fail due to the
        // database being locked by our (now old) Checkpointer.
        while (wk.use_count() != 0)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    }
}
 
std::unique_ptr<std::vector<std::string> const> DatabaseCon::Setup::globalPragma;
 
void
DatabaseCon::setupCheckpointing(JobQueue* q, ServiceRegistry& registry)
{
    if (q == nullptr)
        Throw<std::logic_error>("No JobQueue");
    checkpointer_ = gCheckpointers.create(session_, *q, registry);
}
 
}  // namespace xrpl