// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <fuchsia/media/cpp/fidl.h>
#include <fuchsia/mediacodec/cpp/fidl.h>
#include <inttypes.h>
#include <lib/async/cpp/task.h>
#include <lib/closure-queue/closure_queue.h>
#include <lib/fidl/cpp/clone.h>
#include <lib/fit/defer.h>
#include <lib/media/codec_impl/codec_impl.h>
#include <lib/media/codec_impl/codec_vmo_range.h>
#include <lib/media/codec_impl/log.h>
#include <threads.h>
#include <zircon/compiler.h>
#include <zircon/threads.h>

#include <mutex>
#include <optional>
#include <variant>

#include <bind/fuchsia/sysmem/heap/cpp/bind.h>
#include <fbl/macros.h>

#include "lib/media/codec_impl/codec_port.h"
#include "src/media/lib/metrics/metrics.cb.h"

// "is_bound_checks" - In several lambdas that just send a message, we check
// is_bound() first, only because of ZX_POL_BAD_HANDLE ZX_POL_ACTION_EXCEPTION.
// If it weren't for that, we really wouldn't care about passing
// ZX_HANDLE_INVALID to zx_channel_write(), since the channel error handling is
// async (we Unbind(), sweep the in-proc send queue, and only then delete the
// Binding).

namespace {

constexpr bool kLogTimestampDelay = false;

// The protocol does not permit an unbounded number of in-flight streams, as
// that would potentially result in unbounded data queued in the incoming
// channel with no valid circuit-breaker value for the incoming channel data.
constexpr size_t kMaxInFlightStreams = 10;

void AssertHeld(const std::unique_lock<std::mutex>& auto_lock, const std::mutex& lock)
    __TA_ASSERT(lock) {
  ZX_ASSERT(auto_lock.mutex() == &lock);
  ZX_ASSERT(auto_lock.owns_lock());
}

class ScopedUnlock {
 public:
  explicit ScopedUnlock(std::unique_lock<std::mutex>& unique_lock) : unique_lock_(unique_lock) {
    unique_lock_.unlock();
  }
  ~ScopedUnlock() { unique_lock_.lock(); }

 private:
  std::unique_lock<std::mutex>& unique_lock_;
  ScopedUnlock() = delete;
  DISALLOW_COPY_ASSIGN_AND_MOVE(ScopedUnlock);
};

// Used within ScopedUnlock only.  Normally we'd just leave a std::unique_lock
// locked until it's destructed.
class ScopedRelock {
 public:
  explicit ScopedRelock(std::unique_lock<std::mutex>& unique_lock) : unique_lock_(unique_lock) {
    unique_lock_.lock();
  }
  ~ScopedRelock() { unique_lock_.unlock(); }

 private:
  std::unique_lock<std::mutex>& unique_lock_;
  ScopedRelock() = delete;
  DISALLOW_COPY_ASSIGN_AND_MOVE(ScopedRelock);
};

bool IsStreamErrorRecoverable(fuchsia::media::StreamError e) {
  using StreamError = fuchsia::media::StreamError;
  switch (e) {
    case StreamError::DECRYPTOR_NO_KEY:
      return true;
    default:
      return false;
  }
}

const char* ToString(fuchsia::media::StreamError e) {
  using StreamError = fuchsia::media::StreamError;
  switch (e) {
    case StreamError::UNKNOWN:
      return "UNKNOWN";
    case StreamError::INVALID_INPUT_FORMAT_DETAILS:
      return "INVALID_INPUT_FORMAT_DETAILS";
    case StreamError::INCOMPATIBLE_BUFFERS_PROVIDED:
      return "INCOMPATIBLE_BUFFERS_PROVIDED";
    case StreamError::EOS_PROCESSING:
      return "EOS_PROCESSING";
    case StreamError::DECODER_UNKNOWN:
      return "DECODER_UNKNOWN";
    case StreamError::DECODER_DATA_PARSING:
      return "DECODER_DATA_PARSING";
    case StreamError::ENCODER_UNKNOWN:
      return "ENCODER_UNKNOWN";
    case StreamError::DECRYPTOR_UNKNOWN:
      return "DECRYPTOR_UNKNOWN";
    case StreamError::DECRYPTOR_NO_KEY:
      return "DECRYPTOR_NO_KEY";
  }
}

const char* GetStreamErrorAdditionalHelpText(fuchsia::media::StreamError e) {
  using StreamError = fuchsia::media::StreamError;
  switch (e) {
    case StreamError::DECRYPTOR_NO_KEY:
      return "Retry after keys arrive.";
    default:
      return "";
  }
}

}  // namespace

CodecImpl::CodecImpl(fidl::ClientEnd<fuchsia_sysmem2::Allocator> sysmem,
                     std::unique_ptr<CodecAdmission> codec_admission,
                     async_dispatcher_t* shared_fidl_dispatcher, thrd_t shared_fidl_thread,
                     StreamProcessorParams params,
                     fidl::InterfaceRequest<fuchsia::media::StreamProcessor> request)
    // The parameters to CodecAdapter constructor here aren't important.
    : CodecAdapter(lock_, this),
      codec_admission_(std::move(codec_admission)),
      shared_fidl_dispatcher_(shared_fidl_dispatcher),
      shared_fidl_thread_(shared_fidl_thread),
      shared_fidl_queue_(shared_fidl_dispatcher, shared_fidl_thread),
      params_(std::move(params)),
      tmp_sysmem_(std::move(sysmem)),
      tmp_interface_request_(std::move(request)),
      binding_(this),
      stream_control_loop_(&kAsyncLoopConfigNoAttachToCurrentThread) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  ZX_DEBUG_ASSERT(tmp_sysmem_);
  ZX_DEBUG_ASSERT(tmp_interface_request_);

  if (codec_admission_) {
    codec_admission_->SetChannelToWaitOn(tmp_interface_request_.channel());
  }

  // This is the binding_'s error handler, not the owner_error_handler_ which
  // is related but separate.
  //
  // If client code instead does ~CodecImpl, this error handler is prevented from running by
  // synchronously unbinding in ~CodecImpl.
  binding_.set_error_handler([this](zx_status_t status) {
    // This handler can't run until after binding_ is bound.
    ZX_DEBUG_ASSERT(was_logically_bound_);
    Unbind();
  });

  initial_input_format_details_ = IsDecoder()   ? &decoder_params().input_details()
                                  : IsEncoder() ? &encoder_params().input_details()
                                                : &decryptor_params().input_details();
}

CodecImpl::~CodecImpl() {
  // We need ~binding_ to run on fidl_thread() else it's not safe to
  // un-bind unilaterally.  We could potentially relax this if BindAsync() was
  // never called, but for now we just require this always.
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());

  if (was_logically_bound_) {
    // Ensure that StreamControl is told to stop, which also stops InputData by
    // calling EnsureStreamClosed() as needed.
    Unbind();

    // Wait for StreamControl to be done.
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      // Normally the fidl_thread() waiting for the StreamControl thread to do anything would be
      // bad, because the fidl_thread() is non-blocking and the StreamControl thread can block on
      // stuff, but StreamControl thread behavior after was_unbind_started_ = true and
      // wake_stream_control_condition_.notify_all() does not block and does not wait on
      // fidl_thread().  So in this case it's ok to wait here.
      while (!is_stream_control_done_) {
        stream_control_done_condition_.wait(lock);
      }
    }  // ~lock

    EnsureUnbindCompleted();
  }

  // Ensure the CodecAdmission is deleted entirely after ~this, including after any relevant base
  // class destructors have run.  This posted work may only get deleted, not run, since some
  // environments will Quit() their async::Loop shortly after ~CodecImpl.  So to avoid depending on
  // the destruction order of captures of a lambda, we use a fit::defer which will run it's lambda
  // when deleted.  In this lambda we can force ~CodecAdmission before ~zx::channel, and we know
  // this lambda will run, whether the lambda further down runs or is just deleted.
  auto run_when_deleted = fit::defer([codec_admission = std::move(codec_admission_),
                                      codec_to_close = std::move(codec_to_close_)]() mutable {
    // Ensure codec_to_close is destructed only after the codec_admission is destructed.  We have
    // to be fairly explicit about this since the order of lambda members is explicitly
    // unspecified in C++, so their destruction order is also unspecified.
    //
    // We care about the order because a client is fairly likely to immediately retry on seeing
    // the channel close, and we don't want that to ever bounce off the CodecAdmission for the
    // instance associated with that same channel.
    codec_admission = nullptr;

    // ~codec_to_close (after ~CodecAdmission above).
  });
  // We intentionally don't use shared_fidl_queue_ here.
  PostSerial(shared_fidl_dispatcher_, [run_when_deleted = std::move(run_when_deleted)] {
    // ~run_when_deleted will run the lambda above, whether run at the end of this
    // lambda, or when this lambda is deleted without ever having run during ~async::Loop
    // or async::Loop::Shutdown().
  });

  // Before destruction, we know that EnsureBuffersNotConfigured() got called for both input and
  // output, so we can assert that these are already not set during destruction.
  ZX_DEBUG_ASSERT(!fake_map_range_[kInputPort]);
  ZX_DEBUG_ASSERT(!fake_map_range_[kOutputPort]);
}

std::mutex& CodecImpl::lock() { return lock_; }

void CodecImpl::SetLifetimeTracking(std::vector<zx::eventpair> lifetime_tracking_eventpair) {
  ZX_DEBUG_ASSERT(lifetime_tracking_eventpair.size() <=
                  fuchsia::mediacodec::CODEC_FACTORY_LIFETIME_TRACKING_EVENTPAIR_PER_CREATE_MAX);
  lifetime_tracking_ = std::move(lifetime_tracking_eventpair);
}

void CodecImpl::SetCodecMetrics(CodecMetrics* codec_metrics) {
  ZX_DEBUG_ASSERT(codec_metrics);
  codec_metrics_ = codec_metrics;
}

void CodecImpl::SetCoreCodecAdapter(std::unique_ptr<CodecAdapter> codec_adapter) {
  ZX_DEBUG_ASSERT(codec_adapter);
  ZX_DEBUG_ASSERT(!codec_adapter_);
  codec_adapter_ = std::move(codec_adapter);
  codec_metrics_implementation_dimension_ = codec_adapter_->CoreCodecMetricsImplementation();
}

void CodecImpl::SetCodecDiagnostics(CodecDiagnostics* codec_diagnostics) {
  ZX_DEBUG_ASSERT(codec_adapter_);
  codec_adapter_->SetCodecDiagnostics(codec_diagnostics);
}

zx_koid_t GetKoid(zx_handle_t handle) {
  zx_info_handle_basic_t info;
  zx_status_t status =
      zx_object_get_info(handle, ZX_INFO_HANDLE_BASIC, &info, sizeof(info), nullptr, nullptr);
  return status == ZX_OK ? info.koid : ZX_KOID_INVALID;
}

std::string GetObjectName(zx_handle_t handle) {
  char name[ZX_MAX_NAME_LEN];
  zx_status_t status = zx_object_get_property(handle, ZX_PROP_NAME, name, sizeof(name));
  return status == ZX_OK ? std::string(name) : std::string();
}

void CodecImpl::BindAsync(fit::closure error_handler) {
  // While it would potentially be safe to call Bind() from a thread other than
  // fidl_thread(), we have no reason to permit that.
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  // Up to once only.  No re-use.
  ZX_DEBUG_ASSERT(!was_bind_async_called_);
  ZX_DEBUG_ASSERT(!binding_.is_bound());
  ZX_DEBUG_ASSERT(tmp_interface_request_);
  was_bind_async_called_ = true;

  zx_status_t start_thread_result =
      stream_control_loop_.StartThread("StreamControl_loop", &stream_control_thread_);
  if (start_thread_result != ZX_OK) {
    // Handle the error async, to be consistent with later errors that must
    // occur async anyway.  Inability to start StreamControl is the only case
    // where we just allow the owner to "delete this" without using
    // UnbindLocked(), since UnbindLocked() relies on StreamControl.
    PostToSharedFidl(std::move(error_handler));
    return;
  }
  // Give CodecAdapter a chance to set a scheduler profile.
  CoreCodecSetStreamControlProfile(zx::unowned_thread(thrd_get_zx_handle(stream_control_thread_)));
  stream_control_queue_.SetDispatcher(stream_control_loop_.dispatcher(), stream_control_thread_);

  // From here on, we'll only fail the CodecImpl via UnbindLocked(), or by
  // just calling ~CodecImpl on the FIDL thread.
  was_logically_bound_ = true;

  // This doesn't really need to be set until the start of the posted lambda
  // below, but here is also fine.
  owner_error_handler_ = std::move(error_handler);

  // Do most of the bind work on StreamControl async, since CoreCodecInit()
  // might potentially take a little while longer than makes sense to run on
  // fidl_thread().  Potential examples: if CoreCodecInit() ends up
  // essentially evicting some other CodecImpl, or if setting up HW can take a
  // while, or if getting a scheduling slot on decode HW can require some
  // waiting, or similar.
  PostToStreamControl([this] {
    // This is allowed to take a little while if necessary, using the current
    // StreamControl thread, which is not shared with any other CodecImpl.
    CoreCodecInit(*initial_input_format_details_);
    is_core_codec_init_called_ = true;

    CoreCodecSetSecureMemoryMode(kOutputPort, PortSecureMemoryMode(kOutputPort));
    CoreCodecSetSecureMemoryMode(kInputPort, PortSecureMemoryMode(kInputPort));

    if (IsCoreCodecHwBased(kInputPort) || IsCoreCodecHwBased(kOutputPort)) {
      core_codec_bti_ = CoreCodecBti();
    }

    // We touch FIDL stuff only from the fidl_thread().  While it would
    // be more efficient to post once to bind and send up to two messages below,
    // by posting individually we can share more code and have simpler rules for
    // calling that code.

    // Once this is posted, we can be dispatching incoming FIDL messages,
    // concurrent with the rest of the current lambda.  Aside from Sync(), most
    // of that dispatching would tend to land in FailLocked().  The concurrency
    // is just worth keeping in mind for the rest of the current lambda is all.
    PostToSharedFidl([this] {
      // If the fuchsia::sysmem2::Allocator connection dies, so does this CodecImpl.
      //
      // In case of client code calling ~CodecImpl, this error handler is prevented from running by
      // synchronously unbinding in ~CodecImpl.
      sysmem_.Bind(
          std::move(tmp_sysmem_), shared_fidl_dispatcher_, [this](fidl::UnbindInfo unbind_info) {
            // This handler can't run until after sysmem_ is bound.
            ZX_DEBUG_ASSERT(was_logically_bound_);
            LogEvent(media_metrics::
                         StreamProcessorEvents2MigratedMetricDimensionEvent_SysmemChannelClosed);
            Fail("CodecImpl sysmem_ channel failed: %s", unbind_info.status_string());
          });
      ZX_DEBUG_ASSERT(!tmp_sysmem_);
      fuchsia_sysmem2::AllocatorSetDebugClientInfoRequest request;
      request.name() = GetObjectName(zx_process_self());
      request.id() = GetKoid(zx_process_self());
      auto set_debug_info_result = sysmem_->SetDebugClientInfo(std::move(request));
      if (!set_debug_info_result.is_ok()) {
        LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_FidlError);
        Fail("sysmem_->SetDebugClientInfo() failed: %s",
             set_debug_info_result.error_value().status_string());
        return;
      }

      zx_status_t status =
          binding_.Bind(std::move(tmp_interface_request_), shared_fidl_dispatcher_);
      if (status != ZX_OK) {
        LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_FidlError);
        Fail("binding_.Bind() failed");
        return;
      }
      ZX_DEBUG_ASSERT(!tmp_interface_request_);
    });

    input_constraints_ = CoreCodecBuildNewInputConstraints();

    ZX_DEBUG_ASSERT(input_constraints_);

    sent_buffer_constraints_version_ordinal_[kInputPort] =
        input_constraints_->buffer_constraints_version_ordinal();
    PostToSharedFidl([this] {
      // See "is_bound_checks" comment up top.
      if (binding_.is_bound()) {
        binding_.events().OnInputConstraints(fidl::Clone(*input_constraints_));
      }
    });
  });
}

void CodecImpl::EnableOnStreamFailed() {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  is_on_stream_failed_enabled_ = true;
}

void CodecImpl::AddInputBuffer_StreamControl(CodecBuffer::Info buffer_info,
                                             CodecVmoRange vmo_range) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  if (IsStopping()) {
    return;
  }
  // We must check, because __WARN_UNUSED_RESULT, and it's worth it for the
  // enforcement and consistency.
  if (!AddBufferCommon(std::move(buffer_info), std::move(vmo_range))) {
    return;
  }
}

void CodecImpl::SetInputBufferPartialSettings(
    fuchsia::media::StreamBufferPartialSettings input_settings) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  LogEvent(media_metrics::
               StreamProcessorEvents2MigratedMetricDimensionEvent_InputBufferAllocationStarted);
  PostToStreamControl([this, input_settings = std::move(input_settings)]() mutable {
    SetInputBufferPartialSettings_StreamControl(std::move(input_settings));
  });
}

void CodecImpl::SetInputBufferPartialSettings_StreamControl(
    fuchsia::media::StreamBufferPartialSettings input_partial_settings) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    ZX_DEBUG_ASSERT(sysmem_);
    SetInputBufferSettingsCommon(lock, &input_partial_settings);
  }  // ~lock
}

void CodecImpl::SetInputBufferSettingsCommon(
    std::unique_lock<std::mutex>& lock,
    fuchsia::media::StreamBufferPartialSettings* input_partial_settings) {
  if (IsStoppingLocked()) {
    return;
  }
  if (IsStreamActiveLocked()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("client sent SetInputBuffer*Settings() with stream active");
    return;
  }
  SetBufferSettingsCommon(lock, kInputPort, input_partial_settings, *input_constraints_);
}

void CodecImpl::SetOutputBufferSettingsCommon(
    std::unique_lock<std::mutex>& lock,
    fuchsia::media::StreamBufferPartialSettings* output_partial_settings) {
  if (!output_constraints_) {
    // invalid client behavior
    //
    // client must have received at least the initial OnOutputConstraints()
    // first before sending SetOutputBufferSettings().
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "client sent SetOutputBufferSettings()/SetOutputBufferPartialSettings()"
        " when no output_constraints_");
    return;
  }

  // For a mid-stream output format change, this also enforces that the client
  // can only catch up to the mid-stream format change once.  In other words,
  // if the client has already caught up to the mid-stream config change, the
  // client no longer has an excuse to re-configure again with a stream
  // active.
  //
  // There's a check in SetBufferSettingsCommonLocked() that ignores this
  // message if the client's buffer_constraints_version_ordinal is behind
  // last_required_buffer_constraints_version_ordinal_, which gets updated
  // under the same lock hold interval as the server's de-configuring of
  // output buffers.
  //
  // There's a check in SetBufferSettingsCommonLocked() that closes the
  // channel if the client is sending a buffer_constraints_version_ordinal
  // that's newer than the last sent_buffer_constraints_version_ordinal_.
  if (IsStreamActiveLocked() && IsOutputConfiguredLocked()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "client sent SetOutputBufferSettings()/SetOutputBufferPartialSettings()"
        " with IsStreamActiveLocked() + already-fully-configured output");
    return;
  }

  SetBufferSettingsCommon(lock, kOutputPort, output_partial_settings,
                          output_constraints_->buffer_constraints());
}

void CodecImpl::AddOutputBufferInternal(CodecBuffer::Info buffer_info, CodecVmoRange vmo_range) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());

  bool output_buffers_done_configuring =
      AddBufferCommon(std::move(buffer_info), std::move(vmo_range));
  if (output_buffers_done_configuring) {
    // The StreamControl domain _might_ be waiting for output to be configured.
    wake_stream_control_condition_.notify_all();
  }
}

void CodecImpl::SetOutputBufferPartialSettings(
    fuchsia::media::StreamBufferPartialSettings output_partial_settings) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  VLOGF("CodecImpl::SetOutputBufferPartialSettings");
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    if (!sysmem_) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "client sent SetOutputBufferPartialSettings() to a CodecImpl "
          "that lacks a sysmem_");
      return;
    }
    SetOutputBufferSettingsCommon(lock, &output_partial_settings);
  }  // ~lock
}

void CodecImpl::CompleteOutputBufferPartialSettings(uint64_t buffer_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);

    if (buffer_lifetime_ordinal % 2 == 0) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "CompleteOutputBufferPartialSettings client sent even "
          "buffer_lifetime_ordinal, but must be odd");
      return;
    }

    if (buffer_lifetime_ordinal != protocol_buffer_lifetime_ordinal_[kOutputPort]) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("CompleteOutputBufferPartialSettings bad buffer_lifetime_ordinal");
      return;
    }

    // If the server is not interested in the client's buffer_lifetime_ordinal,
    // the client's buffer_lifetime_ordinal won't match the server's
    // buffer_lifetime_ordinal_.  The client will probably later catch up.
    if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[kOutputPort]) {
      // The case that ends up here is when a client's output configuration
      // (whole or last part) is being ignored because it's not yet caught up
      // with last_required_buffer_constraints_version_ordinal_.

      // Ignore the client's message.  The client will probably catch up later.
      return;
    }

    if (!IsPortBuffersAtLeastPartiallyConfiguredLocked(kOutputPort)) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "CompleteOutputBufferPartialSettings seen without prior "
          "SetOutputBufferPartialSettings");
      return;
    }

    if (port_settings_[kOutputPort]->is_complete_seen_output()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "CompleteOutputBufferPartialSettings permitted exactly once "
          "after each SetOutputBufferPartialSettings");
      return;
    }

    // This will cause IsOutputConfiguredLocked() to start returning true.
    port_settings_[kOutputPort]->SetCompleteSeenOutput();
  }  // ~lock
  wake_stream_control_condition_.notify_all();
}

