This class manages established peer-to-peer connections, handles message exchange, monitors connection health, and graceful shutdown. More...

#include <PeerImp.h>

Inheritance diagram for xrpl::PeerImp:

[legend]

Collaboration diagram for xrpl::PeerImp:

[legend]

Classes
struct	ChargeWithContext

class	Metrics

Public Types
enum class	Tracking { diverged , unknown , converged }
	Whether the peer's view of the ledger converges or diverges from ours. More...

using	ptr = std::shared_ptr< Peer >

using	id_t = std::uint32_t
	Uniquely identifies a peer.

Public Member Functions
	PeerImp (PeerImp const &)=delete

PeerImp &	operator= (PeerImp const &)=delete

	PeerImp (Application &app, id_t id, std::shared_ptr< PeerFinder::Slot > const &slot, http_request_type &&request, PublicKey const &publicKey, ProtocolVersion protocol, Resource::Consumer consumer, std::unique_ptr< stream_type > &&stream_ptr, OverlayImpl &overlay)
	Create an active incoming peer from an established ssl connection.

template<class Buffers >
	PeerImp (Application &app, std::unique_ptr< stream_type > &&stream_ptr, Buffers const &buffers, std::shared_ptr< PeerFinder::Slot > &&slot, http_response_type &&response, Resource::Consumer usage, PublicKey const &publicKey, ProtocolVersion protocol, id_t id, OverlayImpl &overlay)
	Create outgoing, handshaked peer.

virtual	~PeerImp ()

beast::Journal const &	pJournal () const

std::shared_ptr< PeerFinder::Slot > const &	slot ()

virtual void	run ()

void	stop () override

void	send (std::shared_ptr< Message > const &m) override

void	sendTxQueue () override
	Send aggregated transactions' hashes.

void	addTxQueue (uint256 const &hash) override
	Add transaction's hash to the transactions' hashes queue.

void	removeTxQueue (uint256 const &hash) override
	Remove transaction's hash from the transactions' hashes queue.

template<class FwdIt , class = typename std::enable_if_t<std::is_same< typename std::iterator_traits<FwdIt>::value_type, PeerFinder::Endpoint>::value>>
void	sendEndpoints (FwdIt first, FwdIt last)
	Send a set of PeerFinder endpoints as a protocol message.

beast::IP::Endpoint	getRemoteAddress () const override

void	charge (Resource::Charge const &fee, std::string const &context) override
	Adjust this peer's load balance based on the type of load imposed.

Peer::id_t	id () const override

bool	crawl () const
	Returns `true` if this connection will publicly share its IP address.

bool	cluster () const override
	Returns `true` if this connection is a member of the cluster.

void	checkTracking (std::uint32_t validationSeq)
	Check if the peer is tracking.

void	checkTracking (std::uint32_t seq1, std::uint32_t seq2)

PublicKey const &	getNodePublic () const override

std::string	getVersion () const
	Return the version of rippled that the peer is running, if reported.

clock_type::duration	uptime () const

Json::Value	json () override

bool	supportsFeature (ProtocolFeature f) const override

std::optional< std::size_t >	publisherListSequence (PublicKey const &pubKey) const override

void	setPublisherListSequence (PublicKey const &pubKey, std::size_t const seq) override

uint256 const &	getClosedLedgerHash () const override

bool	hasLedger (uint256 const &hash, std::uint32_t seq) const override

void	ledgerRange (std::uint32_t &minSeq, std::uint32_t &maxSeq) const override

bool	hasTxSet (uint256 const &hash) const override

void	cycleStatus () override

bool	hasRange (std::uint32_t uMin, std::uint32_t uMax) override

int	getScore (bool haveItem) const override

bool	isHighLatency () const override

bool	compressionEnabled () const override

bool	txReduceRelayEnabled () const override

void	onMessageUnknown (std::uint16_t type)

void	onMessageBegin (std::uint16_t type, std::shared_ptr<::google::protobuf::Message > const &m, std::size_t size, std::size_t uncompressed_size, bool isCompressed)

void	onMessageEnd (std::uint16_t type, std::shared_ptr<::google::protobuf::Message > const &m)

void	onMessage (std::shared_ptr< protocol::TMManifests > const &m)

void	onMessage (std::shared_ptr< protocol::TMPing > const &m)

void	onMessage (std::shared_ptr< protocol::TMCluster > const &m)

void	onMessage (std::shared_ptr< protocol::TMEndpoints > const &m)

void	onMessage (std::shared_ptr< protocol::TMTransaction > const &m)

void	onMessage (std::shared_ptr< protocol::TMGetLedger > const &m)

void	onMessage (std::shared_ptr< protocol::TMLedgerData > const &m)

void	onMessage (std::shared_ptr< protocol::TMProposeSet > const &m)

void	onMessage (std::shared_ptr< protocol::TMStatusChange > const &m)

void	onMessage (std::shared_ptr< protocol::TMHaveTransactionSet > const &m)

void	onMessage (std::shared_ptr< protocol::TMValidatorList > const &m)

void	onMessage (std::shared_ptr< protocol::TMValidatorListCollection > const &m)

void	onMessage (std::shared_ptr< protocol::TMValidation > const &m)

void	onMessage (std::shared_ptr< protocol::TMGetObjectByHash > const &m)

void	onMessage (std::shared_ptr< protocol::TMHaveTransactions > const &m)

void	onMessage (std::shared_ptr< protocol::TMTransactions > const &m)

void	onMessage (std::shared_ptr< protocol::TMSquelch > const &m)

void	onMessage (std::shared_ptr< protocol::TMProofPathRequest > const &m)

void	onMessage (std::shared_ptr< protocol::TMProofPathResponse > const &m)

void	onMessage (std::shared_ptr< protocol::TMReplayDeltaRequest > const &m)

void	onMessage (std::shared_ptr< protocol::TMReplayDeltaResponse > const &m)

T	shared_from_this (T... args)

