collectors.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2012-2017 Ripple Labs Inc
 
    Permission to use, copy, modify, and/or distribute this software for any
    purpose  with  or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#ifndef RIPPLE_TEST_CSF_COLLECTORS_H_INCLUDED
#define RIPPLE_TEST_CSF_COLLECTORS_H_INCLUDED
 
#include <test/csf/Histogram.h>
#include <test/csf/SimTime.h>
#include <test/csf/events.h>
 
#include <xrpl/basics/UnorderedContainers.h>
 
#include <chrono>
#include <optional>
#include <ostream>
#include <tuple>
 
namespace ripple {
namespace test {
namespace csf {
 
//  A collector is any class that implements
//
//      on(NodeID, SimTime, Event)
//
//  for all events emitted by a Peer.
//
//  This file contains helper functions for composing different collectors
//  and also defines several standard collectors available for simulations.
 
template <class... Cs>
class Collectors
{
    std::tuple<Cs&...> cs;
 
    template <class C, class E>
    static void
    apply(C& c, PeerID who, SimTime when, E e)
    {
        c.on(who, when, e);
    }
 
    template <std::size_t... Is, class E>
    static void
    apply(
        std::tuple<Cs&...>& cs,
        PeerID who,
        SimTime when,
        E e,
        std::index_sequence<Is...>)
    {
        (..., apply(std::get<Is>(cs), who, when, e));
    }
 
public:
    Collectors(Cs&... cs_) : cs(std::tie(cs_...))
    {
    }
 
    template <class E>
    void
    on(PeerID who, SimTime when, E e)
    {
        apply(cs, who, when, e, std::index_sequence_for<Cs...>{});
    }
};
 
template <class... Cs>
Collectors<Cs...>
makeCollectors(Cs&... cs)
{
    return Collectors<Cs...>(cs...);
}
 
template <class CollectorType>
struct CollectByNode
{
    std::map<PeerID, CollectorType> byNode;
 
    CollectorType&
    operator[](PeerID who)
    {
        return byNode[who];
    }
 
    CollectorType const&
    operator[](PeerID who) const
    {
        return byNode[who];
    }
    template <class E>
    void
    on(PeerID who, SimTime when, E const& e)
    {
        byNode[who].on(who, when, e);
    }
};
 
struct NullCollector
{
    template <class E>
    void
    on(PeerID, SimTime, E const& e)
    {
    }
};
 
struct SimDurationCollector
{
    bool init = false;
    SimTime start;
    SimTime stop;
 
    template <class E>
    void
    on(PeerID, SimTime when, E const& e)
    {
        if (!init)
        {
            start = when;
            init = true;
        }
        else
            stop = when;
    }
};
 
struct TxCollector
{
    // Counts
    std::size_t submitted{0};
    std::size_t accepted{0};
    std::size_t validated{0};
 
    struct Tracker
    {
        Tx tx;
        SimTime submitted;
        std::optional<SimTime> accepted;
        std::optional<SimTime> validated;
 
        Tracker(Tx tx_, SimTime submitted_) : tx{tx_}, submitted{submitted_}
        {
        }
    };
 
    hash_map<Tx::ID, Tracker> txs;
 
    using Hist = Histogram<SimTime::duration>;
    Hist submitToAccept;
    Hist submitToValidate;
 
    // Ignore most events by default
    template <class E>
    void
    on(PeerID, SimTime when, E const& e)
    {
    }
 
    void
    on(PeerID who, SimTime when, SubmitTx const& e)
    {
        // save first time it was seen
        if (txs.emplace(e.tx.id(), Tracker{e.tx, when}).second)
        {
            submitted++;
        }
    }
 
    void
    on(PeerID who, SimTime when, AcceptLedger const& e)
    {
        for (auto const& tx : e.ledger.txs())
        {
            auto it = txs.find(tx.id());
            if (it != txs.end() && !it->second.accepted)
            {
                Tracker& tracker = it->second;
                tracker.accepted = when;
                accepted++;
 
                submitToAccept.insert(*tracker.accepted - tracker.submitted);
            }
        }
    }
 