void CodecImpl::FlushEndOfStreamAndCloseStream(uint64_t stream_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_StreamFlushed);
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (!EnsureFutureStreamFlushSeenLocked(stream_lifetime_ordinal)) {
      return;
    }
  }
  PostToStreamControl([this, stream_lifetime_ordinal] {
    FlushEndOfStreamAndCloseStream_StreamControl(stream_lifetime_ordinal);
  });
}

void CodecImpl::FlushEndOfStreamAndCloseStream_StreamControl(uint64_t stream_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    AssertHeld(lock, lock_);
    if (IsStoppingLocked()) {
      return;
    }

    // We re-check some things which were already future-verified a different
    // way, to allow for flexibility in the future-tracking stuff to permit less
    // checking in the Output ordering domain (fidl_thread()) without
    // breaking overall verification of a flush.  Any checking in the Output
    // ordering domain is for the future-tracking's own convenience only. The
    // checking here is the real checking.

    if (!CheckStreamLifetimeOrdinalLocked(stream_lifetime_ordinal)) {
      return;
    }
    ZX_DEBUG_ASSERT(stream_lifetime_ordinal >= stream_lifetime_ordinal_);
    if (!IsStreamActiveLocked() || stream_lifetime_ordinal != stream_lifetime_ordinal_) {
      // TODO(dustingreen): epitaph
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "FlushEndOfStreamAndCloseStream() only valid on an active current "
          "stream (flush does not auto-create a new stream)");
      return;
    }
    // At this point we know that the stream is not discarded, and not already
    // flushed previously (because flush will discard the stream as there's
    // nothing more that the stream is permitted to do).
    ZX_DEBUG_ASSERT(IsStreamActiveLocked());
    ZX_DEBUG_ASSERT(stream_->stream_lifetime_ordinal() == stream_lifetime_ordinal);
    if (stream_->failure_seen()) {
      // Already reported to client.
      return;
    }
    if (!stream_->input_end_of_stream()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "FlushEndOfStreamAndCloseStream() is only permitted after "
          "QueueInputEndOfStream()");
      return;
    }
    auto ensure_closed = fit::defer([this, &lock] {
      // Now that flush is done (or we noticed a stream failure), we close the current stream
      // because there is not any subsequent message for the current stream that's valid.
      AssertHeld(lock, lock_);
      EnsureStreamClosed(lock);
    });
    while (!stream_->output_end_of_stream()) {
      if (stream_->failure_seen()) {
        return;
      }
      // While waiting, we'll continue to send OnOutputPacket(),
      // OnOutputConstraints(), and continue to process RecycleOutputPacket(),
      // until the client catches up to the latest config (as needed) and we've
      // started the send of output end_of_stream packet to the client.
      //
      // There is no way for the client to cancel a
      // FlushEndOfStreamAndCloseStream() short of closing the Codec channel.
      // Before long, the server will either send the OnOutputEndOfStream(), or
      // will send OnOmxStreamFailed(), or will close the Codec channel.  The
      // server must do one of those things before long (not allowed to get
      // stuck while flushing).
      //
      // Some core codecs have no way to report mid-stream input data corruption
      // errors or similar without it being a stream failure, so if there's any
      // stream error it turns into OnStreamFailed(). It's also permitted for a
      // server to set error_detected_ bool(s) on output packets and send
      // OnOutputEndOfStream() despite detected errors, but this is only a
      // reasonable behavior for the server if the server normally would detect
      // and report mid-stream input corruption errors without an
      // OnStreamFailed().
      // TODO(https://fxbug.dev/42119800): Cancel wait immediately on failure without waiting for
      // timeout.
      if (std::cv_status::timeout ==
          output_end_of_stream_seen_.wait_for(lock, std::chrono::seconds(5))) {
        LogEvent(media_metrics::
                     StreamProcessorEvents2MigratedMetricDimensionEvent_EndOfStreamTimeoutError);
        FailLocked("Timeout waiting for end of stream");
        break;
      }
    }
  }  // ~lock
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_CoreFlushed);
}

void CodecImpl::CloseCurrentStream(uint64_t stream_lifetime_ordinal, bool release_input_buffers,
                                   bool release_output_buffers) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (!EnsureFutureStreamCloseSeenLocked(stream_lifetime_ordinal)) {
      return;
    }
  }  // ~lock
  PostToStreamControl(
      [this, stream_lifetime_ordinal, release_input_buffers, release_output_buffers] {
        CloseCurrentStream_StreamControl(stream_lifetime_ordinal, release_input_buffers,
                                         release_output_buffers);
      });
}

void CodecImpl::CloseCurrentStream_StreamControl(uint64_t stream_lifetime_ordinal,
                                                 bool release_input_buffers,
                                                 bool release_output_buffers) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  std::unique_lock<std::mutex> lock(lock_);
  AssertHeld(lock, lock_);
  if (IsStoppingLocked()) {
    return;
  }
  EnsureStreamClosed(lock);
  if (release_input_buffers) {
    EnsureBuffersNotConfigured(lock, kInputPort);
  }
  if (release_output_buffers) {
    // This prevents RecycleOutputPacket() on output domain running concurrently with
    // EnsureBuffersNotConfigured(kOutputPort) on StreamControl.  All EnsureBuffersNotConfigured()
    // called from StreamControl must call this first.
    StartIgnoringClientOldOutputConfig(lock);
    EnsureBuffersNotConfigured(lock, kOutputPort);
  }
}

void CodecImpl::Sync(SyncCallback callback) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  // By posting to StreamControl ordering domain, we sync both Output ordering
  // domain (on fidl_thread()) and the StreamControl ordering domain.
  //
  // If the posted task doesn't run because stream_control_queue_.StopAndClear()
  // happened/happens, it doesn't matter because the whole channel will be
  // closing before long.
  //
  // The callback has affinity with fidl_thread(), including the destructor.
  // This is problematic with respect to the
  // stream_control_queue_.StopAndClear() called on StreamControl domain during
  // unbind. Without special handling, that StopAndClear() would try to delete
  // callback on the StreamControl domain instead of on the fidl_thread().  To
  // prevent that, we ensure that deletion of the lambda without running the
  // lambda will still post destruction of callback to fidl_thread(), and this
  // posting will queue before the lamda that runs
  // shared_fidl_queue_.StopAndClear().
  PostToStreamControl([this, callback_holder = ThreadSafeDeleter<SyncCallback>(
                                 &shared_fidl_queue_, std::move(callback))]() mutable {
    Sync_StreamControl(std::move(callback_holder));
  });
}

void CodecImpl::Sync_StreamControl(ThreadSafeDeleter<SyncCallback> callback_holder) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  if (IsStopping()) {
    // In this case, we rely on ThreadSafeDeleter to delete callback on fidl_thread().
    //
    // The response won't be sent, which is appropriate - the channel is getting closed soon
    // instead, and the client has to tolerate that.
    //
    // ~callback_holder
    return;
  }
  // We post back to FIDL thread to respond to ensure we're not racing with
  // channel close which could lead to attempting to send to handle value 0
  // which can cause process termination.  Also, because this fences
  // BufferAllocation clean close which itself is done async from StreamControl
  // to FIDL in some cases.
  PostToSharedFidl([this, callback_holder = std::move(callback_holder)]() mutable {
    ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
    // call the held callback
    callback_holder.held()();
  });
}

void CodecImpl::RecycleOutputPacket(fuchsia::media::PacketHeader available_output_packet) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  if (kLogTimestampDelay) {
    LOG(INFO, "RecycleOutputPacket");
  }
  CodecPacket* packet = nullptr;
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    if (!available_output_packet.has_buffer_lifetime_ordinal()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("output packet is missing buffer lifetime ordinal");
      return;
    }

    if (!CheckOldBufferLifetimeOrdinalLocked(kOutputPort,
                                             available_output_packet.buffer_lifetime_ordinal())) {
      return;
    }
    if (available_output_packet.buffer_lifetime_ordinal() < buffer_lifetime_ordinal_[kOutputPort]) {
      // ignore arbitrarily-stale required by protocol
      //
      // Thanks to even values from the client being prohibited, this also
      // covers mid-stream output config change where the server has already
      // de-configured output buffers but the client doesn't know about that
      // yet. We include that case here by setting
      // buffer_lifetime_ordinal_[kOutputPort] to the next even value
      // when de-configuring output server-side until the client has
      // re-configured output.
      return;
    }
    ZX_DEBUG_ASSERT(available_output_packet.buffer_lifetime_ordinal() ==
                    buffer_lifetime_ordinal_[kOutputPort]);
    if (!IsOutputConfiguredLocked()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "client sent RecycleOutputPacket() for buffer_lifetime_ordinal that "
          "isn't fully configured yet - bad client behavior");
      return;
    }
    if (!available_output_packet.has_packet_index()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("output packet is missing packet index");
      return;
    }
    ZX_DEBUG_ASSERT(IsOutputConfiguredLocked());
    if (available_output_packet.packet_index() >= all_packets_[kOutputPort].size()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("out of range packet_index from client in RecycleOutputPacket()");
      return;
    }
    uint32_t packet_index = available_output_packet.packet_index();
    if (all_packets_[kOutputPort][packet_index]->is_free()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "packet_index already free at protocol level - invalid client "
          "message");
      return;
    }
    // Mark free at protocol level.
    all_packets_[kOutputPort][packet_index]->SetFree(true);

    // Before handing the packet to the core codec, clear some fields that the
    // core codec is expected to set (or optionally set in the case of
    // timestamp_ish).  In addition to these parameters, a core codec can emit
    // output config changes via onCoreCodecMidStreamOutputConstraintsChange().
    packet = all_packets_[kOutputPort][packet_index].get();
    packet->ClearStartOffset();
    packet->ClearValidLengthBytes();
    packet->ClearTimestampIsh();
  }

  // Recycle to core codec.
  CoreCodecRecycleOutputPacket(packet);
}

void CodecImpl::QueueInputFormatDetails(uint64_t stream_lifetime_ordinal,
                                        fuchsia::media::FormatDetails format_details) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (!EnsureFutureStreamSeenLocked(stream_lifetime_ordinal)) {
      return;
    }
  }  // ~lock

  if (!format_details.has_format_details_version_ordinal()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail(
        "client QueueInputFormatDetails(): Format details have no version "
        "ordinal.");
    return;
  }

  PostToStreamControl(
      [this, stream_lifetime_ordinal, format_details = std::move(format_details)]() mutable {
        QueueInputFormatDetails_StreamControl(stream_lifetime_ordinal, std::move(format_details));
      });
}

// TODO(dustingreen): Need test coverage for this method, to cover at least
// the same format including OOB bytes as were specified during codec creation,
// and codec creation with no OOB bytes then this method setting OOB bytes (not
// the ideal client usage pattern in the long run since the CreateDecoder()
// might decline to provide a optimized but partial Codec implementation, but
// should be allowed nonetheless).
void CodecImpl::QueueInputFormatDetails_StreamControl(
    uint64_t stream_lifetime_ordinal, fuchsia::media::FormatDetails format_details) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);

  std::unique_lock<std::mutex> lock(lock_);
  AssertHeld(lock, lock_);
  if (IsStoppingLocked()) {
    return;
  }
  if (!CheckStreamLifetimeOrdinalLocked(stream_lifetime_ordinal)) {
    return;
  }

  if (!CheckWaitEnsureInputConfigured(lock)) {
    stream_->AssertHeld(this);
    ZX_DEBUG_ASSERT(IsStoppingLocked() || !stream_ || stream_->future_discarded());
    return;
  }

  ZX_DEBUG_ASSERT(stream_lifetime_ordinal >= stream_lifetime_ordinal_);
  if (stream_lifetime_ordinal > stream_lifetime_ordinal_) {
    if (!StartNewStream(lock, stream_lifetime_ordinal)) {
      return;
    }
  }
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal == stream_lifetime_ordinal_);
  if (stream_->failure_seen()) {
    // Already reported to client.
    return;
  }
  if (stream_->input_end_of_stream()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("QueueInputFormatDetails() after QueueInputEndOfStream() unexpected");
    return;
  }
  stream_->AssertHeld(this);
  if (stream_->future_discarded()) {
    // No reason to handle since the stream is future-discarded.
    return;
  }
  stream_->SetInputFormatDetails(
      std::make_unique<fuchsia::media::FormatDetails>(std::move(format_details)));
  // SetOobConfigPending(true) to ensure oob_config_pending() is true.
  //
  // This call is needed only to properly handle a call to
  // QueueInputFormatDetails() mid-stream.  For new streams that lack any calls
  // to QueueInputFormatDetails() before an input packet arrives, the
  // oob_config_pending() will already be true because it starts true for a new
  // stream.  For QueueInputFormatDetails() at the start of a stream before any
  // packets, oob_config_pending() will already be true.
  //
  // For decoders this is basically a pending oob_bytes.  For encoders
  // this pending config change can potentially include uncompressed format
  // details, if mid-stream format change is supported by the encoder.
  stream_->SetOobConfigPending(true);
}

void CodecImpl::QueueInputPacket(fuchsia::media::Packet packet) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (IsStoppingLocked()) {
      return;
    }
    if (!packet.has_stream_lifetime_ordinal()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "client QueueInputPacket() with packet that has no stream lifetime "
          "ordinal");
      return;
    }
    if (!EnsureFutureStreamSeenLocked(packet.stream_lifetime_ordinal())) {
      return;
    }
  }  // ~lock
  if (kLogTimestampDelay) {
    LOG(INFO, "input timestamp: has: %d value: 0x%" PRIx64, packet.has_timestamp_ish(),
        packet.has_timestamp_ish() ? packet.timestamp_ish() : 0);
  }
  PostToStreamControl([this, packet = std::move(packet)]() mutable {
    QueueInputPacket_StreamControl(std::move(packet));
  });
}

void CodecImpl::QueueInputPacket_StreamControl(fuchsia::media::Packet packet) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  ZX_DEBUG_ASSERT(packet.has_stream_lifetime_ordinal());

  if (!packet.has_header()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() with packet has no header");
    return;
  }
  fuchsia::media::PacketHeader temp_header_copy = fidl::Clone(packet.header());

  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    AssertHeld(lock, lock_);
    if (IsStoppingLocked()) {
      return;
    }

    // Unless we cancel this cleanup, we'll free the input packet back to the
    // client.
    auto send_free_input_packet_locked =
        fit::defer([this, header = std::move(temp_header_copy)]() mutable {
          // Mute sending this if FailLocked() was called previously, in case
          // the reason we're here is something horribly wrong with the packet
          // header. This way we avoid repeating gibberish back to the client.
          // While that gibberish might be a slight clue for debugging in some
          // cases, it's not valid protocol, so don't send it.  If
          // IsStoppingLocked(), the Codec channel will close soon, making this
          // response unnecessary.
          if (!IsStoppingLocked()) {
            SendFreeInputPacketLocked(std::move(header));
          }
        });

    if (!packet.header().has_buffer_lifetime_ordinal()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "client QueueInputPacket() with header that has no buffer lifetime "
          "ordinal");
      return;
    }
    if (!CheckOldBufferLifetimeOrdinalLocked(kInputPort,
                                             packet.header().buffer_lifetime_ordinal())) {
      return;
    }

    if (!packet.has_stream_lifetime_ordinal()) {
      FailLocked("client QueueInputPacket() without packet stream_lifetime_ordinal.");
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      return;
    }
    if (!CheckStreamLifetimeOrdinalLocked(packet.stream_lifetime_ordinal())) {
      return;
    }

    if (!CheckWaitEnsureInputConfigured(lock)) {
      stream_->AssertHeld(this);
      ZX_DEBUG_ASSERT(IsStoppingLocked() || (stream_ && stream_->future_discarded()));
      return;
    }

    // For input, mid-stream config changes are not a thing and input buffers
    // are never unilaterally de-configured by the Codec server.
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kInputPort] ==
                    port_settings_[kInputPort]->buffer_lifetime_ordinal());

    // For this message we're strict re. buffer_lifetime_ordinal.
    //
    // In contrast to output, the server doesn't use even values to track config
    // changes that the client doesn't know about yet, since the server can't
    // unilaterally demand any changes to the input settings after initially
    // specifying the input constraints.
    //
    // One could somewhat-convincingly argue that this field in this particular
    // message is a bit pointless, but it might serve to detect client-side
    // bugs faster thanks to this check.
    if (packet.header().buffer_lifetime_ordinal() !=
        port_settings_[kInputPort]->buffer_lifetime_ordinal()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client QueueInputPacket() with invalid buffer_lifetime_ordinal.");
      return;
    }

    ZX_DEBUG_ASSERT(packet.stream_lifetime_ordinal() >= stream_lifetime_ordinal_);

    if (packet.stream_lifetime_ordinal() > stream_lifetime_ordinal_) {
      // This case implicitly starts a new stream.  If the client wanted to
      // ensure that the old stream would be fully processed, the client would
      // have sent FlushEndOfStreamAndCloseStream() previously, whose
      // processing (previous to reaching here) takes care of the flush.
      //
      // Start a new stream, synchronously.
      if (!StartNewStream(lock, packet.stream_lifetime_ordinal())) {
        return;
      }
    }
    ZX_DEBUG_ASSERT(packet.stream_lifetime_ordinal() == stream_lifetime_ordinal_);

    if (!packet.header().has_packet_index()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client QueueInputPacket() with packet has no packet index");
      return;
    }
    if (packet.header().packet_index() >= all_packets_[kInputPort].size()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "client QueueInputPacket() with packet_index out of range - "
          "packet_index: %u size: %u",
          packet.header().packet_index(), all_packets_[kInputPort].size());
      return;
    }
    if (!packet.has_buffer_index()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client QueueInputPacket() with packet has no buffer index");
      return;
    }
    if (packet.buffer_index() >= all_buffers_[kInputPort].size()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client QueueInputPacket() with buffer_index out of range");
      return;
    }

    // Protocol check re. free/busy coherency.  This applies to packets only,
    // not buffers.
    if (!all_packets_[kInputPort][packet.header().packet_index()]->is_free()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client QueueInputPacket() with packet_index !free");
      return;
    }

    if (stream_->input_end_of_stream()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("QueueInputPacket() after QueueInputEndOfStream() unexpeted");
      return;
    }

    if (stream_->failure_seen()) {
      // Already reported to client.
      return;
    }

    stream_->AssertHeld(this);
    if (stream_->future_discarded()) {
      // Don't queue to core codec.  The stream_ may have never fully started,
      // or may have been future-discarded since.  Either way, skip queueing to
      // the core codec.
      //
      // If the stream didn't fully start - as in, the client moved on to
      // another stream before fully configuring output, then the core codec is
      // not presently in a state compatible with queueing input, but the Codec
      // interface is.  So in that case, we must avoid queueing to the core
      // codec for correctness.
      //
      // If the stream was just future-discarded after fully starting, then this
      // is just an optimization to avoid giving the core codec more work to do
      // for a stream the client has already discarded.
      //
      // ~send_free_input_packet_locked
      // ~lock
      return;
    }

    all_packets_[kInputPort][packet.header().packet_index()]->SetFree(false);

    // Sending OnFreeInputPacket() will happen later instead, when the core
    // codec gives back the packet.
    send_free_input_packet_locked.cancel();
  }  // ~lock

  if (stream_->oob_config_pending()) {
    HandlePendingInputFormatDetails();
    stream_->SetOobConfigPending(false);
  }

  CodecPacket* core_codec_packet = all_packets_[kInputPort][packet.header().packet_index()].get();
  core_codec_packet->SetBuffer(all_buffers_[kInputPort][packet.buffer_index()].get());
  if (!packet.has_start_offset()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() with packet has no start offset");
    return;
  }
  core_codec_packet->SetStartOffset(packet.start_offset());
  if (!packet.has_valid_length_bytes()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() with packet has no valid length bytes");
    return;
  }
  core_codec_packet->SetValidLengthBytes(packet.valid_length_bytes());
  if (packet.has_timestamp_ish()) {
    core_codec_packet->SetTimstampIsh(packet.timestamp_ish());
  } else {
    core_codec_packet->ClearTimestampIsh();
  }

  if (core_codec_packet->valid_length_bytes() <= 0) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() with valid_length_bytes 0 - not allowed");
    return;
  }
  if (core_codec_packet->start_offset() + core_codec_packet->valid_length_bytes() <
      core_codec_packet->start_offset()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() start_offset + valid_length_bytes overflow");
    return;
  }
  if (core_codec_packet->start_offset() + core_codec_packet->valid_length_bytes() >
      core_codec_packet->buffer()->size()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    Fail("client QueueInputPacket() with packet end > buffer size");
    return;
  }

  // Flush the data out to RAM if needed.
  if (IsCoreCodecHwBased(kInputPort) &&
      port_settings_[kInputPort]->coherency_domain() == fuchsia_sysmem2::CoherencyDomain::kCpu) {
    // This flushes only the portion of the buffer that the packet is
    // referencing.
    core_codec_packet->CacheFlush();
  }

  // We don't need to be under lock for this, because the fact that we're on the
  // StreamControl domain is enough to guarantee that any other control of the
  // core codec will occur after this.
  CoreCodecQueueInputPacket(core_codec_packet);
}

