From fccfab8083872656f62ae256f4cf9b288d4eb14e Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 10 Mar 2026 01:15:57 -1000 Subject: [PATCH] [socket] Switch UDP rx queue from lock-free SPSC to count-based with LWIP_LOCK With the lwip lock infrastructure in place from the TCP race fix, the lock-free SPSC ring buffer's wasted slot is no longer needed. Switch to a simple rx_count_ approach that uses all 4 queue slots. Also add LWIP_LOCK() to all UDP methods that call lwip APIs (bind, close, sendto, setsockopt, getsockopt, recvfrom, factories). --- .../components/socket/lwip_raw_tcp_impl.cpp | 33 +++++++++++++------ esphome/components/socket/lwip_raw_tcp_impl.h | 18 +++++----- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/esphome/components/socket/lwip_raw_tcp_impl.cpp b/esphome/components/socket/lwip_raw_tcp_impl.cpp index 159bc08a3f..27313b95e6 100644 --- a/esphome/components/socket/lwip_raw_tcp_impl.cpp +++ b/esphome/components/socket/lwip_raw_tcp_impl.cpp @@ -799,6 +799,7 @@ LWIPRawUDPImpl::LWIPRawUDPImpl(sa_family_t family) : family_(family) { } LWIPRawUDPImpl::~LWIPRawUDPImpl() { + LWIP_LOCK(); if (this->pcb_ != nullptr) { udp_remove(this->pcb_); this->pcb_ = nullptr; @@ -806,6 +807,7 @@ LWIPRawUDPImpl::~LWIPRawUDPImpl() { } int LWIPRawUDPImpl::bind_internal_(const struct sockaddr *name, socklen_t addrlen) { + LWIP_LOCK(); if (this->pcb_ == nullptr) { errno = EBADF; return -1; @@ -847,6 +849,7 @@ int LWIPRawUDPImpl::bind_internal_(const struct sockaddr *name, socklen_t addrle int LWIPRawUDPImpl::bind(const struct sockaddr *name, socklen_t addrlen) { return this->bind_internal_(name, addrlen); } int LWIPRawUDPImpl::close() { + LWIP_LOCK(); if (this->pcb_ == nullptr) { errno = EBADF; return -1; @@ -899,6 +902,7 @@ int LWIPRawUDPImpl::ip2sockaddr_(const ip_addr_t *ip, uint16_t port, struct sock ssize_t LWIPRawUDPImpl::sendto(const void *buf, size_t len, int flags, const struct sockaddr *dest_addr, socklen_t addrlen) { (void) flags; // Flags (MSG_DONTWAIT, etc.) are ignored; raw lwip is always non-blocking + LWIP_LOCK(); if (this->pcb_ == nullptr) { errno = EBADF; return -1; @@ -940,6 +944,7 @@ ssize_t LWIPRawUDPImpl::sendto(const void *buf, size_t len, int flags, const str } int LWIPRawUDPImpl::setsockopt(int level, int optname, const void *optval, socklen_t optlen) { + LWIP_LOCK(); if (this->pcb_ == nullptr) { errno = EBADF; return -1; @@ -1001,6 +1006,7 @@ int LWIPRawUDPImpl::setsockopt(int level, int optname, const void *optval, sockl } int LWIPRawUDPImpl::getsockopt(int level, int optname, void *optval, socklen_t *optlen) { + LWIP_LOCK(); if (this->pcb_ == nullptr) { errno = EBADF; return -1; @@ -1036,24 +1042,27 @@ LWIPRawUDPRecvImpl::~LWIPRawUDPRecvImpl() { } int LWIPRawUDPRecvImpl::close() { + LWIP_LOCK(); // Unregister recv callback before removing pcb if (this->pcb_ != nullptr) { udp_recv(this->pcb_, nullptr, nullptr); } // Flush any queued rx packets - while (this->rx_read_idx_ != this->rx_write_idx_) { + while (this->rx_count_ > 0) { auto &pkt = this->rx_queue_[this->rx_read_idx_]; if (pkt.pb != nullptr) { pbuf_free(pkt.pb); pkt.pb = nullptr; } this->rx_read_idx_ = (this->rx_read_idx_ + 1) & UDP_RX_MASK; + this->rx_count_--; } // close() returns EBADF if already closed, which is fine from destructor return LWIPRawUDPImpl::close(); } int LWIPRawUDPRecvImpl::bind(const struct sockaddr *name, socklen_t addrlen) { + LWIP_LOCK(); int ret = this->bind_internal_(name, addrlen); if (ret != 0) return ret; @@ -1073,7 +1082,8 @@ ssize_t LWIPRawUDPRecvImpl::recvfrom(void *buf, size_t len, struct sockaddr *src errno = EINVAL; return -1; } - if (this->rx_read_idx_ == this->rx_write_idx_) { + LWIP_LOCK(); + if (this->rx_count_ == 0) { errno = EWOULDBLOCK; return -1; } @@ -1090,11 +1100,11 @@ ssize_t LWIPRawUDPRecvImpl::recvfrom(void *buf, size_t len, struct sockaddr *src this->ip2sockaddr_(&pkt.src_addr, pkt.src_port, src_addr, addrlen); } - // Free the pbuf and advance the read pointer — must be last, - // as this publishes the slot to the producer (recv callback). + // Free the pbuf and advance the read pointer pbuf_free(pkt.pb); pkt.pb = nullptr; this->rx_read_idx_ = (this->rx_read_idx_ + 1) & UDP_RX_MASK; + this->rx_count_--; return (ssize_t) copy_len; } @@ -1104,24 +1114,25 @@ void LWIPRawUDPRecvImpl::s_recv_fn(void *arg, struct udp_pcb *pcb, struct pbuf * self->recv_fn_(p, addr, port); } +// Called by lwip core which already holds the async_context lock on RP2040. void LWIPRawUDPRecvImpl::recv_fn_(struct pbuf *p, const ip_addr_t *addr, u16_t port) { if (p == nullptr) return; - // Check if queue is full (next write position would collide with read position) - uint8_t next_write = (this->rx_write_idx_ + 1) & UDP_RX_MASK; - if (next_write == this->rx_read_idx_) { + // Check if queue is full + if (this->rx_count_ >= UDP_RX_QUEUE_SIZE) { // Drop packet — queue full pbuf_free(p); return; } - // Enqueue the packet — write data first, then publish by advancing write index. - auto &slot = this->rx_queue_[this->rx_write_idx_]; + // Enqueue the packet + uint8_t write_idx = (this->rx_read_idx_ + this->rx_count_) & UDP_RX_MASK; + auto &slot = this->rx_queue_[write_idx]; slot.pb = p; slot.src_addr = *addr; slot.src_port = port; - this->rx_write_idx_ = next_write; + this->rx_count_++; #if defined(USE_ESP8266) || defined(USE_RP2040) socket_wake(); @@ -1152,6 +1163,7 @@ std::unique_ptr socket_loop_monitored(int domain, int type, int protocol std::unique_ptr socket_udp(int domain, int protocol) { (void) protocol; // Raw lwip UDP ignores protocol; kept for API compatibility + LWIP_LOCK(); auto sock = make_unique((sa_family_t) domain); if (!sock->is_valid()) { errno = ENOMEM; @@ -1162,6 +1174,7 @@ std::unique_ptr socket_udp(int domain, int protocol) { std::unique_ptr socket_udp_recv(int domain, int protocol) { (void) protocol; // Raw lwip UDP ignores protocol; kept for API compatibility + LWIP_LOCK(); auto sock = make_unique((sa_family_t) domain); if (!sock->is_valid()) { errno = ENOMEM; diff --git a/esphome/components/socket/lwip_raw_tcp_impl.h b/esphome/components/socket/lwip_raw_tcp_impl.h index f6cf136e9a..4255611a8f 100644 --- a/esphome/components/socket/lwip_raw_tcp_impl.h +++ b/esphome/components/socket/lwip_raw_tcp_impl.h @@ -269,19 +269,19 @@ class LWIPRawUDPRecvImpl : public LWIPRawUDPImpl { ssize_t recvfrom(void *buf, size_t len, struct sockaddr *src_addr, socklen_t *addrlen); /// Returns true if there are packets available to read. - bool ready() const { return this->rx_read_idx_ != this->rx_write_idx_; } + /// Intentionally unlocked — same rationale as LWIPRawImpl::ready(). + bool ready() const { return this->rx_count_ > 0; } protected: static void s_recv_fn(void *arg, struct udp_pcb *pcb, struct pbuf *p, const ip_addr_t *addr, u16_t port); void recv_fn_(struct pbuf *p, const ip_addr_t *addr, u16_t port); - /// Lock-free SPSC ring buffer for received UDP packets. - /// Producer (recv callback, possibly IRQ context on RP2040) writes rx_write_idx_. - /// Consumer (main loop) writes rx_read_idx_. - /// No shared read-modify-write — safe without locking. - /// One slot is reserved to distinguish full from empty, giving 3 usable slots. + /// Ring buffer for received UDP packets. + /// Both producer (recv callback) and consumer (main loop) are serialized by the + /// lwip lock — the callback runs under lwip core lock, and consumer methods hold + /// LWIP_LOCK(). All 4 slots are usable (no wasted slot for full/empty distinction). /// No heap allocation in the recv callback — packets are dropped if the queue is full. - static constexpr uint8_t UDP_RX_QUEUE_SIZE = 4; // Must be power of 2 + static constexpr uint8_t UDP_RX_QUEUE_SIZE = 4; static constexpr uint8_t UDP_RX_MASK = UDP_RX_QUEUE_SIZE - 1; static_assert((UDP_RX_QUEUE_SIZE & UDP_RX_MASK) == 0, "UDP_RX_QUEUE_SIZE must be power of 2"); struct UDPRxPacket { @@ -290,8 +290,8 @@ class LWIPRawUDPRecvImpl : public LWIPRawUDPImpl { uint16_t src_port{0}; }; std::array rx_queue_{}; - volatile uint8_t rx_read_idx_{0}; ///< Written by consumer (main loop), read by producer - volatile uint8_t rx_write_idx_{0}; ///< Written by producer (recv callback), read by consumer + uint8_t rx_read_idx_{0}; + uint8_t rx_count_{0}; }; } // namespace esphome::socket