Remove recursive mutex in IncomingFrameReader.

PiperOrigin-RevId: 662219648

sharing/BUILD

@@ -144,6 +144,7 @@ cc_library(
         "//internal/platform:types",
         "//sharing/internal/public:logging",
         "//sharing/proto:wire_format_cc_proto",
+        "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/time",
         "@com_google_absl//absl/types:span",
     ],

sharing/incoming_frames_reader.cc

@@ -54,96 +54,92 @@ IncomingFramesReader::IncomingFramesReader(TaskRunner& service_thread,
 }
 
 IncomingFramesReader::~IncomingFramesReader() {
-  MutexLock lock(&mutex_);
   NL_LOG(INFO) << "~IncomingFramesReader is called";
-  Done(std::nullopt);
+  CloseAllPendingReads();
 }
 
 void IncomingFramesReader::ReadFrame(
     std::function<void(std::optional<V1Frame>)> callback) {
-  MutexLock lock(&mutex_);
-  if (!read_frame_info_queue_.empty()) {
+  {
+    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));
-    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();
 }
 
 void IncomingFramesReader::ReadFrame(
     FrameType frame_type, std::function<void(std::optional<V1Frame>)> callback,
     absl::Duration timeout) {
-  MutexLock lock(&mutex_);
-  if (!read_frame_info_queue_.empty()) {
+  {
+    MutexLock lock(&mutex_);
+    if (!read_frame_info_queue_.empty()) {
+      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));
-    return;
+
+    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.";
+            return;
+          }
+          frame_reader->OnTimeout();
+        });
   }
-
-  // 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));
-
-  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.";
-          return;
-        }
-        frame_reader->OnTimeout();
-      });
-
   ReadNextFrame();
 }
 
 void IncomingFramesReader::ReadNextFrame() {
   connection_->Read(
-      [&, reader = GetWeakPtr()](std::optional<std::vector<uint8_t>> bytes) {
+      [reader = GetWeakPtr()](std::optional<std::vector<uint8_t>> bytes) {
         auto frame_reader = reader.lock();
         if (frame_reader == nullptr) {
           NL_LOG(WARNING) << "IncomingFramesReader is released before.";
           return;
         }
-
-        OnDataReadFromConnection(std::move(bytes));
+        frame_reader->OnDataReadFromConnection(std::move(bytes));
       });
 }
 
 void IncomingFramesReader::OnTimeout() {
-  MutexLock lock(&mutex_);
   NL_LOG(WARNING) << __func__ << ": Timed out reading from NearbyConnection.";
-  Done(std::nullopt);
+  CloseAllPendingReads();
 }
 
 void IncomingFramesReader::OnDataReadFromConnection(
     std::optional<std::vector<uint8_t>> bytes) {
-  MutexLock lock(&mutex_);
-  if (read_frame_info_queue_.empty()) {
-    return;
-  }
-
   if (!bytes.has_value()) {
     NL_LOG(WARNING) << __func__ << ": Failed to read frame";
-    Done(std::nullopt);
+    CloseAllPendingReads();
     return;
   }
 
@@ -153,74 +149,84 @@ void IncomingFramesReader::OnDataReadFromConnection(
     NL_LOG(WARNING)
         << __func__
         << ": Cannot decode frame. Not currently bound to nearby process";
-    Done(std::nullopt);
+    CloseAllPendingReads();
     return;
   }
 
-  OnFrameDecoded(std::move(*frame));
+  const V1Frame* v1_frame = OnFrameDecoded(*frame);
+  if (v1_frame != nullptr) {
+    Done(*v1_frame);
+  }
 }
 
-void IncomingFramesReader::OnFrameDecoded(std::optional<Frame> frame) {
-  if (!frame.has_value()) {
-    ReadNextFrame();
-    return;
-  }
-
-  if (frame->version() != Frame::V1) {
+const V1Frame* IncomingFramesReader::OnFrameDecoded(const Frame& frame) {
+  if (frame.version() != Frame::V1) {
     NL_VLOG(1) << __func__ << ": Frame read does not have V1Frame";
     ReadNextFrame();
-    return;
+    return nullptr;
   }
-
-  auto v1_frame = frame->v1();
+  auto v1_frame = frame.v1();
   FrameType v1_frame_type = v1_frame.type();
-
-  const ReadFrameInfo& frame_info = read_frame_info_queue_.front();
-  if (frame_info.frame_type.has_value() &&
-      *frame_info.frame_type != v1_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, std::move(v1_frame)});
-    ReadNextFrame();
-    return;
+  bool cached_frame = false;
+  {
+    MutexLock lock(&mutex_);
+    if (read_frame_info_queue_.empty()) {
+      return nullptr;
+    }
+    const ReadFrameInfo& frame_info = read_frame_info_queue_.front();
+    if (frame_info.frame_type.has_value() &&
+        *frame_info.frame_type != v1_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});
+      cached_frame = true;
+    }
   }
-
-  Done(std::move(v1_frame));
+  if (cached_frame) {
+    ReadNextFrame();
+    return nullptr;
+  }
+  return &frame.v1();
 }
 