    void
    on(PeerID who, SimTime when, FullyValidateLedger const& e)
    {
        for (auto const& tx : e.ledger.txs())
        {
            auto it = txs.find(tx.id());
            if (it != txs.end() && !it->second.validated)
            {
                Tracker& tracker = it->second;
                // Should only validated a previously accepted Tx
                assert(tracker.accepted);
 
                tracker.validated = when;
                validated++;
                submitToValidate.insert(*tracker.validated - tracker.submitted);
            }
        }
    }
 
    // Returns the number of txs which were never accepted
    std::size_t
    orphaned() const
    {
        return std::count_if(txs.begin(), txs.end(), [](auto const& it) {
            return !it.second.accepted;
        });
    }
 
    // Returns the number of txs which were never validated
    std::size_t
    unvalidated() const
    {
        return std::count_if(txs.begin(), txs.end(), [](auto const& it) {
            return !it.second.validated;
        });
    }
 
    template <class T>
    void
    report(SimDuration simDuration, T& log, bool printBreakline = false)
    {
        using namespace std::chrono;
        auto perSec = [&simDuration](std::size_t count) {
            return double(count) / duration_cast<seconds>(simDuration).count();
        };
 
        auto fmtS = [](SimDuration dur) {
            return duration_cast<duration<float>>(dur).count();
        };
 
        if (printBreakline)
        {
            log << std::setw(11) << std::setfill('-') << "-" << "-"
                << std::setw(7) << std::setfill('-') << "-" << "-"
                << std::setw(7) << std::setfill('-') << "-" << "-"
                << std::setw(36) << std::setfill('-') << "-" << std::endl;
            log << std::setfill(' ');
        }
 
        log << std::left << std::setw(11) << "TxStats" << "|" << std::setw(7)
            << "Count" << "|" << std::setw(7) << "Per Sec" << "|"
            << std::setw(15) << "Latency (sec)" << std::right << std::setw(7)
            << "10-ile" << std::setw(7) << "50-ile" << std::setw(7) << "90-ile"
            << std::left << std::endl;
 
        log << std::setw(11) << std::setfill('-') << "-" << "|" << std::setw(7)
            << std::setfill('-') << "-" << "|" << std::setw(7)
            << std::setfill('-') << "-" << "|" << std::setw(36)
            << std::setfill('-') << "-" << std::endl;
        log << std::setfill(' ');
 
        log << std::left << std::setw(11) << "Submit " << "|" << std::right
            << std::setw(7) << submitted << "|" << std::setw(7)
            << std::setprecision(2) << perSec(submitted) << "|" << std::setw(36)
            << "" << std::endl;
 
        log << std::left << std::setw(11) << "Accept " << "|" << std::right
            << std::setw(7) << accepted << "|" << std::setw(7)
            << std::setprecision(2) << perSec(accepted) << "|" << std::setw(15)
            << std::left << "From Submit" << std::right << std::setw(7)
            << std::setprecision(2) << fmtS(submitToAccept.percentile(0.1f))
            << std::setw(7) << std::setprecision(2)
            << fmtS(submitToAccept.percentile(0.5f)) << std::setw(7)
            << std::setprecision(2) << fmtS(submitToAccept.percentile(0.9f))
            << std::endl;
 
        log << std::left << std::setw(11) << "Validate " << "|" << std::right
            << std::setw(7) << validated << "|" << std::setw(7)
            << std::setprecision(2) << perSec(validated) << "|" << std::setw(15)
            << std::left << "From Submit" << std::right << std::setw(7)
            << std::setprecision(2) << fmtS(submitToValidate.percentile(0.1f))
            << std::setw(7) << std::setprecision(2)
            << fmtS(submitToValidate.percentile(0.5f)) << std::setw(7)
            << std::setprecision(2) << fmtS(submitToValidate.percentile(0.9f))
            << std::endl;
 
        log << std::left << std::setw(11) << "Orphan" << "|" << std::right
            << std::setw(7) << orphaned() << "|" << std::setw(7) << "" << "|"
            << std::setw(36) << std::endl;
 
        log << std::left << std::setw(11) << "Unvalidated" << "|" << std::right
            << std::setw(7) << unvalidated() << "|" << std::setw(7) << "" << "|"
            << std::setw(43) << std::endl;
 
        log << std::setw(11) << std::setfill('-') << "-" << "-" << std::setw(7)
            << std::setfill('-') << "-" << "-" << std::setw(7)
            << std::setfill('-') << "-" << "-" << std::setw(36)
            << std::setfill('-') << "-" << std::endl;
        log << std::setfill(' ');
    }
 
