From 93594e42c3f92d82427d2b284ff0f94cdbebe99c Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Wed, 5 Jul 2023 16:22:52 -0400 Subject: [PATCH 1/9] refactor: merge transport serializer and deserializer into Transport class This allows state that is shared between both directions to be encapsulated into a single object. Specifically the v2 transport protocol introduced by BIP324 has sending state (the encryption keys) that depends on received messages (the DH key exchange). Having a single object for both means it can hide logic from callers related to that key exchange and other interactions. --- src/net.cpp | 21 +++++----- src/net.h | 41 ++++++++----------- src/test/fuzz/p2p_transport_serialization.cpp | 15 +++---- src/test/util/net.cpp | 2 +- 4 files changed, 37 insertions(+), 42 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 53a2dcf125..fa20136bb1 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -681,16 +681,16 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) nRecvBytes += msg_bytes.size(); while (msg_bytes.size() > 0) { // absorb network data - int handled = m_deserializer->Read(msg_bytes); + int handled = m_transport->Read(msg_bytes); if (handled < 0) { // Serious header problem, disconnect from the peer. return false; } - if (m_deserializer->Complete()) { + if (m_transport->Complete()) { // decompose a transport agnostic CNetMessage from the deserializer bool reject_message{false}; - CNetMessage msg = m_deserializer->GetMessage(time, reject_message); + CNetMessage msg = m_transport->GetMessage(time, reject_message); if (reject_message) { // Message deserialization failed. Drop the message but don't disconnect the peer. // store the size of the corrupt message @@ -717,7 +717,7 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) return true; } -int V1TransportDeserializer::readHeader(Span msg_bytes) +int V1Transport::readHeader(Span msg_bytes) { // copy data to temporary parsing buffer unsigned int nRemaining = CMessageHeader::HEADER_SIZE - nHdrPos; @@ -757,7 +757,7 @@ int V1TransportDeserializer::readHeader(Span msg_bytes) return nCopy; } -int V1TransportDeserializer::readData(Span msg_bytes) +int V1Transport::readData(Span msg_bytes) { unsigned int nRemaining = hdr.nMessageSize - nDataPos; unsigned int nCopy = std::min(nRemaining, msg_bytes.size()); @@ -774,7 +774,7 @@ int V1TransportDeserializer::readData(Span msg_bytes) return nCopy; } -const uint256& V1TransportDeserializer::GetMessageHash() const +const uint256& V1Transport::GetMessageHash() const { assert(Complete()); if (data_hash.IsNull()) @@ -782,7 +782,7 @@ const uint256& V1TransportDeserializer::GetMessageHash() const return data_hash; } -CNetMessage V1TransportDeserializer::GetMessage(const std::chrono::microseconds time, bool& reject_message) +CNetMessage V1Transport::GetMessage(const std::chrono::microseconds time, bool& reject_message) { // Initialize out parameter reject_message = false; @@ -819,7 +819,7 @@ CNetMessage V1TransportDeserializer::GetMessage(const std::chrono::microseconds return msg; } -void V1TransportSerializer::prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const +void V1Transport::prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const { // create dbl-sha256 checksum uint256 hash = Hash(msg.data); @@ -2822,8 +2822,7 @@ CNode::CNode(NodeId idIn, ConnectionType conn_type_in, bool inbound_onion, CNodeOptions&& node_opts) - : m_deserializer{std::make_unique(V1TransportDeserializer(Params(), idIn, SER_NETWORK, INIT_PROTO_VERSION))}, - m_serializer{std::make_unique(V1TransportSerializer())}, + : m_transport{std::make_unique(Params(), idIn, SER_NETWORK, INIT_PROTO_VERSION)}, m_permission_flags{node_opts.permission_flags}, m_sock{sock}, m_connected{GetTime()}, @@ -2908,7 +2907,7 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) // make sure we use the appropriate network transport format std::vector serializedHeader; - pnode->m_serializer->prepareForTransport(msg, serializedHeader); + pnode->m_transport->prepareForTransport(msg, serializedHeader); size_t nTotalSize = nMessageSize + serializedHeader.size(); size_t nBytesSent = 0; diff --git a/src/net.h b/src/net.h index 3c1221f518..ca6899a83a 100644 --- a/src/net.h +++ b/src/net.h @@ -253,24 +253,31 @@ public: } }; -/** The TransportDeserializer takes care of holding and deserializing the - * network receive buffer. It can deserialize the network buffer into a - * transport protocol agnostic CNetMessage (message type & payload) - */ -class TransportDeserializer { +/** The Transport converts one connection's sent messages to wire bytes, and received bytes back. */ +class Transport { public: + virtual ~Transport() {} + + // 1. Receiver side functions, for decoding bytes received on the wire into transport protocol + // agnostic CNetMessage (message type & payload) objects. Callers must guarantee that none of + // these functions are called concurrently w.r.t. one another. + // returns true if the current deserialization is complete virtual bool Complete() const = 0; - // set the serialization context version + // set the deserialization context version virtual void SetVersion(int version) = 0; /** read and deserialize data, advances msg_bytes data pointer */ virtual int Read(Span& msg_bytes) = 0; // decomposes a message from the context virtual CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) = 0; - virtual ~TransportDeserializer() {} + + // 2. Sending side functions: + + // prepare message for transport (header construction, error-correction computation, payload encryption, etc.) + virtual void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const = 0; }; -class V1TransportDeserializer final : public TransportDeserializer +class V1Transport final : public Transport { private: const CChainParams& m_chain_params; @@ -300,7 +307,7 @@ private: } public: - V1TransportDeserializer(const CChainParams& chain_params, const NodeId node_id, int nTypeIn, int nVersionIn) + V1Transport(const CChainParams& chain_params, const NodeId node_id, int nTypeIn, int nVersionIn) : m_chain_params(chain_params), m_node_id(node_id), hdrbuf(nTypeIn, nVersionIn), @@ -331,19 +338,7 @@ public: return ret; } CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) override; -}; -/** The TransportSerializer prepares messages for the network transport - */ -class TransportSerializer { -public: - // prepare message for transport (header construction, error-correction computation, payload encryption, etc.) - virtual void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const = 0; - virtual ~TransportSerializer() {} -}; - -class V1TransportSerializer : public TransportSerializer { -public: void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const override; }; @@ -359,8 +354,8 @@ struct CNodeOptions class CNode { public: - const std::unique_ptr m_deserializer; // Used only by SocketHandler thread - const std::unique_ptr m_serializer; + /** Transport serializer/deserializer. The receive side functions are only called under cs_vRecv. */ + const std::unique_ptr m_transport; const NetPermissionFlags m_permission_flags; diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index 78350a600e..5e44421f1d 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -24,9 +24,10 @@ void initialize_p2p_transport_serialization() FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serialization) { - // Construct deserializer, with a dummy NodeId - V1TransportDeserializer deserializer{Params(), NodeId{0}, SER_NETWORK, INIT_PROTO_VERSION}; - V1TransportSerializer serializer{}; + // Construct transports for both sides, with dummy NodeIds. + V1Transport recv_transport{Params(), NodeId{0}, SER_NETWORK, INIT_PROTO_VERSION}; + V1Transport send_transport{Params(), NodeId{1}, SER_NETWORK, INIT_PROTO_VERSION}; + FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()}; auto checksum_assist = fuzzed_data_provider.ConsumeBool(); @@ -63,14 +64,14 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial mutable_msg_bytes.insert(mutable_msg_bytes.end(), payload_bytes.begin(), payload_bytes.end()); Span msg_bytes{mutable_msg_bytes}; while (msg_bytes.size() > 0) { - const int handled = deserializer.Read(msg_bytes); + const int handled = recv_transport.Read(msg_bytes); if (handled < 0) { break; } - if (deserializer.Complete()) { + if (recv_transport.Complete()) { const std::chrono::microseconds m_time{std::numeric_limits::max()}; bool reject_message{false}; - CNetMessage msg = deserializer.GetMessage(m_time, reject_message); + CNetMessage msg = recv_transport.GetMessage(m_time, reject_message); assert(msg.m_type.size() <= CMessageHeader::COMMAND_SIZE); assert(msg.m_raw_message_size <= mutable_msg_bytes.size()); assert(msg.m_raw_message_size == CMessageHeader::HEADER_SIZE + msg.m_message_size); @@ -78,7 +79,7 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial std::vector header; auto msg2 = CNetMsgMaker{msg.m_recv.GetVersion()}.Make(msg.m_type, Span{msg.m_recv}); - serializer.prepareForTransport(msg2, header); + send_transport.prepareForTransport(msg2, header); } } } diff --git a/src/test/util/net.cpp b/src/test/util/net.cpp index 3f72384b3b..0031770028 100644 --- a/src/test/util/net.cpp +++ b/src/test/util/net.cpp @@ -73,7 +73,7 @@ void ConnmanTestMsg::NodeReceiveMsgBytes(CNode& node, Span msg_by bool ConnmanTestMsg::ReceiveMsgFrom(CNode& node, CSerializedNetMsg& ser_msg) const { std::vector ser_msg_header; - node.m_serializer->prepareForTransport(ser_msg, ser_msg_header); + node.m_transport->prepareForTransport(ser_msg, ser_msg_header); bool complete; NodeReceiveMsgBytes(node, ser_msg_header, complete); From 27f9ba23efe82531a465c5e63bf7dc62b6a3a8db Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Wed, 26 Jul 2023 13:19:31 -0400 Subject: [PATCH 2/9] net: add V1Transport lock protecting receive state Rather than relying on the caller to prevent concurrent calls to the various receive-side functions of Transport, introduce a private m_cs_recv inside the implementation to protect the lock state. Of course, this does not remove the need for callers to synchronize calls entirely, as it is a stateful object, and e.g. the order in which Receive(), Complete(), and GetMessage() are called matters. It seems impossible to use a Transport object in a meaningful way in a multi-threaded way without some form of external synchronization, but it still feels safer to make the transport object itself responsible for protecting its internal state. --- src/net.cpp | 7 ++++++- src/net.h | 56 +++++++++++++++++++++++++++++++++-------------------- 2 files changed, 41 insertions(+), 22 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index fa20136bb1..b350c58c61 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -719,6 +719,7 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) int V1Transport::readHeader(Span msg_bytes) { + AssertLockHeld(m_recv_mutex); // copy data to temporary parsing buffer unsigned int nRemaining = CMessageHeader::HEADER_SIZE - nHdrPos; unsigned int nCopy = std::min(nRemaining, msg_bytes.size()); @@ -759,6 +760,7 @@ int V1Transport::readHeader(Span msg_bytes) int V1Transport::readData(Span msg_bytes) { + AssertLockHeld(m_recv_mutex); unsigned int nRemaining = hdr.nMessageSize - nDataPos; unsigned int nCopy = std::min(nRemaining, msg_bytes.size()); @@ -776,7 +778,8 @@ int V1Transport::readData(Span msg_bytes) const uint256& V1Transport::GetMessageHash() const { - assert(Complete()); + AssertLockHeld(m_recv_mutex); + assert(CompleteInternal()); if (data_hash.IsNull()) hasher.Finalize(data_hash); return data_hash; @@ -784,9 +787,11 @@ const uint256& V1Transport::GetMessageHash() const CNetMessage V1Transport::GetMessage(const std::chrono::microseconds time, bool& reject_message) { + AssertLockNotHeld(m_recv_mutex); // Initialize out parameter reject_message = false; // decompose a single CNetMessage from the TransportDeserializer + LOCK(m_recv_mutex); CNetMessage msg(std::move(vRecv)); // store message type string, time, and sizes diff --git a/src/net.h b/src/net.h index ca6899a83a..e34ea590cc 100644 --- a/src/net.h +++ b/src/net.h @@ -259,8 +259,7 @@ public: virtual ~Transport() {} // 1. Receiver side functions, for decoding bytes received on the wire into transport protocol - // agnostic CNetMessage (message type & payload) objects. Callers must guarantee that none of - // these functions are called concurrently w.r.t. one another. + // agnostic CNetMessage (message type & payload) objects. // returns true if the current deserialization is complete virtual bool Complete() const = 0; @@ -282,20 +281,22 @@ class V1Transport final : public Transport private: const CChainParams& m_chain_params; const NodeId m_node_id; // Only for logging - mutable CHash256 hasher; - mutable uint256 data_hash; - bool in_data; // parsing header (false) or data (true) - CDataStream hdrbuf; // partially received header - CMessageHeader hdr; // complete header - CDataStream vRecv; // received message data - unsigned int nHdrPos; - unsigned int nDataPos; + mutable Mutex m_recv_mutex; //!< Lock for receive state + mutable CHash256 hasher GUARDED_BY(m_recv_mutex); + mutable uint256 data_hash GUARDED_BY(m_recv_mutex); + bool in_data GUARDED_BY(m_recv_mutex); // parsing header (false) or data (true) + CDataStream hdrbuf GUARDED_BY(m_recv_mutex); // partially received header + CMessageHeader hdr GUARDED_BY(m_recv_mutex); // complete header + CDataStream vRecv GUARDED_BY(m_recv_mutex); // received message data + unsigned int nHdrPos GUARDED_BY(m_recv_mutex); + unsigned int nDataPos GUARDED_BY(m_recv_mutex); - const uint256& GetMessageHash() const; - int readHeader(Span msg_bytes); - int readData(Span msg_bytes); + const uint256& GetMessageHash() const EXCLUSIVE_LOCKS_REQUIRED(m_recv_mutex); + int readHeader(Span msg_bytes) EXCLUSIVE_LOCKS_REQUIRED(m_recv_mutex); + int readData(Span msg_bytes) EXCLUSIVE_LOCKS_REQUIRED(m_recv_mutex); - void Reset() { + void Reset() EXCLUSIVE_LOCKS_REQUIRED(m_recv_mutex) { + AssertLockHeld(m_recv_mutex); vRecv.clear(); hdrbuf.clear(); hdrbuf.resize(24); @@ -306,6 +307,13 @@ private: hasher.Reset(); } + bool CompleteInternal() const noexcept EXCLUSIVE_LOCKS_REQUIRED(m_recv_mutex) + { + AssertLockHeld(m_recv_mutex); + if (!in_data) return false; + return hdr.nMessageSize == nDataPos; + } + public: V1Transport(const CChainParams& chain_params, const NodeId node_id, int nTypeIn, int nVersionIn) : m_chain_params(chain_params), @@ -313,22 +321,28 @@ public: hdrbuf(nTypeIn, nVersionIn), vRecv(nTypeIn, nVersionIn) { + LOCK(m_recv_mutex); Reset(); } - bool Complete() const override + bool Complete() const override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { - if (!in_data) - return false; - return (hdr.nMessageSize == nDataPos); + AssertLockNotHeld(m_recv_mutex); + return WITH_LOCK(m_recv_mutex, return CompleteInternal()); } - void SetVersion(int nVersionIn) override + + void SetVersion(int nVersionIn) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { + AssertLockNotHeld(m_recv_mutex); + LOCK(m_recv_mutex); hdrbuf.SetVersion(nVersionIn); vRecv.SetVersion(nVersionIn); } - int Read(Span& msg_bytes) override + + int Read(Span& msg_bytes) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { + AssertLockNotHeld(m_recv_mutex); + LOCK(m_recv_mutex); int ret = in_data ? readData(msg_bytes) : readHeader(msg_bytes); if (ret < 0) { Reset(); @@ -337,7 +351,7 @@ public: } return ret; } - CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) override; + CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex); void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const override; }; From 649a83c7f73db2ee115f5dce3df16622e318aeba Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Mon, 14 Aug 2023 16:37:05 -0400 Subject: [PATCH 3/9] refactor: rename Transport class receive functions Now that the Transport class deals with both the sending and receiving side of things, make the receive side have function names that clearly indicate they're about receiving. * Transport::Read() -> Transport::ReceivedBytes() * Transport::Complete() -> Transport::ReceivedMessageComplete() * Transport::GetMessage() -> Transport::GetReceivedMessage() * Transport::SetVersion() -> Transport::SetReceiveVersion() Further, also update the comments on these functions to (among others) remove the "deserialization" terminology. That term is better reserved for just the serialization/deserialization between objects and bytes (see serialize.h), and not the conversion from/to wire bytes as performed by the Transport. --- src/net.cpp | 8 +++--- src/net.h | 25 ++++++++++--------- src/test/fuzz/p2p_transport_serialization.cpp | 6 ++--- 3 files changed, 20 insertions(+), 19 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index b350c58c61..338831bb48 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -681,16 +681,16 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) nRecvBytes += msg_bytes.size(); while (msg_bytes.size() > 0) { // absorb network data - int handled = m_transport->Read(msg_bytes); + int handled = m_transport->ReceivedBytes(msg_bytes); if (handled < 0) { // Serious header problem, disconnect from the peer. return false; } - if (m_transport->Complete()) { + if (m_transport->ReceivedMessageComplete()) { // decompose a transport agnostic CNetMessage from the deserializer bool reject_message{false}; - CNetMessage msg = m_transport->GetMessage(time, reject_message); + CNetMessage msg = m_transport->GetReceivedMessage(time, reject_message); if (reject_message) { // Message deserialization failed. Drop the message but don't disconnect the peer. // store the size of the corrupt message @@ -785,7 +785,7 @@ const uint256& V1Transport::GetMessageHash() const return data_hash; } -CNetMessage V1Transport::GetMessage(const std::chrono::microseconds time, bool& reject_message) +CNetMessage V1Transport::GetReceivedMessage(const std::chrono::microseconds time, bool& reject_message) { AssertLockNotHeld(m_recv_mutex); // Initialize out parameter diff --git a/src/net.h b/src/net.h index e34ea590cc..a17ca36652 100644 --- a/src/net.h +++ b/src/net.h @@ -261,14 +261,14 @@ public: // 1. Receiver side functions, for decoding bytes received on the wire into transport protocol // agnostic CNetMessage (message type & payload) objects. - // returns true if the current deserialization is complete - virtual bool Complete() const = 0; - // set the deserialization context version - virtual void SetVersion(int version) = 0; - /** read and deserialize data, advances msg_bytes data pointer */ - virtual int Read(Span& msg_bytes) = 0; - // decomposes a message from the context - virtual CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) = 0; + /** Returns true if the current message is complete (so GetReceivedMessage can be called). */ + virtual bool ReceivedMessageComplete() const = 0; + /** Set the deserialization context version for objects returned by GetReceivedMessage. */ + virtual void SetReceiveVersion(int version) = 0; + /** Feed wire bytes to the transport; chops off consumed bytes off front of msg_bytes. */ + virtual int ReceivedBytes(Span& msg_bytes) = 0; + /** Retrieve a completed message from transport (only when ReceivedMessageComplete). */ + virtual CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) = 0; // 2. Sending side functions: @@ -325,13 +325,13 @@ public: Reset(); } - bool Complete() const override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) + bool ReceivedMessageComplete() const override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { AssertLockNotHeld(m_recv_mutex); return WITH_LOCK(m_recv_mutex, return CompleteInternal()); } - void SetVersion(int nVersionIn) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) + void SetReceiveVersion(int nVersionIn) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { AssertLockNotHeld(m_recv_mutex); LOCK(m_recv_mutex); @@ -339,7 +339,7 @@ public: vRecv.SetVersion(nVersionIn); } - int Read(Span& msg_bytes) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) + int ReceivedBytes(Span& msg_bytes) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { AssertLockNotHeld(m_recv_mutex); LOCK(m_recv_mutex); @@ -351,7 +351,8 @@ public: } return ret; } - CNetMessage GetMessage(std::chrono::microseconds time, bool& reject_message) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex); + + CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex); void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const override; }; diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index 5e44421f1d..dcf7529918 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -64,14 +64,14 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial mutable_msg_bytes.insert(mutable_msg_bytes.end(), payload_bytes.begin(), payload_bytes.end()); Span msg_bytes{mutable_msg_bytes}; while (msg_bytes.size() > 0) { - const int handled = recv_transport.Read(msg_bytes); + const int handled = recv_transport.ReceivedBytes(msg_bytes); if (handled < 0) { break; } - if (recv_transport.Complete()) { + if (recv_transport.ReceivedMessageComplete()) { const std::chrono::microseconds m_time{std::numeric_limits::max()}; bool reject_message{false}; - CNetMessage msg = recv_transport.GetMessage(m_time, reject_message); + CNetMessage msg = recv_transport.GetReceivedMessage(m_time, reject_message); assert(msg.m_type.size() <= CMessageHeader::COMMAND_SIZE); assert(msg.m_raw_message_size <= mutable_msg_bytes.size()); assert(msg.m_raw_message_size == CMessageHeader::HEADER_SIZE + msg.m_message_size); From 0de48fe858a1ffcced340eef2c849165216141c8 Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Fri, 21 Jul 2023 16:31:59 -0400 Subject: [PATCH 4/9] net: abstract sending side of transport serialization further This makes the sending side of P2P transports mirror the receiver side: caller provides message (consisting of type and payload) to be sent, and then asks what bytes must be sent. Once the message has been fully sent, a new message can be provided. This removes the assumption that P2P serialization of messages follows a strict structure of header (a function of type and payload), followed by (unmodified) payload, and instead lets transports decide the structure themselves. It also removes the assumption that a message must always be sent at once, or that no bytes are even sent on the wire when there is no message. This opens the door for supporting traffic shaping mechanisms in the future. --- src/net.cpp | 98 +++++++++++++++---- src/net.h | 63 +++++++++++- src/test/fuzz/p2p_transport_serialization.cpp | 11 ++- src/test/fuzz/process_messages.cpp | 2 +- src/test/util/net.cpp | 21 ++-- src/test/util/net.h | 2 +- 6 files changed, 161 insertions(+), 36 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 338831bb48..1545e36e68 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -824,8 +824,13 @@ CNetMessage V1Transport::GetReceivedMessage(const std::chrono::microseconds time return msg; } -void V1Transport::prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const +bool V1Transport::SetMessageToSend(CSerializedNetMsg& msg) noexcept { + AssertLockNotHeld(m_send_mutex); + // Determine whether a new message can be set. + LOCK(m_send_mutex); + if (m_sending_header || m_bytes_sent < m_message_to_send.data.size()) return false; + // create dbl-sha256 checksum uint256 hash = Hash(msg.data); @@ -834,8 +839,50 @@ void V1Transport::prepareForTransport(CSerializedNetMsg& msg, std::vector CConnman::SocketSendData(CNode& node) const @@ -2910,27 +2957,40 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) msg.data.data() ); - // make sure we use the appropriate network transport format - std::vector serializedHeader; - pnode->m_transport->prepareForTransport(msg, serializedHeader); - size_t nTotalSize = nMessageSize + serializedHeader.size(); - size_t nBytesSent = 0; { LOCK(pnode->cs_vSend); - bool optimisticSend(pnode->vSendMsg.empty()); + const bool queue_was_empty{pnode->vSendMsg.empty()}; - //log total amount of bytes per message type - pnode->AccountForSentBytes(msg.m_type, nTotalSize); - pnode->nSendSize += nTotalSize; + // Give the message to the transport, and add all bytes it wants us to send out as byte + // vectors to vSendMsg. This is temporary code that exists to support the new transport + // sending interface using the old way of queueing data. In a future commit vSendMsg will + // be replaced with a queue of CSerializedNetMsg objects to be sent instead, and this code + // will disappear. + bool queued = pnode->m_transport->SetMessageToSend(msg); + assert(queued); + // In the current transport (V1Transport), GetBytesToSend first returns a header to send, + // and then the payload data (if any), necessitating a loop. + while (true) { + const auto& [bytes, _more, msg_type] = pnode->m_transport->GetBytesToSend(); + if (bytes.empty()) break; + // Update statistics per message type. + pnode->AccountForSentBytes(msg_type, bytes.size()); + // Update number of bytes in the send buffer. + pnode->nSendSize += bytes.size(); + if (pnode->nSendSize > nSendBufferMaxSize) pnode->fPauseSend = true; + pnode->vSendMsg.push_back({bytes.begin(), bytes.end()}); + // Notify transport that bytes have been processed (they're not actually sent yet, + // but pushed onto the vSendMsg queue of bytes to send). + pnode->m_transport->MarkBytesSent(bytes.size()); + } - if (pnode->nSendSize > nSendBufferMaxSize) pnode->fPauseSend = true; - pnode->vSendMsg.push_back(std::move(serializedHeader)); - if (nMessageSize) pnode->vSendMsg.push_back(std::move(msg.data)); - - // If write queue empty, attempt "optimistic write" - bool data_left; - if (optimisticSend) std::tie(nBytesSent, data_left) = SocketSendData(*pnode); + // If the write queue was empty before and isn't now, attempt "optimistic write": + // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually + // doing a send, try sending from the calling thread if the queue was empty before. + if (queue_was_empty && !pnode->vSendMsg.empty()) { + std::tie(nBytesSent, std::ignore) = SocketSendData(*pnode); + } } if (nBytesSent) RecordBytesSent(nBytesSent); } diff --git a/src/net.h b/src/net.h index a17ca36652..83deb4afed 100644 --- a/src/net.h +++ b/src/net.h @@ -270,10 +270,49 @@ public: /** Retrieve a completed message from transport (only when ReceivedMessageComplete). */ virtual CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) = 0; - // 2. Sending side functions: + // 2. Sending side functions, for converting messages into bytes to be sent over the wire. - // prepare message for transport (header construction, error-correction computation, payload encryption, etc.) - virtual void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const = 0; + /** Set the next message to send. + * + * If no message can currently be set (perhaps because the previous one is not yet done being + * sent), returns false, and msg will be unmodified. Otherwise msg is enqueued (and + * possibly moved-from) and true is returned. + */ + virtual bool SetMessageToSend(CSerializedNetMsg& msg) noexcept = 0; + + /** Return type for GetBytesToSend, consisting of: + * - Span to_send: span of bytes to be sent over the wire (possibly empty). + * - bool more: whether there will be more bytes to be sent after the ones in to_send are + * all sent (as signaled by MarkBytesSent()). + * - const std::string& m_type: message type on behalf of which this is being sent. + */ + using BytesToSend = std::tuple< + Span /*to_send*/, + bool /*more*/, + const std::string& /*m_type*/ + >; + + /** Get bytes to send on the wire. + * + * As a const function, it does not modify the transport's observable state, and is thus safe + * to be called multiple times. + * + * The bytes returned by this function act as a stream which can only be appended to. This + * means that with the exception of MarkBytesSent, operations on the transport can only append + * to what is being returned. + * + * Note that m_type and to_send refer to data that is internal to the transport, and calling + * any non-const function on this object may invalidate them. + */ + virtual BytesToSend GetBytesToSend() const noexcept = 0; + + /** Report how many bytes returned by the last GetBytesToSend() have been sent. + * + * bytes_sent cannot exceed to_send.size() of the last GetBytesToSend() result. + * + * If bytes_sent=0, this call has no effect. + */ + virtual void MarkBytesSent(size_t bytes_sent) noexcept = 0; }; class V1Transport final : public Transport @@ -314,6 +353,17 @@ private: return hdr.nMessageSize == nDataPos; } + /** Lock for sending state. */ + mutable Mutex m_send_mutex; + /** The header of the message currently being sent. */ + std::vector m_header_to_send GUARDED_BY(m_send_mutex); + /** The data of the message currently being sent. */ + CSerializedNetMsg m_message_to_send GUARDED_BY(m_send_mutex); + /** Whether we're currently sending header bytes or message bytes. */ + bool m_sending_header GUARDED_BY(m_send_mutex) {false}; + /** How many bytes have been sent so far (from m_header_to_send, or from m_message_to_send.data). */ + size_t m_bytes_sent GUARDED_BY(m_send_mutex) {0}; + public: V1Transport(const CChainParams& chain_params, const NodeId node_id, int nTypeIn, int nVersionIn) : m_chain_params(chain_params), @@ -354,7 +404,9 @@ public: CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex); - void prepareForTransport(CSerializedNetMsg& msg, std::vector& header) const override; + bool SetMessageToSend(CSerializedNetMsg& msg) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); + BytesToSend GetBytesToSend() const noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); + void MarkBytesSent(size_t bytes_sent) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); }; struct CNodeOptions @@ -369,7 +421,8 @@ struct CNodeOptions class CNode { public: - /** Transport serializer/deserializer. The receive side functions are only called under cs_vRecv. */ + /** Transport serializer/deserializer. The receive side functions are only called under cs_vRecv, while + * the sending side functions are only called under cs_vSend. */ const std::unique_ptr m_transport; const NetPermissionFlags m_permission_flags; diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index dcf7529918..d96215e8e0 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -79,7 +79,16 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial std::vector header; auto msg2 = CNetMsgMaker{msg.m_recv.GetVersion()}.Make(msg.m_type, Span{msg.m_recv}); - send_transport.prepareForTransport(msg2, header); + bool queued = send_transport.SetMessageToSend(msg2); + assert(queued); + std::optional known_more; + while (true) { + const auto& [to_send, more, _msg_type] = send_transport.GetBytesToSend(); + if (known_more) assert(!to_send.empty() == *known_more); + if (to_send.empty()) break; + send_transport.MarkBytesSent(to_send.size()); + known_more = more; + } } } } diff --git a/src/test/fuzz/process_messages.cpp b/src/test/fuzz/process_messages.cpp index 2617be3fa8..98962fceb5 100644 --- a/src/test/fuzz/process_messages.cpp +++ b/src/test/fuzz/process_messages.cpp @@ -67,7 +67,7 @@ FUZZ_TARGET(process_messages, .init = initialize_process_messages) CNode& random_node = *PickValue(fuzzed_data_provider, peers); - (void)connman.ReceiveMsgFrom(random_node, net_msg); + (void)connman.ReceiveMsgFrom(random_node, std::move(net_msg)); random_node.fPauseSend = false; try { diff --git a/src/test/util/net.cpp b/src/test/util/net.cpp index 0031770028..c071355bc0 100644 --- a/src/test/util/net.cpp +++ b/src/test/util/net.cpp @@ -41,7 +41,7 @@ void ConnmanTestMsg::Handshake(CNode& node, relay_txs), }; - (void)connman.ReceiveMsgFrom(node, msg_version); + (void)connman.ReceiveMsgFrom(node, std::move(msg_version)); node.fPauseSend = false; connman.ProcessMessagesOnce(node); peerman.SendMessages(&node); @@ -54,7 +54,7 @@ void ConnmanTestMsg::Handshake(CNode& node, assert(statestats.their_services == remote_services); if (successfully_connected) { CSerializedNetMsg msg_verack{mm.Make(NetMsgType::VERACK)}; - (void)connman.ReceiveMsgFrom(node, msg_verack); + (void)connman.ReceiveMsgFrom(node, std::move(msg_verack)); node.fPauseSend = false; connman.ProcessMessagesOnce(node); peerman.SendMessages(&node); @@ -70,14 +70,17 @@ void ConnmanTestMsg::NodeReceiveMsgBytes(CNode& node, Span msg_by } } -bool ConnmanTestMsg::ReceiveMsgFrom(CNode& node, CSerializedNetMsg& ser_msg) const +bool ConnmanTestMsg::ReceiveMsgFrom(CNode& node, CSerializedNetMsg&& ser_msg) const { - std::vector ser_msg_header; - node.m_transport->prepareForTransport(ser_msg, ser_msg_header); - - bool complete; - NodeReceiveMsgBytes(node, ser_msg_header, complete); - NodeReceiveMsgBytes(node, ser_msg.data, complete); + bool queued = node.m_transport->SetMessageToSend(ser_msg); + assert(queued); + bool complete{false}; + while (true) { + const auto& [to_send, _more, _msg_type] = node.m_transport->GetBytesToSend(); + if (to_send.empty()) break; + NodeReceiveMsgBytes(node, to_send, complete); + node.m_transport->MarkBytesSent(to_send.size()); + } return complete; } diff --git a/src/test/util/net.h b/src/test/util/net.h index b2f6ebb163..687ce1e813 100644 --- a/src/test/util/net.h +++ b/src/test/util/net.h @@ -54,7 +54,7 @@ struct ConnmanTestMsg : public CConnman { void NodeReceiveMsgBytes(CNode& node, Span msg_bytes, bool& complete) const; - bool ReceiveMsgFrom(CNode& node, CSerializedNetMsg& ser_msg) const; + bool ReceiveMsgFrom(CNode& node, CSerializedNetMsg&& ser_msg) const; }; constexpr ServiceFlags ALL_SERVICE_FLAGS[]{ From fb2c5edb79656a0b3b04ded6419928102ad696d6 Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Thu, 27 Jul 2023 15:35:41 -0400 Subject: [PATCH 5/9] net: make V1Transport implicitly use current chainparams The rest of net.cpp already uses Params() to determine chainparams in many places (and even V1Transport itself does so in some places). Since the only chainparams dependency is through the message start characters, just store those directly in the transport. --- src/net.cpp | 15 ++++++++++++--- src/net.h | 12 ++---------- src/test/fuzz/p2p_transport_serialization.cpp | 4 ++-- 3 files changed, 16 insertions(+), 15 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 1545e36e68..3a70f3690b 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -717,6 +717,15 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) return true; } +V1Transport::V1Transport(const NodeId node_id, int nTypeIn, int nVersionIn) noexcept : + m_node_id(node_id), hdrbuf(nTypeIn, nVersionIn), vRecv(nTypeIn, nVersionIn) +{ + assert(std::size(Params().MessageStart()) == std::size(m_magic_bytes)); + std::copy(std::begin(Params().MessageStart()), std::end(Params().MessageStart()), m_magic_bytes); + LOCK(m_recv_mutex); + Reset(); +} + int V1Transport::readHeader(Span msg_bytes) { AssertLockHeld(m_recv_mutex); @@ -741,7 +750,7 @@ int V1Transport::readHeader(Span msg_bytes) } // Check start string, network magic - if (memcmp(hdr.pchMessageStart, m_chain_params.MessageStart(), CMessageHeader::MESSAGE_START_SIZE) != 0) { + if (memcmp(hdr.pchMessageStart, m_magic_bytes, CMessageHeader::MESSAGE_START_SIZE) != 0) { LogPrint(BCLog::NET, "Header error: Wrong MessageStart %s received, peer=%d\n", HexStr(hdr.pchMessageStart), m_node_id); return -1; } @@ -835,7 +844,7 @@ bool V1Transport::SetMessageToSend(CSerializedNetMsg& msg) noexcept uint256 hash = Hash(msg.data); // create header - CMessageHeader hdr(Params().MessageStart(), msg.m_type.c_str(), msg.data.size()); + CMessageHeader hdr(m_magic_bytes, msg.m_type.c_str(), msg.data.size()); memcpy(hdr.pchChecksum, hash.begin(), CMessageHeader::CHECKSUM_SIZE); // serialize header @@ -2874,7 +2883,7 @@ CNode::CNode(NodeId idIn, ConnectionType conn_type_in, bool inbound_onion, CNodeOptions&& node_opts) - : m_transport{std::make_unique(Params(), idIn, SER_NETWORK, INIT_PROTO_VERSION)}, + : m_transport{std::make_unique(idIn, SER_NETWORK, INIT_PROTO_VERSION)}, m_permission_flags{node_opts.permission_flags}, m_sock{sock}, m_connected{GetTime()}, diff --git a/src/net.h b/src/net.h index 83deb4afed..b24b52226c 100644 --- a/src/net.h +++ b/src/net.h @@ -318,7 +318,7 @@ public: class V1Transport final : public Transport { private: - const CChainParams& m_chain_params; + CMessageHeader::MessageStartChars m_magic_bytes; const NodeId m_node_id; // Only for logging mutable Mutex m_recv_mutex; //!< Lock for receive state mutable CHash256 hasher GUARDED_BY(m_recv_mutex); @@ -365,15 +365,7 @@ private: size_t m_bytes_sent GUARDED_BY(m_send_mutex) {0}; public: - V1Transport(const CChainParams& chain_params, const NodeId node_id, int nTypeIn, int nVersionIn) - : m_chain_params(chain_params), - m_node_id(node_id), - hdrbuf(nTypeIn, nVersionIn), - vRecv(nTypeIn, nVersionIn) - { - LOCK(m_recv_mutex); - Reset(); - } + V1Transport(const NodeId node_id, int nTypeIn, int nVersionIn) noexcept; bool ReceivedMessageComplete() const override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index d96215e8e0..25e370bbf9 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -25,8 +25,8 @@ void initialize_p2p_transport_serialization() FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serialization) { // Construct transports for both sides, with dummy NodeIds. - V1Transport recv_transport{Params(), NodeId{0}, SER_NETWORK, INIT_PROTO_VERSION}; - V1Transport send_transport{Params(), NodeId{1}, SER_NETWORK, INIT_PROTO_VERSION}; + V1Transport recv_transport{NodeId{0}, SER_NETWORK, INIT_PROTO_VERSION}; + V1Transport send_transport{NodeId{1}, SER_NETWORK, INIT_PROTO_VERSION}; FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()}; From 009ff8d65058430d614c9a0e0e6ae931b7255c37 Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Tue, 25 Jul 2023 17:38:32 -0400 Subject: [PATCH 6/9] fuzz: add bidirectional fragmented transport test This adds a simulation test, with two V1Transport objects, which send messages to each other, with sending and receiving fragmented into multiple pieces that may be interleaved. It primarily verifies that the sending and receiving side are compatible with each other, plus a few sanity checks. --- src/test/fuzz/p2p_transport_serialization.cpp | 241 ++++++++++++++++++ 1 file changed, 241 insertions(+) diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index 25e370bbf9..8363779916 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -9,6 +9,8 @@ #include #include #include +#include +#include #include #include @@ -17,11 +19,19 @@ #include #include +namespace { + +std::vector g_all_messages; + void initialize_p2p_transport_serialization() { SelectParams(ChainType::REGTEST); + g_all_messages = getAllNetMessageTypes(); + std::sort(g_all_messages.begin(), g_all_messages.end()); } +} // namespace + FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serialization) { // Construct transports for both sides, with dummy NodeIds. @@ -92,3 +102,234 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial } } } + +namespace { + +template +void SimulationTest(Transport& initiator, Transport& responder, R& rng, FuzzedDataProvider& provider) +{ + // Simulation test with two Transport objects, which send messages to each other, with + // sending and receiving fragmented into multiple pieces that may be interleaved. It primarily + // verifies that the sending and receiving side are compatible with each other, plus a few + // sanity checks. It does not attempt to introduce errors in the communicated data. + + // Put the transports in an array for by-index access. + const std::array transports = {&initiator, &responder}; + + // Two vectors representing in-flight bytes. inflight[i] is from transport[i] to transport[!i]. + std::array, 2> in_flight; + + // Two queues with expected messages. expected[i] is expected to arrive in transport[!i]. + std::array, 2> expected; + + // Vectors with bytes last returned by GetBytesToSend() on transport[i]. + std::array, 2> to_send; + + // Last returned 'more' values (if still relevant) by transport[i]->GetBytesToSend(). + std::array, 2> last_more; + + // Whether more bytes to be sent are expected on transport[i]. + std::array, 2> expect_more; + + // Function to consume a message type. + auto msg_type_fn = [&]() { + uint8_t v = provider.ConsumeIntegral(); + if (v == 0xFF) { + // If v is 0xFF, construct a valid (but possibly unknown) message type from the fuzz + // data. + std::string ret; + while (ret.size() < CMessageHeader::COMMAND_SIZE) { + char c = provider.ConsumeIntegral(); + // Match the allowed characters in CMessageHeader::IsCommandValid(). Any other + // character is interpreted as end. + if (c < ' ' || c > 0x7E) break; + ret += c; + } + return ret; + } else { + // Otherwise, use it as index into the list of known messages. + return g_all_messages[v % g_all_messages.size()]; + } + }; + + // Function to construct a CSerializedNetMsg to send. + auto make_msg_fn = [&](bool first) { + CSerializedNetMsg msg; + if (first) { + // Always send a "version" message as first one. + msg.m_type = "version"; + } else { + msg.m_type = msg_type_fn(); + } + // Determine size of message to send (limited to 75 kB for performance reasons). + size_t size = provider.ConsumeIntegralInRange(0, 75000); + // Get payload of message from RNG. + msg.data.resize(size); + for (auto& v : msg.data) v = uint8_t(rng()); + // Return. + return msg; + }; + + // The next message to be sent (initially version messages, but will be replaced once sent). + std::array next_msg = { + make_msg_fn(/*first=*/true), + make_msg_fn(/*first=*/true) + }; + + // Wrapper around transport[i]->GetBytesToSend() that performs sanity checks. + auto bytes_to_send_fn = [&](int side) -> Transport::BytesToSend { + const auto& [bytes, more, msg_type] = transports[side]->GetBytesToSend(); + // Compare with expected more. + if (expect_more[side].has_value()) assert(!bytes.empty() == *expect_more[side]); + // Compare with previously reported output. + assert(to_send[side].size() <= bytes.size()); + assert(to_send[side] == Span{bytes}.first(to_send[side].size())); + to_send[side].resize(bytes.size()); + std::copy(bytes.begin(), bytes.end(), to_send[side].begin()); + // Remember 'more' result. + last_more[side] = {more}; + // Return. + return {bytes, more, msg_type}; + }; + + // Function to make side send a new message. + auto new_msg_fn = [&](int side) { + // Don't do anything if there are too many unreceived messages already. + if (expected[side].size() >= 16) return; + // Try to send (a copy of) the message in next_msg[side]. + CSerializedNetMsg msg = next_msg[side].Copy(); + bool queued = transports[side]->SetMessageToSend(msg); + // Update expected more data. + expect_more[side] = std::nullopt; + // Verify consistency of GetBytesToSend after SetMessageToSend + bytes_to_send_fn(/*side=*/side); + if (queued) { + // Remember that this message is now expected by the receiver. + expected[side].emplace_back(std::move(next_msg[side])); + // Construct a new next message to send. + next_msg[side] = make_msg_fn(/*first=*/false); + } + }; + + // Function to make side send out bytes (if any). + auto send_fn = [&](int side, bool everything = false) { + const auto& [bytes, more, msg_type] = bytes_to_send_fn(/*side=*/side); + // Don't do anything if no bytes to send. + if (bytes.empty()) return false; + size_t send_now = everything ? bytes.size() : provider.ConsumeIntegralInRange(0, bytes.size()); + if (send_now == 0) return false; + // Add bytes to the in-flight queue, and mark those bytes as consumed. + in_flight[side].insert(in_flight[side].end(), bytes.begin(), bytes.begin() + send_now); + transports[side]->MarkBytesSent(send_now); + // If all to-be-sent bytes were sent, move last_more data to expect_more data. + if (send_now == bytes.size()) { + expect_more[side] = last_more[side]; + } + // Remove the bytes from the last reported to-be-sent vector. + assert(to_send[side].size() >= send_now); + to_send[side].erase(to_send[side].begin(), to_send[side].begin() + send_now); + // Verify that GetBytesToSend gives a result consistent with earlier. + bytes_to_send_fn(/*side=*/side); + // Return whether anything was sent. + return send_now > 0; + }; + + // Function to make !side receive bytes (if any). + auto recv_fn = [&](int side, bool everything = false) { + // Don't do anything if no bytes in flight. + if (in_flight[side].empty()) return false; + // Decide span to receive + size_t to_recv_len = in_flight[side].size(); + if (!everything) to_recv_len = provider.ConsumeIntegralInRange(0, to_recv_len); + Span to_recv = Span{in_flight[side]}.first(to_recv_len); + // Process those bytes + while (!to_recv.empty()) { + size_t old_len = to_recv.size(); + bool ret = transports[!side]->ReceivedBytes(to_recv); + // Bytes must always be accepted, as this test does not introduce any errors in + // communication. + assert(ret); + // Clear cached expected 'more' information: if certainly no more data was to be sent + // before, receiving bytes makes this uncertain. + if (expect_more[!side] == false) expect_more[!side] = std::nullopt; + // Verify consistency of GetBytesToSend after ReceivedBytes + bytes_to_send_fn(/*side=*/!side); + bool progress = to_recv.size() < old_len; + if (transports[!side]->ReceivedMessageComplete()) { + bool reject{false}; + auto received = transports[!side]->GetReceivedMessage({}, reject); + // Receiving must succeed. + assert(!reject); + // There must be a corresponding expected message. + assert(!expected[side].empty()); + // The m_message_size field must be correct. + assert(received.m_message_size == received.m_recv.size()); + // The m_type must match what is expected. + assert(received.m_type == expected[side].front().m_type); + // The data must match what is expected. + assert(MakeByteSpan(received.m_recv) == MakeByteSpan(expected[side].front().data)); + expected[side].pop_front(); + progress = true; + } + // Progress must be made (by processing incoming bytes and/or returning complete + // messages) until all received bytes are processed. + assert(progress); + } + // Remove the processed bytes from the in_flight buffer. + in_flight[side].erase(in_flight[side].begin(), in_flight[side].begin() + to_recv_len); + // Return whether anything was received. + return to_recv_len > 0; + }; + + // Main loop, interleaving new messages, sends, and receives. + LIMITED_WHILE(provider.remaining_bytes(), 1000) { + CallOneOf(provider, + // (Try to) give the next message to the transport. + [&] { new_msg_fn(/*side=*/0); }, + [&] { new_msg_fn(/*side=*/1); }, + // (Try to) send some bytes from the transport to the network. + [&] { send_fn(/*side=*/0); }, + [&] { send_fn(/*side=*/1); }, + // (Try to) receive bytes from the network, converting to messages. + [&] { recv_fn(/*side=*/0); }, + [&] { recv_fn(/*side=*/1); } + ); + } + + // When we're done, perform sends and receives of existing messages to flush anything already + // in flight. + while (true) { + bool any = false; + if (send_fn(/*side=*/0, /*everything=*/true)) any = true; + if (send_fn(/*side=*/1, /*everything=*/true)) any = true; + if (recv_fn(/*side=*/0, /*everything=*/true)) any = true; + if (recv_fn(/*side=*/1, /*everything=*/true)) any = true; + if (!any) break; + } + + // Make sure nothing is left in flight. + assert(in_flight[0].empty()); + assert(in_flight[1].empty()); + + // Make sure all expected messages were received. + assert(expected[0].empty()); + assert(expected[1].empty()); +} + +std::unique_ptr MakeV1Transport(NodeId nodeid) noexcept +{ + return std::make_unique(nodeid, SER_NETWORK, INIT_PROTO_VERSION); +} + +} // namespace + +FUZZ_TARGET(p2p_transport_bidirectional, .init = initialize_p2p_transport_serialization) +{ + // Test with two V1 transports talking to each other. + FuzzedDataProvider provider{buffer.data(), buffer.size()}; + XoRoShiRo128PlusPlus rng(provider.ConsumeIntegral()); + auto t1 = MakeV1Transport(NodeId{0}); + auto t2 = MakeV1Transport(NodeId{1}); + if (!t1 || !t2) return; + SimulationTest(*t1, *t2, rng, provider); +} From a1a1060fd608a11dc525f76f2f54ab5b177dbd05 Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Mon, 24 Jul 2023 13:23:39 -0400 Subject: [PATCH 7/9] net: measure send buffer fullness based on memory usage This more accurately captures the intent of limiting send buffer size, as many small messages can have a larger overhead that is not counted with the current approach. It also means removing the dependency on the header size (which will become a function of the transport choice) from the send buffer calculations. --- src/init.cpp | 2 +- src/net.cpp | 38 ++++++++++++++++++++++++++++++++------ src/net.h | 12 ++++++++++-- 3 files changed, 43 insertions(+), 9 deletions(-) diff --git a/src/init.cpp b/src/init.cpp index 864c2e278b..2db473ec4b 100644 --- a/src/init.cpp +++ b/src/init.cpp @@ -490,7 +490,7 @@ void SetupServerArgs(ArgsManager& argsman) argsman.AddArg("-listenonion", strprintf("Automatically create Tor onion service (default: %d)", DEFAULT_LISTEN_ONION), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); argsman.AddArg("-maxconnections=", strprintf("Maintain at most connections to peers (default: %u). This limit does not apply to connections manually added via -addnode or the addnode RPC, which have a separate limit of %u.", DEFAULT_MAX_PEER_CONNECTIONS, MAX_ADDNODE_CONNECTIONS), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); argsman.AddArg("-maxreceivebuffer=", strprintf("Maximum per-connection receive buffer, *1000 bytes (default: %u)", DEFAULT_MAXRECEIVEBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); - argsman.AddArg("-maxsendbuffer=", strprintf("Maximum per-connection send buffer, *1000 bytes (default: %u)", DEFAULT_MAXSENDBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); + argsman.AddArg("-maxsendbuffer=", strprintf("Maximum per-connection memory usage for the send buffer, *1000 bytes (default: %u)", DEFAULT_MAXSENDBUFFER), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); argsman.AddArg("-maxtimeadjustment", strprintf("Maximum allowed median peer time offset adjustment. Local perspective of time may be influenced by outbound peers forward or backward by this amount (default: %u seconds).", DEFAULT_MAX_TIME_ADJUSTMENT), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); argsman.AddArg("-maxuploadtarget=", strprintf("Tries to keep outbound traffic under the given target per 24h. Limit does not apply to peers with 'download' permission or blocks created within past week. 0 = no limit (default: %s). Optional suffix units [k|K|m|M|g|G|t|T] (default: M). Lowercase is 1000 base while uppercase is 1024 base", DEFAULT_MAX_UPLOAD_TARGET), ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); argsman.AddArg("-onion=", "Use separate SOCKS5 proxy to reach peers via Tor onion services, set -noonion to disable (default: -proxy)", ArgsManager::ALLOW_ANY, OptionsCategory::CONNECTION); diff --git a/src/net.cpp b/src/net.cpp index 3a70f3690b..47b872a446 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -116,6 +117,14 @@ std::map mapLocalHost GUARDED_BY(g_maplocalhost_mute static bool vfLimited[NET_MAX] GUARDED_BY(g_maplocalhost_mutex) = {}; std::string strSubVersion; +size_t CSerializedNetMsg::GetMemoryUsage() const noexcept +{ + // Don't count the dynamic memory used for the m_type string, by assuming it fits in the + // "small string" optimization area (which stores data inside the object itself, up to some + // size; 15 bytes in modern libstdc++). + return sizeof(*this) + memusage::DynamicUsage(data); +} + void CConnman::AddAddrFetch(const std::string& strDest) { LOCK(m_addr_fetches_mutex); @@ -894,6 +903,14 @@ void V1Transport::MarkBytesSent(size_t bytes_sent) noexcept } } +size_t V1Transport::GetSendMemoryUsage() const noexcept +{ + AssertLockNotHeld(m_send_mutex); + LOCK(m_send_mutex); + // Don't count sending-side fields besides m_message_to_send, as they're all small and bounded. + return m_message_to_send.GetMemoryUsage(); +} + std::pair CConnman::SocketSendData(CNode& node) const { auto it = node.vSendMsg.begin(); @@ -923,8 +940,8 @@ std::pair CConnman::SocketSendData(CNode& node) const nSentSize += nBytes; if (node.nSendOffset == data.size()) { node.nSendOffset = 0; - node.nSendSize -= data.size(); - node.fPauseSend = node.nSendSize > nSendBufferMaxSize; + // Update memory usage of send buffer (as *it will be deleted). + node.m_send_memusage -= sizeof(data) + memusage::DynamicUsage(data); it++; } else { // could not send full message; stop sending more @@ -944,9 +961,11 @@ std::pair CConnman::SocketSendData(CNode& node) const } } + node.fPauseSend = node.m_send_memusage + node.m_transport->GetSendMemoryUsage() > nSendBufferMaxSize; + if (it == node.vSendMsg.end()) { assert(node.nSendOffset == 0); - assert(node.nSendSize == 0); + assert(node.m_send_memusage == 0); } node.vSendMsg.erase(node.vSendMsg.begin(), it); return {nSentSize, !node.vSendMsg.empty()}; @@ -2985,14 +3004,21 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) if (bytes.empty()) break; // Update statistics per message type. pnode->AccountForSentBytes(msg_type, bytes.size()); - // Update number of bytes in the send buffer. - pnode->nSendSize += bytes.size(); - if (pnode->nSendSize > nSendBufferMaxSize) pnode->fPauseSend = true; pnode->vSendMsg.push_back({bytes.begin(), bytes.end()}); + // Update memory usage of send buffer. For now, use static + dynamic memory usage of + // byte vectors in vSendMsg as send memory. In a future commit, vSendMsg will be + // replaced with a queue of CSerializedNetMsg objects, and we'll use their memory usage + // instead. + pnode->m_send_memusage += sizeof(pnode->vSendMsg.back()) + memusage::DynamicUsage(pnode->vSendMsg.back()); // Notify transport that bytes have been processed (they're not actually sent yet, // but pushed onto the vSendMsg queue of bytes to send). pnode->m_transport->MarkBytesSent(bytes.size()); } + // At this point, m_transport->GetSendMemoryUsage() isn't very interesting as the + // transport's message is fully flushed (and converted to byte arrays). It's still included + // here for correctness, and will become relevant in a future commit when a queued message + // inside the transport may survive PushMessage calls. + if (pnode->m_send_memusage + pnode->m_transport->GetSendMemoryUsage() > nSendBufferMaxSize) pnode->fPauseSend = true; // If the write queue was empty before and isn't now, attempt "optimistic write": // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually diff --git a/src/net.h b/src/net.h index b24b52226c..299d8c8e9b 100644 --- a/src/net.h +++ b/src/net.h @@ -122,6 +122,9 @@ struct CSerializedNetMsg { std::vector data; std::string m_type; + + /** Compute total memory usage of this object (own memory + any dynamic memory). */ + size_t GetMemoryUsage() const noexcept; }; /** @@ -313,6 +316,9 @@ public: * If bytes_sent=0, this call has no effect. */ virtual void MarkBytesSent(size_t bytes_sent) noexcept = 0; + + /** Return the memory usage of this transport attributable to buffered data to send. */ + virtual size_t GetSendMemoryUsage() const noexcept = 0; }; class V1Transport final : public Transport @@ -399,6 +405,7 @@ public: bool SetMessageToSend(CSerializedNetMsg& msg) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); BytesToSend GetBytesToSend() const noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); void MarkBytesSent(size_t bytes_sent) noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); + size_t GetSendMemoryUsage() const noexcept override EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex); }; struct CNodeOptions @@ -429,8 +436,9 @@ public: */ std::shared_ptr m_sock GUARDED_BY(m_sock_mutex); - /** Total size of all vSendMsg entries */ - size_t nSendSize GUARDED_BY(cs_vSend){0}; + /** Total memory usage of vSendMsg (counting the vectors and their dynamic usage, but not the + * deque overhead). */ + size_t m_send_memusage GUARDED_BY(cs_vSend){0}; /** Offset inside the first vSendMsg already sent */ size_t nSendOffset GUARDED_BY(cs_vSend){0}; uint64_t nSendBytes GUARDED_BY(cs_vSend){0}; From bb4aab90fd046f2fff61e082a0c0d01c5ee31297 Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Wed, 16 Aug 2023 13:31:50 -0400 Subject: [PATCH 8/9] net: move message conversion to wire bytes from PushMessage to SocketSendData This furthers transport abstraction by removing the assumption that a message can always immediately be converted to wire bytes. This assumption does not hold for the v2 transport proposed by BIP324, as no messages can be sent before the handshake completes. This is done by only keeping (complete) CSerializedNetMsg objects in vSendMsg, rather than the resulting bytes (for header and payload) that need to be sent. In SocketSendData, these objects are handed to the transport as permitted by it, and sending out the bytes the transport tells us to send. This also removes the nSendOffset member variable in CNode, as keeping track of how much has been sent is now a responsability of the transport. This is not a pure refactor, and has the following effects even for the current v1 transport: * Checksum calculation now happens in SocketSendData rather than PushMessage. For non-optimistic-send messages, that means this computation now happens in the network thread rather than the message handler thread (generally a good thing, as the message handler thread is more of a computational bottleneck). * Checksum calculation now happens while holding the cs_vSend lock. This is technically unnecessary for the v1 transport, as messages are encoded independent from one another, but is untenable for the v2 transport anyway. * Statistics updates about per-message sent bytes now happen when those bytes are actually handed to the OS, rather than at PushMessage time. --- src/net.cpp | 92 ++++++++++++++---------------- src/net.h | 9 ++- src/test/denialofservice_tests.cpp | 8 ++- src/test/fuzz/process_messages.cpp | 1 + src/test/util/net.cpp | 14 +++++ src/test/util/net.h | 1 + 6 files changed, 69 insertions(+), 56 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 47b872a446..9fce585b81 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -915,35 +915,49 @@ std::pair CConnman::SocketSendData(CNode& node) const { auto it = node.vSendMsg.begin(); size_t nSentSize = 0; + bool data_left{false}; //!< second return value (whether unsent data remains) - while (it != node.vSendMsg.end()) { - const auto& data = *it; - assert(data.size() > node.nSendOffset); + while (true) { + if (it != node.vSendMsg.end()) { + // If possible, move one message from the send queue to the transport. This fails when + // there is an existing message still being sent. + size_t memusage = it->GetMemoryUsage(); + if (node.m_transport->SetMessageToSend(*it)) { + // Update memory usage of send buffer (as *it will be deleted). + node.m_send_memusage -= memusage; + ++it; + } + } + const auto& [data, more, msg_type] = node.m_transport->GetBytesToSend(); + data_left = !data.empty(); // will be overwritten on next loop if all of data gets sent int nBytes = 0; - { + if (!data.empty()) { LOCK(node.m_sock_mutex); + // There is no socket in case we've already disconnected, or in test cases without + // real connections. In these cases, we bail out immediately and just leave things + // in the send queue and transport. if (!node.m_sock) { break; } int flags = MSG_NOSIGNAL | MSG_DONTWAIT; #ifdef MSG_MORE - if (it + 1 != node.vSendMsg.end()) { + // We have more to send if either the transport itself has more, or if we have more + // messages to send. + if (more || it != node.vSendMsg.end()) { flags |= MSG_MORE; } #endif - nBytes = node.m_sock->Send(reinterpret_cast(data.data()) + node.nSendOffset, data.size() - node.nSendOffset, flags); + nBytes = node.m_sock->Send(reinterpret_cast(data.data()), data.size(), flags); } if (nBytes > 0) { node.m_last_send = GetTime(); node.nSendBytes += nBytes; - node.nSendOffset += nBytes; + // Notify transport that bytes have been processed. + node.m_transport->MarkBytesSent(nBytes); + // Update statistics per message type. + node.AccountForSentBytes(msg_type, nBytes); nSentSize += nBytes; - if (node.nSendOffset == data.size()) { - node.nSendOffset = 0; - // Update memory usage of send buffer (as *it will be deleted). - node.m_send_memusage -= sizeof(data) + memusage::DynamicUsage(data); - it++; - } else { + if ((size_t)nBytes != data.size()) { // could not send full message; stop sending more break; } @@ -956,7 +970,6 @@ std::pair CConnman::SocketSendData(CNode& node) const node.CloseSocketDisconnect(); } } - // couldn't send anything at all break; } } @@ -964,11 +977,10 @@ std::pair CConnman::SocketSendData(CNode& node) const node.fPauseSend = node.m_send_memusage + node.m_transport->GetSendMemoryUsage() > nSendBufferMaxSize; if (it == node.vSendMsg.end()) { - assert(node.nSendOffset == 0); assert(node.m_send_memusage == 0); } node.vSendMsg.erase(node.vSendMsg.begin(), it); - return {nSentSize, !node.vSendMsg.empty()}; + return {nSentSize, data_left}; } /** Try to find a connection to evict when the node is full. @@ -1307,7 +1319,14 @@ Sock::EventsPerSock CConnman::GenerateWaitSockets(Span nodes) for (CNode* pnode : nodes) { bool select_recv = !pnode->fPauseRecv; - bool select_send = WITH_LOCK(pnode->cs_vSend, return !pnode->vSendMsg.empty()); + bool select_send; + { + LOCK(pnode->cs_vSend); + // Sending is possible if either there are bytes to send right now, or if there will be + // once a potential message from vSendMsg is handed to the transport. + const auto& [to_send, _more, _msg_type] = pnode->m_transport->GetBytesToSend(); + select_send = !to_send.empty() || !pnode->vSendMsg.empty(); + } if (!select_recv && !select_send) continue; LOCK(pnode->m_sock_mutex); @@ -2988,42 +3007,19 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) size_t nBytesSent = 0; { LOCK(pnode->cs_vSend); - const bool queue_was_empty{pnode->vSendMsg.empty()}; + const auto& [to_send, _more, _msg_type] = pnode->m_transport->GetBytesToSend(); + const bool queue_was_empty{to_send.empty() && pnode->vSendMsg.empty()}; - // Give the message to the transport, and add all bytes it wants us to send out as byte - // vectors to vSendMsg. This is temporary code that exists to support the new transport - // sending interface using the old way of queueing data. In a future commit vSendMsg will - // be replaced with a queue of CSerializedNetMsg objects to be sent instead, and this code - // will disappear. - bool queued = pnode->m_transport->SetMessageToSend(msg); - assert(queued); - // In the current transport (V1Transport), GetBytesToSend first returns a header to send, - // and then the payload data (if any), necessitating a loop. - while (true) { - const auto& [bytes, _more, msg_type] = pnode->m_transport->GetBytesToSend(); - if (bytes.empty()) break; - // Update statistics per message type. - pnode->AccountForSentBytes(msg_type, bytes.size()); - pnode->vSendMsg.push_back({bytes.begin(), bytes.end()}); - // Update memory usage of send buffer. For now, use static + dynamic memory usage of - // byte vectors in vSendMsg as send memory. In a future commit, vSendMsg will be - // replaced with a queue of CSerializedNetMsg objects, and we'll use their memory usage - // instead. - pnode->m_send_memusage += sizeof(pnode->vSendMsg.back()) + memusage::DynamicUsage(pnode->vSendMsg.back()); - // Notify transport that bytes have been processed (they're not actually sent yet, - // but pushed onto the vSendMsg queue of bytes to send). - pnode->m_transport->MarkBytesSent(bytes.size()); - } - // At this point, m_transport->GetSendMemoryUsage() isn't very interesting as the - // transport's message is fully flushed (and converted to byte arrays). It's still included - // here for correctness, and will become relevant in a future commit when a queued message - // inside the transport may survive PushMessage calls. + // Update memory usage of send buffer. + pnode->m_send_memusage += msg.GetMemoryUsage(); if (pnode->m_send_memusage + pnode->m_transport->GetSendMemoryUsage() > nSendBufferMaxSize) pnode->fPauseSend = true; + // Move message to vSendMsg queue. + pnode->vSendMsg.push_back(std::move(msg)); - // If the write queue was empty before and isn't now, attempt "optimistic write": + // If there was nothing to send before, attempt "optimistic write": // because the poll/select loop may pause for SELECT_TIMEOUT_MILLISECONDS before actually // doing a send, try sending from the calling thread if the queue was empty before. - if (queue_was_empty && !pnode->vSendMsg.empty()) { + if (queue_was_empty) { std::tie(nBytesSent, std::ignore) = SocketSendData(*pnode); } } diff --git a/src/net.h b/src/net.h index 299d8c8e9b..ac26b538d5 100644 --- a/src/net.h +++ b/src/net.h @@ -436,13 +436,12 @@ public: */ std::shared_ptr m_sock GUARDED_BY(m_sock_mutex); - /** Total memory usage of vSendMsg (counting the vectors and their dynamic usage, but not the - * deque overhead). */ + /** Sum of GetMemoryUsage of all vSendMsg entries. */ size_t m_send_memusage GUARDED_BY(cs_vSend){0}; - /** Offset inside the first vSendMsg already sent */ - size_t nSendOffset GUARDED_BY(cs_vSend){0}; + /** Total number of bytes sent on the wire to this peer. */ uint64_t nSendBytes GUARDED_BY(cs_vSend){0}; - std::deque> vSendMsg GUARDED_BY(cs_vSend); + /** Messages still to be fed to m_transport->SetMessageToSend. */ + std::deque vSendMsg GUARDED_BY(cs_vSend); Mutex cs_vSend; Mutex m_sock_mutex; Mutex cs_vRecv; diff --git a/src/test/denialofservice_tests.cpp b/src/test/denialofservice_tests.cpp index 90e5bb34ed..7f5d587cf6 100644 --- a/src/test/denialofservice_tests.cpp +++ b/src/test/denialofservice_tests.cpp @@ -86,9 +86,10 @@ BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction) { LOCK(dummyNode1.cs_vSend); - BOOST_CHECK(dummyNode1.vSendMsg.size() > 0); - dummyNode1.vSendMsg.clear(); + const auto& [to_send, _more, _msg_type] = dummyNode1.m_transport->GetBytesToSend(); + BOOST_CHECK(!to_send.empty()); } + connman.FlushSendBuffer(dummyNode1); int64_t nStartTime = GetTime(); // Wait 21 minutes @@ -96,7 +97,8 @@ BOOST_AUTO_TEST_CASE(outbound_slow_chain_eviction) BOOST_CHECK(peerman.SendMessages(&dummyNode1)); // should result in getheaders { LOCK(dummyNode1.cs_vSend); - BOOST_CHECK(dummyNode1.vSendMsg.size() > 0); + const auto& [to_send, _more, _msg_type] = dummyNode1.m_transport->GetBytesToSend(); + BOOST_CHECK(!to_send.empty()); } // Wait 3 more minutes SetMockTime(nStartTime+24*60); diff --git a/src/test/fuzz/process_messages.cpp b/src/test/fuzz/process_messages.cpp index 98962fceb5..4cb388c20b 100644 --- a/src/test/fuzz/process_messages.cpp +++ b/src/test/fuzz/process_messages.cpp @@ -67,6 +67,7 @@ FUZZ_TARGET(process_messages, .init = initialize_process_messages) CNode& random_node = *PickValue(fuzzed_data_provider, peers); + connman.FlushSendBuffer(random_node); (void)connman.ReceiveMsgFrom(random_node, std::move(net_msg)); random_node.fPauseSend = false; diff --git a/src/test/util/net.cpp b/src/test/util/net.cpp index c071355bc0..8015db3e80 100644 --- a/src/test/util/net.cpp +++ b/src/test/util/net.cpp @@ -25,6 +25,7 @@ void ConnmanTestMsg::Handshake(CNode& node, const CNetMsgMaker mm{0}; peerman.InitializeNode(node, local_services); + FlushSendBuffer(node); // Drop the version message added by InitializeNode. CSerializedNetMsg msg_version{ mm.Make(NetMsgType::VERSION, @@ -45,6 +46,7 @@ void ConnmanTestMsg::Handshake(CNode& node, node.fPauseSend = false; connman.ProcessMessagesOnce(node); peerman.SendMessages(&node); + FlushSendBuffer(node); // Drop the verack message added by SendMessages. if (node.fDisconnect) return; assert(node.nVersion == version); assert(node.GetCommonVersion() == std::min(version, PROTOCOL_VERSION)); @@ -70,6 +72,18 @@ void ConnmanTestMsg::NodeReceiveMsgBytes(CNode& node, Span msg_by } } +void ConnmanTestMsg::FlushSendBuffer(CNode& node) const +{ + LOCK(node.cs_vSend); + node.vSendMsg.clear(); + node.m_send_memusage = 0; + while (true) { + const auto& [to_send, _more, _msg_type] = node.m_transport->GetBytesToSend(); + if (to_send.empty()) break; + node.m_transport->MarkBytesSent(to_send.size()); + } +} + bool ConnmanTestMsg::ReceiveMsgFrom(CNode& node, CSerializedNetMsg&& ser_msg) const { bool queued = node.m_transport->SetMessageToSend(ser_msg); diff --git a/src/test/util/net.h b/src/test/util/net.h index 687ce1e813..1684da777a 100644 --- a/src/test/util/net.h +++ b/src/test/util/net.h @@ -55,6 +55,7 @@ struct ConnmanTestMsg : public CConnman { void NodeReceiveMsgBytes(CNode& node, Span msg_bytes, bool& complete) const; bool ReceiveMsgFrom(CNode& node, CSerializedNetMsg&& ser_msg) const; + void FlushSendBuffer(CNode& node) const; }; constexpr ServiceFlags ALL_SERVICE_FLAGS[]{ From 8a3b6f33873a1075f932f5d9feb6d82e50d83c0c Mon Sep 17 00:00:00 2001 From: Pieter Wuille Date: Sat, 29 Jul 2023 13:59:35 -0400 Subject: [PATCH 9/9] refactor: make Transport::ReceivedBytes just return success/fail --- src/net.cpp | 5 ++-- src/net.h | 23 +++++++++++++++---- src/test/fuzz/p2p_transport_serialization.cpp | 3 +-- 3 files changed, 21 insertions(+), 10 deletions(-) diff --git a/src/net.cpp b/src/net.cpp index 9fce585b81..e66c0ec7f8 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -690,9 +690,8 @@ bool CNode::ReceiveMsgBytes(Span msg_bytes, bool& complete) nRecvBytes += msg_bytes.size(); while (msg_bytes.size() > 0) { // absorb network data - int handled = m_transport->ReceivedBytes(msg_bytes); - if (handled < 0) { - // Serious header problem, disconnect from the peer. + if (!m_transport->ReceivedBytes(msg_bytes)) { + // Serious transport problem, disconnect from the peer. return false; } diff --git a/src/net.h b/src/net.h index ac26b538d5..60a15fea55 100644 --- a/src/net.h +++ b/src/net.h @@ -268,9 +268,22 @@ public: virtual bool ReceivedMessageComplete() const = 0; /** Set the deserialization context version for objects returned by GetReceivedMessage. */ virtual void SetReceiveVersion(int version) = 0; - /** Feed wire bytes to the transport; chops off consumed bytes off front of msg_bytes. */ - virtual int ReceivedBytes(Span& msg_bytes) = 0; - /** Retrieve a completed message from transport (only when ReceivedMessageComplete). */ + + /** Feed wire bytes to the transport. + * + * @return false if some bytes were invalid, in which case the transport can't be used anymore. + * + * Consumed bytes are chopped off the front of msg_bytes. + */ + virtual bool ReceivedBytes(Span& msg_bytes) = 0; + + /** Retrieve a completed message from transport. + * + * This can only be called when ReceivedMessageComplete() is true. + * + * If reject_message=true is returned the message itself is invalid, but (other than false + * returned by ReceivedBytes) the transport is not in an inconsistent state. + */ virtual CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) = 0; // 2. Sending side functions, for converting messages into bytes to be sent over the wire. @@ -387,7 +400,7 @@ public: vRecv.SetVersion(nVersionIn); } - int ReceivedBytes(Span& msg_bytes) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) + bool ReceivedBytes(Span& msg_bytes) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex) { AssertLockNotHeld(m_recv_mutex); LOCK(m_recv_mutex); @@ -397,7 +410,7 @@ public: } else { msg_bytes = msg_bytes.subspan(ret); } - return ret; + return ret >= 0; } CNetMessage GetReceivedMessage(std::chrono::microseconds time, bool& reject_message) override EXCLUSIVE_LOCKS_REQUIRED(!m_recv_mutex); diff --git a/src/test/fuzz/p2p_transport_serialization.cpp b/src/test/fuzz/p2p_transport_serialization.cpp index 8363779916..2fa5de5008 100644 --- a/src/test/fuzz/p2p_transport_serialization.cpp +++ b/src/test/fuzz/p2p_transport_serialization.cpp @@ -74,8 +74,7 @@ FUZZ_TARGET(p2p_transport_serialization, .init = initialize_p2p_transport_serial mutable_msg_bytes.insert(mutable_msg_bytes.end(), payload_bytes.begin(), payload_bytes.end()); Span msg_bytes{mutable_msg_bytes}; while (msg_bytes.size() > 0) { - const int handled = recv_transport.ReceivedBytes(msg_bytes); - if (handled < 0) { + if (!recv_transport.ReceivedBytes(msg_bytes)) { break; } if (recv_transport.ReceivedMessageComplete()) {