Files
rippled/src/libxrpl/rdb/DatabaseCon.cpp

101 lines
2.7 KiB
C++

#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