From 357947f2f08d345f973f6962a9fa5dedb709bcfe Mon Sep 17 00:00:00 2001 From: Per K Date: Mon, 27 Nov 2023 13:03:52 +0100 Subject: [PATCH] Reapply "Refactor AsyncTcpSocket(s) to use rtc::ReceivedPackets" This reverts commit 264547d084d8625c60a31b15843779173d3c95b8. Refactor AsyncTcpSocket(s) to use rtc::ReceivedPackets Patchset 1 contains original cl. Newer patchsets contains fix of the problem from pathset 1. Bug: webrtc:15368, webrtc:11943 Change-Id: Ib8c4c06daf502a5dec8c31beea78eacac8c3c644 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/328820 Commit-Queue: Per Kjellander Reviewed-by: Harald Alvestrand Reviewed-by: Jonas Oreland Cr-Commit-Position: refs/heads/main@{#41255} --- p2p/base/async_stun_tcp_socket.cc | 33 +++++++++------- p2p/base/async_stun_tcp_socket.h | 2 +- p2p/base/async_stun_tcp_socket_unittest.cc | 44 ++++++++++++++++++---- rtc_base/async_tcp_socket.cc | 40 +++++++++++--------- rtc_base/async_tcp_socket.h | 6 ++- 5 files changed, 84 insertions(+), 41 deletions(-) diff --git a/p2p/base/async_stun_tcp_socket.cc b/p2p/base/async_stun_tcp_socket.cc index 4a35903dfe..dcaf7d285d 100644 --- a/p2p/base/async_stun_tcp_socket.cc +++ b/p2p/base/async_stun_tcp_socket.cc @@ -14,9 +14,15 @@ #include #include +#include +#include + +#include "api/array_view.h" #include "api/transport/stun.h" +#include "api/units/timestamp.h" #include "rtc_base/byte_order.h" #include "rtc_base/checks.h" +#include "rtc_base/network/received_packet.h" #include "rtc_base/network/sent_packet.h" #include "rtc_base/time_utils.h" @@ -89,7 +95,7 @@ int AsyncStunTCPSocket::Send(const void* pv, return static_cast(cb); } -void AsyncStunTCPSocket::ProcessInput(char* data, size_t* len) { +size_t AsyncStunTCPSocket::ProcessInput(rtc::ArrayView data) { rtc::SocketAddress remote_addr(GetRemoteAddress()); // STUN packet - First 4 bytes. Total header size is 20 bytes. // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ @@ -101,26 +107,27 @@ void AsyncStunTCPSocket::ProcessInput(char* data, size_t* len) { // | Channel Number | Length | // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + size_t processed_bytes = 0; while (true) { + size_t bytes_left = data.size() - processed_bytes; // We need at least 4 bytes to read the STUN or ChannelData packet length. - if (*len < kPacketLenOffset + kPacketLenSize) - return; + if (bytes_left < kPacketLenOffset + kPacketLenSize) + return processed_bytes; int pad_bytes; - size_t expected_pkt_len = GetExpectedLength(data, *len, &pad_bytes); + size_t expected_pkt_len = GetExpectedLength(data.data() + processed_bytes, + bytes_left, &pad_bytes); size_t actual_length = expected_pkt_len + pad_bytes; - if (*len < actual_length) { - return; + if (bytes_left < actual_length) { + return processed_bytes; } - SignalReadPacket(this, data, expected_pkt_len, remote_addr, - rtc::TimeMicros()); - - *len -= actual_length; - if (*len > 0) { - memmove(data, data + actual_length, *len); - } + rtc::ReceivedPacket received_packet( + data.subview(processed_bytes, expected_pkt_len), remote_addr, + webrtc::Timestamp::Micros(rtc::TimeMicros())); + NotifyPacketReceived(received_packet); + processed_bytes += actual_length; } } diff --git a/p2p/base/async_stun_tcp_socket.h b/p2p/base/async_stun_tcp_socket.h index f0df42b52a..2c43d554fe 100644 --- a/p2p/base/async_stun_tcp_socket.h +++ b/p2p/base/async_stun_tcp_socket.h @@ -37,7 +37,7 @@ class AsyncStunTCPSocket : public rtc::AsyncTCPSocketBase { int Send(const void* pv, size_t cb, const rtc::PacketOptions& options) override; - void ProcessInput(char* data, size_t* len) override; + size_t ProcessInput(rtc::ArrayView data) override; private: // This method returns the message hdr + length written in the header. diff --git a/p2p/base/async_stun_tcp_socket_unittest.cc b/p2p/base/async_stun_tcp_socket_unittest.cc index 72d6a7fde0..853fbb471f 100644 --- a/p2p/base/async_stun_tcp_socket_unittest.cc +++ b/p2p/base/async_stun_tcp_socket_unittest.cc @@ -13,12 +13,17 @@ #include #include +#include #include #include #include #include #include "absl/memory/memory.h" +#include "api/array_view.h" +#include "rtc_base/buffer.h" +#include "rtc_base/byte_buffer.h" +#include "rtc_base/network/received_packet.h" #include "rtc_base/network/sent_packet.h" #include "rtc_base/socket.h" #include "rtc_base/third_party/sigslot/sigslot.h" @@ -96,11 +101,10 @@ class AsyncStunTCPSocketTest : public ::testing::Test, } void OnReadPacket(rtc::AsyncPacketSocket* socket, - const char* data, - size_t len, - const rtc::SocketAddress& remote_addr, - const int64_t& /* packet_time_us */) { - recv_packets_.push_back(std::string(data, len)); + const rtc::ReceivedPacket& packet) { + recv_packets_.push_back( + std::string(reinterpret_cast(packet.payload().data()), + packet.payload().size())); } void OnSentPacket(rtc::AsyncPacketSocket* socket, @@ -111,8 +115,10 @@ class AsyncStunTCPSocketTest : public ::testing::Test, void OnNewConnection(rtc::AsyncListenSocket* /*server*/, rtc::AsyncPacketSocket* new_socket) { recv_socket_ = absl::WrapUnique(new_socket); - new_socket->SignalReadPacket.connect(this, - &AsyncStunTCPSocketTest::OnReadPacket); + new_socket->RegisterReceivedPacketCallback( + [&](rtc::AsyncPacketSocket* socket, const rtc::ReceivedPacket& packet) { + OnReadPacket(socket, packet); + }); } bool Send(const void* data, size_t len) { @@ -164,6 +170,30 @@ TEST_F(AsyncStunTCPSocketTest, TestMultipleStunPackets) { EXPECT_EQ(4u, recv_packets_.size()); } +TEST_F(AsyncStunTCPSocketTest, ProcessInputHandlesMultiplePackets) { + send_socket_->RegisterReceivedPacketCallback( + [&](rtc::AsyncPacketSocket* socket, const rtc::ReceivedPacket& packet) { + recv_packets_.push_back( + std::string(reinterpret_cast(packet.payload().data()), + packet.payload().size())); + }); + rtc::Buffer buffer; + buffer.AppendData(kStunMessageWithZeroLength, + sizeof(kStunMessageWithZeroLength)); + // ChannelData message MUST be padded to + // a multiple of four bytes. + const unsigned char kTurnChannelData[] = { + 0x40, 0x00, 0x00, 0x04, 0x21, 0x12, 0xA4, 0x42, + }; + buffer.AppendData(kTurnChannelData, sizeof(kTurnChannelData)); + + send_socket_->ProcessInput(buffer); + EXPECT_EQ(2u, recv_packets_.size()); + EXPECT_TRUE(CheckData(kStunMessageWithZeroLength, + sizeof(kStunMessageWithZeroLength))); + EXPECT_TRUE(CheckData(kTurnChannelData, sizeof(kTurnChannelData))); +} + // Verifying TURN channel data message with zero length. TEST_F(AsyncStunTCPSocketTest, TestTurnChannelDataWithZeroLength) { EXPECT_TRUE(Send(kTurnChannelDataMessageWithZeroLength, diff --git a/rtc_base/async_tcp_socket.cc b/rtc_base/async_tcp_socket.cc index eed4a31c38..2924340c16 100644 --- a/rtc_base/async_tcp_socket.cc +++ b/rtc_base/async_tcp_socket.cc @@ -14,6 +14,8 @@ #include #include +#include +#include #include #include "api/array_view.h" @@ -209,15 +211,17 @@ void AsyncTCPSocketBase::OnReadEvent(Socket* socket) { return; } - size_t size = inbuf_.size(); - ProcessInput(inbuf_.data(), &size); - - if (size > inbuf_.size()) { + size_t processed = ProcessInput(inbuf_); + size_t bytes_remaining = inbuf_.size() - processed; + if (processed > inbuf_.size()) { RTC_LOG(LS_ERROR) << "input buffer overflow"; RTC_DCHECK_NOTREACHED(); inbuf_.Clear(); } else { - inbuf_.SetSize(size); + if (bytes_remaining > 0) { + memmove(inbuf_.data(), inbuf_.data() + processed, bytes_remaining); + } + inbuf_.SetSize(bytes_remaining); } } @@ -283,24 +287,24 @@ int AsyncTCPSocket::Send(const void* pv, return static_cast(cb); } -void AsyncTCPSocket::ProcessInput(char* data, size_t* len) { +size_t AsyncTCPSocket::ProcessInput(rtc::ArrayView data) { SocketAddress remote_addr(GetRemoteAddress()); + size_t processed_bytes = 0; while (true) { - if (*len < kPacketLenSize) - return; + size_t bytes_left = data.size() - processed_bytes; + if (bytes_left < kPacketLenSize) + return processed_bytes; - PacketLength pkt_len = rtc::GetBE16(data); - if (*len < kPacketLenSize + pkt_len) - return; + PacketLength pkt_len = rtc::GetBE16(data.data() + processed_bytes); + if (bytes_left < kPacketLenSize + pkt_len) + return processed_bytes; - NotifyPacketReceived(rtc::ReceivedPacket::CreateFromLegacy( - data + kPacketLenSize, pkt_len, rtc::TimeMicros(), remote_addr)); - - *len -= kPacketLenSize + pkt_len; - if (*len > 0) { - memmove(data, data + kPacketLenSize + pkt_len, *len); - } + rtc::ReceivedPacket received_packet( + data.subview(processed_bytes + kPacketLenSize, pkt_len), remote_addr, + webrtc::Timestamp::Micros(rtc::TimeMicros())); + NotifyPacketReceived(received_packet); + processed_bytes += kPacketLenSize + pkt_len; } } diff --git a/rtc_base/async_tcp_socket.h b/rtc_base/async_tcp_socket.h index 90f77d618e..d3aff60520 100644 --- a/rtc_base/async_tcp_socket.h +++ b/rtc_base/async_tcp_socket.h @@ -16,6 +16,7 @@ #include #include +#include "api/array_view.h" #include "rtc_base/async_packet_socket.h" #include "rtc_base/buffer.h" #include "rtc_base/socket.h" @@ -38,7 +39,8 @@ class AsyncTCPSocketBase : public AsyncPacketSocket { int Send(const void* pv, size_t cb, const rtc::PacketOptions& options) override = 0; - virtual void ProcessInput(char* data, size_t* len) = 0; + // Must return the number of bytes processed. + virtual size_t ProcessInput(rtc::ArrayView data) = 0; SocketAddress GetLocalAddress() const override; SocketAddress GetRemoteAddress() const override; @@ -100,7 +102,7 @@ class AsyncTCPSocket : public AsyncTCPSocketBase { int Send(const void* pv, size_t cb, const rtc::PacketOptions& options) override; - void ProcessInput(char* data, size_t* len) override; + size_t ProcessInput(rtc::ArrayView) override; }; class AsyncTcpListenSocket : public AsyncListenSocket {