Track peer traffic by category

Builds/VisualStudio2015/RippleD.vcxproj

@@ -2624,6 +2624,12 @@
     </ClCompile>
     <ClInclude Include="..\..\src\ripple\overlay\impl\TMHello.h">
     </ClInclude>
+    <ClCompile Include="..\..\src\ripple\overlay\impl\TrafficCount.cpp">
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='debug|x64'">True</ExcludedFromBuild>
+      <ExcludedFromBuild Condition="'$(Configuration)|$(Platform)'=='release|x64'">True</ExcludedFromBuild>
+    </ClCompile>
+    <ClInclude Include="..\..\src\ripple\overlay\impl\TrafficCount.h">
+    </ClInclude>
     <ClInclude Include="..\..\src\ripple\overlay\impl\Tuning.h">
     </ClInclude>
     <ClInclude Include="..\..\src\ripple\overlay\impl\ZeroCopyStream.h">

Builds/VisualStudio2015/RippleD.vcxproj.filters

@@ -3309,6 +3309,12 @@
     <ClInclude Include="..\..\src\ripple\overlay\impl\TMHello.h">
       <Filter>ripple\overlay\impl</Filter>
     </ClInclude>
+    <ClCompile Include="..\..\src\ripple\overlay\impl\TrafficCount.cpp">
+      <Filter>ripple\overlay\impl</Filter>
+    </ClCompile>
+    <ClInclude Include="..\..\src\ripple\overlay\impl\TrafficCount.h">
+      <Filter>ripple\overlay\impl</Filter>
+    </ClInclude>
     <ClInclude Include="..\..\src\ripple\overlay\impl\Tuning.h">
       <Filter>ripple\overlay\impl</Filter>
     </ClInclude>

src/ripple/overlay/Message.h

@@ -63,6 +63,13 @@ public:
         return mBuffer;
     }
 
+    /** Get the traffic category */
+    int
+    getCategory () const
+    {
+        return mCategory;
+    }
+
     /** Determine bytewise equality. */
     bool operator == (Message const& other) const;
 
@@ -148,10 +155,10 @@ private:
     void encodeHeader (unsigned size, int type);
 
     std::vector <uint8_t> mBuffer;
+
+    int mCategory;
 };
 
 }
 
 #endif
-
-

src/ripple/overlay/impl/Message.cpp

@@ -19,6 +19,7 @@
 
 #include <BeastConfig.h>
 #include <ripple/overlay/Message.h>
+#include <ripple/overlay/impl/TrafficCount.h>
 #include <cstdint>
 
 namespace ripple {
@@ -37,6 +38,9 @@ Message::Message (::google::protobuf::Message const& message, int type)
     {
         message.SerializeToArray (&mBuffer [Message::kHeaderBytes], messageBytes);
     }
+
+    mCategory = static_cast<int>(TrafficCount::categorize
+        (message, type, false));
 }
 
 bool Message::operator== (Message const& other) const

src/ripple/overlay/impl/OverlayImpl.cpp

@@ -32,6 +32,7 @@
 #include <ripple/protocol/STExchange.h>
 #include <beast/ByteOrder.h>
 #include <beast/crypto/base64.h>
+#include <beast/module/core/text/LexicalCast.h>
 #include <beast/http/rfc2616.h>
 #include <beast/utility/ci_char_traits.h>
 #include <beast/utility/WrappedSink.h>
@@ -588,6 +589,28 @@ OverlayImpl::onChildrenStopped ()
 void
 OverlayImpl::onWrite (beast::PropertyStream::Map& stream)
 {
+    beast::PropertyStream::Set set ("traffic", stream);
+    auto stats = m_traffic.getCounts();
+    for (auto& i : stats)
+    {
+        if (! i.second.messagesIn && ! i.second.messagesOut)
+            continue;
+
+        beast::PropertyStream::Map item (set);
+        item["category"] = i.first;
+        item["bytes_in"] =
+            beast::lexicalCast<std::string>
+                (i.second.bytesIn.load());
+        item["messages_in"] =
+            beast::lexicalCast<std::string>
+                (i.second.messagesIn.load());
+        item["bytes_out"] =
+            beast::lexicalCast<std::string>
+                (i.second.bytesOut.load());
+        item["messages_out"] =
+            beast::lexicalCast<std::string>
+                (i.second.messagesOut.load());
+    }
 }
 
 //------------------------------------------------------------------------------