T	weak_from_this (T... args)

Private Types
using	clock_type = std::chrono::steady_clock

using	error_code = boost::system::error_code

using	socket_type = boost::asio::ip::tcp::socket

using	middle_type = boost::beast::tcp_stream

using	stream_type = boost::beast::ssl_stream< middle_type >

using	address_type = boost::asio::ip::address

using	endpoint_type = boost::asio::ip::tcp::endpoint

using	waitable_timer = boost::asio::basic_waitable_timer< std::chrono::steady_clock >

using	Compressed = compression::Compressed

Private Member Functions
void	fail (std::string const &name, error_code ec)
	Handles a failure associated with a specific error code.

void	fail (std::string const &reason)
	Handles a failure described by a reason string.

void	shutdown ()
	Initiates the peer disconnection sequence.

void	tryAsyncShutdown ()
	Attempts to perform a graceful SSL shutdown if conditions are met.

void	onShutdown (error_code ec)
	Handles the completion of the asynchronous SSL shutdown.

void	close ()
	Forcibly closes the underlying socket connection.

void	setTimer (std::chrono::seconds interval)
	Sets and starts the peer timer.

void	onTimer (error_code const &ec)
	Handles the expiration of the peer activity timer.

void	cancelTimer () noexcept
	Cancels any pending wait on the peer activity timer.

void	doAccept ()

std::string	name () const

std::string	domain () const

void	doProtocolStart ()

void	onReadMessage (error_code ec, std::size_t bytes_transferred)

void	onWriteMessage (error_code ec, std::size_t bytes_transferred)

void	handleTransaction (std::shared_ptr< protocol::TMTransaction > const &m, bool eraseTxQueue, bool batch)
	Called from onMessage(TMTransaction(s)).

void	handleHaveTransactions (std::shared_ptr< protocol::TMHaveTransactions > const &m)
	Handle protocol message with hashes of transactions that have not been relayed by an upstream node down to its peers - request transactions, which have not been relayed to this peer.

std::string const &	fingerprint () const override

std::string const &	prefix () const

void	addLedger (uint256 const &hash, std::lock_guard< std::mutex > const &lockedRecentLock)

void	doFetchPack (std::shared_ptr< protocol::TMGetObjectByHash > const &packet)

void	onValidatorListMessage (std::string const &messageType, std::string const &manifest, std::uint32_t version, std::vector< ValidatorBlobInfo > const &blobs)

void	doTransactions (std::shared_ptr< protocol::TMGetObjectByHash > const &packet)
	Process peer's request to send missing transactions.

void	checkTransaction (HashRouterFlags flags, bool checkSignature, std::shared_ptr< STTx const > const &stx, bool batch)

void	checkPropose (bool isTrusted, std::shared_ptr< protocol::TMProposeSet > const &packet, RCLCxPeerPos peerPos)

void	checkValidation (std::shared_ptr< STValidation > const &val, uint256 const &key, std::shared_ptr< protocol::TMValidation > const &packet)

void	sendLedgerBase (std::shared_ptr< Ledger const > const &ledger, protocol::TMLedgerData &ledgerData)

std::shared_ptr< Ledger const >	getLedger (std::shared_ptr< protocol::TMGetLedger > const &m)

std::shared_ptr< SHAMap const >	getTxSet (std::shared_ptr< protocol::TMGetLedger > const &m) const

void	processLedgerRequest (std::shared_ptr< protocol::TMGetLedger > const &m)

Static Private Member Functions
static std::string	makePrefix (std::string const &fingerprint)

Private Attributes
Application &	app_

id_t const	id_

std::string	fingerprint_

std::string	prefix_

beast::WrappedSink	sink_

beast::WrappedSink	p_sink_

beast::Journal const	journal_

beast::Journal const	p_journal_

std::unique_ptr< stream_type >	stream_ptr_

socket_type &	socket_

stream_type &	stream_

boost::asio::strand< boost::asio::executor >	strand_

waitable_timer	timer_

beast::IP::Endpoint const	remote_address_

OverlayImpl &	overlay_

bool const	inbound_

ProtocolVersion	protocol_

std::atomic< Tracking >	tracking_

clock_type::time_point	trackingTime_

PublicKey const	publicKey_

std::string	name_

std::shared_mutex	nameMutex_

LedgerIndex	minLedger_ = 0

LedgerIndex	maxLedger_ = 0

uint256	closedLedgerHash_

uint256	previousLedgerHash_

boost::circular_buffer< uint256 >	recentLedgers_ {128}

boost::circular_buffer< uint256 >	recentTxSets_ {128}

std::optional< std::chrono::milliseconds >	latency_

std::optional< std::uint32_t >	lastPingSeq_

clock_type::time_point	lastPingTime_

clock_type::time_point const	creationTime_

reduce_relay::Squelch< UptimeClock >	squelch_

std::mutex	recentLock_

protocol::TMStatusChange	last_status_

Resource::Consumer	usage_

ChargeWithContext	fee_

std::shared_ptr< PeerFinder::Slot > const	slot_

boost::beast::multi_buffer	read_buffer_

http_request_type	request_

http_response_type	response_

boost::beast::http::fields const &	headers_

std::queue< std::shared_ptr< Message > >	send_queue_

bool	shutdown_ = false

bool	shutdownStarted_ = false

bool	readPending_ = false

bool	writePending_ = false

int	large_sendq_ = 0

std::unique_ptr< LoadEvent >	load_event_

hash_map< PublicKey, std::size_t >	publisherListSequences_

Compressed	compressionEnabled_ = Compressed::Off

hash_set< uint256 >	txQueue_

bool	txReduceRelayEnabled_ = false

bool	ledgerReplayEnabled_ = false

LedgerReplayMsgHandler	ledgerReplayMsgHandler_

struct {

Metrics sent

Metrics recv

}	metrics_

Friends
class	OverlayImpl

Detailed Description

