Merge changes from topic 'vrflinger-prep' into oc-dev am: 5bbb45e038 (8ff501c5) · Commits · e / os / platform_frameworks_native

libs/vr/libbufferhub/include/private/dvr/buffer_hub_client.h

+5 −3

Original line number	Diff line number	Diff line
		@@ -94,9 +94,12 @@ class BufferHubBuffer : public pdx::Client {
		}

		IonBuffer* buffer() { return &slices_[0]; }
		const IonBuffer* buffer() const { return &slices_[0]; }

		// If index is greater than or equal to slice_count(), the result is
		// undefined.
		IonBuffer* slice(size_t index) { return &slices_[index]; }
		const IonBuffer* slice(size_t index) const { return &slices_[index]; }

		int slice_count() const { return static_cast<int>(slices_.size()); }
		int id() const { return id_; }
		@@ -171,9 +174,8 @@ class BufferProducer : public pdx::ClientBase<BufferProducer, BufferHubBuffer> {
		int Post(const LocalHandle& ready_fence) {
		return Post(ready_fence, nullptr, 0);
		}
		template <
		typename Meta,
		typename = typename std::enable_if<!std::is_void<Meta>::value>::type>
		template <typename Meta, typename = typename std::enable_if<
		!std::is_void<Meta>::value>::type>
		int Post(const LocalHandle& ready_fence, const Meta& meta) {
		return Post(ready_fence, &meta, sizeof(meta));
		}

libs/vr/libbufferhub/include/private/dvr/bufferhub_rpc.h

+15 −5

Original line number	Diff line number	Diff line
		@@ -119,6 +119,14 @@ class FenceHandle {
		using LocalFence = FenceHandle<pdx::LocalHandle>;
		using BorrowedFence = FenceHandle<pdx::BorrowedHandle>;

		struct QueueInfo {
		size_t meta_size_bytes;
		int id;

		private:
		PDX_SERIALIZABLE_MEMBERS(QueueInfo, meta_size_bytes, id);
		};

		// BufferHub Service RPC interface. Defines the endpoints, op codes, and method
		// type signatures supported by bufferhubd.
		struct BufferHubRPC {
		@@ -151,6 +159,7 @@ struct BufferHubRPC {
		kOpConsumerSetIgnore,
		kOpCreateProducerQueue,
		kOpCreateConsumerQueue,
		kOpGetQueueInfo,
		kOpProducerQueueAllocateBuffers,
		kOpProducerQueueDetachBuffer,
		kOpConsumerQueueImportBuffers,
		@@ -192,18 +201,19 @@ struct BufferHubRPC {

		// Buffer Queue Methods.
		PDX_REMOTE_METHOD(CreateProducerQueue, kOpCreateProducerQueue,
		int(size_t meta_size_bytes, int usage_set_mask,
		QueueInfo(size_t meta_size_bytes, int usage_set_mask,
		int usage_clear_mask, int usage_deny_set_mask,
		int usage_deny_clear_mask));
		PDX_REMOTE_METHOD(CreateConsumerQueue, kOpCreateConsumerQueue,
		std::pair<LocalChannelHandle, size_t>(Void));
		LocalChannelHandle(Void));
		PDX_REMOTE_METHOD(GetQueueInfo, kOpGetQueueInfo, QueueInfo(Void));
		PDX_REMOTE_METHOD(ProducerQueueAllocateBuffers,
		kOpProducerQueueAllocateBuffers,
		std::vector<std::pair<LocalChannelHandle, size_t>>(
		int width, int height, int format, int usage,
		size_t slice_count, size_t buffer_count));
		PDX_REMOTE_METHOD(ProducerQueueDetachBuffer, kOpProducerQueueDetachBuffer,
		int(size_t slot));
		void(size_t slot));
		PDX_REMOTE_METHOD(ConsumerQueueImportBuffers, kOpConsumerQueueImportBuffers,
		std::vector<std::pair<LocalChannelHandle, size_t>>(Void));
		};

libs/vr/libbufferhub/include/private/dvr/ion_buffer.h

+13 −14

Original line number	Diff line number	Diff line
		@@ -60,21 +60,20 @@ class IonBuffer {
		int LockYUV(int usage, int x, int y, int width, int height,
		struct android_ycbcr* yuv);
		int Unlock();
		buffer_handle_t handle() const { if (buffer_.get()) return buffer_->handle;
		else return nullptr; }
		int width() const { if (buffer_.get()) return buffer_->getWidth();
		else return 0; }
		int height() const { if (buffer_.get()) return buffer_->getHeight();
		else return 0; }
		int layer_count() const { if (buffer_.get()) return buffer_->getLayerCount();
		else return 0; }
		int stride() const { if (buffer_.get()) return buffer_->getStride();
		else return 0; }

		const sp<GraphicBuffer>& buffer() const { return buffer_; }
		buffer_handle_t handle() const {
		return buffer_.get() ? buffer_->handle : nullptr;
		}
		int width() const { return buffer_.get() ? buffer_->getWidth() : 0; }
		int height() const { return buffer_.get() ? buffer_->getHeight() : 0; }
		int layer_count() const {
		return buffer_.get() ? buffer_->getLayerCount() : 0;
		}
		int stride() const { return buffer_.get() ? buffer_->getStride() : 0; }
		int layer_stride() const { return 0; }
		int format() const { if (buffer_.get()) return buffer_->getPixelFormat();
		else return 0; }
		int usage() const { if (buffer_.get()) return buffer_->getUsage();
		else return 0; }
		int format() const { return buffer_.get() ? buffer_->getPixelFormat() : 0; }
		int usage() const { return buffer_.get() ? buffer_->getUsage() : 0; }

		private:
		sp<GraphicBuffer> buffer_;

libs/vr/libbufferhubqueue/buffer_hub_queue_client.cpp

+116 −60

Original line number	Diff line number	Diff line
		@@ -11,34 +11,36 @@
		#include <pdx/file_handle.h>
		#include <private/dvr/bufferhub_rpc.h>

		using android::pdx::ErrorStatus;
		using android::pdx::LocalChannelHandle;
		using android::pdx::Status;

		namespace android {
		namespace dvr {

		BufferHubQueue::BufferHubQueue(LocalChannelHandle channel_handle,
		size_t meta_size)
		BufferHubQueue::BufferHubQueue(LocalChannelHandle channel_handle)
		: Client{pdx::default_transport::ClientChannel::Create(
		std::move(channel_handle))},
		meta_size_(meta_size),
		meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
		meta_size_(0),
		buffers_(BufferHubQueue::kMaxQueueCapacity),
		epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
		available_buffers_(BufferHubQueue::kMaxQueueCapacity),
		fences_(BufferHubQueue::kMaxQueueCapacity),
		capacity_(0) {
		capacity_(0),
		id_(-1) {
		Initialize();
		}

		BufferHubQueue::BufferHubQueue(const std::string& endpoint_path,
		size_t meta_size)
		BufferHubQueue::BufferHubQueue(const std::string& endpoint_path)
		: Client{pdx::default_transport::ClientChannelFactory::Create(
		endpoint_path)},
		meta_size_(meta_size),
		meta_buffer_tmp_(meta_size ? new uint8_t[meta_size] : nullptr),
		meta_size_(0),
		buffers_(BufferHubQueue::kMaxQueueCapacity),
		epollhup_pending_(BufferHubQueue::kMaxQueueCapacity, false),
		available_buffers_(BufferHubQueue::kMaxQueueCapacity),
		fences_(BufferHubQueue::kMaxQueueCapacity),
		capacity_(0) {
		capacity_(0),
		id_(-1) {
		Initialize();
		}

		@@ -55,26 +57,47 @@ void BufferHubQueue::Initialize() {
		BufferHubQueue::kEpollQueueEventIndex)}};
		ret = epoll_fd_.Control(EPOLL_CTL_ADD, event_fd(), &event);
		if (ret < 0) {
		ALOGE("Failed to register ConsumerQueue into epoll event: %s",
		ALOGE("BufferHubQueue::Initialize: Failed to add event fd to epoll set: %s",
		strerror(-ret));
		}
		}

		std::unique_ptr<ConsumerQueue> BufferHubQueue::CreateConsumerQueue() {
		Status<std::pair<LocalChannelHandle, size_t>> status =
		InvokeRemoteMethod<BufferHubRPC::CreateConsumerQueue>();

		Status<void> BufferHubQueue::ImportQueue() {
		auto status = InvokeRemoteMethod<BufferHubRPC::GetQueueInfo>();
		if (!status) {
		ALOGE("Cannot create ConsumerQueue: %s", status.GetErrorMessage().c_str());
		ALOGE("BufferHubQueue::ImportQueue: Failed to import queue: %s",
		status.GetErrorMessage().c_str());
		return ErrorStatus(status.error());
		} else {
		SetupQueue(status.get().meta_size_bytes, status.get().id);
		return {};
		}
		}

		void BufferHubQueue::SetupQueue(size_t meta_size_bytes, int id) {
		meta_size_ = meta_size_bytes;
		id_ = id;
		meta_buffer_tmp_.reset(meta_size_ > 0 ? new uint8_t[meta_size_] : nullptr);
		}

		std::unique_ptr<ConsumerQueue> BufferHubQueue::CreateConsumerQueue() {
		if (auto status = CreateConsumerQueueHandle())
		return std::unique_ptr<ConsumerQueue>(new ConsumerQueue(status.take()));
		else
		return nullptr;
		}

		auto return_value = status.take();
		Status<LocalChannelHandle> BufferHubQueue::CreateConsumerQueueHandle() {
		auto status = InvokeRemoteMethod<BufferHubRPC::CreateConsumerQueue>();
		if (!status) {
		ALOGE(
		"BufferHubQueue::CreateConsumerQueue: Failed to create consumer queue: "
		"%s",
		status.GetErrorMessage().c_str());
		return ErrorStatus(status.error());
		}

		ALOGD_IF(TRACE, "BufferHubQueue::CreateConsumerQueue: meta_size_bytes=%zu",
		return_value.second);
		return ConsumerQueue::Create(std::move(return_value.first),
		return_value.second);
		return status;
		}

		bool BufferHubQueue::WaitForBuffers(int timeout) {
		@@ -89,7 +112,8 @@ bool BufferHubQueue::WaitForBuffers(int timeout) {
		}

		if (ret < 0 && ret != -EINTR) {
		ALOGE("Failed to wait for buffers: %s", strerror(-ret));
		ALOGE("BufferHubQueue::WaitForBuffers: Failed to wait for buffers: %s",
		strerror(-ret));
		return false;
		}

		@@ -108,7 +132,8 @@ bool BufferHubQueue::WaitForBuffers(int timeout) {
		} else if (is_queue_event_index(index)) {
		HandleQueueEvent(events[i]);
		} else {
		ALOGW("Unknown event index: %" PRId64, index);
		ALOGW("BufferHubQueue::WaitForBuffers: Unknown event index: %" PRId64,
		index);
		}
		}
		}
		@@ -134,7 +159,8 @@ void BufferHubQueue::HandleBufferEvent(size_t slot, const epoll_event& event) {
		if (events & EPOLLIN) {
		int ret = OnBufferReady(buffer, &fences_[slot]);
		if (ret < 0) {
		ALOGE("Failed to set buffer ready: %s", strerror(-ret));
		ALOGE("BufferHubQueue::HandleBufferEvent: Failed to set buffer ready: %s",
		strerror(-ret));
		return;
		}
		Enqueue(buffer, slot);
		@@ -144,8 +170,8 @@ void BufferHubQueue::HandleBufferEvent(size_t slot, const epoll_event& event) {
		// 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",
		"BufferHubQueue::HandleBufferEvent: Received EPOLLHUP at slot: %zu, "
		"buffer event fd: %d, EPOLLHUP pending: %d",
		slot, buffer->event_fd(), int{epollhup_pending_[slot]});
		if (epollhup_pending_[slot]) {
		epollhup_pending_[slot] = false;
		@@ -153,7 +179,10 @@ void BufferHubQueue::HandleBufferEvent(size_t slot, const epoll_event& event) {
		DetachBuffer(slot);
		}
		} else {
		ALOGW("Unknown event, slot=%zu, epoll events=%d", slot, events);
		ALOGW(
		"BufferHubQueue::HandleBufferEvent: Unknown event, slot=%zu, epoll "
		"events=%d",
		slot, events);
		}
		}

		@@ -169,12 +198,13 @@ void BufferHubQueue::HandleQueueEvent(const epoll_event& event) {
		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));
		auto buffer_status = OnBufferAllocated();
		if (!buffer_status) {
		ALOGE("BufferHubQueue::HandleQueueEvent: Failed to import buffer: %s",
		buffer_status.GetErrorMessage().c_str());
		}
		} else {
		ALOGW("Unknown epoll events=%d", events);
		ALOGW("BufferHubQueue::HandleQueueEvent: Unknown epoll events=%d", events);
		}
		}

		@@ -233,7 +263,7 @@ int BufferHubQueue::DetachBuffer(size_t slot) {
		void BufferHubQueue::Enqueue(std::shared_ptr<BufferHubBuffer> buf,
		size_t slot) {
		if (count() == capacity_) {
		ALOGE("Buffer queue is full!");
		ALOGE("BufferHubQueue::Enqueue: Buffer queue is full!");
		return;
		}

		@@ -274,7 +304,7 @@ std::shared_ptr<BufferHubBuffer> BufferHubQueue::Dequeue(int timeout,
		available_buffers_.PopFront();

		if (!buf) {
		ALOGE("Dequeue: Buffer to be dequeued is nullptr");
		ALOGE("BufferHubQueue::Dequeue: Buffer to be dequeued is nullptr");
		return nullptr;
		}

		@@ -289,15 +319,22 @@ std::shared_ptr<BufferHubBuffer> BufferHubQueue::Dequeue(int timeout,
		ProducerQueue::ProducerQueue(size_t meta_size)
		: ProducerQueue(meta_size, 0, 0, 0, 0) {}

		ProducerQueue::ProducerQueue(LocalChannelHandle handle, size_t meta_size)
		: BASE(std::move(handle), meta_size) {}
		ProducerQueue::ProducerQueue(LocalChannelHandle handle)
		: BASE(std::move(handle)) {
		auto status = ImportQueue();
		if (!status) {
		ALOGE("ProducerQueue::ProducerQueue: Failed to import queue: %s",
		status.GetErrorMessage().c_str());
		Close(-status.error());
		}
		}

		ProducerQueue::ProducerQueue(size_t meta_size, int usage_set_mask,
		int usage_clear_mask, int usage_deny_set_mask,
		int usage_deny_clear_mask)
		: BASE(BufferHubRPC::kClientPath, meta_size) {
		: BASE(BufferHubRPC::kClientPath) {
		auto status = InvokeRemoteMethod<BufferHubRPC::CreateProducerQueue>(
		meta_size_, usage_set_mask, usage_clear_mask, usage_deny_set_mask,
		meta_size, usage_set_mask, usage_clear_mask, usage_deny_set_mask,
		usage_deny_clear_mask);
		if (!status) {
		ALOGE("ProducerQueue::ProducerQueue: Failed to create producer queue: %s",
		@@ -305,12 +342,14 @@ ProducerQueue::ProducerQueue(size_t meta_size, int usage_set_mask,
		Close(-status.error());
		return;
		}

		SetupQueue(status.get().meta_size_bytes, status.get().id);
		}

		int ProducerQueue::AllocateBuffer(int width, int height, int format, int usage,
		size_t slice_count, size_t* out_slot) {
		if (out_slot == nullptr) {
		ALOGE("Parameter out_slot cannot be null.");
		ALOGE("ProducerQueue::AllocateBuffer: Parameter out_slot cannot be null.");
		return -EINVAL;
		}

		@@ -362,7 +401,7 @@ int ProducerQueue::AddBuffer(const std::shared_ptr<BufferProducer>& buf,
		}

		int ProducerQueue::DetachBuffer(size_t slot) {
		Status<int> status =
		auto status =
		InvokeRemoteMethod<BufferHubRPC::ProducerQueueDetachBuffer>(slot);
		if (!status) {
		ALOGE(
		@@ -378,7 +417,9 @@ int ProducerQueue::DetachBuffer(size_t slot) {
		std::shared_ptr<BufferProducer> ProducerQueue::Dequeue(
		int timeout, size_t* slot, LocalHandle* release_fence) {
		if (slot == nullptr \|\| release_fence == nullptr) {
		ALOGE("invalid parameter, slot=%p, release_fence=%p", slot, release_fence);
		ALOGE(
		"ProducerQueue::Dequeue: invalid parameter, slot=%p, release_fence=%p",
		slot, release_fence);
		return nullptr;
		}

		@@ -392,21 +433,27 @@ int ProducerQueue::OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
		return buffer->Gain(release_fence);
		}

		ConsumerQueue::ConsumerQueue(LocalChannelHandle handle, size_t meta_size)
		: BASE(std::move(handle), meta_size) {
		// TODO(b/34387835) Import consumer queue in case the ProducerQueue we are
		ConsumerQueue::ConsumerQueue(LocalChannelHandle handle)
		: BufferHubQueue(std::move(handle)) {
		auto status = ImportQueue();
		if (!status) {
		ALOGE("ConsumerQueue::ConsumerQueue: Failed to import queue: %s",
		status.GetErrorMessage().c_str());
		Close(-status.error());
		}

		// TODO(b/34387835) Import buffers in case the ProducerQueue we are
		// based on was not empty.
		}

		int ConsumerQueue::ImportBuffers() {
		Status<std::vector<std::pair<LocalChannelHandle, size_t>>> status =
		InvokeRemoteMethod<BufferHubRPC::ConsumerQueueImportBuffers>();
		Status<size_t> ConsumerQueue::ImportBuffers() {
		auto status = InvokeRemoteMethod<BufferHubRPC::ConsumerQueueImportBuffers>();
		if (!status) {
		ALOGE(
		"ConsumerQueue::ImportBuffers failed to import consumer buffer through "
		"BufferBub, error: %s",
		status.GetErrorMessage().c_str());
		return -status.error();
		return ErrorStatus(status.error());
		}

		int last_error = 0;
		@@ -431,7 +478,10 @@ int ConsumerQueue::ImportBuffers() {
		}
		}

		return imported_buffers > 0 ? imported_buffers : last_error;
		if (imported_buffers > 0)
		return {imported_buffers};
		else
		return ErrorStatus(-last_error);
		}

		int ConsumerQueue::AddBuffer(const std::shared_ptr<BufferConsumer>& buf,
		@@ -445,15 +495,17 @@ std::shared_ptr<BufferConsumer> ConsumerQueue::Dequeue(
		LocalHandle* acquire_fence) {
		if (meta_size != meta_size_) {
		ALOGE(
		"metadata size (%zu) for the dequeuing buffer does not match metadata "
		"size (%zu) for the queue.",
		"ConsumerQueue::Dequeue: Metadata size (%zu) for the dequeuing buffer "
		"does not match metadata size (%zu) for the queue.",
		meta_size, meta_size_);
		return nullptr;
		}

		if (slot == nullptr \|\| meta == nullptr \|\| acquire_fence == nullptr) {
		ALOGE("invalid parameter, slot=%p, meta=%p, acquire_fence=%p", slot, meta,
		acquire_fence);
		ALOGE(
		"ConsumerQueue::Dequeue: Invalid parameter, slot=%p, meta=%p, "
		"acquire_fence=%p",
		slot, meta, acquire_fence);
		return nullptr;
		}

		@@ -467,15 +519,19 @@ int ConsumerQueue::OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
		return buffer->Acquire(acquire_fence, meta_buffer_tmp_.get(), meta_size_);
		}

		int ConsumerQueue::OnBufferAllocated() {
		const int ret = ImportBuffers();
		if (ret == 0) {
		ALOGW("No new buffer can be imported on buffer allocated event.");
		} else if (ret < 0) {
		ALOGE("Failed to import buffers on buffer allocated event.");
		Status<void> ConsumerQueue::OnBufferAllocated() {
		auto status = ImportBuffers();
		if (!status) {
		ALOGE("ConsumerQueue::OnBufferAllocated: Failed to import buffers: %s",
		status.GetErrorMessage().c_str());
		return ErrorStatus(status.error());
		} else if (status.get() == 0) {
		ALOGW("ConsumerQueue::OnBufferAllocated: No new buffers allocated!");
		return ErrorStatus(ENOBUFS);
		} else {
		ALOGD_IF(TRACE, "Imported %zu consumer buffers.", status.get());
		return {};
		}
		ALOGD_IF(TRACE, "Imported %d consumer buffers.", ret);
		return ret;
		}

		} // namespace dvr

libs/vr/libbufferhubqueue/include/private/dvr/buffer_hub_queue_client.h

+34 −17

Original line number	Diff line number	Diff line
		@@ -4,6 +4,7 @@
		#include <gui/BufferQueueDefs.h>

		#include <pdx/client.h>
		#include <pdx/status.h>
		#include <private/dvr/buffer_hub_client.h>
		#include <private/dvr/epoll_file_descriptor.h>
		#include <private/dvr/ring_buffer.h>
		@@ -41,6 +42,9 @@ class BufferHubQueue : public pdx::Client {
		// Return the default buffer format of this buffer queue.
		int32_t default_format() const { return default_format_; }

		// Create a new consumer in handle form for immediate transport over RPC.
		Status<LocalChannelHandle> CreateConsumerQueueHandle();

		// Return the number of buffers avaiable for dequeue.
		size_t count() const { return available_buffers_.GetSize(); }

		@@ -83,9 +87,18 @@ class BufferHubQueue : public pdx::Client {
		// timeout.
		static constexpr int kNoTimeOut = -1;

		int id() const { return id_; }

		protected:
		BufferHubQueue(LocalChannelHandle channel, size_t meta_size);
		BufferHubQueue(const std::string& endpoint_path, size_t meta_size);
		BufferHubQueue(LocalChannelHandle channel);
		BufferHubQueue(const std::string& endpoint_path);

		// Imports the queue parameters by querying BufferHub for the parameters for
		// this channel.
		Status<void> ImportQueue();

		// Sets up the queue with the given parameters.
		void SetupQueue(size_t meta_size_bytes_, int id);

		// Called by ProducerQueue::AddBuffer and ConsumerQueue::AddBuffer only. to
		// register a buffer for epoll and internal bookkeeping.
		@@ -112,7 +125,7 @@ class BufferHubQueue : public pdx::Client {
		LocalHandle* fence) = 0;

		// Called when a buffer is allocated remotely.
		virtual int OnBufferAllocated() = 0;
		virtual Status<void> OnBufferAllocated() { return {}; }

		// Data members to handle arbitrary metadata passed through BufferHub. It is
		// fair to enforce that all buffers in the same queue share the same metadata
		@@ -235,6 +248,9 @@ class BufferHubQueue : public pdx::Client {
		// Epoll fd used to wait for BufferHub events.
		EpollFileDescriptor epoll_fd_;

		// Global id for the queue that is consistent across processes.
		int id_;

		BufferHubQueue(const BufferHubQueue&) = delete;
		void operator=(BufferHubQueue&) = delete;
		};
		@@ -267,9 +283,8 @@ class ProducerQueue : public pdx::ClientBase<ProducerQueue, BufferHubQueue> {
		}

		// Import a \|ProducerQueue\| from a channel handle.
		template <typename Meta>
		static std::unique_ptr<ProducerQueue> Import(LocalChannelHandle handle) {
		return BASE::Create(std::move(handle), sizeof(Meta));
		return BASE::Create(std::move(handle));
		}

		// Get a buffer producer. Note that the method doesn't check whether the
		@@ -311,18 +326,15 @@ class ProducerQueue : public pdx::ClientBase<ProducerQueue, BufferHubQueue> {
		// static template methods inherited from ClientBase, which take the same
		// arguments as the constructors.
		explicit ProducerQueue(size_t meta_size);
		ProducerQueue(LocalChannelHandle handle, size_t meta_size);
		ProducerQueue(LocalChannelHandle handle);
		ProducerQueue(size_t meta_size, int usage_set_mask, int usage_clear_mask,
		int usage_deny_set_mask, int usage_deny_clear_mask);

		int OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
		LocalHandle* release_fence) override;

		// Producer buffer is always allocated from the client (i.e. local) side.
		int OnBufferAllocated() override { return 0; }
		};

		class ConsumerQueue : public pdx::ClientBase<ConsumerQueue, BufferHubQueue> {
		class ConsumerQueue : public BufferHubQueue {
		public:
		// Get a buffer consumer. Note that the method doesn't check whether the
		// buffer slot has a valid buffer that has been imported already. When no
		@@ -333,10 +345,14 @@ class ConsumerQueue : public pdx::ClientBase<ConsumerQueue, BufferHubQueue> {
		BufferHubQueue::GetBuffer(slot));
		}

		// Import a \|ConsumerQueue\| from a channel handle.
		static std::unique_ptr<ConsumerQueue> Import(LocalChannelHandle handle) {
		return std::unique_ptr<ConsumerQueue>(new ConsumerQueue(std::move(handle)));
		}

		// Import newly created buffers from the service side.
		// Returns number of buffers successfully imported; or negative error code
		// when buffer import fails.
		int ImportBuffers();
		// Returns number of buffers successfully imported or an error.
		Status<size_t> ImportBuffers();

		// Dequeue a consumer buffer to read. The returned buffer in \|Acquired\|'ed
		// mode, and caller should call Releasse() once it's done writing to release
		@@ -353,10 +369,11 @@ class ConsumerQueue : public pdx::ClientBase<ConsumerQueue, BufferHubQueue> {
		std::shared_ptr<BufferConsumer> Dequeue(int timeout, size_t* slot, void* meta,
		size_t meta_size,
		LocalHandle* acquire_fence);

		private:
		friend BASE;
		friend BufferHubQueue;

		ConsumerQueue(LocalChannelHandle handle, size_t meta_size);
		ConsumerQueue(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\|
		@@ -367,7 +384,7 @@ class ConsumerQueue : public pdx::ClientBase<ConsumerQueue, BufferHubQueue> {
		int OnBufferReady(std::shared_ptr<BufferHubBuffer> buf,
		LocalHandle* acquire_fence) override;

		int OnBufferAllocated() override;
		Status<void> OnBufferAllocated() override;
		};

		} // namespace dvr