void CodecImpl::QueueInputEndOfStream(uint64_t stream_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  LogEvent(
      media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_StreamEndOfStreamInput);
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (!EnsureFutureStreamSeenLocked(stream_lifetime_ordinal)) {
      return;
    }
  }  // ~lock
  PostToStreamControl([this, stream_lifetime_ordinal] {
    QueueInputEndOfStream_StreamControl(stream_lifetime_ordinal);
  });
}

void CodecImpl::QueueInputEndOfStream_StreamControl(uint64_t stream_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    AssertHeld(lock, lock_);
    if (IsStoppingLocked()) {
      return;
    }
    if (!CheckStreamLifetimeOrdinalLocked(stream_lifetime_ordinal)) {
      return;
    }

    if (!CheckWaitEnsureInputConfigured(lock)) {
      stream_->AssertHeld(this);
      ZX_DEBUG_ASSERT(IsStoppingLocked() || (stream_ && stream_->future_discarded()));
      return;
    }

    ZX_DEBUG_ASSERT(stream_lifetime_ordinal >= stream_lifetime_ordinal_);
    if (stream_lifetime_ordinal > stream_lifetime_ordinal_) {
      // We start a new stream given an end-of-stream for a stream we've not
      // seen before, since allowing empty streams to not be errors may be nicer
      // to use.
      if (!StartNewStream(lock, stream_lifetime_ordinal)) {
        return;
      }
    }

    if (stream_->failure_seen()) {
      // Already reported to client.
      return;
    }

    if (stream_->input_end_of_stream()) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked("client already sent QueueInputEndOfStream() for this stream");
      return;
    }
    stream_->SetInputEndOfStream();

    stream_->AssertHeld(this);
    if (stream_->future_discarded()) {
      // Don't queue to core codec.  The stream_ may have never fully started,
      // or may have been future-discarded since. Either way, skip queueing to
      // core codec. We only really must do this because the stream may not have
      // ever fully started, in the case where the client moves on to a new
      // stream before catching up to latest output config.
      return;
    }
  }  // ~lock

  CoreCodecQueueInputEndOfStream();
}

zx_status_t CodecImpl::Pin(uint32_t options, const zx::vmo& vmo, uint64_t offset, uint64_t size,
                           zx_paddr_t* addrs, size_t addrs_count, zx::pmt* pmt) {
  ZX_DEBUG_ASSERT(*core_codec_bti_);
  return core_codec_bti_->pin(options, vmo, offset, size, addrs, addrs_count, pmt);
}

bool CodecImpl::CheckWaitEnsureInputConfigured(std::unique_lock<std::mutex>& lock) {
  // Ensure/finish input configuration.
  if (!IsPortBuffersAtLeastPartiallyConfiguredLocked(kInputPort)) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "client QueueInput*() with input buffers not at least partially "
        "configured");
    return false;
  }
  ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kInputPort] % 2 == 1);
  // The client is required to know that sysmem is in fact done allocating the
  // BufferCollection successfully before the client sends
  // QueueInput...StreamControl.  We can't trust a client to necessarily get
  // that right however, so rather than just getting stuck indefinitely in that
  // case, we detect by asking sysmem to verify that it has allocated the
  // BufferCollection successfully.  This verification happens async, but will
  // shortly cause WaitEnsureSysmemReadyOnInput() to return and
  // IsStoppingLocked() to return true if verification fails.
  if (!IsInputConfiguredLocked()) {
    PostToSharedFidl([this, buffer_lifetime_ordinal = buffer_lifetime_ordinal_[kInputPort]] {
      std::unique_lock<std::mutex> lock(lock_);
      if (IsStoppingLocked()) {
        return;
      }
      if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[kInputPort]) {
        // stale; no problem; old buffers were allocated fine and client already
        // moved on after that.
        return;
      }
      // Else previous buffer_lifetime_ordinal check would have noticed.
      ZX_DEBUG_ASSERT(port_settings_[kInputPort]);
      // paranoid check - assert above believed to be valid
      if (!port_settings_[kInputPort]) {
        return;
      }
      auto& buffer_collection = port_settings_[kInputPort]->buffer_collection();
      // Else IsStoppingLocked() check above would have returned.
      ZX_DEBUG_ASSERT(!!buffer_collection && buffer_collection->is_valid());
      // paranoid check - assert above believed to be valid
      if (!buffer_collection || !buffer_collection->is_valid()) {
        return;
      }
      (*buffer_collection)
          ->CheckAllBuffersAllocated()
          .Then([this, buffer_lifetime_ordinal](
                    fidl::Result<::fuchsia_sysmem2::BufferCollection::CheckAllBuffersAllocated>&
                        result) {
            std::unique_lock<std::mutex> lock(lock_);
            if (IsStoppingLocked()) {
              return;
            }
            if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[kInputPort]) {
              // stale; no problem; old buffers were allocated fine and client
              // already moved on after that.
              return;
            }
            if (result.is_error()) {
              // This will cause any in-progress WaitEnsureSysmemReadyOnInput()
              // to return shortly and IsStoppingLocked() will be true.
              LogEvent(media_metrics::
                           StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
              FailLocked(
                  "Probably client did QueueInput* before the client "
                  "determined that sysmem was done successfully allocating "
                  "buffers after most recent SetInputBufferPartialSettings(): %d",
                  result.error_value());
              return;
            }
          });
    });
    if (!WaitEnsureSysmemReadyOnInput(lock)) {
      ZX_DEBUG_ASSERT(IsStoppingLocked());
      return false;
    }
  }
  if (!IsInputConfiguredLocked()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("client QueueInput*() with input buffers not configured");
    return false;
  }
  return true;
}

void CodecImpl::UnbindLocked() {
  // We must have first gotten far enough through BindAsync() before calling
  // UnbindLocked().
  ZX_DEBUG_ASSERT(was_logically_bound_);

  if (was_unbind_started_) {
    // Ignore the second trigger if we have a near-simultaneous failure from
    // StreamControl thread (for example) and from fidl_thread() (for
    // example).  The first will start unbinding, and the second will be
    // ignored.  Since completion of the Unbind() call doesn't imply anything
    // about how done the unbind is, there's no need for the second caller to
    // be blocked waiting for the first caller's unbind to be done.
    return;
  }

  if (codec_admission_) {
    codec_admission_->SetCodecIsClosing();
  }

  // Tell StreamControl to not start any more work.
  was_unbind_started_ = true;
  wake_stream_control_condition_.notify_all();

  // Unbind() / UnbindLocked() can be called from any thread.
  //
  // Regardless of what thread UnbindLocked() is called on, "this" will remain
  // allocated at least until the caller of UnbindLocked() releases lock_.
  //
  // In all cases, this posted lambda runs after BindAsync()'s work that's
  // posted to StreamControl, because any/all calls to UnbindLocked() happen
  // after BindAsync() has posted to StreamControl.
  //
  // We know the stream_control_queue_ isn't stopped yet, because the present
  // method is idempotent and the lambda being posted just below has the only
  // call to stream_control_queue_.StopAndClear().
  ZX_DEBUG_ASSERT(!stream_control_queue_.is_stopped());
  PostToStreamControl([this] {
    // At this point we know that no more streams will be started by
    // StreamControl ordering domain (thanks to was_unbind_started_ /
    // IsStoppingLocked() checks), but lambdas posted to the StreamControl
    // ordering domain (by the fidl_thread() or by core codec) may still
    // be creating other activity such as posting lambdas to StreamControl or
    // fidl_thread().
    {  // scope lock
      std::unique_lock<std::mutex> lock(lock_);
      AssertHeld(lock, lock_);
      // Stop core codec associated with this CodecImpl, partly to make sure it
      // stops running code that could make calls into this CodecImpl, and
      // partly to ensure the core codec isn't in the middle of anything when it
      // gets deleted.
      //
      // We know the core codec won't start more activity because the core codec
      // isn't allowed to initiate actions while there's no active stream, and
      // because no new active stream will be created.  All _StreamControl
      // methods check IsStoppingLocked() at the start, and the StreamControl
      // ordering domain is the only domain that ever starts a stream.
      //
      // We intentionally don't check for IsStoppingLocked() in protocol
      // dispatch methods running on fidl_thread(). For example the codec
      // must tolerate calls to configure buffers after EnsureStreamClosed()
      // here.  The Unbind() later is what silences the protocol message
      // dispatch methods.  Checking for IsStoppingLocked() in protocol dispatch
      // methods would only decrease the probability of certain event orderings,
      // not eliminate those orderings, so it's actually better to let them
      // happen to get more coverage of those orderings.
      if (is_core_codec_init_called_) {
        EnsureStreamClosed(lock);
        EnsureBuffersNotConfigured(lock, kInputPort);
      }

      // Because the current path is the only path that sets this bool to true,
      // and the current path is run-once.
      ZX_DEBUG_ASSERT(!is_stream_control_done_);
      // Because stream_control_done_ is false, and ~CodecImpl waits for
      // is_stream_control_done_ true before shared_fidl_queue_.StopAndClear().
      ZX_DEBUG_ASSERT(!shared_fidl_queue_.is_stopped());

      // We do this from here so we know that this thread won't run any more
      // tasks after the currently-running task.
      //
      // The currently-running StreamControl task (this method) still gets to
      // run to completion.
      //
      // TODO(dustingreen): We probably could lean more heavily on this Quit()
      // and do less checking of IsStoppingLocked() in StreamControl tasks.
      // This TODO is not meant to imply that all current checking of
      // IsStoppingLocked() is ok to remove (less, not none).
      stream_control_loop_.Quit();

      // This deletes any further tasks already queued to StreamControl, and will immediately
      // delete any additional tasks that try to queue to StreamControl.  We also need to ensure the
      // first time stream_control_queue_.StopAndClear() runs is on stream_control_thread_, per
      // ClosureQueue's usage rules.
      stream_control_queue_.StopAndClear();

      // We're ready to let EnsureUnbindCompleted() and ~CodecImpl do the rest.
      //
      // The core codec has been stopped, so it has no current stream.  The core codec is required
      // to be delete-able when it has no current stream, and required not to asynchronously post
      // more work to the CodecImpl (because calling onCoreCodec... methods is not allowed when
      // there is no current stream).
      //
      // The binding_.Unbind() will run during EnsureUnbindCompleted() on the FIDL thread, so no
      // more FIDL dispatching to this CodecImpl after that.
      //
      // The stream_control_loop_.JoinThreads() will run during ~CodecImpl, so no more activity from
      // stream_control_thread_ after that.
      //
      // Anything posted using PostToSharedFidl() can be deleted instead of run since the whole
      // CodecImpl is going away, and shared_fidl_queue_ makes it safe for ~CodecImpl to complete
      // without needing to wait/fence past previously-posted labmdas to FIDL thread.
      is_stream_control_done_ = true;
      // Must notify_all() under lock_ in this case since ~CodecImpl can run as soon as
      // stream_control_done_ = true just above.
      stream_control_done_condition_.notify_all();

      // If we're not running from ~CodecImpl, we need to run the owner_error_handler_ on the FIDL
      // thread, which will in turn call ~CodecImpl.  If we are running from ~CodecImpl, then we're
      // already on the FIDL thread, and this posted work won't run thanks to shared_fidl_queue_
      // just deleting the posted task instead, in which case the owner_error_handler_ just gets
      // deleted instead of running (the usual semantics in response to unsolicited destruction).
      //
      // Must post under lock_ in this case else ~CodecImpl can have already finished as soon as
      // stream_control_done_ = true above.
      PostToSharedFidl([this, client_error_handler = std::move(owner_error_handler_)] {
        ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
        // We go ahead and finish up the un-binding aspects (because we can free up
        // resources prior to the client code potentially running ~CodecImpl async
        // later).
        //
        // However, this doesn't finish up aspects related to ordering release of
        // resources before acquisition of new resources.  In particular, this call
        // unbinds the channel, but intentionally doesn't close the channel itself until
        // after ~CodecImpl and after ~CodecAdmission.  The intent is to prevent the
        // possibility that overly-agressive client retries on channel closure by the
        // server could build up many CodecImpl instances, even if different instances
        // happen to use different FIDL threads (also potentially different than FIDL
        // thread on which a new CodecAdmission is created). By only closing the channel
        // itself as the last thing after all other cleanup is fully done, we don't
        // trigger the client to create a new CodecImpl while the old one still exists.
        EnsureUnbindCompleted();
        // This call is expected to run ~CodecImpl, either synchronously during this
        // call or shortly later async.
        client_error_handler();
      });
    }  // ~lock

    // "this" will be deleted shortly async when lambda posted just above runs, or we're returning
    // back to rest of ~CodecImpl, or ~CodecImpl is racing/running separately and completing
    // immediately after ~lock just above.  Regardless, done here.
    return;
  });
  // "this" remains allocated until caller releases lock_.
}

void CodecImpl::Unbind() {
  std::lock_guard<std::mutex> lock(lock_);
  UnbindLocked();
  // ~lock
  //
  // "this" may be deleted very shortly after ~lock, depending on what thread
  // Unbind() is called from.
}

void CodecImpl::EnsureUnbindCompleted() {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  ZX_DEBUG_ASSERT(was_logically_bound_);
  if (was_unbind_completed_) {
    return;
  }
  // Or will be, before this method returns.
  was_unbind_completed_ = true;

  // Unbind from the channel so we won't see any more incoming FIDL messages. This binding doesn't
  // own "this".
  //
  // The Unbind() stops any additional FIDL dispatching re. this CodecImpl.
  if (binding_.is_bound()) {
    codec_to_close_ = binding_.Unbind().TakeChannel();
  }

  // This isn't strictly necessary, but since we can potentially delete a queued
  // task here (before a client-called ~CodecImpl), we go ahead and do that now.
  //
  // This is partly a very minor potential resource deletion, and partly so we
  // get a nicer stack if anything should go wrong during that deletion; partly
  // so we get a nicer stack if somehow the JoinThreads() gets stuck (it
  // shouldn't since Quit() already happened).
  stream_control_loop_.JoinThreads();
  stream_control_loop_.Shutdown();

  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);

    EnsureBuffersNotConfigured(lock, kOutputPort);

    // By this point both PortSettings should have already been deleted.
    ZX_DEBUG_ASSERT(!port_settings_[kInputPort]);
    ZX_DEBUG_ASSERT(!port_settings_[kOutputPort]);

    // Unbind the sysmem_ fuchsia::sysmem2::Allocator connection - this also ensures that any
    // in-flight requests' completions will not run.
    //
    // We check sysmem_.is_valid() here because the current fidl_thread work item can be be running
    // ~CodecImpl (or fidl_thread part of UnbindLocked) before the remainder of BindAsync that would
    // make sysmem_ valid. In that case, the current work item will finish running, but remaining
    // work items on shared_fidl_queue_ won't run because of the shared_fidl_queue_.StopAndClear()
    // below. So in that case, sysmem_ will remain ~is_valid() until destruction.
    if (sysmem_.is_valid()) {
      (void)sysmem_.UnbindMaybeGetEndpoint();
    }
  }  // ~lock

  // Any previously-posted tasks via shared_fidl_queue_ are deleted here without running.
  //
  // If we're shutting down because UnbindLocked() was run first upon discovery of an
  // internally-noticed error, then previously-queued sending of FIDL messages on the FIDL thread
  // already ran before the EnsureUnbindCompleted(), which was posted after the sends.
  //
  // If we're running ~CodecImpl because the client code is just deleting CodecImpl for whatever
  // client-initiated reason, then previously queueud sending of FIDL messages can be just deleted
  // here without the sends actually occurring, which is fine since in that case the client code
  // has no particular expectation that any particular messages were sent before deletion vs. not
  // getting sent due to deletion.
  shared_fidl_queue_.StopAndClear();
}

fuchsia::mediacodec::SecureMemoryMode CodecImpl::OutputSecureMemoryMode() {
  if (!IsDecoder() && !IsDecryptor()) {
    return fuchsia::mediacodec::SecureMemoryMode::OFF;
  }
  if (IsDecoder()) {
    if (!decoder_params().has_secure_output_mode()) {
      return fuchsia::mediacodec::SecureMemoryMode::OFF;
    }
    return decoder_params().secure_output_mode();
  } else {
    ZX_DEBUG_ASSERT(IsDecryptor());
    if (!decryptor_params().has_require_secure_mode()) {
      return fuchsia::mediacodec::SecureMemoryMode::OFF;
    }
    return decryptor_params().require_secure_mode() ? fuchsia::mediacodec::SecureMemoryMode::ON
                                                    : fuchsia::mediacodec::SecureMemoryMode::OFF;
  }
}

fuchsia::mediacodec::SecureMemoryMode CodecImpl::InputSecureMemoryMode() {
  if (!IsDecoder()) {
    return fuchsia::mediacodec::SecureMemoryMode::OFF;
  }
  if (!decoder_params().has_secure_input_mode()) {
    return fuchsia::mediacodec::SecureMemoryMode::OFF;
  }
  return decoder_params().secure_input_mode();
}

fuchsia::mediacodec::SecureMemoryMode CodecImpl::PortSecureMemoryMode(CodecPort port) {
  if (port == kOutputPort) {
    return OutputSecureMemoryMode();
  } else {
    ZX_DEBUG_ASSERT(port == kInputPort);
    return InputSecureMemoryMode();
  }
}

bool CodecImpl::IsPortSecureRequired(CodecPort port) {
  // Return false for DYNAMIC, if/when we add that.
  return PortSecureMemoryMode(port) == fuchsia::mediacodec::SecureMemoryMode::ON;
}

bool CodecImpl::IsPortSecurePermitted(CodecPort port) {
  // Return true for DYNAMIC, if/when we add that.
  return PortSecureMemoryMode(port) != fuchsia::mediacodec::SecureMemoryMode::OFF;
}

bool CodecImpl::IsStreamActiveLocked() {
  ZX_DEBUG_ASSERT(!!stream_ == (stream_lifetime_ordinal_ % 2 == 1));
  return !!stream_;
}

