PerfLogImp.h

#ifndef XRPL_BASICS_PERFLOGIMP_H
#define XRPL_BASICS_PERFLOGIMP_H
 
#include <xrpld/perflog/PerfLog.h>
#include <xrpld/rpc/detail/Handler.h>
 
#include <xrpl/beast/utility/Journal.h>
 
#include <boost/asio/ip/host_name.hpp>
 
#include <condition_variable>
#include <cstdint>
#include <fstream>
#include <memory>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
 
namespace ripple {
namespace perf {
 
template <typename T>
struct Locked
{
    T value;
    mutable std::mutex mutex;
 
    Locked() = default;
    Locked(T const& value) : value(value)
    {
    }
    Locked(T&& value) : value(std::move(value))
    {
    }
    Locked(Locked const& rhs) : value(rhs.value)
    {
    }
    Locked(Locked&& rhs) : value(std::move(rhs.value))
    {
    }
};
 
class PerfLogImp : public PerfLog
{
    struct Counters
    {
    public:
        using MethodStart = std::pair<char const*, steady_time_point>;
        struct Rpc
        {
            // Counters for each time a method starts and then either
            // finishes successfully or with an exception.
            std::uint64_t started{0};
            std::uint64_t finished{0};
            std::uint64_t errored{0};
            // Cumulative duration of all finished and errored method calls.
            microseconds duration{0};
        };
 
        struct Jq
        {
            // Counters for each time a job is enqueued, begins to run,
            // finishes.
            std::uint64_t queued{0};
            std::uint64_t started{0};
            std::uint64_t finished{0};
            // Cumulative duration of all jobs' queued and running times.
            microseconds queuedDuration{0};
            microseconds runningDuration{0};
        };
 
        // rpc_ and jq_ do not need mutex protection because all
        // keys and values are created before more threads are started.
        std::unordered_map<std::string, Locked<Rpc>> rpc_;
        std::unordered_map<JobType, Locked<Jq>> jq_;
        std::vector<std::pair<JobType, steady_time_point>> jobs_;
        mutable std::mutex jobsMutex_;
        std::unordered_map<std::uint64_t, MethodStart> methods_;
        mutable std::mutex methodsMutex_;
 
        Counters(std::set<char const*> const& labels, JobTypes const& jobTypes);
        Json::Value
        countersJson() const;
        Json::Value
        currentJson() const;
    };
 
    Setup const setup_;
    Application& app_;
    beast::Journal const j_;
    std::function<void()> const signalStop_;
    Counters counters_{ripple::RPC::getHandlerNames(), JobTypes::instance()};
    std::ofstream logFile_;
    std::thread thread_;
    std::mutex mutex_;
    std::condition_variable cond_;
    system_time_point lastLog_;
    std::string const hostname_{boost::asio::ip::host_name()};
    bool stop_{false};
    bool rotate_{false};
 
    void
    openLog();
    void
    run();
    void
    report();
    void
    rpcEnd(
        std::string const& method,
        std::uint64_t const requestId,
        bool finish);
 
public:
    PerfLogImp(
        Setup const& setup,
        Application& app,
        beast::Journal journal,
        std::function<void()>&& signalStop);
 
    ~PerfLogImp() override;
 
    void
    rpcStart(std::string const& method, std::uint64_t const requestId) override;
 
    void
    rpcFinish(std::string const& method, std::uint64_t const requestId) override
    {
        rpcEnd(method, requestId, true);
    }
 
    void
    rpcError(std::string const& method, std::uint64_t const requestId) override
    {
        rpcEnd(method, requestId, false);
    }
 
    void
    jobQueue(JobType const type) override;
    void
    jobStart(
        JobType const type,
        microseconds dur,
        steady_time_point startTime,
        int instance) override;
    void
    jobFinish(JobType const type, microseconds dur, int instance) override;
 
    Json::Value
    countersJson() const override
    {
        return counters_.countersJson();
    }
 
    Json::Value
    currentJson() const override
    {
        return counters_.currentJson();
    }
 
    void
    resizeJobs(int const resize) override;
    void
    rotate() override;
 
    void
    start() override;
 
    void
    stop() override;
};
 
}  // namespace perf
}  // namespace ripple
 
#endif  // XRPL_BASICS_PERFLOGIMP_H