BufferHubQueue: ClearAvailable on consumer import

This fixes the issue that ConsumerQueue keeps receiving
|OnBufferAllocated| events when using UDS based libpdx. It also hardens
error handling during consumer imports.

Bug: 34690159
Test: Build, flash, buffer_hub_queue-test pases
Change-Id: I4f739a2869df851ed836f39122b3eba95697f3f6

1 files changed, 73 insertions, 25 deletions

diff --git a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
index 0576b21a21..bad9503383 100644
--- a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
+++ b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
@@ -1,5 +1,7 @@
 #include "include/private/dvr/buffer_hub_queue_client.h"
 
+//#define LOG_NDEBUG 0
+
 #include <inttypes.h>
 #include <log/log.h>
 #include <sys/epoll.h>
@@ -24,6 +26,7 @@ BufferHubQueue::BufferHubQueue(LocalChannelHandle channel_handle,
       meta_size_(meta_size),
       meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
       buffers_(BufferHubQueue::kMaxQueueCapacity),
+      epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
       available_buffers_(BufferHubQueue::kMaxQueueCapacity),
       capacity_(0) {
   Initialize();
@@ -36,6 +39,7 @@ BufferHubQueue::BufferHubQueue(const std::string& endpoint_path,
       meta_size_(meta_size),
       meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
       buffers_(BufferHubQueue::kMaxQueueCapacity),
+      epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
       available_buffers_(BufferHubQueue::kMaxQueueCapacity),
       capacity_(0) {
   Initialize();
@@ -101,31 +105,12 @@ bool BufferHubQueue::WaitForBuffers(int timeout) {
 
       ALOGD("New BufferHubQueue event %d: index=%" PRId64, i, index);
 
-      if (is_buffer_event_index(index) && (events[i].events & EPOLLIN)) {
-        auto buffer = buffers_[index];
-        ret = OnBufferReady(buffer);
-        if (ret < 0) {
-          ALOGE("Failed to set buffer ready: %s", strerror(-ret));
-          continue;
-        }
-        Enqueue(buffer, index);
-      } else if (is_buffer_event_index(index) &&
-                 (events[i].events & EPOLLHUP)) {
-        // This maybe caused by producer replacing an exising buffer slot.
-        // Currently the epoll FD is cleaned up when the replacement consumer
-        // client is imported.
-        ALOGW("Receives EPOLLHUP at slot: %" PRId64, index);
-      } else if (is_queue_event_index(index) && (events[i].events & EPOLLIN)) {
-        // Note that after buffer imports, if |count()| still returns 0, epoll
-        // wait will be tried again to acquire the newly imported buffer.
-        ret = OnBufferAllocated();
-        if (ret < 0) {
-          ALOGE("Failed to import buffer: %s", strerror(-ret));
-          continue;
-        }
+      if (is_buffer_event_index(index)) {
+        HandleBufferEvent(static_cast<size_t>(index), events[i]);
+      } else if (is_queue_event_index(index)) {
+        HandleQueueEvent(events[i]);
       } else {
-        ALOGW("Unknown event %d: u64=%" PRId64 ": events=%" PRIu32, i, index,
-              events[i].events);
+        ALOGW("Unknown event index: %" PRId64, index);
       }
     }
   }
@@ -133,6 +118,68 @@ bool BufferHubQueue::WaitForBuffers(int timeout) {
   return true;
 }
 
+void BufferHubQueue::HandleBufferEvent(size_t slot, const epoll_event& event) {
+  auto buffer = buffers_[slot];
+  if (!buffer) {
+    ALOGW("BufferHubQueue::HandleBufferEvent: Invalid buffer slot: %zu", slot);
+    return;
+  }
+
+  auto status = buffer->GetEventMask(event.events);
+  if (!status) {
+    ALOGW("BufferHubQueue::HandleBufferEvent: Failed to get event mask: %s",
+          status.GetErrorMessage().c_str());
+    return;
+  }
+
+  int events = status.get();
+  if (events & EPOLLIN) {
+    int ret = OnBufferReady(buffer);
+    if (ret < 0) {
+      ALOGE("Failed to set buffer ready: %s", strerror(-ret));
+      return;
+    }
+    Enqueue(buffer, slot);
+  } else if (events & EPOLLHUP) {
+    // This might be caused by producer replacing an existing buffer slot, or
+    // when BufferHubQueue is shutting down. For the first case, currently the
+    // epoll FD is cleaned up when the replacement consumer client is imported,
+    // we shouldn't detach again if |epollhub_pending_[slot]| is set.
+    ALOGW(
+        "Receives EPOLLHUP at slot: %zu, buffer event fd: %d, EPOLLHUP "
+        "pending: %d",
+        slot, buffer->event_fd(), epollhup_pending_[slot]);
+    if (epollhup_pending_[slot]) {
+      epollhup_pending_[slot] = false;
+    } else {
+      DetachBuffer(slot);
+    }
+  } else {
+    ALOGW("Unknown event, slot=%zu, epoll events=%d", slot, events);
+  }
+}
+
+void BufferHubQueue::HandleQueueEvent(const epoll_event& event) {
+  auto status = GetEventMask(event.events);
+  if (!status) {
+    ALOGW("BufferHubQueue::HandleQueueEvent: Failed to get event mask: %s",
+          status.GetErrorMessage().c_str());
+    return;
+  }
+
+  int events = status.get();
+  if (events & EPOLLIN) {
+    // Note that after buffer imports, if |count()| still returns 0, epoll
+    // wait will be tried again to acquire the newly imported buffer.
+    int ret = OnBufferAllocated();
+    if (ret < 0) {
+      ALOGE("Failed to import buffer: %s", strerror(-ret));
+    }
+  } else {
+    ALOGW("Unknown epoll events=%d", events);
+  }
+}
+
 int BufferHubQueue::AddBuffer(const std::shared_ptr<BufferHubBuffer>& buf,
                               size_t slot) {
   if (is_full()) {
@@ -146,8 +193,9 @@ int BufferHubQueue::AddBuffer(const std::shared_ptr<BufferHubBuffer>& buf,
   if (buffers_[slot] != nullptr) {
     // Replace the buffer if the slot is preoccupied. This could happen when the
     // producer side replaced the slot with a newly allocated buffer. Detach the
-    // buffer and set up with the new one.
+    // buffer before setting up with the new one.
     DetachBuffer(slot);
+    epollhup_pending_[slot] = true;
   }
 
   epoll_event event = {.events = EPOLLIN | EPOLLET, .data = {.u64 = slot}};