void CodecImpl::SetBufferSettingsCommon(
    std::unique_lock<std::mutex>& lock, CodecPort port,
    fuchsia::media::StreamBufferPartialSettings* partial_settings,
    const fuchsia::media::StreamBufferConstraints& stream_constraints) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  ZX_DEBUG_ASSERT(!IsStoppingLocked());

  if (!partial_settings->has_buffer_lifetime_ordinal()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("partial_settings do not have buffer lifetime ordinal");
    return;
  }
  if (!partial_settings->has_buffer_constraints_version_ordinal()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("partial_settings do not have buffer constraints version ordinal");
    return;
  }
  if ((!partial_settings->has_sysmem_token() || !partial_settings->sysmem_token().is_valid()) &&
      (!partial_settings->has_sysmem2_token() || !partial_settings->sysmem2_token().is_valid())) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("partial_settings missing valid sysmem2_token (sysmem_token also missing)");
    return;
  }
  if (partial_settings->has_sysmem_token() && partial_settings->has_sysmem2_token()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "partial_settings must have only sysmem2_token or sysmem_token, not both (prefer sysmem2_token)");
    return;
  }
  if (partial_settings->has_sysmem_token() && !partial_settings->has_sysmem2_token()) {
    LOG(WARNING,
        "client is using deprecated sysmem_token field; client should switch to sysmem2_token field");
    // Token channels served by sysmem serve both sysmem(1) and sysmem2 BufferCollectionToken on the
    // same channel.
    partial_settings->set_sysmem2_token(
        fidl::InterfaceHandle<fuchsia::sysmem2::BufferCollectionToken>(
            partial_settings->mutable_sysmem_token()->TakeChannel()));
    partial_settings->clear_sysmem_token();
  }

  ZX_DEBUG_ASSERT(
      !port_settings_[port] ||
      (buffer_lifetime_ordinal_[port] >= port_settings_[port]->buffer_lifetime_ordinal() &&
       buffer_lifetime_ordinal_[port] <= port_settings_[port]->buffer_lifetime_ordinal() + 1));

  // Extract buffer_lifetime_ordinal and buffer_constraints_version_ordinal from
  // StreamBufferPartialSettings
  // is providing.
  uint64_t buffer_lifetime_ordinal = partial_settings->buffer_lifetime_ordinal();

  uint64_t buffer_constraints_version_ordinal =
      partial_settings->buffer_constraints_version_ordinal();

  if (buffer_lifetime_ordinal <= protocol_buffer_lifetime_ordinal_[port]) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "buffer_lifetime_ordinal <= "
        "protocol_buffer_lifetime_ordinal_[port] - port: %d",
        port);
    return;
  }
  if (buffer_lifetime_ordinal % 2 == 0) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked(
        "Only odd values for buffer_lifetime_ordinal are permitted - port: %d "
        "value %lu",
        port, buffer_lifetime_ordinal);
    return;
  }
  protocol_buffer_lifetime_ordinal_[port] = buffer_lifetime_ordinal;

  if (buffer_constraints_version_ordinal > sent_buffer_constraints_version_ordinal_[port]) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("Client sent too-new buffer_constraints_version_ordinal - port: %d", port);
    return;
  }

  if (buffer_constraints_version_ordinal <
      last_required_buffer_constraints_version_ordinal_[port]) {
    // ignore - client will probably catch up later
    return;
  }

  // We've peeled off too new and too old above.
  ZX_DEBUG_ASSERT(buffer_constraints_version_ordinal >=
                      last_required_buffer_constraints_version_ordinal_[port] &&
                  buffer_constraints_version_ordinal <=
                      sent_buffer_constraints_version_ordinal_[port]);

  // We've already checked above that the buffer_lifetime_ordinal is in
  // sequence.
  ZX_DEBUG_ASSERT(!port_settings_[port] ||
                  buffer_lifetime_ordinal > buffer_lifetime_ordinal_[port]);

  if (!ValidatePartialBufferSettingsVsConstraintsLocked(port, *partial_settings,
                                                        stream_constraints)) {
    // This assert is safe only because this thread still holds lock_.  This
    // is asserting that ValidateBufferSettingsVsConstraintsLocked() already
    // called FailLocked().
    ZX_DEBUG_ASSERT(IsStoppingLocked());
    return;
  }

  // Little if any reason to do this outside the lock.
  EnsureBuffersNotConfigured(lock, port);

  // This also starts the new buffer_lifetime_ordinal.
  {  // scope port_settings, to enforce not using it after we've moved it out
    std::unique_ptr<PortSettings> port_settings;
    port_settings = std::make_unique<PortSettings>(this, port, std::move(*partial_settings));
    port_settings_[port] = std::move(port_settings);
  }  // ~port_settings, which has been moved out, so we can't use it anyway
  buffer_lifetime_ordinal_[port] = port_settings_[port]->buffer_lifetime_ordinal();

  fidl::ClientEnd<fuchsia_sysmem2::BufferCollectionToken> token = port_settings_[port]->TakeToken();
  // We intentionally don't want to hand the sysmem token directly to the core
  // codec, at least for now (maybe later it'll be necessary).
  ZX_DEBUG_ASSERT(!port_settings_[port]->partial_settings().has_sysmem_token());
  ZX_DEBUG_ASSERT(!port_settings_[port]->partial_settings().has_sysmem2_token());
  fuchsia_sysmem2::BufferCollectionConstraints buffer_collection_constraints =
      [this, port, &lock, &stream_constraints]() {
        // port_settings_[port] can only change on this thread so are safe to
        // read outside the lock.
        ScopedUnlock unlock(lock);
        auto collection_constraints = CoreCodecGetBufferCollectionConstraints2(
            port, stream_constraints, port_settings_[port]->partial_settings());
        return collection_constraints;
      }();
  // The core codec doesn't fill out usage directly.  Instead we fill it out
  // here.
  if (!FixupBufferCollectionConstraintsLocked(port, stream_constraints,
                                              port_settings_[port]->partial_settings(),
                                              &buffer_collection_constraints)) {
    // FixupBufferCollectionConstraints() already called Fail().
    ZX_DEBUG_ASSERT(IsStoppingLocked());
    return;
  }
  // For output, the only reason we re-post here is to share the lock
  // acquisition code with input.
  PostToSharedFidl(
      [this, port, buffer_lifetime_ordinal = buffer_lifetime_ordinal_[port],
       token = std::move(token),
       buffer_collection_constraints = std::move(buffer_collection_constraints)]() mutable {
        std::lock_guard<std::mutex> lock(lock_);
        if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[port]) {
          return;
        }
        if (!sysmem_) {
          return;
        }
        if (IsStoppingLocked()) {
          return;
        }

        auto buffer_collection_request = port_settings_[port]->NewBufferCollectionRequest(
            shared_fidl_dispatcher_,
            [this, port, buffer_lifetime_ordinal](fidl::UnbindInfo unbind_info) {
              std::lock_guard<std::mutex> lock(lock_);
              if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[port]) {
                // It's fine if a BufferCollection fails after we're already
                // done using it.
                return;
              }
              // We're intentionally picky about the BufferCollection failing
              // too soon, as all clean closes should use Close(), which will
              // avoid causing this.  If we find a case where a client
              // legitimately needs to try one way then if that fails try
              // another way, we should see if we can avoid the need to do
              // that by expressing in sysmem constraints, or more likely just
              // accept that such a client will need to start with a new codec
              // instance for the 2nd try.
              UnbindLocked();
            });
        fuchsia_sysmem2::AllocatorBindSharedCollectionRequest bind_request;
        bind_request.token() = std::move(token);
        bind_request.buffer_collection_request() = std::move(buffer_collection_request);
        auto bind_result = sysmem_->BindSharedCollection(std::move(bind_request));
        // one-way message; Allocator disconnection is fatal
        ZX_ASSERT(bind_result.is_ok());

        std::string buffer_name = codec_adapter_->CoreCodecGetName();
        switch (port) {
          case kInputPort:
            buffer_name += "Input";
            break;
          case kOutputPort:
            buffer_name += "Output";
            break;
          default:
            buffer_name += "Unknown";
            break;
        }

        auto& buffer_collection = port_settings_[port]->buffer_collection();

        fuchsia_sysmem2::NodeSetNameRequest set_name_request;
        set_name_request.name() = std::move(buffer_name);
        set_name_request.priority() = 11;
        // We can ignore one-way send failure because the error handler for ZX_CHANNEL_PEER_CLOSED.
        (void)(*buffer_collection)->SetName(std::move(set_name_request));

        fuchsia_sysmem2::NodeSetDebugClientInfoRequest set_client_info_request;
        set_client_info_request.name() = codec_adapter_->CoreCodecGetName();
        set_client_info_request.id() = 0;
        // We can ignore one-way send failure because the error handler for ZX_CHANNEL_PEER_CLOSED.
        (void)(*buffer_collection)->SetDebugClientInfo(std::move(set_client_info_request));

        fuchsia_sysmem2::BufferCollectionSetConstraintsRequest set_constraints_request;
        set_constraints_request.constraints() = std::move(buffer_collection_constraints);
        // We can ignore one-way send failure because the error handler for ZX_CHANNEL_PEER_CLOSED.
        (void)(*buffer_collection)->SetConstraints(std::move(set_constraints_request));

        (*buffer_collection)
            ->WaitForAllBuffersAllocated()
            .Then([this, port, buffer_lifetime_ordinal](
                      fidl::Result<fuchsia_sysmem2::BufferCollection::WaitForAllBuffersAllocated>&
                          result) mutable {
              zx_status_t status;
              fuchsia_sysmem2::BufferCollectionInfo buffer_collection_info;
              if (result.is_error()) {
                if (result.error_value().is_framework_error()) {
                  status = result.error_value().framework_error().status();
                } else {
                  ZX_DEBUG_ASSERT(result.error_value().is_domain_error());
                  status = sysmem::V1CopyFromV2Error(result.error_value().domain_error());
                }
              } else {
                ZX_DEBUG_ASSERT(result.is_ok());
                status = ZX_OK;
                buffer_collection_info = std::move(result->buffer_collection_info().value());
              }
              OnBufferCollectionInfo(port, buffer_lifetime_ordinal, status,
                                     std::move(buffer_collection_info));
            });
      });
}

void CodecImpl::OnBufferCollectionInfo(
    CodecPort port, uint64_t buffer_lifetime_ordinal, zx_status_t status,
    fuchsia_sysmem2::BufferCollectionInfo buffer_collection_info) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());

  if (port == kInputPort) {
    PostSysmemCompletion([this, port, buffer_lifetime_ordinal, status,
                          buffer_collection_info = std::move(buffer_collection_info)]() mutable {
      OnBufferCollectionInfoInternal(port, buffer_lifetime_ordinal, status,
                                     std::move(buffer_collection_info));
    });
  } else {
    ZX_DEBUG_ASSERT(port == kOutputPort);
    OnBufferCollectionInfoInternal(port, buffer_lifetime_ordinal, status,
                                   std::move(buffer_collection_info));
  }
}

void CodecImpl::OnBufferCollectionInfoInternal(
    CodecPort port, uint64_t buffer_lifetime_ordinal, zx_status_t allocate_status,
    fuchsia_sysmem2::BufferCollectionInfo buffer_collection_info) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());

  if (port == kOutputPort) {
    LogEvent(
        media_metrics::
            StreamProcessorEvents2MigratedMetricDimensionEvent_OutputBufferAllocationCompleted);
  } else {
    LogEvent(media_metrics::
                 StreamProcessorEvents2MigratedMetricDimensionEvent_InputBufferAllocationCompleted);
  }

  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    if (IsStoppingLocked()) {
      return;
    }

    // The buffer_lifetime_ordinal_[port] can only change on the current thread.
    if (buffer_lifetime_ordinal != buffer_lifetime_ordinal_[port]) {
      // stale response
      return;
    }
    if (allocate_status != ZX_OK) {
      if (port == kOutputPort) {
        LogEvent(
            media_metrics::
                StreamProcessorEvents2MigratedMetricDimensionEvent_OutputBufferAllocationFailure);
      } else {
        LogEvent(
            media_metrics::
                StreamProcessorEvents2MigratedMetricDimensionEvent_InputBufferAllocationFailure);
      }
      FailLocked(
          "OnBufferCollectionInfoLocked() sees failure - port: %d "
          "allocate_status: %d",
          port, allocate_status);
      return;
    }
  }  // ~lock

  uint32_t buffer_count = static_cast<uint32_t>(buffer_collection_info.buffers()->size());

  // This code trusts sysmem to really be sysmem and to behave correctly, but
  // doesn't hurt to double-check some things in debug build.
  ZX_DEBUG_ASSERT(buffer_count >= 1);
  ZX_DEBUG_ASSERT(buffer_collection_info.buffers()->at(buffer_count - 1).vmo().has_value());
  ZX_DEBUG_ASSERT(buffer_collection_info.buffers()->at(buffer_count - 1).vmo()->is_valid());

  // Let's move the VMO handles out first, so that the BufferCollectionInfo_2 we
  // send to the core codec doesn't have the VMO handles.  We want the core
  // codec to get its VMO handles via the CodecBuffer*(s) we'll provide shortly
  // below.
  std::vector<zx::vmo> vmos;
  vmos.reserve(buffer_count);
  for (uint32_t i = 0; i < buffer_count; ++i) {
    vmos.emplace_back(std::move(buffer_collection_info.buffers()->at(i).vmo().value()));
    buffer_collection_info.buffers()->at(i).vmo().reset();
  }
  ZX_DEBUG_ASSERT(vmos.size() == buffer_count);

  // Now we can tell the core codec about the collection info.  The core codec
  // can clone the FIDL struct if it wants, or can just copy out any info it
  // wants from specific fields.
  CoreCodecSetBufferCollectionInfo(port, buffer_collection_info);

  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);

    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal == buffer_lifetime_ordinal_[port]);

    // The only way port_settings_[port] gets cleared is if
    // buffer_lifetime_ordinal changes.
    ZX_DEBUG_ASSERT(port_settings_[port]);

    // This completes the settings, analogous to having completed
    // SetInputBufferSettings()/SetOutputBufferSettings().
    port_settings_[port]->SetBufferCollectionInfo(std::move(buffer_collection_info));
  }

  if (IsPortSecureRequired(port) && !port_settings_[port]->is_secure()) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    Fail("IsPortSecureRequired(port) && !port_settings_[port]->is_secure() - port: %d", port);
    return;
  }
  if (!IsPortSecurePermitted(port) && port_settings_[port]->is_secure()) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    Fail("!IsPortSecurePermitted(port) && port_settings_[port]->is_secure() - port: %d", port);
    return;
  }

  ZX_DEBUG_ASSERT(!fake_map_range_[port]);
  if (port_settings_[port]->is_secure()) {
    if (IsCoreCodecMappedBufferUseful(port)) {
      zx_status_t status =
          FakeMapRange::Create(port_settings_[port]->vmo_usable_size(), &fake_map_range_[port]);
      if (status != ZX_OK) {
        LogEvent(
            media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_InitializationError);
        Fail("FakeMapRange::Init() failed");
        return;
      }
    }
  }

  // We convert the buffer_collection_info into AddInputBuffer_StreamControl()
  // and AddOutputBufferInternal() calls, almost as if the client were adding
  // the buffers itself (but without the check that the client isn't adding
  // buffers itself while using sysmem).
  for (uint32_t i = 0; i < buffer_count; i++) {
    // While under the lock we'll move out the stuff we need into locals
    uint64_t vmo_usable_start = 0;
    uint64_t vmo_usable_size = 0;
    bool is_secure = false;
    {  // scope lock
      std::lock_guard<std::mutex> lock(lock_);

      ZX_DEBUG_ASSERT(buffer_lifetime_ordinal == buffer_lifetime_ordinal_[port]);
      ZX_DEBUG_ASSERT(port_settings_[port]);

      vmo_usable_start = port_settings_[port]->vmo_usable_start(i);
      vmo_usable_size = port_settings_[port]->vmo_usable_size();
      is_secure = port_settings_[port]->is_secure();
    }  // ~lock

    CodecBuffer::Info buffer_info{.port = port,
                                  .lifetime_ordinal = buffer_lifetime_ordinal,
                                  .index = i,
                                  .is_secure = is_secure};
    CodecVmoRange vmo_range(std::move(vmos[i]), vmo_usable_start, vmo_usable_size);
    if (port == kInputPort) {
      AddInputBuffer_StreamControl(std::move(buffer_info), std::move(vmo_range));
    } else {
      ZX_DEBUG_ASSERT(port == kOutputPort);
      AddOutputBufferInternal(std::move(buffer_info), std::move(vmo_range));
    }
  }

  if (port == kOutputPort) {
    LogEvent(media_metrics::
                 StreamProcessorEvents2MigratedMetricDimensionEvent_OutputBufferAllocationSuccess);
  } else {
    LogEvent(media_metrics::
                 StreamProcessorEvents2MigratedMetricDimensionEvent_InputBufferAllocationSuccess);
  }
}

void CodecImpl::EnsureBuffersNotConfigured(std::unique_lock<std::mutex>& lock, CodecPort port) {
  // This method can be called on input only if there's no current stream.
  //
  // On output, this method can be called if there's no current stream or if
  // we're in the middle of an output config change.
  //
  // On input, this can only be called on stream_control_thread_.
  //
  // On output, this can be called on stream_control_thread_ or fidl_thread().
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && (thrd_current() == stream_control_thread_ ||
                                          thrd_current() == fidl_thread()));

  is_port_buffers_configured_[port] = false;
  if (buffer_lifetime_ordinal_[port] % 2 == 1) {
    buffer_lifetime_ordinal_[port]++;
  }
  if (port_settings_[port]) {
    // This will close the BufferCollection (async as-needed) cleanly, without causing the
    // LogicalBufferCollection to fail.  Mainly we care so we can more easily tell during debugging
    // whether a LogicalBufferCollection was cleanly closed by all participants, vs. potentially
    // getting failed by a participant exiting or non-cleanly closing.  A Sync() by the client is
    // sufficient to ensure this async close is done.
    port_settings_[port] = nullptr;
  }

  // Ensure that buffers aren't with the core codec.
  {  // scope unlock
    ScopedUnlock unlock(lock);
    CoreCodecEnsureBuffersNotConfigured(port);
  }  // ~unlock

  // For mid-stream output config change, the caller is responsible for ensuring
  // that buffers are not with the HW first.
  //
  // TODO(dustingreen): Check anything relevant to buffers not presently being
  // with the HW.
  // ZX_DEBUG_ASSERT(all_packets_[port].empty() ||
  // !all_packets_[port][0]->is_with_hw());

  // This ~FakeMapRange (which calls zx::vmar::destroy()) is among the motivations for calling
  // EnsureBuffersNotConfigured() during the Unbind() sequence / during ~CodecImpl.
  if (fake_map_range_[port]) {
    fake_map_range_[port] = std::nullopt;
  }

  all_packets_[port].clear();
  all_buffers_[port].clear();
  ZX_DEBUG_ASSERT(all_packets_[port].empty());
  ZX_DEBUG_ASSERT(all_buffers_[port].empty());
}

bool CodecImpl::ValidatePartialBufferSettingsVsConstraintsLocked(
    CodecPort port, const fuchsia::media::StreamBufferPartialSettings& partial_settings,
    const fuchsia::media::StreamBufferConstraints& constraints) {
  // Most of the constraints will be handled by telling sysmem about them, not
  // via the client, so there's not a ton to validate here.
  if (partial_settings.has_single_buffer_mode()) {
    if (partial_settings.single_buffer_mode()) {
      LogEvent(media_metrics::
                   StreamProcessorEvents2MigratedMetricDimensionEvent_ClientConstraintsFailure);
      FailLocked("single_buffer_mode (deprecated; obsolete)");
      return false;
    } else {
      LOG(WARNING,
          "has_single_buffer_mode() (set to false) seen - client should stop setting this, even to "
          "false (deprecated; obsolete)");
    }
  }
  ZX_DEBUG_ASSERT(partial_settings.sysmem2_token().is_valid());
  return true;
}

bool CodecImpl::AddBufferCommon(CodecBuffer::Info buffer_info, CodecVmoRange vmo_range) {
  const CodecPort port = buffer_info.port;
  ZX_DEBUG_ASSERT(port == kInputPort && (thrd_current() == stream_control_thread_) ||
                  port == kOutputPort && (thrd_current() == fidl_thread()));
  bool buffers_done_configuring = false;

  std::unique_lock<std::mutex> lock(lock_);

  if (buffer_info.lifetime_ordinal % 2 == 0) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked("Client sent even buffer_lifetime_ordinal, but must be odd - exiting - port: %u\n",
               port);
    return false;
  }

  if (buffer_info.lifetime_ordinal != protocol_buffer_lifetime_ordinal_[port]) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked(
        "Incoherent SetOutputBufferSettings()/SetInputBufferSettings() + "
        "AddOutputBuffer()/AddInputBuffer()s - exiting - port: %d\n",
        port);
    return false;
  }

  // If the server is not interested in the client's buffer_lifetime_ordinal,
  // the client's buffer_lifetime_ordinal won't match the server's
  // buffer_lifetime_ordinal_.  The client will probably later catch up.
  if (buffer_info.lifetime_ordinal != buffer_lifetime_ordinal_[port]) {
    // The case that ends up here is when a client's output configuration
    // (whole or last part) is being ignored because it's not yet caught up
    // with last_required_buffer_constraints_version_ordinal_.

    // This case won't happen for input, at least for now.  This is an assert
    // rather than a client behavior check, because previous client protocol
    // checks have already peeled off any invalid client behavior that might
    // otherwise cause this assert to trigger.
    ZX_DEBUG_ASSERT(port == kOutputPort);

    // Ignore the client's message.  The client will probably catch up later.
    return false;
  }

  ZX_DEBUG_ASSERT(buffer_info.index == all_buffers_[port].size());
  uint32_t required_buffer_count = port_settings_[port]->buffer_count();
  ZX_DEBUG_ASSERT(buffer_info.index < required_buffer_count);

  std::unique_ptr<CodecBuffer> local_buffer = std::unique_ptr<CodecBuffer>(
      new CodecBuffer(this, std::move(buffer_info), std::move(vmo_range)));

  if (IsCoreCodecMappedBufferUseful(port)) {
    if (fake_map_range_[port]) {
      // The fake_map_range_[port]->base() is % PAGE_SIZE == 0, which is the same as a mapping
      // would be.  There are sufficient virtual pages starting at FakeMapRange::base() to permit
      // CodecBuffer to include the low-order vmo_usable_start % PAGE_SIZE bits in
      // CodecBuffer::base(), for any vmo_usable_start() value (even the worst case of
      // PAGE_SIZE - 1, and buffer size % PAGE_SIZE == 2).  By including those low-order
      // intra-page-offset bits, we can treat non-secure and secure cases similarly.
      local_buffer->FakeMap(fake_map_range_[port]->base());
    } else {
      // So far, there's little reason to avoid doing the Map() part under the
      // lock, even if it can be a bit more time consuming, since there's no data
      // processing happening at this point anyway, and there wouldn't be any
      // happening in any other code location where we could potentially move the
      // Map() either.
      if (!local_buffer->Map()) {
        LogEvent(
            media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_InitializationError);
        FailLocked("AddOutputBuffer()/AddInputBuffer() couldn't Map() new buffer - port: %d", port);
        return false;
      }
    }
  }

  // We keep the buffers pinned for DMA continuously, since there's not much benefit to un-pinning
  // and re-pinning them (so far).  By pinning, we prevent sysmem from recycling the
  // BufferCollection VMOs until the driver has re-started and un-quarantined pinned pages (via
  // its BTI), after ensuring the HW is no longer doing DMA from/to the pages.
  //
  // TODO(https://fxbug.dev/42114424): All CodecAdapter(s) that start memory access that can
  // continue beyond VMO handle closure during process death/termination should have a BTI.
  // Resolving this
  // TODO will require updating at least the amlogic-video VP9 decoder to provide a BTI.
  //
  // TODO(https://fxbug.dev/42114425): Currently OEMCrypto's indirect (via FIDL) SMC calls that take
  // physical addresses are not guaranteed to be fully over/done before VMO handles are auto-closed
  // by OEMCrypto assuming OEMCryto's process dies/terminates.
  if (IsCoreCodecHwBased(port) && *core_codec_bti_) {
    zx_status_t status = local_buffer->Pin();
    if (status != ZX_OK) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_InitializationError);
      FailLocked("buffer->Pin() failed - status: %d port: %d", status, port);
      return false;
    }
  }

  {
    ScopedUnlock unlock(lock);
    // Inform the core codec up-front about each buffer.
    CoreCodecAddBuffer(port, local_buffer.get());
  }
  all_buffers_[port].push_back(std::move(local_buffer));
  if (all_buffers_[port].size() == required_buffer_count) {
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[port] ==
                    port_settings_[port]->buffer_lifetime_ordinal());
    // Stash this while we can, before the client de-configures.
    last_provided_buffer_constraints_version_ordinal_[port] =
        port_settings_[port]->buffer_constraints_version_ordinal();
    // Now we allocate all_packets_[port].
    ZX_DEBUG_ASSERT(all_packets_[port].empty());
    uint32_t packet_count = port_settings_[port]->packet_count();
    for (uint32_t i = 0; i < packet_count; i++) {
      // Private constructor to prevent core codec maybe creating its own
      // Packet instances (which isn't the intent) seems worth the hassle of
      // not using make_unique<>() here.
      all_packets_[port].push_back(std::unique_ptr<CodecPacket>(
          new CodecPacket(port_settings_[port]->buffer_lifetime_ordinal(), i)));
    }

    {  // scope unlock
      ScopedUnlock unlock(lock);

      // A core codec can take action here to finish configuring buffers if
      // it's able, or can delay configuring buffers until
      // CoreCodecStartStream() or
      // CoreCodecMidStreamOutputBufferReConfigFinish() if that works better
      // for the core codec.
      //
      // In any case, during a mid-stream output constraints change, the core
      // codec must not call any onCoreCodecOutput* methods until the core
      // codec sees CoreCodecStopStream() (after stopping the stream, in
      // preparation for the next stream), or
      // CoreCodecMidStreamOutputBufferReConfigFinish().
      //
      // In other words, this call does /not/ imply un-pausing output.
      CoreCodecConfigureBuffers(port, all_packets_[port]);

      // All output packets need to start with the core codec.  This is
      // implicit for the Codec interface (implied by adding the last output
      // buffer) but explicit in the CodecAdapter interface.
      if (port == kOutputPort) {
        for (uint32_t i = 0; i < packet_count; i++) {
          CoreCodecRecycleOutputPacket(all_packets_[kOutputPort][i].get());
        }
      }
    }  // ~unlock

    is_port_buffers_configured_[port] = true;
    buffers_done_configuring = true;

    // For client-called AddOutputBuffer(), the last buffer being added is
    // analogous to CompleteOutputBufferPartialSettings(); we handle that
    // analogous-ness in IsOutputConfiguredLocked() (not by pretending we got
    // a CompleteOutputBufferPartialSettings() here), so
    // is_port_buffers_configured_[port] = true above is enough to make
    // IsOutputConfiguredLocked() return true if this is a client-driven
    // AddOutputBuffer().
  }
  return buffers_done_configuring;
}