    template <class T, class Tag>
    void
    csv(SimDuration simDuration,
        T& log,
        Tag const& tag,
        bool printHeaders = false)
    {
        using namespace std::chrono;
        auto perSec = [&simDuration](std::size_t count) {
            return double(count) / duration_cast<seconds>(simDuration).count();
        };
 
        auto fmtS = [](SimDuration dur) {
            return duration_cast<duration<float>>(dur).count();
        };
 
        if (printHeaders)
        {
            log << "tag" << "," << "txNumSubmitted" << "," << "txNumAccepted"
                << "," << "txNumValidated" << "," << "txNumOrphaned" << ","
                << "txUnvalidated" << "," << "txRateSumbitted" << ","
                << "txRateAccepted" << "," << "txRateValidated" << ","
                << "txLatencySubmitToAccept10Pctl" << ","
                << "txLatencySubmitToAccept50Pctl" << ","
                << "txLatencySubmitToAccept90Pctl" << ","
                << "txLatencySubmitToValidatet10Pctl" << ","
                << "txLatencySubmitToValidatet50Pctl" << ","
                << "txLatencySubmitToValidatet90Pctl" << std::endl;
        }
 
        log << tag
            << ","
            // txNumSubmitted
            << submitted
            << ","
            // txNumAccepted
            << accepted
            << ","
            // txNumValidated
            << validated
            << ","
            // txNumOrphaned
            << orphaned()
            << ","
            // txNumUnvalidated
            << unvalidated()
            << ","
            // txRateSubmitted
            << std::setprecision(2) << perSec(submitted)
            << ","
            // txRateAccepted
            << std::setprecision(2) << perSec(accepted)
            << ","
            // txRateValidated
            << std::setprecision(2) << perSec(validated)
            << ","
            // txLatencySubmitToAccept10Pctl
            << std::setprecision(2) << fmtS(submitToAccept.percentile(0.1f))
            << ","
            // txLatencySubmitToAccept50Pctl
            << std::setprecision(2) << fmtS(submitToAccept.percentile(0.5f))
            << ","
            // txLatencySubmitToAccept90Pctl
            << std::setprecision(2) << fmtS(submitToAccept.percentile(0.9f))
            << ","
            // txLatencySubmitToValidate10Pctl
            << std::setprecision(2) << fmtS(submitToValidate.percentile(0.1f))
            << ","
            // txLatencySubmitToValidate50Pctl
            << std::setprecision(2) << fmtS(submitToValidate.percentile(0.5f))
            << ","
            // txLatencySubmitToValidate90Pctl
            << std::setprecision(2) << fmtS(submitToValidate.percentile(0.9f))
            << "," << std::endl;
    }
};
 
struct LedgerCollector
{
    std::size_t accepted{0};
    std::size_t fullyValidated{0};
 
    struct Tracker
    {
        SimTime accepted;
        std::optional<SimTime> fullyValidated;
 
        Tracker(SimTime accepted_) : accepted{accepted_}
        {
        }
    };
 
    hash_map<Ledger::ID, Tracker> ledgers_;
 
    using Hist = Histogram<SimTime::duration>;
    Hist acceptToFullyValid;
    Hist acceptToAccept;
    Hist fullyValidToFullyValid;
 
    // Ignore most events by default
    template <class E>
    void
    on(PeerID, SimTime, E const& e)
    {
    }
 
    void
    on(PeerID who, SimTime when, AcceptLedger const& e)
    {
        // First time this ledger accepted
        if (ledgers_.emplace(e.ledger.id(), Tracker{when}).second)
        {
            ++accepted;
            // ignore jumps?
            if (e.prior.id() == e.ledger.parentID())
            {
                auto const it = ledgers_.find(e.ledger.parentID());
                if (it != ledgers_.end())
                {
                    acceptToAccept.insert(when - it->second.accepted);
                }
            }
        }
    }
 
