clear error first

This commit is contained in:
J. Nick Koston
2026-03-10 14:46:44 -10:00
2 changed files with 84 additions and 30 deletions
+64 -25
View File
@@ -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) {
@@ -668,10 +672,15 @@ LWIPRawListenImpl::~LWIPRawListenImpl() {
// Clear the error callback first — tcp_abort triggers it, and we don't
// want s_accepted_pcb_err_fn writing to slots during destruction.
for (uint8_t i = 0; i < this->accepted_socket_count_; i++) {
if (this->accepted_pcbs_[i] != nullptr) {
tcp_err(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_err(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;
@@ -700,12 +709,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<struct tcp_pcb **>(arg);
*slot = nullptr;
auto *entry = reinterpret_cast<QueuedPcb *>(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<QueuedPcb *>(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) {
@@ -723,32 +752,39 @@ std::unique_ptr<LWIPRawImpl> 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<LWIPRawImpl>(this->family_, pcb);
sock->init();
// heap allocation in IRQ context on RP2040. Transfer any data received while queued.
auto sock = make_unique<LWIPRawImpl>(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);
}
@@ -803,12 +839,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.
+20 -5
View File
@@ -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<LWIPRawImpl> 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<struct tcp_pcb *, MAX_ACCEPTED_SOCKETS> accepted_pcbs_{};
uint8_t accepted_socket_count_ = 0; // Number of PCBs currently in queue
std::array<QueuedPcb, MAX_ACCEPTED_SOCKETS> accepted_pcbs_{};
uint8_t accepted_socket_count_ = 0; // Number of entries currently in queue
};
} // namespace esphome::socket