TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-16 15:36:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

fixed l2cap server socket hanging

Browse Source
This commit is contained in:
kidfromjupiter
2026-02-06 11:48:26 +00:00
parent 4d2f34ccfb
commit 2718d9942b
4 changed files with 159 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/platform/implementation/linux/ble_l2cap_server_socket.cc
+148 -111
View File
@@ -14,10 +14,12 @@
#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 <sys/socket.h>
#include <unistd.h>
#include <cerrno>
#include <cstring>
@@ -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<api::ble::BleL2capSocket> 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<BleL2capSocket>(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<api::ble::BleL2capSocket> 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<api::ble::BleL2capSocket> BleL2capServerSocket::Accept() {
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 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<void()> notifier) {
absl::MutexLock lock(&mutex_);
close_notifier_ = std::move(notifier);
}
} // namespace linux
} // namespace nearby
internal/platform/implementation/linux/ble_l2cap_server_socket.h
+11 -12
View File
@@ -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 <bluetooth/bluetooth.h>
#include <bluetooth/l2cap.h>
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<api::ble::BleL2capSocket> Accept() override
ABSL_LOCKS_EXCLUDED(mutex_);
Exception Close() override ABSL_LOCKS_EXCLUDED(mutex_);
std::unique_ptr<api::ble::BleL2capSocket> Accept() override;
Exception Close() override ;
void SetCloseNotifier(absl::AnyInvocable<void()> notifier)
ABSL_LOCKS_EXCLUDED(mutex_);
void SetCloseNotifier(absl::AnyInvocable<void()> notifier);
private:
Exception DoClose() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
mutable absl::Mutex mutex_;
absl::CondVar cond_;
absl::AnyInvocable<void()> 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_;
// <server_fd, <client_fd, peripheral_id>>
absl::flat_hash_map<int, std::pair<int,api::ble::BlePeripheral::UniqueId>> accepted_fds_ ABSL_GUARDED_BY(mutex_);
absl::flat_hash_map<int, std::pair<int,api::ble::BlePeripheral::UniqueId>> accepted_fds_;
};
} // namespace linux
internal/platform/implementation/linux/ble_l2cap_socket.cc
-15
View File
@@ -124,8 +124,6 @@ ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
if (size <= 0) {
return ExceptionOr<ByteArray>(ByteArray{});
}
LOG(INFO) << "Reading";
auto poller = Poller::CreateOutputPoller(fd);
while (true) {
@@ -144,12 +142,10 @@ ExceptionOr<ByteArray> 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<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
// LOG(INFO) << __func__ << ": socket closed (EOF)";
// return {Exception::kIo};
// }
LOG(INFO)<< "Got this many "<< static_cast<size_t>(n);
buffer.resize(static_cast<size_t>(n));
@@ -172,9 +167,6 @@ ExceptionOr<ByteArray> BleL2capInputStream::Read(std::int64_t size) {
// close to l2test semantics (one recv consumes one frame).
size_t to_return = std::min(static_cast<size_t>(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>(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<BleL2capInputStream>(fd)),
output_stream_(std::make_unique<BleL2capOutputStream>(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);
internal/platform/implementation/linux/ble_v2_medium.cc
-2
View File
@@ -468,8 +468,6 @@ BleV2Medium::OpenL2capServerSocket(const std::string &service_id) {
auto psm = 0x80 + (prng.NextUint32() % 0x80);
auto server_socket = std::make_unique<linux::BleL2capServerSocket>(psm);
LOG(INFO) << __func__ << ": L2CAP server socket created with PSM: "
<< server_socket->GetPSM();
return server_socket;
}