diff --git a/Builds/QtCreator/rippled.pro b/Builds/QtCreator/rippled.pro index d8a9ac5d5d..a7b6aa4016 100644 --- a/Builds/QtCreator/rippled.pro +++ b/Builds/QtCreator/rippled.pro @@ -67,6 +67,7 @@ SOURCES += \ ../../src/ripple/beast/ripple_beastc.c \ ../../src/ripple/http/ripple_http.cpp \ ../../src/ripple/json/ripple_json.cpp \ + ../../src/ripple/peerfinder/ripple_peerfinder.cpp \ ../../src/ripple/rpc/ripple_rpc.cpp \ ../../src/ripple/sophia/ripple_sophia.c \ ../../src/ripple/sslutil/ripple_sslutil.cpp \ diff --git a/Builds/VisualStudio2012/RippleD.vcxproj b/Builds/VisualStudio2012/RippleD.vcxproj index 44ae01cd19..d6c4fd5f10 100644 --- a/Builds/VisualStudio2012/RippleD.vcxproj +++ b/Builds/VisualStudio2012/RippleD.vcxproj @@ -66,6 +66,49 @@ true + + true + true + true + true + + + true + true + true + true + + + true + true + true + true + + + true + true + true + true + + + true + true + true + true + + + true + true + true + true + + + true + true + true + true + + true true @@ -1022,12 +1065,6 @@ true true - - true - true - true - true - true @@ -1587,6 +1624,20 @@ + + + + + + + + + + + + + + @@ -1610,6 +1661,7 @@ + @@ -1635,7 +1687,6 @@ - @@ -1793,7 +1844,6 @@ - diff --git a/Builds/VisualStudio2012/RippleD.vcxproj.filters b/Builds/VisualStudio2012/RippleD.vcxproj.filters index 71a7cab512..22216cf48c 100644 --- a/Builds/VisualStudio2012/RippleD.vcxproj.filters +++ b/Builds/VisualStudio2012/RippleD.vcxproj.filters @@ -103,9 +103,6 @@ {548037f2-eb8a-41bd-95dc-05f58cdbc041} - - {d1648d3f-7d71-495d-afc9-576ed00d7185} - {df861e00-baa6-43d4-bbbf-df61d66ee414} @@ -232,6 +229,15 @@ {794194cf-e641-4838-ba78-82482c6e87c0} + + {6bda410a-7901-479d-9cc1-63c339057c92} + + + {908c91d8-3d3b-4c5e-ba85-dffc6f9b41f0} + + + {cd1585a6-bc32-477d-88cd-275159dafa83} + @@ -789,9 +795,6 @@ [0] Libraries\protobuf - - [2] Old Ripple\ripple_core\peerfinder - [2] Old Ripple\ripple_core\functional @@ -1059,6 +1062,30 @@ [2] Old Ripple\ripple_core\functional + + [1] Ripple\peerfinder + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + @@ -1728,9 +1755,6 @@ [2] Old Ripple\ripple_app\misc - - [2] Old Ripple\ripple_core\peerfinder - [2] Old Ripple\ripple_core\functional @@ -2106,15 +2130,57 @@ [2] Old Ripple\ripple_basics\containers - - [1] Ripple\validators\impl - [2] Old Ripple\ripple_core\functional [2] Old Ripple\ripple_core\functional + + [1] Ripple\types\api + + + [1] Ripple\peerfinder + + + [1] Ripple\peerfinder\api + + + [1] Ripple\peerfinder\api + + + [1] Ripple\peerfinder\api + + + [1] Ripple\peerfinder\api + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\impl + + + [1] Ripple\peerfinder\api + diff --git a/SConstruct b/SConstruct index 33701dc72f..ba4b92814b 100644 --- a/SConstruct +++ b/SConstruct @@ -13,6 +13,7 @@ OSX = bool(platform.mac_ver()[0]) FreeBSD = bool('FreeBSD' == platform.system()) Linux = bool('Linux' == platform.system()) Ubuntu = bool(Linux and 'Ubuntu' == platform.linux_distribution()[0]) +Debian = bool(Linux and 'debian' == platform.linux_distribution()[0]) Archlinux = bool(Linux and ('','','') == platform.linux_distribution()) #Arch still has issues with the platform module # @@ -21,7 +22,7 @@ Archlinux = bool(Linux and ('','','') == platform.linux_distribution()) #Arch s BOOST_HOME = os.environ.get("RIPPLED_BOOST_HOME", None) -if OSX or Ubuntu or Archlinux: +if OSX or Ubuntu or Debian or Archlinux: CTAGS = 'ctags' elif FreeBSD: CTAGS = 'exctags' @@ -157,6 +158,7 @@ COMPILED_FILES.extend (['src/ripple/beast/ripple_beastc.c']) COMPILED_FILES.extend([ 'src/ripple/http/ripple_http.cpp', 'src/ripple/json/ripple_json.cpp', + 'src/ripple/peerfinder/ripple_peerfinder.cpp', 'src/ripple/rpc/ripple_rpc.cpp', 'src/ripple/sophia/ripple_sophia.c', 'src/ripple/sslutil/ripple_sslutil.cpp', diff --git a/src/BeastConfig.h b/src/BeastConfig.h index b7257530f0..e5e47c58bc 100644 --- a/src/BeastConfig.h +++ b/src/BeastConfig.h @@ -17,7 +17,6 @@ */ //============================================================================== - #ifndef BEAST_BEASTCONFIG_H_INCLUDED #define BEAST_BEASTCONFIG_H_INCLUDED @@ -164,13 +163,10 @@ #define RIPPLE_USE_NEW_VALIDATORS 0 #endif -// This is only here temporarily. Use it to turn off the sending of -// "ANNOUNCE" messages if you suspect that you're having problems -// because of it. -#ifndef RIPPLE_USE_MT_ANNOUNCE -#define RIPPLE_USE_MT_ANNOUNCE 0 -#endif - +// Turning this on will use the new PeerFinder logic to establish connections +// to other peers. Even with this off, PeerFinder will still send mtENDPOINTS +// messages as needed, and collect legacy IP endpoint information. +// #ifndef RIPPLE_USE_PEERFINDER #define RIPPLE_USE_PEERFINDER 0 #endif diff --git a/src/ripple/peerfinder/api/Callback.h b/src/ripple/peerfinder/api/Callback.h new file mode 100644 index 0000000000..121b9acf59 --- /dev/null +++ b/src/ripple/peerfinder/api/Callback.h @@ -0,0 +1,50 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_CALLBACK_H_INCLUDED +#define RIPPLE_PEERFINDER_CALLBACK_H_INCLUDED + +#include "Endpoint.h" +#include "Types.h" + +namespace ripple { +namespace PeerFinder { + +/** The Callback receives PeerFinder notifications. + The notifications are sent on a thread owned by the PeerFinder, + so it is best not to do too much work in here. Just post functor + to another worker thread or job queue and return. +*/ +struct Callback +{ + /** Sends a set of Endpoint records to the specified peer. */ + virtual void sendPeerEndpoints (PeerID const& id, + std::vector const& endpoints) = 0; + + /** Initiate outgoing Peer connections to the specified set of endpoints. */ + virtual void connectPeerEndpoints (std::vector const& list) = 0; + + /** Impose a load charge on the specified peer. */ + virtual void chargePeerLoadPenalty (PeerID const& id) = 0; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/api/Config.h b/src/ripple/peerfinder/api/Config.h new file mode 100644 index 0000000000..ef63574fe9 --- /dev/null +++ b/src/ripple/peerfinder/api/Config.h @@ -0,0 +1,45 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_CONFIG_H_INCLUDED +#define RIPPLE_PEERFINDER_CONFIG_H_INCLUDED + +namespace ripple { +namespace PeerFinder { + +/** Configuration for the Manager. */ +struct Config +{ + Config (); + + static int const minOutCount = 10; + static int const outPercent = 15; + int maxPeerCount; + + /** True if we want to accept incoming connections. */ + bool wantIncoming; + + uint16 listeningPort; + std::string featureList; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/api/Endpoint.h b/src/ripple/peerfinder/api/Endpoint.h new file mode 100644 index 0000000000..b1adc3a7f7 --- /dev/null +++ b/src/ripple/peerfinder/api/Endpoint.h @@ -0,0 +1,55 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_ENDPOINT_H_INCLUDED +#define RIPPLE_PEERFINDER_ENDPOINT_H_INCLUDED + +#include "Types.h" + +namespace ripple { +namespace PeerFinder { + +/** Describes a connectible peer address along with some metadata. */ +struct Endpoint +{ + Endpoint (); + + IPEndpoint address; + uint16 port; + int hops; + uint32 incomingSlotsAvailable; + uint32 incomingSlotsMax; + uint32 uptimeMinutes; + std::string featureList; +}; + +inline bool operator< (Endpoint const& lhs, Endpoint const& rhs) +{ + return lhs.address < rhs.address; +} + +inline bool operator== (Endpoint const& lhs, Endpoint const& rhs) +{ + return lhs.address == rhs.address; +} + +} +} + +#endif diff --git a/src/ripple/peerfinder/api/Manager.h b/src/ripple/peerfinder/api/Manager.h new file mode 100644 index 0000000000..1cd6a22f72 --- /dev/null +++ b/src/ripple/peerfinder/api/Manager.h @@ -0,0 +1,90 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_MANAGER_H_INCLUDED +#define RIPPLE_PEERFINDER_MANAGER_H_INCLUDED + +namespace ripple { +namespace PeerFinder { + +/** Maintains a set of IP addresses used for getting into the network. */ +class Manager : public Stoppable +{ +protected: + explicit Manager (Stoppable& parent); + +public: + /** Create a new Manager. */ + static Manager* New (Stoppable& parent, + Callback& callback, Journal journal); + + /** Destroy the object. + Any pending source fetch operations are aborted. + There may be some listener calls made before the + destructor returns. + */ + virtual ~Manager () { } + + /** Set the configuration for the manager. + The new settings will be applied asynchronously. + Thread safety: + Can be called from any threads at any time. + */ + virtual void setConfig (Config const& config) = 0; + + /** Add a set of strings as fallback IPEndpoint sources. + @param name A label used for diagnostics. + */ + virtual void addStrings (std::string const& name, + std::vector const& strings) = 0; + + /** Add a URL as a fallback location to obtain IPEndpoint sources. + @param name A label used for diagnostics. + */ + virtual void addURL (std::string const& name, + std::string const& url) = 0; + + /** Called when a new peer connection is established. + Internally, we add the peer to our tracking table, validate that + we can connect to it, and begin advertising it to others after + we are sure that its connection is stable. + */ + virtual void onPeerConnected (PeerID const& id, + IPEndpoint const& address, + bool inbound) = 0; + + /** Called when an existing peer connection drops for whatever reason. + Internally, we mark the peer as no longer connected, calculate + stability metrics, and consider whether we should try to reconnect + to it or drop it from our list. + */ + virtual void onPeerDisconnected (PeerID const& id) = 0; + + /** Called when mtENDPOINTS is received. */ + virtual void onPeerEndpoints (PeerID const& id, + std::vector const& endpoints) = 0; + + /** Called when a legacy IP/port address is received (from mtPEER). */ + virtual void onPeerLegacyEndpoint (IPEndpoint const& ep) = 0; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/api/Types.h b/src/ripple/peerfinder/api/Types.h new file mode 100644 index 0000000000..f9cf3d6e93 --- /dev/null +++ b/src/ripple/peerfinder/api/Types.h @@ -0,0 +1,32 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_TYPES_H_INCLUDED +#define RIPPLE_PEERFINDER_TYPES_H_INCLUDED + +namespace ripple { +namespace PeerFinder { + +/** Used to identify peers. */ +typedef RipplePublicKey PeerID; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Config.cpp b/src/ripple/peerfinder/impl/Config.cpp new file mode 100644 index 0000000000..d3e5b6e732 --- /dev/null +++ b/src/ripple/peerfinder/impl/Config.cpp @@ -0,0 +1,33 @@ +//------------------------------------------------------------------------------ +/* + 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 "../api/Config.h" + +namespace ripple { +namespace PeerFinder { + +Config::Config () + : maxPeerCount (20) + , wantIncoming (false) + , listeningPort (0) +{ +} + +} +} diff --git a/src/ripple/peerfinder/impl/Endpoint.cpp b/src/ripple/peerfinder/impl/Endpoint.cpp new file mode 100644 index 0000000000..44979d1e4b --- /dev/null +++ b/src/ripple/peerfinder/impl/Endpoint.cpp @@ -0,0 +1,35 @@ +//------------------------------------------------------------------------------ +/* + 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 "../api/Endpoint.h" + +namespace ripple { +namespace PeerFinder { + +Endpoint::Endpoint () + : port (0) + , hops (0) + , incomingSlotsAvailable (0) + , incomingSlotsMax (0) + , uptimeMinutes (0) +{ +} + +} +} diff --git a/src/ripple/peerfinder/impl/EndpointCache.cpp b/src/ripple/peerfinder/impl/EndpointCache.cpp new file mode 100644 index 0000000000..398cc19215 --- /dev/null +++ b/src/ripple/peerfinder/impl/EndpointCache.cpp @@ -0,0 +1,19 @@ +//------------------------------------------------------------------------------ +/* + 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. +*/ +//============================================================================== + diff --git a/src/ripple/peerfinder/impl/EndpointCache.h b/src/ripple/peerfinder/impl/EndpointCache.h new file mode 100644 index 0000000000..6463cafe00 --- /dev/null +++ b/src/ripple/peerfinder/impl/EndpointCache.h @@ -0,0 +1,46 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_ENDPOINTCACHE_H_INCLUDED +#define RIPPLE_PEERFINDER_ENDPOINTCACHE_H_INCLUDED + +#include "../../ripple/types/api/AgedHistory.h" + +#include "../api/Types.h" + +namespace ripple { +namespace PeerFinder { + +/** This container holds the master set of Endpoints. */ +class EndpointCache +{ +public: + EndpointCache (); + ~EndpointCache (); + + // Informs the cache we've received an endpoint. + void update (Endpoint const& ep); + +private: +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Logic.h b/src/ripple/peerfinder/impl/Logic.h new file mode 100644 index 0000000000..a0bed834d3 --- /dev/null +++ b/src/ripple/peerfinder/impl/Logic.h @@ -0,0 +1,496 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_LOGIC_H_INCLUDED +#define RIPPLE_PEERFINDER_LOGIC_H_INCLUDED + +#include "../../ripple/types/api/AgedHistory.h" + +#include "PeerInfo.h" +#include "Slots.h" +#include "Store.h" + +#include +#include "beast/modules/beast_core/system/BeforeBoost.h" +#include +#include +#include +#include +#include + +namespace ripple { +namespace PeerFinder { + +// Tunable constants +enum +{ + // How often we will try to make outgoing connections + secondsPerConnect = 10, + + // How often we send or accept mtENDPOINTS messages per peer + secondsPerEndpoints = 5, + + // How many Endpoint to send in each mtENDPOINTS + numberOfEndpoints = 10, + + // The most Endpoint we will accept in mtENDPOINTS + numberOfEndpointsMax = 20, + + // How many legacy endpoints to keep in our cache + numberOfLegacyEndpoints = 1000, + + // How often legacy endpoints are updated in the database + legacyEndpointUpdateSeconds = 60 * 60 +}; + +//-------------------------------------------------------------------------- + +/* +typedef boost::multi_index_container < + Endpoint, boost::multi_index::indexed_by < + + boost::multi_index::hashed_unique < + BOOST_MULTI_INDEX_MEMBER(PeerFinder::Endpoint,IPEndpoint,address)> + > +> EndpointCache; +*/ + +// Describes an Endpoint in the global Endpoint table +// This includes the Endpoint as well as some additional information +// +struct EndpointInfo +{ + Endpoint endpoint; +}; + +inline bool operator< (EndpointInfo const& lhs, EndpointInfo const& rhs) +{ + return lhs.endpoint < rhs.endpoint; +} + +inline bool operator== (EndpointInfo const& lhs, EndpointInfo const& rhs) +{ + return lhs.endpoint == rhs.endpoint; +} + +typedef AgedHistory > LegacyEndpoints; + +//-------------------------------------------------------------------------- + +typedef boost::multi_index_container < + PeerInfo, boost::multi_index::indexed_by < + boost::multi_index::hashed_unique < + BOOST_MULTI_INDEX_MEMBER(PeerFinder::PeerInfo,PeerID,id), + PeerID::hasher>, + boost::multi_index::hashed_unique < + BOOST_MULTI_INDEX_MEMBER(PeerFinder::PeerInfo,beast::IPEndpoint,address), + IPEndpoint::hasher> + > +> Peers; + +//------------------------------------------------------------------------------ + +/** The Logic for maintaining the list of Peer addresses. + We keep this in a separate class so it can be instantiated + for unit tests. +*/ +class Logic +{ +public: + struct State + { + State () + { + } + + // Fresh endpoints are ones we have seen recently via mtENDPOINTS. + // These are best to give out to someone who needs additional + // connections as quickly as possible, since it is very likely + // that the fresh endpoints have open incoming slots. + // + //EndpointCache fresh; + + // Reliable endpoints are ones which are highly likely to be + // connectible over long periods of time. They might not necessarily + // have an incoming slot, but they are good for bootstrapping when + // there are no peers yet. Typically these are what we would want + // to store in a database or local config file for a future launch. + //Endpoints reliable; + }; + + //---------------------------------------------------------------------- + + Callback& m_callback; + Store& m_store; + Journal m_journal; + Config m_config; + + // A list of dynamic sources consulted as a fallback + std::vector > m_sources; + + // The current tally of peer slot statistics + Slots m_slots; + + // Our view of the current set of connected peers. + Peers m_peers; + + LegacyEndpoints m_legacyEndpoints; + bool m_legacyEndpointsDirty; + + //---------------------------------------------------------------------- + + Logic (Callback& callback, Store& store, Journal journal) + : m_callback (callback) + , m_store (store) + , m_journal (journal) + , m_legacyEndpointsDirty (false) + { + } + + //---------------------------------------------------------------------- + + // Load persistent state information from the Store + // + void load () + { + typedef std::vector List; + List list; + m_store.loadLegacyEndpoints (list); + for (List::const_iterator iter (list.begin()); + iter != list.end(); ++iter) + m_legacyEndpoints->insert (*iter); + m_legacyEndpoints.swap(); + m_journal.debug << "Loaded " << list.size() << " legacy endpoints"; + } + + // Called when a peer's id is unexpectedly not found + // + void peerNotFound (PeerID const& id) + { + m_journal.fatal << "Missing peer " << id; + } + + // Returns a suitable Endpoint representing us. + // + Endpoint thisEndpoint () + { + // Why would someone call this if we don't want incoming? + bassert (m_config.wantIncoming); + + Endpoint ep; + // ep.address = ? + ep.port = m_config.listeningPort; + ep.hops = 0; + ep.incomingSlotsAvailable = m_slots.inboundSlots; + ep.incomingSlotsMax = m_slots.inboundSlotsMaximum; + ep.uptimeMinutes = m_slots.uptimeMinutes(); + + return ep; + } + + // Returns true if the Endpoint contains no invalid data. + // + bool validEndpoint (Endpoint const& endpoint) + { + if (! endpoint.address.isPublic()) + return false; + if (endpoint.port == 0) + return false; + return false; + } + + // Prunes invalid endpoints from a list + // + void pruneEndpoints (std::vector & list) + { + for (std::vector ::iterator iter (list.begin()); + iter != list.end(); ++iter) + { + while (! validEndpoint (*iter)) + { + m_journal.error << "Pruned invalid endpoint " << iter->address; + iter = list.erase (iter); + if (iter == list.end()) + break; + } + } + } + + // Send mtENDPOINTS for the specified peer + // + void sendEndpoints (PeerInfo const& peer) + { + typedef std::vector List; + std::vector endpoints; + + // fill in endpoints + + // Add us to the list if we want incoming + if (m_slots.inboundSlots > 0) + endpoints.push_back (thisEndpoint ()); + + if (! endpoints.empty()) + m_callback.sendPeerEndpoints (peer.id, endpoints); + } + + // Assembles a list from the legacy endpoint container + // + void createLegacyEndpointList (std::vector & list) + { + list.clear (); + list.reserve (m_legacyEndpoints.front().size() + + m_legacyEndpoints.back().size()); + + for (LegacyEndpoints::container_type::const_iterator iter ( + m_legacyEndpoints.front().begin()); iter != m_legacyEndpoints.front().end(); ++iter) + list.push_back (*iter); + + for (LegacyEndpoints::container_type::const_iterator iter ( + m_legacyEndpoints.back().begin()); iter != m_legacyEndpoints.back().end(); ++iter) + list.push_back (*iter); + } + + // Make outgoing connections to bring us up to desired out count + // + void makeOutgoingConnections () + { + if (m_slots.outDesired > m_slots.outboundCount) + { + std::vector list; + createLegacyEndpointList (list); + std::random_shuffle (list.begin(), list.end()); + + int needed = m_slots.outDesired - m_slots.outboundCount; + if (needed > list.size()) + needed = list.size(); + +#if RIPPLE_USE_PEERFINDER + m_callback.connectPeerEndpoints (list); +#endif + } + } + + // Fetch the list of IPEndpoint from the specified source + // + void fetch (Source& source) + { + m_journal.debug << "Fetching " << source.name(); + + Source::IPEndpoints endpoints; + source.fetch (endpoints, m_journal); + + if (! endpoints.empty()) + { + for (Source::IPEndpoints::const_iterator iter (endpoints.begin()); + iter != endpoints.end(); ++iter) + m_legacyEndpoints->insert (*iter); + + if (m_legacyEndpoints->size() > (numberOfLegacyEndpoints/2)) + { + m_legacyEndpoints.swap(); + m_legacyEndpoints->clear(); + } + + m_legacyEndpointsDirty = true; + } + } + + //---------------------------------------------------------------------- + + void setConfig (Config const& config) + { + m_config = config; + m_slots.update (m_config); + } + + void addStaticSource (Source* source) + { + ScopedPointer p (source); + fetch (*source); + } + + void addSource (Source* source) + { + m_sources.push_back (source); + } + + void onUpdate () + { + m_journal.debug << "Processing Update"; + } + + // Send mtENDPOINTS for each peer as needed + // + void sendEndpoints () + { + if (! m_peers.empty()) + { + m_journal.debug << "Sending mtENDPOINTS"; + + RelativeTime const now (RelativeTime::fromStartup()); + + for (Peers::iterator iter (m_peers.begin()); + iter != m_peers.end(); ++iter) + { + PeerInfo const& peer (*iter); + if (peer.whenSendEndpoints <= now) + { + sendEndpoints (peer); + peer.whenSendEndpoints = now + + RelativeTime (secondsPerEndpoints); + } + } + } + } + + void onPeerConnected (PeerID const& id, + IPEndpoint const& address, bool inbound) + { + m_journal.debug << "Peer connected: " << address; + + std::pair result ( + m_peers.insert ( + PeerInfo (id, address, inbound))); + if (result.second) + { + //PeerInfo const& peer (*result.first); + m_slots.addPeer (m_config, inbound); + } + else + { + // already exists! + m_journal.error << "Duplicate peer " << id; + //m_callback.disconnectPeer (id); + } + } + + void onPeerDisconnected (PeerID const& id) + { + Peers::iterator iter (m_peers.find (id)); + if (iter != m_peers.end()) + { + // found + PeerInfo const& peer (*iter); + m_journal.debug << "Peer disconnected: " << peer.address; + m_slots.dropPeer (m_config, peer.inbound); + m_peers.erase (iter); + } + else + { + m_journal.debug << "Peer disconnected: " << id; + peerNotFound (id); + } + } + + // Processes a list of Endpoint received from a peer. + // + void onPeerEndpoints (PeerID const& id, std::vector endpoints) + { + pruneEndpoints (endpoints); + + Peers::iterator iter (m_peers.find (id)); + if (iter != m_peers.end()) + { + RelativeTime const now (RelativeTime::fromStartup()); + PeerInfo const& peer (*iter); + + if (now >= peer.whenReceiveEndpoints) + { + m_journal.debug << "Received " << endpoints.size() << + "Endpoint descriptors from " << peer.address; + + // We charge a load penalty if the peer sends us more than + // numberOfEndpoints peers in a single message + if (endpoints.size() > numberOfEndpoints) + { + m_journal.warning << "Charging peer " << peer.address << + " for sending too many endpoints"; + + m_callback.chargePeerLoadPenalty(id); + } + + // process the list + + peer.whenReceiveEndpoints = now + secondsPerEndpoints; + } + else + { + m_journal.warning << "Charging peer " << peer.address << + " for sending too quickly"; + + // Peer sent mtENDPOINTS too often + m_callback.chargePeerLoadPenalty (id); + } + } + else + { + peerNotFound (id); + } + } + + void onPeerLegacyEndpoint (IPEndpoint const& ep) + { + if (ep.isPublic()) + { + // insert into front container + std::pair result ( + m_legacyEndpoints->insert (ep)); + + // erase from back container if its new + if (result.second) + { + std::size_t const n (m_legacyEndpoints.back().erase (ep)); + if (n == 0) + { + m_legacyEndpointsDirty = true; + m_journal.trace << "Legacy endpoint: " << ep; + } + } + + if (m_legacyEndpoints->size() > (numberOfLegacyEndpoints/2)) + { + m_legacyEndpoints.swap(); + m_legacyEndpoints->clear(); + } + } + } + + // Updates the Store with the current set of legacy endpoints + // + void storeLegacyEndpoints () + { + if (!m_legacyEndpointsDirty) + return; + + std::vector list; + + createLegacyEndpointList (list); + + m_journal.debug << "Updating " << list.size() << " legacy endpoints"; + + m_store.storeLegacyEndpoints (list); + + m_legacyEndpointsDirty = false; + } +}; + +} +} + +#endif diff --git a/src/ripple_core/peerfinder/PeerFinder.cpp b/src/ripple/peerfinder/impl/Manager.cpp similarity index 62% rename from src/ripple_core/peerfinder/PeerFinder.cpp rename to src/ripple/peerfinder/impl/Manager.cpp index 6811aae9d7..fdf9c52ee1 100644 --- a/src/ripple_core/peerfinder/PeerFinder.cpp +++ b/src/ripple/peerfinder/impl/Manager.cpp @@ -78,10 +78,10 @@ network graph We define these values: - PeerCount (calculated) + peerCount (calculated) The number of currently connected and established peers - OutCount (calculated) + outCount (calculated) The number of peers in PeerCount that are outbound connections. MinOutCount (hard-coded constant) @@ -95,7 +95,7 @@ We define these values: that a peer wishes to maintain. Setting MaxPeerCount equal to or below MinOutCount would disallow incoming connections. - OutDesiredPercent (a baked-in program constant for now) + OutPercent (a baked-in program constant for now) The peer's target value for OutCount. When the value of OutCount is below this number, the peer will employ the Outgoing Strategy to raise its value of OutCount. This value is initially a constant @@ -170,214 +170,214 @@ Gnutella 0.6 Protocol Revised Gnutella Ping Pong Scheme By Christopher Rohrs and Vincent Falco http://rfc-gnutella.sourceforge.net/src/pong-caching.html - */ -//------------------------------------------------------------------------------ -class PeerFinderImp - : public PeerFinder - , private ThreadWithCallQueue::EntryPoints - , private DeadlineTimer::Listener - , LeakChecked +#include "Logic.h" +#include "StoreSqdb.h" + +namespace ripple { +namespace PeerFinder { + +class ManagerImp + : public Manager + , public Thread + , public DeadlineTimer::Listener + , public LeakChecked { public: - // Tunable constants - enum - { - // How often our timer goes off to consult outside sources for IPs - secondsPerUpdate = 1 * 60 * 60, // once per hour - // How often we announce our IP - secondsPerBroadcast = 5 * 60, - - // The minimum number of peers we want - numberOfPeersMinimum = 4, - numberOfPeersMaximum = 10, - - // The minimum number of seconds a connection ought to be sustained - // before we consider it "stable" - secondsForStability = 60, // one minute - }; + ServiceQueue m_queue; + Journal m_journal; + StoreSqdb m_store; + Logic m_logic; + DeadlineTimer m_connectTimer; + DeadlineTimer m_endpointsTimer; + RelativeTime m_whenStoreLegacyEndpoints; //-------------------------------------------------------------------------- - /** The Logic for maintaining the list of Peer addresses. - We keep this in a separate class so it can be instantiated - for unit tests. - */ - class Logic + ManagerImp (Stoppable& stoppable, Callback& callback, Journal journal) + : Manager (stoppable) + , Thread ("PeerFinder") + , m_journal (journal) + , m_store (journal) + , m_logic (callback, m_store, journal) + , m_connectTimer (this) + , m_endpointsTimer (this) { - Callback &m_callback; - - public: - explicit Logic (Callback& callback) - : m_callback (callback) +#if BEAST_MSVC + if (beast_isRunningUnderDebugger()) { + m_journal.sink().set_console (true); + m_journal.sink().set_severity (Journal::kLowestSeverity); } - - // Called on the PeerFinder thread - void onUpdateConnectionsStatus ( - Connections const& connections) - { - if (connections.numberTotal () < numberOfPeersMinimum) - { - // do something - } - else - { - // do something? - } - } - - void onPeerConnected ( - const PeerId& id) - { - - } - - void onPeerDisconnected ( - const PeerId& id) - { - - } - - void onAcceptTimer() - { - m_callback.onAnnounceAddress (); - } - }; - - //-------------------------------------------------------------------------- - -public: - explicit PeerFinderImp (Callback& callback) - : m_logic (callback) - , m_thread ("PeerFinder") - , m_acceptTimer (this) - , m_updateTimer (this) - { - m_thread.start (this); - } - - ~PeerFinderImp () - { - } - - void updateConnectionsStatus (Connections& connections) - { - // Queue the call to the logic - m_thread.call (&Logic::onUpdateConnectionsStatus, - &m_logic, connections); - } - - void onPeerConnected(const PeerId& id) - { - m_thread.call (&Logic::onPeerConnected, - &m_logic, id); - } - - void onPeerDisconnected(const PeerId& id) - { - m_thread.call (&Logic::onPeerDisconnected, - &m_logic, id); - } - - //-------------------------------------------------------------------------- - void onAcceptTimer () - { -#if 0 - static int x = 0; - - if(x == 0) - Debug::breakPoint (); - - x++; #endif } + ~ManagerImp () + { + } + + //-------------------------------------------------------------------------- + // + // PeerFinder + // + + void setConfig (Config const& config) + { + m_queue.dispatch (bind (&Logic::setConfig, &m_logic, config)); + } + + void addStrings (std::string const& name, + std::vector const& strings) + { + m_queue.dispatch (bind (&Logic::addStaticSource, &m_logic, + SourceStrings::New (name, strings))); + } + + void addURL (std::string const& name, std::string const& url) + { + } + + void onPeerConnected (PeerID const& id, + IPEndpoint const& address, bool incoming) + { + m_queue.dispatch (bind (&Logic::onPeerConnected, &m_logic, + id, address, incoming)); + } + + void onPeerDisconnected (const PeerID& id) + { + m_queue.dispatch (bind (&Logic::onPeerDisconnected, &m_logic, id)); + } + + void onPeerLegacyEndpoint (IPEndpoint const& ep) + { + m_queue.dispatch (bind (&Logic::onPeerLegacyEndpoint, &m_logic, + ep)); + } + + void onPeerEndpoints (PeerID const& id, + std::vector const& endpoints) + { + m_queue.dispatch (beast::bind (&Logic::onPeerEndpoints, &m_logic, + id, endpoints)); + } + + //-------------------------------------------------------------------------- + // + // Stoppable + // + + void onPrepare (Journal journal) + { + } + + void onStart (Journal journal) + { + startThread(); + } + + void onStop (Journal journal) + { + if (this->Thread::isThreadRunning ()) + { + m_journal.debug << "Stopping"; + m_connectTimer.cancel(); + m_endpointsTimer.cancel(); + m_queue.dispatch (bind (&Thread::signalThreadShouldExit, this)); + } + else + { + stopped(); + } + } + + //-------------------------------------------------------------------------- + void onDeadlineTimer (DeadlineTimer& timer) { - // This will make us fall into the idle proc as needed - // - if (timer == m_updateTimer) - m_thread.interrupt (); - else if (timer == m_acceptTimer) - m_thread.call (&Logic::onAcceptTimer, &m_logic); + if (timer == m_connectTimer) + { + m_queue.dispatch (bind (&Logic::makeOutgoingConnections, &m_logic)); + m_connectTimer.setExpiration (secondsPerConnect); + } + else if (timer == m_endpointsTimer) + { + m_queue.dispatch (bind (&Logic::sendEndpoints, &m_logic)); + m_endpointsTimer.setExpiration (secondsPerEndpoints); + } } - void threadInit () + // Checks to see if its time to update legacy endpoints + void storeLegacyEndpoints() { - m_updateTimer.setRecurringExpiration (secondsPerUpdate); - m_acceptTimer.setRecurringExpiration (secondsPerBroadcast); + RelativeTime const now (RelativeTime::fromStartup()); + if (now >= m_whenStoreLegacyEndpoints) + { + m_logic.storeLegacyEndpoints (); + m_whenStoreLegacyEndpoints = now + + RelativeTime (legacyEndpointUpdateSeconds); + } } - void threadExit () + void init () { + m_journal.debug << "Initializing"; + + File const file (File::getSpecialLocation ( + File::userDocumentsDirectory).getChildFile ("PeerFinder.sqlite")); + + m_journal.debug << "Opening database at '" << file.getFullPathName() << "'"; + + Error error (m_store.open (file)); + + if (error) + { + m_journal.fatal << + "Failed to open '" << file.getFullPathName() << "'"; + } + + if (! error) + { + m_logic.load (); + } + + m_connectTimer.setExpiration (secondsPerConnect); + m_endpointsTimer.setExpiration (secondsPerEndpoints); + + m_queue.post (bind (&Logic::makeOutgoingConnections, &m_logic)); } - bool threadIdle () + void run () { - bool interrupted = false; + m_journal.debug << "Started"; - // This is where you can go into a loop and do stuff - // like process the lists, and what not. Just be - // sure to call: - // - // @code - // interrupted = interruptionPoint (); - // @encode - // - // From time to time. If it returns true then you - // need to exit this function so that Thread can - // process its asynchronous call queue and then come - // back into threadIdle() + m_whenStoreLegacyEndpoints = RelativeTime::fromStartup() + + RelativeTime (legacyEndpointUpdateSeconds); - return interrupted; + init (); + + while (! this->threadShouldExit()) + { + storeLegacyEndpoints(); + m_queue.run_one(); + } + + stopped(); } - -private: - Logic m_logic; - ThreadWithCallQueue m_thread; - DeadlineTimer m_acceptTimer; - DeadlineTimer m_updateTimer; }; //------------------------------------------------------------------------------ -PeerFinder* PeerFinder::New (PeerFinder::Callback& callback) +Manager::Manager (Stoppable& parent) + : Stoppable ("PeerFinder", parent) { - return new PeerFinderImp (callback); } -//------------------------------------------------------------------------------ - -class PeerFinderTests : public UnitTest, - public PeerFinder::Callback +Manager* Manager::New (Stoppable& parent, Callback& callback, Journal journal) { -public: - void testValidityChecks () - { - beginTestCase ("ip validation"); - - fail ("there's no code!"); - } - - void runTest () - { - PeerFinderImp::Logic logic (*this); - - beginTestCase ("logic"); - logic.onAcceptTimer (); - } - - void onAnnounceAddress () - { - - } - - PeerFinderTests () : UnitTest ("PeerFinder", "ripple", runManual) - { - } -}; - -static PeerFinderTests peerFinderTests; + return new ManagerImp (parent, callback, journal); +} +} +} diff --git a/src/ripple/peerfinder/impl/PeerInfo.h b/src/ripple/peerfinder/impl/PeerInfo.h new file mode 100644 index 0000000000..84a55a19ce --- /dev/null +++ b/src/ripple/peerfinder/impl/PeerInfo.h @@ -0,0 +1,70 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_PEERINFO_H_INCLUDED +#define RIPPLE_PEERFINDER_PEERINFO_H_INCLUDED + +#include "../../ripple/types/api/AgedHistory.h" + +#include "../api/Types.h" + +#include + +namespace ripple { +namespace PeerFinder { + +typedef AgedHistory > Endpoints; + +//-------------------------------------------------------------------------- + +// we keep one of these for each connected peer +struct PeerInfo +{ + PeerInfo (PeerID const& id_, + IPEndpoint const& address_, + bool inbound_) + : id (id_) + , address (address_) + , inbound (inbound_) + , whenSendEndpoints (RelativeTime::fromStartup()) + , whenReceiveEndpoints (RelativeTime::fromStartup()) + { + } + + PeerID id; + IPEndpoint address; + bool inbound; + + // The time after which we will send the peer mtENDPOINTS + RelativeTime mutable whenSendEndpoints; + + // The time after which we will accept mtENDPOINTS from the peer + // This is to prevent flooding or spamming. Receipt of mtENDPOINTS + // sooner than the allotted time should impose a load charge. + // + RelativeTime mutable whenReceiveEndpoints; + + // All the Endpoint records we have received from this peer + Endpoints mutable endpoints; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Slots.cpp b/src/ripple/peerfinder/impl/Slots.cpp new file mode 100644 index 0000000000..c2ed4763f7 --- /dev/null +++ b/src/ripple/peerfinder/impl/Slots.cpp @@ -0,0 +1,99 @@ +//------------------------------------------------------------------------------ +/* + 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. +*/ +//============================================================================== + +namespace ripple { +namespace PeerFinder { + +Slots::Slots () + : startTime (0) + , peerCount (0) + , inboundCount (0) + , outboundCount (0) + , outDesired (0) + , inboundSlots (0) + , inboundSlotsMaximum (0) + , m_roundUpwards (Random::getSystemRandom().nextBool()) +{ +} + +void Slots::update (Config const& config) +{ + if (! config.wantIncoming) + { + inboundSlots = 0; + inboundSlotsMaximum = 0; + } + else + { + double outDesiredFraction ( + config.maxPeerCount * (Config::outPercent * .01)); + + if (m_roundUpwards) + outDesired = int (std::ceil (outDesiredFraction)); + else + outDesired = int (std::floor (outDesiredFraction)); + if (outDesired < Config::minOutCount) + outDesired = Config::minOutCount; + + if (config.maxPeerCount >= outDesired) + inboundSlotsMaximum = config.maxPeerCount - outDesired; + else + inboundSlotsMaximum = 0; + + inboundSlots = std::max (inboundSlotsMaximum - inboundCount, 0); + } +} + +void Slots::addPeer (Config const& config, bool inbound) +{ + if (peerCount == 0) + startTime = Time::getCurrentTime(); + + ++peerCount; + if (inbound) + ++inboundCount; + else + ++outboundCount; + + update (config); +} + +void Slots::dropPeer (Config const& config, bool inbound) +{ + --peerCount; + if (inbound) + --inboundCount; + else + --outboundCount; + + if (peerCount == 0) + startTime = Time (0); + + update (config); +} + +uint32 Slots::uptimeMinutes () const +{ + if (startTime.isNotNull()) + return (Time::getCurrentTime()-startTime).inMinutes(); + return 0; +} + +} +} diff --git a/src/ripple/peerfinder/impl/Slots.h b/src/ripple/peerfinder/impl/Slots.h new file mode 100644 index 0000000000..6c2b8edae0 --- /dev/null +++ b/src/ripple/peerfinder/impl/Slots.h @@ -0,0 +1,66 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_SLOTS_H_INCLUDED +#define RIPPLE_PEERFINDER_SLOTS_H_INCLUDED + +#include "../api/Config.h" + +namespace ripple { +namespace PeerFinder { + +class Slots +{ +public: + Slots (); + + void update (Config const& config); + void addPeer (Config const& config, bool inbound); + void dropPeer (Config const& config, bool inbound); + uint32 uptimeMinutes () const; + + // Most recent time when we went from 0 to 1 peers + Time startTime; + + // Current total of connected peers that have HELLOed + int peerCount; + + // The portion of peers which are incoming connections + int inboundCount; + + // The portion of peers which are outgoing connections + int outboundCount; + + // The number of outgoing peer connections we want (calculated) + int outDesired; + + // The number of available incoming slots (calculated) + int inboundSlots; + + // The maximum number of incoming slots (calculated) + int inboundSlotsMaximum; + +private: + bool m_roundUpwards; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Source.h b/src/ripple/peerfinder/impl/Source.h new file mode 100644 index 0000000000..4492f0c425 --- /dev/null +++ b/src/ripple/peerfinder/impl/Source.h @@ -0,0 +1,42 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_SOURCE_H_INCLUDED +#define RIPPLE_PEERFINDER_SOURCE_H_INCLUDED + +namespace ripple { +namespace PeerFinder { + +/** A source of IPEndpoint for peers. */ +class Source +{ +public: + typedef std::vector IPEndpoints; + + virtual ~Source () { } + virtual std::string const& name () = 0; + + virtual void cancel () { } + virtual void fetch (IPEndpoints& list, Journal journal) = 0; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/SourceStrings.cpp b/src/ripple/peerfinder/impl/SourceStrings.cpp new file mode 100644 index 0000000000..3712b97284 --- /dev/null +++ b/src/ripple/peerfinder/impl/SourceStrings.cpp @@ -0,0 +1,71 @@ +//------------------------------------------------------------------------------ +/* + 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 "SourceStrings.h" + +namespace ripple { +namespace PeerFinder { + +class SourceStringsImp : public SourceStrings +{ +public: + SourceStringsImp (std::string const& name, Strings const& strings) + : m_name (name) + , m_strings (strings) + { + } + + ~SourceStringsImp () + { + } + + std::string const& name () + { + return m_name; + } + + void fetch (IPEndpoints& list, Journal journal) + { + list.resize (0); + list.reserve (m_strings.size()); + + for (int i = 0; i < m_strings.size (); ++i) + { + IPEndpoint ep ( + IPEndpoint::from_string_altform ( + m_strings [i])); + if (! ep.empty()) + list.push_back (ep); + } + } + +private: + std::string m_name; + Strings m_strings; +}; + +//------------------------------------------------------------------------------ + +SourceStrings* SourceStrings::New (std::string const& name, Strings const& strings) +{ + return new SourceStringsImp (name, strings); +} + +} +} diff --git a/src/ripple/peerfinder/impl/SourceStrings.h b/src/ripple/peerfinder/impl/SourceStrings.h new file mode 100644 index 0000000000..867d0faf97 --- /dev/null +++ b/src/ripple/peerfinder/impl/SourceStrings.h @@ -0,0 +1,40 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_SOURCESTRINGS_H_INCLUDED +#define RIPPLE_PEERFINDER_SOURCESTRINGS_H_INCLUDED + +#include "Source.h" + +namespace ripple { +namespace PeerFinder { + +/** Provides an IPEndpoint list from a set of strings. */ +class SourceStrings : public Source +{ +public: + typedef std::vector Strings; + + static SourceStrings* New (std::string const& name, Strings const& strings); +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Store.h b/src/ripple/peerfinder/impl/Store.h new file mode 100644 index 0000000000..828d8f2799 --- /dev/null +++ b/src/ripple/peerfinder/impl/Store.h @@ -0,0 +1,42 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_STORE_H_INCLUDED +#define RIPPLE_PEERFINDER_STORE_H_INCLUDED + +namespace ripple { +namespace PeerFinder { + +/** Abstract persistence for PeerFinder data. */ +class Store +{ +public: + virtual ~Store () { } + + virtual void loadLegacyEndpoints ( + std::vector & list) = 0; + + virtual void storeLegacyEndpoints ( + std::vector const& list) = 0; +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/StoreSqdb.h b/src/ripple/peerfinder/impl/StoreSqdb.h new file mode 100644 index 0000000000..61057ce838 --- /dev/null +++ b/src/ripple/peerfinder/impl/StoreSqdb.h @@ -0,0 +1,202 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_STORESQDB_H_INCLUDED +#define RIPPLE_PEERFINDER_STORESQDB_H_INCLUDED + +#include "beast/modules/beast_sqdb/beast_sqdb.h" + +#include "Store.h" + +namespace ripple { +namespace PeerFinder { + +/** Database persistence for PeerFinder using SQLite */ +class StoreSqdb : public Store +{ +private: + Journal m_journal; + sqdb::session m_session; + +public: + explicit StoreSqdb (Journal journal = Journal()) + : m_journal (journal) + { + } + + ~StoreSqdb () + { + } + + Error open (File const& file) + { + Error error (m_session.open (file.getFullPathName ())); + + if (!error) + error = init (); + + return error; + } + + void loadLegacyEndpoints ( + std::vector & list) + { + list.clear (); + + Error error; + + // Get the count + std::size_t count; + if (! error) + { + m_session.once (error) << + "SELECT COUNT(*) FROM LegacyEndpoints " + ,sqdb::into (count) + ; + } + + if (error) + { + report (error, __FILE__, __LINE__); + return; + } + + list.reserve (count); + + { + std::string s; + sqdb::statement st = (m_session.prepare << + "SELECT ipv4 FROM LegacyEndpoints " + ,sqdb::into (s) + ); + + if (st.execute_and_fetch (error)) + { + do + { + IPEndpoint ep (IPEndpoint::from_string (s)); + if (! ep.empty()) + list.push_back (ep); + } + while (st.fetch (error)); + } + } + + if (error) + { + report (error, __FILE__, __LINE__); + } + } + + void storeLegacyEndpoints ( + std::vector const& list) + { + typedef std::vector List; + + Error error; + + sqdb::transaction tr (m_session); + + m_session.once (error) << + "DELETE FROM LegacyEndpoints"; + + if (! error) + { + std::string s; + sqdb::statement st = (m_session.prepare << + "INSERT INTO LegacyEndpoints ( " + " ipv4 " + ") VALUES ( " + " ? " + ");" + ,sqdb::use (s) + ); + + for (List::const_iterator iter (list.begin()); + !error && iter != list.end(); ++iter) + { + IPEndpoint const& ep (*iter); + s = ep.to_string(); + st.execute_and_fetch (error); + } + } + + if (! error) + { + error = tr.commit(); + } + + if (error) + { + tr.rollback (); + report (error, __FILE__, __LINE__); + } + } + +private: + Error init () + { + Error error; + sqdb::transaction tr (m_session); + + if (! error) + { + m_session.once (error) << + "PRAGMA encoding=\"UTF-8\""; + } + + if (! error) + { + m_session.once (error) << + "CREATE TABLE IF NOT EXISTS LegacyEndpoints ( " + " id INTEGER PRIMARY KEY AUTOINCREMENT, " + " ipv4 TEXT UNIQUE NOT NULL " + ");" + ; + } + + if (! error) + { + error = tr.commit(); + } + + if (error) + { + tr.rollback (); + report (error, __FILE__, __LINE__); + } + + return error; + } + + void report (Error const& error, char const* fileName, int lineNumber) + { + if (error) + { + m_journal.error << + "Failure: '"<< error.getReasonText() << "' " << + " at " << Debug::getSourceLocation (fileName, lineNumber); + } + } +}; + +} +} + +#endif diff --git a/src/ripple/peerfinder/impl/Tests.cpp b/src/ripple/peerfinder/impl/Tests.cpp new file mode 100644 index 0000000000..e84c6e909e --- /dev/null +++ b/src/ripple/peerfinder/impl/Tests.cpp @@ -0,0 +1,42 @@ +//------------------------------------------------------------------------------ +/* + 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 "Logic.h" + +namespace ripple { +namespace PeerFinder { + +class PeerFinderTests : public UnitTest +{ +public: + void runTest () + { + beginTestCase ("logic"); + pass (); + } + + PeerFinderTests () : UnitTest ("PeerFinder", "ripple", runManual) + { + } +}; + +static PeerFinderTests peerFinderTests; + +} +} diff --git a/src/ripple/peerfinder/ripple_peerfinder.cpp b/src/ripple/peerfinder/ripple_peerfinder.cpp new file mode 100644 index 0000000000..4553f547fc --- /dev/null +++ b/src/ripple/peerfinder/ripple_peerfinder.cpp @@ -0,0 +1,37 @@ +//------------------------------------------------------------------------------ +/* + 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_peerfinder.h" + +namespace ripple { +using namespace beast; +} + +# include "impl/Source.h" +# include "impl/SourceStrings.h" +#include "impl/SourceStrings.cpp" +#include "impl/Config.cpp" +#include "impl/Endpoint.cpp" +#include "impl/EndpointCache.cpp" +#include "impl/Manager.cpp" +#include "impl/Slots.cpp" +#include "impl/Tests.cpp" diff --git a/src/ripple/peerfinder/ripple_peerfinder.h b/src/ripple/peerfinder/ripple_peerfinder.h new file mode 100644 index 0000000000..69985f0b42 --- /dev/null +++ b/src/ripple/peerfinder/ripple_peerfinder.h @@ -0,0 +1,37 @@ +//------------------------------------------------------------------------------ +/* + 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_PEERFINDER_H_INCLUDED +#define RIPPLE_PEERFINDER_H_INCLUDED + +#include "beast/modules/beast_core/beast_core.h" + +namespace ripple { +using namespace beast; +} + +#include "../types/api/RipplePublicKey.h" + +#include "api/Types.h" +#include "api/Endpoint.h" +#include "api/Config.h" +#include "api/Callback.h" +#include "api/Manager.h" + +#endif diff --git a/src/ripple/validators/impl/AgedHistory.h b/src/ripple/types/api/AgedHistory.h similarity index 94% rename from src/ripple/validators/impl/AgedHistory.h rename to src/ripple/types/api/AgedHistory.h index 30950aedd9..457b56c9cb 100644 --- a/src/ripple/validators/impl/AgedHistory.h +++ b/src/ripple/types/api/AgedHistory.h @@ -18,11 +18,10 @@ //============================================================================== -#ifndef RIPPLE_VALIDATORS_AGEDHISTORY_H_INCLUDED -#define RIPPLE_VALIDATORS_AGEDHISTORY_H_INCLUDED +#ifndef RIPPLE_TYPES_AGEDHISTORY_H_INCLUDED +#define RIPPLE_TYPES_AGEDHISTORY_H_INCLUDED namespace ripple { -namespace Validators { // Simple container swapping template template @@ -71,7 +70,6 @@ private: Container* m_p2; }; -} } #endif diff --git a/src/ripple/types/api/RipplePublicKey.h b/src/ripple/types/api/RipplePublicKey.h index 4c3ccc92a5..c8a81c89fd 100644 --- a/src/ripple/types/api/RipplePublicKey.h +++ b/src/ripple/types/api/RipplePublicKey.h @@ -21,6 +21,9 @@ #ifndef RIPPLE_TYPES_RIPPLEPUBLICKEY_H_INCLUDED #define RIPPLE_TYPES_RIPPLEPUBLICKEY_H_INCLUDED +#include "CryptoIdentifier.h" +#include "IdentifierType.h" + namespace ripple { class RipplePublicKeyTraits diff --git a/src/ripple/types/ripple_types.h b/src/ripple/types/ripple_types.h index 4afeadce0b..f2517bdb5d 100644 --- a/src/ripple/types/ripple_types.h +++ b/src/ripple/types/ripple_types.h @@ -42,6 +42,7 @@ namespace ripple { using namespace beast; } +#include "api/AgedHistory.h" # include "api/Blob.h" # include "api/Base58.h" # include "api/ByteOrder.h" diff --git a/src/ripple/validators/impl/Logic.h b/src/ripple/validators/impl/Logic.h index e09111b567..0e90381d58 100644 --- a/src/ripple/validators/impl/Logic.h +++ b/src/ripple/validators/impl/Logic.h @@ -20,8 +20,6 @@ #ifndef RIPPLE_VALIDATORS_LOGIC_H_INCLUDED #define RIPPLE_VALIDATORS_LOGIC_H_INCLUDED -#include "AgedHistory.h" - namespace ripple { namespace Validators { diff --git a/src/ripple/validators/impl/Manager.cpp b/src/ripple/validators/impl/Manager.cpp index dde7f8e971..58c7e6c3a0 100644 --- a/src/ripple/validators/impl/Manager.cpp +++ b/src/ripple/validators/impl/Manager.cpp @@ -253,7 +253,7 @@ public: void onPrepare (Journal journal) { #if RIPPLE_USE_NEW_VALIDATORS - journal.info << "Preparing"; + journal.info << "Validators preparing"; addRPCHandlers(); #endif @@ -262,7 +262,7 @@ public: void onStart (Journal journal) { #if RIPPLE_USE_NEW_VALIDATORS - journal.info << "Starting"; + journal.info << "Validators starting"; // Do this late so the sources have a chance to be added. m_queue.dispatch (bind (&ManagerImp::setCheckSources, this)); @@ -273,7 +273,7 @@ public: void onStop (Journal journal) { - journal.info << "Stopping"; + journal.info << "Validators stopping"; if (this->Thread::isThreadRunning()) { diff --git a/src/ripple/validators/impl/StoreSqdb.cpp b/src/ripple/validators/impl/StoreSqdb.cpp index 51e6602c37..ef62b63264 100644 --- a/src/ripple/validators/impl/StoreSqdb.cpp +++ b/src/ripple/validators/impl/StoreSqdb.cpp @@ -159,11 +159,12 @@ void StoreSqdb::update (SourceDesc& desc, bool updateFetchResults) if (! error) { - error = tr.commit (); + error = tr.commit(); } if (error) { + tr.rollback (); report (error, __FILE__, __LINE__); } } diff --git a/src/ripple_app/main/Application.cpp b/src/ripple_app/main/Application.cpp index 7fddf75f02..e50d506b54 100644 --- a/src/ripple_app/main/Application.cpp +++ b/src/ripple_app/main/Application.cpp @@ -51,8 +51,7 @@ class ApplicationImp : public Application , public RootStoppable , public DeadlineTimer::Listener - , LeakChecked - , PeerFinder::Callback + , public LeakChecked { private: static ApplicationImp* s_instance; @@ -136,8 +135,6 @@ public: , m_loadManager (LoadManager::New (*this, LogJournal::get ())) - , mPeerFinder (PeerFinder::New (*this)) - , m_sweepTimer (this) , mShutdown (false) @@ -283,11 +280,6 @@ public: return *m_peers; } - PeerFinder& getPeerFinder () - { - return *mPeerFinder; - } - // VFALCO TODO Move these to the .cpp bool running () { @@ -861,7 +853,6 @@ private: ScopedPointer mValidations; ScopedPointer mProofOfWorkFactory; ScopedPointer m_loadManager; - ScopedPointer mPeerFinder; DeadlineTimer m_sweepTimer; bool volatile mShutdown; diff --git a/src/ripple_app/main/Application.h b/src/ripple_app/main/Application.h index 5e7c08fca4..0db7b24225 100644 --- a/src/ripple_app/main/Application.h +++ b/src/ripple_app/main/Application.h @@ -43,7 +43,6 @@ class SerializedLedgerEntry; class TransactionMaster; class TxQueue; class LocalCredentials; -class PeerFinder; class DatabaseCon; @@ -113,7 +112,6 @@ public: virtual TransactionMaster& getMasterTransaction () = 0; virtual TxQueue& getTxQueue () = 0; virtual LocalCredentials& getLocalCredentials () = 0; - virtual PeerFinder& getPeerFinder () = 0; virtual DatabaseCon* getRpcDB () = 0; virtual DatabaseCon* getTxnDB () = 0; diff --git a/src/ripple_app/peers/Peer.cpp b/src/ripple_app/peers/Peer.cpp index 9cbc31bb71..11b9eda255 100644 --- a/src/ripple_app/peers/Peer.cpp +++ b/src/ripple_app/peers/Peer.cpp @@ -17,13 +17,12 @@ */ //============================================================================== - SETUP_LOG (Peer) class PeerImp; // Don't try to run past receiving nonsense from a peer -// #define TRUST_NETWORK +// #define TRUST_NETWORKHEAD // Node has this long to verify its identity from connection accepted or connection attempt. #define NODE_VERIFY_SECONDS 15 @@ -41,11 +40,6 @@ private: public: - //--------------------------------------------------------------------------- - // - // - // - ScopedPointer m_socket; boost::asio::io_service::strand m_strand; @@ -64,12 +58,6 @@ public: return *m_socket; } - // - // - // - //--------------------------------------------------------------------------- - -public: PeerImp (boost::asio::io_service& io_service, boost::asio::ssl::context& ssl_context, uint64 peerID, @@ -89,6 +77,7 @@ public: , mMinLedger (0) , mMaxLedger (0) , mActivityTimer (io_service) + , m_remoteAddressSet (false) { WriteLog (lsDEBUG, Peer) << "CREATING PEER: " << addressToString (this); } @@ -126,6 +115,9 @@ private: protocol::TMStatusChange mLastStatus; protocol::TMHello mHello; + bool m_remoteAddressSet; + IPEndpoint m_remoteAddress; + public: static char const* getCountedObjectName () { return "Peer"; } @@ -308,6 +300,14 @@ private: if (proxyInfo.protocol == "TCP4") { + m_remoteAddressSet = true; + m_remoteAddress = IPEndpoint (IPEndpoint::V4 ( + proxyInfo.sourceAddress.value [0], + proxyInfo.sourceAddress.value [1], + proxyInfo.sourceAddress.value [2], + proxyInfo.sourceAddress.value [3]), + proxyInfo.sourcePort); + // Set remote IP and port number from PROXY handshake mIpPort.first = proxyInfo.sourceAddress.toString ().toStdString (); mIpPort.second = proxyInfo.sourcePort; @@ -362,6 +362,7 @@ private: void recvGetContacts (protocol::TMGetContacts & packet); void recvGetPeers (protocol::TMGetPeers & packet, Application::ScopedLockType& masterLockHolder); void recvPeers (protocol::TMPeers & packet); + void recvEndpoints (protocol::TMEndpoints & packet); void recvGetObjectByHash (const boost::shared_ptr& packet); void recvPing (protocol::TMPing & packet); void recvErrorMessage (protocol::TMErrorMsg & packet); @@ -390,7 +391,7 @@ void PeerImp::handleWrite (const boost::system::error_code& error, size_t bytes_ // Call on IO strand #ifdef BEAST_DEBUG // if (!error) - // Log::out() << "PeerImp::handleWrite bytes: "<< bytes_transferred; + // Log::out() << "Peer::handleWrite bytes: "<< bytes_transferred; #endif mSendingPacket.reset (); @@ -463,6 +464,8 @@ void PeerImp::detach (const char* rsn, bool onIOStrand) { getApp().getPeers ().peerDisconnected (shared_from_this (), mNodePublic); + getApp().getPeers().getPeerFinder().onPeerDisconnected (RipplePublicKey (mNodePublic)); + mNodePublic.clear (); // Be idempotent. } @@ -739,9 +742,9 @@ void PeerImp::processReadBuffer () // Log::out() << "PRB(" << type << "), len=" << (mReadbuf.size()-PackedMessage::kHeaderBytes); #endif - // Log::out() << "PeerImp::processReadBuffer: " << mIpPort.first << " " << mIpPort.second; + // Log::out() << "Peer::processReadBuffer: " << mIpPort.first << " " << mIpPort.second; - LoadEvent::autoptr event (getApp().getJobQueue ().getLoadEventAP (jtPEER, "PeerImp::read")); + LoadEvent::autoptr event (getApp().getJobQueue ().getLoadEventAP (jtPEER, "Peer::read")); { Application::ScopedLockType lock (getApp ().getMasterLock (), __FILE__, __LINE__); @@ -758,7 +761,7 @@ void PeerImp::processReadBuffer () { case protocol::mtHELLO: { - event->reName ("PeerImp::hello"); + event->reName ("Peer::hello"); protocol::TMHello msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -770,7 +773,7 @@ void PeerImp::processReadBuffer () case protocol::mtCLUSTER: { - event->reName ("PeerImp::cluster"); + event->reName ("Peer::cluster"); protocol::TMCluster msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -781,7 +784,7 @@ void PeerImp::processReadBuffer () case protocol::mtERROR_MSG: { - event->reName ("PeerImp::errormessage"); + event->reName ("Peer::errormessage"); protocol::TMErrorMsg msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -793,7 +796,7 @@ void PeerImp::processReadBuffer () case protocol::mtPING: { - event->reName ("PeerImp::ping"); + event->reName ("Peer::ping"); protocol::TMPing msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -805,7 +808,7 @@ void PeerImp::processReadBuffer () case protocol::mtGET_CONTACTS: { - event->reName ("PeerImp::getcontacts"); + event->reName ("Peer::getcontacts"); protocol::TMGetContacts msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -817,7 +820,7 @@ void PeerImp::processReadBuffer () case protocol::mtCONTACT: { - event->reName ("PeerImp::contact"); + event->reName ("Peer::contact"); protocol::TMContact msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -829,7 +832,7 @@ void PeerImp::processReadBuffer () case protocol::mtGET_PEERS: { - event->reName ("PeerImp::getpeers"); + event->reName ("Peer::getpeers"); protocol::TMGetPeers msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -841,7 +844,7 @@ void PeerImp::processReadBuffer () case protocol::mtPEERS: { - event->reName ("PeerImp::peers"); + event->reName ("Peer::peers"); protocol::TMPeers msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -851,9 +854,21 @@ void PeerImp::processReadBuffer () } break; + case protocol::mtENDPOINTS: + { + event->reName ("Peer::endpoints"); + protocol::TMEndpoints msg; + + if(msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size() - PackedMessage::kHeaderBytes)) + recvEndpoints (msg); + else + WriteLog (lsWARNING, Peer) << "parse error: " << type; + } + break; + case protocol::mtSEARCH_TRANSACTION: { - event->reName ("PeerImp::searchtransaction"); + event->reName ("Peer::searchtransaction"); protocol::TMSearchTransaction msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -865,7 +880,7 @@ void PeerImp::processReadBuffer () case protocol::mtGET_ACCOUNT: { - event->reName ("PeerImp::getaccount"); + event->reName ("Peer::getaccount"); protocol::TMGetAccount msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -877,7 +892,7 @@ void PeerImp::processReadBuffer () case protocol::mtACCOUNT: { - event->reName ("PeerImp::account"); + event->reName ("Peer::account"); protocol::TMAccount msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -889,7 +904,7 @@ void PeerImp::processReadBuffer () case protocol::mtTRANSACTION: { - event->reName ("PeerImp::transaction"); + event->reName ("Peer::transaction"); protocol::TMTransaction msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -901,7 +916,7 @@ void PeerImp::processReadBuffer () case protocol::mtSTATUS_CHANGE: { - event->reName ("PeerImp::statuschange"); + event->reName ("Peer::statuschange"); protocol::TMStatusChange msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -913,7 +928,7 @@ void PeerImp::processReadBuffer () case protocol::mtPROPOSE_LEDGER: { - event->reName ("PeerImp::propose"); + event->reName ("Peer::propose"); boost::shared_ptr msg = boost::make_shared (); if (msg->ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -925,7 +940,7 @@ void PeerImp::processReadBuffer () case protocol::mtGET_LEDGER: { - event->reName ("PeerImp::getledger"); + event->reName ("Peer::getledger"); protocol::TMGetLedger msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -937,7 +952,7 @@ void PeerImp::processReadBuffer () case protocol::mtLEDGER_DATA: { - event->reName ("PeerImp::ledgerdata"); + event->reName ("Peer::ledgerdata"); boost::shared_ptr msg = boost::make_shared (); if (msg->ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -949,7 +964,7 @@ void PeerImp::processReadBuffer () case protocol::mtHAVE_SET: { - event->reName ("PeerImp::haveset"); + event->reName ("Peer::haveset"); protocol::TMHaveTransactionSet msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -961,7 +976,7 @@ void PeerImp::processReadBuffer () case protocol::mtVALIDATION: { - event->reName ("PeerImp::validation"); + event->reName ("Peer::validation"); boost::shared_ptr msg = boost::make_shared (); if (msg->ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -987,7 +1002,7 @@ void PeerImp::processReadBuffer () case protocol::mtGET_OBJECTS: { - event->reName ("PeerImp::getobjects"); + event->reName ("Peer::getobjects"); boost::shared_ptr msg = boost::make_shared (); if (msg->ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -999,7 +1014,7 @@ void PeerImp::processReadBuffer () case protocol::mtPROOFOFWORK: { - event->reName ("PeerImp::proofofwork"); + event->reName ("Peer::proofofwork"); protocol::TMProofWork msg; if (msg.ParseFromArray (&mReadbuf[PackedMessage::kHeaderBytes], mReadbuf.size () - PackedMessage::kHeaderBytes)) @@ -1011,7 +1026,7 @@ void PeerImp::processReadBuffer () default: - event->reName ("PeerImp::unknown"); + event->reName ("Peer::unknown"); WriteLog (lsWARNING, Peer) << "Unknown Msg: " << type; WriteLog (lsWARNING, Peer) << strHex (&mReadbuf[0], mReadbuf.size ()); } @@ -1033,7 +1048,6 @@ void PeerImp::recvHello (protocol::TMHello& packet) uint32 maxTime = ourTime + 20; #ifdef BEAST_DEBUG - if (packet.has_nettime ()) { int64 to = ourTime; @@ -1137,6 +1151,23 @@ void PeerImp::recvHello (protocol::TMHello& packet) // Consider us connected. No longer accepting mtHELLO. mHelloed = true; + { + if (! m_remoteAddressSet) + { + boost::asio::ip::address addr ( + getNativeSocket().remote_endpoint().address()); + if (addr.is_v4()) + { + boost::asio::ip::address_v4::bytes_type bytes (addr.to_v4().to_bytes()); + m_remoteAddress = IPEndpoint (IPEndpoint::V4 ( + bytes[0], bytes[1], bytes[2], bytes[3]), 0); + } + } + + getApp().getPeers().getPeerFinder().onPeerConnected ( + RipplePublicKey (mNodePublic), m_remoteAddress, m_isInbound); + } + // XXX Set timer: connection is in grace period to be useful. // XXX Set timer: connection idle (idle may vary depending on connection type.) @@ -1617,6 +1648,12 @@ void PeerImp::recvPeers (protocol::TMPeers& packet) addr.s_addr = packet.nodes (i).ipv4 (); + { + IPEndpoint::V4 v4 (ntohl (addr.s_addr)); + IPEndpoint ep (v4, packet.nodes (i).ipv4port ()); + getApp().getPeers().getPeerFinder().onPeerLegacyEndpoint (ep); + } + std::string strIP (inet_ntoa (addr)); int iPort = packet.nodes (i).ipv4port (); @@ -1629,6 +1666,58 @@ void PeerImp::recvPeers (protocol::TMPeers& packet) } } +void PeerImp::recvEndpoints (protocol::TMEndpoints& packet) +{ + std::vector endpoints; + + endpoints.reserve (packet.endpoints().size()); + + for (int i = 0; i < packet.endpoints ().size (); ++i) + { + PeerFinder::Endpoint endpoint; + protocol::TMEndpoint const& tm (packet.endpoints(i)); + + // hops + endpoint.hops = tm.hops(); + + // ipv4 + if (endpoint.hops > 0) + { + in_addr addr; + addr.s_addr = tm.ipv4().ipv4(); + IPEndpoint::V4 v4 (ntohl (addr.s_addr)); + endpoint.address = IPEndpoint (v4, 0); + endpoint.port = tm.ipv4().ipv4port (); + } + else + { + // This Endpoint describes the peer we are connected to. + // We will take the remote address seen on the socket and + // store that in the Endpoint. If this is the first time, + // then we'll verify that their listener can receive incoming + // by performing a connectivity test. + // + bassert (m_remoteAddressSet); + endpoint.address = m_remoteAddress.withPort (0); + endpoint.port = tm.ipv4().ipv4port (); + } + + // slots + endpoint.incomingSlotsAvailable = tm.slots(); + + // maxSlots + endpoint.incomingSlotsMax = tm.maxslots(); + + // uptimeMinutes + endpoint.uptimeMinutes = tm.uptimeminutes(); + + endpoints.push_back (endpoint); + } + + getApp().getPeers().getPeerFinder().onPeerEndpoints ( + PeerFinder::PeerID (mNodePublic), endpoints); +} + void PeerImp::recvGetObjectByHash (const boost::shared_ptr& ptr) { protocol::TMGetObjectByHash& packet = *ptr; diff --git a/src/ripple_app/peers/Peers.cpp b/src/ripple_app/peers/Peers.cpp index ec0e8339ec..5031fd752b 100644 --- a/src/ripple_app/peers/Peers.cpp +++ b/src/ripple_app/peers/Peers.cpp @@ -17,26 +17,82 @@ */ //============================================================================== - SETUP_LOG (Peers) +class PeerFinderLog; +template <> char const* LogPartition::getPartitionName () { return "PeerFinder"; } + class PeersImp : public Peers , public Stoppable + , public PeerFinder::Callback , public LeakChecked { public: enum { - /** Frequency of policy enforcement. - */ + /** Frequency of policy enforcement. */ policyIntervalSeconds = 5 }; + typedef RippleRecursiveMutex LockType; + typedef LockType::ScopedLockType ScopedLockType; + typedef std::pair naPeer; + typedef std::pair pipPeer; + typedef std::map::value_type vtPeer; + + ScopedPointer m_peerFinder; + + boost::asio::io_service& m_io_service; + boost::asio::ssl::context& m_ssl_context; + + LockType mPeerLock; + + uint64 mLastPeer; + int mPhase; + + // PeersImp we are connecting with and non-thin peers we are connected to. + // Only peers we know the connection ip for are listed. + // We know the ip and port for: + // - All outbound connections + // - Some inbound connections (which we figured out). + boost::unordered_map mIpMap; + + // Non-thin peers which we are connected to. + // PeersImp we have the public key for. + typedef boost::unordered_map::value_type vtConMap; + boost::unordered_map mConnectedMap; + + // Connections with have a 64-bit identifier + boost::unordered_map mPeerIdMap; + + Peer::pointer mScanning; + boost::asio::deadline_timer mScanTimer; + std::string mScanIp; + int mScanPort; + + void scanHandler (const boost::system::error_code& ecResult); + + boost::asio::deadline_timer mPolicyTimer; + + void policyHandler (const boost::system::error_code& ecResult); + + // PeersImp we are establishing a connection with as a client. + // int miConnectStarting; + + bool peerAvailable (std::string& strIp, int& iPort); + bool peerScanSet (const std::string& strIp, int iPort); + + Peer::pointer peerConnect (const std::string& strIp, int iPort); + + //-------------------------------------------------------------------------- + PeersImp (Stoppable& parent, boost::asio::io_service& io_service, boost::asio::ssl::context& ssl_context) : Stoppable ("Peers", parent) + , m_peerFinder (PeerFinder::Manager::New ( + *this, *this, LogJournal::get ())) , m_io_service (io_service) , m_ssl_context (ssl_context) , mPeerLock (this, "PeersImp", __FILE__, __LINE__) @@ -47,6 +103,127 @@ public: { } + //-------------------------------------------------------------------------- + // + // PeerFinder + // + + // Maps Config settings to PeerFinder::Config + void preparePeerFinder() + { + PeerFinder::Config config; + + // config.maxPeerCount = ? + + config.wantIncoming = + (! getConfig ().PEER_PRIVATE) && + (getConfig().peerListeningPort != 0); + + if (config.wantIncoming) + config.listeningPort = getConfig().peerListeningPort; + + config.featureList = ""; + + m_peerFinder->setConfig (config); + + // Add the static IPs from the rippled.cfg file + m_peerFinder->addStrings ("rippled.cfg", getConfig().IPS); + } + + void sendPeerEndpoints (PeerFinder::PeerID const& id, + std::vector const& endpoints) + { + bassert (! endpoints.empty()); + + typedef std::vector List; + protocol::TMEndpoints tm; + + for (List::const_iterator iter (endpoints.begin()); + iter != endpoints.end(); ++iter) + { + PeerFinder::Endpoint const& ep (*iter); + protocol::TMEndpoint& tme (*tm.add_endpoints()); + + if (ep.address.isV4()) + tme.mutable_ipv4()->set_ipv4( + toNetworkByteOrder (ep.address.v4().value)); + else + tme.mutable_ipv4()->set_ipv4(0); + tme.mutable_ipv4()->set_ipv4port (ep.port); + + tme.set_hops (ep.hops); + tme.set_slots (ep.incomingSlotsAvailable); + tme.set_maxslots (ep.incomingSlotsMax); + tme.set_uptimeminutes (ep.uptimeMinutes); + tme.set_features (ep.featureList); + } + + PackedMessage::pointer msg ( + boost::make_shared ( + tm, protocol::mtENDPOINTS)); + + std::vector list = getPeerVector (); + BOOST_FOREACH (Peer::ref peer, list) + { + if (peer->isConnected() && + PeerFinder::PeerID (peer->getNodePublic()) == id) + { + peer->sendPacket (msg, false); + break; + } + } + } + + void connectPeerEndpoints (std::vector const& list) + { + typedef std::vector List; + + for (List::const_iterator iter (list.begin()); + iter != list.end(); ++iter) + peerConnect (iter->withPort (0), iter->port()); + } + + void chargePeerLoadPenalty (PeerFinder::PeerID const& id) + { + std::vector list = getPeerVector (); + BOOST_FOREACH (Peer::ref peer, list) + { + if (peer->isConnected() && + PeerFinder::PeerID (peer->getNodePublic()) == id) + { + peer->applyLoadCharge (LT_UnwantedData); + break; + } + } + } + + //-------------------------------------------------------------------------- + // + // Stoppable + // + + void onPrepare (Journal) + { + preparePeerFinder(); + } + + void onStart (Journal) + { + } + + void onStop (Journal) + { + // VFALCO TODO Clean this up and do it right, based on sockets + stopped(); + } + + //-------------------------------------------------------------------------- + + PeerFinder::Manager& getPeerFinder() + { + return *m_peerFinder; + } + // Begin enforcing connection policy. void start (); @@ -102,55 +279,6 @@ public: // configured connections void makeConfigured (); - -private: - typedef RippleRecursiveMutex LockType; - typedef LockType::ScopedLockType ScopedLockType; - typedef std::pair naPeer; - typedef std::pair pipPeer; - typedef std::map::value_type vtPeer; - - boost::asio::io_service& m_io_service; - boost::asio::ssl::context& m_ssl_context; - - LockType mPeerLock; - - uint64 mLastPeer; - int mPhase; - - // PeersImp we are connecting with and non-thin peers we are connected to. - // Only peers we know the connection ip for are listed. - // We know the ip and port for: - // - All outbound connections - // - Some inbound connections (which we figured out). - boost::unordered_map mIpMap; - - // Non-thin peers which we are connected to. - // PeersImp we have the public key for. - typedef boost::unordered_map::value_type vtConMap; - boost::unordered_map mConnectedMap; - - // Connections with have a 64-bit identifier - boost::unordered_map mPeerIdMap; - - Peer::pointer mScanning; - boost::asio::deadline_timer mScanTimer; - std::string mScanIp; - int mScanPort; - - void scanHandler (const boost::system::error_code& ecResult); - - boost::asio::deadline_timer mPolicyTimer; - - void policyHandler (const boost::system::error_code& ecResult); - - // PeersImp we are establishing a connection with as a client. - // int miConnectStarting; - - bool peerAvailable (std::string& strIp, int& iPort); - bool peerScanSet (const std::string& strIp, int iPort); - - Peer::pointer peerConnect (const std::string& strIp, int iPort); }; void splitIpPort (const std::string& strIpPort, std::string& strIp, int& iPort) @@ -167,16 +295,18 @@ void PeersImp::start () if (getConfig ().RUN_STANDALONE) return; +#if ! RIPPLE_USE_PEERFINDER // Start running policy. policyEnforce (); // Start scanning. scanRefresh (); +#endif } bool PeersImp::getTopNAddrs (int n, std::vector& addrs) { - +#if ! RIPPLE_USE_PEERFINDER // Try current connections first std::vector peers = getPeerVector(); BOOST_FOREACH(Peer::ref peer, peers) @@ -206,6 +336,7 @@ bool PeersImp::getTopNAddrs (int n, std::vector& addrs) } // FIXME: Should uniqify addrs +#endif return true; } @@ -214,6 +345,7 @@ bool PeersImp::savePeer (const std::string& strIp, int iPort, char code) { bool bNew = false; +#if ! RIPPLE_USE_PEERFINDER Database* db = getApp().getWalletDB ()->getDB (); std::string ipAndPort = sqlEscape (str (boost::format ("%s %d") % strIp % iPort)); @@ -246,6 +378,7 @@ bool PeersImp::savePeer (const std::string& strIp, int iPort, char code) if (bNew) scanRefresh (); +#endif return bNew; } @@ -362,6 +495,7 @@ void PeersImp::policyLowWater () void PeersImp::policyEnforce () { +#if ! RIPPLE_USE_PEERFINDER // Cancel any in progress timer. (void) mPolicyTimer.cancel (); @@ -378,6 +512,7 @@ void PeersImp::policyEnforce () // Schedule next enforcement. mPolicyTimer.expires_at (boost::posix_time::second_clock::universal_time () + boost::posix_time::seconds (policyIntervalSeconds)); mPolicyTimer.async_wait (BIND_TYPE (&PeersImp::policyHandler, this, P_1)); +#endif } void PeersImp::policyHandler (const boost::system::error_code& ecResult) @@ -829,6 +964,7 @@ void PeersImp::makeConfigured () // Scan ips as per db entries. void PeersImp::scanRefresh () { +#if ! RIPPLE_USE_PEERFINDER if (getConfig ().RUN_STANDALONE) { nothing (); @@ -922,6 +1058,7 @@ void PeersImp::scanRefresh () mScanTimer.async_wait (BIND_TYPE (&PeersImp::scanHandler, this, P_1)); } } +#endif } //------------------------------------------------------------------------------ diff --git a/src/ripple_app/peers/Peers.h b/src/ripple_app/peers/Peers.h index 52d95aff39..3614cf070a 100644 --- a/src/ripple_app/peers/Peers.h +++ b/src/ripple_app/peers/Peers.h @@ -17,12 +17,14 @@ */ //============================================================================== - #ifndef RIPPLE_PEERS_H_INCLUDED #define RIPPLE_PEERS_H_INCLUDED -/** Manages the set of connected peers. -*/ +namespace PeerFinder { +class Manager; +} + +/** Manages the set of connected peers. */ class Peers { public: @@ -32,6 +34,8 @@ public: virtual ~Peers () { } + virtual PeerFinder::Manager& getPeerFinder() = 0; + // Begin enforcing connection policy. virtual void start () = 0; diff --git a/src/ripple_app/ripple_app_pt5.cpp b/src/ripple_app/ripple_app_pt5.cpp index 8a92f4080f..7ae574f835 100644 --- a/src/ripple_app/ripple_app_pt5.cpp +++ b/src/ripple_app/ripple_app_pt5.cpp @@ -23,6 +23,7 @@ #include "ripple_app.h" #include "../ripple_net/ripple_net.h" +#include "../ripple/peerfinder/ripple_peerfinder.h" #include "../ripple/validators/ripple_validators.h" namespace ripple @@ -40,5 +41,6 @@ namespace ripple # include "misc/ProofOfWorkFactory.h" #include "peers/Peer.cpp" #include "peers/PackedMessage.cpp" +#include "peers/Peers.cpp" } diff --git a/src/ripple_app/ripple_app_pt7.cpp b/src/ripple_app/ripple_app_pt7.cpp index 8fd560f287..e53910ed65 100644 --- a/src/ripple_app/ripple_app_pt7.cpp +++ b/src/ripple_app/ripple_app_pt7.cpp @@ -29,7 +29,6 @@ namespace ripple #include "ledger/LedgerHistory.cpp" #include "misc/SerializedLedger.cpp" #include "tx/TransactionAcquire.cpp" -#include "peers/Peers.cpp" # include "tx/TxQueueEntry.h" # include "tx/TxQueue.h" diff --git a/src/ripple_core/peerfinder/PeerFinder.h b/src/ripple_core/peerfinder/PeerFinder.h deleted file mode 100644 index 59104ac868..0000000000 --- a/src/ripple_core/peerfinder/PeerFinder.h +++ /dev/null @@ -1,159 +0,0 @@ -//------------------------------------------------------------------------------ -/* - 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_PEERFINDER_H_INCLUDED -#define RIPPLE_PEERFINDER_H_INCLUDED - -/** The identifier we use to track peers in peerfinder -*/ -typedef uint160 PeerId; - - -/** Maintains a set of IP addresses used for getting into the network. -*/ -class PeerFinder : public Uncopyable -{ -public: - //-------------------------------------------------------------------------- - - /** Describes the state of our currently connected peers - */ - struct Connections - { - int numberIncoming; // number of inbound Peers - int numberOutgoing; // number of outbound Peers - - inline int numberTotal () const noexcept - { - return numberIncoming + numberOutgoing; - } - }; - - //-------------------------------------------------------------------------- - - /** An abstract address that can be turned into a socket endpoint. */ - struct Address - { - virtual String asString () = 0; - }; - - /** An IPv4 address. */ -#if 0 - struct AddressIPv4 : Address - { - AddressIPv4 (InputParser::IPv4Address const& address, uint16 port) - : m_address (address) - , m_port (port) - { - - } - - String asString () - { - return String () + - String (m_address.value [0]) + "." + - String (m_address.value [1]) + "." + - String (m_address.value [2]) + "." + - String (m_address.value [3]) + ":" + - String (m_port); - } - - private: - InputParser::IPv4Address m_address; - uint16 m_port; - }; -#endif - - //-------------------------------------------------------------------------- - - /** The Callback receives Peerfinder notifications. - The notifications are sent on a thread owned by the PeerFinder, - so it is best not to do too much work in here. Just post functor - to another worker thread or job queue and return. - */ - struct Callback - { - /** Announces our listening ip/port combinations to the network. - - @param address The address to broadcast. - */ - virtual void onAnnounceAddress () = 0; - - /** Indicates whether or not incoming connections should be accepted. - When we are full on incoming connections, future incoming - connections from valid peers should be politely turned away, - after giving them a random sample of other addresses to try - from our cache. - */ - //virtual void onSetAcceptStatus (bool shouldAcceptIncoming) = 0; - - /** Called periodically to update the callback's list of eligible addresses. - This is used for making new outgoing connections, for - handing out addresses to peers, and for periodically seeding the - network wth hop-limited broadcasts of IP addresses. - */ - //virtual void onNewAddressesAvailable (std::vector
const& list) = 0; - }; - - //-------------------------------------------------------------------------- - - /** Create a new PeerFinder object. - */ - static PeerFinder* New (Callback& callback); - - /** Destroy the object. - - Any pending source fetch operations are aborted. - - There may be some listener calls made before the - destructor returns. - */ - virtual ~PeerFinder () { } - - /** Inform the PeerFinder of the status of our connections. - - This call queues an asynchronous operation to the PeerFinder's thread - and returns immediately. Normally this is called by the Peer code - when the counts change. - - Thread-safety: - Safe to call from any thread - - @see Peer - */ - virtual void updateConnectionsStatus (Connections& connections) = 0; - - - /** Called when a new peer connection is established. - Internally, we add the peer to our tracking table, validate that - we can connect to it, and begin advertising it to others after - we are sure that its connection is stable. - */ - virtual void onPeerConnected(const PeerId& id) = 0; - - /** Called when an existing peer connection drops for whatever reason. - Internally, we mark the peer as no longer connected, calculate - stability metrics, and consider whether we should try to reconnect - to it or drop it from our list. - */ - virtual void onPeerDisconnected(const PeerId& id) = 0; -}; - -#endif diff --git a/src/ripple_core/ripple_core.cpp b/src/ripple_core/ripple_core.cpp index 1fc02db779..b9a908129f 100644 --- a/src/ripple_core/ripple_core.cpp +++ b/src/ripple_core/ripple_core.cpp @@ -29,8 +29,6 @@ #include "beast/modules/beast_core/system/BeforeBoost.h" #include #include -#include -#include #include "nodestore/NodeStore.cpp" @@ -47,6 +45,4 @@ namespace ripple #include "functional/LoadEvent.cpp" #include "functional/LoadMonitor.cpp" -#include "peerfinder/PeerFinder.cpp" - } diff --git a/src/ripple_core/ripple_core.h b/src/ripple_core/ripple_core.h index 04fa71d3d0..05dcc2cf54 100644 --- a/src/ripple_core/ripple_core.h +++ b/src/ripple_core/ripple_core.h @@ -44,8 +44,6 @@ namespace ripple # include "functional/LoadType.h" #include "functional/LoadSource.h" -#include "peerfinder/PeerFinder.h" - } #endif diff --git a/src/ripple_data/protocol/ripple.proto b/src/ripple_data/protocol/ripple.proto index 39df5aa5d1..1443d78766 100644 --- a/src/ripple_data/protocol/ripple.proto +++ b/src/ripple_data/protocol/ripple.proto @@ -14,6 +14,7 @@ enum MessageType mtCONTACT = 11; mtGET_PEERS = 12; mtPEERS = 13; + mtENDPOINTS = 14; // operations for 'small' nodes mtSEARCH_TRANSACTION = 20; @@ -231,6 +232,23 @@ message TMPeers repeated TMIPv4EndPoint nodes = 1; } +// An Endpoint describes a network peer that can accept incoming connections +message TMEndpoint +{ + required TMIPv4EndPoint ipv4 = 1; + required uint32 hops = 2; + required uint32 slots = 3; // the number of available incoming slots + required uint32 maxSlots = 4; // the maximum number of incoming slots + required uint32 uptimeMinutes = 5; // uptime in minutes + required string features = 6; +} + +// An array of Endpoint messages +message TMEndpoints +{ + repeated TMEndpoint endpoints = 1; +}; + message TMSearchTransaction { required uint32 maxTrans = 1;