bool CodecImpl::CheckOldBufferLifetimeOrdinalLocked(CodecPort port,
                                                    uint64_t buffer_lifetime_ordinal) {
  // The client must only send odd values.  0 is even so we don't need a
  // separate check for that.
  if (buffer_lifetime_ordinal % 2 == 0) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked(
        "CheckOldBufferLifetimeOrdinalLocked() - buffer_lifetime_ordinal must "
        "be odd");
    return false;
  }
  if (buffer_lifetime_ordinal > protocol_buffer_lifetime_ordinal_[port]) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked(
        "client sent new buffer_lifetime_ordinal in message type that doesn't "
        "allow new buffer_lifetime_ordinals");
    return false;
  }
  return true;
}

bool CodecImpl::CheckStreamLifetimeOrdinalLocked(uint64_t stream_lifetime_ordinal) {
  if (stream_lifetime_ordinal % 2 != 1) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked("stream_lifetime_ordinal must be odd.\n");
    return false;
  }
  if (stream_lifetime_ordinal < stream_lifetime_ordinal_) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked("client sent stream_lifetime_ordinal that went backwards");
    return false;
  }
  return true;
}

bool CodecImpl::StartNewStream(std::unique_lock<std::mutex>& lock,
                               uint64_t stream_lifetime_ordinal) {
  VLOGF("StartNewStream()");
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  ZX_DEBUG_ASSERT((stream_lifetime_ordinal % 2 == 1) && "new stream_lifetime_ordinal must be odd");

  if (IsStoppingLocked()) {
    // Don't start a new stream if the whole CodecImpl is already stopping.
    //
    // A completely different path will take care of calling
    // EnsureStreamClosed() during CodecImpl stop.
    //
    // TODO(dustingreen): If all callers are already checking this at the top
    // of each relevant .*_StreamControl method, then we don't necessarily need
    // this check, but consider any intervals where lock_ isn't held also - we
    // don't want the wait for stream_control_thread_ to exit to ever be long
    // when stopping this CodecImpl.
    return false;
  }

  EnsureStreamClosed(lock);
  ZX_DEBUG_ASSERT(!IsStreamActiveLocked());

  // Now it's time to start the new stream.  We start the new stream at
  // Codec layer first then core codec layer.

  if (!IsInputConfiguredLocked()) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked("input not configured before start of stream (QueueInputPacket())");
    return false;
  }

  // The lock_ is held.  We don't need lock_ to keep stream_ alive.  That's not necessary because
  // only StreamControl domain will remove items from stream_queue_.  This hold interval is to
  // protect stream_queue_ against concurrent modification by output domain (FIDL thread) only.
  ZX_DEBUG_ASSERT(stream_queue_.size() >= 1);
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal == stream_queue_.front()->stream_lifetime_ordinal());
  stream_ = stream_queue_.front().get();

  // Update the stream_lifetime_ordinal_ to the new stream.  We need to do
  // this before we send new output config, since the output config will be
  // generated using the current stream ordinal.
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal > stream_lifetime_ordinal_);
  stream_lifetime_ordinal_ = stream_lifetime_ordinal;
  ZX_DEBUG_ASSERT(stream_->stream_lifetime_ordinal() == stream_lifetime_ordinal_);

  // The client is not permitted to unilaterally re-configure output while a
  // stream is active, but the client may still be responding to a previous
  // server-initiated mid-stream format change.
  //
  // ###########################################################################
  // We don't attempt to optimize every case as much as might be possible here.
  // The main overall optimization is that it's possible to switch streams
  // without reallocating buffers.  We also need to make sure it's possible to
  // detect output format at the start of a stream regardless of what happened
  // before, and possible to perform a mid-stream format change.
  // ###########################################################################
  //
  // Given the above, our *main concern* here is that we get to a state where we
  // *know* the client isn't trying to re-configure output during format
  // detection, which at best would be confusing to allow, so we avoid that
  // possibility here by forcing a client to catch up with the server, if
  // there's *any possibility* that the client might still be working on
  // catching up with the server.
  //
  // If the client's most recently fully-completed output config is less than
  // the most recently sent output constraints with action_required true, then
  // we force an even fresher output constraints here tagged as being relevant
  // to the current stream, and wait for the client to catch up to that before
  // continuing.  By marking as being for this stream, we ensure that the client
  // will bother to finish configuring output, which gets us to a state where we
  // know it's safe to do another mid-stream format change as needed (vs. the
  // client maybe finishing the old config or maybe not).
  //
  // We also force the client to catch up if the core codec previously indicated
  // that the current config is "meh".  This may not be strictly necessary since
  // the "meh" was with respect to the old stream, but just in case a core codec
  // cares, we move on from the old config before delivering new stream data.
  //
  // Some core codecs may require the output to be configured to _something_ as
  // they don't support giving us the real output config unless the output is
  // configured to at least something at first.
  //
  // Other core codecs (such as some HW-based codecs) can deal with no output
  // configured while detecting the output format, but even for those codecs, we
  // only do this if the above cases don't apply.  These codecs have to deal
  // with an output config that's already set across a stream switch anyway, to
  // permit buffers to stay configured across a stream switch when possible, so
  // the cases above potentially setting an output config that's not super
  // relevant to the new stream doesn't really complicate the core codec since
  // an old stream's config might not be super relevant to a new stream either.
  //
  // Format detection is separate and handled like a mid-stream format change.
  // This stuff here is just getting output config into a non-changing state
  // before we start format detection.
  bool is_new_config_needed;
  // The statement below could obviously be re-written as a giant boolean
  // expression, but this way seems easier to comment.
  if (last_provided_buffer_constraints_version_ordinal_[kOutputPort] <
      last_required_buffer_constraints_version_ordinal_[kOutputPort]) {
    // The client _might_ still be trying to catch up, so to disambiguate,
    // require an even fresher config with respect to this new stream to
    // unambiguously force the client to catch up to the even newer config.
    is_new_config_needed = true;
  } else if (IsCoreCodecRequiringOutputConfigForFormatDetection() && !IsOutputConfiguredLocked()) {
    // The core codec requires output to be configured before format detection,
    // so we force the client to provide an output config before format
    // detection.
    is_new_config_needed = true;
  } else if (IsOutputConfiguredLocked() &&
             port_settings_[kOutputPort]->buffer_constraints_version_ordinal() <=
                 core_codec_meh_output_buffer_constraints_version_ordinal_) {
    // The core codec previously expressed "meh" regarding the current config's
    // buffer_constraints_version_ordinal, so to avoid mixing that with core
    // codec stream switch, force the client to configure output buffers before
    // format detection for the new stream.
    is_new_config_needed = true;
  } else {
    // The core codec is ok to perform format detection in the current state,
    // and we know that a well-behaved client is not currently trying to change
    // the output config.
    is_new_config_needed = false;
  }

  if (is_new_config_needed) {
    StartIgnoringClientOldOutputConfig(lock);
    EnsureBuffersNotConfigured(lock, kOutputPort);
    // This does count as a mid-stream output config change, even when this is
    // at the start of a stream - it's still while a stream is active, and still
    // prevents this stream from outputting any data to the Codec client until
    // the Codec client re-configures output while this stream is active.
    GenerateAndSendNewOutputConstraints(lock);

    // Now we can wait for the client to catch up to the current output config
    // or for the client to tell the server to discard the current stream.
    stream_->AssertHeld(this);
    while (!IsStoppingLocked() && !stream_->future_discarded() && !IsOutputConfiguredLocked()) {
      RunAnySysmemCompletionsOrWait(lock);
    }

    if (IsStoppingLocked()) {
      return false;
    }

    if (stream_->future_discarded()) {
      // A discarded stream isn't an error for the CodecImpl instance.
      return true;
    }
  }

  // Now we have input configured, and output configured if needed by the core
  // codec, so we can move the core codec to running state.
  {  // scope unlock
    ScopedUnlock unlock(lock);
    CoreCodecStartStream();
  }  // ~unlock

  // Track this so the core codec doesn't have to bother with "ensure"
  // semantics, just start/stop, where stop isn't called unless the core codec
  // has a started stream.
  is_core_codec_stream_started_ = true;

  return true;
}

void CodecImpl::EnsureStreamClosed(std::unique_lock<std::mutex>& lock) {
  VLOGF("EnsureStreamClosed()");
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);

  // Ensure the old stream is closed at CodecAdapter layer.  The stream may already be closed at
  // CodecAdapter layer, such as if the CodecAdapter previously used onCoreCodecFailStream() and the
  // async work posted by that call has already completed.
  EnsureCoreCodecStreamStopped(lock);

  // Now close the old stream at the Codec layer.
  EnsureCodecStreamClosedLockedInternal();

  ZX_DEBUG_ASSERT(!IsStreamActiveLocked());
}

void CodecImpl::EnsureCoreCodecStreamStopped(std::unique_lock<std::mutex>& lock) {
  // Stop the core codec, by using this thread to directly drive the core codec
  // from running to stopped (if not already stopped).  We do this first so the
  // core codec won't try to send us output while we have no stream at the Codec
  // layer.
  if (is_core_codec_stream_started_) {
    {  // scope unlock
      ScopedUnlock unlock(lock);
      VLOGF("CoreCodecStopStream()...");
      CoreCodecStopStream();
      VLOGF("CoreCodecStopStream() done.");
    }
    is_core_codec_stream_started_ = false;
  }
}

// The only valid caller of this is EnsureStreamClosed().  We have this in a
// separate method only to make it easier to assert a couple things in the
// caller.
void CodecImpl::EnsureCodecStreamClosedLockedInternal() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  if (stream_lifetime_ordinal_ % 2 == 0) {
    // Already closed.
    return;
  }

  // The lock_ is held which protects against concurrent modification by output domain (FIDL
  // thread), and is also held while deleting first item in stream_queue_ so that output domain can
  // hold lock_ to ensure a stream obtained from stream_queue_ stays alive until output domain is
  // done marking the stream as future_flushed() or similar.
  ZX_DEBUG_ASSERT(stream_queue_.front()->stream_lifetime_ordinal() == stream_lifetime_ordinal_);
  stream_ = nullptr;
  stream_queue_.pop_front();

  stream_lifetime_ordinal_++;
  // Even values mean no current stream.
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal_ % 2 == 0);
}

bool CodecImpl::RunAnySysmemCompletions(std::unique_lock<std::mutex>& lock) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  // Typically this loop will run once, but on return we want the queue to be
  // empty even if more showed up while in this method, for condition_variable
  // signalling reasons.
  bool any_ran = false;
  while (!sysmem_completion_queue_.empty()) {
    // We'll run them all, so extract all the items and run them all.
    std::queue<fit::closure> local_batch_to_run;
    local_batch_to_run.swap(sysmem_completion_queue_);
    // The unlock doesn't cause queue re-ordering, though so far none of these
    // items care anyway.
    ScopedUnlock unlock(lock);
    while (!local_batch_to_run.empty()) {
      any_ran = true;
      fit::closure to_run = std::move(local_batch_to_run.front());
      local_batch_to_run.pop();
      to_run();
    }
  }
  return any_ran;
}

void CodecImpl::PostSysmemCompletion(fit::closure to_run) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());

  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    sysmem_completion_queue_.emplace(std::move(to_run));
    // In case there is no WaitEnsureSysmemReadyOnInput(), we post to
    // StreamControl to ensure that RunAnySysmemCompletions() runs soon.
    // Don't let them accumulate though.
    if (!is_sysmem_runner_pending_) {
      is_sysmem_runner_pending_ = true;
      PostToStreamControl([this] {
        std::unique_lock<std::mutex> lock(lock_);
        RunAnySysmemCompletions(lock);
        ZX_DEBUG_ASSERT(sysmem_completion_queue_.empty());
        is_sysmem_runner_pending_ = false;
      });
    }
  }  // ~lock

  // In case to_run needs to get run by a QueueInput...StreamControl() method
  // via WaitEnsureSysmemReadyOnInput(), we wake the StreamControl thread.  We
  // must do this even if is_sysmem_runner_pending_, in case that runner won't
  // run for a while due to WaitEnsureSysmemReadyOnInput() blocking
  // StreamControl.
  wake_stream_control_condition_.notify_all();
}

bool CodecImpl::WaitEnsureSysmemReadyOnInput(std::unique_lock<std::mutex>& lock) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  // Input buffer re-config is not permitted unless there's no current stream.
  ZX_DEBUG_ASSERT(!IsStreamActiveLocked());
  while (!IsInputConfiguredLocked()) {
    RunAnySysmemCompletionsOrWait(lock);
    // No need to check for stream switch since it's not permitted for a client
    // to be sending any message that can cause a new stream until after the
    // client is done configuring input buffers (enforced elsewhere).
    if (IsStoppingLocked()) {
      return false;
    }
  }
  return true;
}

void CodecImpl::RunAnySysmemCompletionsOrWait(std::unique_lock<std::mutex>& lock) {
  // If any sysmem completions ran, we immediately return, so that conditions
  // can be checked again in the caller immediately.
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  bool any_completions_ran = RunAnySysmemCompletions(lock);
  ZX_DEBUG_ASSERT(sysmem_completion_queue_.empty());
  if (!any_completions_ran) {
    // We know sysmem_completion_queue_.empty() and the lock is held just before
    // this wait().
    wake_stream_control_condition_.wait(lock);
  }
}

// This is called on Output ordering domain (FIDL thread) any time a message is
// received which would be able to start a new stream.
//
// More complete protocol validation happens on StreamControl ordering domain.
// The validation here is just to validate to degree needed to not break our
// stream_queue_ and future_stream_lifetime_ordinal_.
bool CodecImpl::EnsureFutureStreamSeenLocked(uint64_t stream_lifetime_ordinal) {
  if (future_stream_lifetime_ordinal_ == stream_lifetime_ordinal) {
    return true;
  }
  if (stream_lifetime_ordinal < future_stream_lifetime_ordinal_) {
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolFailure);
    FailLocked("stream_lifetime_ordinal went backward - exiting\n");
    return false;
  }
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal > future_stream_lifetime_ordinal_);
  if (future_stream_lifetime_ordinal_ % 2 == 1) {
    if (!EnsureFutureStreamCloseSeenLocked(future_stream_lifetime_ordinal_)) {
      return false;
    }
  }
  future_stream_lifetime_ordinal_ = stream_lifetime_ordinal;

  // The lock_ is held, protecting stream_queue_ against concurrent modification by StreamControl
  // domain.
  stream_queue_.push_back(std::make_unique<Stream>(this, stream_lifetime_ordinal));
  if (stream_queue_.size() > kMaxInFlightStreams) {
    LogEvent(
        media_metrics::
            StreamProcessorEvents2MigratedMetricDimensionEvent_MaxInFlightStreamsExceededError);
    FailLocked(
        "kMaxInFlightStreams reached - clients capable of causing this are "
        "instead supposed to wait/postpone to prevent this from occurring - "
        "exiting\n");
    return false;
  }
  return true;
}

// This is called on Output ordering domain (FIDL thread) any time a message is
// received which would close a stream.
//
// More complete protocol validation happens on StreamControl ordering domain.
// The validation here is just to validate to degree needed to not break our
// stream_queue_ and future_stream_lifetime_ordinal_.
bool CodecImpl::EnsureFutureStreamCloseSeenLocked(uint64_t stream_lifetime_ordinal) {
  if (future_stream_lifetime_ordinal_ % 2 == 0) {
    // Already closed.
    if (stream_lifetime_ordinal != future_stream_lifetime_ordinal_ - 1) {
      LogEvent(
          media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
      FailLocked(
          "CloseCurrentStream() seen with stream_lifetime_ordinal != "
          "most-recent seen stream");
      return false;
    }
    return true;
  }
  if (stream_lifetime_ordinal != future_stream_lifetime_ordinal_) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("attempt to close a stream other than the latest seen stream");
    return false;
  }
  ZX_DEBUG_ASSERT(stream_lifetime_ordinal == future_stream_lifetime_ordinal_);

  // The lock_ is held, ensuring closing_stream stays alive until after call to
  // SetFutureDiscarded().
  if (stream_queue_.empty()) {
    // Latest stream already failed and removed by StreamControl domain.
    return true;
  }
  ZX_DEBUG_ASSERT(!stream_queue_.empty());
  Stream* closing_stream = stream_queue_.back().get();
  ZX_DEBUG_ASSERT(closing_stream->stream_lifetime_ordinal() == stream_lifetime_ordinal);
  // It is permitted to see a FlushCurrentStream() before a CloseCurrentStream()
  // and this can make sense if a client just wants to inform the server of all
  // stream closes, or if the client wants to release_input_buffers or
  // release_output_buffers after the flush is done.
  //
  // If we didn't previously flush, then this close is discarding.
  closing_stream->AssertHeld(this);
  if (!closing_stream->future_flush_end_of_stream()) {
    closing_stream->SetFutureDiscarded();
  }

  future_stream_lifetime_ordinal_++;
  ZX_DEBUG_ASSERT(future_stream_lifetime_ordinal_ % 2 == 0);
  return true;
}

// This is called on Output ordering domain (FIDL thread) any time a flush is
// seen.
//
// More complete protocol validation happens on StreamControl ordering domain.
// The validation here is just to validate to degree needed to not break our
// stream_queue_ and future_stream_lifetime_ordinal_.
bool CodecImpl::EnsureFutureStreamFlushSeenLocked(uint64_t stream_lifetime_ordinal) {
  if (stream_lifetime_ordinal != future_stream_lifetime_ordinal_) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("FlushCurrentStream() stream_lifetime_ordinal inconsistent");
    return false;
  }

  // The lock_ is held, ensuring flushing_stream stays alive until after call to
  // SetFutureFlushEndOfStream().
  if (stream_queue_.empty()) {
    // Latest stream already failed and removed by StreamControl domain.
    return false;
  }
  ZX_DEBUG_ASSERT(!stream_queue_.empty());
  Stream* flushing_stream = stream_queue_.back().get();
  // Thanks to the above future_stream_lifetime_ordinal_ check, we know the
  // future stream is not discarded yet.
  flushing_stream->AssertHeld(this);
  ZX_DEBUG_ASSERT(!flushing_stream->future_discarded());
  flushing_stream->AssertHeld(this);
  if (flushing_stream->future_flush_end_of_stream()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_ClientProtocolError);
    FailLocked("FlushCurrentStream() used twice on same stream");
    return false;
  }

  // We don't future-verify that we have a QueueInputEndOfStream(). We'll verify
  // that later when StreamControl catches up to this stream.

  // Remember the flush so we later know that a close doesn't imply discard.
  flushing_stream->SetFutureFlushEndOfStream();

  // A FlushEndOfStreamAndCloseStream() is also a close, after the flush.  This
  // keeps future_stream_lifetime_ordinal_ consistent.
  if (!EnsureFutureStreamCloseSeenLocked(stream_lifetime_ordinal)) {
    return false;
  }
  return true;
}