This class manages established peer-to-peer connections, handles message exchange, monitors connection health, and graceful shutdown.

The PeerImp shutdown mechanism is a multi-stage process designed to ensure graceful connection termination while handling ongoing I/O operations safely. The shutdown can be initiated from multiple points and follows a deterministic state machine.

The shutdown process can be triggered from several entry points:

External requests: stop() method called by overlay management
Error conditions: fail(error_code) or fail(string) on protocol violations
Timer expiration: Various timeout scenarios (ping timeout, large send queue)
Connection health: Peer tracking divergence or unknown state timeouts

The shutdown follows this progression:

Normal Operation → shutdown() → tryAsyncShutdown() → onShutdown() → close() ↓ ↓ ↓ ↓ Set shutdown_ SSL graceful Timer cancel Socket close Cancel timer shutdown start & cleanup & metrics 5s safety timer Set shutdownStarted_ update

Two primary flags coordinate the shutdown process:

shutdown_: Set when shutdown is requested
shutdownStarted_: Set when SSL shutdown begins

The shutdown mechanism carefully coordinates with ongoing read/write operations:

**Read Operations (onReadMessage)**:

Checks shutdown_ flag after processing each message batch
If shutdown initiated during processing, calls tryAsyncShutdown()

**Write Operations (onWriteMessage)**:

Checks shutdown_ flag before queuing new writes
Calls tryAsyncShutdown() when shutdown flag detected

Multiple timers require coordination during shutdown:

Peer Timer: Regular ping/pong timer cancelled immediately in shutdown()
Shutdown Timer: 5-second safety timer ensures shutdown completion
Operation Cancellation: All pending async operations are cancelled

The shutdown implements fallback mechanisms:

Graceful Path: SSL shutdown → Socket close → Cleanup
Forced Path: If SSL shutdown fails or times out, proceeds to socket close
Safety Timer: 5-second timeout prevents hanging shutdowns

All shutdown operations are serialized through the boost::asio::strand to ensure thread safety. The strand guarantees that shutdown state changes and I/O operation callbacks are executed sequentially.

Note: This class requires careful coordination between async operations, timer management, and shutdown procedures to ensure no resource leaks or hanging connections in high-throughput networking scenarios.

Definition at line 96 of file PeerImp.h.

Member Typedef Documentation

◆ clock_type

using xrpl::PeerImp::clock_type = std::chrono::steady_clock

private

Definition at line 105 of file PeerImp.h.

◆ error_code

using xrpl::PeerImp::error_code = boost::system::error_code

private

Definition at line 106 of file PeerImp.h.

◆ socket_type

using xrpl::PeerImp::socket_type = boost::asio::ip::tcp::socket

private

Definition at line 107 of file PeerImp.h.

◆ middle_type

using xrpl::PeerImp::middle_type = boost::beast::tcp_stream

private

Definition at line 108 of file PeerImp.h.

◆ stream_type

using xrpl::PeerImp::stream_type = boost::beast::ssl_stream<middle_type>

private

Definition at line 109 of file PeerImp.h.

◆ address_type

using xrpl::PeerImp::address_type = boost::asio::ip::address

private

Definition at line 110 of file PeerImp.h.

◆ endpoint_type

using xrpl::PeerImp::endpoint_type = boost::asio::ip::tcp::endpoint

private

Definition at line 111 of file PeerImp.h.

◆ waitable_timer

using xrpl::PeerImp::waitable_timer = boost::asio::basic_waitable_timer<std::chrono::steady_clock>

private

Definition at line 112 of file PeerImp.h.

◆ Compressed

using xrpl::PeerImp::Compressed = compression::Compressed

private

Definition at line 114 of file PeerImp.h.

◆ ptr

using xrpl::Peer::ptr = std::shared_ptr<Peer>

inherited

Definition at line 27 of file xrpld/overlay/Peer.h.

◆ id_t

using xrpl::Peer::id_t = std::uint32_t

inherited

Uniquely identifies a peer.

This can be stored in tables to find the peer later. Callers can discover if the peer is no longer connected and make adjustments as needed.

Definition at line 34 of file xrpld/overlay/Peer.h.

Member Enumeration Documentation

◆ Tracking

enum class xrpl::PeerImp::Tracking

strong

Whether the peer's view of the ledger converges or diverges from ours.

Enumerator
diverged
unknown
converged

Definition at line 102 of file PeerImp.h.

Constructor & Destructor Documentation

◆ PeerImp() [1/3]

xrpl::PeerImp::PeerImp ( PeerImp const & )

delete

◆ PeerImp() [2/3]

xrpl::PeerImp::PeerImp	(	Application &	app,
		id_t	id,
		std::shared_ptr< PeerFinder::Slot > const &	slot,
		http_request_type &&	request,
		PublicKey const &	publicKey,
		ProtocolVersion	protocol,
		Resource::Consumer	consumer,
		std::unique_ptr< stream_type > &&	stream_ptr,
		OverlayImpl &	overlay
	)

Create an active incoming peer from an established ssl connection.

Definition at line 53 of file PeerImp.cpp.

◆ PeerImp() [3/3]

template<class Buffers >

xrpl::PeerImp::PeerImp	(	Application &	app,
		std::unique_ptr< stream_type > &&	stream_ptr,
		Buffers const &	buffers,
		std::shared_ptr< PeerFinder::Slot > &&	slot,
		http_response_type &&	response,
		Resource::Consumer	usage,
		PublicKey const &	publicKey,
		ProtocolVersion	protocol,
		id_t	id,
		OverlayImpl &	overlay
	)

Create outgoing, handshaked peer.

Definition at line 816 of file PeerImp.h.

◆ ~PeerImp()

xrpl::PeerImp::~PeerImp ( )

virtual

Definition at line 121 of file PeerImp.cpp.

Member Function Documentation

◆ operator=()

PeerImp & xrpl::PeerImp::operator= ( PeerImp const & )

delete

