Add shared memory based buffer metadata
This CLs reduces BufferHub CPU consumption by adding asynchronous
state transition so that out-of-process VR composition can run on 2016
pixel devices smoothly. In addition, this CL addresses a couple corner
cases in the existing bufferhub logic, which fixes various blackscreen
issues.
1/ Tracks buffer transition states (gained, posted, acquired, released)
from the client side via atomic shared memory and adds
PostAsync/AcquireAsync/ReleaseAsync/GainAsync with metadata and
fence support.
2/ Adds dequeue order guarantee for buffers enqueued with
dvrWriteBufferQueuePostBuffer.
3/ Synchronous BuffeHub operations are still supported.
4/ Bump up the bufferhubd's soft limit of open file descriptor.
5/ Handle orphaned consumer in acquired state. This is a corner case
that consumer process goes aways (most likely due to a crash) leaving
buffer stuck in acquired state with inconsistent buffer state.
6/ Fixes a race condition for released buffer to be Gain'ed and
Acquire'd when a new consumer is created in released state.
7/ Improve silent consumer queue efficiency: Silent queues no longer
import buffers or receive signals about new buffers and they are
limited to only spawning other consumers and notifications about
producers hanging up.
8/ Modify PDX/UDS channel event signaling to work around epoll
behavior. PDX UDS uses a combination of an eventfd and an epoll set
to simulate the original PDX transport channel events. An odd
behavior discovered in the kernel implementation of epoll was found
that causes the epoll fd to "unsignal" itself whenever epoll_wait()
is called on it, regardless of whether it should still be
pending. This breaks the edge triggerd behavior in nested epoll sets
that channel events depend on. Since this is unlikely to ever be
fixed in the kernel we work around the behavior by using the epoll
set only as a logical OR of two eventfds and never calling
epoll_wait() on it. When polling is required we use regluar poll()
with the eventfds and data fd to avoid the bad behavior in
epoll_wait().
9/ Keep reading data after PDX hangup signal. UDS will signal hangup
when the other end of the socket closes. However, data could still be
in the kerenl buffer and should be consumed. Fix an issue where the
service misses an impulse sent right before the socket is closed.
Bug: 65455724
Bug: 65458354
Bug: 65458312
Bug: 64027135
Bug: 67424527
Test: libpdx_uds_tests
bufferhub_tests
buffer_hub_queue-test
buffer_hub_queue_producer-test
dvr_api-test
Change-Id: Id07db1f206ccf4e06f7ee3c671193334408971ca
diff --git a/libs/vr/libbufferhubqueue/Android.bp b/libs/vr/libbufferhubqueue/Android.bp
index 0b3b2f0..93ccd0f 100644
--- a/libs/vr/libbufferhubqueue/Android.bp
+++ b/libs/vr/libbufferhubqueue/Android.bp
@@ -48,6 +48,7 @@
cflags: [
"-DLOG_TAG=\"libbufferhubqueue\"",
"-DTRACE=0",
+ "-DATRACE_TAG=ATRACE_TAG_GRAPHICS",
],
srcs: sourceFiles,
export_include_dirs: includeFiles,
diff --git a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
index f9f87ff..8bea0cd 100644
--- a/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
+++ b/libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp
@@ -10,6 +10,7 @@
#include <pdx/default_transport/client_channel.h>
#include <pdx/default_transport/client_channel_factory.h>
#include <pdx/file_handle.h>
+#include <pdx/trace.h>
#define RETRY_EINTR(fnc_call) \
([&]() -> decltype(fnc_call) { \
@@ -44,17 +45,6 @@
}
}
-// Polls a buffer for the given events, taking care to do the proper
-// translation.
-Status<int> PollEvents(const std::shared_ptr<BufferHubBuffer>& buffer,
- short events) {
- auto poll_status = PollEvents(buffer->event_fd(), events);
- if (!poll_status)
- return poll_status;
-
- return buffer->GetEventMask(poll_status.get());
-}
-
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))};
@@ -115,27 +105,27 @@
default_width_ = queue_info.producer_config.default_width;
default_height_ = queue_info.producer_config.default_height;
default_format_ = queue_info.producer_config.default_format;
- meta_size_ = queue_info.producer_config.meta_size_bytes;
+ user_metadata_size_ = queue_info.producer_config.user_metadata_size;
id_ = queue_info.id;
}
std::unique_ptr<ConsumerQueue> BufferHubQueue::CreateConsumerQueue() {
- if (auto status = CreateConsumerQueueHandle())
+ if (auto status = CreateConsumerQueueHandle(/*silent*/ false))
return std::unique_ptr<ConsumerQueue>(new ConsumerQueue(status.take()));
else
return nullptr;
}
std::unique_ptr<ConsumerQueue> BufferHubQueue::CreateSilentConsumerQueue() {
- if (auto status = CreateConsumerQueueHandle())
- return std::unique_ptr<ConsumerQueue>(
- new ConsumerQueue(status.take(), true));
+ if (auto status = CreateConsumerQueueHandle(/*silent*/ true))
+ return std::unique_ptr<ConsumerQueue>(new ConsumerQueue(status.take()));
else
return nullptr;
}
-Status<LocalChannelHandle> BufferHubQueue::CreateConsumerQueueHandle() {
- auto status = InvokeRemoteMethod<BufferHubRPC::CreateConsumerQueue>();
+Status<LocalChannelHandle> BufferHubQueue::CreateConsumerQueueHandle(
+ bool silent) {
+ auto status = InvokeRemoteMethod<BufferHubRPC::CreateConsumerQueue>(silent);
if (!status) {
ALOGE(
"BufferHubQueue::CreateConsumerQueue: Failed to create consumer queue: "
@@ -148,6 +138,7 @@
}
bool BufferHubQueue::WaitForBuffers(int timeout) {
+ ATRACE_NAME("BufferHubQueue::WaitForBuffers");
std::array<epoll_event, kMaxEvents> events;
// Loop at least once to check for hangups.
@@ -178,13 +169,18 @@
const int num_events = ret;
// A BufferQueue's epoll fd tracks N+1 events, where there are N events,
- // one for each buffer, in the queue and one extra event for the queue
+ // one for each buffer in the queue, and one extra event for the queue
// client itself.
for (int i = 0; i < num_events; i++) {
int32_t event_fd;
int32_t index;
std::tie(event_fd, index) = Unstuff(events[i].data.u64);
+ PDX_TRACE_FORMAT(
+ "epoll_event|queue_id=%d;num_events=%d;event_index=%d;event_fd=%d;"
+ "slot=%d|",
+ id(), num_events, i, event_fd, index);
+
ALOGD_IF(TRACE,
"BufferHubQueue::WaitForBuffers: event %d: event_fd=%d index=%d",
i, event_fd, index);
@@ -208,6 +204,7 @@
Status<void> BufferHubQueue::HandleBufferEvent(size_t slot, int event_fd,
int poll_events) {
+ ATRACE_NAME("BufferHubQueue::HandleBufferEvent");
if (!buffers_[slot]) {
ALOGW("BufferHubQueue::HandleBufferEvent: Invalid buffer slot: %zu", slot);
return ErrorStatus(ENOENT);
@@ -221,58 +218,19 @@
}
const int events = status.get();
+ PDX_TRACE_FORMAT(
+ "buffer|queue_id=%d;buffer_id=%d;slot=%zu;event_fd=%d;poll_events=%x;"
+ "events=%d|",
+ id(), buffers_[slot]->id(), slot, event_fd, poll_events, events);
+
if (events & EPOLLIN) {
- auto entry_status = OnBufferReady(buffers_[slot], slot);
- if (entry_status.ok() || entry_status.error() == EALREADY) {
- // Only enqueue the buffer if it moves to or is already in the state
- // requested in OnBufferReady().
- return Enqueue(entry_status.take());
- } else if (entry_status.error() == EBUSY) {
- // If the buffer is busy this means that the buffer moved from released to
- // posted when a new consumer was created before the ProducerQueue had a
- // chance to regain it. This is a valid transition that we have to handle
- // because edge triggered poll events latch the ready state even if it is
- // later de-asserted -- don't enqueue or print an error log in this case.
- } else {
- ALOGE(
- "BufferHubQueue::HandleBufferEvent: Failed to set buffer ready, "
- "queue_id=%d buffer_id=%d: %s",
- id(), buffers_[slot]->id(), entry_status.GetErrorMessage().c_str());
- }
+ return Enqueue({buffers_[slot], slot, buffers_[slot]->GetQueueIndex()});
} else if (events & EPOLLHUP) {
- // Check to see if the current buffer in the slot hung up. This is a bit of
- // paranoia to deal with the epoll set getting out of sync with the buffer
- // slots.
- auto poll_status = PollEvents(buffers_[slot], POLLIN);
- if (!poll_status && poll_status.error() != ETIMEDOUT) {
- ALOGE("BufferHubQueue::HandleBufferEvent: Failed to poll buffer: %s",
- poll_status.GetErrorMessage().c_str());
- return poll_status.error_status();
- }
-
- const bool hangup_pending = status.ok() && (poll_status.get() & EPOLLHUP);
-
ALOGW(
"BufferHubQueue::HandleBufferEvent: Received EPOLLHUP event: slot=%zu "
- "event_fd=%d buffer_id=%d hangup_pending=%d poll_status=%x",
- slot, buffers_[slot]->event_fd(), buffers_[slot]->id(), hangup_pending,
- poll_status.get());
-
- if (hangup_pending) {
- return RemoveBuffer(slot);
- } else {
- // Clean up the bookkeeping for the event fd. This is a bit of paranoia to
- // deal with the epoll set getting out of sync with the buffer slots.
- // Hitting this path should be very unusual.
- const int ret = epoll_fd_.Control(EPOLL_CTL_DEL, event_fd, nullptr);
- if (ret < 0) {
- ALOGE(
- "BufferHubQueue::HandleBufferEvent: Failed to remove fd=%d from "
- "epoll set: %s",
- event_fd, strerror(-ret));
- return ErrorStatus(-ret);
- }
- }
+ "event_fd=%d buffer_id=%d",
+ slot, buffers_[slot]->event_fd(), buffers_[slot]->id());
+ return RemoveBuffer(slot);
} else {
ALOGW(
"BufferHubQueue::HandleBufferEvent: Unknown event, slot=%zu, epoll "
@@ -284,6 +242,7 @@
}
Status<void> BufferHubQueue::HandleQueueEvent(int poll_event) {
+ ATRACE_NAME("BufferHubQueue::HandleQueueEvent");
auto status = GetEventMask(poll_event);
if (!status) {
ALOGW("BufferHubQueue::HandleQueueEvent: Failed to get event mask: %s",
@@ -330,13 +289,16 @@
return remove_status.error_status();
}
- epoll_event event = {.events = EPOLLIN | EPOLLET,
- .data = {.u64 = Stuff(buffer->event_fd(), slot)}};
- const int ret = epoll_fd_.Control(EPOLL_CTL_ADD, buffer->event_fd(), &event);
- if (ret < 0) {
- ALOGE("BufferHubQueue::AddBuffer: Failed to add buffer to epoll set: %s",
- strerror(-ret));
- return ErrorStatus(-ret);
+ for (const auto& event_source : buffer->GetEventSources()) {
+ epoll_event event = {.events = event_source.event_mask | EPOLLET,
+ .data = {.u64 = Stuff(buffer->event_fd(), slot)}};
+ const int ret =
+ epoll_fd_.Control(EPOLL_CTL_ADD, event_source.event_fd, &event);
+ if (ret < 0) {
+ ALOGE("BufferHubQueue::AddBuffer: Failed to add buffer to epoll set: %s",
+ strerror(-ret));
+ return ErrorStatus(-ret);
+ }
}
buffers_[slot] = buffer;
@@ -348,15 +310,16 @@
ALOGD_IF(TRACE, "BufferHubQueue::RemoveBuffer: slot=%zu", slot);
if (buffers_[slot]) {
- const int ret =
- epoll_fd_.Control(EPOLL_CTL_DEL, buffers_[slot]->event_fd(), nullptr);
- if (ret < 0) {
- ALOGE(
- "BufferHubQueue::RemoveBuffer: Failed to remove buffer from epoll "
- "set: "
- "%s",
- strerror(-ret));
- return ErrorStatus(-ret);
+ for (const auto& event_source : buffers_[slot]->GetEventSources()) {
+ const int ret =
+ epoll_fd_.Control(EPOLL_CTL_DEL, event_source.event_fd, nullptr);
+ if (ret < 0) {
+ ALOGE(
+ "BufferHubQueue::RemoveBuffer: Failed to remove buffer from epoll "
+ "set: %s",
+ strerror(-ret));
+ return ErrorStatus(-ret);
+ }
}
// Trigger OnBufferRemoved callback if registered.
@@ -372,7 +335,7 @@
Status<void> BufferHubQueue::Enqueue(Entry entry) {
if (!is_full()) {
- available_buffers_.Append(std::move(entry));
+ available_buffers_.push(std::move(entry));
// Trigger OnBufferAvailable callback if registered.
if (on_buffer_available_)
@@ -385,25 +348,26 @@
}
}
-Status<std::shared_ptr<BufferHubBuffer>> BufferHubQueue::Dequeue(
- int timeout, size_t* slot, void* meta, LocalHandle* fence) {
+Status<std::shared_ptr<BufferHubBuffer>> BufferHubQueue::Dequeue(int timeout,
+ size_t* slot) {
ALOGD_IF(TRACE, "BufferHubQueue::Dequeue: count=%zu, timeout=%d", count(),
timeout);
- if (!WaitForBuffers(timeout))
- return ErrorStatus(ETIMEDOUT);
+ PDX_TRACE_FORMAT("BufferHubQueue::Dequeue|count=%zu|", count());
- auto& entry = available_buffers_.Front();
+ if (count() == 0) {
+ if (!WaitForBuffers(timeout))
+ return ErrorStatus(ETIMEDOUT);
+ }
+
+ auto& entry = available_buffers_.top();
+ PDX_TRACE_FORMAT("buffer|buffer_id=%d;slot=%zu|", entry.buffer->id(),
+ entry.slot);
std::shared_ptr<BufferHubBuffer> buffer = std::move(entry.buffer);
*slot = entry.slot;
- *fence = std::move(entry.fence);
- if (meta && entry.metadata) {
- std::copy(entry.metadata.get(), entry.metadata.get() + meta_size_,
- reinterpret_cast<uint8_t*>(meta));
- }
- available_buffers_.PopFront();
+ available_buffers_.pop();
return {std::move(buffer)};
}
@@ -419,7 +383,8 @@
pdx::Status<void> BufferHubQueue::FreeAllBuffers() {
// Clear all available buffers.
- available_buffers_.Clear();
+ while (!available_buffers_.empty())
+ available_buffers_.pop();
pdx::Status<void> last_error; // No error.
// Clear all buffers this producer queue is tracking.
@@ -429,7 +394,7 @@
if (!status) {
ALOGE(
"ProducerQueue::FreeAllBuffers: Failed to remove buffer at "
- "slot=%d.",
+ "slot=%zu.",
slot);
last_error = status.error_status();
}
@@ -548,7 +513,7 @@
if (!status)
return status;
- return Enqueue(buffer, slot);
+ return BufferHubQueue::Enqueue({buffer, slot, 0ULL});
}
Status<void> ProducerQueue::RemoveBuffer(size_t slot) {
@@ -565,40 +530,33 @@
Status<std::shared_ptr<BufferProducer>> ProducerQueue::Dequeue(
int timeout, size_t* slot, LocalHandle* release_fence) {
- if (slot == nullptr || release_fence == nullptr) {
- ALOGE("ProducerQueue::Dequeue: Invalid parameter: slot=%p release_fence=%p",
- slot, release_fence);
+ DvrNativeBufferMetadata canonical_meta;
+ return Dequeue(timeout, slot, &canonical_meta, release_fence);
+}
+
+pdx::Status<std::shared_ptr<BufferProducer>> ProducerQueue::Dequeue(
+ int timeout, size_t* slot, DvrNativeBufferMetadata* out_meta,
+ pdx::LocalHandle* release_fence) {
+ ATRACE_NAME("ProducerQueue::Dequeue");
+ if (slot == nullptr || out_meta == nullptr || release_fence == nullptr) {
+ ALOGE("ProducerQueue::Dequeue: Invalid parameter.");
return ErrorStatus(EINVAL);
}
- auto buffer_status =
- BufferHubQueue::Dequeue(timeout, slot, nullptr, release_fence);
- if (!buffer_status)
- return buffer_status.error_status();
+ auto status = BufferHubQueue::Dequeue(timeout, slot);
+ if (!status)
+ return status.error_status();
- return {std::static_pointer_cast<BufferProducer>(buffer_status.take())};
-}
-
-Status<BufferHubQueue::Entry> ProducerQueue::OnBufferReady(
- const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot) {
- ALOGD_IF(TRACE,
- "ProducerQueue::OnBufferReady: queue_id=%d buffer_id=%d slot=%zu",
- id(), buffer->id(), slot);
-
- // Avoid taking a transient reference, buffer is valid for the duration of
- // this method call.
- auto* producer_buffer = static_cast<BufferProducer*>(buffer.get());
- LocalHandle release_fence;
-
- const int ret = producer_buffer->Gain(&release_fence);
- if (ret < 0)
+ auto buffer = std::static_pointer_cast<BufferProducer>(status.take());
+ const int ret = buffer->GainAsync(out_meta, release_fence);
+ if (ret < 0 && ret != -EALREADY)
return ErrorStatus(-ret);
- else
- return {{buffer, nullptr, std::move(release_fence), slot}};
+
+ return {std::move(buffer)};
}
-ConsumerQueue::ConsumerQueue(LocalChannelHandle handle, bool ignore_on_import)
- : BufferHubQueue(std::move(handle)), ignore_on_import_(ignore_on_import) {
+ConsumerQueue::ConsumerQueue(LocalChannelHandle handle)
+ : BufferHubQueue(std::move(handle)) {
auto status = ImportQueue();
if (!status) {
ALOGE("ConsumerQueue::ConsumerQueue: Failed to import queue: %s",
@@ -619,9 +577,17 @@
Status<size_t> ConsumerQueue::ImportBuffers() {
auto status = InvokeRemoteMethod<BufferHubRPC::ConsumerQueueImportBuffers>();
if (!status) {
- ALOGE("ConsumerQueue::ImportBuffers: Failed to import consumer buffer: %s",
+ if (status.error() == EBADR) {
+ ALOGI(
+ "ConsumerQueue::ImportBuffers: Queue is silent, no buffers "
+ "imported.");
+ return {0};
+ } else {
+ ALOGE(
+ "ConsumerQueue::ImportBuffers: Failed to import consumer buffer: %s",
status.GetErrorMessage().c_str());
- return status.error_status();
+ return status.error_status();
+ }
}
int ret;
@@ -642,22 +608,6 @@
continue;
}
- // Setup ignore state before adding buffer to the queue.
- if (ignore_on_import_) {
- ALOGD_IF(TRACE,
- "ConsumerQueue::ImportBuffers: Setting buffer to ignored state: "
- "buffer_id=%d",
- buffer_consumer->id());
- ret = buffer_consumer->SetIgnore(true);
- if (ret < 0) {
- ALOGE(
- "ConsumerQueue::ImportBuffers: Failed to set ignored state on "
- "imported buffer buffer_id=%d: %s",
- buffer_consumer->id(), strerror(-ret));
- last_error = ErrorStatus(-ret);
- }
- }
-
auto add_status =
AddBuffer(std::move(buffer_consumer), buffer_handle_slot.second);
if (!add_status) {
@@ -685,7 +635,7 @@
// 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 and edge transition when adding new buffers to the epoll
+ // 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);
@@ -703,51 +653,53 @@
}
Status<std::shared_ptr<BufferConsumer>> ConsumerQueue::Dequeue(
- int timeout, size_t* slot, void* meta, size_t meta_size,
+ int timeout, size_t* slot, void* meta, size_t user_metadata_size,
LocalHandle* acquire_fence) {
- if (meta_size != meta_size_) {
+ if (user_metadata_size != user_metadata_size_) {
ALOGE(
"ConsumerQueue::Dequeue: Metadata size (%zu) for the dequeuing buffer "
"does not match metadata size (%zu) for the queue.",
- meta_size, meta_size_);
+ user_metadata_size, user_metadata_size_);
return ErrorStatus(EINVAL);
}
- if (slot == nullptr || acquire_fence == nullptr) {
- ALOGE(
- "ConsumerQueue::Dequeue: Invalid parameter: slot=%p meta=%p "
- "acquire_fence=%p",
- slot, meta, acquire_fence);
- return ErrorStatus(EINVAL);
+ DvrNativeBufferMetadata canonical_meta;
+ auto status = Dequeue(timeout, slot, &canonical_meta, acquire_fence);
+ if (!status)
+ return status.error_status();
+
+ if (meta && user_metadata_size) {
+ void* metadata_src =
+ reinterpret_cast<void*>(canonical_meta.user_metadata_ptr);
+ if (metadata_src) {
+ memcpy(meta, metadata_src, user_metadata_size);
+ } else {
+ ALOGW("ConsumerQueue::Dequeue: no user-defined metadata.");
+ }
}
- auto buffer_status =
- BufferHubQueue::Dequeue(timeout, slot, meta, acquire_fence);
- if (!buffer_status)
- return buffer_status.error_status();
-
- return {std::static_pointer_cast<BufferConsumer>(buffer_status.take())};
+ return status;
}
-Status<BufferHubQueue::Entry> ConsumerQueue::OnBufferReady(
- const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot) {
- ALOGD_IF(TRACE,
- "ConsumerQueue::OnBufferReady: queue_id=%d buffer_id=%d slot=%zu",
- id(), buffer->id(), slot);
+Status<std::shared_ptr<BufferConsumer>> ConsumerQueue::Dequeue(
+ int timeout, size_t* slot, DvrNativeBufferMetadata* out_meta,
+ pdx::LocalHandle* acquire_fence) {
+ ATRACE_NAME("ConsumerQueue::Dequeue");
+ if (slot == nullptr || out_meta == nullptr || acquire_fence == nullptr) {
+ ALOGE("ConsumerQueue::Dequeue: Invalid parameter.");
+ return ErrorStatus(EINVAL);
+ }
- // Avoid taking a transient reference, buffer is valid for the duration of
- // this method call.
- auto* consumer_buffer = static_cast<BufferConsumer*>(buffer.get());
- std::unique_ptr<uint8_t[]> metadata(meta_size_ ? new uint8_t[meta_size_]
- : nullptr);
- LocalHandle acquire_fence;
+ auto status = BufferHubQueue::Dequeue(timeout, slot);
+ if (!status)
+ return status.error_status();
- const int ret =
- consumer_buffer->Acquire(&acquire_fence, metadata.get(), meta_size_);
+ auto buffer = std::static_pointer_cast<BufferConsumer>(status.take());
+ const int ret = buffer->AcquireAsync(out_meta, acquire_fence);
if (ret < 0)
return ErrorStatus(-ret);
- else
- return {{buffer, std::move(metadata), std::move(acquire_fence), slot}};
+
+ return {std::move(buffer)};
}
Status<void> ConsumerQueue::OnBufferAllocated() {
diff --git a/libs/vr/libbufferhubqueue/buffer_hub_queue_producer.cpp b/libs/vr/libbufferhubqueue/buffer_hub_queue_producer.cpp
index 53eed89..221bc4f 100644
--- a/libs/vr/libbufferhubqueue/buffer_hub_queue_producer.cpp
+++ b/libs/vr/libbufferhubqueue/buffer_hub_queue_producer.cpp
@@ -328,7 +328,7 @@
LocalHandle fence_fd(fence->isValid() ? fence->dup() : -1);
- DvrNativeBufferMetadata meta_data = {};
+ DvrNativeBufferMetadata meta_data;
meta_data.timestamp = timestamp;
meta_data.is_auto_timestamp = static_cast<int32_t>(is_auto_timestamp);
meta_data.dataspace = static_cast<int32_t>(dataspace);
@@ -339,7 +339,7 @@
meta_data.scaling_mode = static_cast<int32_t>(scaling_mode);
meta_data.transform = static_cast<int32_t>(transform);
- buffer_producer->Post(fence_fd, &meta_data, sizeof(meta_data));
+ buffer_producer->PostAsync(&meta_data, fence_fd);
buffers_[slot].mBufferState.queue();
output->width = buffer_producer->width();
@@ -384,7 +384,7 @@
}
auto buffer_producer = buffers_[slot].mBufferProducer;
- queue_->Enqueue(buffer_producer, slot);
+ queue_->Enqueue(buffer_producer, slot, 0ULL);
buffers_[slot].mBufferState.cancel();
buffers_[slot].mFence = fence;
ALOGD_IF(TRACE, "cancelBuffer: slot %d", slot);
diff --git a/libs/vr/libbufferhubqueue/include/private/dvr/buffer_hub_queue_client.h b/libs/vr/libbufferhubqueue/include/private/dvr/buffer_hub_queue_client.h
index 0699fef..6962d6c 100644
--- a/libs/vr/libbufferhubqueue/include/private/dvr/buffer_hub_queue_client.h
+++ b/libs/vr/libbufferhubqueue/include/private/dvr/buffer_hub_queue_client.h
@@ -5,12 +5,13 @@
#include <pdx/client.h>
#include <pdx/status.h>
-#include <private/dvr/bufferhub_rpc.h>
#include <private/dvr/buffer_hub_client.h>
+#include <private/dvr/bufferhub_rpc.h>
#include <private/dvr/epoll_file_descriptor.h>
#include <private/dvr/ring_buffer.h>
#include <memory>
+#include <queue>
#include <vector>
namespace android {
@@ -50,19 +51,22 @@
uint32_t default_format() const { return default_format_; }
// Creates a new consumer in handle form for immediate transport over RPC.
- pdx::Status<pdx::LocalChannelHandle> CreateConsumerQueueHandle();
+ pdx::Status<pdx::LocalChannelHandle> CreateConsumerQueueHandle(
+ bool silent = false);
// Returns the number of buffers avaiable for dequeue.
- size_t count() const { return available_buffers_.GetSize(); }
+ size_t count() const { return available_buffers_.size(); }
// Returns the total number of buffers that the queue is tracking.
size_t capacity() const { return capacity_; }
// Returns the size of metadata structure associated with this queue.
- size_t metadata_size() const { return meta_size_; }
+ size_t metadata_size() const { return user_metadata_size_; }
// Returns whether the buffer queue is full.
- bool is_full() const { return available_buffers_.IsFull(); }
+ bool is_full() const {
+ return available_buffers_.size() >= kMaxQueueCapacity;
+ }
explicit operator bool() const { return epoll_fd_.IsValid(); }
@@ -136,8 +140,8 @@
// block. Specifying a timeout of -1 causes Dequeue() to block indefinitely,
// while specifying a timeout equal to zero cause Dequeue() to return
// immediately, even if no buffers are available.
- pdx::Status<std::shared_ptr<BufferHubBuffer>> Dequeue(
- int timeout, size_t* slot, void* meta, pdx::LocalHandle* fence);
+ pdx::Status<std::shared_ptr<BufferHubBuffer>> Dequeue(int timeout,
+ size_t* slot);
// Waits for buffers to become available and adds them to the available queue.
bool WaitForBuffers(int timeout);
@@ -150,8 +154,9 @@
// per-buffer data.
struct Entry {
Entry() : slot(0) {}
- Entry(const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot)
- : buffer(buffer), slot(slot) {}
+ Entry(const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot,
+ uint64_t index)
+ : buffer(buffer), slot(slot), index(index) {}
Entry(const std::shared_ptr<BufferHubBuffer>& buffer,
std::unique_ptr<uint8_t[]> metadata, pdx::LocalHandle fence,
size_t slot)
@@ -166,20 +171,24 @@
std::unique_ptr<uint8_t[]> metadata;
pdx::LocalHandle fence;
size_t slot;
+ uint64_t index;
+ };
+
+ struct EntryComparator {
+ bool operator()(const Entry& lhs, const Entry& rhs) {
+ return lhs.index > rhs.index;
+ }
};
// Enqueues a buffer to the available list (Gained for producer or Acquireed
// for consumer).
pdx::Status<void> Enqueue(Entry entry);
- virtual pdx::Status<Entry> OnBufferReady(
- const std::shared_ptr<BufferHubBuffer>& buf, size_t slot) = 0;
-
// Called when a buffer is allocated remotely.
virtual pdx::Status<void> OnBufferAllocated() { return {}; }
// Size of the metadata that buffers in this queue cary.
- size_t meta_size_{0};
+ size_t user_metadata_size_{0};
private:
void Initialize();
@@ -226,7 +235,9 @@
std::array<std::shared_ptr<BufferHubBuffer>, kMaxQueueCapacity> buffers_;
// Buffers and related data that are available for dequeue.
- RingBuffer<Entry> available_buffers_{kMaxQueueCapacity};
+ // RingBuffer<Entry> available_buffers_{kMaxQueueCapacity};
+ std::priority_queue<Entry, std::vector<Entry>, EntryComparator>
+ available_buffers_;
// Keeps track with how many buffers have been added into the queue.
size_t capacity_{0};
@@ -316,11 +327,14 @@
// to the consumer side.
pdx::Status<std::shared_ptr<BufferProducer>> Dequeue(
int timeout, size_t* slot, pdx::LocalHandle* release_fence);
+ pdx::Status<std::shared_ptr<BufferProducer>> Dequeue(
+ int timeout, size_t* slot, DvrNativeBufferMetadata* out_meta,
+ pdx::LocalHandle* release_fence);
// Enqueues a producer buffer in the queue.
pdx::Status<void> Enqueue(const std::shared_ptr<BufferProducer>& buffer,
- size_t slot) {
- return BufferHubQueue::Enqueue({buffer, slot});
+ size_t slot, uint64_t index) {
+ return BufferHubQueue::Enqueue({buffer, slot, index});
}
private:
@@ -331,9 +345,6 @@
// arguments as the constructors.
explicit ProducerQueue(pdx::LocalChannelHandle handle);
ProducerQueue(const ProducerQueueConfig& config, const UsagePolicy& usage);
-
- pdx::Status<Entry> OnBufferReady(
- const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot) override;
};
class ConsumerQueue : public BufferHubQueue {
@@ -352,10 +363,9 @@
// used to avoid participation in the buffer lifecycle by a consumer queue
// that is only used to spawn other consumer queues, such as in an
// intermediate service.
- static std::unique_ptr<ConsumerQueue> Import(pdx::LocalChannelHandle handle,
- bool ignore_on_import = false) {
+ static std::unique_ptr<ConsumerQueue> Import(pdx::LocalChannelHandle handle) {
return std::unique_ptr<ConsumerQueue>(
- new ConsumerQueue(std::move(handle), ignore_on_import));
+ new ConsumerQueue(std::move(handle)));
}
// Import newly created buffers from the service side.
@@ -379,13 +389,16 @@
}
pdx::Status<std::shared_ptr<BufferConsumer>> Dequeue(
- int timeout, size_t* slot, void* meta, size_t meta_size,
+ int timeout, size_t* slot, void* meta, size_t user_metadata_size,
+ pdx::LocalHandle* acquire_fence);
+ pdx::Status<std::shared_ptr<BufferConsumer>> Dequeue(
+ int timeout, size_t* slot, DvrNativeBufferMetadata* out_meta,
pdx::LocalHandle* acquire_fence);
private:
friend BufferHubQueue;
- ConsumerQueue(pdx::LocalChannelHandle handle, bool ignore_on_import = false);
+ ConsumerQueue(pdx::LocalChannelHandle handle);
// Add a consumer buffer to populate the queue. Once added, a consumer buffer
// is NOT available to use until the producer side |Post| it. |WaitForBuffers|
@@ -394,14 +407,7 @@
pdx::Status<void> AddBuffer(const std::shared_ptr<BufferConsumer>& buffer,
size_t slot);
- pdx::Status<Entry> OnBufferReady(
- const std::shared_ptr<BufferHubBuffer>& buffer, size_t slot) override;
-
pdx::Status<void> OnBufferAllocated() override;
-
- // Flag indicating that imported (consumer) buffers should be ignored when
- // imported to avoid participating in the buffer ownership flow.
- bool ignore_on_import_;
};
} // namespace dvr
diff --git a/libs/vr/libbufferhubqueue/tests/Android.bp b/libs/vr/libbufferhubqueue/tests/Android.bp
index 865573c..8bd1ef1 100644
--- a/libs/vr/libbufferhubqueue/tests/Android.bp
+++ b/libs/vr/libbufferhubqueue/tests/Android.bp
@@ -1,4 +1,7 @@
+header_libraries = [
+ "libdvr_headers",
+]
shared_libraries = [
"libbase",
@@ -21,6 +24,7 @@
cc_test {
srcs: ["buffer_hub_queue-test.cpp"],
+ header_libs: header_libraries,
static_libs: static_libraries,
shared_libs: shared_libraries,
cflags: [
@@ -35,6 +39,7 @@
cc_test {
srcs: ["buffer_hub_queue_producer-test.cpp"],
+ header_libs: header_libraries,
static_libs: static_libraries,
shared_libs: shared_libraries,
cflags: [
diff --git a/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp b/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
index 7581a06..8a72531 100644
--- a/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
+++ b/libs/vr/libbufferhubqueue/tests/buffer_hub_queue-test.cpp
@@ -3,6 +3,8 @@
#include <private/dvr/buffer_hub_queue_client.h>
#include <gtest/gtest.h>
+#include <poll.h>
+#include <sys/eventfd.h>
#include <vector>
@@ -46,9 +48,9 @@
void AllocateBuffer(size_t* slot_out = nullptr) {
// Create producer buffer.
- auto status = producer_queue_->AllocateBuffer(
- kBufferWidth, kBufferHeight, kBufferLayerCount, kBufferFormat,
- kBufferUsage);
+ auto status = producer_queue_->AllocateBuffer(kBufferWidth, kBufferHeight,
+ kBufferLayerCount,
+ kBufferFormat, kBufferUsage);
ASSERT_TRUE(status.ok());
size_t slot = status.take();
@@ -56,6 +58,23 @@
*slot_out = slot;
}
+ bool WaitAndHandleOnce(BufferHubQueue* queue, int timeout_ms) {
+ pollfd pfd{queue->queue_fd(), POLLIN, 0};
+ int ret;
+ do {
+ ret = poll(&pfd, 1, timeout_ms);
+ } while (ret == -1 && errno == EINTR);
+
+ if (ret < 0) {
+ ALOGW("Failed to poll queue %d's event fd, error: %s.", queue->id(),
+ strerror(errno));
+ return false;
+ } else if (ret == 0) {
+ return false;
+ }
+ return queue->HandleQueueEvents();
+ }
+
protected:
ProducerQueueConfigBuilder config_builder_;
std::unique_ptr<ProducerQueue> producer_queue_;
@@ -75,7 +94,7 @@
for (size_t i = 0; i < nb_dequeue_times; i++) {
size_t slot;
LocalHandle fence;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_NE(nullptr, p1);
@@ -113,31 +132,26 @@
// Dequeue returns timeout since no buffer is ready to consumer, but
// this implicitly triggers buffer import and bump up |capacity|.
LocalHandle fence;
- auto status = consumer_queue_->Dequeue(0, &slot, &seq, &fence);
+ auto status = consumer_queue_->Dequeue(100, &slot, &seq, &fence);
ASSERT_FALSE(status.ok());
ASSERT_EQ(ETIMEDOUT, status.error());
ASSERT_EQ(consumer_queue_->capacity(), i + 1);
}
- // Use /dev/zero as a stand-in for a fence. As long as BufferHub does not need
- // to merge fences, which only happens when multiple consumers release the
- // same buffer with release fences, the file object should simply pass
- // through.
- LocalHandle post_fence("/dev/zero", O_RDONLY);
- struct stat post_fence_stat;
- ASSERT_EQ(0, fstat(post_fence.Get(), &post_fence_stat));
+ // Use eventfd as a stand-in for a fence.
+ LocalHandle post_fence(eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK));
for (size_t i = 0; i < kBufferCount; i++) {
LocalHandle fence;
// First time there is no buffer available to dequeue.
- auto consumer_status = consumer_queue_->Dequeue(0, &slot, &seq, &fence);
+ auto consumer_status = consumer_queue_->Dequeue(100, &slot, &seq, &fence);
ASSERT_FALSE(consumer_status.ok());
ASSERT_EQ(ETIMEDOUT, consumer_status.error());
// Make sure Producer buffer is POSTED so that it's ready to Accquire
// in the consumer's Dequeue() function.
- auto producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
auto producer = producer_status.take();
ASSERT_NE(nullptr, producer);
@@ -147,20 +161,10 @@
// Second time the just the POSTED buffer should be dequeued.
uint64_t seq_out = 0;
- consumer_status = consumer_queue_->Dequeue(0, &slot, &seq_out, &fence);
+ consumer_status = consumer_queue_->Dequeue(100, &slot, &seq_out, &fence);
ASSERT_TRUE(consumer_status.ok());
EXPECT_TRUE(fence.IsValid());
- struct stat acquire_fence_stat;
- ASSERT_EQ(0, fstat(fence.Get(), &acquire_fence_stat));
-
- // The file descriptors should refer to the same file object. Testing the
- // device id and inode is a proxy for testing that the fds refer to the same
- // file object.
- EXPECT_NE(post_fence.Get(), fence.Get());
- EXPECT_EQ(post_fence_stat.st_dev, acquire_fence_stat.st_dev);
- EXPECT_EQ(post_fence_stat.st_ino, acquire_fence_stat.st_ino);
-
auto consumer = consumer_status.take();
ASSERT_NE(nullptr, consumer);
ASSERT_EQ(seq_in, seq_out);
@@ -196,12 +200,11 @@
for (size_t i = 0; i < kBufferCount; i++) {
Entry* entry = &buffers[i];
- auto producer_status =
- producer_queue_->Dequeue(0, &entry->slot, &entry->fence);
+ auto producer_status = producer_queue_->Dequeue(
+ /*timeout_ms=*/100, &entry->slot, &entry->fence);
ASSERT_TRUE(producer_status.ok());
entry->buffer = producer_status.take();
ASSERT_NE(nullptr, entry->buffer);
- EXPECT_EQ(i, entry->slot);
}
// Remove a buffer and make sure both queues reflect the change.
@@ -218,8 +221,8 @@
buffers[0].buffer = nullptr;
// Now the consumer queue should know it's gone.
- EXPECT_FALSE(consumer_queue_->HandleQueueEvents());
- EXPECT_EQ(kBufferCount - 1, consumer_queue_->capacity());
+ EXPECT_FALSE(WaitAndHandleOnce(consumer_queue_.get(), /*timeout_ms=*/100));
+ ASSERT_EQ(kBufferCount - 1, consumer_queue_->capacity());
// Allocate a new buffer. This should take the first empty slot.
size_t slot;
@@ -286,17 +289,20 @@
auto silent_queue = producer_queue_->CreateSilentConsumerQueue();
ASSERT_NE(nullptr, silent_queue);
- // Check that buffers are correctly imported on construction.
- EXPECT_EQ(kBufferCount, silent_queue->capacity());
+ // Check that silent queue doesn't import buffers on creation.
+ EXPECT_EQ(0, silent_queue->capacity());
// Dequeue and post a buffer.
size_t slot;
LocalHandle fence;
- auto producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto producer_status =
+ producer_queue_->Dequeue(/*timeout_ms=*/100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
auto producer_buffer = producer_status.take();
ASSERT_NE(nullptr, producer_buffer);
ASSERT_EQ(0, producer_buffer->Post<void>({}));
+ // After post, check the number of remaining available buffers.
+ EXPECT_EQ(kBufferCount - 1, producer_queue_->count());
// Currently we expect no buffer to be available prior to calling
// WaitForBuffers/HandleQueueEvents.
@@ -314,23 +320,30 @@
EXPECT_EQ(1u, consumer_queue_->count());
// Reclaim released/ignored buffers.
- producer_queue_->HandleQueueEvents();
+ ASSERT_EQ(kBufferCount - 1, producer_queue_->count());
+
+ usleep(10000);
+ WaitAndHandleOnce(producer_queue_.get(), /*timeout_ms=*/100);
ASSERT_EQ(kBufferCount - 1, producer_queue_->count());
// Post another buffer.
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(/*timeout_ms=*/100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
producer_buffer = producer_status.take();
ASSERT_NE(nullptr, producer_buffer);
ASSERT_EQ(0, producer_buffer->Post<void>({}));
// Verify that the consumer queue receives it.
- EXPECT_EQ(1u, consumer_queue_->count());
- EXPECT_TRUE(consumer_queue_->HandleQueueEvents());
- EXPECT_EQ(2u, consumer_queue_->count());
+ size_t consumer_queue_count = consumer_queue_->count();
+ WaitAndHandleOnce(consumer_queue_.get(), /*timeout_ms=*/100);
+ EXPECT_LT(consumer_queue_count, consumer_queue_->count());
+
+ // Save the current consumer queue buffer count to compare after the dequeue.
+ consumer_queue_count = consumer_queue_->count();
// Dequeue and acquire/release (discard) buffers on the consumer end.
- auto consumer_status = consumer_queue_->Dequeue(0, &slot, &fence);
+ auto consumer_status =
+ consumer_queue_->Dequeue(/*timeout_ms=*/100, &slot, &fence);
ASSERT_TRUE(consumer_status.ok());
auto consumer_buffer = consumer_status.take();
ASSERT_NE(nullptr, consumer_buffer);
@@ -338,7 +351,7 @@
// Buffer should be returned to the producer queue without being handled by
// the silent consumer queue.
- EXPECT_EQ(1u, consumer_queue_->count());
+ EXPECT_GT(consumer_queue_count, consumer_queue_->count());
EXPECT_EQ(kBufferCount - 2, producer_queue_->count());
EXPECT_TRUE(producer_queue_->HandleQueueEvents());
EXPECT_EQ(kBufferCount - 1, producer_queue_->count());
@@ -362,13 +375,13 @@
for (auto mi : ms) {
size_t slot;
LocalHandle fence;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_NE(nullptr, p1);
ASSERT_EQ(p1->Post(LocalHandle(-1), &mi, sizeof(mi)), 0);
TestMetadata mo;
- auto c1_status = consumer_queue_->Dequeue(0, &slot, &mo, &fence);
+ auto c1_status = consumer_queue_->Dequeue(100, &slot, &mo, &fence);
ASSERT_TRUE(c1_status.ok());
auto c1 = c1_status.take();
ASSERT_EQ(mi.a, mo.a);
@@ -387,7 +400,7 @@
int64_t mi = 3;
size_t slot;
LocalHandle fence;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_NE(nullptr, p1);
@@ -395,7 +408,7 @@
int32_t mo;
// Acquire a buffer with mismatched metadata is not OK.
- auto c1_status = consumer_queue_->Dequeue(0, &slot, &mo, &fence);
+ auto c1_status = consumer_queue_->Dequeue(100, &slot, &mo, &fence);
ASSERT_FALSE(c1_status.ok());
}
@@ -406,14 +419,14 @@
size_t slot;
LocalHandle fence;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_NE(nullptr, p1);
int64_t mo;
- producer_queue_->Enqueue(p1, slot);
- auto c1_status = consumer_queue_->Dequeue(0, &slot, &mo, &fence);
+ producer_queue_->Enqueue(p1, slot, 0ULL);
+ auto c1_status = consumer_queue_->Dequeue(100, &slot, &mo, &fence);
ASSERT_FALSE(c1_status.ok());
}
@@ -424,14 +437,14 @@
size_t s1;
AllocateBuffer();
LocalHandle fence;
- auto p1_status = producer_queue_->Dequeue(0, &s1, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &s1, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_NE(nullptr, p1);
// producer queue is exhausted
size_t s2;
- auto p2_status = producer_queue_->Dequeue(0, &s2, &fence);
+ auto p2_status = producer_queue_->Dequeue(100, &s2, &fence);
ASSERT_FALSE(p2_status.ok());
ASSERT_EQ(ETIMEDOUT, p2_status.error());
@@ -441,7 +454,7 @@
ASSERT_EQ(producer_queue_->capacity(), 2U);
// now we can dequeue again
- p2_status = producer_queue_->Dequeue(0, &s2, &fence);
+ p2_status = producer_queue_->Dequeue(100, &s2, &fence);
ASSERT_TRUE(p2_status.ok());
auto p2 = p2_status.take();
ASSERT_NE(nullptr, p2);
@@ -456,7 +469,7 @@
int64_t seq = 1;
ASSERT_EQ(p1->Post(LocalHandle(), seq), 0);
size_t cs1, cs2;
- auto c1_status = consumer_queue_->Dequeue(0, &cs1, &seq, &fence);
+ auto c1_status = consumer_queue_->Dequeue(100, &cs1, &seq, &fence);
ASSERT_TRUE(c1_status.ok());
auto c1 = c1_status.take();
ASSERT_NE(nullptr, c1);
@@ -465,7 +478,7 @@
ASSERT_EQ(cs1, s1);
ASSERT_EQ(p2->Post(LocalHandle(), seq), 0);
- auto c2_status = consumer_queue_->Dequeue(0, &cs2, &seq, &fence);
+ auto c2_status = consumer_queue_->Dequeue(100, &cs2, &seq, &fence);
ASSERT_TRUE(c2_status.ok());
auto c2 = c2_status.take();
ASSERT_NE(nullptr, c2);
@@ -485,7 +498,7 @@
LocalHandle fence;
size_t slot;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_EQ(p1->usage() & set_mask, set_mask);
@@ -504,7 +517,7 @@
LocalHandle fence;
size_t slot;
- auto p1_status = producer_queue_->Dequeue(0, &slot, &fence);
+ auto p1_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(p1_status.ok());
auto p1 = p1_status.take();
ASSERT_EQ(0u, p1->usage() & clear_mask);
@@ -543,9 +556,9 @@
ASSERT_TRUE(status.ok());
// While allocation without those bits should fail.
- status = producer_queue_->AllocateBuffer(
- kBufferWidth, kBufferHeight, kBufferLayerCount, kBufferFormat,
- kBufferUsage & ~deny_clear_mask);
+ status = producer_queue_->AllocateBuffer(kBufferWidth, kBufferHeight,
+ kBufferLayerCount, kBufferFormat,
+ kBufferUsage & ~deny_clear_mask);
ASSERT_FALSE(status.ok());
ASSERT_EQ(EINVAL, status.error());
}
@@ -603,7 +616,7 @@
// Free all buffers when one buffer is dequeued.
CHECK_NO_BUFFER_THEN_ALLOCATE(kBufferCount);
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
status = producer_queue_->FreeAllBuffers();
EXPECT_TRUE(status.ok());
@@ -611,7 +624,7 @@
// Free all buffers when all buffers are dequeued.
CHECK_NO_BUFFER_THEN_ALLOCATE(kBufferCount);
for (size_t i = 0; i < kBufferCount; i++) {
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
}
status = producer_queue_->FreeAllBuffers();
@@ -619,7 +632,7 @@
// Free all buffers when one buffer is posted.
CHECK_NO_BUFFER_THEN_ALLOCATE(kBufferCount);
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
producer_buffer = producer_status.take();
ASSERT_NE(nullptr, producer_buffer);
@@ -630,7 +643,7 @@
// Free all buffers when all buffers are posted.
CHECK_NO_BUFFER_THEN_ALLOCATE(kBufferCount);
for (size_t i = 0; i < kBufferCount; i++) {
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
producer_buffer = producer_status.take();
ASSERT_NE(nullptr, producer_buffer);
@@ -642,12 +655,12 @@
// Free all buffers when all buffers are acquired.
CHECK_NO_BUFFER_THEN_ALLOCATE(kBufferCount);
for (size_t i = 0; i < kBufferCount; i++) {
- producer_status = producer_queue_->Dequeue(0, &slot, &fence);
+ producer_status = producer_queue_->Dequeue(100, &slot, &fence);
ASSERT_TRUE(producer_status.ok());
producer_buffer = producer_status.take();
ASSERT_NE(nullptr, producer_buffer);
ASSERT_EQ(0, producer_buffer->Post(fence, &seq, sizeof(seq)));
- consumer_status = consumer_queue_->Dequeue(0, &slot, &seq, &fence);
+ consumer_status = consumer_queue_->Dequeue(100, &slot, &seq, &fence);
ASSERT_TRUE(consumer_status.ok());
}