media/fuchsia/common/stream_processor_helper.cc

// Copyright 2019 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/stream_processor_helper.h"

#include "base/fuchsia/fuchsia_logging.h"
#include "base/functional/bind.h"
#include "media/base/timestamp_constants.h"

namespace media {

namespace {

// Input buffers are allocate once per decoder, the |buffer_lifetime_ordinal| is
// always the same.
constexpr uint64_t kInputBufferLifetimeOrdinal = 1;

}  // namespace

StreamProcessorHelper::IoPacket::IoPacket(size_t index,
                                          size_t offset,
                                          size_t size,
                                          base::TimeDelta timestamp,
                                          bool unit_end,
                                          bool key_frame,
                                          base::OnceClosure destroy_cb)
    : index_(index),
      offset_(offset),
      size_(size),
      timestamp_(timestamp),
      unit_end_(unit_end),
      key_frame_(key_frame) {
  destroy_callbacks_.push_front(std::move(destroy_cb));
}

StreamProcessorHelper::IoPacket::~IoPacket() {
  for (auto& cb : destroy_callbacks_) {
    std::move(cb).Run();
  }
}

StreamProcessorHelper::IoPacket::IoPacket(IoPacket&&) = default;
StreamProcessorHelper::IoPacket& StreamProcessorHelper::IoPacket::operator=(
    IoPacket&&) = default;

void StreamProcessorHelper::IoPacket::AddOnDestroyClosure(
    base::OnceClosure closure) {
  destroy_callbacks_.push_front(std::move(closure));
}

StreamProcessorHelper::StreamProcessorHelper(
    fuchsia::media::StreamProcessorPtr processor,
    Client* client)
    : processor_(std::move(processor)), client_(client), weak_factory_(this) {
  DCHECK(processor_);
  DCHECK(client_);
  weak_this_ = weak_factory_.GetWeakPtr();

  processor_.set_error_handler(
      [this](zx_status_t status) {
        ZX_LOG(ERROR, status)
            << "The fuchsia.media.StreamProcessor channel was terminated.";
        OnError();
      });

  processor_.events().OnStreamFailed =
      fit::bind_member(this, &StreamProcessorHelper::OnStreamFailed);
  processor_.events().OnFreeInputPacket =
      fit::bind_member(this, &StreamProcessorHelper::OnFreeInputPacket);
  processor_.events().OnInputConstraints =
      fit::bind_member(this, &StreamProcessorHelper::OnInputConstraints);
  processor_.events().OnOutputConstraints =
      fit::bind_member(this, &StreamProcessorHelper::OnOutputConstraints);
  processor_.events().OnOutputFormat =
      fit::bind_member(this, &StreamProcessorHelper::OnOutputFormat);
  processor_.events().OnOutputPacket =
      fit::bind_member(this, &StreamProcessorHelper::OnOutputPacket);
  processor_.events().OnOutputEndOfStream =
      fit::bind_member(this, &StreamProcessorHelper::OnOutputEndOfStream);

  processor_->EnableOnStreamFailed();
}

StreamProcessorHelper::~StreamProcessorHelper() {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
}

void StreamProcessorHelper::Process(IoPacket input) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  DCHECK(processor_);

  fuchsia::media::Packet packet;
  packet.mutable_header()->set_buffer_lifetime_ordinal(
      kInputBufferLifetimeOrdinal);
  packet.mutable_header()->set_packet_index(input.buffer_index());
  packet.set_buffer_index(packet.header().packet_index());
  packet.set_timestamp_ish(input.timestamp().InNanoseconds());
  packet.set_stream_lifetime_ordinal(stream_lifetime_ordinal_);
  packet.set_start_offset(input.offset());
  packet.set_valid_length_bytes(input.size());
  packet.set_known_end_access_unit(input.unit_end());

  active_stream_ = true;

  if (!input.format().IsEmpty()) {
    processor_->QueueInputFormatDetails(stream_lifetime_ordinal_,
                                        fidl::Clone(input.format()));
  }

  DCHECK(!input_packets_.contains(input.buffer_index()));
  input_packets_.insert_or_assign(input.buffer_index(), std::move(input));
  processor_->QueueInputPacket(std::move(packet));
}

