media/filters/decoder_stream.cc

   1 // Copyright 2014 The Chromium Authors
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "media/filters/decoder_stream.h"
   6
   7 #include <utility>
   8
   9 #include "base/feature_list.h"
  10 #include "base/functional/bind.h"
  11 #include "base/functional/callback_helpers.h"
  12 #include "base/location.h"
  13 #include "base/logging.h"
  14 #include "base/task/bind_post_task.h"
  15 #include "base/task/sequenced_task_runner.h"
  16 #include "base/trace_event/trace_event.h"
  17 #include "media/base/cdm_context.h"
  18 #include "media/base/decoder_buffer.h"
  19 #include "media/base/limits.h"
  20 #include "media/base/media_log.h"
  21 #include "media/base/media_switches.h"
  22 #include "media/base/media_util.h"
  23 #include "media/base/timestamp_constants.h"
  24 #include "media/base/video_decoder.h"
  25 #include "media/base/video_frame.h"
  26 #include "media/filters/decrypting_demuxer_stream.h"
  27
  28 namespace media {
  29
  30 #define FUNCTION_DVLOG(level) \
  31   DVLOG(level) << __func__ << "<" << GetStreamTypeString() << ">"
  32
  33 template <DemuxerStream::Type StreamType>
  34 static const char* GetDecodeTraceString();
  35 template <DemuxerStream::Type StreamType>
  36 static const char* GetReadTraceString();
  37 template <DemuxerStream::Type StreamType>
  38 static const char* GetDemuxerReadTraceString();
  39 template <DemuxerStream::Type StreamType>
  40 static const char* GetPrepareTraceString();
  41
  42 template <>
  43 const char* GetDecodeTraceString<DemuxerStream::VIDEO>() {
  44   return "VideoDecoderStream::Decode";
  45 }
  46
  47 template <>
  48 const char* GetDecodeTraceString<DemuxerStream::AUDIO>() {
  49   return "AudioDecoderStream::Decode";
  50 }
  51
  52 template <>
  53 const char* GetReadTraceString<DemuxerStream::VIDEO>() {
  54   return "VideoDecoderStream::Read";
  55 }
  56
  57 template <>
  58 const char* GetReadTraceString<DemuxerStream::AUDIO>() {
  59   return "AudioDecoderStream::Read";
  60 }
  61
  62 template <>
  63 const char* GetDemuxerReadTraceString<DemuxerStream::VIDEO>() {
  64   return "VideoDecoderStream::ReadFromDemuxerStream";
  65 }
  66
  67 template <>
  68 const char* GetDemuxerReadTraceString<DemuxerStream::AUDIO>() {
  69   return "AudioDecoderStream::ReadFromDemuxerStream";
  70 }
  71
  72 template <>
  73 const char* GetPrepareTraceString<DemuxerStream::VIDEO>() {
  74   return "VideoDecoderStream::PrepareOutput";
  75 }
  76
  77 template <>
  78 const char* GetPrepareTraceString<DemuxerStream::AUDIO>() {
  79   return "AudioDecoderStream::PrepareOutput";
  80 }
  81
  82 const char* GetStatusString(const DecoderStatus& status) {
  83   // TODO(crbug.com/1129662): Replace this with generic Status-to-string.
  84   switch (status.code()) {
  85     case DecoderStatus::Codes::kOk:
  86       return "okay";
  87     case DecoderStatus::Codes::kAborted:
  88       return "aborted";
  89     default:
  90       return "decode_error";
  91   }
  92
  93   NOTREACHED_NORETURN();
  94 }
  95
  96 template <DemuxerStream::Type StreamType>
  97 DecoderStream<StreamType>::DecoderStream(
  98     std::unique_ptr<DecoderStreamTraits<StreamType>> traits,
  99     scoped_refptr<base::SequencedTaskRunner> task_runner,
 100     CreateDecodersCB create_decoders_cb,
 101     MediaLog* media_log)
 102     : traits_(std::move(traits)),
 103       task_runner_(std::move(task_runner)),
 104       media_log_(media_log),
 105       state_(STATE_UNINITIALIZED),
 106       stream_(nullptr),
 107       cdm_context_(nullptr),
 108       decoder_produced_a_frame_(false),
 109       decoder_selector_(task_runner_, std::move(create_decoders_cb), media_log),
 110       decoding_eos_(false),
 111       preparing_output_(false),
 112       pending_decode_requests_(0),
 113       duration_tracker_(8),
 114       received_config_change_during_reinit_(false),
 115       pending_demuxer_read_(false) {
 116   FUNCTION_DVLOG(1);
 117 }
 118
 119 template <DemuxerStream::Type StreamType>
 120 DecoderStream<StreamType>::~DecoderStream() {
 121   FUNCTION_DVLOG(1);
 122   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 123
 124   if (init_cb_) {
 125     task_runner_->PostTask(FROM_HERE,
 126                            base::BindOnce(std::move(init_cb_), false));
 127   }
 128   if (read_cb_) {
 129     read_cb_ = base::BindPostTaskToCurrentDefault(std::move(read_cb_));
 130     SatisfyRead(DecoderStatus::Codes::kAborted);
 131   }
 132   if (reset_cb_)
 133     task_runner_->PostTask(FROM_HERE, std::move(reset_cb_));
 134
 135   if (preparing_output_)
 136     CompletePrepare(nullptr);
 137
 138   // Don't manually reset anything here; rely on the order of member variables
 139   // within the header, which enforces WeakPtrFactory invalidation first.
 140 }
 141
 142 template <DemuxerStream::Type StreamType>
 143 std::string DecoderStream<StreamType>::GetStreamTypeString() {
 144   return DecoderStreamTraits<StreamType>::ToString();
 145 }
 146
 147 template <DemuxerStream::Type StreamType>
 148 void DecoderStream<StreamType>::Initialize(DemuxerStream* stream,
 149                                            InitCB init_cb,
 150                                            CdmContext* cdm_context,
 151                                            StatisticsCB statistics_cb,
 152                                            WaitingCB waiting_cb) {
 153   FUNCTION_DVLOG(1);
 154   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 155   DCHECK_EQ(state_, STATE_UNINITIALIZED);
 156   DCHECK(!init_cb_);
 157   DCHECK(init_cb);
 158
 159   stream_ = stream;
 160   init_cb_ = std::move(init_cb);
 161   cdm_context_ = cdm_context;
 162   statistics_cb_ = std::move(statistics_cb);
 163
 164   // Make a copy here since it's also passed to |decoder_selector_| below.
 165   waiting_cb_ = waiting_cb;
 166
 167   traits_->OnStreamReset(stream_);
 168   decoder_selector_.Initialize(traits_.get(), stream, cdm_context,
 169                                std::move(waiting_cb));
 170
 171   state_ = STATE_INITIALIZING;
 172   BeginDecoderSelection();
 173 }
 174
 175 template <DemuxerStream::Type StreamType>
 176 void DecoderStream<StreamType>::Read(ReadCB read_cb) {
 177   FUNCTION_DVLOG(3);
 178   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 179   DCHECK(state_ != STATE_UNINITIALIZED && state_ != STATE_INITIALIZING)
 180       << state_;
 181   // No two reads in the flight at any time.
 182   DCHECK(!read_cb_);
 183   // No read during resetting or stopping process.
 184   DCHECK(!reset_cb_);
 185
 186   TRACE_EVENT_ASYNC_BEGIN0("media", GetReadTraceString<StreamType>(), this);
 187   if (state_ == STATE_ERROR) {
 188     read_cb_ = base::BindPostTaskToCurrentDefault(std::move(read_cb));
 189     // OnDecodeDone, OnBufferReady, and CompleteDecoderReinitialization all set
 190     // STATE_ERROR and call SatisfyRead, passing the error back to a ReadCB.
 191     SatisfyRead(DecoderStatus::Codes::kDecoderStreamInErrorState);
 192     return;
 193   }
 194
 195   if (state_ == STATE_END_OF_STREAM && ready_outputs_.empty() &&
 196       unprepared_outputs_.empty()) {
 197     read_cb_ = base::BindPostTaskToCurrentDefault(std::move(read_cb));
 198     SatisfyRead(StreamTraits::CreateEOSOutput());
 199     return;
 200   }
 201
 202   if (!ready_outputs_.empty()) {
 203     read_cb_ = base::BindPostTaskToCurrentDefault(std::move(read_cb));
 204     SatisfyRead(ready_outputs_.front());
 205     ready_outputs_.pop_front();
 206     MaybePrepareAnotherOutput();
 207   } else {
 208     read_cb_ = std::move(read_cb);
 209   }
 210
 211   if (state_ == STATE_NORMAL && CanDecodeMore())
 212     ReadFromDemuxerStream();
 213 }
 214
 215 template <DemuxerStream::Type StreamType>
 216 void DecoderStream<StreamType>::Reset(base::OnceClosure closure) {
 217   FUNCTION_DVLOG(2);
 218   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 219   DCHECK_NE(state_, STATE_UNINITIALIZED);
 220   DCHECK(!reset_cb_);
 221
 222   reset_cb_ = std::move(closure);
 223
 224   if (read_cb_) {
 225     read_cb_ = base::BindPostTaskToCurrentDefault(std::move(read_cb_));
 226     SatisfyRead(DecoderStatus::Codes::kAborted);
 227   }
 228
 229   ClearOutputs();
 230   traits_->OnStreamReset(stream_);
 231
 232   // It's possible to have received a DECODE_ERROR and entered STATE_ERROR right
 233   // before a Reset() is executed. If we are still waiting for a demuxer read,
 234   // OnBufferReady() will handle the reset callback.
 235   // See crbug.com/597605 and crbug.com/607454.
 236   if (state_ == STATE_ERROR && !pending_demuxer_read_) {
 237     task_runner_->PostTask(FROM_HERE, std::move(reset_cb_));
 238     return;
 239   }
 240
 241   // During decoder reinitialization, the Decoder does not need to be and
 242   // cannot be Reset(). |decrypting_demuxer_stream_| was reset before decoder
 243   // reinitialization.
 244   if (state_ == STATE_REINITIALIZING_DECODER)
 245     return;
 246
 247   // |decrypting_demuxer_stream_| will fire all of its read requests when
 248   // it resets. |reset_cb_| will be fired in OnDecoderReset(), after the
 249   // decrypting demuxer stream finishes its reset.
 250   if (decrypting_demuxer_stream_) {
 251     decrypting_demuxer_stream_->Reset(base::BindOnce(
 252         &DecoderStream<StreamType>::ResetDecoder, weak_factory_.GetWeakPtr()));
 253     return;
 254   }
 255
 256   // During pending demuxer read and when not using DecryptingDemuxerStream,
 257   // the Decoder will be reset after demuxer read is returned
 258   // (in OnBufferReady()).
 259   if (pending_demuxer_read_)
 260     return;
 261
 262   ResetDecoder();
 263 }
 264
 265 template <DemuxerStream::Type StreamType>
 266 bool DecoderStream<StreamType>::CanReadWithoutStalling() const {
 267   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 268   return !ready_outputs_.empty() || !unprepared_outputs_.empty() ||
 269          (decoder_ && decoder_->CanReadWithoutStalling());
 270 }
 271
 272 template <>
 273 bool DecoderStream<DemuxerStream::AUDIO>::CanReadWithoutStalling() const {
 274   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 275   return true;
 276 }
 277
 278 template <DemuxerStream::Type StreamType>
 279 int DecoderStream<StreamType>::GetMaxDecodeRequests() const {
 280   // The decoder is owned by |decoder_selector_| during reinitialization, so
 281   // during that time we disallow decode requests.
 282   return state_ != STATE_REINITIALIZING_DECODER
 283              ? decoder_->GetMaxDecodeRequests()
 284              : 0;
 285 }
 286
 287 template <>
 288 int DecoderStream<DemuxerStream::AUDIO>::GetMaxDecodeRequests() const {
 289   return 1;
 290 }
 291
 292 template <DemuxerStream::Type StreamType>
 293 int DecoderStream<StreamType>::GetMaxReadyOutputs() const {
 294   // The decoder is owned by |decoder_selector_| during reinitialization, so
 295   // during that time we assume the minimum viable number of max ready outputs.
 296   return state_ != STATE_REINITIALIZING_DECODER
 297              ? decoder_->GetMaxDecodeRequests()
 298              : 1;
 299 }
 300
 301 template <>
 302 int DecoderStream<DemuxerStream::AUDIO>::GetMaxReadyOutputs() const {
 303   return 1;
 304 }
 305
 306 template <DemuxerStream::Type StreamType>
 307 bool DecoderStream<StreamType>::CanDecodeMore() const {
 308   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 309
 310   bool buffers_left = !(fallback_buffers_.empty() && decoding_eos_);
 311
 312   // Limit total number of outputs stored and being decoded. It only makes sense
 313   // to saturate decoder completely when our output queues are empty.
 314   int num_decodes = ready_outputs_.size() + unprepared_outputs_.size() +
 315                     pending_decode_requests_;
 316   return buffers_left && num_decodes < GetMaxDecodeRequests();
 317 }
 318
 319 template <DemuxerStream::Type StreamType>
 320 base::TimeDelta DecoderStream<StreamType>::AverageDuration() const {
 321   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 322   return duration_tracker_.Mean();
 323 }
 324
 325 template <DemuxerStream::Type StreamType>
 326 void DecoderStream<StreamType>::SetPrepareCB(PrepareCB prepare_cb) {
 327   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 328   prepare_cb_ = std::move(prepare_cb);
 329 }
 330
 331 template <DemuxerStream::Type StreamType>
 332 void DecoderStream<StreamType>::SkipPrepareUntil(
 333     base::TimeDelta start_timestamp) {
 334   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 335   skip_prepare_until_timestamp_ = start_timestamp;
 336 }
 337
 338 template <DemuxerStream::Type StreamType>
 339 void DecoderStream<StreamType>::BeginDecoderSelection() {
 340   decoder_selector_.BeginDecoderSelection(
 341       base::BindOnce(&DecoderStream<StreamType>::OnDecoderSelected,
 342                      weak_factory_.GetWeakPtr()),
 343       base::BindRepeating(&DecoderStream<StreamType>::OnDecodeOutputReady,
 344                           fallback_weak_factory_.GetWeakPtr()));
 345 }
 346
 347 template <DemuxerStream::Type StreamType>
 348 void DecoderStream<StreamType>::ResumeDecoderSelection(
 349     DecoderStatus&& reinit_cause) {
 350   decoder_selector_.ResumeDecoderSelection(
 351       base::BindOnce(&DecoderStream<StreamType>::OnDecoderSelected,
 352                      weak_factory_.GetWeakPtr()),
 353       base::BindRepeating(&DecoderStream<StreamType>::OnDecodeOutputReady,
 354                           fallback_weak_factory_.GetWeakPtr()),
 355       std::move(reinit_cause));
 356 }
 357
 358 template <DemuxerStream::Type StreamType>
 359 void DecoderStream<StreamType>::OnDecoderSelected(
 360     DecoderStatus::Or<std::unique_ptr<Decoder>> decoder_or_error,
 361     std::unique_ptr<DecryptingDemuxerStream> decrypting_demuxer_stream) {
 362   FUNCTION_DVLOG(1) << ": "
 363                     << (decoder_or_error.has_value()
 364                             ? GetDecoderName(decoder_or_error->GetDecoderType())
 365                             : "No decoder selected.");
 366   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 367   DCHECK(state_ == STATE_INITIALIZING || state_ == STATE_REINITIALIZING_DECODER)
 368       << state_;
 369
 370   if (state_ == STATE_INITIALIZING) {
 371     DCHECK(init_cb_);
 372     DCHECK(!read_cb_);
 373     DCHECK(!reset_cb_);
 374   }
 375
 376   auto* original_stream = stream_.get();
 377   bool is_decrypting_demuxer_stream_selected = !!decrypting_demuxer_stream;
 378
 379   if (decrypting_demuxer_stream) {
 380     // Override |stream_| with the decrypted stream provided by
 381     // DecryptingDemuxerStream.
 382     decrypting_demuxer_stream_ = std::move(decrypting_demuxer_stream);
 383     stream_ = decrypting_demuxer_stream_.get();
 384
 385     // Also clear |cdm_context_|, it shouldn't be passed during reinitialize for
 386     // a stream that isn't encrypted.
 387     cdm_context_ = nullptr;
 388   }
 389
 390   if (decoder_change_observer_cb_) {
 391     decoder_change_observer_cb_.Run(
 392         decoder_or_error.has_value() ? (*decoder_or_error).get() : nullptr);
 393   }
 394
 395   // TODO(tguilbert): crbug.com/603713 support config changes on decoder reinit.
 396   if (received_config_change_during_reinit_) {
 397     CompleteDecoderReinitialization(DecoderStatus::Codes::kInterrupted);
 398     return;
 399   }
 400
 401   // Attempt to decode buffers from previous decoders (when those decoders have
 402   // never successfully outputed a frame).
 403   fallback_buffers_ = pending_buffers_;
 404
 405   if (!decoder_or_error.has_value()) {
 406     if (state_ == STATE_INITIALIZING) {
 407       state_ = STATE_UNINITIALIZED;
 408       MEDIA_LOG(ERROR, media_log_)
 409           << GetStreamTypeString() << " decoder initialization failed";
 410       std::move(init_cb_).Run(false);
 411       // Node that |decoder_or_error| is not actually lost in this case, as
 412       // DecoderSelector is keeping track of it to use in case there are no
 413       // successfully initialized decoders.
 414     } else {
 415       CompleteDecoderReinitialization(std::move(decoder_or_error).error());
 416     }
 417     return;
 418   }
 419
 420   DCHECK(decoder_or_error.has_value());
 421   decoder_ = std::move(decoder_or_error).value();
 422
 423   // Send logs and statistics updates including the decoder name.
 424   traits_->SetIsPlatformDecoder(decoder_->IsPlatformDecoder());
 425   traits_->SetIsDecryptingDemuxerStream(!!decrypting_demuxer_stream_);
 426   traits_->ReportStatistics(statistics_cb_, 0);
 427
 428   media_log_->SetProperty<StreamTraits::kIsDecryptingDemuxerStream>(
 429       !!decrypting_demuxer_stream_);
 430   media_log_->SetProperty<StreamTraits::kDecoderName>(
 431       decoder_->GetDecoderType());
 432   media_log_->SetProperty<StreamTraits::kIsPlatformDecoder>(
 433       decoder_->IsPlatformDecoder());
 434
 435   if (is_decrypting_demuxer_stream_selected) {
 436     MEDIA_LOG(INFO, media_log_)
 437         << "Selected DecryptingDemuxerStream for " << GetStreamTypeString()
 438         << " decryption, config: "
 439         << traits_->GetDecoderConfig(original_stream).AsHumanReadableString();
 440   }
 441
 442   MEDIA_LOG(INFO, media_log_)
 443       << "Selected " << decoder_->GetDecoderType() << " for "
 444       << GetStreamTypeString() << " decoding, config: "
 445       << traits_->GetDecoderConfig(stream_).AsHumanReadableString();
 446
 447   if (state_ == STATE_REINITIALIZING_DECODER) {
 448     CompleteDecoderReinitialization(OkStatus());
 449     return;
 450   }
 451
 452   // Initialization succeeded.
 453   state_ = STATE_NORMAL;
 454   if (StreamTraits::NeedsBitstreamConversion(decoder_.get()))
 455     stream_->EnableBitstreamConverter();
 456   std::move(init_cb_).Run(true);
 457 }
 458
 459 template <DemuxerStream::Type StreamType>
 460 void DecoderStream<StreamType>::SatisfyRead(ReadResult result) {
 461   DCHECK(read_cb_);
 462   TRACE_EVENT_ASYNC_END1("media", GetReadTraceString<StreamType>(), this,
 463                          "status", GetStatusString(result.code()));
 464   std::move(read_cb_).Run(std::move(result));
 465 }
 466
 467 template <DemuxerStream::Type StreamType>
 468 void DecoderStream<StreamType>::Decode(scoped_refptr<DecoderBuffer> buffer) {
 469   FUNCTION_DVLOG(3);
 470
 471   // We don't know if the decoder will error out on first decode yet. Save the
 472   // buffer to feed it to the fallback decoder later if needed.
 473   if (!decoder_produced_a_frame_) {
 474     pending_buffers_.push_back(buffer);
 475   }
 476
 477   // It's possible for a buffer to arrive from the demuxer right after the
 478   // fallback decoder successfully completed its initialization. At this point
 479   // |pending_buffers_| has already been copied to |fallback_buffers_| and we
 480   // need to append it ourselves.
 481   if (!fallback_buffers_.empty() || fallback_buffers_being_decoded_ > 0) {
 482     fallback_buffers_.push_back(std::exchange(buffer, nullptr));
 483
 484     // There may already be a pending buffer being decoded after decoder
 485     // change. Since decoders can have different max decode requests, we need to
 486     // make sure we can actually decode more buffers here.
 487     if (!CanDecodeMore()) {
 488       return;
 489     }
 490   }
 491
 492   // TODO(https://crbug.com/1324732): We should DCHECK(CanDecodeMore()) here,
 493   // but this breaks a number of tests.
 494
 495   if (!fallback_buffers_.empty()) {
 496     buffer = std::move(fallback_buffers_.front());
 497     fallback_buffers_.pop_front();
 498     ++fallback_buffers_being_decoded_;
 499   }
 500
 501   DecodeInternal(std::move(buffer));
 502 }
 503
 504 template <DemuxerStream::Type StreamType>
 505 void DecoderStream<StreamType>::DecodeInternal(
 506     scoped_refptr<DecoderBuffer> buffer) {
 507   FUNCTION_DVLOG(3);
 508   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER) << state_;
 509   DCHECK_LT(pending_decode_requests_, GetMaxDecodeRequests());
 510   DCHECK(!reset_cb_);
 511   DCHECK(buffer);
 512
 513   std::unique_ptr<ScopedDecodeTrace> trace_event;
 514   if (MediaTraceIsEnabled()) {
 515     // Because multiple Decode() calls may be in flight, each call needs a
 516     // unique trace event class to identify it. This scoped event is bound
 517     // into the OnDecodeDone callback to ensure the trace is always closed.
 518     trace_event = std::make_unique<ScopedDecodeTrace>(
 519         GetDecodeTraceString<StreamType>(), *buffer);
 520   }
 521
 522   traits_->OnDecode(*buffer);
 523
 524   const bool is_eos = buffer->end_of_stream();
 525   if (is_eos)
 526     decoding_eos_ = true;
 527   else if (buffer->duration() != kNoTimestamp)
 528     duration_tracker_.AddSample(buffer->duration());
 529
 530   ++pending_decode_requests_;
 531
 532   const int buffer_size = is_eos ? 0 : buffer->data_size();
 533   decoder_->Decode(
 534       std::move(buffer),
 535       base::BindOnce(&DecoderStream<StreamType>::OnDecodeDone,
 536                      fallback_weak_factory_.GetWeakPtr(), buffer_size,
 537                      decoding_eos_, std::move(trace_event)));
 538 }
 539
 540 template <DemuxerStream::Type StreamType>
 541 void DecoderStream<StreamType>::FlushDecoder() {
 542   // Send the EOS directly to the decoder, bypassing a potential add to
 543   // |pending_buffers_|.
 544   DecodeInternal(DecoderBuffer::CreateEOSBuffer());
 545 }
 546
 547 template <DemuxerStream::Type StreamType>
 548 void DecoderStream<StreamType>::OnDecodeDone(
 549     int buffer_size,
 550     bool end_of_stream,
 551     std::unique_ptr<ScopedDecodeTrace> trace_event,
 552     DecoderStatus status) {
 553   FUNCTION_DVLOG(status.is_ok() ? 3 : 1)
 554       << ": " << static_cast<int>(status.code());
 555   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
 556          state_ == STATE_ERROR)
 557       << state_;
 558   DCHECK_GT(pending_decode_requests_, 0);
 559
 560   --pending_decode_requests_;
 561   if (trace_event)
 562     trace_event->EndTrace(status);
 563
 564   if (end_of_stream) {
 565     DCHECK(!pending_decode_requests_);
 566     decoding_eos_ = false;
 567     if (status.is_ok()) {
 568       // Even if no frames were decoded, completing a flush counts as
 569       // successfully selecting a decoder. This allows back-to-back config
 570       // changes to select from all decoders.
 571       decoder_selector_.FinalizeDecoderSelection();
 572     }
 573   }
 574
 575   if (state_ == STATE_ERROR) {
 576     DCHECK(!read_cb_);
 577     return;
 578   }
 579
 580   // Drop decoding result if Reset() was called during decoding.
 581   // The resetting process will be handled when the decoder is reset.
 582   if (reset_cb_)
 583     return;
 584
 585   switch (status.code()) {
 586     case DecoderStatus::Codes::kAborted:
 587       // Decoder can return kAborted during Reset() or during destruction.
 588       return;
 589
 590     case DecoderStatus::Codes::kOk:
 591       // Any successful decode counts!
 592       if (buffer_size > 0)
 593         traits_->ReportStatistics(statistics_cb_, buffer_size);
 594
 595       if (fallback_buffers_being_decoded_ > 0) {
 596         --fallback_buffers_being_decoded_;
 597       }
 598
 599       if (state_ == STATE_NORMAL) {
 600         if (end_of_stream) {
 601           state_ = STATE_END_OF_STREAM;
 602           if (ready_outputs_.empty() && unprepared_outputs_.empty() && read_cb_)
 603             SatisfyRead(StreamTraits::CreateEOSOutput());
 604           return;
 605         }
 606
 607         if (CanDecodeMore())
 608           ReadFromDemuxerStream();
 609         return;
 610       }
 611
 612       if (state_ == STATE_FLUSHING_DECODER && !pending_decode_requests_)
 613         ReinitializeDecoder();
 614       return;
 615
 616     default:
 617       if (!decoder_produced_a_frame_ &&
 618           base::FeatureList::IsEnabled(kFallbackAfterDecodeError)) {
 619         MEDIA_LOG(WARNING, media_log_)
 620             << GetStreamTypeString()
 621             << " decoder fallback after initial decode error.";
 622
 623         // Prevent all pending decode requests and outputs from those requests
 624         // from being called back.
 625         fallback_weak_factory_.InvalidateWeakPtrs();
 626         pending_decode_requests_ = 0;
 627         decoding_eos_ = false;
 628         state_ = STATE_REINITIALIZING_DECODER;
 629         if (fallback_cb_) {
 630           DecoderStatus copy = status;
 631           PipelineStatus fallback_status = {
 632               PipelineStatus::Codes::PIPELINE_ERROR_DECODE, std::move(copy)};
 633           fallback_cb_.Run(fallback_status);
 634         }
 635         ResumeDecoderSelection(std::move(status));
 636       } else {
 637         media_log_->NotifyError(status);
 638         MEDIA_LOG(ERROR, media_log_)
 639             << GetStreamTypeString() << " decode error!";
 640
 641         state_ = STATE_ERROR;
 642         ClearOutputs();
 643         if (read_cb_)
 644           SatisfyRead(std::move(status));
 645       }
 646       return;
 647   }
 648 }
 649
 650 template <DemuxerStream::Type StreamType>
 651 void DecoderStream<StreamType>::OnDecodeOutputReady(
 652     scoped_refptr<Output> output) {
 653   FUNCTION_DVLOG(3) << ": " << output->timestamp().InMilliseconds() << " ms";
 654   DCHECK(output);
 655   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
 656          state_ == STATE_ERROR)
 657       << state_;
 658
 659   if (state_ == STATE_ERROR) {
 660     DCHECK(!read_cb_);
 661     return;
 662   }
 663
 664   // Drop decoding result if Reset() was called during decoding.
 665   // The resetting process will be handled when the decoder is reset.
 666   if (reset_cb_)
 667     return;
 668
 669   // |decoder_| successfully decoded a frame. No need to keep buffers for a
 670   // fallback decoder.
 671   // Note: |fallback_buffers_| might still have buffers, and we will keep
 672   // reading from there before requesting new buffers from |stream_|.
 673   if (!decoder_produced_a_frame_) {
 674     decoder_produced_a_frame_ = true;
 675     decoder_selector_.FinalizeDecoderSelection();
 676     pending_buffers_.clear();
 677   }
 678
 679   // If the frame should be dropped, exit early and decode another frame.
 680   if (traits_->OnDecodeDone(output.get()) == PostDecodeAction::DROP)
 681     return;
 682
 683   if (prepare_cb_ && output->timestamp() + AverageDuration() >=
 684                          skip_prepare_until_timestamp_) {
 685     unprepared_outputs_.push_back(std::move(output));
 686     MaybePrepareAnotherOutput();
 687     return;
 688   }
 689
 690   traits_->OnOutputReady(output.get());
 691
 692   if (read_cb_) {
 693     // If |ready_outputs_| was non-empty, the read would have already been
 694     // satisifed by Read().
 695     DCHECK(ready_outputs_.empty());
 696     SatisfyRead(std::move(output));
 697     return;
 698   }
 699
 700   // Store decoded output.
 701   ready_outputs_.push_back(std::move(output));
 702 }
 703
 704 template <DemuxerStream::Type StreamType>
 705 void DecoderStream<StreamType>::ReadFromDemuxerStream() {
 706   FUNCTION_DVLOG(3);
 707   DCHECK_EQ(state_, STATE_NORMAL);
 708   DCHECK(CanDecodeMore());
 709   DCHECK(!reset_cb_);
 710
 711   if (!fallback_buffers_.empty()) {
 712     scoped_refptr<DecoderBuffer> buffer = std::move(fallback_buffers_.front());
 713     fallback_buffers_.pop_front();
 714     ++fallback_buffers_being_decoded_;
 715
 716     // Decode the buffer without re-appending it to |pending_buffers_|.
 717     DecodeInternal(std::move(buffer));
 718     return;
 719   }
 720
 721   // We may get here when a read is already pending, ignore this.
 722   if (pending_demuxer_read_)
 723     return;
 724
 725   TRACE_EVENT_ASYNC_BEGIN0("media", GetDemuxerReadTraceString<StreamType>(),
 726                            this);
 727   pending_demuxer_read_ = true;
 728   uint32_t buffer_read_count = 1;
 729   if (base::FeatureList::IsEnabled(kVideoDecodeBatching)) {
 730     buffer_read_count = GetMaxDecodeRequests() - pending_decode_requests_;
 731   }
 732   {
 733     TRACE_EVENT2("media", "DecodeStreamRead",
 734                  "StreamType:", GetStreamTypeString(),
 735                  "buffer_read_count:", buffer_read_count);
 736     stream_->Read(buffer_read_count,
 737                   base::BindOnce(&DecoderStream<StreamType>::OnBuffersReady,
 738                                  weak_factory_.GetWeakPtr()));
 739   }
 740 }
 741
 742 template <DemuxerStream::Type StreamType>
 743 void DecoderStream<StreamType>::OnBuffersReady(
 744     DemuxerStream::Status status,
 745     DemuxerStream::DecoderBufferVector buffers) {
 746   if (status == DemuxerStream::kOk && buffers.empty()) {
 747     MEDIA_LOG(ERROR, media_log_) << "Empty buffer received.";
 748     pending_demuxer_read_ = false;
 749     return;
 750   }
 751
 752   TRACE_EVENT_ASYNC_END1("media", GetDemuxerReadTraceString<StreamType>(), this,
 753                          "status", DemuxerStream::GetStatusName(status));
 754
 755   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 756   DCHECK(pending_demuxer_read_);
 757   if (!decoder_produced_a_frame_) {
 758     DCHECK(state_ == STATE_ERROR || state_ == STATE_REINITIALIZING_DECODER ||
 759            state_ == STATE_NORMAL)
 760         << state_;
 761   }
 762   pending_demuxer_read_ = false;
 763
 764   // If parallel decode requests are supported, multiple read requests might
 765   // have been sent to the demuxer. The buffers might arrive while the decoder
 766   // is reinitializing after falling back on first decode error.
 767   if (state_ == STATE_REINITIALIZING_DECODER && !decoder_produced_a_frame_) {
 768     switch (status) {
 769       case DemuxerStream::kOk:
 770         // Save valid buffers to be consumed by the new decoder.
 771         // |pending_buffers_| is copied to |fallback_buffers_| in
 772         // OnDecoderSelected().
 773         for (auto buffer : buffers) {
 774           pending_buffers_.push_back(std::move(buffer));
 775         }
 776         buffers.clear();
 777         break;
 778       case DemuxerStream::kConfigChanged:
 779         // TODO(tguilbert): crbug.com/603713
 780         // |decoder_| might have a stale config by the time it is reinitialized.
 781         // Ideally, we would save the config from |stream_| and reinitialize the
 782         // decoder by playing back the sequence of buffers and config changes.
 783         received_config_change_during_reinit_ = true;
 784         pending_buffers_.clear();
 785         break;
 786       case DemuxerStream::kAborted:
 787       case DemuxerStream::kError:
 788         // Will read from the demuxer stream again in OnDecoderSelected().
 789         pending_buffers_.clear();
 790         break;
 791     }
 792     return;
 793   }
 794
 795   if (status == DemuxerStream::kError) {
 796     FUNCTION_DVLOG(1) << ": Demuxer stream read error!";
 797     state_ = STATE_ERROR;
 798     MEDIA_LOG(ERROR, media_log_)
 799         << GetStreamTypeString() << " demuxer stream read error!";
 800     pending_buffers_.clear();
 801     ClearOutputs();
 802     // TODO(crbug.com/c/1326324): Convert |status| into a typed status so that
 803     // it can be set as a cause here.
 804     if (read_cb_) {
 805       SatisfyRead(DecoderStatus::Codes::kDecoderStreamDemuxerError);
 806     }
 807   }
 808
 809   // Decoding has been stopped.
 810   if (state_ == STATE_ERROR) {
 811     DCHECK(!read_cb_);
 812
 813     if (reset_cb_) {
 814       // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
 815       // which will continue the resetting process in its callback.
 816       if (!decrypting_demuxer_stream_) {
 817         Reset(std::move(reset_cb_));
 818       }
 819     }
 820     return;
 821   }
 822
 823   state_ = STATE_NORMAL;
 824
 825   if (status == DemuxerStream::kConfigChanged) {
 826     FUNCTION_DVLOG(2) << ": ConfigChanged";
 827     DCHECK(stream_->SupportsConfigChanges());
 828
 829     // Pending buffers might not match the reinitialized decoder's new config.
 830     //
 831     // Note: as part of crbug.com/603713, we should record the config in order
 832     // to play it back to the fallback decoder.
 833     //
 834     // Clearing the buffers is an acceptable workaround for the time being. It
 835     // assures us that we maintain a consistent state, at the cost of
 836     // potentially dropping some frames. Flushing the decoder will cause one of
 837     // the following outcomes:
 838     //   - The decoder outputs a valid frame during flushing (we no longer
 839     //   care about |pending_buffers_| and fallback scenarios).
 840     //   - The decoder returns a DECODE_ERROR via OnDecodeDone() without having
 841     //   outputted a frame (we fallback to a new decoder which will read
 842     //   straight from the demuxer, dropping some frames).
 843     //   - The decoder is flushed without returning a frame or without a
 844     //   DECODE_ERROR (we reinitialize the decoder as if a normal flush
 845     //   happened, and read straight from the demuxer, which could lead to some
 846     //   lost frames if we were to fallback then).
 847     pending_buffers_.clear();
 848
 849     const DecoderConfig& config = traits_->GetDecoderConfig(stream_);
 850
 851     MEDIA_LOG(INFO, media_log_)
 852         << GetStreamTypeString()
 853         << " decoder config changed midstream, new config: "
 854         << config.AsHumanReadableString();
 855
 856     if (config_change_observer_cb_) {
 857       config_change_observer_cb_.Run(config);
 858     }
 859
 860     state_ = STATE_FLUSHING_DECODER;
 861     if (reset_cb_) {
 862       // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
 863       // which will continue the resetting process in its callback.
 864       if (!decrypting_demuxer_stream_) {
 865         Reset(std::move(reset_cb_));
 866       }
 867       // Reinitialization will continue after Reset() is done.
 868     } else {
 869       FlushDecoder();
 870     }
 871     return;
 872   }
 873
 874   if (reset_cb_) {
 875     // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
 876     // which will continue the resetting process in its callback.
 877     if (!decrypting_demuxer_stream_) {
 878       Reset(std::move(reset_cb_));
 879     }
 880     return;
 881   }
 882
 883   if (status == DemuxerStream::kAborted) {
 884     if (read_cb_) {
 885       SatisfyRead(DecoderStatus::Codes::kAborted);
 886     }
 887     return;
 888   }
 889
 890   DCHECK(status == DemuxerStream::kOk) << status;
 891
 892   // Report encryption type of the stream. For simplicity, we only report it
 893   // once on the first buffer of the first config, even if there may be config
 894   // changes later, which is fine for metrics purposes.
 895   if (!encryption_type_reported_) {
 896     encryption_type_reported_ = true;
 897     ReportEncryptionType(buffers[0]);
 898   }
 899
 900   for (auto buffer : buffers) {
 901     Decode(std::move(buffer));
 902   }
 903   buffers.clear();
 904
 905   // Read more data if the decoder supports multiple parallel decoding requests.
 906   if (CanDecodeMore()) {
 907     ReadFromDemuxerStream();
 908   }
 909 }
 910
 911 template <DemuxerStream::Type StreamType>
 912 void DecoderStream<StreamType>::ReinitializeDecoder() {
 913   FUNCTION_DVLOG(2);
 914   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 915   DCHECK_EQ(state_, STATE_FLUSHING_DECODER);
 916   DCHECK_EQ(pending_decode_requests_, 0);
 917
 918   state_ = STATE_REINITIALIZING_DECODER;
 919   decoder_selector_.PrependDecoder(std::move(decoder_));
 920   BeginDecoderSelection();
 921 }
 922
 923 template <DemuxerStream::Type StreamType>
 924 void DecoderStream<StreamType>::CompleteDecoderReinitialization(
 925     DecoderStatus status) {
 926   FUNCTION_DVLOG(2);
 927   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 928   DCHECK_EQ(state_, STATE_REINITIALIZING_DECODER);
 929
 930   state_ = status.is_ok() ? STATE_NORMAL : STATE_ERROR;
 931
 932   // If there's a pending read and no pending reset, report error via
 933   // `read_cb_`, otherwise report it via MediaLog.
 934   if (!status.is_ok() && (reset_cb_ || !read_cb_)) {
 935     media_log_->NotifyError(std::move(status));
 936     MEDIA_LOG(ERROR, media_log_)
 937         << GetStreamTypeString() << " decoder reinitialization failed";
 938   }
 939
 940   if (reset_cb_) {
 941     std::move(reset_cb_).Run();
 942     return;
 943   }
 944
 945   if (!read_cb_)
 946     return;
 947
 948   if (state_ == STATE_ERROR) {
 949     SatisfyRead(std::move(status));
 950     return;
 951   }
 952
 953   // Re-enable fallback to software after reinitialization. This is the last
 954   // place we can clear that state, and as such is the least likely to interfere
 955   // with the rest of the fallback algorithm.
 956   // TODO(tguilbert): investigate setting this flag at an earlier time. This
 957   // could fix the hypothetical edge case of receiving a decode error when
 958   // flushing the decoder during a seek operation.
 959   decoder_produced_a_frame_ = false;
 960
 961   // We may still have too many |ready_outputs_| or |unprepared_outputs_| to
 962   // initiate another read to the demuxer stream. If so, the read will be
 963   // initiated later once we vended enough outputs to read again.
 964   if (CanDecodeMore())
 965     ReadFromDemuxerStream();
 966 }
 967
 968 template <DemuxerStream::Type StreamType>
 969 void DecoderStream<StreamType>::ResetDecoder() {
 970   FUNCTION_DVLOG(2);
 971   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 972   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
 973          state_ == STATE_ERROR || state_ == STATE_END_OF_STREAM)
 974       << state_;
 975   DCHECK(reset_cb_);
 976
 977   decoder_->Reset(base::BindOnce(&DecoderStream<StreamType>::OnDecoderReset,
 978                                  weak_factory_.GetWeakPtr()));
 979 }
 980
 981 template <DemuxerStream::Type StreamType>
 982 void DecoderStream<StreamType>::OnDecoderReset() {
 983   FUNCTION_DVLOG(2);
 984   DCHECK(task_runner_->RunsTasksInCurrentSequence());
 985   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
 986          state_ == STATE_ERROR || state_ == STATE_END_OF_STREAM)
 987       << state_;
 988   // If Reset() was called during pending read, read callback should be fired
 989   // before the reset callback is fired.
 990   DCHECK(!read_cb_);
 991   DCHECK(reset_cb_);
 992
 993   // Make sure we read directly from the demuxer after a reset.
 994   fallback_buffers_.clear();
 995   pending_buffers_.clear();
 996   fallback_buffers_being_decoded_ = 0;
 997
 998   if (state_ != STATE_FLUSHING_DECODER) {
 999     state_ = STATE_NORMAL;
1000     // Pending read, on failure, could have fired the reset callback already.
1001     if (reset_cb_)
1002       std::move(reset_cb_).Run();
1003     return;
1004   }
1005
1006   // The resetting process will be continued in OnDecoderReinitialized().
1007   ReinitializeDecoder();
1008 }
1009
1010 template <DemuxerStream::Type StreamType>
1011 void DecoderStream<StreamType>::ClearOutputs() {
1012   if (preparing_output_)
1013     CompletePrepare(nullptr);
1014   ready_outputs_.clear();
1015   unprepared_outputs_.clear();
1016   prepare_weak_factory_.InvalidateWeakPtrs();
1017 }
1018
1019 template <DemuxerStream::Type StreamType>
1020 void DecoderStream<StreamType>::MaybePrepareAnotherOutput() {
1021   FUNCTION_DVLOG(3);
1022   DCHECK(task_runner_->RunsTasksInCurrentSequence());
1023   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
1024          state_ == STATE_END_OF_STREAM ||
1025          state_ == STATE_REINITIALIZING_DECODER)
1026       << state_;
1027
1028   // If there's nothing to prepare or a prepare is underway, we're done.
1029   if (!prepare_cb_ || unprepared_outputs_.empty() || preparing_output_)
1030     return;
1031
1032   // If there's too many ready outputs, we're done.
1033   if (ready_outputs_.size() >= static_cast<size_t>(GetMaxReadyOutputs()))
1034     return;
1035
1036   // Retain a copy to avoid dangling reference in OnPreparedOutputReady().
1037   const scoped_refptr<Output> output = unprepared_outputs_.front();
1038   TRACE_EVENT_ASYNC_BEGIN1("media", GetPrepareTraceString<StreamType>(), this,
1039                            "timestamp_us",
1040                            output->timestamp().InMicroseconds());
1041   preparing_output_ = true;
1042   prepare_cb_.Run(
1043       output, base::BindOnce(&DecoderStream<StreamType>::OnPreparedOutputReady,
1044                              prepare_weak_factory_.GetWeakPtr()));
1045 }
1046
1047 template <DemuxerStream::Type StreamType>
1048 void DecoderStream<StreamType>::OnPreparedOutputReady(
1049     scoped_refptr<Output> output) {
1050   FUNCTION_DVLOG(2);
1051   DCHECK(task_runner_->RunsTasksInCurrentSequence());
1052
1053   // Errors and reset invalidate the WeakPtr factory for this function, so it
1054   // should only be called in normal and flush states.
1055   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
1056          state_ == STATE_END_OF_STREAM ||
1057          state_ == STATE_REINITIALIZING_DECODER)
1058       << state_;
1059   DCHECK(!reset_cb_);
1060   DCHECK(!unprepared_outputs_.empty());
1061   DCHECK(preparing_output_);
1062
1063   traits_->OnOutputReady(output.get());
1064   CompletePrepare(output.get());
1065   unprepared_outputs_.pop_front();
1066   if (!read_cb_)
1067     ready_outputs_.emplace_back(std::move(output));
1068   else
1069     SatisfyRead(std::move(output));
1070
1071   MaybePrepareAnotherOutput();
1072
1073   // Since decoding may have been stalled by unprepared outputs, we should try
1074   // to queue another decode here if one has been returned.
1075   if (state_ == STATE_NORMAL && CanDecodeMore())
1076     ReadFromDemuxerStream();
1077 }
1078
1079 template <DemuxerStream::Type StreamType>
1080 void DecoderStream<StreamType>::CompletePrepare(const Output* output) {
1081   DCHECK(preparing_output_);
1082   TRACE_EVENT_ASYNC_END1(
1083       "media", GetPrepareTraceString<StreamType>(), this, "timestamp_us",
1084       (output ? output->timestamp() : kNoTimestamp).InMicroseconds());
1085   preparing_output_ = false;
1086 }
1087
1088 template <DemuxerStream::Type StreamType>
1089 void DecoderStream<StreamType>::ReportEncryptionType(
1090     const scoped_refptr<DecoderBuffer>& buffer) {
1091   auto encryption_type = EncryptionType::kClear;
1092   if (decrypting_demuxer_stream_) {
1093     encryption_type = decrypting_demuxer_stream_->HasClearLead()
1094                           ? EncryptionType::kEncryptedWithClearLead
1095                           : EncryptionType::kEncrypted;
1096   } else if (traits_->GetDecoderConfig(stream_).is_encrypted()) {
1097     // Treat EOS as clear buffer which should be rare.
1098     bool is_buffer_encrypted =
1099         !buffer->end_of_stream() && buffer->decrypt_config();
1100     encryption_type = !is_buffer_encrypted
1101                           ? EncryptionType::kEncryptedWithClearLead
1102                           : EncryptionType::kEncrypted;
1103   }
1104
1105   if (encryption_type == EncryptionType::kEncryptedWithClearLead) {
1106     MEDIA_LOG(INFO, media_log_)
1107         << GetStreamTypeString() << "stream is encrypted with clear lead";
1108   }
1109
1110   traits_->SetEncryptionType(encryption_type);
1111   traits_->ReportStatistics(statistics_cb_, 0);
1112 }
1113
1114 template class DecoderStream<DemuxerStream::VIDEO>;
1115 template class DecoderStream<DemuxerStream::AUDIO>;
1116
1117 }  // namespace media