From 2c8a7d84144bc4c2c1e411ee44d5a3d21ea68766 Mon Sep 17 00:00:00 2001 From: Lasan Mahaliyana Date: Thu, 25 Jun 2026 16:03:47 +0530 Subject: [PATCH] converted linux stream to just have simple integer backed fds. I don't understand the ownership heirarchy to safely use smart pointers --- .../implementation/linux/ble_l2cap_socket.h | 13 ++++--- .../linux/bluetooth_classic_socket.h | 12 +++++-- .../platform/implementation/linux/stream.cc | 30 ++++++++-------- .../platform/implementation/linux/stream.h | 14 ++++---- .../implementation/linux/tcp_server_socket.h | 35 ++++++++++++------- .../linux/wifi_hotspot_socket.h | 10 ++++-- 6 files changed, 70 insertions(+), 44 deletions(-) diff --git a/internal/platform/implementation/linux/ble_l2cap_socket.h b/internal/platform/implementation/linux/ble_l2cap_socket.h index 80964624..65ad2a75 100644 --- a/internal/platform/implementation/linux/ble_l2cap_socket.h +++ b/internal/platform/implementation/linux/ble_l2cap_socket.h @@ -21,6 +21,7 @@ #include #include #include +#include #include "absl/functional/any_invocable.h" #include "absl/synchronization/mutex.h" @@ -37,7 +38,7 @@ class BleL2capSocket; class BleL2capInputStream : public nearby::InputStream { public: - BleL2capInputStream(sdbus::UnixFd fd) : stream_(fd) {} + explicit BleL2capInputStream(int fd) : stream_(fd) {} ExceptionOr Read(std::int64_t size) override; Exception Close() override { @@ -57,7 +58,7 @@ class BleL2capInputStream : public nearby::InputStream { class BleL2capOutputStream : public nearby::OutputStream { public: - BleL2capOutputStream(sdbus::UnixFd fd) : stream_(fd) {} + explicit BleL2capOutputStream(int fd) : stream_(fd) {} Exception Write(absl::string_view data) override; Exception Flush() override; @@ -78,7 +79,7 @@ class BleL2capSocket final : public api::ble::BleL2capSocket { public: BleL2capSocket(int fd, api::ble::BlePeripheral::UniqueId peripheral_id, std::string service_id = "") - : fd_(sdbus::UnixFd(fd)), + : fd_(fd), output_stream_(fd_), input_stream_(fd_), peripheral_id_(peripheral_id) {}; @@ -88,6 +89,10 @@ class BleL2capSocket final : public api::ble::BleL2capSocket { Exception Close() override { input_stream_.Close(); output_stream_.Close(); + if (fd_ >= 0) { + close(fd_); + fd_ = -1; + } return Exception{Exception::kSuccess}; }; @@ -96,7 +101,7 @@ class BleL2capSocket final : public api::ble::BleL2capSocket { } private: - sdbus::UnixFd fd_; + int fd_; BleL2capOutputStream output_stream_; BleL2capInputStream input_stream_; api::ble::BlePeripheral::UniqueId peripheral_id_; diff --git a/internal/platform/implementation/linux/bluetooth_classic_socket.h b/internal/platform/implementation/linux/bluetooth_classic_socket.h index a1353713..d1e14443 100644 --- a/internal/platform/implementation/linux/bluetooth_classic_socket.h +++ b/internal/platform/implementation/linux/bluetooth_classic_socket.h @@ -21,6 +21,7 @@ #include #include +#include #include #include "absl/synchronization/mutex.h" @@ -38,20 +39,27 @@ class BluetoothSocket final : public api::BluetoothSocket { public: BluetoothSocket(std::shared_ptr device, sdbus::UnixFd fd) - :fd_(fd), device_(std::move(device)), output_stream_(fd_), input_stream_(fd_) {} + : fd_(fd.release()), + device_(std::move(device)), + output_stream_(fd_), + input_stream_(fd_) {} InputStream &GetInputStream() override { return input_stream_; } OutputStream &GetOutputStream() override { return output_stream_; } Exception Close() override { input_stream_.Close(); output_stream_.Close(); + if (fd_ >= 0) { + close(fd_); + fd_ = -1; + } return Exception{Exception::kSuccess}; } api::BluetoothDevice *GetRemoteDevice() override { return device_.get(); }; private: - sdbus::UnixFd fd_; + int fd_; std::shared_ptr device_; OutputStream output_stream_; InputStream input_stream_; diff --git a/internal/platform/implementation/linux/stream.cc b/internal/platform/implementation/linux/stream.cc index 48997f3f..cd457159 100644 --- a/internal/platform/implementation/linux/stream.cc +++ b/internal/platform/implementation/linux/stream.cc @@ -15,10 +15,9 @@ #include #include #include -#include -#include #include #include +#include #include "absl/strings/escaping.h" #include "internal/platform/byte_array.h" @@ -34,7 +33,7 @@ ExceptionOr InputStream::Read(std::int64_t size) { return ExceptionOr(ByteArray(std::string())); } - if (!fd_ || !fd_->isValid()) { + if (closed_ || fd_ < 0) { return {Exception::kIo}; } @@ -43,7 +42,7 @@ ExceptionOr InputStream::Read(std::int64_t size) { while (true) { pollfd pfd{}; - pfd.fd = fd_->get(); + pfd.fd = fd_; pfd.events = POLLIN; int poll_result = poll(&pfd, 1, -1); @@ -64,7 +63,7 @@ ExceptionOr InputStream::Read(std::int64_t size) { } if (pfd.revents & (POLLIN | POLLHUP)) { - ssize_t bytes_read = recv(fd_->get(), buffer.data(), buffer.size(), 0); + ssize_t bytes_read = recv(fd_, buffer.data(), buffer.size(), 0); if (bytes_read > 0) { buffer.resize(static_cast(bytes_read)); @@ -93,17 +92,18 @@ ExceptionOr InputStream::Read(std::int64_t size) { } Exception InputStream::Close() { - if (!fd_->isValid()) return Exception{Exception::kIo}; - fd_.reset(); - return {}; + if (closed_ || fd_ < 0) return Exception{Exception::kSuccess}; + closed_ = true; + shutdown(fd_, SHUT_RD); + return Exception{Exception::kSuccess}; } Exception OutputStream::Write(absl::string_view data) { - if (!fd_ || !fd_->isValid()) { + if (closed_ || fd_ < 0) { return {Exception::kIo}; } - const int fd = fd_->get(); + const int fd = fd_; size_t sent = 0; while (sent < data.size()) { @@ -165,12 +165,10 @@ Exception OutputStream::Flush() { } Exception OutputStream::Close() { - if (!fd_->isValid()) return Exception{Exception::kIo}; - - auto ret = close(fd_->get()) < 0 ? Exception{Exception::kIo} - : Exception{Exception::kSuccess}; - fd_.reset(); - return ret; + if (closed_ || fd_ < 0) return Exception{Exception::kSuccess}; + closed_ = true; + shutdown(fd_, SHUT_WR); + return Exception{Exception::kSuccess}; } } // namespace linux diff --git a/internal/platform/implementation/linux/stream.h b/internal/platform/implementation/linux/stream.h index b4b42f19..4223a8e0 100644 --- a/internal/platform/implementation/linux/stream.h +++ b/internal/platform/implementation/linux/stream.h @@ -15,10 +15,6 @@ #ifndef PLATFORM_IMPL_LINUX_STREAM_H_ #define PLATFORM_IMPL_LINUX_STREAM_H_ -#include - -#include - #include "internal/platform/input_stream.h" #include "internal/platform/output_stream.h" @@ -26,26 +22,28 @@ namespace nearby { namespace linux { class InputStream : public nearby::InputStream { public: - explicit InputStream(sdbus::UnixFd fd) : fd_(std::make_shared(fd)){}; + explicit InputStream(int fd) : fd_(fd){}; ExceptionOr Read(std::int64_t size) override; Exception Close() override; private: - std::shared_ptr fd_; + int fd_; + bool closed_ = false; }; class OutputStream : public nearby::OutputStream { public: - explicit OutputStream(sdbus::UnixFd fd) : fd_(std::make_shared(fd)){}; + explicit OutputStream(int fd) : fd_(fd){}; Exception Write(absl::string_view data) override; Exception Flush() override; Exception Close() override; private: - std::shared_ptr fd_; + int fd_; + bool closed_ = false; }; } // namespace linux diff --git a/internal/platform/implementation/linux/tcp_server_socket.h b/internal/platform/implementation/linux/tcp_server_socket.h index a2e983d7..cb016fc9 100644 --- a/internal/platform/implementation/linux/tcp_server_socket.h +++ b/internal/platform/implementation/linux/tcp_server_socket.h @@ -18,9 +18,10 @@ #include #include #include +#include +#include #include - -#include +#include #include "internal/platform/exception.h" #include "internal/platform/implementation/linux/stream.h" @@ -30,8 +31,8 @@ namespace nearby { namespace linux { class TCPSocket { public: - explicit TCPSocket(const sdbus::UnixFd& fd) - : closed_(false), output_stream_(fd), input_stream_(fd) {} + explicit TCPSocket(int fd) + : fd_(fd), closed_(false), output_stream_(fd), input_stream_(fd) {} static std::optional Connect(const std::string& ip_address, int port) { @@ -54,10 +55,11 @@ class TCPSocket { if (ret < 0) { LOG(ERROR) << __func__ << ": Error connecting to socket: " << std::strerror(errno); + close(sock); return std::nullopt; } - return TCPSocket(sdbus::UnixFd(sock)); + return TCPSocket(sock); } InputStream& GetInputStream() { return input_stream_; } @@ -69,11 +71,16 @@ class TCPSocket { closed_ = true; input_stream_.Close(); output_stream_.Close(); + if (fd_ >= 0) { + close(fd_); + fd_ = -1; + } return {Exception::kSuccess}; }; private: + int fd_; bool closed_; OutputStream output_stream_; @@ -125,19 +132,23 @@ class TCPServerSocket { socklen_t len = sizeof(addr); auto conn = - accept(fd_.get(), reinterpret_cast(&addr), &len); + accept(fd_, reinterpret_cast(&addr), &len); if (conn < 0) { LOG(ERROR) << __func__ << ": Error accepting incoming connections on socket " - << fd_.get() << ": " << std::strerror(errno); + << fd_ << ": " << std::strerror(errno); return std::nullopt; } - return TCPSocket(sdbus::UnixFd(conn)); + return TCPSocket(conn); }; Exception Close() { - int fd = fd_.release(); + if (fd_ < 0) { + return {Exception::kSuccess}; + } + int fd = fd_; + fd_ = -1; shutdown(fd, SHUT_RDWR); auto ret = close(fd); if (ret < 0) { @@ -153,11 +164,11 @@ class TCPServerSocket { struct sockaddr_in sin; socklen_t len = sizeof(sin); auto ret = - getsockname(fd_.get(), reinterpret_cast(&sin), &len); + getsockname(fd_, reinterpret_cast(&sin), &len); if (ret < 0) { LOG(ERROR) << __func__ << ": Error getting information for socket " - << fd_.get() << ": " << std::strerror(errno); + << fd_ << ": " << std::strerror(errno); return 0; } @@ -165,7 +176,7 @@ class TCPServerSocket { } private: - sdbus::UnixFd fd_; + int fd_; }; } // namespace linux } // namespace nearby diff --git a/internal/platform/implementation/linux/wifi_hotspot_socket.h b/internal/platform/implementation/linux/wifi_hotspot_socket.h index e94587ff..27fda3e7 100644 --- a/internal/platform/implementation/linux/wifi_hotspot_socket.h +++ b/internal/platform/implementation/linux/wifi_hotspot_socket.h @@ -18,12 +18,14 @@ #include "internal/platform/implementation/linux/stream.h" #include "internal/platform/implementation/wifi_hotspot.h" +#include + namespace nearby { namespace linux { class WifiHotspotSocket : public api::WifiHotspotSocket { public: explicit WifiHotspotSocket(int connection_fd) - : fd_(sdbus::UnixFd(connection_fd)), + : fd_(connection_fd), output_stream_(fd_), input_stream_(fd_) {} @@ -32,12 +34,16 @@ class WifiHotspotSocket : public api::WifiHotspotSocket { Exception Close() override { input_stream_.Close(); output_stream_.Close(); + if (fd_ >= 0) { + close(fd_); + fd_ = -1; + } return Exception{Exception::kSuccess}; }; private: - sdbus::UnixFd fd_; + int fd_; OutputStream output_stream_; InputStream input_stream_; };