mirror of
https://github.com/kidfromjupiter/nearby.git
synced 2026-09-14 14:46:12 -04:00
rewrote l2cap server socket to fix deadlocks
This commit is contained in:
Lasan Mahaliyana
@@ -14,188 +14,224 @@
|
||||
|
||||
#include "internal/platform/implementation/linux/ble_l2cap_server_socket.h"
|
||||
|
||||
#include <poll.h>
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
#include <bluetooth/bluetooth.h>
|
||||
#include <bluetooth/l2cap.h>
|
||||
#include <poll.h>
|
||||
#include <sys/socket.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <cerrno>
|
||||
#include <cstdint>
|
||||
#include <cstring>
|
||||
#include <utility>
|
||||
|
||||
#include <bluetooth/bluetooth.h>
|
||||
#include <bluetooth/l2cap.h>
|
||||
|
||||
#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<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 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<sockaddr*>(&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<sockaddr*>(&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<api::ble::BleL2capSocket> 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<api::ble::BleL2capSocket> 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<uint8_t>(client_addr.l2_bdaddr.b[i]);
|
||||
}
|
||||
|
||||
accepted_fds_.emplace(server_fd_, std::pair(client_fd, peripheral_id));
|
||||
return std::make_unique<BleL2capSocket>(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<void()> 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<void()> notifier) {
|
||||
absl::AnyInvocable<void()> 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
|
||||
|
||||
@@ -16,15 +16,12 @@
|
||||
#define PLATFORM_IMPL_LINUX_BLE_L2CAP_SERVER_SOCKET_H_
|
||||
|
||||
#include <memory>
|
||||
#include <sdbus-c++/Types.h>
|
||||
|
||||
#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 <bluetooth/bluetooth.h>
|
||||
#include <bluetooth/l2cap.h>
|
||||
@@ -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<api::ble::BleL2capSocket> Accept() override;
|
||||
Exception Close() override ;
|
||||
std::unique_ptr<api::ble::BleL2capSocket> Accept() override
|
||||
ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
|
||||
|
||||
void SetCloseNotifier(absl::AnyInvocable<void()> notifier);
|
||||
void SetCloseNotifier(absl::AnyInvocable<void()> 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);
|
||||
|
||||
// <server_fd, <client_fd, peripheral_id>>
|
||||
absl::flat_hash_map<int, std::pair<int,api::ble::BlePeripheral::UniqueId>> accepted_fds_;
|
||||
absl::Mutex mutex_;
|
||||
bool closed_ ABSL_GUARDED_BY(mutex_) = false;
|
||||
absl::AnyInvocable<void()> 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
|
||||
|
||||
Reference in New Issue
Block a user