diff --git a/esphome/components/socket/lwip_raw_tcp_impl.cpp b/esphome/components/socket/lwip_raw_tcp_impl.cpp index 799b09e844d..63c577769fb 100644 --- a/esphome/components/socket/lwip_raw_tcp_impl.cpp +++ b/esphome/components/socket/lwip_raw_tcp_impl.cpp @@ -425,12 +425,16 @@ LWIPRawImpl::~LWIPRawImpl() { // Base class destructor handles pcb_ cleanup via tcp_abort } -void LWIPRawImpl::init() { +void LWIPRawImpl::init(struct pbuf *initial_rx) { LWIP_LOCK(); LWIP_LOG("init(%p)", this->pcb_); tcp_arg(this->pcb_, this); tcp_recv(this->pcb_, LWIPRawImpl::s_recv_fn); tcp_err(this->pcb_, LWIPRawImpl::s_err_fn); + if (initial_rx != nullptr) { + this->rx_buf_ = initial_rx; + this->rx_buf_offset_ = 0; + } } void LWIPRawImpl::s_err_fn(void *arg, err_t err) { @@ -666,9 +670,14 @@ LWIPRawListenImpl::~LWIPRawListenImpl() { LWIP_LOCK(); // Abort any queued PCBs that were never accepted by the main loop for (uint8_t i = 0; i < this->accepted_socket_count_; i++) { - if (this->accepted_pcbs_[i] != nullptr) { - tcp_abort(this->accepted_pcbs_[i]); - this->accepted_pcbs_[i] = nullptr; + auto &entry = this->accepted_pcbs_[i]; + if (entry.pcb != nullptr) { + tcp_abort(entry.pcb); + entry.pcb = nullptr; + } + if (entry.rx_buf != nullptr) { + pbuf_free(entry.rx_buf); + entry.rx_buf = nullptr; } } this->accepted_socket_count_ = 0; @@ -697,12 +706,32 @@ void LWIPRawListenImpl::s_err_fn(void *arg, err_t err) { arg_this->pcb_ = nullptr; } -void LWIPRawListenImpl::s_accepted_pcb_err_fn(void *arg, err_t err) { +void LWIPRawListenImpl::s_queued_err_fn(void *arg, err_t err) { // Called when a queued (not yet accepted) PCB errors — e.g., remote sent RST. // The PCB is already freed by lwip. Null our pointer so accept() skips it. (void) err; - auto *slot = reinterpret_cast(arg); - *slot = nullptr; + auto *entry = reinterpret_cast(arg); + entry->pcb = nullptr; + // Don't free rx_buf here — accept() will clean it up when it sees pcb==nullptr +} + +err_t LWIPRawListenImpl::s_queued_recv_fn(void *arg, struct tcp_pcb *pcb, struct pbuf *pb, err_t err) { + // Temporary recv callback for PCBs queued between accept_fn_ and accept(). + // Without this, lwip's default tcp_recv_null handler would ACK and drop the data, + // causing the API handshake to silently fail (client sends Hello, server never sees it). + (void) pcb; + auto *entry = reinterpret_cast(arg); + if (pb == nullptr || err != ERR_OK) { + // Remote closed or error + entry->rx_closed = true; + return ERR_OK; + } + if (entry->rx_buf == nullptr) { + entry->rx_buf = pb; + } else { + pbuf_cat(entry->rx_buf, pb); + } + return ERR_OK; } err_t LWIPRawListenImpl::s_accept_fn(void *arg, struct tcp_pcb *newpcb, err_t err) { @@ -720,32 +749,39 @@ std::unique_ptr LWIPRawListenImpl::accept(struct sockaddr *addr, so errno = EWOULDBLOCK; return nullptr; } - // Take raw PCB from front of queue - struct tcp_pcb *pcb = this->accepted_pcbs_[0]; - // Shift remaining PCBs forward + // Take entry from front of queue + QueuedPcb entry = this->accepted_pcbs_[0]; + // Shift remaining entries forward for (uint8_t i = 1; i < this->accepted_socket_count_; i++) { this->accepted_pcbs_[i - 1] = this->accepted_pcbs_[i]; } - this->accepted_pcbs_[this->accepted_socket_count_ - 1] = nullptr; + this->accepted_pcbs_[this->accepted_socket_count_ - 1] = {}; this->accepted_socket_count_--; // Update tcp_arg for remaining queued PCBs — their array slots shifted by one. - // Safe because we hold LWIP_LOCK, so err callbacks can't fire during the update. + // Safe because we hold LWIP_LOCK, so err/recv callbacks can't fire during the update. for (uint8_t i = 0; i < this->accepted_socket_count_; i++) { - if (this->accepted_pcbs_[i] != nullptr) { - tcp_arg(this->accepted_pcbs_[i], &this->accepted_pcbs_[i]); + if (this->accepted_pcbs_[i].pcb != nullptr) { + tcp_arg(this->accepted_pcbs_[i].pcb, &this->accepted_pcbs_[i]); } } LWIP_LOG("Connection accepted by application, queue size: %d", this->accepted_socket_count_); - if (pcb == nullptr) { + if (entry.pcb == nullptr) { // PCB was freed by lwip (RST/timeout) while queued — the temporary error callback - // nulled our pointer. Return EWOULDBLOCK so the caller retries next loop. + // nulled our pointer. Free any buffered data and return EWOULDBLOCK. + if (entry.rx_buf != nullptr) { + pbuf_free(entry.rx_buf); + } errno = EWOULDBLOCK; return nullptr; } // Create socket wrapper on the main loop (not in accept callback) to avoid - // heap allocation in IRQ context on RP2040. - auto sock = make_unique(this->family_, pcb); - sock->init(); + // heap allocation in IRQ context on RP2040. Transfer any data received while queued. + auto sock = make_unique(this->family_, entry.pcb); + sock->init(entry.rx_buf); + if (entry.rx_closed) { + // Remote closed while queued — mark so read() returns EOF after buffered data + sock->rx_closed_ = true; + } if (addr != nullptr) { sock->getpeername(addr, addrlen); } @@ -800,12 +836,15 @@ err_t LWIPRawListenImpl::accept_fn_(struct tcp_pcb *newpcb, err_t err) { // Store the raw PCB — LWIPRawImpl creation is deferred to the main-loop accept(). // This avoids heap allocation in this callback, which is unsafe from IRQ context on RP2040. uint8_t idx = this->accepted_socket_count_++; - this->accepted_pcbs_[idx] = newpcb; - // Register a temporary error callback so that if the connection errors (RST, timeout) - // before accept() picks it up, we null our pointer instead of leaving a dangling reference. - // tcp_arg points to our array slot; accept() updates these pointers after shifting. + this->accepted_pcbs_[idx] = {newpcb, nullptr, false}; + // Register temporary callbacks so that while the PCB is queued: + // - err: nulls our pointer if the connection errors (RST, timeout) + // - recv: buffers any data that arrives before accept() creates the LWIPRawImpl + // (without this, lwip's default tcp_recv_null would ACK and drop the data) + // tcp_arg points to our queue entry; accept() updates these pointers after shifting. tcp_arg(newpcb, &this->accepted_pcbs_[idx]); - tcp_err(newpcb, LWIPRawListenImpl::s_accepted_pcb_err_fn); + tcp_err(newpcb, LWIPRawListenImpl::s_queued_err_fn); + tcp_recv(newpcb, LWIPRawListenImpl::s_queued_recv_fn); LWIP_LOG("Accepted connection, queue size: %d", this->accepted_socket_count_); #if (defined(USE_ESP8266) || defined(USE_RP2040)) // Wake the main loop immediately so it can accept the new connection. diff --git a/esphome/components/socket/lwip_raw_tcp_impl.h b/esphome/components/socket/lwip_raw_tcp_impl.h index 1bba8ecf0e8..0fb8516b862 100644 --- a/esphome/components/socket/lwip_raw_tcp_impl.h +++ b/esphome/components/socket/lwip_raw_tcp_impl.h @@ -66,7 +66,7 @@ class LWIPRawImpl : public LWIPRawCommon { using LWIPRawCommon::LWIPRawCommon; ~LWIPRawImpl(); - void init(); + void init(struct pbuf *initial_rx = nullptr); // Non-listening sockets return error std::unique_ptr accept(struct sockaddr *, socklen_t *) { @@ -121,6 +121,8 @@ class LWIPRawImpl : public LWIPRawCommon { static void s_err_fn(void *arg, err_t err); static err_t s_recv_fn(void *arg, struct tcp_pcb *pcb, struct pbuf *pb, err_t err); + friend class LWIPRawListenImpl; // accept() transfers queued rx data + protected: ssize_t internal_write_(const void *buf, size_t len); int internal_output_(); @@ -177,20 +179,33 @@ class LWIPRawListenImpl : public LWIPRawCommon { int loop() { return 0; } static void s_err_fn(void *arg, err_t err); - static void s_accepted_pcb_err_fn(void *arg, err_t err); private: err_t accept_fn_(struct tcp_pcb *newpcb, err_t err); static err_t s_accept_fn(void *arg, struct tcp_pcb *newpcb, err_t err); - // Accept queue — stores raw tcp_pcb pointers instead of heap-allocated LWIPRawImpl objects. + // Temporary callbacks for queued PCBs (between accept_fn_ and accept()) + static void s_queued_err_fn(void *arg, err_t err); + static err_t s_queued_recv_fn(void *arg, struct tcp_pcb *pcb, struct pbuf *pb, err_t err); + + // Accept queue entry — stores a raw tcp_pcb and any data received while queued. + // lwip's default tcp_recv_null handler drops data and ACKs it, so we must register + // a temporary recv callback to buffer any data that arrives between accept_fn_ + // (which stores the PCB) and accept() (which creates the LWIPRawImpl). + struct QueuedPcb { + struct tcp_pcb *pcb{nullptr}; + struct pbuf *rx_buf{nullptr}; // Data received while queued (before accept() picks it up) + bool rx_closed{false}; // Remote sent FIN while queued + }; + + // Accept queue — stores raw tcp_pcb entries instead of heap-allocated LWIPRawImpl objects. // LWIPRawImpl creation is deferred to the main-loop accept() call. This avoids: // - Heap allocation in the accept callback (unsafe from IRQ context on RP2040) // - Dangling LWIPRawImpl if the connection errors before accept() picks it up // 2 slots is plenty since the main loop drains the queue every iteration. static constexpr size_t MAX_ACCEPTED_SOCKETS = 2; - std::array accepted_pcbs_{}; - uint8_t accepted_socket_count_ = 0; // Number of PCBs currently in queue + std::array accepted_pcbs_{}; + uint8_t accepted_socket_count_ = 0; // Number of entries currently in queue }; } // namespace esphome::socket