◆ pJournal()

beast::Journal const & xrpl::PeerImp::pJournal ( ) const

Definition at line 326 of file PeerImp.h.

◆ slot()

std::shared_ptr< PeerFinder::Slot > const & xrpl::PeerImp::slot ( )

Definition at line 332 of file PeerImp.h.

◆ run()

void xrpl::PeerImp::run ( )

virtual

Reimplemented in xrpl::test::tx_reduce_relay_test::PeerTest.

Definition at line 144 of file PeerImp.cpp.

◆ stop()

void xrpl::PeerImp::stop ( )

overridevirtual

Implements xrpl::OverlayImpl::Child.

Definition at line 202 of file PeerImp.cpp.

◆ send()

void xrpl::PeerImp::send ( std::shared_ptr< Message > const & m )

overridevirtual

Implements xrpl::Peer.

Reimplemented in xrpl::test::tx_reduce_relay_test::PeerTest.

Definition at line 222 of file PeerImp.cpp.

◆ sendTxQueue()

void xrpl::PeerImp::sendTxQueue ( )

overridevirtual

Send aggregated transactions' hashes.

Implements xrpl::Peer.

Definition at line 289 of file PeerImp.cpp.

◆ addTxQueue()

void xrpl::PeerImp::addTxQueue ( uint256 const & hash )

overridevirtual

Add transaction's hash to the transactions' hashes queue.

Parameters

hash	transaction's hash

Implements xrpl::Peer.

Reimplemented in xrpl::test::tx_reduce_relay_test::PeerTest.

Definition at line 308 of file PeerImp.cpp.

◆ removeTxQueue()

void xrpl::PeerImp::removeTxQueue ( uint256 const & hash )

overridevirtual

Remove transaction's hash from the transactions' hashes queue.

Parameters

hash	transaction's hash

Implements xrpl::Peer.

Definition at line 325 of file PeerImp.cpp.

◆ sendEndpoints()

template<class FwdIt , class >

void xrpl::PeerImp::sendEndpoints	(	FwdIt	first,
		FwdIt	last
	)

Send a set of PeerFinder endpoints as a protocol message.

Definition at line 890 of file PeerImp.h.

◆ getRemoteAddress()

beast::IP::Endpoint xrpl::PeerImp::getRemoteAddress ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 378 of file PeerImp.h.

◆ charge()

void xrpl::PeerImp::charge	(	Resource::Charge const &	fee,
		std::string const &	context
	)

overridevirtual

Adjust this peer's load balance based on the type of load imposed.

Implements xrpl::Peer.

Definition at line 337 of file PeerImp.cpp.

◆ id()

Peer::id_t xrpl::PeerImp::id ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 391 of file PeerImp.h.

◆ crawl()

bool xrpl::PeerImp::crawl ( ) const

Returns true if this connection will publicly share its IP address.

Definition at line 351 of file PeerImp.cpp.

◆ cluster()

bool xrpl::PeerImp::cluster ( ) const

overridevirtual

Returns true if this connection is a member of the cluster.

Implements xrpl::Peer.

Definition at line 360 of file PeerImp.cpp.

◆ checkTracking() [1/2]

void xrpl::PeerImp::checkTracking ( std::uint32_t validationSeq )

Check if the peer is tracking.

Parameters

validationSeq The ledger sequence of a recently-validated ledger

Definition at line 2062 of file PeerImp.cpp.

◆ checkTracking() [2/2]

void xrpl::PeerImp::checkTracking	(	std::uint32_t	seq1,
		std::uint32_t	seq2
	)

Definition at line 2081 of file PeerImp.cpp.

◆ getNodePublic()

PublicKey const & xrpl::PeerImp::getNodePublic ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 413 of file PeerImp.h.

◆ getVersion()

std::string xrpl::PeerImp::getVersion ( ) const

Return the version of rippled that the peer is running, if reported.

Definition at line 366 of file PeerImp.cpp.

◆ uptime()

clock_type::duration xrpl::PeerImp::uptime ( ) const

Definition at line 424 of file PeerImp.h.

◆ json()

Json::Value xrpl::PeerImp::json ( )

overridevirtual

Implements xrpl::Peer.

Definition at line 374 of file PeerImp.cpp.

◆ supportsFeature()

