[socket] Restore read_locked_, SO_RCVTIMEO, and wait_for_data_ lost in merge

These features from upstream/dev were dropped when resolving conflicts
with the PR's remote branch: read_locked_/wait_for_data_ (blocking read
with SO_RCVTIMEO timeout support), recv_timeout_cs_ field, SO_RCVTIMEO
and SO_SNDTIMEO setsockopt/getsockopt handling, and the setblocking()
implementation that accepts blocking mode for SO_RCVTIMEO.

esphome/components/socket/lwip_raw_tcp_impl.cpp

@@ -5,6 +5,7 @@
 
 #include <cerrno>
 #include <cstring>
+#include <sys/time.h>
 
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
@@ -422,6 +423,18 @@ int LWIPRawCommon::getsockopt(int level, int optname, void *optval, socklen_t *o
     *optlen = 4;
     return 0;
   }
+  if (level == SOL_SOCKET && optname == SO_RCVTIMEO) {
+    if (*optlen < sizeof(struct timeval)) {
+      errno = EINVAL;
+      return -1;
+    }
+    uint32_t ms = this->recv_timeout_cs_ * 10;
+    auto *tv = reinterpret_cast<struct timeval *>(optval);
+    tv->tv_sec = ms / 1000;
+    tv->tv_usec = (ms % 1000) * 1000;
+    *optlen = sizeof(struct timeval);
+    return 0;
+  }
   if (level == IPPROTO_TCP && optname == TCP_NODELAY) {
     if (*optlen < 4) {
       errno = EINVAL;
@@ -451,6 +464,21 @@ int LWIPRawCommon::setsockopt(int level, int optname, const void *optval, sockle
     // to prevent warnings
     return 0;
   }
+  if (level == SOL_SOCKET && optname == SO_RCVTIMEO) {
+    if (optlen < sizeof(struct timeval)) {
+      errno = EINVAL;
+      return -1;
+    }
+    const auto *tv = reinterpret_cast<const struct timeval *>(optval);
+    uint32_t ms = tv->tv_sec * 1000 + tv->tv_usec / 1000;
+    uint32_t cs = (ms + 9) / 10;  // round up to nearest centisecond
+    this->recv_timeout_cs_ = cs > 255 ? 255 : static_cast<uint8_t>(cs);
+    return 0;
+  }
+  if (level == SOL_SOCKET && optname == SO_SNDTIMEO) {
+    // Raw TCP writes are non-blocking (tcp_write), so send timeout is a no-op.
+    return 0;
+  }
   if (level == IPPROTO_TCP && optname == TCP_NODELAY) {
     if (optlen != 4) {
       errno = EINVAL;
@@ -546,8 +574,25 @@ err_t LWIPRawImpl::recv_fn(struct pbuf *pb, err_t err) {
   return ERR_OK;
 }
 
-ssize_t LWIPRawImpl::read(void *buf, size_t len) {
-  LWIP_LOCK();
+void LWIPRawImpl::wait_for_data_() {
+  // Wait for data without holding LWIP_LOCK so recv_fn() can run on RP2040
+  // (needs async_context lock).
+  //
+  // Loop until data arrives, connection closes, or the full timeout elapses.
+  // socket_delay() may return early due to other sockets waking the global
+  // socket_wake() flag, so we re-enter for the remaining time.
+  uint32_t timeout_ms = this->recv_timeout_cs_ * 10;
+  uint32_t start = millis();
+  while (this->waiting_for_data_()) {
+    uint32_t elapsed = millis() - start;
+    if (elapsed >= timeout_ms)
+      break;
+    socket_delay(timeout_ms - elapsed);
+  }
+}
+
+ssize_t LWIPRawImpl::read_locked_(void *buf, size_t len) {
+  // Caller must hold LWIP_LOCK. Copies available data from rx_buf_ into buf.
   if (this->pcb_ == nullptr) {
     errno = ECONNRESET;
     return -1;
@@ -606,11 +651,26 @@ ssize_t LWIPRawImpl::read(void *buf, size_t len) {
   return read;
 }
 
+ssize_t LWIPRawImpl::read(void *buf, size_t len) {
+  // See waiting_for_data_() for safety of unlocked reads.
+  if (this->recv_timeout_cs_ > 0 && this->waiting_for_data_()) {
+    this->wait_for_data_();
+  }
+
+  LWIP_LOCK();
+  return this->read_locked_(buf, len);
+}
+
 ssize_t LWIPRawImpl::readv(const struct iovec *iov, int iovcnt) {
+  // See waiting_for_data_() for safety of unlocked reads.
+  if (this->recv_timeout_cs_ > 0 && this->waiting_for_data_()) {
+    this->wait_for_data_();
+  }
+
   LWIP_LOCK();  // Hold for entire scatter-gather operation
   ssize_t ret = 0;
   for (int i = 0; i < iovcnt; i++) {
-    ssize_t err = this->read(reinterpret_cast<uint8_t *>(iov[i].iov_base), iov[i].iov_len);
+    ssize_t err = this->read_locked_(reinterpret_cast<uint8_t *>(iov[i].iov_base), iov[i].iov_len);
     if (err == -1) {
       if (ret != 0) {
         // if we already read some don't return an error

esphome/components/socket/lwip_raw_tcp_impl.h

@@ -58,6 +58,7 @@ class LWIPRawCommon {
   // instead use it for determining whether to call lwip_output
   bool nodelay_ = false;
   sa_family_t family_ = 0;
+  uint8_t recv_timeout_cs_ = 0;  // SO_RCVTIMEO in centiseconds (0 = no timeout, max 2.55s)
 };
 
 /// Connected socket implementation for LWIP raw TCP.
@@ -108,11 +109,8 @@ class LWIPRawImpl : public LWIPRawCommon {
       errno = ECONNRESET;
       return -1;
     }
-    if (blocking) {
-      // blocking operation not supported
-      errno = EINVAL;
-      return -1;
-    }
+    // Raw TCP doesn't use a blocking flag directly. Blocking behavior
+    // is provided by SO_RCVTIMEO which makes read() wait via socket_delay().
     return 0;
   }
   int loop() { return 0; }
@@ -123,6 +121,14 @@ class LWIPRawImpl : public LWIPRawCommon {
   static err_t s_recv_fn(void *arg, struct tcp_pcb *pcb, struct pbuf *pb, err_t err);
 
  protected:
+  // True when the socket could receive data but none has arrived yet.
+  // Safe to call without LWIP_LOCK — only null-checks pointers and reads a bool,
+  // all atomic on ARM/Xtensa. A stale value is harmless: the caller either does
+  // an unnecessary wait (stale true) or skips it (stale false), and the
+  // authoritative recheck happens under LWIP_LOCK afterward.
+  bool waiting_for_data_() const { return this->rx_buf_ == nullptr && !this->rx_closed_ && this->pcb_ != nullptr; }
+  void wait_for_data_();
+  ssize_t read_locked_(void *buf, size_t len);
   ssize_t internal_write_(const void *buf, size_t len);
   int internal_output_();