blob: f4daf110685b5bd8232aabc7dfd1e96860160533 [file] [log] [blame]
// Copyright 2021 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "media/fuchsia/common/vmo_buffer_writer_queue.h"
#include <zircon/rights.h>
#include <algorithm>
#include "base/fuchsia/fuchsia_logging.h"
#include "base/process/process_metrics.h"
#include "media/base/decoder_buffer.h"
namespace media {
VmoBufferWriterQueue::PendingBuffer::PendingBuffer(
scoped_refptr<DecoderBuffer> buffer)
: buffer(buffer) {
DCHECK(buffer);
}
VmoBufferWriterQueue::PendingBuffer::~PendingBuffer() = default;
VmoBufferWriterQueue::PendingBuffer::PendingBuffer(PendingBuffer&& other) =
default;
const uint8_t* VmoBufferWriterQueue::PendingBuffer::data() const {
return buffer->data() + buffer_pos;
}
size_t VmoBufferWriterQueue::PendingBuffer::bytes_left() const {
return buffer->data_size() - buffer_pos;
}
void VmoBufferWriterQueue::PendingBuffer::AdvanceCurrentPos(size_t bytes) {
DCHECK_LE(bytes, bytes_left());
buffer_pos += bytes;
}
VmoBufferWriterQueue::VmoBufferWriterQueue() {
DETACH_FROM_THREAD(thread_checker_);
}
VmoBufferWriterQueue::~VmoBufferWriterQueue() = default;
void VmoBufferWriterQueue::EnqueueBuffer(scoped_refptr<DecoderBuffer> buffer) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
pending_buffers_.emplace_back(std::move(buffer));
PumpPackets();
}
void VmoBufferWriterQueue::Start(std::vector<VmoBuffer> buffers,
SendPacketCB send_packet_cb,
EndOfStreamCB end_of_stream_cb) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
DCHECK(buffers_.empty());
DCHECK(!buffers.empty());
buffers_ = std::move(buffers);
send_packet_cb_ = std::move(send_packet_cb);
end_of_stream_cb_ = std::move(end_of_stream_cb);
// Initialize |unused_buffers_|.
unused_buffers_.reserve(buffers_.size());
for (size_t i = 0; i < buffers_.size(); ++i) {
unused_buffers_.push_back(i);
}
PumpPackets();
}
bool VmoBufferWriterQueue::IsBlocked() const {
return unused_buffers_.empty();
}
void VmoBufferWriterQueue::PumpPackets() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
auto weak_this = weak_factory_.GetWeakPtr();
while (!buffers_.empty() && !is_paused_ &&
input_queue_position_ < pending_buffers_.size()) {
PendingBuffer* current_buffer = &pending_buffers_[input_queue_position_];
if (current_buffer->buffer->end_of_stream()) {
// Pop the EndOfStream buffer if it's the only buffer. Otherwise, mark it
// as complete so that ReleaseBuffer will pop it.
if (input_queue_position_ == 0) {
pending_buffers_.pop_front();
DCHECK(pending_buffers_.empty());
} else {
current_buffer->is_complete = true;
input_queue_position_ += 1;
}
end_of_stream_cb_.Run();
if (!weak_this)
return;
continue;
}
if (unused_buffers_.empty()) {
// No input buffer available.
return;
}
size_t buffer_index = unused_buffers_.back();
unused_buffers_.pop_back();
size_t bytes_filled = buffers_[buffer_index].Write(
base::make_span(current_buffer->data(), current_buffer->bytes_left()));
current_buffer->AdvanceCurrentPos(bytes_filled);
bool buffer_end = current_buffer->bytes_left() == 0;
StreamProcessorHelper::IoPacket packet(
buffer_index, /*offset=*/0, bytes_filled,
current_buffer->buffer->timestamp(), buffer_end, /*key_frame=*/false,
base::BindOnce(&VmoBufferWriterQueue::ReleaseBuffer,
weak_factory_.GetWeakPtr(), buffer_index));
if (buffer_end) {
current_buffer->tail_sysmem_buffer_index = buffer_index;
input_queue_position_ += 1;
}
send_packet_cb_.Run(current_buffer->buffer.get(), std::move(packet));
if (!weak_this)
return;
}
}
void VmoBufferWriterQueue::ResetQueue() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
pending_buffers_.clear();
input_queue_position_ = 0;
is_paused_ = false;
}
void VmoBufferWriterQueue::ResetBuffers() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
buffers_.clear();
send_packet_cb_ = SendPacketCB();
end_of_stream_cb_ = EndOfStreamCB();
// Invalidate weak pointers, so ReleaseBuffer() is not called for the old
// buffers.
weak_factory_.InvalidateWeakPtrs();
}
void VmoBufferWriterQueue::ResetPositionAndPause() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
for (auto& buffer : pending_buffers_) {
buffer.buffer_pos = 0;
buffer.is_complete = false;
// All packets that were pending will need to be resent. Reset
// |tail_sysmem_buffer_index| to ensure that these packets are not removed
// from the queue in ReleaseBuffer().
buffer.tail_sysmem_buffer_index = absl::nullopt;
}
input_queue_position_ = 0;
is_paused_ = true;
}
void VmoBufferWriterQueue::Unpause() {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
DCHECK(is_paused_);
is_paused_ = false;
PumpPackets();
}
void VmoBufferWriterQueue::ReleaseBuffer(size_t buffer_index) {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
DCHECK(!buffers_.empty());
// Mark the input buffer as complete.
for (size_t i = 0; i < input_queue_position_; ++i) {
if (pending_buffers_[i].tail_sysmem_buffer_index == buffer_index)
pending_buffers_[i].is_complete = true;
}
// Remove all complete buffers from the head of the queue since we no longer
// need them. Note that currently StreamProcessor doesn't guarantee that input
// buffers are released in the same order they were sent (see
// https://fuchsia.googlesource.com/fuchsia/+/3b12c8c5/sdk/fidl/fuchsia.media/stream_processor.fidl#1646
// ). This means that some complete buffers will need to stay in the queue
// until all preceding packets are released as well.
while (!pending_buffers_.empty() && pending_buffers_.front().is_complete) {
pending_buffers_.pop_front();
DCHECK_GT(input_queue_position_, 0U);
input_queue_position_--;
}
unused_buffers_.push_back(buffer_index);
PumpPackets();
}
size_t VmoBufferWriterQueue::num_buffers() const {
DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
return buffers_.size();
}
} // namespace media