void StreamProcessorHelper::ProcessEos() {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  DCHECK(processor_);

  active_stream_ = true;
  processor_->QueueInputEndOfStream(stream_lifetime_ordinal_);
  processor_->FlushEndOfStreamAndCloseStream(stream_lifetime_ordinal_);
}

void StreamProcessorHelper::Reset() {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!active_stream_) {
    // Nothing to do if we don't have an active stream.
    return;
  }

  if (processor_) {
    processor_->CloseCurrentStream(stream_lifetime_ordinal_,
                                   /*release_input_buffers=*/false,
                                   /*release_output_buffers=*/false);
  }

  stream_lifetime_ordinal_ += 2;
  active_stream_ = false;
}

void StreamProcessorHelper::OnStreamFailed(uint64_t stream_lifetime_ordinal,
                                           fuchsia::media::StreamError error) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (stream_lifetime_ordinal_ != stream_lifetime_ordinal) {
    return;
  }

  if (error == fuchsia::media::StreamError::DECRYPTOR_NO_KEY) {
    // Always reset the stream since the current one has failed.
    Reset();

    client_->OnStreamProcessorNoKey();
    return;
  }

  OnError();
}

void StreamProcessorHelper::OnFreeInputPacket(
    fuchsia::media::PacketHeader free_input_packet) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!free_input_packet.has_packet_index()) {
    DLOG(ERROR) << "Received OnFreeInputPacket() with missing required fields.";
    OnError();
    return;
  }

  auto it = input_packets_.find(free_input_packet.packet_index());
  if (it == input_packets_.end()) {
    DLOG(ERROR) << "Received OnFreeInputPacket() with invalid packet index.";
    OnError();
    return;
  }

  // The packet should be destroyed only after it's removed from
  // |input_packets_|. Otherwise packet destruction observer may call Process()
  // for the next packet while the current packet is still in |input_packets_|.
  auto packet = std::move(it->second);
  input_packets_.erase(it);
}

void StreamProcessorHelper::OnInputConstraints(
    fuchsia::media::StreamBufferConstraints input_constraints) {
  if (!input_constraints.has_buffer_constraints_version_ordinal()) {
    DLOG(ERROR)
        << "Received OnInputConstraints() with missing required fields.";
    OnError();
    return;
  }

  if (input_constraints.buffer_constraints_version_ordinal() !=
      kInputBufferLifetimeOrdinal) {
    DLOG(ERROR) << "Received OnInputConstraints() unexpected version ordinal: "
                << input_constraints.buffer_constraints_version_ordinal();
    OnError();
    return;
  }

  client_->OnStreamProcessorAllocateInputBuffers(input_constraints);
}

void StreamProcessorHelper::OnOutputConstraints(
    fuchsia::media::StreamOutputConstraints output_constraints) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!output_constraints.has_stream_lifetime_ordinal()) {
    DLOG(ERROR)
        << "Received OnOutputConstraints() with missing required fields.";
    OnError();
    return;
  }

  if (output_constraints.stream_lifetime_ordinal() !=
      stream_lifetime_ordinal_) {
    return;
  }

  if (!output_constraints.has_buffer_constraints_action_required() ||
      !output_constraints.buffer_constraints_action_required()) {
    return;
  }

  if (!output_constraints.has_buffer_constraints()) {
    DLOG(ERROR) << "Received OnOutputConstraints() which requires buffer "
                   "constraints action, but without buffer constraints.";
    OnError();
    return;
  }

  // StreamProcessor API expects odd buffer lifetime ordinal, which is
  // incremented by 2 for each buffer generation.
  output_buffer_lifetime_ordinal_ += 2;

  output_buffer_constraints_ =
      std::move(*output_constraints.mutable_buffer_constraints());

  client_->OnStreamProcessorAllocateOutputBuffers(output_buffer_constraints_);
}

void StreamProcessorHelper::OnOutputFormat(
    fuchsia::media::StreamOutputFormat output_format) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!output_format.has_stream_lifetime_ordinal() ||
      !output_format.has_format_details()) {
    DLOG(ERROR) << "Received OnOutputFormat() with missing required fields.";
    OnError();
    return;
  }

  if (output_format.stream_lifetime_ordinal() != stream_lifetime_ordinal_) {
    return;
  }

  client_->OnStreamProcessorOutputFormat(std::move(output_format));
}

