Add outgoing connections to the raw lwIP TCP socket implementation

This commit is contained in:
J. Nick Koston
2026-09-05 13:37:42 +02:00
parent 84f78831f9
commit 8ce65558e1
12 changed files with 284 additions and 84 deletions
@@ -42,7 +42,16 @@ bool AsyncClient::connect(const char *host, uint16_t port) {
return false;
}
socket_->setblocking(false);
if (socket_->setblocking(false) != 0) {
// Capture before the log and reset() below can clobber errno; a blocking
// connect()/read() would otherwise stall the whole loop
const int saved_errno = errno;
ESP_LOGE(TAG, "Failed to set nonblocking: errno %d", saved_errno);
socket_.reset();
if (error_cb_)
error_cb_(error_arg_, this, saved_errno);
return false;
}
int err = socket_->connect((struct sockaddr *) &addr, addrlen);
if (err == 0) {
@@ -97,45 +106,22 @@ void AsyncClient::loop() {
return;
if (connecting_) {
// For connecting, we need to check writability, not readability
// The Application's select() only monitors read FDs, so we do our own check here
// For ESP platforms lwip_select() might be faster, but this code isn't used
// on those platforms anyway. If it was, we'd fix the Application select()
// to report writability instead of doing it this way.
int fd = socket_->get_fd();
if (fd < 0) {
ESP_LOGW(TAG, "Invalid socket fd");
close();
return;
}
fd_set writefds;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
struct timeval tv = {0, 0};
int ret = select(fd + 1, nullptr, &writefds, nullptr, &tv);
if (ret > 0 && FD_ISSET(fd, &writefds)) {
int error = 0;
socklen_t len = sizeof(error);
if (socket_->getsockopt(SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) {
int err = 0;
switch (socket::poll_connect(*socket_, err)) {
case socket::ConnectPollResult::CONNECT_POLL_PENDING:
break;
case socket::ConnectPollResult::CONNECT_POLL_CONNECTED:
connecting_ = false;
connected_ = true;
if (connect_cb_)
connect_cb_(connect_arg_, this);
} else {
ESP_LOGW(TAG, "Connection failed: %d", error);
break;
case socket::ConnectPollResult::CONNECT_POLL_ERROR:
ESP_LOGW(TAG, "Connection failed: %d", err);
close();
if (error_cb_)
error_cb_(error_arg_, this, error);
}
} else if (ret < 0) {
const int err = errno;
ESP_LOGE(TAG, "Select error: %d", err);
close();
if (error_cb_)
error_cb_(error_arg_, this, err);
error_cb_(error_arg_, this, err);
break;
}
} else if (connected_) {
// For connected sockets, use the Application's select() results
@@ -407,7 +407,10 @@ void ESPHomeOTAComponent::handle_data_() {
tv.tv_usec = 0;
this->client_->setsockopt(SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
this->client_->setsockopt(SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
this->client_->setblocking(true);
if (this->client_->setblocking(true) != 0) {
this->log_socket_error_(LOG_STR("blocking"));
goto error; // NOLINT(cppcoreguidelines-avoid-goto)
}
// Acknowledge auth OK - 1 byte
this->data_write_byte_(ota::OTA_RESPONSE_AUTH_OK);
@@ -59,13 +59,15 @@ int BSDSocketImpl::close() {
int BSDSocketImpl::setblocking(bool blocking) {
int fl = ::fcntl(this->fd_, F_GETFL, 0);
if (fl < 0) {
return fl;
}
if (blocking) {
fl &= ~O_NONBLOCK;
} else {
fl |= O_NONBLOCK;
}
::fcntl(this->fd_, F_SETFL, fl);
return 0;
return ::fcntl(this->fd_, F_SETFL, fl);
}
size_t BSDSocketImpl::getpeername_to(std::span<char, SOCKADDR_STR_LEN> buf) {
+7
View File
@@ -205,6 +205,13 @@ static constexpr size_t SOCKADDR_STR_LEN = 46; // INET6_ADDRSTRLEN
static constexpr size_t SOCKADDR_STR_LEN = 16; // INET_ADDRSTRLEN
#endif
/// Outcome of polling a non-blocking connect(); see socket::poll_connect().
enum class ConnectPollResult : uint8_t {
CONNECT_POLL_PENDING,
CONNECT_POLL_CONNECTED,
CONNECT_POLL_ERROR,
};
} // namespace esphome::socket
#endif
+145 -41
View File
@@ -63,7 +63,8 @@ static constexpr uint32_t ESP8266_YIELD_INTERVAL_US = 1000;
// tcp_abort() triggers the err callback synchronously, which would
// otherwise call back into a partially-destroyed object.
// tcp_sent/tcp_poll are not cleared because this implementation
// never registers them.
// never registers them, and the tcp_connect callback only fires on
// SYN_SENT -> ESTABLISHED, which cannot follow an abort or close.
static void pcb_detach_abort(struct tcp_pcb *pcb) {
tcp_arg(pcb, nullptr);
tcp_recv(pcb, nullptr);
@@ -76,8 +77,8 @@ static void pcb_detach_abort(struct tcp_pcb *pcb) {
// After tcp_close(), the PCB remains alive during the TCP close handshake
// (FIN_WAIT, TIME_WAIT states). Without clearing callbacks first, LWIP
// would call recv/err on a destroyed socket object, corrupting the heap.
// tcp_sent/tcp_poll are not cleared because this implementation
// never registers them.
// tcp_sent/tcp_poll and the tcp_connect callback are not cleared for the
// same reasons as in pcb_detach_abort().
// Returns ERR_OK on success; on failure the PCB is aborted instead.
static err_t pcb_detach_close(struct tcp_pcb *pcb) {
tcp_arg(pcb, nullptr);
@@ -101,53 +102,61 @@ LWIPRawCommon::~LWIPRawCommon() {
}
}
bool LWIPRawCommon::sockaddr2ip_(const struct sockaddr *name, socklen_t addrlen, ip_addr_t *ip, uint16_t *port) const {
if (name == nullptr) {
errno = EINVAL;
return false;
}
#if LWIP_IPV6
if (this->family_ == AF_INET) {
if (addrlen < sizeof(sockaddr_in)) {
errno = EINVAL;
return false;
}
auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
*port = ntohs(addr4->sin_port);
ip->type = IPADDR_TYPE_V4;
ip->u_addr.ip4.addr = addr4->sin_addr.s_addr;
return true;
}
if (this->family_ == AF_INET6) {
if (addrlen < sizeof(sockaddr_in6)) {
errno = EINVAL;
return false;
}
auto *addr6 = reinterpret_cast<const sockaddr_in6 *>(name);
*port = ntohs(addr6->sin6_port);
// ANY lets bind() accept both families; connect() picks the concrete type
ip->type = IPADDR_TYPE_ANY;
memcpy(&ip->u_addr.ip6.addr, &addr6->sin6_addr.un.u8_addr, 16);
return true;
}
errno = EINVAL;
return false;
#else
if (this->family_ != AF_INET || addrlen < sizeof(sockaddr_in)) {
errno = EINVAL;
return false;
}
auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
*port = ntohs(addr4->sin_port);
ip->addr = addr4->sin_addr.s_addr;
return true;
#endif
}
int LWIPRawCommon::bind(const struct sockaddr *name, socklen_t addrlen) {
LWIP_LOCK();
if (this->pcb_ == nullptr) {
errno = EBADF;
return -1;
}
if (name == nullptr) {
errno = EINVAL;
return -1;
}
ip_addr_t ip;
in_port_t port;
#if LWIP_IPV6
if (this->family_ == AF_INET) {
if (addrlen < sizeof(sockaddr_in)) {
errno = EINVAL;
return -1;
}
auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
port = ntohs(addr4->sin_port);
ip.type = IPADDR_TYPE_V4;
ip.u_addr.ip4.addr = addr4->sin_addr.s_addr;
LWIP_LOG("tcp_bind(%p ip=%s port=%u)", this->pcb_, ip4addr_ntoa(&ip.u_addr.ip4), port);
} else if (this->family_ == AF_INET6) {
if (addrlen < sizeof(sockaddr_in6)) {
errno = EINVAL;
return -1;
}
auto *addr6 = reinterpret_cast<const sockaddr_in6 *>(name);
port = ntohs(addr6->sin6_port);
ip.type = IPADDR_TYPE_ANY;
memcpy(&ip.u_addr.ip6.addr, &addr6->sin6_addr.un.u8_addr, 16);
LWIP_LOG("tcp_bind(%p ip=%s port=%u)", this->pcb_, ip6addr_ntoa(&ip.u_addr.ip6), port);
} else {
errno = EINVAL;
uint16_t port;
if (!this->sockaddr2ip_(name, addrlen, &ip, &port)) {
return -1;
}
#else
if (this->family_ != AF_INET) {
errno = EINVAL;
return -1;
}
auto *addr4 = reinterpret_cast<const sockaddr_in *>(name);
port = ntohs(addr4->sin_port);
ip.addr = addr4->sin_addr.s_addr;
LWIP_LOG("tcp_bind(%p ip=%u port=%u)", this->pcb_, ip.addr, port);
#endif
LWIP_LOG("tcp_bind(%p ip=%s port=%u)", this->pcb_, ipaddr_ntoa(&ip), port);
err_t err = tcp_bind(this->pcb_, &ip, port);
if (err == ERR_USE) {
LWIP_LOG(" -> err ERR_USE");
@@ -425,9 +434,104 @@ void LWIPRawImpl::s_err_fn(void *arg, err_t err) {
// ERR_ABRT: aborted through tcp_abort or TCP timer
auto *arg_this = reinterpret_cast<LWIPRawImpl *>(arg);
ESP_LOGVV(TAG, "socket %p: err(err=%d)", arg_this, err);
if (arg_this->connect_err_ == EINPROGRESS) {
// A connect that never established: RST is a refusal, anything else is
// the SYN retransmits giving up. Written before pcb_ so poll_connect()
// never sees a dead pcb without its reason.
arg_this->connect_err_ = err == ERR_RST ? ECONNREFUSED : ETIMEDOUT;
}
arg_this->pcb_ = nullptr;
esphome::wake_loop_any_context();
}
err_t LWIPRawImpl::s_connected_fn(void *arg, struct tcp_pcb *pcb, err_t err) {
// LWIP CALLBACK — same constraints as s_err_fn. lwip always passes ERR_OK
// here; a failed connect arrives through s_err_fn instead.
auto *arg_this = reinterpret_cast<LWIPRawImpl *>(arg);
arg_this->connect_err_ = err == ERR_OK ? 0 : ECONNRESET;
esphome::wake_loop_any_context();
return ERR_OK;
}
int LWIPRawImpl::connect(const struct sockaddr *addr, socklen_t addrlen) {
LWIP_LOCK();
if (this->pcb_ == nullptr) {
errno = EBADF;
return -1;
}
if (this->connect_err_ != 0) {
errno = EALREADY;
return -1;
}
ip_addr_t ip;
uint16_t port;
if (!this->sockaddr2ip_(addr, addrlen, &ip, &port)) {
return -1;
}
#if LWIP_IPV6
// tcp_connect needs a concrete address type. A remembered IPv4 peer on an
// IPv6 build arrives as a v4-mapped address and must be dialed as IPv4.
if (IP_IS_ANY_TYPE_VAL(ip)) {
if (ip6_addr_isipv4mappedipv6(ip_2_ip6(&ip))) {
unmap_ipv4_mapped_ipv6(ip_2_ip4(&ip), ip_2_ip6(&ip));
IP_SET_TYPE_VAL(ip, IPADDR_TYPE_V4);
} else {
IP_SET_TYPE_VAL(ip, IPADDR_TYPE_V6);
}
}
#endif
LWIP_LOG("tcp_connect(%p ip=%s port=%u)", this->pcb_, ipaddr_ntoa(&ip), port);
err_t err = tcp_connect(this->pcb_, &ip, port, LWIPRawImpl::s_connected_fn);
switch (err) {
case ERR_OK:
this->connect_err_ = EINPROGRESS;
errno = EINPROGRESS;
return -1;
case ERR_RTE:
errno = EHOSTUNREACH; // no route or no address yet; callers retry
break;
case ERR_USE:
errno = EADDRINUSE;
break;
case ERR_ISCONN:
errno = EISCONN;
break;
case ERR_BUF:
errno = EAGAIN; // no free local port
break;
case ERR_MEM:
errno = ENOMEM;
break;
default:
errno = EINVAL;
break;
}
LWIP_LOG(" -> err %d", err);
return -1;
}
ConnectPollResult LWIPRawImpl::poll_connect(int &err_out) const {
// pcb_ first: s_err_fn records the reason before it clears the pcb
if (this->pcb_ == nullptr) {
err_out = this->connect_err_ == 0 || this->connect_err_ == EINPROGRESS ? ECONNRESET : this->connect_err_;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
if (this->connect_err_ == EINPROGRESS) {
#ifdef USE_ESP8266
// Let SYS process the SYN-ACK between polls; see read()
optimistic_yield(ESP8266_YIELD_INTERVAL_US);
#endif
return ConnectPollResult::CONNECT_POLL_PENDING;
}
if (this->connect_err_ != 0) {
err_out = this->connect_err_;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
return ConnectPollResult::CONNECT_POLL_CONNECTED;
}
ConnectPollResult poll_connect(Socket &sock, int &err_out) { return sock.poll_connect(err_out); }
err_t LWIPRawImpl::s_recv_fn(void *arg, struct tcp_pcb *pcb, struct pbuf *pb, err_t err) {
auto *arg_this = reinterpret_cast<LWIPRawImpl *>(arg);
return arg_this->recv_fn(pb, err);
@@ -50,6 +50,9 @@ class LWIPRawCommon {
protected:
int ip2sockaddr_(ip_addr_t *ip, uint16_t port, struct sockaddr *name, socklen_t *addrlen);
/// Convert a sockaddr of this socket's family to an lwip address and port.
/// Returns false with errno set on a family or length mismatch.
bool sockaddr2ip_(const struct sockaddr *name, socklen_t addrlen, ip_addr_t *ip, uint16_t *port) const;
// Member ordering optimized to minimize padding on 32-bit systems
struct tcp_pcb *pcb_;
@@ -58,6 +61,12 @@ class LWIPRawCommon {
bool nodelay_ = false;
sa_family_t family_ = 0;
uint8_t recv_timeout_cs_ = 0; // SO_RCVTIMEO in centiseconds (0 = no timeout, max 2.55s)
// State of a connect() started on this socket: 0 when none is pending (or
// it completed), EINPROGRESS while the SYN is out, otherwise the errno the
// lwip callbacks recorded for its failure. Fits the padding byte here.
uint8_t connect_err_ = 0;
static_assert(EINPROGRESS < 256 && ECONNREFUSED < 256 && ECONNRESET < 256 && ETIMEDOUT < 256,
"connect_err_ stores errno values in a byte");
};
/// Connected socket implementation for LWIP raw TCP.
@@ -83,6 +92,13 @@ class LWIPRawImpl : public LWIPRawCommon {
errno = EOPNOTSUPP;
return -1;
}
/// Start a non-blocking connect. Always returns -1 with errno EINPROGRESS
/// when the SYN was queued; completion is reported by poll_connect().
int connect(const struct sockaddr *addr, socklen_t addrlen);
// Intentionally unlocked like ready(): reads one pointer and one byte that
// the callbacks write in the order the checks depend on (error byte first,
// then pcb_), so a torn read only costs one extra poll.
ConnectPollResult poll_connect(int &err_out) const;
ssize_t read(void *buf, size_t len);
ssize_t readv(const struct iovec *iov, int iovcnt);
ssize_t recvfrom(void *, size_t, sockaddr *, socklen_t *) {
@@ -120,6 +136,7 @@ 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);
static err_t s_connected_fn(void *arg, struct tcp_pcb *pcb, err_t err);
protected:
// True when the socket could receive data but none has arrived yet.
@@ -49,13 +49,15 @@ int LwIPSocketImpl::close() {
int LwIPSocketImpl::setblocking(bool blocking) {
int fl = lwip_fcntl(this->fd_, F_GETFL, 0);
if (fl < 0) {
return fl;
}
if (blocking) {
fl &= ~O_NONBLOCK;
} else {
fl |= O_NONBLOCK;
}
lwip_fcntl(this->fd_, F_SETFL, fl);
return 0;
return lwip_fcntl(this->fd_, F_SETFL, fl);
}
size_t LwIPSocketImpl::getpeername_to(std::span<char, SOCKADDR_STR_LEN> buf) {
+54 -2
View File
@@ -2,6 +2,9 @@
#if defined(USE_SOCKET_IMPL_LWIP_TCP) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) || defined(USE_SOCKET_IMPL_BSD_SOCKETS)
#include <cerrno>
#include <cstring>
#ifdef USE_SOCKET_IMPL_BSD_SOCKETS
#include <sys/select.h>
#endif
#include <string>
#include "esphome/core/log.h"
#include "esphome/core/application.h"
@@ -165,7 +168,10 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
#else
// Use LWIP-specific functions
ip6_addr_t ip6;
inet6_aton(ip_address, &ip6);
if (inet6_aton(ip_address, &ip6) == 0) {
errno = EINVAL;
return 0;
}
memcpy(server->sin6_addr.un.u32_addr, ip6.addr, sizeof(ip6.addr));
#endif
return sizeof(sockaddr_in6);
@@ -185,12 +191,58 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
return 0;
}
#else
server->sin_addr.s_addr = inet_addr(ip_address);
// Unlike inet_addr(), inet_aton() can signal failure while still
// accepting the broadcast address 255.255.255.255
if (inet_aton(ip_address, &server->sin_addr) == 0) {
errno = EINVAL;
return 0;
}
#endif
server->sin_port = htons(port);
return sizeof(sockaddr_in);
}
#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
ConnectPollResult poll_connect(Socket &sock, int &err_out) {
int fd = sock.get_fd();
if (fd < 0 || fd >= FD_SETSIZE) {
// FD_SET on either is undefined behavior
err_out = EBADF;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
// Connect completion is a write event; the main loop only selects on reads
fd_set writefds;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
struct timeval tv = {0, 0};
#ifdef USE_SOCKET_IMPL_LWIP_SOCKETS
// LWIP_COMPAT_SOCKETS may be off (LibreTiny), so use the lwip symbol directly
int ret = lwip_select(fd + 1, nullptr, &writefds, nullptr, &tv);
#else
// Global-scope select: the entity namespace esphome::select shadows it here
int ret = ::select(fd + 1, nullptr, &writefds, nullptr, &tv);
#endif
if (ret < 0) {
err_out = errno;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
if (ret == 0 || !FD_ISSET(fd, &writefds)) {
return ConnectPollResult::CONNECT_POLL_PENDING;
}
int error = 0;
socklen_t len = sizeof(error);
if (sock.getsockopt(SOL_SOCKET, SO_ERROR, &error, &len) != 0) {
err_out = errno;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
if (error != 0) {
err_out = error;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
return ConnectPollResult::CONNECT_POLL_CONNECTED;
}
#endif
socklen_t set_sockaddr_any(struct sockaddr *addr, socklen_t addrlen, uint16_t port) {
#if USE_NETWORK_IPV6
if (addrlen < sizeof(sockaddr_in6)) {
+6
View File
@@ -145,6 +145,12 @@ inline socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const st
/// Set a sockaddr to the any address and specified port for the IP version used by socket_ip().
socklen_t set_sockaddr_any(struct sockaddr *addr, socklen_t addrlen, uint16_t port);
/// Check a non-blocking connect() for completion without blocking. On
/// CONNECT_POLL_ERROR, err_out holds the socket's SO_ERROR (or errno when the
/// poll itself failed) on fd based implementations, and the failure recorded
/// by the lwip callbacks on the raw lwip implementation.
ConnectPollResult poll_connect(Socket &sock, int &err_out);
/// Format sockaddr into caller-provided buffer, returns length written (excluding null)
size_t format_sockaddr_to(const struct sockaddr *addr_ptr, socklen_t len, std::span<char, SOCKADDR_STR_LEN> buf);
+15 -2
View File
@@ -13,9 +13,16 @@ void UDPComponent::setup() {
#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
for (const auto &address : this->addresses_) {
struct sockaddr saddr {};
socket::set_sockaddr(&saddr, sizeof(saddr), address, this->broadcast_port_);
if (socket::set_sockaddr(&saddr, sizeof(saddr), address, this->broadcast_port_) == 0) {
ESP_LOGW(TAG, "Invalid address %s", address);
continue;
}
this->sockaddrs_.push_back(saddr);
}
if (this->sockaddrs_.size() != this->addresses_.size()) {
// A dropped address silently receives nothing; surface the misconfiguration
this->status_set_warning(LOG_STR("invalid address"));
}
// set up broadcast socket
if (this->should_broadcast_) {
this->broadcast_socket_ = socket::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
@@ -94,9 +101,15 @@ void UDPComponent::setup() {
// 8266 and RP2040 `Duino
for (const auto &address : this->addresses_) {
auto ipaddr = IPAddress();
ipaddr.fromString(address);
if (!ipaddr.fromString(address)) {
ESP_LOGW(TAG, "Invalid address %s", address);
continue;
}
this->ipaddrs_.push_back(ipaddr);
}
if (this->ipaddrs_.size() != this->addresses_.size()) {
this->status_set_warning(LOG_STR("invalid address"));
}
if (this->should_listen_)
this->udp_client_.begin(this->listen_port_);
#endif
@@ -34,6 +34,10 @@ void WakeOnLanButton::press_action() {
struct sockaddr_storage saddr {};
auto addr_len =
socket::set_sockaddr(reinterpret_cast<sockaddr *>(&saddr), sizeof(saddr), "255.255.255.255", this->port_);
if (addr_len == 0) {
ESP_LOGW(TAG, "Invalid broadcast address");
return;
}
uint8_t buffer[6 + sizeof this->macaddr_ * 16];
memcpy(buffer, PREFIX, sizeof(PREFIX));
for (size_t i = 0; i != 16; i++) {
@@ -0,0 +1,4 @@
substitutions:
network_enable_ipv6: "true"
<<: !include common.yaml