From d238fb589f0d206aa9b81751ee88b2cd283ad0a2 Mon Sep 17 00:00:00 2001 From: Lasan Mahaliyana Date: Fri, 6 Feb 2026 23:29:32 +0530 Subject: [PATCH] rewrote l2cap server socket to fix deadlocks --- .../linux/ble_l2cap_server_socket.cc | 349 +++++++++++------- .../linux/ble_l2cap_server_socket.h | 28 +- 2 files changed, 230 insertions(+), 147 deletions(-) diff --git a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc index 3df1e631..97c7897b 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc @@ -14,188 +14,224 @@ #include "internal/platform/implementation/linux/ble_l2cap_server_socket.h" -#include #include -#include -#include -#include +#include #include +#include #include +#include #include +#include + +#include +#include #include "internal/platform/logging.h" -#include "internal/platform/prng.h" namespace nearby { namespace linux { +namespace { bool SetNonBlocking(int fd) { int flags = fcntl(fd, F_GETFL, 0); if (flags < 0) return false; return fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0; } + void DrainFd(int fd) { char buf[64]; - for (;;) { - ssize_t n = read(fd, buf, sizeof(buf)); - if (n > 0) continue; - if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) break; + while (true) { + ssize_t read_count = read(fd, buf, sizeof(buf)); + if (read_count > 0) continue; + if (read_count < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) break; break; } } -BleL2capServerSocket::BleL2capServerSocket() : psm_(0) {} -BleL2capServerSocket::BleL2capServerSocket(int psm) : psm_(psm) { -} +} // namespace + +BleL2capServerSocket::BleL2capServerSocket() = default; + +BleL2capServerSocket::BleL2capServerSocket(int psm) : psm_(psm) {} BleL2capServerSocket::~BleL2capServerSocket() { Close(); } -void BleL2capServerSocket::SetPSM(int psm) { psm_ = psm; } +void BleL2capServerSocket::SetPSM(int psm) { + absl::MutexLock lock(&mutex_); + psm_ = psm; +} -void BleL2capServerSocket::AcceptPoll(int& client_fd, sockaddr_l2& client_addr, socklen_t& client_len) { - for (;;) { - struct pollfd fds[2]; - fds[0].fd = server_fd_; +bool BleL2capServerSocket::InitializeServerSocketLocked() { + if (closed_) { + errno = EINTR; + return false; + } + + if (server_fd_ >= 0) { + return true; + } + + if ((stop_pipe_[0] == -1) != (stop_pipe_[1] == -1)) { + if (stop_pipe_[0] != -1) close(stop_pipe_[0]); + if (stop_pipe_[1] != -1) close(stop_pipe_[1]); + stop_pipe_[0] = -1; + stop_pipe_[1] = -1; + } + + if (stop_pipe_[0] == -1) { + if (pipe(stop_pipe_) < 0) { + LOG(ERROR) << "Failed to create stop pipe: " << std::strerror(errno); + return false; + } + if (!SetNonBlocking(stop_pipe_[0]) || !SetNonBlocking(stop_pipe_[1])) { + LOG(ERROR) << "Failed to set non-blocking mode on stop pipe: " + << std::strerror(errno); + close(stop_pipe_[0]); + close(stop_pipe_[1]); + stop_pipe_[0] = -1; + stop_pipe_[1] = -1; + return false; + } + } + + server_fd_ = socket(AF_BLUETOOTH, SOCK_SEQPACKET, BTPROTO_L2CAP); + if (server_fd_ < 0) { + LOG(ERROR) << "Failed to create L2CAP server socket: " + << std::strerror(errno); + return false; + } + + sockaddr_l2 addr; + std::memset(&addr, 0, sizeof(addr)); + addr.l2_family = AF_BLUETOOTH; + addr.l2_psm = htobs(psm_); + addr.l2_bdaddr_type = BDADDR_LE_PUBLIC; + std::memset(&addr.l2_bdaddr, 0, sizeof(addr.l2_bdaddr)); + + if (bind(server_fd_, reinterpret_cast(&addr), sizeof(addr)) < 0) { + LOG(ERROR) << "Failed to bind L2CAP server socket: " << std::strerror(errno) + << " (errno: " << errno << ")"; + close(server_fd_); + server_fd_ = -1; + return false; + } + + constexpr uint16_t kReceiveMtu = 672; + if (setsockopt(server_fd_, SOL_BLUETOOTH, BT_RCVMTU, &kReceiveMtu, + sizeof(kReceiveMtu)) < 0) { + LOG(WARNING) << "Failed to set receive MTU on L2CAP server socket: " + << std::strerror(errno); + } + + if (listen(server_fd_, 5) < 0) { + LOG(ERROR) << "Failed to listen on L2CAP server socket: " + << std::strerror(errno); + close(server_fd_); + server_fd_ = -1; + return false; + } + + if (!SetNonBlocking(server_fd_)) { + LOG(ERROR) << "Failed to set non-blocking on L2CAP server socket: " + << std::strerror(errno); + close(server_fd_); + server_fd_ = -1; + return false; + } + + socklen_t addr_len = sizeof(addr); + if (getsockname(server_fd_, reinterpret_cast(&addr), &addr_len) == + 0) { + psm_ = btohs(addr.l2_psm); + LOG(INFO) << "L2CAP server socket listening on PSM: " << psm_; + } else { + LOG(WARNING) << "Failed to get socket name: " << std::strerror(errno); + } + + return true; +} + +void BleL2capServerSocket::AcceptPoll(int server_fd, int stop_fd, int& client_fd, + sockaddr_l2& client_addr, + socklen_t& client_len) { + while (true) { + pollfd fds[2]; + fds[0].fd = server_fd; fds[0].events = POLLIN; fds[0].revents = 0; - - fds[1].fd = stop_pipe_[0]; + fds[1].fd = stop_fd; fds[1].events = POLLIN; fds[1].revents = 0; - int r = poll(fds, 2, -1); // wait forever; stop pipe will wake us - if (r < 0) { + int result = poll(fds, 2, -1); + if (result < 0) { if (errno == EINTR) continue; LOG(ERROR) << "poll() failed: " << std::strerror(errno); + client_fd = -1; return; } - // Stop requested - if (fds[1].revents & POLLIN) { - absl::MutexLock l(&mutex_); - DrainFd(stop_pipe_[0]); + if (fds[1].revents & (POLLIN | POLLERR | POLLHUP | POLLNVAL)) { + if (fds[1].revents & POLLIN) { + DrainFd(stop_fd); + } client_fd = -1; - errno = EINTR; // optional: helps caller treat as "interrupted" - return; // <-- THIS is the key + errno = EINTR; + return; } - // Listen socket error/hangup if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) { LOG(ERROR) << "poll() listen fd error revents=" << fds[0].revents; + client_fd = -1; + errno = EIO; + return; } - // Incoming connection(s) - if (fds[0].revents & POLLIN) { - for (;;) { - client_fd = accept(server_fd_, (struct sockaddr*)&client_addr, &client_len); - if (client_fd >= 0) break; + if ((fds[0].revents & POLLIN) == 0) continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) { - // No more queued connections; go back to poll(). - client_fd = -1; - break; - } - if (errno == EINTR) continue; - - LOG(ERROR) << "Failed to accept L2CAP connection: " << std::strerror(errno); + while (true) { + client_fd = accept(server_fd, reinterpret_cast(&client_addr), + &client_len); + if (client_fd >= 0) { return; } - if (client_fd >= 0) break; + if (errno == EINTR) continue; + if (errno == EAGAIN || errno == EWOULDBLOCK) { + client_fd = -1; + break; + } + + LOG(ERROR) << "Failed to accept L2CAP connection: " << std::strerror(errno); + return; } } } + std::unique_ptr BleL2capServerSocket::Accept() { - // blocking accept until the previous server_fd_ is closed. - // We need to keep accepting connections on the same psm. That's why we do this - absl::MutexLock lock(&mutex_); - auto idle = [this] {return server_fd_ == -1;}; - mutex_.Await(absl::Condition(&idle)); - - server_fd_ = socket(AF_BLUETOOTH, SOCK_SEQPACKET, BTPROTO_L2CAP); - if (server_fd_ < 0) { - LOG(ERROR) << "Failed to create L2CAP server socket: " - << std::strerror(errno); - return nullptr; - } - // Create stop pipe once (used to wake poll() from another thread). - if (stop_pipe_[0] == -1 && stop_pipe_[1] == -1) { - if (pipe(stop_pipe_) < 0) { - LOG(ERROR) << "Failed to create stop pipe: " << std::strerror(errno); - close(server_fd_); - server_fd_ = -1; - return nullptr; - } - // Optional, but avoids any chance of blocking on drain/write. - (void)SetNonBlocking(stop_pipe_[0]); - (void)SetNonBlocking(stop_pipe_[1]); - } - - struct sockaddr_l2 addr; - std::memset(&addr, 0, sizeof(addr)); - addr.l2_family = AF_BLUETOOTH; - addr.l2_psm = htobs(psm_); - addr.l2_bdaddr_type = BDADDR_LE_PUBLIC; - // Set BDADDR_ANY (all zeros) - std::memset(&addr.l2_bdaddr, 0, sizeof(addr.l2_bdaddr)); - - if (bind(server_fd_, (struct sockaddr*)&addr, sizeof(addr)) < 0) { - LOG(ERROR) << "Failed to bind L2CAP server socket: " - << std::strerror(errno) << " (errno: " << errno << ")"; - close(server_fd_); - server_fd_ = -1; - return nullptr; - } - - struct l2cap_options opts; - opts.omtu = 0; - opts.imtu = 672; - - if (setsockopt(server_fd_, SOL_BLUETOOTH, BT_RCVMTU, &opts.imtu, - sizeof(opts.imtu)) < 0) { - LOG(ERROR) << "Failed to set socket options on L2CAP server socket" ; - } - - if (listen(server_fd_, 5) < 0) { - LOG(ERROR) << "Failed to listen on L2CAP server socket: " - << std::strerror(errno); - close(server_fd_); - server_fd_ = -1; - return nullptr; - } - - // Make accept() non-blocking; we will block in poll() instead. - if (!SetNonBlocking(server_fd_)) { - LOG(ERROR) << "Failed to set non-blocking on L2CAP server socket: " << std::strerror(errno); - close(server_fd_); - server_fd_ = -1; - return nullptr; - } - socklen_t addr_len = sizeof(addr); - if (getsockname(server_fd_, (struct sockaddr*)&addr, &addr_len) == 0) { - psm_ = btohs(addr.l2_psm); - LOG(INFO) << "L2CAP server socket listening on PSM: " << psm_; - } else { - LOG(ERROR) << "Failed to get socket name: " << std::strerror(errno); - } - - if (server_fd_ < 0) { - LOG(ERROR) << "Server socket not initialized"; - return nullptr; + int server_fd = -1; + int stop_fd = -1; + int listening_psm = 0; + { + absl::MutexLock lock(&mutex_); + if (!InitializeServerSocketLocked()) return nullptr; + server_fd = server_fd_; + stop_fd = stop_pipe_[0]; + listening_psm = psm_; } - struct sockaddr_l2 client_addr; - socklen_t client_len = sizeof(client_addr); + sockaddr_l2 client_addr; std::memset(&client_addr, 0, sizeof(client_addr)); + socklen_t client_len = sizeof(client_addr); - LOG(INFO) << "Waiting for L2CAP connection on PSM " << psm_ << "..."; + LOG(INFO) << "Waiting for L2CAP connection on PSM " << listening_psm << "..."; int client_fd = -1; - AcceptPoll(client_fd, client_addr, client_len); + AcceptPoll(server_fd, stop_fd, client_fd, client_addr, client_len); if (client_fd < 0) { if (errno == EINTR || errno == EAGAIN) { - LOG(WARNING) << "Accept interrupted, returning nullptr"; + LOG(INFO) << "Accept interrupted, returning nullptr"; return nullptr; } LOG(ERROR) << "Failed to accept L2CAP connection: " << std::strerror(errno); @@ -207,25 +243,72 @@ std::unique_ptr BleL2capServerSocket::Accept() { LOG(INFO) << "Accepted L2CAP connection from " << client_addr_str << " on PSM " << btohs(client_addr.l2_psm); - LOG(INFO) << __func__ << ": Connected to client_fd: " << client_fd; - // Create a unique ID from the MAC address api::ble::BlePeripheral::UniqueId peripheral_id = 0; - for (int i = 0; i < 6; i++) { - peripheral_id = (peripheral_id << 8) | client_addr.l2_bdaddr.b[i]; + for (int i = 0; i < 6; ++i) { + peripheral_id = + (peripheral_id << 8) | + static_cast(client_addr.l2_bdaddr.b[i]); } - accepted_fds_.emplace(server_fd_, std::pair(client_fd, peripheral_id)); return std::make_unique(client_fd, peripheral_id); } -Exception BleL2capServerSocket::Close() { - LOG(ERROR) << __func__ << ": closing l2cap server socket"; - if (stop_pipe_[1] != -1) (void)write(stop_pipe_[1], "x", 1); - LOG(ERROR) << __func__ << ": l2cap server socket closed"; +Exception BleL2capServerSocket::Close() { + absl::AnyInvocable notifier; + int server_fd = -1; + int stop_read_fd = -1; + int stop_write_fd = -1; + { + absl::MutexLock lock(&mutex_); + if (closed_) { + return {Exception::kSuccess}; + } + closed_ = true; + notifier = std::move(close_notifier_); + server_fd = std::exchange(server_fd_, -1); + stop_read_fd = std::exchange(stop_pipe_[0], -1); + stop_write_fd = std::exchange(stop_pipe_[1], -1); + } + + if (stop_write_fd != -1) { + char wake = 'x'; + ssize_t ignored = write(stop_write_fd, &wake, 1); + (void)ignored; + } + + if (server_fd != -1 && close(server_fd) != 0) { + LOG(WARNING) << "Failed to close L2CAP server socket: " << std::strerror(errno); + } + if (stop_read_fd != -1 && close(stop_read_fd) != 0) { + LOG(WARNING) << "Failed to close stop pipe read fd: " << std::strerror(errno); + } + if (stop_write_fd != -1 && close(stop_write_fd) != 0) { + LOG(WARNING) << "Failed to close stop pipe write fd: " + << std::strerror(errno); + } + + if (notifier) { + notifier(); + } + return {Exception::kSuccess}; } - +void BleL2capServerSocket::SetCloseNotifier( + absl::AnyInvocable notifier) { + absl::AnyInvocable notifier_to_run; + { + absl::MutexLock lock(&mutex_); + if (!closed_) { + close_notifier_ = std::move(notifier); + return; + } + notifier_to_run = std::move(notifier); + } + if (notifier_to_run) { + notifier_to_run(); + } +} } // namespace linux } // namespace nearby diff --git a/internal/platform/implementation/linux/ble_l2cap_server_socket.h b/internal/platform/implementation/linux/ble_l2cap_server_socket.h index bee205ce..c4772097 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.h +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.h @@ -16,15 +16,12 @@ #define PLATFORM_IMPL_LINUX_BLE_L2CAP_SERVER_SOCKET_H_ #include -#include #include "absl/functional/any_invocable.h" #include "absl/synchronization/mutex.h" #include "internal/platform/exception.h" #include "internal/platform/implementation/ble.h" -#include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/linux/ble_l2cap_socket.h" -#include "absl/container/flat_hash_map.h" #include #include @@ -39,22 +36,25 @@ class BleL2capServerSocket final : public api::ble::BleL2capServerSocket { int GetPSM() const override { return psm_; } void SetPSM(int psm); - void AcceptPoll(int &client_fd, sockaddr_l2 &client_addr, - socklen_t &client_len); - std::unique_ptr Accept() override; - Exception Close() override ; + std::unique_ptr Accept() override + ABSL_LOCKS_EXCLUDED(mutex_); + Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); - void SetCloseNotifier(absl::AnyInvocable notifier); + void SetCloseNotifier(absl::AnyInvocable notifier) + ABSL_LOCKS_EXCLUDED(mutex_); private: - absl::Mutex mutex_; - int psm_ = 0; - int server_fd_ = -1; - int stop_pipe_[2] = {-1, -1}; // read end [0], write end [1] + bool InitializeServerSocketLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + void AcceptPoll(int server_fd, int stop_fd, int& client_fd, + sockaddr_l2& client_addr, socklen_t& client_len); - // > - absl::flat_hash_map> accepted_fds_; + absl::Mutex mutex_; + bool closed_ ABSL_GUARDED_BY(mutex_) = false; + absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); + int psm_ = 0; + int server_fd_ ABSL_GUARDED_BY(mutex_) = -1; + int stop_pipe_[2] ABSL_GUARDED_BY(mutex_) = {-1, -1}; // read [0], write [1] }; } // namespace linux