BatchWriter.cpp

#include <xrpl/nodestore/detail/BatchWriter.h>
 
namespace xrpl {
namespace NodeStore {
 
BatchWriter::BatchWriter(Callback& callback, Scheduler& scheduler)
    : m_callback(callback), m_scheduler(scheduler), mWriteLoad(0), mWritePending(false)
{
    mWriteSet.reserve(batchWritePreallocationSize);
}
 
BatchWriter::~BatchWriter()
{
    waitForWriting();
}
 
void
BatchWriter::store(std::shared_ptr<NodeObject> const& object)
{
    std::unique_lock<decltype(mWriteMutex)> sl(mWriteMutex);
 
    // If the batch has reached its limit, we wait
    // until the batch writer is finished
    while (mWriteSet.size() >= batchWriteLimitSize)
        mWriteCondition.wait(sl);
 
    mWriteSet.push_back(object);
 
    if (!mWritePending)
    {
        mWritePending = true;
 
        m_scheduler.scheduleTask(*this);
    }
}
 
int
BatchWriter::getWriteLoad()
{
    std::lock_guard sl(mWriteMutex);
 
    return std::max(mWriteLoad, static_cast<int>(mWriteSet.size()));
}
 
void
BatchWriter::performScheduledTask()
{
    writeBatch();
}
 
void
BatchWriter::writeBatch()
{
    for (;;)
    {
        std::vector<std::shared_ptr<NodeObject>> set;
 
        set.reserve(batchWritePreallocationSize);
 
        {
            std::lock_guard sl(mWriteMutex);
 
            mWriteSet.swap(set);
            XRPL_ASSERT(mWriteSet.empty(), "xrpl::NodeStore::BatchWriter::writeBatch : writes not set");
            mWriteLoad = set.size();
 
            if (set.empty())
            {
                mWritePending = false;
                mWriteCondition.notify_all();
 
                // VFALCO NOTE Fix this function to not return from the middle
                return;
            }
        }
 
        BatchWriteReport report;
        report.writeCount = set.size();
        auto const before = std::chrono::steady_clock::now();
 
        m_callback.writeBatch(set);
 
        report.elapsed =
            std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - before);
 
        m_scheduler.onBatchWrite(report);
    }
}
 
void
BatchWriter::waitForWriting()
{
    std::unique_lock<decltype(mWriteMutex)> sl(mWriteMutex);
 
    while (mWritePending)
        mWriteCondition.wait(sl);
}
 
}  // namespace NodeStore
}  // namespace xrpl