    void
    on(PeerID who, SimTime when, FullyValidateLedger const& e)
    {
        // ignore jumps
        if (e.prior.id() == e.ledger.parentID())
        {
            auto const it = ledgers_.find(e.ledger.id());
            assert(it != ledgers_.end());
            auto& tracker = it->second;
            // first time fully validated
            if (!tracker.fullyValidated)
            {
                ++fullyValidated;
                tracker.fullyValidated = when;
                acceptToFullyValid.insert(when - tracker.accepted);
 
                auto const parentIt = ledgers_.find(e.ledger.parentID());
                if (parentIt != ledgers_.end())
                {
                    auto& parentTracker = parentIt->second;
                    if (parentTracker.fullyValidated)
                    {
                        fullyValidToFullyValid.insert(
                            when - *parentTracker.fullyValidated);
                    }
                }
            }
        }
    }
 
    std::size_t
    unvalidated() const
    {
        return std::count_if(
            ledgers_.begin(), ledgers_.end(), [](auto const& it) {
                return !it.second.fullyValidated;
            });
    }
 
    template <class T>
    void
    report(SimDuration simDuration, T& log, bool printBreakline = false)
    {
        using namespace std::chrono;
        auto perSec = [&simDuration](std::size_t count) {
            return double(count) / duration_cast<seconds>(simDuration).count();
        };
 
        auto fmtS = [](SimDuration dur) {
            return duration_cast<duration<float>>(dur).count();
        };
 
        if (printBreakline)
        {
            log << std::setw(11) << std::setfill('-') << "-" << "-"
                << std::setw(7) << std::setfill('-') << "-" << "-"
                << std::setw(7) << std::setfill('-') << "-" << "-"
                << std::setw(36) << std::setfill('-') << "-" << std::endl;
            log << std::setfill(' ');
        }
 
        log << std::left << std::setw(11) << "LedgerStats" << "|"
            << std::setw(7) << "Count" << "|" << std::setw(7) << "Per Sec"
            << "|" << std::setw(15) << "Latency (sec)" << std::right
            << std::setw(7) << "10-ile" << std::setw(7) << "50-ile"
            << std::setw(7) << "90-ile" << std::left << std::endl;
 
        log << std::setw(11) << std::setfill('-') << "-" << "|" << std::setw(7)
            << std::setfill('-') << "-" << "|" << std::setw(7)
            << std::setfill('-') << "-" << "|" << std::setw(36)
            << std::setfill('-') << "-" << std::endl;
        log << std::setfill(' ');
 
        log << std::left << std::setw(11) << "Accept " << "|" << std::right
            << std::setw(7) << accepted << "|" << std::setw(7)
            << std::setprecision(2) << perSec(accepted) << "|" << std::setw(15)
            << std::left << "From Accept" << std::right << std::setw(7)
            << std::setprecision(2) << fmtS(acceptToAccept.percentile(0.1f))
            << std::setw(7) << std::setprecision(2)
            << fmtS(acceptToAccept.percentile(0.5f)) << std::setw(7)
            << std::setprecision(2) << fmtS(acceptToAccept.percentile(0.9f))
            << std::endl;
 
        log << std::left << std::setw(11) << "Validate " << "|" << std::right
            << std::setw(7) << fullyValidated << "|" << std::setw(7)
            << std::setprecision(2) << perSec(fullyValidated) << "|"
            << std::setw(15) << std::left << "From Validate " << std::right
            << std::setw(7) << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.1f)) << std::setw(7)
            << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.5f)) << std::setw(7)
            << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.9f)) << std::endl;
 
        log << std::setw(11) << std::setfill('-') << "-" << "-" << std::setw(7)
            << std::setfill('-') << "-" << "-" << std::setw(7)
            << std::setfill('-') << "-" << "-" << std::setw(36)
            << std::setfill('-') << "-" << std::endl;
        log << std::setfill(' ');
    }
 
    template <class T, class Tag>
    void
    csv(SimDuration simDuration,
        T& log,
        Tag const& tag,
        bool printHeaders = false)
    {
        using namespace std::chrono;
        auto perSec = [&simDuration](std::size_t count) {
            return double(count) / duration_cast<seconds>(simDuration).count();
        };
 
        auto fmtS = [](SimDuration dur) {
            return duration_cast<duration<float>>(dur).count();
        };
 
        if (printHeaders)
        {
            log << "tag" << "," << "ledgerNumAccepted" << ","
                << "ledgerNumFullyValidated" << "," << "ledgerRateAccepted"
                << "," << "ledgerRateFullyValidated" << ","
                << "ledgerLatencyAcceptToAccept10Pctl" << ","
                << "ledgerLatencyAcceptToAccept50Pctl" << ","
                << "ledgerLatencyAcceptToAccept90Pctl" << ","
                << "ledgerLatencyFullyValidToFullyValid10Pctl" << ","
                << "ledgerLatencyFullyValidToFullyValid50Pctl" << ","
                << "ledgerLatencyFullyValidToFullyValid90Pctl" << std::endl;
        }
 
        log << tag
            << ","
            // ledgerNumAccepted
            << accepted
            << ","
            // ledgerNumFullyValidated
            << fullyValidated
            << ","
            // ledgerRateAccepted
            << std::setprecision(2) << perSec(accepted)
            << ","
            // ledgerRateFullyValidated
            << std::setprecision(2) << perSec(fullyValidated)
            << ","
            // ledgerLatencyAcceptToAccept10Pctl
            << std::setprecision(2) << fmtS(acceptToAccept.percentile(0.1f))
            << ","
            // ledgerLatencyAcceptToAccept50Pctl
            << std::setprecision(2) << fmtS(acceptToAccept.percentile(0.5f))
            << ","
            // ledgerLatencyAcceptToAccept90Pctl
            << std::setprecision(2) << fmtS(acceptToAccept.percentile(0.9f))
            << ","
            // ledgerLatencyFullyValidToFullyValid10Pctl
            << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.1f))
            << ","
            // ledgerLatencyFullyValidToFullyValid50Pctl
            << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.5f))
            << ","
            // ledgerLatencyFullyValidToFullyValid90Pctl
            << std::setprecision(2)
            << fmtS(fullyValidToFullyValid.percentile(0.9f)) << std::endl;
    }
};
 
