4#ifdef USE_SOCKET_IMPL_LWIP_TCP
27static volatile bool s_socket_woke =
false;
33 s_socket_woke =
false;
34 esp_delay(ms, []() {
return !s_socket_woke; });
43static const char *
const TAG =
"socket.lwip";
47#define LWIP_LOG(msg, ...) ESP_LOGVV(TAG, "socket %p: " msg, this, ##__VA_ARGS__)
49#define LWIP_LOG(msg, ...)
52class LWIPRawImpl :
public Socket {
54 LWIPRawImpl(
sa_family_t family,
struct tcp_pcb *pcb) : pcb_(pcb), family_(family) {}
55 ~LWIPRawImpl()
override {
56 if (pcb_ !=
nullptr) {
57 LWIP_LOG(
"tcp_abort(%p)", pcb_);
64 LWIP_LOG(
"init(%p)", pcb_);
66 tcp_recv(pcb_, LWIPRawImpl::s_recv_fn);
67 tcp_err(pcb_, LWIPRawImpl::s_err_fn);
70 std::unique_ptr<Socket> accept(
struct sockaddr *addr,
socklen_t *addrlen)
override {
76 if (pcb_ ==
nullptr) {
80 if (name ==
nullptr) {
87 if (family_ == AF_INET) {
92 auto *addr4 =
reinterpret_cast<const sockaddr_in *
>(name);
93 port = ntohs(addr4->sin_port);
94 ip.type = IPADDR_TYPE_V4;
95 ip.u_addr.ip4.addr = addr4->sin_addr.s_addr;
96 LWIP_LOG(
"tcp_bind(%p ip=%s port=%u)", pcb_, ip4addr_ntoa(&ip.u_addr.ip4), port);
97 }
else if (family_ == AF_INET6) {
102 auto *addr6 =
reinterpret_cast<const sockaddr_in6 *
>(name);
103 port = ntohs(addr6->sin6_port);
104 ip.type = IPADDR_TYPE_ANY;
105 memcpy(&ip.u_addr.ip6.addr, &addr6->sin6_addr.un.u8_addr, 16);
106 LWIP_LOG(
"tcp_bind(%p ip=%s port=%u)", pcb_, ip6addr_ntoa(&ip.u_addr.ip6), port);
112 if (family_ != AF_INET) {
116 auto *addr4 =
reinterpret_cast<const sockaddr_in *
>(name);
117 port = ntohs(addr4->sin_port);
118 ip.addr = addr4->sin_addr.s_addr;
119 LWIP_LOG(
"tcp_bind(%p ip=%u port=%u)", pcb_, ip.addr, port);
121 err_t err = tcp_bind(pcb_, &ip, port);
122 if (err == ERR_USE) {
123 LWIP_LOG(
" -> err ERR_USE");
127 if (err == ERR_VAL) {
128 LWIP_LOG(
" -> err ERR_VAL");
133 LWIP_LOG(
" -> err %d", err);
139 int close()
override {
140 if (pcb_ ==
nullptr) {
144 LWIP_LOG(
"tcp_close(%p)", pcb_);
145 err_t err = tcp_close(pcb_);
147 LWIP_LOG(
" -> err %d", err);
150 errno = err == ERR_MEM ? ENOMEM : EIO;
156 int shutdown(
int how)
override {
157 if (pcb_ ==
nullptr) {
161 bool shut_rx =
false, shut_tx =
false;
162 if (how == SHUT_RD) {
164 }
else if (how == SHUT_WR) {
166 }
else if (how == SHUT_RDWR) {
167 shut_rx = shut_tx =
true;
172 LWIP_LOG(
"tcp_shutdown(%p shut_rx=%d shut_tx=%d)", pcb_, shut_rx ? 1 : 0, shut_tx ? 1 : 0);
173 err_t err = tcp_shutdown(pcb_, shut_rx, shut_tx);
175 LWIP_LOG(
" -> err %d", err);
176 errno = err == ERR_MEM ? ENOMEM : EIO;
183 if (pcb_ ==
nullptr) {
187 if (name ==
nullptr || addrlen ==
nullptr) {
191 return this->ip2sockaddr_(&pcb_->remote_ip, pcb_->remote_port, name, addrlen);
193 std::string getpeername()
override {
194 if (pcb_ ==
nullptr) {
198 return this->format_ip_address_(pcb_->remote_ip);
201 if (pcb_ ==
nullptr) {
205 if (name ==
nullptr || addrlen ==
nullptr) {
209 return this->ip2sockaddr_(&pcb_->local_ip, pcb_->local_port, name, addrlen);
211 std::string getsockname()
override {
212 if (pcb_ ==
nullptr) {
216 return this->format_ip_address_(pcb_->local_ip);
218 int getsockopt(
int level,
int optname,
void *optval,
socklen_t *optlen)
override {
219 if (pcb_ ==
nullptr) {
223 if (optlen ==
nullptr || optval ==
nullptr) {
227 if (level == SOL_SOCKET && optname == SO_REUSEADDR) {
235 *
reinterpret_cast<int *
>(optval) = 1;
239 if (level == IPPROTO_TCP && optname == TCP_NODELAY) {
244 *
reinterpret_cast<int *
>(optval) = nodelay_;
252 int setsockopt(
int level,
int optname,
const void *optval,
socklen_t optlen)
override {
253 if (pcb_ ==
nullptr) {
257 if (level == SOL_SOCKET && optname == SO_REUSEADDR) {
267 if (level == IPPROTO_TCP && optname == TCP_NODELAY) {
272 int val = *
reinterpret_cast<const int *
>(optval);
280 int listen(
int backlog)
override {
286 ssize_t read(
void *buf,
size_t len)
override {
287 if (pcb_ ==
nullptr) {
291 if (rx_closed_ && rx_buf_ ==
nullptr) {
297 if (rx_buf_ ==
nullptr) {
303 uint8_t *buf8 =
reinterpret_cast<uint8_t *
>(buf);
304 while (
len && rx_buf_ !=
nullptr) {
305 size_t pb_len = rx_buf_->len;
306 size_t pb_left = pb_len - rx_buf_offset_;
309 size_t copysize = std::min(
len, pb_left);
310 memcpy(buf8,
reinterpret_cast<uint8_t *
>(rx_buf_->payload) + rx_buf_offset_, copysize);
312 if (pb_left == copysize) {
314 if (rx_buf_->next ==
nullptr) {
320 auto *old_buf = rx_buf_;
321 rx_buf_ = rx_buf_->next;
327 rx_buf_offset_ += copysize;
329 LWIP_LOG(
"tcp_recved(%p %u)", pcb_, copysize);
330 tcp_recved(pcb_, copysize);
344 ssize_t readv(
const struct iovec *iov,
int iovcnt)
override {
346 for (
int i = 0; i < iovcnt; i++) {
347 ssize_t err = read(
reinterpret_cast<uint8_t *
>(iov[i].iov_base), iov[i].iov_len);
356 if ((
size_t) err != iov[i].iov_len)
367 ssize_t internal_write(
const void *buf,
size_t len) {
368 if (pcb_ ==
nullptr) {
374 if (buf ==
nullptr) {
378 auto space = tcp_sndbuf(pcb_);
383 size_t to_send = std::min((
size_t) space,
len);
384 LWIP_LOG(
"tcp_write(%p buf=%p %u)", pcb_, buf, to_send);
385 err_t err = tcp_write(pcb_, buf, to_send, TCP_WRITE_FLAG_COPY);
386 if (err == ERR_MEM) {
387 LWIP_LOG(
" -> err ERR_MEM");
392 LWIP_LOG(
" -> err %d", err);
398 int internal_output() {
399 LWIP_LOG(
"tcp_output(%p)", pcb_);
400 err_t err = tcp_output(pcb_);
401 if (err == ERR_ABRT) {
402 LWIP_LOG(
" -> err ERR_ABRT");
410 LWIP_LOG(
" -> err %d", err);
416 ssize_t write(
const void *buf,
size_t len)
override {
425 int err = internal_output();
431 ssize_t writev(
const struct iovec *iov,
int iovcnt)
override {
433 for (
int i = 0; i < iovcnt; i++) {
434 ssize_t err = internal_write(
reinterpret_cast<uint8_t *
>(iov[i].iov_base), iov[i].iov_len);
443 if ((
size_t) err != iov[i].iov_len)
451 int err = internal_output();
462 int setblocking(
bool blocking)
override {
463 if (pcb_ ==
nullptr) {
475 void err_fn(err_t err) {
476 LWIP_LOG(
"err(err=%d)", err);
484 err_t recv_fn(
struct pbuf *pb, err_t err) {
485 LWIP_LOG(
"recv(pb=%p err=%d)", pb, err);
496 if (rx_buf_ ==
nullptr) {
501 pbuf_cat(rx_buf_, pb);
510 static void s_err_fn(
void *arg, err_t err) {
511 LWIPRawImpl *arg_this =
reinterpret_cast<LWIPRawImpl *
>(arg);
512 arg_this->err_fn(err);
515 static err_t s_recv_fn(
void *arg,
struct tcp_pcb *pcb,
struct pbuf *pb, err_t err) {
516 LWIPRawImpl *arg_this =
reinterpret_cast<LWIPRawImpl *
>(arg);
517 return arg_this->recv_fn(pb, err);
521 std::string format_ip_address_(
const ip_addr_t &ip) {
522 char buffer[50] = {};
523 if (IP_IS_V4_VAL(ip)) {
524 inet_ntoa_r(ip, buffer,
sizeof(buffer));
527 else if (IP_IS_V6_VAL(ip)) {
528 inet6_ntoa_r(ip, buffer,
sizeof(buffer));
531 return std::string(buffer);
535 if (family_ == AF_INET) {
545 inet_addr_from_ip4addr(&addr->
sin_addr, ip_2_ip4(ip));
549 else if (family_ == AF_INET6) {
563 ip4_2_ipv4_mapped_ipv6(ip_2_ip6(&mapped), ip_2_ip4(ip));
564 inet6_addr_from_ip6addr(&addr->
sin6_addr, ip_2_ip6(&mapped));
566 inet6_addr_from_ip6addr(&addr->
sin6_addr, ip_2_ip6(ip));
576 struct tcp_pcb *pcb_;
577 pbuf *rx_buf_ =
nullptr;
578 size_t rx_buf_offset_ = 0;
579 bool rx_closed_ =
false;
582 bool nodelay_ =
false;
588class LWIPRawListenImpl :
public LWIPRawImpl {
590 LWIPRawListenImpl(
sa_family_t family,
struct tcp_pcb *pcb) : LWIPRawImpl(family, pcb) {}
593 LWIP_LOG(
"init(%p)", pcb_);
595 tcp_accept(pcb_, LWIPRawListenImpl::s_accept_fn);
596 tcp_err(pcb_, LWIPRawImpl::s_err_fn);
599 std::unique_ptr<Socket> accept(
struct sockaddr *addr,
socklen_t *addrlen)
override {
600 if (pcb_ ==
nullptr) {
604 if (accepted_socket_count_ == 0) {
609 std::unique_ptr<LWIPRawImpl> sock = std::move(accepted_sockets_[0]);
611 for (uint8_t i = 1; i < accepted_socket_count_; i++) {
612 accepted_sockets_[i - 1] = std::move(accepted_sockets_[i]);
614 accepted_socket_count_--;
615 LWIP_LOG(
"Connection accepted by application, queue size: %d", accepted_socket_count_);
616 if (addr !=
nullptr) {
617 sock->getpeername(addr, addrlen);
619 LWIP_LOG(
"accept(%p)", sock.get());
620 return std::unique_ptr<Socket>(std::move(sock));
623 int listen(
int backlog)
override {
624 if (pcb_ ==
nullptr) {
628 LWIP_LOG(
"tcp_listen_with_backlog(%p backlog=%d)", pcb_, backlog);
629 struct tcp_pcb *listen_pcb = tcp_listen_with_backlog(pcb_, backlog);
630 if (listen_pcb ==
nullptr) {
639 LWIP_LOG(
"tcp_arg(%p)", pcb_);
641 tcp_accept(pcb_, LWIPRawListenImpl::s_accept_fn);
646 err_t accept_fn_(
struct tcp_pcb *newpcb, err_t err) {
647 LWIP_LOG(
"accept(newpcb=%p err=%d)", newpcb, err);
648 if (err != ERR_OK || newpcb ==
nullptr) {
656 if (accepted_socket_count_ >= MAX_ACCEPTED_SOCKETS) {
657 LWIP_LOG(
"Rejecting connection, queue full (%d)", accepted_socket_count_);
663 auto sock = make_unique<LWIPRawImpl>(family_, newpcb);
665 accepted_sockets_[accepted_socket_count_++] = std::move(sock);
666 LWIP_LOG(
"Accepted connection, queue size: %d", accepted_socket_count_);
674 static err_t s_accept_fn(
void *arg,
struct tcp_pcb *newpcb, err_t err) {
675 LWIPRawListenImpl *arg_this =
reinterpret_cast<LWIPRawListenImpl *
>(arg);
676 return arg_this->accept_fn_(newpcb, err);
693 static constexpr size_t MAX_ACCEPTED_SOCKETS = 3;
694 std::array<std::unique_ptr<LWIPRawImpl>, MAX_ACCEPTED_SOCKETS> accepted_sockets_;
695 uint8_t accepted_socket_count_ = 0;
698std::unique_ptr<Socket>
socket(
int domain,
int type,
int protocol) {
699 auto *pcb = tcp_new();
704 auto *sock =
new LWIPRawListenImpl((
sa_family_t) domain, pcb);
706 return std::unique_ptr<Socket>{sock};
void socket_wake()
Called by lwip callbacks to signal socket activity and wake delay.
std::unique_ptr< Socket > socket(int domain, int type, int protocol)
Create a socket of the given domain, type and protocol.
std::unique_ptr< Socket > socket_loop_monitored(int domain, int type, int protocol)
Create a socket and monitor it for data in the main loop.
void socket_delay(uint32_t ms)
Delay that can be woken early by socket activity.
Providing packet encoding functions for exchanging data with a remote host.
struct in6_addr sin6_addr