// This method is only called when buffer_constraints_action_required will be
// true in an OnOutputConstraints() message sent shortly after this method call.
//
// Even if the client is switching streams rapidly without configuring output,
// this method and GenerateAndSendNewOutputConstraints() with
// buffer_constraints_action_required true always run in pairs.
//
// If the client is in the middle of configuring output, we'll start ignoring
// the client's messages re. the old buffer_lifetime_ordinal and old
// buffer_constraints_version_ordinal until the client catches up to the new
// last_required_buffer_constraints_version_ordinal_[kOutputPort].
void CodecImpl::StartIgnoringClientOldOutputConfig(std::unique_lock<std::mutex>& lock) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);

  // The buffer_lifetime_ordinal_[kOutputPort] can be even on entry due to at
  // least two cases: 0, and when the client is switching streams repeatedly
  // without setting a new buffer_lifetime_ordinal_[kOutputPort].
  if (buffer_lifetime_ordinal_[kOutputPort] % 2 == 1) {
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kOutputPort] % 2 == 1);
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kOutputPort] ==
                    port_settings_[kOutputPort]->buffer_lifetime_ordinal());
    buffer_lifetime_ordinal_[kOutputPort]++;
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kOutputPort] % 2 == 0);
    ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[kOutputPort] ==
                    port_settings_[kOutputPort]->buffer_lifetime_ordinal() + 1);
  }

  // When buffer_constraints_action_required true, we can assert in
  // GenerateAndSendNewOutputConstraints() that this value is still the
  // next_output_buffer_constraints_version_ordinal_ in that method.
  last_required_buffer_constraints_version_ordinal_[kOutputPort] =
      next_output_buffer_constraints_version_ordinal_;

  // Now that we've stopped any new calls to CoreCodecRecycleOutputPacket(),
  // fence through any previously-started call to CoreCodecRecycleOutputPacket()
  // that maybe have been started previously, before returning from this method.
  //
  // We can't just be holding lock_ during the call to
  // CoreCodecRecycleOutputPacket() because it acquires the video_decoder_lock_
  // and in other paths the video_decoder_lock_ is held while acquiring lock_.
  //
  // It's ok for the StreamControl domain to wait on the Output domain (but not
  // the other way around).
  bool is_output_ordering_domain_done_with_recycle_output_packet = false;
  std::condition_variable condition_changed;
  PostToSharedFidl(
      [this, &is_output_ordering_domain_done_with_recycle_output_packet, &condition_changed] {
        {  // scope lock
          std::lock_guard<std::mutex> lock(lock_);
          is_output_ordering_domain_done_with_recycle_output_packet = true;
        }
        condition_changed.notify_all();
      });
  while (!is_output_ordering_domain_done_with_recycle_output_packet) {
    condition_changed.wait(lock);
  }
  ZX_DEBUG_ASSERT(is_output_ordering_domain_done_with_recycle_output_packet);
}

void CodecImpl::GenerateAndSendNewOutputConstraints(std::unique_lock<std::mutex>& lock) {
  // When client action is required, this can only happen on the StreamControl
  // ordering domain.  When client action is not required, it can happen from
  // the InputData ordering domain.
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);

  uint64_t current_stream_lifetime_ordinal = stream_lifetime_ordinal_;
  uint64_t new_output_buffer_constraints_version_ordinal =
      next_output_buffer_constraints_version_ordinal_++;

  // If buffer_constraints_action_required true, the caller bumped the
  // last_required_buffer_constraints_version_ordinal_[kOutputPort] before
  // calling this method (using StartIgnoringClientOldOutputConfig()), to
  // ensure any output config messages from the client are ignored until the
  // client catches up to at least
  // last_required_buffer_constraints_version_ordinal_.
  ZX_DEBUG_ASSERT(last_required_buffer_constraints_version_ordinal_[kOutputPort] ==
                  new_output_buffer_constraints_version_ordinal);

  std::unique_ptr<const fuchsia::media::StreamOutputConstraints> output_constraints;
  {  // scope unlock
    ScopedUnlock unlock(lock);

    // Don't call the core codec under the lock_, because we can avoid doing so,
    // and to allow the core codec to use this thread to call back into
    // CodecImpl using this stack if needed.  So far we don't have any actual
    // known examples of a core codec using this thread to call back into
    // CodecImpl using this stack.
    output_constraints = CoreCodecBuildNewOutputConstraints(
        current_stream_lifetime_ordinal, new_output_buffer_constraints_version_ordinal,
        /*buffer_constraints_action_required=*/true);
  }  // ~unlock

  // We only call GenerateAndSendNewOutputConstraints() from contexts that won't
  // be changing the stream_lifetime_ordinal_, so the fact that we released the
  // lock above doesn't mean the stream_lifetime_ordinal_ could have changed, so
  // we can assert here that it's still the same as above.
  ZX_DEBUG_ASSERT(current_stream_lifetime_ordinal == stream_lifetime_ordinal_);

  output_constraints_ = std::move(output_constraints);

  // Stay under lock after setting output_constraints_, to get proper ordering
  // of sent messages even if a hostile client deduces the content of this
  // message before we've sent it and manages to get the server to send another
  // subsequent OnOutputConstraints().

  ZX_DEBUG_ASSERT(sent_buffer_constraints_version_ordinal_[kOutputPort] + 1 ==
                  new_output_buffer_constraints_version_ordinal);

  // Setting this within same lock hold interval as we queue the message to be
  // sent in order vs. other OnOutputConstraints() messages.  This way we can
  // verify that the client's incoming messages are not trying to configure with
  // respect to a buffer_constraints_version_ordinal that is newer than we've
  // actually sent the client.
  sent_buffer_constraints_version_ordinal_[kOutputPort] =
      new_output_buffer_constraints_version_ordinal;

  // Intentional copy of fuchsia::media::OutputConfig output_constraints_ here,
  // as we want output_constraints_ to remain valid (at least for debugging
  // reasons for now).
  PostToSharedFidl([this, output_constraints = fidl::Clone(*output_constraints_)]() mutable {
    // See "is_bound_checks" comment up top.
    if (binding_.is_bound()) {
      binding_.events().OnOutputConstraints(std::move(output_constraints));
    }
  });
}

void CodecImpl::MidStreamOutputConstraintsChange(uint64_t stream_lifetime_ordinal) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  VLOGF("CodecImpl::MidStreamOutputConstraintsChange - stream: %lu", stream_lifetime_ordinal);
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    AssertHeld(lock, lock_);
    VLOGF("lock aquired 1");
    if (stream_lifetime_ordinal < stream_lifetime_ordinal_) {
      // ignore; The omx_meh_output_buffer_constraints_version_ordinal_ took
      // care of it.
      VLOGF("CodecImpl::MidStreamOutputConstraintsChange - stale stream");
      return;
    }
    ZX_DEBUG_ASSERT(stream_lifetime_ordinal == stream_lifetime_ordinal_);

    // We can work through the mid-stream output constraints change step by step
    // using this thread.

    // This is what starts the interval during which we'll ignore any
    // in-progress client output config until the client catches up.
    VLOGF("StartIngoringClientOldOutputConfig()...");
    StartIgnoringClientOldOutputConfig(lock);

    {  // scope unlock
      ScopedUnlock unlock(lock);
      VLOGF("CoreCodecMidStreamOutputBufferReConfigPrepare()...");
      CoreCodecMidStreamOutputBufferReConfigPrepare();
    }  // ~unlock

    VLOGF("EnsureBuffersNotConfigured()...");
    EnsureBuffersNotConfigured(lock, kOutputPort);

    VLOGF("GenerateAndSendNewOutputConstraints()...");
    GenerateAndSendNewOutputConstraints(lock);

    // Now we can wait for the client to catch up to the current output config
    // or for the client to tell the server to discard the current stream.
    VLOGF("RunAnySysmemCompletionsOrWait()...");
    stream_->AssertHeld(this);
    while (!IsStoppingLocked() && !stream_->future_discarded() && !IsOutputConfiguredLocked()) {
      RunAnySysmemCompletionsOrWait(lock);
    }

    if (IsStoppingLocked()) {
      VLOGF("CodecImpl::MidStreamOutputConstraintsChange IsStoppingLocked()");
      return;
    }

    if (stream_->future_discarded()) {
      // We already know how to handle this case, and
      // core_codec_meh_output_buffer_constraints_version_ordinal_ is still set
      // such that the client will be forced to re-configure output buffers at
      // the start of the new stream.
      VLOGF("CodecImpl::MidStreamOutputConstraintsChange future_discarded()");
      return;
    }

    // For asserts.
    VLOGF("ClearMidStreamOutputConstraintsChangeActive()...");
    stream_->ClearMidStreamOutputConstraintsChangeActive();
  }  // ~lock

  VLOGF("CoreCodecMidStreamOutputBufferReConfigFinish()...");
  CoreCodecMidStreamOutputBufferReConfigFinish();

  VLOGF("Done with mid-stream format change.");
}

// TODO(dustingreen): Consider whether we ever intend to plumb anything coming from the core codec
// from a different proc.  If not (probably this is the case), we can change several of the checks
// in here to ZX_DEBUG_ASSERT() instead.
bool CodecImpl::FixupBufferCollectionConstraintsLocked(
    CodecPort port, const fuchsia::media::StreamBufferConstraints& stream_buffer_constraints,
    const fuchsia::media::StreamBufferPartialSettings& partial_settings,
    fuchsia_sysmem2::BufferCollectionConstraints* buffer_collection_constraints) {
  if (!buffer_collection_constraints->usage().has_value()) {
    buffer_collection_constraints->usage().emplace();
  }
  fuchsia_sysmem2::BufferUsage& usage = *buffer_collection_constraints->usage();

  if (IsCoreCodecMappedBufferUseful(port)) {
    // Not surprisingly, both decoders and encoders read from input and write to
    // output.
    if (port == kInputPort) {
      uint32_t cpu_usage = 0;
      if (usage.cpu().has_value()) {
        cpu_usage = *usage.cpu();
      }
      if (cpu_usage & ~(fuchsia_sysmem2::kCpuUsageRead | fuchsia_sysmem2::kCpuUsageReadOften)) {
        LogEvent(
            media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
        FailLocked("Core codec set disallowed CPU usage bits (input port).");
        return false;
      }
      if (!IsPortSecureRequired(kInputPort)) {
        if (!usage.cpu().has_value()) {
          usage.cpu().emplace(0);
        }
        *usage.cpu() |= fuchsia_sysmem2::kCpuUsageRead | fuchsia_sysmem2::kCpuUsageReadOften;
      } else {
        usage.cpu().reset();
      }
    } else {
      uint32_t cpu_usage = 0;
      if (usage.cpu().has_value()) {
        cpu_usage = *usage.cpu();
      }
      if (cpu_usage & ~(fuchsia_sysmem2::kCpuUsageWrite | fuchsia_sysmem2::kCpuUsageWriteOften)) {
        LogEvent(
            media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
        FailLocked("Core codec set disallowed CPU usage bit(s) (output port).");
        return false;
      }
      if (!IsPortSecureRequired(kOutputPort)) {
        if (!usage.cpu().has_value()) {
          usage.cpu().emplace(0);
        }
        *usage.cpu() |= fuchsia_sysmem2::kCpuUsageWrite | fuchsia_sysmem2::kCpuUsageWriteOften;
      } else {
        usage.cpu().reset();
      }
    }
  } else {
    if (usage.cpu().has_value()) {
      LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
      FailLocked("Core codec set usage.cpu despite !IsCoreCodecMappedBufferUseful()");
      return false;
    }
    // The CPU won't touch the buffers at all.
    usage.cpu().reset();
  }
  if (usage.vulkan().has_value()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec set usage.vulkan bits");
    return false;
  }
  ZX_DEBUG_ASSERT(!usage.vulkan().has_value());
  if (usage.display().has_value()) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec set usage.display bits");
    return false;
  }
  ZX_DEBUG_ASSERT(!usage.display().has_value());
  if (IsDecryptor()) {
    // DecryptorAdapter should not be setting video usage bits.
    if (usage.video().has_value()) {
      LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
      FailLocked("Core codec set disallowed video usage bits for decryptor");
      return false;
    }
    if (port == kOutputPort) {
      usage.video().emplace(fuchsia_sysmem2::kVideoUsageDecryptorOutput);
    }
  } else if (IsCoreCodecHwBased(port)) {
    // Let's see if we can deprecate videoUsageHwProtected, since it's redundant
    // with secure_required.
    uint32_t video_usage = 0;
    if (usage.video().has_value()) {
      video_usage = *usage.video();
    }
    uint32_t allowed_video_usage_bits =
        IsDecoder() ? fuchsia_sysmem2::kVideoUsageHwDecoder : fuchsia_sysmem2::kVideoUsageHwEncoder;
    if (video_usage & ~allowed_video_usage_bits) {
      LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
      FailLocked(
          "Core codec set disallowed video usage bit(s) - port: %d, usage: "
          "0x%08x, allowed: 0x%08x",
          port, video_usage, allowed_video_usage_bits);
      return false;
    }
    if (IsDecoder()) {
      if (!usage.video().has_value()) {
        usage.video().emplace(0);
      }
      *usage.video() |= fuchsia_sysmem2::kVideoUsageHwDecoder;
    } else if (IsEncoder()) {
      if (!usage.video().has_value()) {
        usage.video().emplace(0);
      }
      *usage.video() |= fuchsia_sysmem2::kVideoUsageHwEncoder;
    }
  } else {
    // Despite being a video decoder or encoder, a SW decoder or encoder doesn't
    // count as videoUsageHwDecoder or videoUsageHwEncoder.
    usage.video().reset();
  }

  if (!buffer_collection_constraints->min_buffer_count_for_camping().has_value() ||
      buffer_collection_constraints->min_buffer_count_for_camping().value() < 1) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec set min_buffer_count_for_camping to 0; must set to at least 1.");
    return false;
  }

  if (!buffer_collection_constraints->buffer_memory_constraints().has_value()) {
    // Leaving all fields set to their defaults is fine if that's really true, but this encourages
    // CodecAdapter implementations to set fields in here.
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec must set has_buffer_memory_constraints");
    return false;
  }
  fuchsia_sysmem2::BufferMemoryConstraints& buffer_memory_constraints =
      buffer_collection_constraints->buffer_memory_constraints().value();

  // Sysmem will fail the BufferCollection if the core codec provides constraints that are
  // inconsistent, but we need to check here that the core codec is being consistent with
  // SecureMemoryMode, since sysmem doesn't know about SecureMemoryMode.  Essentially
  // SecureMemoryMode translates into secure_required and secure_permitted in sysmem.  The former
  // is just a bool.  The latter is indicated by listing at least one secure heap.

  // secure_required consistency check
  //
  // CoreCodecSetSecureMemoryMode() informed the core codec of the mode previously.
  bool secure_required = false;
  if (buffer_memory_constraints.secure_required().has_value()) {
    secure_required = *buffer_memory_constraints.secure_required();
  }
  if (!!IsPortSecureRequired(port) != !!secure_required) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec secure_required inconsistent with SecureMemoryMode");
    return false;
  }

  // secure_permitted consistency check
  //
  // If secure is permitted, then the core codec must support at least one non-SYSTEM_RAM heap, as
  // specifying support for a secure heap is how sysmem knows secure_permitted.  We can't directly
  // tell that the non-RAM heap is secure, so this is an approximate check.  In any case
  // secure_required by any sysmem participant will be enforced by sysmem with respect to specific
  // heaps and whether they're secure.  The approximate-ness is ok since this only comes from
  // in-proc, so the check is just for trying to notice if the core codec is filling out
  // inconsistent constraints in a way that sysmem wouldn't otherwise notice.
  bool is_non_ram_heap_found = false;
  if (buffer_memory_constraints.permitted_heaps().has_value()) {
    for (uint32_t iter = 0; iter < buffer_memory_constraints.permitted_heaps()->size(); ++iter) {
      if (buffer_memory_constraints.permitted_heaps()->at(iter).heap_type().value() !=
          bind_fuchsia_sysmem_heap::HEAP_TYPE_SYSTEM_RAM) {
        is_non_ram_heap_found = true;
        break;
      }
    }
  }
  if (IsPortSecurePermitted(port) && !is_non_ram_heap_found) {
    LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_UnreachableError);
    FailLocked("Core codec must specify at least one non-RAM heap when secure_required");
    return false;
  }

  // The rest of the constraints are entirely up to the core codec, and it's up to the core codec
  // to specify self-consistent constraints.  Sysmem will perform additional consistency checks on
  // the constraints.

  return true;
}

thrd_t CodecImpl::fidl_thread() { return shared_fidl_thread_; }

void CodecImpl::SendFreeInputPacketLocked(fuchsia::media::PacketHeader header) {
  // We allow calling this method on StreamControl or InputData ordering domain.
  // Because the InputData ordering domain thread isn't visible to this code,
  // if this isn't the StreamControl then we can only assert that this thread
  // isn't the FIDL thread, because we know the codec's InputData thread isn't
  // the FIDL thread.
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_ || thrd_current() != fidl_thread());
  // We only send using fidl_thread().
  PostToSharedFidl([this, header = std::move(header)]() mutable {
    // See "is_bound_checks" comment up top.
    if (binding_.is_bound()) {
      binding_.events().OnFreeInputPacket(std::move(header));
    }
  });
}

bool CodecImpl::IsInputConfiguredLocked() {
  return IsPortBuffersConfiguredCommonLocked(kInputPort);
}

bool CodecImpl::IsOutputConfiguredLocked() {
  if (!IsPortBuffersConfiguredCommonLocked(kOutputPort)) {
    return false;
  }
  ZX_DEBUG_ASSERT(port_settings_[kOutputPort]);
  if (!port_settings_[kOutputPort]->is_complete_seen_output()) {
    return false;
  }
  return true;
}

bool CodecImpl::IsPortBuffersConfiguredCommonLocked(CodecPort port) {
  // In addition to what we're able to assert here, when
  // is_port_buffers_configured_[port], the core codec also has the port
  // configured.
  ZX_DEBUG_ASSERT(!is_port_buffers_configured_[port] ||
                  port_settings_[port] &&
                      all_buffers_[port].size() == port_settings_[port]->buffer_count());
  return is_port_buffers_configured_[port];
}

bool CodecImpl::IsPortBuffersAtLeastPartiallyConfiguredLocked(CodecPort port) {
  if (IsPortBuffersConfiguredCommonLocked(port)) {
    return true;
  }
  if (!port_settings_[port]) {
    return false;
  }
  ZX_DEBUG_ASSERT(port_settings_[port]);
  ZX_DEBUG_ASSERT(buffer_lifetime_ordinal_[port] % 2 == 1);
  return true;
}

void CodecImpl::Fail(const char* format, ...) {
  va_list args;
  va_start(args, format);
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    vFailLocked(false, format, args);
  }  // ~lock
  // "this" can be deallocated by this point (as soon as ~lock above).
  va_end(args);
}

void CodecImpl::FailLocked(const char* format, ...) {
  va_list args;
  va_start(args, format);
  vFailLocked(false, format, args);
  va_end(args);
  // At this point know "this" is still allocated only because we still hold
  // lock_.  As soon as lock_ is released by the caller, "this" can immediately
  // be deallocated by another thread, if this isn't currently the
  // fidl_thread().
}

void CodecImpl::FailFatalLocked(const char* format, ...) {
  va_list args;
  va_start(args, format);
  // This doesn't return.
  vFailLocked(true, format, args);
  va_end(args);
}

void CodecImpl::vFail(bool is_fatal, const char* format, va_list args) {
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);
    vFailLocked(is_fatal, format, args);
  }  // ~lock
}

// Only meant to be called from Fail() and FailLocked().  Only meant to be
// called for async failure cases after was_logically_bound_ has become true.
// Failures before that point are handled separately.
void CodecImpl::vFailLocked(bool is_fatal, const char* format, va_list args) {
  // TODO(dustingreen): Send epitaph when possible.

  // Let's not have a buffer on the stack, not because it couldn't be done
  // safely, but because we'd potentially run into stack size vs. message length
  // tradeoffs, stack expansion granularity fun, or whatever else.

  va_list args2;
  va_copy(args2, args);

  size_t buffer_bytes = vsnprintf(nullptr, 0, format, args) + 1;

  // ~buffer never actually runs since this method never returns
  std::unique_ptr<char[]> buffer(new char[buffer_bytes]);

  size_t buffer_bytes_2 = vsnprintf(buffer.get(), buffer_bytes, format, args2) + 1;
  (void)buffer_bytes_2;
  // sanity check; should match so go ahead and assert that it does.
  ZX_DEBUG_ASSERT(buffer_bytes == buffer_bytes_2);
  va_end(args2);

  // TODO(dustingreen): It might be worth wiring this up to the log in a more
  // official way, especially if doing so would print a timestamp automatically
  // and/or provide filtering goodness etc.
  const char* message = is_fatal ? "devhost will fail" : "Codec channel will close async";

  // TODO(dustingreen): Send string in buffer via epitaph, when possible.  First
  // we should switch to events so we'll only have the Codec channel not the
  // CodecEvents channel. Note to self: The channel failing server-side may race
  // with trying to send.

  LogEvent(media_metrics::
               StreamProcessorEvents2MigratedMetricDimensionEvent_StreamProcessorFailureAnyReason);
  if (is_fatal) {
    // Default logging to stderr for both driver and non-driver clients
    LOG(ERROR, "%s -- %s", buffer.get(), message);

    abort();
  } else {
    // Default logging to stderr for both driver and non-driver clients
    LOG(WARNING, "%s -- %s", buffer.get(), message);

    UnbindLocked();
  }

  // At this point we know "this" is still allocated only because we still hold
  // lock_.  As soon as lock_ is released by the caller, "this" can immediately
  // be deallocated by another thread, if this isn't currently the
  // fidl_thread().
}

void CodecImpl::PostSerial(async_dispatcher_t* async, fit::closure to_run) {
  zx_status_t result = async::PostTask(async, std::move(to_run));
  ZX_ASSERT(result == ZX_OK);
}

