diff --git a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc index 14aa15d7..3c03ff3d 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.cc +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.cc @@ -14,10 +14,12 @@ #include "internal/platform/implementation/linux/ble_l2cap_server_socket.h" +#include +#include +#include #include #include #include -#include #include #include @@ -28,6 +30,20 @@ namespace nearby { namespace linux { +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; + break; + } +} BleL2capServerSocket::BleL2capServerSocket() : psm_(0) {} BleL2capServerSocket::BleL2capServerSocket(int psm) : psm_(psm) { @@ -37,91 +53,141 @@ BleL2capServerSocket::~BleL2capServerSocket() { Close(); } void BleL2capServerSocket::SetPSM(int psm) { 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_; + fds[0].events = POLLIN; + fds[0].revents = 0; + + fds[1].fd = stop_pipe_[0]; + fds[1].events = POLLIN; + fds[1].revents = 0; + + int r = poll(fds, 2, -1); // wait forever; stop pipe will wake us + if (r < 0) { + if (errno == EINTR) continue; + LOG(ERROR) << "poll() failed: " << std::strerror(errno); + return; + } + + // Stop requested +if (fds[1].revents & POLLIN) { + DrainFd(stop_pipe_[0]); + client_fd = -1; + errno = EINTR; // optional: helps caller treat as "interrupted" + return; // <-- THIS is the key +} + + // Listen socket error/hangup + if (fds[0].revents & (POLLERR | POLLHUP | POLLNVAL)) { + LOG(ERROR) << "poll() listen fd error revents=" << fds[0].revents; + } + + // 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 (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); + return; + } + + if (client_fd >= 0) break; + } + } +} std::unique_ptr BleL2capServerSocket::Accept() { - if (stopped_.Cancelled()) { - LOG(ERROR) << __func__ << ": server socket has been stopped"; - return nullptr; + Prng prng; + psm_ = 0x80 + (prng.NextUint32() % 0x80); + + 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; } - absl::MutexLock lock(&mutex_); - - Prng prng; - psm_ = 0x80 + (prng.NextUint32() % 0x80); - - 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; - } - - - LOG(INFO) << __func__ << ": Using server_fd: " << server_fd_; - - 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; - } - - 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); - } - // // Return cached socket if it exists - // auto it = accepted_fds_.find(server_fd_); - // if (it != accepted_fds_.end()) { - // LOG(INFO) << __func__ << ": Socket exists. Returning cached socket"; - // return std::make_unique(it->second.first, it->second.second); // I hate how this looks - // } - - if (server_fd_ < 0) { - LOG(ERROR) << "Server socket not initialized"; - return nullptr; - } - - // Release the mutex while waiting for incoming connection - mutex_.Unlock(); - struct sockaddr_l2 client_addr; socklen_t client_len = sizeof(client_addr); std::memset(&client_addr, 0, sizeof(client_addr)); LOG(INFO) << "Waiting for L2CAP connection on PSM " << psm_ << "..."; - int client_fd = accept(server_fd_, (struct sockaddr*)&client_addr, &client_len); - - // Re-acquire the mutex - mutex_.Lock(); + int client_fd = -1; + AcceptPoll(client_fd, client_addr, client_len); if (client_fd < 0) { if (errno == EINTR || errno == EAGAIN) { @@ -132,14 +198,9 @@ std::unique_ptr BleL2capServerSocket::Accept() { return nullptr; } - if (closed_) { - close(client_fd); - return nullptr; - } - char client_addr_str[18]; ba2str(&client_addr.l2_bdaddr, client_addr_str); - LOG(INFO) << "Accepted L2CAP connection from " << client_addr_str + 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; @@ -152,38 +213,14 @@ std::unique_ptr BleL2capServerSocket::Accept() { 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 bluetooth server socket"; - stopped_.Cancel(); + LOG(ERROR) << __func__ << ": closing l2cap server socket"; - return DoClose(); -} - -Exception BleL2capServerSocket::DoClose() { - closed_ = true; - - if (server_fd_ >= 0) { - shutdown(server_fd_, SHUT_RDWR); - close(server_fd_); - server_fd_ = -1; - } - - if (close_notifier_) { - auto notifier = std::move(close_notifier_); - mutex_.Unlock(); - notifier(); - mutex_.Lock(); - } - + if (stop_pipe_[1] != -1) (void)write(stop_pipe_[1], "x", 1); + LOG(ERROR) << __func__ << ": l2cap server socket closed"; return {Exception::kSuccess}; } -void BleL2capServerSocket::SetCloseNotifier( - absl::AnyInvocable notifier) { - absl::MutexLock lock(&mutex_); - close_notifier_ = std::move(notifier); -} } // 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 a3f1e0dc..e77203c5 100644 --- a/internal/platform/implementation/linux/ble_l2cap_server_socket.h +++ b/internal/platform/implementation/linux/ble_l2cap_server_socket.h @@ -25,6 +25,9 @@ #include "internal/platform/implementation/ble.h" #include "internal/platform/implementation/linux/ble_l2cap_socket.h" #include "absl/container/flat_hash_map.h" + +#include +#include namespace nearby { namespace linux { @@ -36,27 +39,23 @@ 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 - ABSL_LOCKS_EXCLUDED(mutex_); - Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_); + std::unique_ptr Accept() override; + Exception Close() override ; - void SetCloseNotifier(absl::AnyInvocable notifier) - ABSL_LOCKS_EXCLUDED(mutex_); + void SetCloseNotifier(absl::AnyInvocable notifier); private: - Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - mutable absl::Mutex mutex_; absl::CondVar cond_; - absl::AnyInvocable close_notifier_ ABSL_GUARDED_BY(mutex_); - bool closed_ ABSL_GUARDED_BY(mutex_) = false; int psm_ = 0; - int server_fd_ ABSL_GUARDED_BY(mutex_) = -1; + int server_fd_ = -1; + int stop_pipe_[2] = {-1, -1}; // read end [0], write end [1] - CancellationFlag stopped_; // > - absl::flat_hash_map> accepted_fds_ ABSL_GUARDED_BY(mutex_); + absl::flat_hash_map> accepted_fds_; }; } // namespace linux diff --git a/internal/platform/implementation/linux/ble_l2cap_socket.cc b/internal/platform/implementation/linux/ble_l2cap_socket.cc index 7c605233..205a8800 100644 --- a/internal/platform/implementation/linux/ble_l2cap_socket.cc +++ b/internal/platform/implementation/linux/ble_l2cap_socket.cc @@ -124,8 +124,6 @@ ExceptionOr BleL2capInputStream::Read(std::int64_t size) { if (size <= 0) { return ExceptionOr(ByteArray{}); } - LOG(INFO) << "Reading"; - auto poller = Poller::CreateOutputPoller(fd); while (true) { @@ -144,12 +142,10 @@ ExceptionOr BleL2capInputStream::Read(std::int64_t size) { if (fd_.load() != fd) return {Exception::kIo}; ssize_t n = ::recv(fd, buffer.data(), buffer.size(), 0); - LOG(INFO)<< "Received something"; if (n < 0) { if (errno == EINTR) continue; if (errno == EAGAIN || errno == EWOULDBLOCK) continue; if (errno == EBADF) { - LOG(INFO) << __func__ << ": socket was closed during read"; return {Exception::kIo}; } LOG(ERROR) << __func__ << ": error reading data on bluetooth socket: " @@ -160,7 +156,6 @@ ExceptionOr BleL2capInputStream::Read(std::int64_t size) { // LOG(INFO) << __func__ << ": socket closed (EOF)"; // return {Exception::kIo}; // } - LOG(INFO)<< "Got this many "<< static_cast(n); buffer.resize(static_cast(n)); @@ -172,9 +167,6 @@ ExceptionOr BleL2capInputStream::Read(std::int64_t size) { // close to l2test semantics (one recv consumes one frame). size_t to_return = std::min(static_cast(size), buffer.size()); std::string out = buffer.substr(0, to_return); - LOG(INFO) << __func__ << ": returning " << to_return - << " bytes from SDU size " << buffer.size() << ", data=0x" - << HexPreview(out.data(), out.size()); return ExceptionOr(ByteArray(std::move(out))); } } @@ -205,10 +197,6 @@ Exception BleL2capOutputStream::Write(absl::string_view data) { max_chunk_size = kDefaultBleL2capMtu; } - LOG(INFO) << "BleL2capOutputStream::Write bytes=" << data.size() - << " mtu=" << max_chunk_size - << " data=0x" << HexPreview(data.data(), data.size()); - size_t offset = 0; while (offset < data.size()) { if (fd_.load() != fd) return {Exception::kIo}; @@ -280,9 +268,6 @@ BleL2capSocket::BleL2capSocket(int fd, input_stream_(std::make_unique(fd)), output_stream_(std::make_unique(fd)) { - LOG(INFO) << "fd_ " << fd; - LOG(INFO) << "input_stream_ :" << input_stream_.get(); - LOG(INFO) << "output_stream_ :" << output_stream_.get(); struct l2cap_options opts; size_t snd_mtu = GetSocketMtu(fd, BT_SNDMTU); size_t rcv_mtu = GetSocketMtu(fd, BT_RCVMTU); diff --git a/internal/platform/implementation/linux/ble_v2_medium.cc b/internal/platform/implementation/linux/ble_v2_medium.cc index 01135b90..7d4ff3cf 100644 --- a/internal/platform/implementation/linux/ble_v2_medium.cc +++ b/internal/platform/implementation/linux/ble_v2_medium.cc @@ -468,8 +468,6 @@ BleV2Medium::OpenL2capServerSocket(const std::string &service_id) { auto psm = 0x80 + (prng.NextUint32() % 0x80); auto server_socket = std::make_unique(psm); - LOG(INFO) << __func__ << ": L2CAP server socket created with PSM: " - << server_socket->GetPSM(); return server_socket; }