-void IncomingFramesReader::Done(std::optional<V1Frame> frame) {
-  if (read_frame_info_queue_.empty()) {
-    return;
+void IncomingFramesReader::CloseAllPendingReads() {
+  std::queue<ReadFrameInfo> queue;
+  {
+    MutexLock lock(&mutex_);
+    queue.swap(read_frame_info_queue_);
   }
+  while (!queue.empty()) {
+    ReadFrameInfo read_frame_info = std::move(queue.front());
+    queue.pop();
+    read_frame_info.callback(std::nullopt);
+  }
+}
 
-  timeout_timer_.reset();
-
-  bool is_empty_frame = !frame.has_value();
-  ReadFrameInfo read_frame_info = std::move(read_frame_info_queue_.front());
-  read_frame_info_queue_.pop();
+void IncomingFramesReader::Done(const V1Frame& frame) {
+  ReadFrameInfo read_frame_info;
+  {
+    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));
 
-  if (is_empty_frame) {
-    // should complete all pending readers.
-    while (!read_frame_info_queue_.empty()) {
-      read_frame_info = std::move(read_frame_info_queue_.front());
-      read_frame_info_queue_.pop();
-      read_frame_info.callback(std::nullopt);
+  {
+    MutexLock lock(&mutex_);
+    if (read_frame_info_queue_.empty()) {
+      return;
     }
-    return;
+    read_frame_info = std::move(read_frame_info_queue_.front());
+    read_frame_info_queue_.pop();
   }
 
-  if (!read_frame_info_queue_.empty()) {
-    ReadFrameInfo read_frame_info = std::move(read_frame_info_queue_.front());
-    read_frame_info_queue_.pop();
-
-    if (read_frame_info.timeout.has_value()) {
-      ReadFrame(*read_frame_info.frame_type,
-                std::move(read_frame_info.callback), *read_frame_info.timeout);
-    } else {
-      ReadFrame(std::move(read_frame_info.callback));
-    }
+  if (read_frame_info.timeout.has_value()) {
+    ReadFrame(*read_frame_info.frame_type,
+              std::move(read_frame_info.callback), *read_frame_info.timeout);
+  } else {
+    ReadFrame(std::move(read_frame_info.callback));
   }
 }
 

sharing/incoming_frames_reader.h

@@ -24,6 +24,7 @@
 #include <queue>
 #include <vector>
 
+#include "absl/base/thread_annotations.h"
 #include "absl/time/time.h"
 #include "internal/platform/mutex.h"
 #include "internal/platform/task_runner.h"
@@ -55,7 +56,7 @@ class IncomingFramesReader
   virtual void ReadFrame(
       std::function<
           void(std::optional<nearby::sharing::service::proto::V1Frame>)>
-          callback);
+          callback) ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Reads a frame of type |frame_type| from |connection|. |callback| is called
   // with the frame read from connection or nullopt if connection socket is
@@ -68,7 +69,7 @@ class IncomingFramesReader
       std::function<
           void(std::optional<nearby::sharing::service::proto::V1Frame>)>
           callback,
-      absl::Duration timeout);
+      absl::Duration timeout) ABSL_LOCKS_EXCLUDED(mutex_);
 
   std::weak_ptr<IncomingFramesReader> GetWeakPtr() {
     return this->weak_from_this();
@@ -83,30 +84,33 @@ class IncomingFramesReader
     std::optional<absl::Duration> timeout = std::nullopt;
   };
 
-  void ReadNextFrame();
-  void OnDataReadFromConnection(std::optional<std::vector<uint8_t>> bytes);
-  void OnFrameDecoded(
-      std::optional<nearby::sharing::service::proto::Frame> frame);
+  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)
+      ABSL_LOCKS_EXCLUDED(mutex_);
   void OnTimeout();
-  void Done(std::optional<nearby::sharing::service::proto::V1Frame> frame);
+  void Done(const 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>
-          frame_type);
+          frame_type) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
 
   TaskRunner& service_thread_;
   const NearbySharingDecoder& decoder_;
-  NearbyConnection* connection_;
+  NearbyConnection* const connection_;
 
   RecursiveMutex mutex_;
-  std::queue<ReadFrameInfo> read_frame_info_queue_;
-  std::function<void()> timeout_callback_;
+  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>>
-      cached_frames_;
+      cached_frames_ ABSL_GUARDED_BY(mutex_);
 
-  std::unique_ptr<ThreadTimer> timeout_timer_;
+  std::unique_ptr<ThreadTimer> timeout_timer_ ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace sharing

sharing/nearby_sharing_service_impl_test.cc

@@ -3204,6 +3204,11 @@ TEST_F(NearbySharingServiceImplTest,
   connection_->AppendReadableData(bytes);
   FlushTesting();
   EXPECT_CALL(*mock_app_info_, SetActiveFlag());
+  EXPECT_CALL(callback,
+              OnTransferUpdate(
+                  testing::_, testing::_,
+                  nearby::sharing::HasStatus(
+                      TransferMetadata::Status::kPairedKeyVerificationFailed)));
 
   service_->OnIncomingConnection(kEndpointId, GetValidV1EndpointInfo(),
                                  connection_.get());