From 7b018b9dcad0dec6af34f52a964315b8f6717d8f Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 2 Oct 2026 09:07:35 -0500 Subject: [PATCH] [socket] Move the outgoing buffer into the TCP client link (#20010) --- esphome/components/socket/tcp_client_link.cpp | 24 ++++++++++++++- esphome/components/socket/tcp_client_link.h | 28 ++++++++++++++++-- esphome/components/tcp_uart/tcp_uart.cpp | 29 ++++--------------- esphome/components/tcp_uart/tcp_uart.h | 8 ++--- .../tcp_client_link_test_component.cpp | 10 +++++-- 5 files changed, 64 insertions(+), 35 deletions(-) diff --git a/esphome/components/socket/tcp_client_link.cpp b/esphome/components/socket/tcp_client_link.cpp index f2c1b8e4c1..141695a4bb 100644 --- a/esphome/components/socket/tcp_client_link.cpp +++ b/esphome/components/socket/tcp_client_link.cpp @@ -7,6 +7,7 @@ #include #include +#include namespace esphome::socket { @@ -114,7 +115,7 @@ ssize_t TcpClientLink::read(uint8_t *buf, size_t len) { return 0; } -ssize_t TcpClientLink::write(const uint8_t *buf, size_t len) { +ssize_t TcpClientLink::write_(const uint8_t *buf, size_t len) { if (!this->connected_ || len == 0) { return 0; } @@ -129,6 +130,26 @@ ssize_t TcpClientLink::write(const uint8_t *buf, size_t len) { return -1; } +size_t TcpClientLink::queue(const uint8_t *data, size_t len) { + size_t room = this->tx_free(); + if (len > room) { + len = room; + } + std::memcpy(this->tx_ + this->tx_len_, data, len); + this->tx_len_ += static_cast(len); + return len; +} + +void TcpClientLink::flush_tx_slow_() { + ssize_t sent = this->write_(this->tx_, this->tx_len_); + if (sent > 0) { + this->tx_len_ -= static_cast(sent); + if (this->tx_len_ != 0) { + std::memmove(this->tx_, this->tx_ + sent, this->tx_len_); + } + } +} + void TcpClientLink::close() { if (this->sock_ != nullptr) { this->sock_->shutdown(SHUT_RDWR); @@ -136,6 +157,7 @@ void TcpClientLink::close() { this->sock_.reset(); } this->connected_ = false; + this->tx_len_ = 0; this->resolved_.forget(); } diff --git a/esphome/components/socket/tcp_client_link.h b/esphome/components/socket/tcp_client_link.h index 065c4df562..f77ee05055 100644 --- a/esphome/components/socket/tcp_client_link.h +++ b/esphome/components/socket/tcp_client_link.h @@ -16,8 +16,9 @@ namespace esphome::socket { /// A reconnecting TCP stream driven from loop(). Owns the socket, the DNS -/// lookup and the retry backoff. A fatal read/write error closes the link -/// and schedules the next attempt; the caller sees the edge via connected(). +/// lookup, the retry backoff and the outgoing buffer. A fatal read/write +/// error closes the link and schedules the next attempt; the caller sees +/// the edge via connected(). class TcpClientLink { public: void set_host(const char *host) { this->host_ = StringRef(host); } @@ -41,7 +42,21 @@ class TcpClientLink { void adopt(std::unique_ptr sock); /// Returns bytes moved, 0 when nothing can move now, -1 when the link dropped. ssize_t read(uint8_t *buf, size_t len); - ssize_t write(const uint8_t *buf, size_t len); + /// Copy into the outgoing buffer; returns how many bytes fit. + size_t queue(const uint8_t *data, size_t len); + /// Direct access to the buffer's free tail. Fill at most tx_free() bytes, + /// then tx_commit() the count; neither is bounds checked. + uint8_t *tx_tail() { return this->tx_ + this->tx_len_; } + void tx_commit(size_t len) { this->tx_len_ += static_cast(len); } + size_t tx_free() const { return this->connected_ ? TX_BUFFER_SIZE - this->tx_len_ : 0; } + /// Send the front of the buffer; true once it is empty. + /// A partial write keeps the rest; inline no-op while nothing is queued. + bool flush_tx() { + if (this->tx_len_ != 0) { + this->flush_tx_slow_(); + } + return this->tx_len_ == 0; + } /// Close without scheduling a reconnect (shutdown). void close(); @@ -54,6 +69,11 @@ class TcpClientLink { } protected: + static constexpr size_t TX_BUFFER_SIZE = 1024; + + /// The raw stream write behind flush_tx(); drops the link on a fatal error. + ssize_t write_(const uint8_t *buf, size_t len); + void flush_tx_slow_(); void poll_slow_(); void try_connect_(); /// Close after a failure, log what and errno, schedule the next attempt. @@ -66,7 +86,9 @@ class TcpClientLink { uint32_t reconnect_interval_ms_{5000}; Ipv4Resolve resolved_; uint16_t port_{0}; + uint16_t tx_len_{0}; bool connected_{false}; + uint8_t tx_[TX_BUFFER_SIZE]{}; }; } // namespace esphome::socket diff --git a/esphome/components/tcp_uart/tcp_uart.cpp b/esphome/components/tcp_uart/tcp_uart.cpp index fc81f6ecfe..00b59d82ba 100644 --- a/esphome/components/tcp_uart/tcp_uart.cpp +++ b/esphome/components/tcp_uart/tcp_uart.cpp @@ -33,7 +33,6 @@ void TcpUart::sync_link_() { this->link_was_up_ = up; if (!up) { this->rx_start_ = this->rx_end_ = 0; - this->tx_len_ = 0; } if (this->connected_sensor_ != nullptr) { this->connected_sensor_->publish_state(up); @@ -63,14 +62,6 @@ void TcpUart::read_socket_() { this->rx_pending_ = static_cast(count) == room; } -void TcpUart::flush_tx_() { - ssize_t sent = this->link_.write(this->tx_, this->tx_len_); - if (sent > 0) { - this->tx_len_ -= static_cast(sent); - std::memmove(this->tx_, this->tx_ + sent, this->tx_len_); - } -} - void TcpUart::loop() { this->link_.poll(); if (this->link_.connected() != this->link_was_up_) { @@ -82,25 +73,20 @@ void TcpUart::loop() { if (this->rx_pending_ || this->link_.ready()) { this->read_socket_(); } - if (this->tx_len_ != 0) { - this->flush_tx_(); - } + this->link_.flush_tx(); } void TcpUart::write_array(const uint8_t *data, size_t len) { - size_t room = this->link_.connected() ? sizeof(this->tx_) - this->tx_len_ : 0; - if (len > room) { + size_t queued = this->link_.queue(data, len); + if (queued < len) { uint32_t now = App.get_loop_component_start_time(); if (this->last_drop_log_ms_ == 0 || now - this->last_drop_log_ms_ >= DROP_LOG_INTERVAL_MS) { ESP_LOGW(TAG, "%s, dropped %u bytes", this->link_.connected() ? LOG_STR_LITERAL("TX buffer full") : LOG_STR_LITERAL("Not connected"), - static_cast(len - room)); + static_cast(len - queued)); this->last_drop_log_ms_ = now; } - len = room; } - std::memcpy(this->tx_ + this->tx_len_, data, len); - this->tx_len_ += static_cast(len); } bool TcpUart::peek_byte(uint8_t *data) { @@ -121,11 +107,8 @@ bool TcpUart::read_array(uint8_t *data, size_t len) { } uart::UARTFlushResult TcpUart::flush() { - this->flush_tx_(); - if (this->tx_len_ == 0) { - return uart::UARTFlushResult::UART_FLUSH_RESULT_SUCCESS; - } - return uart::UARTFlushResult::UART_FLUSH_RESULT_TIMEOUT; + return this->link_.flush_tx() ? uart::UARTFlushResult::UART_FLUSH_RESULT_SUCCESS + : uart::UARTFlushResult::UART_FLUSH_RESULT_TIMEOUT; } } // namespace esphome::tcp_uart diff --git a/esphome/components/tcp_uart/tcp_uart.h b/esphome/components/tcp_uart/tcp_uart.h index c13d2ef6f0..7ce822c2fb 100644 --- a/esphome/components/tcp_uart/tcp_uart.h +++ b/esphome/components/tcp_uart/tcp_uart.h @@ -32,7 +32,7 @@ class TcpUart : public uart::UARTComponent, public Component { bool read_array(uint8_t *data, size_t len) override; size_t available() override { return static_cast(this->rx_end_ - this->rx_start_); } // Same room write_array() grants, so consumers can apply backpressure. - size_t available_for_write() override { return this->link_.connected() ? sizeof(this->tx_) - this->tx_len_ : 0; } + size_t available_for_write() override { return this->link_.tx_free(); } uart::UARTFlushResult flush() override; bool is_connected() override { return this->link_.connected(); } #if defined(USE_ESP8266) || defined(USE_ESP32) @@ -43,24 +43,20 @@ class TcpUart : public uart::UARTComponent, public Component { void check_logger_conflict() override {} void sync_link_(); void read_socket_(); - void flush_tx_(); static constexpr size_t RX_BUFFER_SIZE = 1024; - static constexpr size_t TX_BUFFER_SIZE = 1024; socket::TcpClientLink link_; binary_sensor::BinarySensor *connected_sensor_{nullptr}; uint32_t last_drop_log_ms_{0}; - uint16_t tx_len_{0}; // rx_[rx_start_, rx_end_) holds unread bytes; read_socket_() compacts to the front. uint16_t rx_start_{0}; uint16_t rx_end_{0}; - // The link state loop() saw last; edges clear the buffers and publish the sensor. + // The link state loop() saw last; edges clear rx_ and publish the sensor. bool link_was_up_{false}; // A read stopped before EAGAIN. ready() stays false until new data arrives. bool rx_pending_{false}; uint8_t rx_[RX_BUFFER_SIZE]{}; - uint8_t tx_[TX_BUFFER_SIZE]{}; }; } // namespace esphome::tcp_uart diff --git a/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp b/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp index 7f2af8add8..bd45e60a00 100644 --- a/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp +++ b/tests/integration/fixtures/external_components/tcp_client_link_test_component/tcp_client_link_test_component.cpp @@ -14,14 +14,20 @@ void TcpClientLinkTestComponent::loop() { this->was_up_ = up; ESP_LOGI(TAG, "Link %s", up ? LOG_STR_LITERAL("up") : LOG_STR_LITERAL("down")); } - if (!up || !this->link_.ready()) { + if (!up) { + return; + } + // Retries a partial echo; inline no-op when nothing is queued. + this->link_.flush_tx(); + if (!this->link_.ready()) { return; } uint8_t buf[64]; ssize_t count = this->link_.read(buf, sizeof(buf)); if (count > 0) { ESP_LOGI(TAG, "Echoing %d bytes", static_cast(count)); - this->link_.write(buf, static_cast(count)); + this->link_.queue(buf, static_cast(count)); + this->link_.flush_tx(); } }