// The implementation of PostToSharedFidl() permits queuing lambdas that use
// "this", despite the fact that the client can call ~CodecImpl at any time
// using the fidl_thread().  If ~CodecImpl is called before the lambda runs, the
// lambda will be deleted instead of run, and the deletion will occur during
// ~CodecImpl while essentially all of CodecImpl is still valid (in case ~lambda
// itself touches any of CodecImpl).
void CodecImpl::PostToSharedFidl(fit::closure to_run) {
  // If shared_fidl_queue_.is_stopped(), then to_run will just be deleted here.
  shared_fidl_queue_.Enqueue(std::move(to_run));
}

// The implementation of PostToStreamControl() doesn't strongly need to guard
// against ~CodecImpl because ~CodecImpl will do
// stream_control_loop_.Shutdown(), which deletes any tasks that haven't already
// run on StreamControl.  We use a ClosureQueue anyway, for at least a couple
// reasons.
//
// Not very importantly, by using a ClosureQueue here, we eliminate a window
// between is_stream_control_done_ = true and the lambda posted to FIDL thread
// shortly after that, during which hypothetically many FIDL dispatches could
// queue to StreamControl without them being consumed by StreamControl.
//
// More importantly, assuming we add an over-full threshold detection to
// ClosureQueue, that can help avoid the server being overwhelmed by a
// badly-behaving client that queues more messages than make any sense given the
// StreamProcessor protocol (which overall limits the number of concurrent
// messages that are allowed / make any sense, but any given message isn't
// necessarily checked for making sense until we're on StreamControl).
void CodecImpl::PostToStreamControl(fit::closure to_run) {
  // If stream_control_queue_.is_stopped(), then to_run will just be deleted
  // here.
  stream_control_queue_.Enqueue(std::move(to_run));
}

bool CodecImpl::IsStoppingLocked() { return was_unbind_started_; }

bool CodecImpl::IsStopping() {
  std::lock_guard<std::mutex> lock(lock_);
  return IsStoppingLocked();
}

bool CodecImpl::IsDecoder() const { return params_.index() == 0; }

bool CodecImpl::IsEncoder() const { return params_.index() == 1; }

bool CodecImpl::IsDecryptor() const { return params_.index() == 2; }

const fuchsia::mediacodec::CreateDecoder_Params& CodecImpl::decoder_params() const {
  ZX_DEBUG_ASSERT(IsDecoder());
  return std::get<fuchsia::mediacodec::CreateDecoder_Params>(params_);
}

const fuchsia::mediacodec::CreateEncoder_Params& CodecImpl::encoder_params() const {
  ZX_DEBUG_ASSERT(IsEncoder());
  return std::get<fuchsia::mediacodec::CreateEncoder_Params>(params_);
}

const fuchsia::media::drm::DecryptorParams& CodecImpl::decryptor_params() const {
  ZX_DEBUG_ASSERT(IsDecryptor());
  return std::get<fuchsia::media::drm::DecryptorParams>(params_);
}

void CodecImpl::LogEvent(
    media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent event_code) const {
  if (!codec_metrics_) {
    return;
  }
  if (!codec_metrics_implementation_dimension_) {
    return;
  }
  codec_metrics_->LogEvent(codec_metrics_implementation_dimension_.value(), event_code);
}

// true - maybe it's the core codec thread.
// false - it's definitely not the core codec thread.
bool CodecImpl::IsPotentiallyCoreCodecThread() {
  return (thrd_current() != stream_control_thread_) && (thrd_current() != fidl_thread());
}

void CodecImpl::HandlePendingInputFormatDetails() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  const fuchsia::media::FormatDetails* input_details = nullptr;
  if (stream_->input_format_details()) {
    input_details = stream_->input_format_details();
  } else {
    input_details = initial_input_format_details_;
  }
  ZX_DEBUG_ASSERT(input_details);
  CoreCodecQueueInputFormatDetails(*input_details);
}

void CodecImpl::onCoreCodecFailCodec(const char* format, ...) {
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_CoreFailureAnyReason);
  std::string local_format = std::string("onCoreCodecFailCodec() called -- ") + format;
  va_list args;
  va_start(args, format);
  vFail(false, local_format.c_str(), args);
  // "this" can be deallocated by this point (as soon as ~lock above).
  va_end(args);
}

void CodecImpl::onCoreCodecFailStream(fuchsia::media::StreamError error) {
  {  // scope lock
    LogEvent(
        media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_StreamFailureAnyReason);
    std::unique_lock<std::mutex> lock(lock_);
    AssertHeld(lock, lock_);
    if (IsStoppingLocked()) {
      // This CodecImpl is already stopping due to a previous FailLocked(),
      // which will result in the Codec channel getting closed soon.  So don't
      // send OnStreamFailed().
      return;
    }

    // We rely on the CodecAdapter and the rest of CodecImpl to only call this method when there's a
    // current stream.
    ZX_DEBUG_ASSERT(stream_ && stream_->stream_lifetime_ordinal() == stream_lifetime_ordinal_);

    if (stream_->output_end_of_stream()) {
      // Tolerate a CodecAdapter failing the stream after output EndOfStream
      // seen, and avoid notifying the client of a stream failure that's too
      // late to matter.
      return;
    }

    if (stream_->failure_seen()) {
      // We already know.  We don't auto-close the stream because the client is
      // in control of stream lifetime, so it's plausible that a CodecAdapter
      // could notify of stream failure more than once.  We can ignore the
      // redundant stream failure to avoid sending OnStreamFailed() again.
      return;
    }
    stream_->SetFailureSeen();
    // avoid hang in FlushEndOfStreamAndCloseStream_StreamControl
    // TODO(https://fxbug.dev/42119800): Clean this up.
    output_end_of_stream_seen_.notify_all();

    if (IsStreamErrorRecoverable(error)) {
      LOG(INFO, "Stream %lu failed: %s. %s", stream_lifetime_ordinal_, ToString(error),
          GetStreamErrorAdditionalHelpText(error));
    } else {
      LOG(ERROR, "Stream %lu failed: %s", stream_lifetime_ordinal_, ToString(error));
    }

    PostToStreamControl([this, stream_lifetime_ordinal = stream_lifetime_ordinal_] {
      ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
      std::unique_lock<std::mutex> lock(lock_);
      if (IsStoppingLocked()) {
        return;
      }
      if (stream_lifetime_ordinal != stream_lifetime_ordinal_) {
        ZX_DEBUG_ASSERT(stream_lifetime_ordinal_ > stream_lifetime_ordinal);
        return;
      }
      // CodecImpl will preserve the ordering of onCoreCodecInputPacketDone() and
      // onCoreCodecFailStream() when sending messages to the client.  Some CodecAdapter(s) _may_
      // order these messages to free input packets which were successfully processed before stream
      // failure, and free input packets which were not successfully processed after stream failure.
      //
      // Other CodecAdapter(s) may simply free all pending input packets, either before
      // onCoreCodecFailStream() or after onCoreCodecFailStream(), without any particular ordering
      // with respect to onCoreCodecFailStream().  In particular, a CodecAdapter may free an input
      // packet before that input packet's data is fully processed so that the input packet can be
      // re-filled with new data (a good thing for performance and efficiency), then later the data
      // that was obtained from that input packet may turn out to cause a processing error later.
      // In such a pipelined CodecAdapter, it's not really feasible or desireable for the
      // CodecAdapter to free an input packet only when the input packet has been fully processed
      // into any output it may generate.
      //
      // If a client needs to determine which input packets generated output packets, and possibly
      // also which output packets those are, use of timstamp_ish values is recommended, as the
      // timstamp_ish mechanism is designed to establish correspondence between input packets and
      // output packets for all StreamProcessor implementations, without imposing any onerous
      // requirement that the CodecAdapter hold onto an input packet until the input packet's data
      // has been fully processed through a processing pipeline.  In short, the relative ordering of
      // OnFreeInputPacket() and OnStreamFailed() varies among StreamProcessor implementations, and
      // should not be relied on in general.  For some specific scenarios with specific known
      // StreamProcessor implementations, a client _may_ be able to reason about their relative
      // order without interference by CodecImpl, but any reliance on their relative order is
      // discouraged and deprecated.
      //
      // To assist the CodecAdapter in freeing any remaining input packets back to the client after
      // a stream failure, we call CoreCodecStopStream().  The CodecAdapter may choose to return any
      // pending input packets during CoreCodecStopStream(), or it's also fine for the CodecAdapter
      // to free any pending input packets after onCoreCodecFailStream() and before the CodecAdapter
      // sees CoreCodecStopStream().  By the end of CoreCodecStopStream(), the CodecAdapter must
      // have called onCoreCodecInputPacketDone() on all previously-pending input packets and must
      // be holding zero pending input packets.
      //
      // For some CodecAdapter implementations, it is acceptable but not preferred for some of the
      // input packets which led to stream failure to be freed before onCoreCodecFailStream(), but
      // again this is not preferred and not recommended.
      EnsureCoreCodecStreamStopped(lock);
    });

    // We're failing the current stream.  We should still queue to the output
    // ordering domain to ensure ordering vs. any previously-sent output on this
    // stream that was sent directly from codec processing thread.
    //
    // This failure is async, in the sense that the client may still be sending
    // input data, and the core codec is expected to just hold onto those
    // packets until the client has moved on from this stream.

    stream_->AssertHeld(this);
    if (stream_->future_discarded()) {
      // No reason to report a stream failure to the client for an obsolete stream.  The client has
      // already moved on from the current stream anyway.  This path won't be taken if the client
      // flushed the stream before moving on.  This permits core codecs to indicate
      // onCoreCodecFailStream() on a stream being cancelled due to a newer stream, without that
      // causing FailLocked() of the whole codec (important), and without sending an extraneous
      // OnStreamFailed() (less important since the client is expected to ignore messages for an
      // obsolete stream).  Ideally a core codec wouldn't trigger onCoreCodecFailStream() during
      // CoreCodecStopStream(), but this path tolerates it.
      return;
    }

    if (!is_on_stream_failed_enabled_) {
      FailLocked(
          "onStreamFailed() with a client that didn't send "
          "EnableOnStreamFailed(), so closing the Codec channel instead.");
      return;
    }

    // The client needs to move on from the failed stream to a new stream, or close the Codec
    // channel.
    //
    // Because we're holding lock continously to this point, we know that this message will be
    // queued to the shared fidl thread before any returned input packets from CodecAdapter after
    // onCoreCodecFailedStream(), and before any messages queued to the fidl thread by the post to
    // StreamControl above.
    PostToSharedFidl([this, stream_lifetime_ordinal = stream_lifetime_ordinal_, error] {
      // See "is_bound_checks" comment up top.
      if (binding_.is_bound()) {
        binding_.events().OnStreamFailed(stream_lifetime_ordinal, error);
      }
    });
  }  // ~lock
}

void CodecImpl::onCoreCodecResetStreamAfterCurrentFrame() {
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    // Calls to onCoreCodecResetStreamAfterCurrentFrame() must be fenced out (by the core codec)
    // during CoreCodecStopStream(), so we know we still have the current stream here.
    ZX_DEBUG_ASSERT(stream_);
    // By the time we post over to StreamControl however, the current stream may no longer be
    // current.  If we've moved on to another stream, it's fine to just ignore the reset stream
    // request for a stream that's no longer current.
    uint64_t stream_lifetime_ordinal = stream_->stream_lifetime_ordinal();
    PostToStreamControl([this, stream_lifetime_ordinal] {
      ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
      {  // scope lock
        std::lock_guard<std::mutex> lock(lock_);

        // Only StreamControl messes with stream_.
        if (!stream_) {
          return;
        }
        ZX_DEBUG_ASSERT(stream_);
        if (stream_->stream_lifetime_ordinal() != stream_lifetime_ordinal) {
          return;
        }
        ZX_DEBUG_ASSERT(stream_->stream_lifetime_ordinal() == stream_lifetime_ordinal);
        stream_->AssertHeld(this);
        if (stream_->future_discarded()) {
          // Ignore since this stream will be gone soon anyway.
          return;
        }
        if (stream_->failure_seen()) {
          // Ignore since this stream has already failed anyway.
          return;
        }
        ZX_DEBUG_ASSERT(is_core_codec_stream_started_);
      }  // ~lock
      CoreCodecResetStreamAfterCurrentFrame();
      return;
    });
  }  // ~lock
}

void CodecImpl::onCoreCodecMidStreamOutputConstraintsChange(bool output_re_config_required) {
  VLOGF("CodecImpl::onCoreCodecMidStreamOutputConstraintsChange(): re-config: %d",
        output_re_config_required);
  // For now, the core codec thread is the only thread this gets called from.
  ZX_DEBUG_ASSERT(IsPotentiallyCoreCodecThread());

  // Passing false for this parameter is deprecated (and not done by any current CodecAdapter).
  // The old use of false here has been superseded by onCoreCodecOutputFormatChange().
  //
  // We have an output constraints change that does demand output buffer re-config before more
  // output data.
  ZX_DEBUG_ASSERT(output_re_config_required);

  // We post over to StreamControl domain because we need to synchronize
  // with any changes to stream state that might be driven by the client.
  // When we get over there to StreamControl, we'll check if we're still
  // talking about the same stream_lifetime_ordinal, and if not, we ignore
  // the event, because a new stream may or may not have the same output
  // settings, and we'll be re-generating an OnOutputConstraints() as needed
  // from current/later core codec output constraints anyway.  Here are the
  // possibilities:
  //   * Prior to the client moving to a new stream, we process this event
  //     on StreamControl ordering domain and have bumped
  //     buffer_lifetime_ordinal by the time we start any subsequent
  //     new stream from the client, which means we'll require the client
  //     to catch up to the new buffer_lifetime_ordinal before we start
  //     that new stream.
  //   * The client moves to a new stream before this event gets over to
  //     StreamControl.  In this case we ignore the event on StreamControl
  //     domain since its stale by that point, but instead we use
  //     omx_meh_output_buffer_constraints_version_ordinal_ to cause the
  //     client's next stream to start with a new OnOutputConstraints() that
  //     the client must catch up to before the stream can fully start.
  //     This way we know we're not ignoring a potential change to
  //     nBufferCountMin or anything like that.
  uint64_t local_stream_lifetime_ordinal;
  {  // scope lock
    std::lock_guard<std::mutex> lock(lock_);

    // The core codec is only allowed to call this mehtod while there's an
    // active stream.
    ZX_DEBUG_ASSERT(IsStreamActiveLocked());

    // The client is allowed to essentially forget what the format is on any
    // mid-stream buffer config change, so remember to re-send the format to the
    // client before the next output packet of this stream.
    stream_->SetOutputFormatPending();

    // For asserts.
    stream_->SetMidStreamOutputConstraintsChangeActive();

    // This part is not speculative.  The core codec has indicated that it's at
    // least meh about the current output config, so ensure we do a required
    // OnOutputConstraints() before the next stream starts, even if the client
    // moves on to a new stream such that the speculative part below becomes
    // stale.
    core_codec_meh_output_buffer_constraints_version_ordinal_ =
        port_settings_[kOutputPort]
            ? port_settings_[kOutputPort]->buffer_constraints_version_ordinal()
            : 0;
    // Speculative part - this part is speculative, in that we don't know if
    // this post over to StreamControl will beat any client driving to a new
    // stream.  So we snap the stream_lifetime_ordinal so we know whether to
    // ignore the post once it reaches StreamControl.
    local_stream_lifetime_ordinal = stream_lifetime_ordinal_;
  }  // ~lock
  PostToStreamControl([this, stream_lifetime_ordinal = local_stream_lifetime_ordinal] {
    MidStreamOutputConstraintsChange(stream_lifetime_ordinal);
  });
}

void CodecImpl::onCoreCodecOutputFormatChange() {
  ZX_DEBUG_ASSERT(IsPotentiallyCoreCodecThread());
  std::lock_guard<std::mutex> lock(lock_);
  ZX_DEBUG_ASSERT(IsStreamActiveLocked());
  // In future we could relax this requirement, but for now we don't allow
  // output format changes, output packets, or EOS while mid-stream constraints
  // change is active.
  ZX_DEBUG_ASSERT(!stream_->is_mid_stream_output_constraints_change_active());
  // Next time the core codec asks to output a packet, we'll send the format
  // first.
  stream_->SetOutputFormatPending();
}

void CodecImpl::onCoreCodecInputPacketDone(CodecPacket* packet) {
  // Free/busy coherency from Codec interface to core codec doesn't involve
  // trusting the client, so assert we're doing it right server-side.
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    // The core codec says the buffer-referening in-flight lifetime of this
    // packet is over.  We'll set the buffer again when this packet get's used
    // by the client again to deliver more input data.
    packet->SetBuffer(nullptr);
    // Unfortunately we have to insist that the core codec not call
    // onCoreCodecInputPacketDone() arbitrarily late because we need to know
    // when it's safe to deallocate binding_, and the core codec, etc.  So the
    // rule is the core codec needs to ensure that all calls to stream-related
    // callbacks have completed (to structure-touching degree; not
    // code-unloading degree) before CoreCodecStopStream() returns.
    ZX_DEBUG_ASSERT(is_core_codec_stream_started_);
    ZX_DEBUG_ASSERT(!all_packets_[kInputPort][packet->packet_index()]->is_free());
    all_packets_[kInputPort][packet->packet_index()]->SetFree(true);
    fuchsia::media::PacketHeader header;
    header.set_buffer_lifetime_ordinal(packet->buffer_lifetime_ordinal());
    header.set_packet_index(packet->packet_index());
    SendFreeInputPacketLocked(std::move(header));
  }  // ~lock
}

void CodecImpl::onCoreCodecOutputPacket(CodecPacket* packet, bool error_detected_before,
                                        bool error_detected_during) {
  ZX_DEBUG_ASSERT(IsPotentiallyCoreCodecThread());

  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);

    // The core codec shouldn't output a packet until after
    // CoreCodecStartStream() and input data availability in the case that
    // output buffer config was already suitable, or until after
    // CoreCodecMidStreamOutputBufferReConfigFinish() in the case that output
    // buffer config wasn't suitable (not configured or not suitable) or
    // changed mid-stream.  See also comments in codec_adapter.h.
    ZX_DEBUG_ASSERT(IsOutputConfiguredLocked());

    // Before we send the packet, we check whether the stream has output format
    // pending, which means we need to send the output format before the output
    // packet (and clear the pending state).
    ZX_DEBUG_ASSERT(IsStreamActiveLocked());

    ZX_DEBUG_ASSERT(!stream_->is_mid_stream_output_constraints_change_active());

    if (stream_->output_format_pending()) {
      stream_->ClearOutputFormatPending();
      uint64_t stream_lifetime_ordinal = stream_lifetime_ordinal_;
      uint64_t new_output_format_details_version_ordinal =
          next_output_format_details_version_ordinal_++;
      fuchsia::media::StreamOutputFormat output_format;
      {  // scope unlock
        ScopedUnlock unlock(lock);
        output_format = CoreCodecGetOutputFormat(stream_lifetime_ordinal,
                                                 new_output_format_details_version_ordinal);
      }  // ~unlock
      ZX_DEBUG_ASSERT(output_format.has_format_details());
      // Stream change while unlocked above won't happen because we're on
      // InputData domain which is fenced as part of stream switch.
      ZX_DEBUG_ASSERT(stream_lifetime_ordinal == stream_lifetime_ordinal_);
      ZX_DEBUG_ASSERT(new_output_format_details_version_ordinal ==
                      next_output_format_details_version_ordinal_ - 1);
      ZX_DEBUG_ASSERT(sent_format_details_version_ordinal_[kOutputPort] + 1 ==
                      new_output_format_details_version_ordinal);
      sent_format_details_version_ordinal_[kOutputPort] = new_output_format_details_version_ordinal;
      PostToSharedFidl([this, output_format = std::move(output_format)]() mutable {
        // See "is_bound_checks" comment up top.
        if (binding_.is_bound()) {
          binding_.events().OnOutputFormat(std::move(output_format));
        }
      });
    }
  }

  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    // This helps verify that packet lifetimes are coherent, but we can't do the
    // same for buffer_index because VP9 has show_existing_frame which is
    // allowed to output the same buffer repeatedly.
    //
    // TODO(dustingreen): We could _optionally_ verify that buffer lifetimes are
    // coherent for codecs that don't output the same buffer repeatedly and
    // concurrently.
    all_packets_[kOutputPort][packet->packet_index()]->SetFree(false);
    ZX_DEBUG_ASSERT(packet->has_start_offset());
    ZX_DEBUG_ASSERT(packet->has_valid_length_bytes());
    // packet->has_timestamp_ish() is optional even if
    // promise_separate_access_units_on_input is true.  We do want to enforce
    // that the client gets no set timestamp_ish values if the client didn't
    // promise_separate_access_units_on_input.
    bool has_timestamp_ish =
        (!IsDecoder() || (decoder_params().has_promise_separate_access_units_on_input() &&
                          decoder_params().promise_separate_access_units_on_input())) &&
        packet->has_timestamp_ish();
    fuchsia::media::Packet p;
    p.mutable_header()->set_buffer_lifetime_ordinal(packet->buffer_lifetime_ordinal());
    p.mutable_header()->set_packet_index(packet->packet_index());
    p.set_buffer_index(packet->buffer()->index());
    p.set_stream_lifetime_ordinal(stream_lifetime_ordinal_);
    p.set_start_offset(packet->start_offset());
    p.set_valid_length_bytes(packet->valid_length_bytes());
    if (has_timestamp_ish) {
      p.set_timestamp_ish(packet->timestamp_ish());
    }
    if (packet->has_key_frame()) {
      p.set_key_frame(packet->key_frame());
    }
    p.set_start_access_unit(true);
    p.set_known_end_access_unit(true);
    PostToSharedFidl(
        [this, p = std::move(p), error_detected_before, error_detected_during]() mutable {
          // See "is_bound_checks" comment up top.
          if (binding_.is_bound()) {
            if (kLogTimestampDelay) {
              LOG(INFO, "output timestamp: has: %d value: 0x%" PRIx64, p.has_timestamp_ish(),
                  p.has_timestamp_ish() ? p.timestamp_ish() : 0);
            }
            binding_.events().OnOutputPacket(std::move(p), error_detected_before,
                                             error_detected_during);
          }
        });
  }  // ~lock
}

