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

Really remove recursive locks in IncomingFramesReader.

Browse Source 
PiperOrigin-RevId: 672616279
This commit is contained in:
Francis Tsui
2024-09-09 11:39:15 -07:00
committed by Copybara-Service
parent 686abb539f
commit 48001e65cd
4 changed files with 156 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sharing/BUILD
+2 -1
View File
@@ -143,8 +143,9 @@ cc_library(
"//sharing/internal/public:logging",
"//sharing/proto:wire_format_cc_proto",
"@com_google_absl//absl/base:core_headers",
"@com_google_absl//absl/memory",
"@com_google_absl//absl/synchronization",
"@com_google_absl//absl/time",
"@com_google_absl//absl/types:span",
],
)
sharing/incoming_frames_reader.cc
+75 -87
View File
@@ -16,17 +16,17 @@
#include <stdint.h>
#include <algorithm>
#include <functional>
#include <map>
#include <memory>
#include <optional>
#include <queue>
#include <utility>
#include <vector>
#include "absl/memory/memory.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
#include "absl/types/span.h"
#include "internal/platform/mutex_lock.h"
#include "internal/platform/task_runner.h"
#include "sharing/internal/public/logging.h"
#include "sharing/nearby_connection.h"
@@ -41,11 +41,12 @@ using FrameType = ::nearby::sharing::service::proto::V1Frame_FrameType;
using V1Frame = ::nearby::sharing::service::proto::V1Frame;
using Frame = ::nearby::sharing::service::proto::Frame;
std::unique_ptr<Frame> DecodeFrame(absl::Span<const uint8_t> data) {
std::unique_ptr<V1Frame> DecodeV1Frame(const std::vector<uint8_t>& data) {
auto frame = std::make_unique<Frame>();
if (frame->ParseFromArray(data.data(), data.size())) {
return frame;
if (frame->ParseFromArray(data.data(), data.size()) &&
frame->version() == Frame::V1) {
return absl::WrapUnique(frame->release_v1());
} else {
return nullptr;
}
@@ -67,60 +68,55 @@ IncomingFramesReader::~IncomingFramesReader() {
void IncomingFramesReader::ReadFrame(
std::function<void(std::optional<V1Frame>)> callback) {
{
MutexLock lock(&mutex_);
if (!read_frame_info_queue_.empty()) {
ReadFrameInfo read_fame_info{std::nullopt, std::move(callback),
std::nullopt};
read_frame_info_queue_.push(std::move(read_fame_info));
return;
}
// Check in the cache for frame.
std::optional<V1Frame> cached_frame = GetCachedFrame(std::nullopt);
if (cached_frame.has_value()) {
callback(std::move(cached_frame));
return;
}
ReadFrameInfo read_fame_info{std::nullopt, std::move(callback),
std::nullopt};
read_frame_info_queue_.push(std::move(read_fame_info));
}
ReadNextFrame();
ProcessReadRequest(std::nullopt, std::move(callback), absl::ZeroDuration());
}
void IncomingFramesReader::ReadFrame(
FrameType frame_type, std::function<void(std::optional<V1Frame>)> callback,
absl::Duration timeout) {
ProcessReadRequest(frame_type, std::move(callback), timeout);
}
void IncomingFramesReader::ProcessReadRequest(
std::optional<FrameType> frame_type,
std::function<void(std::optional<V1Frame>)> callback,
absl::Duration timeout) {
std::unique_ptr<V1Frame> cached_frame;
{
MutexLock lock(&mutex_);
absl::MutexLock lock(&mutex_);
if (!read_frame_info_queue_.empty()) {
// There are already outstanding read requests, just queue this up.
ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout};
read_frame_info_queue_.push(std::move(read_fame_info));
return;
}
// Check in the cache for frame.
std::optional<V1Frame> cached_frame = GetCachedFrame(frame_type);
if (cached_frame.has_value()) {
callback(std::move(cached_frame));
return;
}
ReadFrameInfo read_fame_info{frame_type, std::move(callback), timeout};
read_frame_info_queue_.push(std::move(read_fame_info));
cached_frame = PopCachedFrame(frame_type);
}
if (cached_frame) {
callback(*cached_frame);
return;
}
{
// No matching cached frame, queue this request, then read more frames.
absl::MutexLock lock(&mutex_);
ReadFrameInfo read_frame_info{frame_type, std::move(callback), timeout};
read_frame_info_queue_.push(std::move(read_frame_info));
if (timeout != absl::ZeroDuration()) {
timeout_timer_ = std::make_unique<ThreadTimer>(
service_thread_, "frame_reader_timeout", timeout,
[reader = GetWeakPtr()]() {
auto frame_reader = reader.lock();
if (frame_reader == nullptr) {
NL_LOG(WARNING) << "IncomingFramesReader is released before.";
NL_LOG(WARNING) << "IncomingFramesReader has already been released "
"before read timeout.";
return;
}
frame_reader->OnTimeout();
});
}
}
ReadNextFrame();
}
@@ -133,7 +129,12 @@ void IncomingFramesReader::ReadNextFrame() {
NL_LOG(WARNING) << "IncomingFramesReader is released before.";
return;
}
frame_reader->OnDataReadFromConnection(std::move(bytes));
if (!bytes.has_value()) {
NL_LOG(WARNING) << __func__ << ": Failed to read frame";
frame_reader->CloseAllPendingReads();
return;
}
frame_reader->OnDataReadFromConnection(*bytes);
});
}
@@ -143,64 +144,43 @@ void IncomingFramesReader::OnTimeout() {
}
void IncomingFramesReader::OnDataReadFromConnection(
std::optional<std::vector<uint8_t>> bytes) {
if (!bytes.has_value()) {
NL_LOG(WARNING) << __func__ << ": Failed to read frame";
CloseAllPendingReads();
return;
}
std::unique_ptr<Frame> frame =
DecodeFrame(absl::MakeSpan(bytes->data(), bytes->size()));
const std::vector<uint8_t>& bytes) {
std::unique_ptr<V1Frame> frame = DecodeV1Frame(bytes);
if (frame == nullptr) {
NL_LOG(WARNING)
<< __func__
<< ": Cannot decode frame. Not currently bound to nearby process";
CloseAllPendingReads();
ReadNextFrame();
return;
}
const V1Frame* v1_frame = OnFrameDecoded(*frame);
if (v1_frame != nullptr) {
Done(*v1_frame);
}
}
const V1Frame* IncomingFramesReader::OnFrameDecoded(const Frame& frame) {
if (frame.version() != Frame::V1) {
NL_VLOG(1) << __func__ << ": Frame read does not have V1Frame";
ReadNextFrame();
return nullptr;
}
auto v1_frame = frame.v1();
FrameType v1_frame_type = v1_frame.type();
FrameType frame_type = frame->type();
bool cached_frame = false;
{
MutexLock lock(&mutex_);
absl::MutexLock lock(&mutex_);
if (read_frame_info_queue_.empty()) {
return nullptr;
return;
}
const ReadFrameInfo& frame_info = read_frame_info_queue_.front();
if (frame_info.frame_type.has_value() &&
*frame_info.frame_type != v1_frame_type) {
*frame_info.frame_type != frame_type) {
NL_LOG(WARNING) << __func__ << ": Failed to read frame of type "
<< *frame_info.frame_type << ", but got frame of type "
<< v1_frame_type << ". Cached for later.";
cached_frames_.insert({v1_frame_type, v1_frame});
<< frame_type << ". Cached for later.";
cached_frames_.push_back(std::move(frame));
cached_frame = true;
}
}
if (cached_frame) {
ReadNextFrame();
return nullptr;
return;
}
return &frame.v1();
Done(std::move(frame));
}
void IncomingFramesReader::CloseAllPendingReads() {
std::queue<ReadFrameInfo> queue;
{
MutexLock lock(&mutex_);
absl::MutexLock lock(&mutex_);
queue.swap(read_frame_info_queue_);
}
while (!queue.empty()) {
@@ -210,18 +190,18 @@ void IncomingFramesReader::CloseAllPendingReads() {
}
}
void IncomingFramesReader::Done(const V1Frame& frame) {
void IncomingFramesReader::Done(std::unique_ptr<V1Frame> frame) {
ReadFrameInfo read_frame_info;
{
MutexLock lock(&mutex_);
absl::MutexLock lock(&mutex_);
timeout_timer_.reset();
read_frame_info = std::move(read_frame_info_queue_.front());
read_frame_info_queue_.pop();
}
read_frame_info.callback(std::move(frame));
read_frame_info.callback(*frame);
{
MutexLock lock(&mutex_);
absl::MutexLock lock(&mutex_);
if (read_frame_info_queue_.empty()) {
return;
}
@@ -229,28 +209,36 @@ void IncomingFramesReader::Done(const V1Frame& frame) {
read_frame_info_queue_.pop();
}
if (read_frame_info.timeout.has_value()) {
if (read_frame_info.timeout != absl::ZeroDuration()) {
ReadFrame(*read_frame_info.frame_type,
std::move(read_frame_info.callback), *read_frame_info.timeout);
std::move(read_frame_info.callback), read_frame_info.timeout);
} else {
ReadFrame(std::move(read_frame_info.callback));
}
}
std::optional<V1Frame> IncomingFramesReader::GetCachedFrame(
std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
frame_type) {
std::unique_ptr<V1Frame> IncomingFramesReader::PopCachedFrame(
std::optional<V1Frame::FrameType> frame_type) {
NL_VLOG(1) << __func__ << ": Fetching cached frame";
if (frame_type.has_value())
NL_VLOG(1) << __func__ << ": Requested frame type - " << *frame_type;
if (cached_frames_.empty()) {
return nullptr;
}
if (!frame_type.has_value()) {
std::unique_ptr<V1Frame> frame = std::move(cached_frames_.front());
cached_frames_.pop_front();
return frame;
}
NL_VLOG(1) << __func__ << ": Requested frame type - " << *frame_type;
auto iter = frame_type.has_value() ? cached_frames_.find(*frame_type)
: cached_frames_.begin();
if (iter == cached_frames_.end()) return std::nullopt;
auto iter =
std::find_if(cached_frames_.begin(), cached_frames_.end(),
[frame_type = *frame_type](std::unique_ptr<V1Frame>& frame) {
return frame->type() == frame_type;
});
if (iter == cached_frames_.end()) return nullptr;
NL_VLOG(1) << __func__ << ": Successfully read cached frame";
std::optional<V1Frame> frame = std::move(iter->second);
std::unique_ptr<V1Frame> frame = std::move(*iter);
cached_frames_.erase(iter);
return frame;
}
sharing/incoming_frames_reader.h
+18 -15
View File
@@ -18,7 +18,7 @@
#include <stdint.h>
#include <functional>
#include <map>
#include <list>
#include <memory>
#include <optional>
#include <queue>
@@ -26,7 +26,7 @@
#include "absl/base/thread_annotations.h"
#include "absl/time/time.h"
#include "internal/platform/mutex.h"
#include "absl/synchronization/mutex.h"
#include "internal/platform/task_runner.h"
#include "sharing/nearby_connection.h"
#include "sharing/proto/wire_format.pb.h"
@@ -63,7 +63,7 @@ class IncomingFramesReader
// Note: Callers are expected wait for |callback| to be run before scheduling
// subsequent calls to ReadFrame(..).
virtual void ReadFrame(
nearby::sharing::service::proto::V1Frame_FrameType frame_type,
nearby::sharing::service::proto::V1Frame::FrameType frame_type,
std::function<
void(std::optional<nearby::sharing::service::proto::V1Frame>)>
callback,
@@ -75,36 +75,39 @@ class IncomingFramesReader
private:
struct ReadFrameInfo {
std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
frame_type = std::nullopt;
std::function<void(std::optional<nearby::sharing::service::proto::V1Frame>)>
callback = nullptr;
std::optional<absl::Duration> timeout = std::nullopt;
absl::Duration timeout = absl::ZeroDuration();
};
void ProcessReadRequest(
std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
frame_type,
std::function<
void(std::optional<nearby::sharing::service::proto::V1Frame>)>
callback,
absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_);
void CloseAllPendingReads() ABSL_LOCKS_EXCLUDED(mutex_);
void ReadNextFrame() ABSL_LOCKS_EXCLUDED(mutex_);
void OnDataReadFromConnection(std::optional<std::vector<uint8_t>> bytes)
ABSL_LOCKS_EXCLUDED(mutex_);
const nearby::sharing::service::proto::V1Frame* OnFrameDecoded(
const nearby::sharing::service::proto::Frame& frame)
void OnDataReadFromConnection(const std::vector<uint8_t>& bytes)
ABSL_LOCKS_EXCLUDED(mutex_);
void OnTimeout();
void Done(const nearby::sharing::service::proto::V1Frame& frame)
void Done(std::unique_ptr<nearby::sharing::service::proto::V1Frame> frame)
ABSL_LOCKS_EXCLUDED(mutex_);
std::optional<nearby::sharing::service::proto::V1Frame> GetCachedFrame(
std::optional<nearby::sharing::service::proto::V1Frame_FrameType>
std::unique_ptr<nearby::sharing::service::proto::V1Frame> PopCachedFrame(
std::optional<nearby::sharing::service::proto::V1Frame::FrameType>
frame_type) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
TaskRunner& service_thread_;
NearbyConnection* const connection_;
RecursiveMutex mutex_;
absl::Mutex mutex_;
std::queue<ReadFrameInfo> read_frame_info_queue_ ABSL_GUARDED_BY(mutex_);
// Caches frames read from NearbyConnection which are not used immediately.
std::map<nearby::sharing::service::proto::V1Frame_FrameType,
std::optional<nearby::sharing::service::proto::V1Frame>>
std::list<std::unique_ptr<nearby::sharing::service::proto::V1Frame>>
cached_frames_ ABSL_GUARDED_BY(mutex_);
std::unique_ptr<ThreadTimer> timeout_timer_ ABSL_GUARDED_BY(mutex_);
sharing/incoming_frames_reader_test.cc
+61 -5
View File
@@ -35,6 +35,7 @@ namespace nearby {
namespace sharing {
namespace {
using ::nearby::sharing::service::proto::ConnectionResponseFrame;
using ::nearby::sharing::service::proto::V1Frame;
constexpr absl::Duration kTimeout = absl::Seconds(1);
@@ -48,7 +49,7 @@ std::optional<std::vector<uint8_t>> GetIntroductionFrame() {
v1frame->mutable_introduction();
std::vector<uint8_t> data;
data.resize(frame.ByteSize());
data.resize(frame.ByteSizeLong());
if (frame.SerializeToArray(data.data(), data.size())) {
return data;
}
@@ -64,7 +65,25 @@ std::optional<std::vector<uint8_t>> GetCancelFrame() {
v1frame->set_type(service::proto::V1Frame::CANCEL);
std::vector<uint8_t> data;
data.resize(frame.ByteSize());
data.resize(frame.ByteSizeLong());
if (frame.SerializeToArray(data.data(), data.size())) {
return data;
}
return std::nullopt;
}
std::optional<std::vector<uint8_t>> GetResponseFrame() {
nearby::sharing::service::proto::Frame frame =
nearby::sharing::service::proto::Frame();
frame.set_version(nearby::sharing::service::proto::Frame::V1);
V1Frame* v1frame = frame.mutable_v1();
v1frame->set_type(service::proto::V1Frame::RESPONSE);
v1frame->mutable_connection_response()->set_status(
ConnectionResponseFrame::ACCEPT);
std::vector<uint8_t> data;
data.resize(frame.ByteSizeLong());
if (frame.SerializeToArray(data.data(), data.size())) {
return data;
}
@@ -137,6 +156,29 @@ TEST_F(IncomingFramesReaderTest, ReadTimedOut) {
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
}
TEST_F(IncomingFramesReaderTest, ReadNonV1FrameSkipped) {
nearby::sharing::service::proto::Frame frame =
nearby::sharing::service::proto::Frame();
V1Frame* v1frame = frame.mutable_v1();
v1frame->set_type(service::proto::V1Frame::CANCEL);
std::vector<uint8_t> data;
data.resize(frame.ByteSizeLong());
ASSERT_GT(data.size(), 0);
ASSERT_TRUE(frame.SerializeToArray(data.data(), data.size()));
connection().WriteMessage(data);
std::optional<std::vector<uint8_t>> introduction_frame =
GetIntroductionFrame();
ASSERT_TRUE(introduction_frame.has_value());
connection().WriteMessage(*introduction_frame);
absl::Notification notification;
frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
EXPECT_EQ(frame->type(), service::proto::V1Frame::INTRODUCTION);
notification.Notify();
});
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
}
TEST_F(IncomingFramesReaderTest, ReadAnyFrameSuccessful) {
std::optional<std::vector<uint8_t>> introduction_frame =
GetIntroductionFrame();
@@ -193,6 +235,9 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
std::optional<std::vector<uint8_t>> cancel_frame = GetCancelFrame();
ASSERT_TRUE(cancel_frame.has_value());
connection().WriteMessage(*cancel_frame);
std::optional<std::vector<uint8_t>> response_frame = GetResponseFrame();
ASSERT_TRUE(response_frame.has_value());
connection().WriteMessage(*response_frame);
std::optional<std::vector<uint8_t>> introduction_frame =
GetIntroductionFrame();
@@ -209,7 +254,7 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
kTimeout);
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
// Reading any frame should return CancelFrame.
// Reading any frame should return cancel frame, then response frame.
absl::Notification cancel_notification;
frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
ASSERT_NE(frame, std::nullopt);
@@ -217,6 +262,13 @@ TEST_F(IncomingFramesReaderTest, JumbledFramesOrdering_ReadFromCache) {
cancel_notification.Notify();
});
EXPECT_TRUE(cancel_notification.WaitForNotificationWithTimeout(kTimeout));
absl::Notification response_notification;
frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
ASSERT_NE(frame, std::nullopt);
EXPECT_EQ(frame->type(), service::proto::V1Frame::RESPONSE);
response_notification.Notify();
});
EXPECT_TRUE(response_notification.WaitForNotificationWithTimeout(kTimeout));
}
TEST_F(IncomingFramesReaderTest, ReadAfterConnectionClosed) {
@@ -298,19 +350,23 @@ TEST_F(IncomingFramesReaderTest, ReleaseFrameReaderDuringRead) {
EXPECT_EQ(frames_reader(), nullptr);
}
TEST_F(IncomingFramesReaderTest, ReadInvalidFrame) {
TEST_F(IncomingFramesReaderTest, SkipInvalidFrame) {
absl::Notification notification;
frames_reader()->ReadFrame([&](std::optional<V1Frame> frame) {
EXPECT_EQ(frame, std::nullopt);
EXPECT_EQ(frame->type(), service::proto::V1Frame::CANCEL);
notification.Notify();
});
std::optional<std::vector<uint8_t>> invalid_frame = GetInvalidFrame();
ASSERT_TRUE(invalid_frame.has_value());
connection().WriteMessage(*invalid_frame);
std::optional<std::vector<uint8_t>> cancel_frame = GetCancelFrame();
ASSERT_TRUE(cancel_frame.has_value());
connection().WriteMessage(*cancel_frame);
Sync();
EXPECT_TRUE(notification.WaitForNotificationWithTimeout(kTimeout));
ReleaseFrameReader();
}
} // namespace