@@ -711,6 +734,15 @@ OverlayImpl::onManifests (
     }
 }
 
+void
+OverlayImpl::reportTraffic (
+    TrafficCount::category cat,
+    bool isInbound,
+    int number)
+{
+    m_traffic.addCount (cat, isInbound, number);
+}
+
 std::size_t
 OverlayImpl::selectPeers (PeerSet& set, std::size_t limit,
     std::function<bool(std::shared_ptr<Peer> const&)> score)

src/ripple/overlay/impl/OverlayImpl.h

@@ -24,6 +24,7 @@
 #include <ripple/core/Job.h>
 #include <ripple/overlay/Overlay.h>
 #include <ripple/overlay/impl/Manifest.h>
+#include <ripple/overlay/impl/TrafficCount.h>
 #include <ripple/server/Handoff.h>
 #include <ripple/server/ServerHandler.h>
 #include <ripple/basics/Resolver.h>
@@ -111,6 +112,7 @@ private:
     ServerHandler& serverHandler_;
     Resource::Manager& m_resourceManager;
     std::unique_ptr <PeerFinder::Manager> m_peerFinder;
+    TrafficCount m_traffic;
     hash_map <PeerFinder::Slot::ptr,
         std::weak_ptr <PeerImp>> m_peers;
     hash_map<RippleAddress, std::weak_ptr<PeerImp>> m_publicKeyMap;
@@ -267,6 +269,12 @@ public:
     std::string
     makePrefix (std::uint32_t id);
 
+    void
+    reportTraffic (
+        TrafficCount::category cat,
+        bool isInbound,
+        int bytes);
+
 private:
     std::shared_ptr<HTTP::Writer>
     makeRedirectResponse (PeerFinder::Slot::ptr const& slot,

src/ripple/overlay/impl/PeerImp.cpp

@@ -192,6 +192,10 @@ PeerImp::send (Message::pointer const& m)
     if(detaching_)
         return;
 
+    overlay_.reportTraffic (
+        static_cast<TrafficCount::category>(m->getCategory()),
+        false, static_cast<int>(m->getBuffer().size()));
+
     auto sendq_size = send_queue_.size();
 
     if (sendq_size < Tuning::targetSendQueue)
@@ -813,11 +817,14 @@ PeerImp::onMessageUnknown (std::uint16_t type)
 
 PeerImp::error_code
 PeerImp::onMessageBegin (std::uint16_t type,
-    std::shared_ptr <::google::protobuf::Message> const& m)
+    std::shared_ptr <::google::protobuf::Message> const& m,
+    std::size_t size)
 {
     load_event_ = app_.getJobQueue ().getLoadEventAP (
         jtPEER, protocolMessageName(type));
     fee_ = Resource::feeLightPeer;
+    overlay_.reportTraffic (TrafficCount::categorize (*m, type, true),
+        true, static_cast<int>(size));
     return error_code{};
 }
 

src/ripple/overlay/impl/PeerImp.h