bool xrpl::PeerImp::supportsFeature ( ProtocolFeature f ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 492 of file PeerImp.cpp.

◆ publisherListSequence()

std::optional< std::size_t > xrpl::PeerImp::publisherListSequence ( PublicKey const & pubKey ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 436 of file PeerImp.h.

◆ setPublisherListSequence()

void xrpl::PeerImp::setPublisherListSequence	(	PublicKey const &	pubKey,
		std::size_t const	seq
	)

overridevirtual

Implements xrpl::Peer.

Definition at line 447 of file PeerImp.h.

◆ getClosedLedgerHash()

uint256 const & xrpl::PeerImp::getClosedLedgerHash ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 460 of file PeerImp.h.

◆ hasLedger()

bool xrpl::PeerImp::hasLedger	(	uint256 const &	hash,
		std::uint32_t	seq
	)		const

overridevirtual

Implements xrpl::Peer.

Definition at line 509 of file PeerImp.cpp.

◆ ledgerRange()

void xrpl::PeerImp::ledgerRange	(	std::uint32_t &	minSeq,
		std::uint32_t &	maxSeq
	)		const

overridevirtual

Implements xrpl::Peer.

Definition at line 524 of file PeerImp.cpp.

◆ hasTxSet()

bool xrpl::PeerImp::hasTxSet ( uint256 const & hash ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 533 of file PeerImp.cpp.

◆ cycleStatus()

void xrpl::PeerImp::cycleStatus ( )

overridevirtual

Implements xrpl::Peer.

Definition at line 541 of file PeerImp.cpp.

◆ hasRange()

bool xrpl::PeerImp::hasRange	(	std::uint32_t	uMin,
		std::uint32_t	uMax
	)

overridevirtual

Implements xrpl::Peer.

Definition at line 551 of file PeerImp.cpp.

◆ getScore()

int xrpl::PeerImp::getScore ( bool haveItem ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 3574 of file PeerImp.cpp.

◆ isHighLatency()

bool xrpl::PeerImp::isHighLatency ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 3612 of file PeerImp.cpp.

◆ compressionEnabled()

bool xrpl::PeerImp::compressionEnabled ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 488 of file PeerImp.h.

◆ txReduceRelayEnabled()

bool xrpl::PeerImp::txReduceRelayEnabled ( ) const

overridevirtual

Implements xrpl::Peer.

Definition at line 494 of file PeerImp.h.

◆ fail() [1/2]

void xrpl::PeerImp::fail	(	std::string const &	name,
		error_code	ec
	)

private

Handles a failure associated with a specific error code.

This function is called when an operation fails with an error code. It logs the warning message and gracefully shutdowns the connection.

The function will do nothing if the connection is already closed or if a shutdown is already in progress.

Parameters

name	The name of the operation that failed (e.g., "read", "write").
ec	The error code associated with the failure.

Note: This function must be called from within the object's strand.

Definition at line 561 of file PeerImp.cpp.

◆ fail() [2/2]

void xrpl::PeerImp::fail ( std::string const & reason )

private

Handles a failure described by a reason string.

This overload is used for logical errors or protocol violations not associated with a specific error code. It logs a warning with the given reason, then initiates a graceful shutdown.

The function will do nothing if the connection is already closed or if a shutdown is already in progress.

Parameters

reason A descriptive string explaining the reason for the failure.

Note: This function must be called from within the object's strand.

Definition at line 576 of file PeerImp.cpp.

◆ shutdown()

void xrpl::PeerImp::shutdown ( )

private

Initiates the peer disconnection sequence.

This is the primary entry point to start closing a peer connection. It marks the peer for shutdown and cancels any outstanding asynchronous operations. This cancellation allows the graceful shutdown to proceed once the handlers for the cancelled operations have completed.

Note: This method must be called on the peer's strand.

Definition at line 624 of file PeerImp.cpp.

◆ tryAsyncShutdown()

void xrpl::PeerImp::tryAsyncShutdown ( )

private

Attempts to perform a graceful SSL shutdown if conditions are met.

This helper function checks if the peer is in a state where a graceful SSL shutdown can be performed (i.e., shutdown has been requested and no I/O operations are currently in progress).

Note: This method must be called on the peer's strand.

Definition at line 600 of file PeerImp.cpp.

◆ onShutdown()

void xrpl::PeerImp::onShutdown ( error_code ec )

private

Handles the completion of the asynchronous SSL shutdown.

This function is the callback for the async_shutdown operation started in shutdown(). Its first action is to cancel the timer. It then inspects the error code to determine the outcome.

Regardless of the result, this function proceeds to call close() to ensure the underlying socket is fully closed.

Parameters

ec	The error code resulting from the `async_shutdown` operation.

Definition at line 641 of file PeerImp.cpp.

◆ close()

void xrpl::PeerImp::close ( )

private

Forcibly closes the underlying socket connection.

This function provides the final, non-graceful shutdown of the peer connection. It ensures any pending timers are cancelled and then immediately closes the TCP socket, bypassing the SSL shutdown handshake.

After closing, it notifies the overlay manager of the disconnection.

Note: This function must be called from within the object's strand.

Definition at line 667 of file PeerImp.cpp.

◆ setTimer()

void xrpl::PeerImp::setTimer ( std::chrono::seconds interval )

private

Sets and starts the peer timer.

This function starts timer, which is used to detect inactivity and prevent stalled connections. It sets the timer to expire after the predefined peerTimerInterval.

Note: This function will terminate the connection in case of any errors.

Definition at line 693 of file PeerImp.cpp.

◆ onTimer()

void xrpl::PeerImp::onTimer ( error_code const & ec )

private

Handles the expiration of the peer activity timer.

This callback is invoked when the timer set by setTimer expires. It watches the peer connection, checking for various timeout and health conditions.

Parameters

ec	The error code associated with the timer's expiration. `operation_aborted` is expected if the timer was cancelled.

Definition at line 722 of file PeerImp.cpp.

◆ cancelTimer()

void xrpl::PeerImp::cancelTimer ( )

privatenoexcept

Cancels any pending wait on the peer activity timer.

This function is called to stop the timer. It gracefully manages any errors that might occur during the cancellation process.

Definition at line 789 of file PeerImp.cpp.

◆ makePrefix()

std::string xrpl::PeerImp::makePrefix ( std::string const & fingerprint )

staticprivate

Definition at line 714 of file PeerImp.cpp.

◆ doAccept()

void xrpl::PeerImp::doAccept ( )

private

Definition at line 803 of file PeerImp.cpp.

◆ name()

std::string xrpl::PeerImp::name ( ) const

private

Definition at line 873 of file PeerImp.cpp.

◆ domain()

std::string xrpl::PeerImp::domain ( ) const

private

Definition at line 880 of file PeerImp.cpp.

◆ doProtocolStart()

void xrpl::PeerImp::doProtocolStart ( )

private

Definition at line 890 of file PeerImp.cpp.

◆ onReadMessage()

void xrpl::PeerImp::onReadMessage	(	error_code	ec,
		std::size_t	bytes_transferred
	)

private

Definition at line 932 of file PeerImp.cpp.

◆ onWriteMessage()

void xrpl::PeerImp::onWriteMessage	(	error_code	ec,
		std::size_t	bytes_transferred
	)

private

Definition at line 1023 of file PeerImp.cpp.

◆ handleTransaction()

void xrpl::PeerImp::handleTransaction	(	std::shared_ptr< protocol::TMTransaction > const &	m,
		bool	eraseTxQueue,
		bool	batch
	)

private

Called from onMessage(TMTransaction(s)).

Parameters

m	Transaction protocol message
eraseTxQueue	is true when called from onMessage(TMTransaction) and is false when called from onMessage(TMTransactions). If true then the transaction hash is erased from txQueue_. Don't need to erase from the queue when called from onMessage(TMTransactions) because this message is a response to the missing transactions request and the queue would not have any of these transactions.
batch	is false when called from onMessage(TMTransaction) and is true when called from onMessage(TMTransactions). If true, then the transaction is part of a batch, and should not be charged an extra fee.

Definition at line 1339 of file PeerImp.cpp.

◆ handleHaveTransactions()

void xrpl::PeerImp::handleHaveTransactions ( std::shared_ptr< protocol::TMHaveTransactions > const & m )

private

Handle protocol message with hashes of transactions that have not been relayed by an upstream node down to its peers - request transactions, which have not been relayed to this peer.

Parameters

m	protocol message with transactions' hashes

Definition at line 2706 of file PeerImp.cpp.

◆ fingerprint()

std::string const & xrpl::PeerImp::fingerprint ( ) const

overrideprivatevirtual

Implements xrpl::Peer.

Definition at line 676 of file PeerImp.h.

◆ prefix()

std::string const & xrpl::PeerImp::prefix ( ) const

private

Definition at line 682 of file PeerImp.h.

◆ onMessageUnknown()

void xrpl::PeerImp::onMessageUnknown ( std::uint16_t type )

Definition at line 1089 of file PeerImp.cpp.

◆ onMessageBegin()

void xrpl::PeerImp::onMessageBegin	(	std::uint16_t	type,
		std::shared_ptr<::google::protobuf::Message > const &	m,
		std::size_t	size,
		std::size_t	uncompressed_size,
		bool	isCompressed
	)

Definition at line 1095 of file PeerImp.cpp.

◆ onMessageEnd()

void xrpl::PeerImp::onMessageEnd	(	std::uint16_t	type,
		std::shared_ptr<::google::protobuf::Message > const &	m
	)

Definition at line 1138 of file PeerImp.cpp.

◆ onMessage() [1/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMManifests > const & m )

Definition at line 1147 of file PeerImp.cpp.

◆ onMessage() [2/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMPing > const & m )

Definition at line 1167 of file PeerImp.cpp.

◆ onMessage() [3/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMCluster > const & m )

Definition at line 1204 of file PeerImp.cpp.

◆ onMessage() [4/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMEndpoints > const & m )

Definition at line 1276 of file PeerImp.cpp.

◆ onMessage() [5/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMTransaction > const & m )

Definition at line 1333 of file PeerImp.cpp.

◆ onMessage() [6/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMGetLedger > const & m )

Definition at line 1476 of file PeerImp.cpp.

◆ onMessage() [7/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMLedgerData > const & m )

Definition at line 1675 of file PeerImp.cpp.

◆ onMessage() [8/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMProposeSet > const & m )

Definition at line 1766 of file PeerImp.cpp.

◆ onMessage() [9/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMStatusChange > const & m )

Definition at line 1887 of file PeerImp.cpp.

◆ onMessage() [10/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMHaveTransactionSet > const & m )

Definition at line 2103 of file PeerImp.cpp.

◆ onMessage() [11/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMValidatorList > const & m )

Definition at line 2338 of file PeerImp.cpp.

◆ onMessage() [12/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMValidatorListCollection > const & m )

Definition at line 2366 of file PeerImp.cpp.

◆ onMessage() [13/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMValidation > const & m )

Definition at line 2406 of file PeerImp.cpp.

◆ onMessage() [14/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMGetObjectByHash > const & m )

Definition at line 2531 of file PeerImp.cpp.

◆ onMessage() [15/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMHaveTransactions > const & m )

Definition at line 2687 of file PeerImp.cpp.

◆ onMessage() [16/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMTransactions > const & m )

Definition at line 2756 of file PeerImp.cpp.

◆ onMessage() [17/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMSquelch > const & m )

Definition at line 2780 of file PeerImp.cpp.

◆ onMessage() [18/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMProofPathRequest > const & m )

Definition at line 1565 of file PeerImp.cpp.

◆ onMessage() [19/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMProofPathResponse > const & m )

Definition at line 1604 of file PeerImp.cpp.

◆ onMessage() [20/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMReplayDeltaRequest > const & m )

Definition at line 1620 of file PeerImp.cpp.

◆ onMessage() [21/21]

void xrpl::PeerImp::onMessage ( std::shared_ptr< protocol::TMReplayDeltaResponse > const & m )

Definition at line 1659 of file PeerImp.cpp.

◆ addLedger()

void xrpl::PeerImp::addLedger	(	uint256 const &	hash,
		std::lock_guard< std::mutex > const &	lockedRecentLock
	)

private

Definition at line 2826 of file PeerImp.cpp.

◆ doFetchPack()

void xrpl::PeerImp::doFetchPack ( std::shared_ptr< protocol::TMGetObjectByHash > const & packet )

private

Definition at line 2842 of file PeerImp.cpp.

◆ onValidatorListMessage()

void xrpl::PeerImp::onValidatorListMessage	(	std::string const &	messageType,
		std::string const &	manifest,
		std::uint32_t	version,
		std::vector< ValidatorBlobInfo > const &	blobs
	)

private

Definition at line 2129 of file PeerImp.cpp.

◆ doTransactions()

void xrpl::PeerImp::doTransactions ( std::shared_ptr< protocol::TMGetObjectByHash > const & packet )

private

Process peer's request to send missing transactions.

The request is sent in response to TMHaveTransactions.

Parameters

packet protocol message containing missing transactions' hashes.

Definition at line 2876 of file PeerImp.cpp.

◆ checkTransaction()

void xrpl::PeerImp::checkTransaction	(	HashRouterFlags	flags,
		bool	checkSignature,
		std::shared_ptr< STTx const > const &	stx,
		bool	batch
	)

private

Definition at line 2931 of file PeerImp.cpp.

◆ checkPropose()

void xrpl::PeerImp::checkPropose	(	bool	isTrusted,
		std::shared_ptr< protocol::TMProposeSet > const &	packet,
		RCLCxPeerPos	peerPos
	)

private

Definition at line 3084 of file PeerImp.cpp.

◆ checkValidation()

void xrpl::PeerImp::checkValidation	(	std::shared_ptr< STValidation > const &	val,
		uint256 const &	key,
		std::shared_ptr< protocol::TMValidation > const &	packet
	)

private

Definition at line 3127 of file PeerImp.cpp.

◆ sendLedgerBase()

void xrpl::PeerImp::sendLedgerBase	(	std::shared_ptr< Ledger const > const &	ledger,
		protocol::TMLedgerData &	ledgerData
	)

private

Definition at line 3224 of file PeerImp.cpp.

◆ getLedger()

std::shared_ptr< Ledger const > xrpl::PeerImp::getLedger ( std::shared_ptr< protocol::TMGetLedger > const & m )

private

Definition at line 3264 of file PeerImp.cpp.

◆ getTxSet()

std::shared_ptr< SHAMap const > xrpl::PeerImp::getTxSet ( std::shared_ptr< protocol::TMGetLedger > const & m ) const

private

Definition at line 3360 of file PeerImp.cpp.

◆ processLedgerRequest()

void xrpl::PeerImp::processLedgerRequest ( std::shared_ptr< protocol::TMGetLedger > const & m )

private

Definition at line 3395 of file PeerImp.cpp.

Friends And Related Symbol Documentation

◆ OverlayImpl

friend class OverlayImpl

friend

Definition at line 256 of file PeerImp.h.

Member Data Documentation

◆ app_

Application& xrpl::PeerImp::app_

private

Definition at line 116 of file PeerImp.h.

◆ id_

id_t const xrpl::PeerImp::id_

private

Definition at line 117 of file PeerImp.h.

◆ fingerprint_

std::string xrpl::PeerImp::fingerprint_

private

Definition at line 118 of file PeerImp.h.

◆ prefix_

std::string xrpl::PeerImp::prefix_

private

Definition at line 119 of file PeerImp.h.

◆ sink_

beast::WrappedSink xrpl::PeerImp::sink_

private

Definition at line 120 of file PeerImp.h.

◆ p_sink_

beast::WrappedSink xrpl::PeerImp::p_sink_

private

Definition at line 121 of file PeerImp.h.

◆ journal_

beast::Journal const xrpl::PeerImp::journal_

private

Definition at line 122 of file PeerImp.h.

◆ p_journal_

beast::Journal const xrpl::PeerImp::p_journal_

private

Definition at line 123 of file PeerImp.h.

◆ stream_ptr_

std::unique_ptr<stream_type> xrpl::PeerImp::stream_ptr_

private

Definition at line 124 of file PeerImp.h.

◆ socket_

socket_type& xrpl::PeerImp::socket_

private

Definition at line 125 of file PeerImp.h.

◆ stream_

stream_type& xrpl::PeerImp::stream_

private

Definition at line 126 of file PeerImp.h.

◆ strand_

boost::asio::strand<boost::asio::executor> xrpl::PeerImp::strand_

private

Definition at line 127 of file PeerImp.h.

◆ timer_

waitable_timer xrpl::PeerImp::timer_

private

Definition at line 130 of file PeerImp.h.

◆ remote_address_

beast::IP::Endpoint const xrpl::PeerImp::remote_address_

private

Definition at line 134 of file PeerImp.h.

◆ overlay_

OverlayImpl& xrpl::PeerImp::overlay_

private

Definition at line 138 of file PeerImp.h.

◆ inbound_

bool const xrpl::PeerImp::inbound_

private

Definition at line 139 of file PeerImp.h.

◆ protocol_

ProtocolVersion xrpl::PeerImp::protocol_

private

Definition at line 142 of file PeerImp.h.

◆ tracking_

std::atomic<Tracking> xrpl::PeerImp::tracking_

private

Definition at line 144 of file PeerImp.h.

◆ trackingTime_

clock_type::time_point xrpl::PeerImp::trackingTime_

private

Definition at line 145 of file PeerImp.h.

◆ publicKey_

PublicKey const xrpl::PeerImp::publicKey_

private

Definition at line 147 of file PeerImp.h.

◆ name_

std::string xrpl::PeerImp::name_

private

Definition at line 148 of file PeerImp.h.

◆ nameMutex_

std::shared_mutex xrpl::PeerImp::nameMutex_

mutableprivate

Definition at line 149 of file PeerImp.h.

◆ minLedger_

LedgerIndex xrpl::PeerImp::minLedger_ = 0

private

Definition at line 153 of file PeerImp.h.

◆ maxLedger_

LedgerIndex xrpl::PeerImp::maxLedger_ = 0

private

Definition at line 154 of file PeerImp.h.

◆ closedLedgerHash_

uint256 xrpl::PeerImp::closedLedgerHash_

private

Definition at line 155 of file PeerImp.h.

◆ previousLedgerHash_

uint256 xrpl::PeerImp::previousLedgerHash_

private

Definition at line 156 of file PeerImp.h.

◆ recentLedgers_

boost::circular_buffer<uint256> xrpl::PeerImp::recentLedgers_ {128}

private

Definition at line 158 of file PeerImp.h.

◆ recentTxSets_

boost::circular_buffer<uint256> xrpl::PeerImp::recentTxSets_ {128}

private

Definition at line 159 of file PeerImp.h.

◆ latency_

std::optional<std::chrono::milliseconds> xrpl::PeerImp::latency_

private

Definition at line 161 of file PeerImp.h.

◆ lastPingSeq_

std::optional<std::uint32_t> xrpl::PeerImp::lastPingSeq_

private

Definition at line 162 of file PeerImp.h.

◆ lastPingTime_

clock_type::time_point xrpl::PeerImp::lastPingTime_

private

Definition at line 163 of file PeerImp.h.

◆ creationTime_

clock_type::time_point const xrpl::PeerImp::creationTime_

private

Definition at line 164 of file PeerImp.h.

◆ squelch_

reduce_relay::Squelch<UptimeClock> xrpl::PeerImp::squelch_

private

Definition at line 166 of file PeerImp.h.

◆ recentLock_

std::mutex xrpl::PeerImp::recentLock_

mutableprivate

Definition at line 215 of file PeerImp.h.

◆ last_status_

protocol::TMStatusChange xrpl::PeerImp::last_status_

private

Definition at line 216 of file PeerImp.h.

◆ usage_

Resource::Consumer xrpl::PeerImp::usage_

private

Definition at line 217 of file PeerImp.h.

◆ fee_

ChargeWithContext xrpl::PeerImp::fee_

private

Definition at line 218 of file PeerImp.h.

◆ slot_

std::shared_ptr<PeerFinder::Slot> const xrpl::PeerImp::slot_

private

Definition at line 219 of file PeerImp.h.

◆ read_buffer_

boost::beast::multi_buffer xrpl::PeerImp::read_buffer_

private

Definition at line 220 of file PeerImp.h.

◆ request_

http_request_type xrpl::PeerImp::request_

private

Definition at line 221 of file PeerImp.h.

◆ response_

http_response_type xrpl::PeerImp::response_

private

Definition at line 222 of file PeerImp.h.

◆ headers_

boost::beast::http::fields const& xrpl::PeerImp::headers_

private

Definition at line 223 of file PeerImp.h.

◆ send_queue_

std::queue<std::shared_ptr<Message> > xrpl::PeerImp::send_queue_

private

Definition at line 224 of file PeerImp.h.

◆ shutdown_

bool xrpl::PeerImp::shutdown_ = false

private

Definition at line 227 of file PeerImp.h.

◆ shutdownStarted_

bool xrpl::PeerImp::shutdownStarted_ = false

private

Definition at line 230 of file PeerImp.h.

◆ readPending_

bool xrpl::PeerImp::readPending_ = false

private

Definition at line 233 of file PeerImp.h.

◆ writePending_

bool xrpl::PeerImp::writePending_ = false

private

Definition at line 236 of file PeerImp.h.

◆ large_sendq_

int xrpl::PeerImp::large_sendq_ = 0

private

Definition at line 238 of file PeerImp.h.

◆ load_event_

std::unique_ptr<LoadEvent> xrpl::PeerImp::load_event_

private

Definition at line 239 of file PeerImp.h.

◆ publisherListSequences_

hash_map<PublicKey, std::size_t> xrpl::PeerImp::publisherListSequences_

private

Definition at line 242 of file PeerImp.h.

◆ compressionEnabled_

Compressed xrpl::PeerImp::compressionEnabled_ = Compressed::Off

private

Definition at line 244 of file PeerImp.h.

◆ txQueue_

hash_set<uint256> xrpl::PeerImp::txQueue_

private

Definition at line 249 of file PeerImp.h.

◆ txReduceRelayEnabled_

bool xrpl::PeerImp::txReduceRelayEnabled_ = false

private

Definition at line 251 of file PeerImp.h.

◆ ledgerReplayEnabled_

bool xrpl::PeerImp::ledgerReplayEnabled_ = false

private

Definition at line 253 of file PeerImp.h.

◆ ledgerReplayMsgHandler_

LedgerReplayMsgHandler xrpl::PeerImp::ledgerReplayMsgHandler_

private

Definition at line 254 of file PeerImp.h.

◆ sent

Metrics xrpl::PeerImp::sent

Definition at line 287 of file PeerImp.h.

◆ recv

Metrics xrpl::PeerImp::recv

Definition at line 288 of file PeerImp.h.

◆ [struct]

struct { ... } xrpl::PeerImp::metrics_

Classes

Public Types

Public Member Functions

Private Types

Private Member Functions

Static Private Member Functions

Private Attributes

Friends

Detailed Description

Member Typedef Documentation

◆ clock_type

◆ error_code

◆ socket_type

◆ middle_type

◆ stream_type

◆ address_type

◆ endpoint_type

◆ waitable_timer

◆ Compressed

◆ ptr

◆ id_t

Member Enumeration Documentation

◆ Tracking

Constructor & Destructor Documentation

◆ PeerImp() [1/3]

◆ PeerImp() [2/3]

◆ PeerImp() [3/3]

◆ ~PeerImp()

Member Function Documentation

◆ operator=()

◆ pJournal()

◆ slot()

◆ run()

◆ stop()

◆ send()

◆ sendTxQueue()

◆ addTxQueue()

◆ removeTxQueue()

◆ sendEndpoints()

◆ getRemoteAddress()

◆ charge()

◆ id()

◆ crawl()

◆ cluster()

◆ checkTracking() [1/2]

◆ checkTracking() [2/2]

◆ getNodePublic()

◆ getVersion()

◆ uptime()

◆ json()

◆ supportsFeature()

◆ publisherListSequence()

◆ setPublisherListSequence()

◆ getClosedLedgerHash()

◆ hasLedger()

◆ ledgerRange()

◆ hasTxSet()

◆ cycleStatus()

◆ hasRange()

◆ getScore()

◆ isHighLatency()

◆ compressionEnabled()

◆ txReduceRelayEnabled()

◆ fail() [1/2]

◆ fail() [2/2]

◆ shutdown()

◆ tryAsyncShutdown()

◆ onShutdown()

◆ close()

◆ setTimer()

◆ onTimer()

◆ cancelTimer()

◆ makePrefix()

◆ doAccept()

◆ name()

◆ domain()

◆ doProtocolStart()

◆ onReadMessage()

◆ onWriteMessage()

◆ handleTransaction()