void StreamProcessorHelper::OnOutputPacket(fuchsia::media::Packet output_packet,
                                           bool error_detected_before,
                                           bool error_detected_during) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!output_packet.has_header() ||
      !output_packet.header().has_buffer_lifetime_ordinal() ||
      !output_packet.header().has_packet_index() ||
      !output_packet.has_stream_lifetime_ordinal() ||
      !output_packet.has_buffer_index()) {
    DLOG(ERROR) << "Received OnOutputPacket() with missing required fields.";
    OnError();
    return;
  }

  if (output_packet.header().buffer_lifetime_ordinal() !=
      output_buffer_lifetime_ordinal_) {
    return;
  }

  if (output_packet.stream_lifetime_ordinal() != stream_lifetime_ordinal_) {
    // Output packets from old streams still need to be recycled.
    OnRecycleOutputBuffer(output_buffer_lifetime_ordinal_,
                          output_packet.header().packet_index());
    return;
  }

  auto buffer_index = output_packet.buffer_index();
  auto packet_index = output_packet.header().packet_index();
  base::TimeDelta timestamp =
      output_packet.has_timestamp_ish()
          ? base::Nanoseconds(output_packet.timestamp_ish())
          : kNoTimestamp;

  client_->OnStreamProcessorOutputPacket(IoPacket(
      buffer_index, output_packet.start_offset(),
      output_packet.valid_length_bytes(), timestamp,
      output_packet.known_end_access_unit(),
      output_packet.has_key_frame() ? output_packet.key_frame() : false,
      base::BindOnce(&StreamProcessorHelper::OnRecycleOutputBuffer, weak_this_,
                     output_buffer_lifetime_ordinal_, packet_index)));
}

void StreamProcessorHelper::OnOutputEndOfStream(
    uint64_t stream_lifetime_ordinal,
    bool error_detected_before) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (stream_lifetime_ordinal != stream_lifetime_ordinal_) {
    return;
  }

  stream_lifetime_ordinal_ += 2;
  active_stream_ = false;

  client_->OnStreamProcessorEndOfStream();
}

void StreamProcessorHelper::OnError() {
  processor_.Unbind();
  client_->OnStreamProcessorError();
}

void StreamProcessorHelper::SetInputBufferCollectionToken(
    fuchsia::sysmem::BufferCollectionTokenPtr sysmem_token) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  fuchsia::media::StreamBufferPartialSettings settings;
  settings.set_buffer_lifetime_ordinal(kInputBufferLifetimeOrdinal);
  settings.set_buffer_constraints_version_ordinal(0);
  settings.set_sysmem_token(std::move(sysmem_token));
  processor_->SetInputBufferPartialSettings(std::move(settings));
}

void StreamProcessorHelper::CompleteOutputBuffersAllocation(
    fuchsia::sysmem::BufferCollectionTokenPtr collection_token) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  DCHECK(!output_buffer_constraints_.IsEmpty());

  // Pass new output buffer settings to the stream processor.
  fuchsia::media::StreamBufferPartialSettings settings;
  settings.set_buffer_lifetime_ordinal(output_buffer_lifetime_ordinal_);
  settings.set_buffer_constraints_version_ordinal(
      output_buffer_constraints_.buffer_constraints_version_ordinal());
  settings.set_sysmem_token(std::move(collection_token));
  processor_->SetOutputBufferPartialSettings(std::move(settings));
  processor_->CompleteOutputBufferPartialSettings(
      output_buffer_lifetime_ordinal_);
}

void StreamProcessorHelper::OnRecycleOutputBuffer(
    uint64_t buffer_lifetime_ordinal,
    uint32_t packet_index) {
  DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);

  if (!processor_)
    return;

  if (buffer_lifetime_ordinal != output_buffer_lifetime_ordinal_)
    return;

  fuchsia::media::PacketHeader header;
  header.set_buffer_lifetime_ordinal(buffer_lifetime_ordinal);
  header.set_packet_index(packet_index);
  processor_->RecycleOutputPacket(std::move(header));
}

}  // namespace media