Fix flaky ConsumerQueue::Dequeue after import
We recently noticed some flakiness around buffer_hub_queue-test's
BufferHubQueueTest.TestAllocateBuffer test case. After digging deep here
is the actual sympton:
There are two consecutive ConsumerQueue::Dequeue() calls in the test,
right after the consumer queue get created. The second call gets a
certain chance to fail and it appears that the queue logic is dequeuing
the same (i.e. the first) buffer again. Since the buffer already
acquired during the first Dequeue() call, the second Dequeue() will
eventually fail due to invalid buffer state transition.
The problem here is that we Enqueue() the first buffer twice:
(1) During buffer import, we always check whether the buffer has been
signaled. If so, we Enqueue() it, then Acquire it asynchronously during
Dequeue().
(2) During Dequeue(), the buffer's event_fd is already in the
ConsumerQeue's epoll set and is will **possibly** trigger the same
buffer being Enqueue()'d the second time. Note that this may not
happened if the AcquireAsync in step (1) race ahead of the epoll_wait
and clears the buffer available signal.
We had the logic described in step (1) as a safe guard to explicitly
catch the cases where buffers are already available when added into an
epoll set. Under this assuption (1) and (2) should happen exclusively:
either a buffer is available on import, thus edge triggered epoll won't
fire and we should only trigger (1); or the buffer is not available on
import, but becomes available later, which should only
trigger (2). However, this is not how epoll/eventfd works. We found
though unit test on epoll that adding a already signaled event_fd into a
epoll set will always trigger the epoll fd no matter whether the
event_fd is already signaled when being added into the epoll set. We
verified this on Android and mainline Linux kernel.
Thus, the check in step (1) is redundant and should be removed. Once (1)
is removed, we will always catch new buffer events through the same
edge-triggered epoll_wait in (2). Since it is edge triggered, it will
be trigged only once when the buffer becomes available, so it's
resilient to the race condition related to AcquireAsync().
Bug: 70306222
Test: buffer_hub_queue-test, buffer_hub-test
Change-Id: Ic24188ad84664bc102b85ca93ed24c354cb5c5a3
diff --git a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
index a5cefd9..c75c67f 100644
--- a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
+++ b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
@@ -31,20 +31,6 @@
namespace {
-// Polls an fd for the given events.
-Status<int> PollEvents(int fd, short events) {
- const int kTimeoutMs = 0;
- pollfd pfd{fd, events, 0};
- const int count = RETRY_EINTR(poll(&pfd, 1, kTimeoutMs));
- if (count < 0) {
- return ErrorStatus(errno);
- } else if (count == 0) {
- return ErrorStatus(ETIMEDOUT);
- } else {
- return {pfd.revents};
- }
-}
-
std::pair<int32_t, int32_t> Unstuff(uint64_t value) {
return {static_cast<int32_t>(value >> 32),
static_cast<int32_t>(value & ((1ull << 32) - 1))};
@@ -670,27 +656,7 @@
const std::shared_ptr<BufferConsumer>& buffer, size_t slot) {
ALOGD_IF(TRACE, "ConsumerQueue::AddBuffer: queue_id=%d buffer_id=%d slot=%zu",
id(), buffer->id(), slot);
- auto status = BufferHubQueue::AddBuffer(buffer, slot);
- if (!status)
- return status;
-
- // Check to see if the buffer is already signaled. This is necessary to catch
- // cases where buffers are already available; epoll edge triggered mode does
- // not fire until an edge transition when adding new buffers to the epoll
- // set. Note that we only poll the fd events because HandleBufferEvent() takes
- // care of checking the translated buffer events.
- auto poll_status = PollEvents(buffer->event_fd(), POLLIN);
- if (!poll_status && poll_status.error() != ETIMEDOUT) {
- ALOGE("ConsumerQueue::AddBuffer: Failed to poll consumer buffer: %s",
- poll_status.GetErrorMessage().c_str());
- return poll_status.error_status();
- }
-
- // Update accounting if the buffer is available.
- if (poll_status)
- return HandleBufferEvent(slot, buffer->event_fd(), poll_status.get());
- else
- return {};
+ return BufferHubQueue::AddBuffer(buffer, slot);
}
Status<std::shared_ptr<BufferConsumer>> ConsumerQueue::Dequeue(
diff --git a/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp b/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
index 3efa723..47a2734 100644
--- a/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
+++ b/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
@@ -329,7 +329,9 @@
// Check that buffers are correctly imported on construction.
EXPECT_EQ(consumer_queue_->capacity(), kBufferCount);
- EXPECT_EQ(consumer_queue_->count(), 1U);
+ // Buffers are only imported, but their availability is not checked until
+ // first call to Dequeue().
+ EXPECT_EQ(consumer_queue_->count(), 0U);
// Reclaim released/ignored buffers.
EXPECT_EQ(producer_queue_->count(), kBufferCount - 1);