void CodecImpl::onCoreCodecOutputEndOfStream(bool error_detected_before) {
  VLOGF("CodecImpl::onCoreCodecOutputEndOfStream()");
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_CoreEndOfStreamOuput);
  {  // scope lock
    std::unique_lock<std::mutex> lock(lock_);
    ZX_DEBUG_ASSERT(IsStreamActiveLocked());
    ZX_DEBUG_ASSERT(!stream_->is_mid_stream_output_constraints_change_active());
    stream_->SetOutputEndOfStream();
    output_end_of_stream_seen_.notify_all();
    PostToSharedFidl([this, stream_lifetime_ordinal = stream_lifetime_ordinal_,
                      error_detected_before] {
      // See "is_bound_checks" comment up top.
      if (binding_.is_bound()) {
        LogEvent(media_metrics::
                     StreamProcessorEvents2MigratedMetricDimensionEvent_StreamEndOfStreamOutput);
        binding_.events().OnOutputEndOfStream(stream_lifetime_ordinal, error_detected_before);
      }
    });
  }  // ~lock
}

void CodecImpl::onCoreCodecLogEvent(
    media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent event_code) {
  // A CodecAdapter sub-class that ever calls LogEvent() must override
  // CoreCodecMetricsImplementation() and must not return std::nullopt.
  ZX_DEBUG_ASSERT(codec_metrics_implementation_dimension_);
  LogEvent(event_code);
}

CodecImpl::Stream::Stream(const CodecImpl* const parent, uint64_t stream_lifetime_ordinal)
    : parent_(parent), stream_lifetime_ordinal_(stream_lifetime_ordinal) {
  // nothing else to do here
}

void CodecImpl::Stream::AssertHeld(const CodecImpl* const parent) {
  ZX_ASSERT(parent == parent_);
  // And of course the lock is the same lock also.
  ZX_ASSERT(&parent->lock_ == &parent_->lock_);
}

uint64_t CodecImpl::Stream::stream_lifetime_ordinal() { return stream_lifetime_ordinal_; }

void CodecImpl::Stream::SetFutureDiscarded() {
  ZX_DEBUG_ASSERT(!future_discarded_);
  future_discarded_ = true;
}

bool CodecImpl::Stream::future_discarded() { return future_discarded_; }

void CodecImpl::Stream::SetFutureFlushEndOfStream() {
  ZX_DEBUG_ASSERT(!future_flush_end_of_stream_);
  future_flush_end_of_stream_ = true;
}

bool CodecImpl::Stream::future_flush_end_of_stream() { return future_flush_end_of_stream_; }

CodecImpl::Stream::~Stream() {
  VLOGF("~Stream() stream_lifetime_ordinal: %lu", stream_lifetime_ordinal_);
}

void CodecImpl::Stream::SetInputFormatDetails(
    std::unique_ptr<fuchsia::media::FormatDetails> input_format_details) {
  // This is allowed to happen multiple times per stream.
  input_format_details_ = std::move(input_format_details);
}

const fuchsia::media::FormatDetails* CodecImpl::Stream::input_format_details() {
  return input_format_details_.get();
}

void CodecImpl::Stream::SetOobConfigPending(bool pending) {
  // SetOobConfigPending(true) is legal regardless of current state, but
  // SetOobConfigPending(false) is only legal if the state is currently true.
  ZX_DEBUG_ASSERT(pending || oob_config_pending_);
  oob_config_pending_ = pending;
}

bool CodecImpl::Stream::oob_config_pending() { return oob_config_pending_; }

void CodecImpl::Stream::SetInputEndOfStream() {
  ZX_DEBUG_ASSERT(!input_end_of_stream_);
  input_end_of_stream_ = true;
}

bool CodecImpl::Stream::input_end_of_stream() { return input_end_of_stream_; }

void CodecImpl::Stream::SetOutputEndOfStream() {
  ZX_DEBUG_ASSERT(!output_end_of_stream_);
  output_end_of_stream_ = true;
}

bool CodecImpl::Stream::output_end_of_stream() { return output_end_of_stream_; }

void CodecImpl::Stream::SetFailureSeen() {
  ZX_DEBUG_ASSERT(!failure_seen_);
  failure_seen_ = true;
}

bool CodecImpl::Stream::failure_seen() { return failure_seen_; }

void CodecImpl::Stream::SetOutputFormatPending() { output_format_pending_ = true; }

void CodecImpl::Stream::ClearOutputFormatPending() { output_format_pending_ = false; }

bool CodecImpl::Stream::output_format_pending() { return output_format_pending_; }

void CodecImpl::Stream::SetMidStreamOutputConstraintsChangeActive() {
  ZX_DEBUG_ASSERT(!is_mid_stream_output_constraints_change_active_);
  is_mid_stream_output_constraints_change_active_ = true;
}

void CodecImpl::Stream::ClearMidStreamOutputConstraintsChangeActive() {
  ZX_DEBUG_ASSERT(is_mid_stream_output_constraints_change_active_);
  is_mid_stream_output_constraints_change_active_ = false;
}

bool CodecImpl::Stream::is_mid_stream_output_constraints_change_active() {
  return is_mid_stream_output_constraints_change_active_;
}

CodecImpl::PortSettings::PortSettings(CodecImpl* parent, CodecPort port,
                                      fuchsia::media::StreamBufferPartialSettings partial_settings)
    : parent_(parent),
      port_(port),
      partial_settings_(std::make_unique<fuchsia::media::StreamBufferPartialSettings>(
          std::move(partial_settings))),
      gone_marker_(std::make_shared<std::monostate>()) {
  // nothing else to do here
}

CodecImpl::PortSettings::~PortSettings() {
  // To be safe, the unbind needs to occur on the FIDL thread.  In addition, we want to send a clean
  // Close() to avoid causing the LogicalBufferCollection to fail.  Since we're not a crashing
  // process, this is a clean close by definition.
  //
  // TODO(https://fxbug.dev/42112876): Consider _not_ sending Close() for unexpected failures
  // initiated by the server. Consider whether to have a Close() on StreamProcessor to disambiguate
  // clean vs. unexpected StreamProcessor channel close.
  if (thrd_current() != parent_->fidl_thread()) {
    parent_->PostToSharedFidl([buffer_collection = std::move(buffer_collection_)] {
      // Sysmem will notice the Close() before the PEER_CLOSED.
      if (!!buffer_collection && buffer_collection->is_valid()) {
        // ignore potential one-way send failure
        (void)(*buffer_collection)->Release();
      }
      // ~buffer_collection on FIDL thread
    });
    buffer_collection_.reset();
  } else {
    if (!!buffer_collection_) {
      // ignore potential one-way send failure
      (void)(*buffer_collection_)->Release();
    }
  }
}

void CodecImpl::PortSettings::SetBufferCollectionInfo(
    fuchsia_sysmem2::BufferCollectionInfo buffer_collection_info) {
  ZX_DEBUG_ASSERT(!buffer_collection_info_);
  buffer_collection_info_ =
      std::make_unique<fuchsia_sysmem2::BufferCollectionInfo>(std::move(buffer_collection_info));
}

const fuchsia_sysmem2::BufferCollectionInfo& CodecImpl::PortSettings::buffer_collection_info() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return *buffer_collection_info_;
}

uint64_t CodecImpl::PortSettings::buffer_lifetime_ordinal() {
  return partial_settings_->buffer_lifetime_ordinal();
}

uint64_t CodecImpl::PortSettings::buffer_constraints_version_ordinal() {
  return partial_settings_->buffer_constraints_version_ordinal();
}

uint32_t CodecImpl::PortSettings::packet_count() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return static_cast<uint32_t>(buffer_collection_info_->buffers()->size());
}

uint32_t CodecImpl::PortSettings::buffer_count() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return static_cast<uint32_t>(buffer_collection_info_->buffers()->size());
}

fuchsia_sysmem2::CoherencyDomain CodecImpl::PortSettings::coherency_domain() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return buffer_collection_info_->settings()->buffer_settings()->coherency_domain().value();
}

const fuchsia::media::StreamBufferPartialSettings& CodecImpl::PortSettings::partial_settings() {
  return *partial_settings_;
}

fidl::ClientEnd<fuchsia_sysmem2::BufferCollectionToken> CodecImpl::PortSettings::TakeToken() {
  ZX_DEBUG_ASSERT(!partial_settings_->has_sysmem_token());
  ZX_DEBUG_ASSERT(partial_settings_->has_sysmem2_token());
  auto token = fidl::ClientEnd<fuchsia_sysmem2::BufferCollectionToken>(
      partial_settings_->mutable_sysmem2_token()->TakeChannel());
  partial_settings_->clear_sysmem2_token();
  return token;
}

zx::vmo CodecImpl::PortSettings::TakeVmo(uint32_t buffer_index) {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  ZX_DEBUG_ASSERT(buffer_index < buffer_collection_info_->buffers()->size());
  return std::move(buffer_collection_info_->buffers()->at(buffer_index).vmo().value());
}

fidl::ServerEnd<fuchsia_sysmem2::BufferCollection>
CodecImpl::PortSettings::NewBufferCollectionRequest(
    async_dispatcher_t* dispatcher,
    CodecImpl::Client<fuchsia_sysmem2::BufferCollection>::ErrorFunction on_error) {
  ZX_DEBUG_ASSERT(thrd_current() == parent_->fidl_thread());
  ZX_DEBUG_ASSERT(!buffer_collection_);
  auto collection_endpoints = fidl::CreateEndpoints<fuchsia_sysmem2::BufferCollection>();
  ZX_ASSERT(collection_endpoints.is_ok());
  buffer_collection_ = std::make_unique<Client<fuchsia_sysmem2::BufferCollection>>();
  auto wrapping_error_handler = [this, weak_gone_marker = std::weak_ptr(gone_marker_),
                                 on_error = std::move(on_error)](fidl::UnbindInfo unbind_info) {
    if (!weak_gone_marker.lock()) {
      // ~on_error - don't call CodecImpl's error handler because CodecImpl already deleted "this",
      // so doesn't expect its error handler to be called.
      return;
    }
    // We're on the fidl thread so "this" isn't gone yet (the lock() above isn't sufficient alone,
    // even if we held the shared_ptr until here, since it's not a shared_ptr<CodecImpl>). Go ahead
    // and use "this" to be able to assert we're on the expected thread, and hope that if we're not
    // on the expected thread, that parent_ is still alive so we can fail this assert, or that we
    // crash here so we'd know something is wrong.
    ZX_DEBUG_ASSERT(thrd_current() == parent_->fidl_thread());
    // Call CodecImpl's error handler, but only if "this" still exists (see above).
    on_error(unbind_info);
  };
  buffer_collection_->Bind(std::move(collection_endpoints->client), dispatcher,
                           std::move(wrapping_error_handler));
  return std::move(collection_endpoints->server);
}

std::unique_ptr<CodecImpl::Client<fuchsia_sysmem2::BufferCollection>>&
CodecImpl::PortSettings::buffer_collection() {
  ZX_DEBUG_ASSERT(thrd_current() == parent_->fidl_thread());
  return buffer_collection_;
}

void CodecImpl::PortSettings::UnbindBufferCollection() {
  ZX_DEBUG_ASSERT(thrd_current() == parent_->fidl_thread());
  // Unbind even if there are outstanding requests; delete context for those if they exist.
  buffer_collection_.reset();
}

bool CodecImpl::PortSettings::is_complete_seen_output() {
  ZX_DEBUG_ASSERT(port_ == kOutputPort);
  return is_complete_seen_output_;
}

void CodecImpl::PortSettings::SetCompleteSeenOutput() {
  ZX_DEBUG_ASSERT(port_ == kOutputPort);
  ZX_DEBUG_ASSERT(thrd_current() == parent_->fidl_thread());
  ZX_DEBUG_ASSERT(!is_complete_seen_output_);
  is_complete_seen_output_ = true;
}

uint64_t CodecImpl::PortSettings::vmo_usable_start(uint32_t buffer_index) {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  ZX_DEBUG_ASSERT(buffer_index < buffer_collection_info_->buffers()->size());
  return buffer_collection_info_->buffers()->at(buffer_index).vmo_usable_start().value();
}

uint64_t CodecImpl::PortSettings::vmo_usable_size() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return buffer_collection_info_->settings()->buffer_settings()->size_bytes().value();
}

bool CodecImpl::PortSettings::is_secure() {
  ZX_DEBUG_ASSERT(buffer_collection_info_);
  return buffer_collection_info_->settings()->buffer_settings()->is_secure().value();
}

//
// CoreCodec wrappers, for the asserts.  These asserts, and the way we ensure
// at compile time that this class has a method for every method of
// CodecAdapter, are essentially costing a double vtable call instead of a
// single vtable call.  If we don't like that at some point, we can remove the
// private CodecAdapter inheritance from CodecImpl and have these be normal
// methods instead of virtual methods.
//

void CodecImpl::CoreCodecInit(const fuchsia::media::FormatDetails& initial_input_format_details) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecInit(initial_input_format_details);
}

void CodecImpl::CoreCodecSetSecureMemoryMode(
    CodecPort port, fuchsia::mediacodec::SecureMemoryMode secure_memory_mode) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecSetSecureMemoryMode(port, secure_memory_mode);
}

fuchsia_sysmem2::BufferCollectionConstraints CodecImpl::CoreCodecGetBufferCollectionConstraints2(
    CodecPort port, const fuchsia::media::StreamBufferConstraints& stream_buffer_constraints,
    const fuchsia::media::StreamBufferPartialSettings& partial_settings) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  // We don't intend to send the sysmem token to the core codec directly, just
  // because it doesn't really need to participate directly that way, and this
  // lets us keep direct interaction with sysmem in CodecImpl instead of each
  // core codec.
  ZX_DEBUG_ASSERT(!partial_settings.has_sysmem2_token());
  ZX_DEBUG_ASSERT(!partial_settings.has_sysmem_token());
  return codec_adapter_->CoreCodecGetBufferCollectionConstraints2(port, stream_buffer_constraints,
                                                                  partial_settings);
}

void CodecImpl::CoreCodecSetBufferCollectionInfo(
    CodecPort port, const fuchsia_sysmem2::BufferCollectionInfo& buffer_collection_info) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  codec_adapter_->CoreCodecSetBufferCollectionInfo(port, buffer_collection_info);
}

void CodecImpl::CoreCodecAddBuffer(CodecPort port, const CodecBuffer* buffer) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  codec_adapter_->CoreCodecAddBuffer(port, buffer);
}

void CodecImpl::CoreCodecConfigureBuffers(
    CodecPort port, const std::vector<std::unique_ptr<CodecPacket>>& packets) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  codec_adapter_->CoreCodecConfigureBuffers(port, packets);
}

void CodecImpl::CoreCodecEnsureBuffersNotConfigured(CodecPort port) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && (thrd_current() == fidl_thread() ||
                                          thrd_current() == stream_control_thread_));
  codec_adapter_->CoreCodecEnsureBuffersNotConfigured(port);
}

void CodecImpl::CoreCodecStartStream() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_StreamCreated);
  codec_adapter_->CoreCodecStartStream();
}

void CodecImpl::CoreCodecQueueInputFormatDetails(
    const fuchsia::media::FormatDetails& per_stream_override_format_details) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecQueueInputFormatDetails(per_stream_override_format_details);
}

void CodecImpl::CoreCodecQueueInputPacket(CodecPacket* packet) {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecQueueInputPacket(packet);
}

void CodecImpl::CoreCodecQueueInputEndOfStream() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_CoreEndOfStreamInput);
  codec_adapter_->CoreCodecQueueInputEndOfStream();
}

void CodecImpl::CoreCodecStopStream() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  LogEvent(media_metrics::StreamProcessorEvents2MigratedMetricDimensionEvent_StreamDeleted);
  codec_adapter_->CoreCodecStopStream();
}

void CodecImpl::CoreCodecResetStreamAfterCurrentFrame() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecResetStreamAfterCurrentFrame();
}

std::optional<media_metrics::StreamProcessorEvents2MigratedMetricDimensionImplementation>
CodecImpl::CoreCodecMetricsImplementation() {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  return codec_adapter_->CoreCodecMetricsImplementation();
}

bool CodecImpl::IsCoreCodecRequiringOutputConfigForFormatDetection() {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread() || thrd_current() == stream_control_thread_);
  return codec_adapter_->IsCoreCodecRequiringOutputConfigForFormatDetection();
}

bool CodecImpl::IsCoreCodecMappedBufferUseful(CodecPort port) {
  ZX_DEBUG_ASSERT(port == kInputPort && thrd_current() == stream_control_thread_ ||
                  port == kOutputPort && thrd_current() == fidl_thread());
  return codec_adapter_->IsCoreCodecMappedBufferUseful(port);
}

bool CodecImpl::IsCoreCodecHwBased(CodecPort port) {
  return codec_adapter_->IsCoreCodecHwBased(port);
}

zx::unowned_bti CodecImpl::CoreCodecBti() {
  ZX_DEBUG_ASSERT(IsCoreCodecHwBased(kInputPort) || IsCoreCodecHwBased(kOutputPort));
  return codec_adapter_->CoreCodecBti();
}

std::unique_ptr<const fuchsia::media::StreamBufferConstraints>
CodecImpl::CoreCodecBuildNewInputConstraints() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  std::unique_ptr<const fuchsia::media::StreamBufferConstraints> constraints =
      codec_adapter_->CoreCodecBuildNewInputConstraints();
  ZX_DEBUG_ASSERT(constraints);
  ZX_DEBUG_ASSERT(constraints->has_buffer_constraints_version_ordinal());
  return constraints;
}
// Caller must ensure that this is called only on one thread at a time, only
// during setup, during a core codec initiated mid-stream format change, or
// during stream start before any input data has been delivered for the new
// stream.
std::unique_ptr<const fuchsia::media::StreamOutputConstraints>
CodecImpl::CoreCodecBuildNewOutputConstraints(
    uint64_t stream_lifetime_ordinal, uint64_t new_output_buffer_constraints_version_ordinal,
    bool buffer_constraints_action_required) {
  ZX_DEBUG_ASSERT(IsPotentiallyCoreCodecThread() || thrd_current() == stream_control_thread_);
  std::unique_ptr<const fuchsia::media::StreamOutputConstraints> constraints =
      codec_adapter_->CoreCodecBuildNewOutputConstraints(
          stream_lifetime_ordinal, new_output_buffer_constraints_version_ordinal,
          buffer_constraints_action_required);
  ZX_DEBUG_ASSERT(constraints);
  ZX_DEBUG_ASSERT(constraints->has_stream_lifetime_ordinal());
  ZX_DEBUG_ASSERT(constraints->stream_lifetime_ordinal() == stream_lifetime_ordinal);
  ZX_DEBUG_ASSERT(constraints->has_buffer_constraints());
  ZX_DEBUG_ASSERT(constraints->buffer_constraints().has_buffer_constraints_version_ordinal());
  ZX_DEBUG_ASSERT(constraints->buffer_constraints().buffer_constraints_version_ordinal() ==
                  new_output_buffer_constraints_version_ordinal);
  ZX_DEBUG_ASSERT(constraints->has_buffer_constraints_action_required());
  ZX_DEBUG_ASSERT(constraints->buffer_constraints_action_required() ==
                  buffer_constraints_action_required);
  return constraints;
}

fuchsia::media::StreamOutputFormat CodecImpl::CoreCodecGetOutputFormat(
    uint64_t stream_lifetime_ordinal, uint64_t new_output_format_details_version_ordinal) {
  ZX_DEBUG_ASSERT(IsPotentiallyCoreCodecThread());
  fuchsia::media::StreamOutputFormat format = codec_adapter_->CoreCodecGetOutputFormat(
      stream_lifetime_ordinal, new_output_format_details_version_ordinal);
  ZX_DEBUG_ASSERT(format.has_stream_lifetime_ordinal());
  ZX_DEBUG_ASSERT(format.stream_lifetime_ordinal() == stream_lifetime_ordinal);
  ZX_DEBUG_ASSERT(format.has_format_details());
  ZX_DEBUG_ASSERT(format.format_details().has_format_details_version_ordinal());
  ZX_DEBUG_ASSERT(format.format_details().format_details_version_ordinal() ==
                  new_output_format_details_version_ordinal);
  return format;
}

void CodecImpl::CoreCodecMidStreamOutputBufferReConfigPrepare() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecMidStreamOutputBufferReConfigPrepare();
}

void CodecImpl::CoreCodecMidStreamOutputBufferReConfigFinish() {
  ZX_DEBUG_ASSERT(thrd_current() == stream_control_thread_);
  codec_adapter_->CoreCodecMidStreamOutputBufferReConfigFinish();
}

void CodecImpl::CoreCodecRecycleOutputPacket(CodecPacket* packet) {
  ZX_DEBUG_ASSERT(thrd_current() == fidl_thread());
  codec_adapter_->CoreCodecRecycleOutputPacket(packet);
}

void CodecImpl::CoreCodecSetStreamControlProfile(zx::unowned_thread stream_control_thread) {
  codec_adapter_->CoreCodecSetStreamControlProfile(std::move(stream_control_thread));
}