struct StreamCollector
{
    std::ostream& out;
 
    // Ignore most events by default
    template <class E>
    void
    on(PeerID, SimTime, E const& e)
    {
    }
 
    void
    on(PeerID who, SimTime when, AcceptLedger const& e)
    {
        out << when.time_since_epoch().count() << ": Node " << who
            << " accepted " << "L" << e.ledger.id() << " " << e.ledger.txs()
            << "\n";
    }
 
    void
    on(PeerID who, SimTime when, FullyValidateLedger const& e)
    {
        out << when.time_since_epoch().count() << ": Node " << who
            << " fully-validated " << "L" << e.ledger.id() << " "
            << e.ledger.txs() << "\n";
    }
};
 
struct JumpCollector
{
    struct Jump
    {
        PeerID id;
        SimTime when;
        Ledger from;
        Ledger to;
    };
 
    std::vector<Jump> closeJumps;
    std::vector<Jump> fullyValidatedJumps;
 
    // Ignore most events by default
    template <class E>
    void
    on(PeerID, SimTime, E const& e)
    {
    }
 
    void
    on(PeerID who, SimTime when, AcceptLedger const& e)
    {
        // Not a direct child -> parent switch
        if (e.ledger.parentID() != e.prior.id())
            closeJumps.emplace_back(Jump{who, when, e.prior, e.ledger});
    }
 
    void
    on(PeerID who, SimTime when, FullyValidateLedger const& e)
    {
        // Not a direct child -> parent switch
        if (e.ledger.parentID() != e.prior.id())
            fullyValidatedJumps.emplace_back(
                Jump{who, when, e.prior, e.ledger});
    }
};
 
}  // namespace csf
}  // namespace test
}  // namespace ripple
 
#endif