TimeoutCounter.cpp

#include <xrpld/app/ledger/detail/TimeoutCounter.h>
 
#include <xrpld/app/main/Application.h>
 
#include <xrpl/basics/Log.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/core/JobQueue.h>
 
#include <boost/asio/error.hpp>
#include <boost/system/detail/error_code.hpp>
 
#include <chrono>
#include <utility>
 
namespace xrpl {
 
using namespace std::chrono_literals;
 
TimeoutCounter::TimeoutCounter(
    Application& app,
    UInt256 const& hash,
    std::chrono::milliseconds interval,
    QueueJobParameter&& jobParameter,
    beast::Journal journal)
    : app_(app)
    , journal_(journal)
    , hash_(hash)
    , timerInterval_(interval)
    , queueJobParameter_(std::move(jobParameter))
    , timer_(app_.getIOContext())
{
    XRPL_ASSERT(
        (timerInterval_ > 10ms) && (timerInterval_ < 30s),
        "xrpl::TimeoutCounter::TimeoutCounter : interval input inside range");
}
 
void
TimeoutCounter::setTimer(ScopedLockType& sl)
{
    if (isDone())
        return;
    timer_.expires_after(timerInterval_);
    timer_.async_wait([wptr = pmDowncast()](boost::system::error_code const& ec) {
        if (ec == boost::asio::error::operation_aborted)
            return;
 
        if (auto ptr = wptr.lock())
        {
            ScopedLockType sl(ptr->mtx_);
            ptr->queueJob(sl);
        }
    });
}
 
void
TimeoutCounter::queueJob(ScopedLockType& sl)
{
    if (isDone())
        return;
    if (queueJobParameter_.jobLimit &&
        app_.getJobQueue().getJobCountTotal(queueJobParameter_.jobType) >=
            queueJobParameter_.jobLimit)
    {
        JLOG(journal_.debug()) << "Deferring " << queueJobParameter_.jobName
                               << " timer due to load";
        setTimer(sl);
        return;
    }
 
    app_.getJobQueue().addJob(
        queueJobParameter_.jobType, queueJobParameter_.jobName, [wptr = pmDowncast()]() {
            if (auto sptr = wptr.lock(); sptr)
                sptr->invokeOnTimer();
        });
}
 
void
TimeoutCounter::invokeOnTimer()
{
    ScopedLockType sl(mtx_);
 
    if (isDone())
        return;
 
    if (!progress_)
    {
        ++timeouts_;
        JLOG(journal_.debug()) << "Timeout(" << timeouts_ << ") "
                               << " acquiring " << hash_;
        onTimer(false, sl);
    }
    else
    {
        progress_ = false;
        onTimer(true, sl);
    }
 
    if (!isDone())
        setTimer(sl);
}
 
void
TimeoutCounter::cancel()
{
    ScopedLockType const sl(mtx_);
    if (!isDone())
    {
        failed_ = true;
        JLOG(journal_.info()) << "Cancel " << hash_;
    }
}
 
}  // namespace xrpl