@@ -403,7 +403,8 @@ public:
 
     error_code
     onMessageBegin (std::uint16_t type,
-        std::shared_ptr <::google::protobuf::Message> const& m);
+        std::shared_ptr <::google::protobuf::Message> const& m,
+        std::size_t size);
 
     void
     onMessageEnd (std::uint16_t type,

src/ripple/overlay/impl/ProtocolMessage.h

@@ -78,7 +78,8 @@ invoke (int type, Buffers const& buffers,
     if (! m->ParseFromZeroCopyStream(&stream))
         return boost::system::errc::make_error_code(
             boost::system::errc::invalid_argument);
-    auto ec = handler.onMessageBegin (type, m);
+    auto ec = handler.onMessageBegin (type, m,
+       Message::kHeaderBytes + Message::size (buffers));
     if (! ec)
     {
         handler.onMessage (m);

src/ripple/overlay/impl/TrafficCount.cpp

@@ -0,0 +1,135 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2012, 2013 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.
+*/
+//==============================================================================
+
+#include <BeastConfig.h>
+#include <ripple/overlay/impl/TrafficCount.h>
+
+namespace ripple {
+
+const char* TrafficCount::getName (category c)
+{
+    switch (c)
+    {
+        case category::CT_base:
+            return "overhead";
+        case category::CT_overlay:
+            return "overlay";
+        case category::CT_transaction:
+            return "transactions";
+        case category::CT_proposal:
+            return "proposals";
+        case category::CT_validation:
+            return "validations";
+        case category::CT_get_ledger:
+            return "ledger_get";
+        case category::CT_share_ledger:
+            return "ledger_share";
+        case category::CT_get_trans:
+            return "transaction_set_get";
+        case category::CT_share_trans:
+            return "transaction_set_share";
+        case category::CT_unknown:
+            assert (false);
+            return "unknown";
+        default:
+            assert (false);
+            return "truly_unknow";
+    }
+}
+
+TrafficCount::category TrafficCount::categorize (
+    ::google::protobuf::Message const& message,
+    int type, bool inbound)
+{
+    if ((type == protocol::mtHELLO) ||
+            (type == protocol::mtPING) ||
+            (type == protocol::mtCLUSTER) ||
+            (type == protocol::mtSTATUS_CHANGE))
+        return TrafficCount::category::CT_base;
+
+    if ((type == protocol::mtMANIFESTS) ||
+            (type == protocol::mtENDPOINTS) ||
+            (type == protocol::mtPEERS) ||
+            (type == protocol::mtGET_PEERS))
+        return TrafficCount::category::CT_overlay;
+
+    if (type == protocol::mtTRANSACTION)
+        return TrafficCount::category::CT_transaction;
+
+    if (type == protocol::mtVALIDATION)
+        return TrafficCount::category::CT_validation;
+
+    if (type == protocol::mtPROPOSE_LEDGER)
+        return TrafficCount::category::CT_proposal;
+
+    if (type == protocol::mtHAVE_SET)
+        return inbound ? TrafficCount::category::CT_get_trans :
+            TrafficCount::category::CT_share_trans;
+
+    {
+        auto msg = dynamic_cast
+            <protocol::TMLedgerData const*> (&message);
+        if (msg)
+        {
+            // We have received ledger data
+            if (msg->type() == protocol::liTS_CANDIDATE)
+                return (inbound && !msg->has_requestcookie()) ?
+                    TrafficCount::category::CT_get_trans :
+                    TrafficCount::category::CT_share_trans;
+            return (inbound && !msg->has_requestcookie()) ?
+                TrafficCount::category::CT_get_ledger :
+                TrafficCount::category::CT_share_ledger;
+        }
+    }
+
+    {
+        auto msg =
+            dynamic_cast <protocol::TMGetLedger const*>
+                (&message);
+        if (msg)
+        {
+            if (msg->itype() == protocol::liTS_CANDIDATE)
+                return (inbound || msg->has_requestcookie()) ?
+                    TrafficCount::category::CT_share_trans :
+                    TrafficCount::category::CT_get_trans;
+            return (inbound || msg->has_requestcookie()) ?
+                TrafficCount::category::CT_share_ledger :
+                TrafficCount::category::CT_get_ledger;
+        }
+    }
+
+    {
+        auto msg =
+            dynamic_cast <protocol::TMGetObjectByHash const*>
+                (&message);
+        if (msg)
+        {
+            // inbound queries and outbound responses are sharing
+            // outbound queries and inbound responses are getting
+            return (msg->query() == inbound) ?
+                TrafficCount::category::CT_share_ledger :
+                TrafficCount::category::CT_get_ledger;
+        }
+    }
+
+    assert (false);
+    return TrafficCount::category::CT_unknown;
+}
+
+} // ripple

src/ripple/overlay/impl/TrafficCount.h

@@ -0,0 +1,124 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2012, 2013 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_OVERLAY_TRAFFIC_H_INCLUDED
+#define RIPPLE_OVERLAY_TRAFFIC_H_INCLUDED
+
+#include "ripple.pb.h"
+
+#include <atomic>
+#include <map>
+
+namespace ripple {
+
+class TrafficCount
+{
+
+public:
+
+    using count_t = std::atomic <unsigned long>;
+
+    class TrafficStats
+    {
+        public:
+
+        count_t bytesIn;
+        count_t bytesOut;
+        count_t messagesIn;
+        count_t messagesOut;
+
+        TrafficStats() : bytesIn(0), bytesOut(0),
+            messagesIn(0), messagesOut(0)
+        { ; }
+
+        TrafficStats(const TrafficStats& ts)
+            : bytesIn (ts.bytesIn.load())
+            , bytesOut (ts.bytesOut.load())
+            , messagesIn (ts.messagesIn.load())
+            , messagesOut (ts.messagesOut.load())
+        { ; }
+    };
+
+
+    enum class category
+    {
+        CT_base,           // basic peer overhead, must be first
+        CT_overlay,        // overlay management
+        CT_transaction,
+        CT_proposal,
+        CT_validation,
+        CT_get_ledger,     // ledgers we try to get
+        CT_share_ledger,   // ledgers we share
+        CT_get_trans,      // transaction sets we try to get
+        CT_share_trans,    // transaction sets we get
+        CT_unknown         // must be last
+    };
+
+    static const char* getName (category c);
+
+    static category categorize (
+        ::google::protobuf::Message const& message,
+        int type, bool inbound);
+
+    void addCount (category cat, bool inbound, int number)
+    {
+        if (inbound)
+        {
+            counts_[cat].bytesIn += number;
+            ++counts_[cat].messagesIn;
+        }
+        else
+        {
+            counts_[cat].bytesOut += number;
+            ++counts_[cat].messagesOut;
+        }
+    }
+
+    TrafficCount()
+    {
+        for (category i = category::CT_base;
+            i <= category::CT_unknown;
+            i = static_cast<category>(static_cast<int>(i) + 1))
+        {
+            counts_[i];
+        }
+    }
+
+    std::map <std::string, TrafficStats>
+    getCounts () const
+    {
+        std::map <std::string, TrafficStats> ret;
+
+        for (auto& i : counts_)
+        {
+            ret.emplace (std::piecewise_construct,
+                std::forward_as_tuple (getName (i.first)),
+                std::forward_as_tuple (i.second));
+        }
+
+        return ret;
+    }
+
+    protected:
+
+    std::map <category, TrafficStats> counts_;
+};
+
+}
+#endif

src/ripple/protocol/JsonFields.h

@@ -275,7 +275,7 @@ JSS ( paths_computed );             // out: PathRequest, RipplePathFind
 JSS ( peer );                       // in: AccountLines
 JSS ( peer_authorized );            // out: AccountLines
 JSS ( peer_id );                    // out: LedgerProposal
-JSS ( peers );                      // out: InboundLedger, handlers/Peers
+JSS ( peers );                      // out: InboundLedger, handlers/Peers, Overlay
 JSS ( port );                       // in: Connect
 JSS ( previous_ledger );            // out: LedgerPropose
 JSS ( proof );                      // in: BookOffers
@@ -344,6 +344,7 @@ JSS ( taker_pays );                 // in: Subscribe, Unsubscribe, BookOffers
 JSS ( taker_pays_funded );          // out: NetworkOPs
 JSS ( threshold );                  // in: Blacklist
 JSS ( timeouts );                   // out: InboundLedger
+JSS ( traffic );                    // out: Overlay
 JSS ( totalCoins );                 // out: LedgerToJson
 JSS ( total_coins );                // out: LedgerToJson
 JSS ( transTreeHash );              // out: ledger/Ledger.cpp

src/ripple/unity/overlay.cpp

@@ -26,6 +26,7 @@
 #include <ripple/overlay/impl/PeerImp.cpp>
 #include <ripple/overlay/impl/PeerSet.cpp>
 #include <ripple/overlay/impl/TMHello.cpp>
+#include <ripple/overlay/impl/TrafficCount.cpp>
 
 #include <ripple/overlay/tests/manifest_test.cpp>
 #include <ripple/overlay/tests/short_read.test.cpp>