Home | History | Annotate | Download | only in filters 
      1 // Copyright 2014 The Chromium Authors. All rights reserved.
      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 "base/bind.h"
      8 #include "base/callback_helpers.h"
      9 #include "base/debug/trace_event.h"
     10 #include "base/location.h"
     11 #include "base/logging.h"
     12 #include "base/single_thread_task_runner.h"
     13 #include "media/base/audio_decoder.h"
     14 #include "media/base/bind_to_current_loop.h"
     15 #include "media/base/decoder_buffer.h"
     16 #include "media/base/demuxer_stream.h"
     17 #include "media/base/video_decoder.h"
     18 #include "media/filters/decrypting_demuxer_stream.h"
     19 
     20 namespace media {
     21 
     22 // TODO(rileya): Devise a better way of specifying trace/UMA/etc strings for
     23 // templated classes such as this.
     24 template <DemuxerStream::Type StreamType>
     25 static const char* GetTraceString();
     26 
     27 #define FUNCTION_DVLOG(level) \
     28   DVLOG(level) << __FUNCTION__ << \
     29   "<" << DecoderStreamTraits<StreamType>::ToString() << ">"
     30 
     31 template <>
     32 const char* GetTraceString<DemuxerStream::VIDEO>() {
     33   return "DecoderStream<VIDEO>::Decode";
     34 }
     35 
     36 template <>
     37 const char* GetTraceString<DemuxerStream::AUDIO>() {
     38   return "DecoderStream<AUDIO>::Decode";
     39 }
     40 
     41 template <DemuxerStream::Type StreamType>
     42 DecoderStream<StreamType>::DecoderStream(
     43     const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
     44     ScopedVector<Decoder> decoders,
     45     const SetDecryptorReadyCB& set_decryptor_ready_cb,
     46     const scoped_refptr<MediaLog>& media_log)
     47     : task_runner_(task_runner),
     48       media_log_(media_log),
     49       state_(STATE_UNINITIALIZED),
     50       stream_(NULL),
     51       low_delay_(false),
     52       decoder_selector_(
     53           new DecoderSelector<StreamType>(task_runner,
     54                                           decoders.Pass(),
     55                                           set_decryptor_ready_cb)),
     56       active_splice_(false),
     57       decoding_eos_(false),
     58       pending_decode_requests_(0),
     59       weak_factory_(this) {}
     60 
     61 template <DemuxerStream::Type StreamType>
     62 DecoderStream<StreamType>::~DecoderStream() {
     63   FUNCTION_DVLOG(2);
     64   DCHECK(task_runner_->BelongsToCurrentThread());
     65 
     66   decoder_selector_.reset();
     67 
     68   if (!init_cb_.is_null()) {
     69     task_runner_->PostTask(FROM_HERE,
     70                            base::Bind(base::ResetAndReturn(&init_cb_), false));
     71   }
     72   if (!read_cb_.is_null()) {
     73     task_runner_->PostTask(FROM_HERE, base::Bind(
     74         base::ResetAndReturn(&read_cb_), ABORTED, scoped_refptr<Output>()));
     75   }
     76   if (!reset_cb_.is_null())
     77     task_runner_->PostTask(FROM_HERE, base::ResetAndReturn(&reset_cb_));
     78 
     79   stream_ = NULL;
     80   decoder_.reset();
     81   decrypting_demuxer_stream_.reset();
     82 }
     83 
     84 template <DemuxerStream::Type StreamType>
     85 void DecoderStream<StreamType>::Initialize(DemuxerStream* stream,
     86                                            bool low_delay,
     87                                            const StatisticsCB& statistics_cb,
     88                                            const InitCB& init_cb) {
     89   FUNCTION_DVLOG(2);
     90   DCHECK(task_runner_->BelongsToCurrentThread());
     91   DCHECK_EQ(state_, STATE_UNINITIALIZED) << state_;
     92   DCHECK(init_cb_.is_null());
     93   DCHECK(!init_cb.is_null());
     94 
     95   statistics_cb_ = statistics_cb;
     96   init_cb_ = init_cb;
     97   stream_ = stream;
     98   low_delay_ = low_delay;
     99 
    100   state_ = STATE_INITIALIZING;
    101   // TODO(xhwang): DecoderSelector only needs a config to select a decoder.
    102   decoder_selector_->SelectDecoder(
    103       stream, low_delay,
    104       base::Bind(&DecoderStream<StreamType>::OnDecoderSelected,
    105                  weak_factory_.GetWeakPtr()),
    106       base::Bind(&DecoderStream<StreamType>::OnDecodeOutputReady,
    107                  weak_factory_.GetWeakPtr()));
    108 }
    109 
    110 template <DemuxerStream::Type StreamType>
    111 void DecoderStream<StreamType>::Read(const ReadCB& read_cb) {
    112   FUNCTION_DVLOG(2);
    113   DCHECK(task_runner_->BelongsToCurrentThread());
    114   DCHECK(state_ != STATE_UNINITIALIZED && state_ != STATE_INITIALIZING)
    115       << state_;
    116   // No two reads in the flight at any time.
    117   DCHECK(read_cb_.is_null());
    118   // No read during resetting or stopping process.
    119   DCHECK(reset_cb_.is_null());
    120 
    121   if (state_ == STATE_ERROR) {
    122     task_runner_->PostTask(
    123         FROM_HERE, base::Bind(read_cb, DECODE_ERROR, scoped_refptr<Output>()));
    124     return;
    125   }
    126 
    127   if (state_ == STATE_END_OF_STREAM && ready_outputs_.empty()) {
    128     task_runner_->PostTask(
    129         FROM_HERE, base::Bind(read_cb, OK, StreamTraits::CreateEOSOutput()));
    130     return;
    131   }
    132 
    133   if (!ready_outputs_.empty()) {
    134     task_runner_->PostTask(FROM_HERE,
    135                            base::Bind(read_cb, OK, ready_outputs_.front()));
    136     ready_outputs_.pop_front();
    137   } else {
    138     read_cb_ = read_cb;
    139   }
    140 
    141   if (state_ == STATE_NORMAL && CanDecodeMore())
    142     ReadFromDemuxerStream();
    143 }
    144 
    145 template <DemuxerStream::Type StreamType>
    146 void DecoderStream<StreamType>::Reset(const base::Closure& closure) {
    147   FUNCTION_DVLOG(2);
    148   DCHECK(task_runner_->BelongsToCurrentThread());
    149   DCHECK(state_ != STATE_UNINITIALIZED)<< state_;
    150   DCHECK(reset_cb_.is_null());
    151 
    152   reset_cb_ = closure;
    153 
    154   if (!read_cb_.is_null()) {
    155     task_runner_->PostTask(FROM_HERE, base::Bind(
    156         base::ResetAndReturn(&read_cb_), ABORTED, scoped_refptr<Output>()));
    157   }
    158 
    159   ready_outputs_.clear();
    160 
    161   // During decoder reinitialization, the Decoder does not need to be and
    162   // cannot be Reset(). |decrypting_demuxer_stream_| was reset before decoder
    163   // reinitialization.
    164   if (state_ == STATE_REINITIALIZING_DECODER)
    165     return;
    166 
    167   // During pending demuxer read and when not using DecryptingDemuxerStream,
    168   // the Decoder will be reset after demuxer read is returned
    169   // (in OnBufferReady()).
    170   if (state_ == STATE_PENDING_DEMUXER_READ && !decrypting_demuxer_stream_)
    171     return;
    172 
    173   if (decrypting_demuxer_stream_) {
    174     decrypting_demuxer_stream_->Reset(base::Bind(
    175         &DecoderStream<StreamType>::ResetDecoder, weak_factory_.GetWeakPtr()));
    176     return;
    177   }
    178 
    179   ResetDecoder();
    180 }
    181 
    182 template <DemuxerStream::Type StreamType>
    183 bool DecoderStream<StreamType>::CanReadWithoutStalling() const {
    184   DCHECK(task_runner_->BelongsToCurrentThread());
    185   return !ready_outputs_.empty() || decoder_->CanReadWithoutStalling();
    186 }
    187 
    188 template <>
    189 bool DecoderStream<DemuxerStream::AUDIO>::CanReadWithoutStalling() const {
    190   DCHECK(task_runner_->BelongsToCurrentThread());
    191   return true;
    192 }
    193 
    194 template <DemuxerStream::Type StreamType>
    195 int DecoderStream<StreamType>::GetMaxDecodeRequests() const {
    196   return decoder_->GetMaxDecodeRequests();
    197 }
    198 
    199 template <>
    200 int DecoderStream<DemuxerStream::AUDIO>::GetMaxDecodeRequests() const {
    201   return 1;
    202 }
    203 
    204 template <DemuxerStream::Type StreamType>
    205 bool DecoderStream<StreamType>::CanDecodeMore() const {
    206   DCHECK(task_runner_->BelongsToCurrentThread());
    207 
    208   // Limit total number of outputs stored in |ready_outputs_| and being decoded.
    209   // It only makes sense to saturate decoder completely when output queue is
    210   // empty.
    211   int num_decodes =
    212       static_cast<int>(ready_outputs_.size()) + pending_decode_requests_;
    213   return !decoding_eos_ && num_decodes < GetMaxDecodeRequests();
    214 }
    215 
    216 template <DemuxerStream::Type StreamType>
    217 void DecoderStream<StreamType>::OnDecoderSelected(
    218     scoped_ptr<Decoder> selected_decoder,
    219     scoped_ptr<DecryptingDemuxerStream> decrypting_demuxer_stream) {
    220   FUNCTION_DVLOG(2);
    221   DCHECK(task_runner_->BelongsToCurrentThread());
    222   DCHECK_EQ(state_, STATE_INITIALIZING) << state_;
    223   DCHECK(!init_cb_.is_null());
    224   DCHECK(read_cb_.is_null());
    225   DCHECK(reset_cb_.is_null());
    226 
    227   decoder_selector_.reset();
    228   if (decrypting_demuxer_stream)
    229     stream_ = decrypting_demuxer_stream.get();
    230 
    231   if (!selected_decoder) {
    232     state_ = STATE_UNINITIALIZED;
    233     base::ResetAndReturn(&init_cb_).Run(false);
    234     return;
    235   }
    236 
    237   state_ = STATE_NORMAL;
    238   decoder_ = selected_decoder.Pass();
    239   decrypting_demuxer_stream_ = decrypting_demuxer_stream.Pass();
    240 
    241   const std::string stream_type = DecoderStreamTraits<StreamType>::ToString();
    242   media_log_->SetBooleanProperty((stream_type + "_dds").c_str(),
    243                                  decrypting_demuxer_stream_);
    244   media_log_->SetStringProperty((stream_type + "_decoder").c_str(),
    245                                 decoder_->GetDisplayName());
    246 
    247   if (StreamTraits::NeedsBitstreamConversion(decoder_.get()))
    248     stream_->EnableBitstreamConverter();
    249   base::ResetAndReturn(&init_cb_).Run(true);
    250 }
    251 
    252 template <DemuxerStream::Type StreamType>
    253 void DecoderStream<StreamType>::SatisfyRead(
    254     Status status,
    255     const scoped_refptr<Output>& output) {
    256   DCHECK(!read_cb_.is_null());
    257   base::ResetAndReturn(&read_cb_).Run(status, output);
    258 }
    259 
    260 template <DemuxerStream::Type StreamType>
    261 void DecoderStream<StreamType>::Decode(
    262     const scoped_refptr<DecoderBuffer>& buffer) {
    263   FUNCTION_DVLOG(2);
    264   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER) << state_;
    265   DCHECK_LT(pending_decode_requests_, GetMaxDecodeRequests());
    266   DCHECK(reset_cb_.is_null());
    267   DCHECK(buffer.get());
    268 
    269   int buffer_size = buffer->end_of_stream() ? 0 : buffer->data_size();
    270 
    271   TRACE_EVENT_ASYNC_BEGIN0("media", GetTraceString<StreamType>(), this);
    272 
    273   if (buffer->end_of_stream())
    274     decoding_eos_ = true;
    275 
    276   ++pending_decode_requests_;
    277   decoder_->Decode(buffer,
    278                    base::Bind(&DecoderStream<StreamType>::OnDecodeDone,
    279                               weak_factory_.GetWeakPtr(),
    280                               buffer_size,
    281                               buffer->end_of_stream()));
    282 }
    283 
    284 template <DemuxerStream::Type StreamType>
    285 void DecoderStream<StreamType>::FlushDecoder() {
    286   Decode(DecoderBuffer::CreateEOSBuffer());
    287 }
    288 
    289 template <DemuxerStream::Type StreamType>
    290 void DecoderStream<StreamType>::OnDecodeDone(int buffer_size,
    291                                              bool end_of_stream,
    292                                              typename Decoder::Status status) {
    293   FUNCTION_DVLOG(2) << status;
    294   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
    295          state_ == STATE_PENDING_DEMUXER_READ || state_ == STATE_ERROR)
    296       << state_;
    297   DCHECK_GT(pending_decode_requests_, 0);
    298 
    299   --pending_decode_requests_;
    300 
    301   TRACE_EVENT_ASYNC_END0("media", GetTraceString<StreamType>(), this);
    302 
    303   if (end_of_stream)
    304     decoding_eos_ = false;
    305 
    306   if (state_ == STATE_ERROR) {
    307     DCHECK(read_cb_.is_null());
    308     return;
    309   }
    310 
    311   // Drop decoding result if Reset() was called during decoding.
    312   // The resetting process will be handled when the decoder is reset.
    313   if (!reset_cb_.is_null())
    314     return;
    315 
    316   switch (status) {
    317     case Decoder::kDecodeError:
    318     case Decoder::kDecryptError:
    319       state_ = STATE_ERROR;
    320       ready_outputs_.clear();
    321       if (!read_cb_.is_null())
    322         SatisfyRead(DECODE_ERROR, NULL);
    323       return;
    324 
    325     case Decoder::kAborted:
    326       // Decoder can return kAborted only when Reset is pending.
    327       NOTREACHED();
    328       return;
    329 
    330     case Decoder::kOk:
    331       // Any successful decode counts!
    332       if (buffer_size > 0)
    333         StreamTraits::ReportStatistics(statistics_cb_, buffer_size);
    334 
    335       if (state_ == STATE_NORMAL) {
    336         if (end_of_stream) {
    337           state_ = STATE_END_OF_STREAM;
    338           if (ready_outputs_.empty() && !read_cb_.is_null())
    339             SatisfyRead(OK, StreamTraits::CreateEOSOutput());
    340           return;
    341         }
    342 
    343         if (CanDecodeMore())
    344           ReadFromDemuxerStream();
    345         return;
    346       }
    347 
    348       if (state_ == STATE_FLUSHING_DECODER && !pending_decode_requests_)
    349         ReinitializeDecoder();
    350       return;
    351   }
    352 }
    353 
    354 template <DemuxerStream::Type StreamType>
    355 void DecoderStream<StreamType>::OnDecodeOutputReady(
    356     const scoped_refptr<Output>& output) {
    357   FUNCTION_DVLOG(2) << ": " << output->timestamp().InMilliseconds() << " ms";
    358   DCHECK(output.get());
    359   DCHECK(!output->end_of_stream());
    360   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
    361          state_ == STATE_PENDING_DEMUXER_READ || state_ == STATE_ERROR)
    362       << state_;
    363 
    364   if (state_ == STATE_ERROR) {
    365     DCHECK(read_cb_.is_null());
    366     return;
    367   }
    368 
    369   // Drop decoding result if Reset() was called during decoding.
    370   // The resetting process will be handled when the decoder is reset.
    371   if (!reset_cb_.is_null())
    372     return;
    373 
    374   // TODO(xhwang): VideoDecoder doesn't need to return EOS after it's flushed.
    375   // Fix all decoders and remove this block.
    376   // Store decoded output.
    377   ready_outputs_.push_back(output);
    378 
    379   if (read_cb_.is_null())
    380     return;
    381 
    382   // Satisfy outstanding read request, if any.
    383   scoped_refptr<Output> read_result = ready_outputs_.front();
    384   ready_outputs_.pop_front();
    385   SatisfyRead(OK, output);
    386 }
    387 
    388 template <DemuxerStream::Type StreamType>
    389 void DecoderStream<StreamType>::ReadFromDemuxerStream() {
    390   FUNCTION_DVLOG(2);
    391   DCHECK_EQ(state_, STATE_NORMAL) << state_;
    392   DCHECK(CanDecodeMore());
    393   DCHECK(reset_cb_.is_null());
    394 
    395   state_ = STATE_PENDING_DEMUXER_READ;
    396   stream_->Read(base::Bind(&DecoderStream<StreamType>::OnBufferReady,
    397                            weak_factory_.GetWeakPtr()));
    398 }
    399 
    400 template <DemuxerStream::Type StreamType>
    401 void DecoderStream<StreamType>::OnBufferReady(
    402     DemuxerStream::Status status,
    403     const scoped_refptr<DecoderBuffer>& buffer) {
    404   FUNCTION_DVLOG(2) << ": " << status << ", "
    405                     << (buffer.get() ? buffer->AsHumanReadableString()
    406                                      : "NULL");
    407 
    408   DCHECK(task_runner_->BelongsToCurrentThread());
    409   DCHECK(state_ == STATE_PENDING_DEMUXER_READ || state_ == STATE_ERROR)
    410       << state_;
    411   DCHECK_EQ(buffer.get() != NULL, status == DemuxerStream::kOk) << status;
    412 
    413   // Decoding has been stopped (e.g due to an error).
    414   if (state_ != STATE_PENDING_DEMUXER_READ) {
    415     DCHECK(state_ == STATE_ERROR);
    416     DCHECK(read_cb_.is_null());
    417     return;
    418   }
    419 
    420   state_ = STATE_NORMAL;
    421 
    422   if (status == DemuxerStream::kConfigChanged) {
    423     FUNCTION_DVLOG(2) << ": " << "ConfigChanged";
    424     DCHECK(stream_->SupportsConfigChanges());
    425 
    426     if (!config_change_observer_cb_.is_null())
    427       config_change_observer_cb_.Run();
    428 
    429     state_ = STATE_FLUSHING_DECODER;
    430     if (!reset_cb_.is_null()) {
    431       // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
    432       // which will continue the resetting process in it's callback.
    433       if (!decrypting_demuxer_stream_)
    434         Reset(base::ResetAndReturn(&reset_cb_));
    435       // Reinitialization will continue after Reset() is done.
    436     } else {
    437       FlushDecoder();
    438     }
    439     return;
    440   }
    441 
    442   if (!reset_cb_.is_null()) {
    443     // If we are using DecryptingDemuxerStream, we already called DDS::Reset()
    444     // which will continue the resetting process in it's callback.
    445     if (!decrypting_demuxer_stream_)
    446       Reset(base::ResetAndReturn(&reset_cb_));
    447     return;
    448   }
    449 
    450   if (status == DemuxerStream::kAborted) {
    451     if (!read_cb_.is_null())
    452       SatisfyRead(DEMUXER_READ_ABORTED, NULL);
    453     return;
    454   }
    455 
    456   if (!splice_observer_cb_.is_null() && !buffer->end_of_stream()) {
    457     const bool has_splice_ts = buffer->splice_timestamp() != kNoTimestamp();
    458     if (active_splice_ || has_splice_ts) {
    459       splice_observer_cb_.Run(buffer->splice_timestamp());
    460       active_splice_ = has_splice_ts;
    461     }
    462   }
    463 
    464   DCHECK(status == DemuxerStream::kOk) << status;
    465   Decode(buffer);
    466 
    467   // Read more data if the decoder supports multiple parallel decoding requests.
    468   if (CanDecodeMore())
    469     ReadFromDemuxerStream();
    470 }
    471 
    472 template <DemuxerStream::Type StreamType>
    473 void DecoderStream<StreamType>::ReinitializeDecoder() {
    474   FUNCTION_DVLOG(2);
    475   DCHECK(task_runner_->BelongsToCurrentThread());
    476   DCHECK_EQ(state_, STATE_FLUSHING_DECODER) << state_;
    477   DCHECK_EQ(pending_decode_requests_, 0);
    478 
    479   DCHECK(StreamTraits::GetDecoderConfig(*stream_).IsValidConfig());
    480   state_ = STATE_REINITIALIZING_DECODER;
    481   DecoderStreamTraits<StreamType>::Initialize(
    482       decoder_.get(),
    483       StreamTraits::GetDecoderConfig(*stream_),
    484       low_delay_,
    485       base::Bind(&DecoderStream<StreamType>::OnDecoderReinitialized,
    486                  weak_factory_.GetWeakPtr()),
    487       base::Bind(&DecoderStream<StreamType>::OnDecodeOutputReady,
    488                  weak_factory_.GetWeakPtr()));
    489 }
    490 
    491 template <DemuxerStream::Type StreamType>
    492 void DecoderStream<StreamType>::OnDecoderReinitialized(PipelineStatus status) {
    493   FUNCTION_DVLOG(2);
    494   DCHECK(task_runner_->BelongsToCurrentThread());
    495   DCHECK_EQ(state_, STATE_REINITIALIZING_DECODER) << state_;
    496 
    497   // ReinitializeDecoder() can be called in two cases:
    498   // 1, Flushing decoder finished (see OnDecodeOutputReady()).
    499   // 2, Reset() was called during flushing decoder (see OnDecoderReset()).
    500   // Also, Reset() can be called during pending ReinitializeDecoder().
    501   // This function needs to handle them all!
    502 
    503   state_ = (status == PIPELINE_OK) ? STATE_NORMAL : STATE_ERROR;
    504 
    505   if (!reset_cb_.is_null()) {
    506     base::ResetAndReturn(&reset_cb_).Run();
    507     return;
    508   }
    509 
    510   if (read_cb_.is_null())
    511     return;
    512 
    513   if (state_ == STATE_ERROR) {
    514     SatisfyRead(DECODE_ERROR, NULL);
    515     return;
    516   }
    517 
    518   ReadFromDemuxerStream();
    519 }
    520 
    521 template <DemuxerStream::Type StreamType>
    522 void DecoderStream<StreamType>::ResetDecoder() {
    523   FUNCTION_DVLOG(2);
    524   DCHECK(task_runner_->BelongsToCurrentThread());
    525   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
    526          state_ == STATE_ERROR || state_ == STATE_END_OF_STREAM) << state_;
    527   DCHECK(!reset_cb_.is_null());
    528 
    529   decoder_->Reset(base::Bind(&DecoderStream<StreamType>::OnDecoderReset,
    530                              weak_factory_.GetWeakPtr()));
    531 }
    532 
    533 template <DemuxerStream::Type StreamType>
    534 void DecoderStream<StreamType>::OnDecoderReset() {
    535   FUNCTION_DVLOG(2);
    536   DCHECK(task_runner_->BelongsToCurrentThread());
    537   DCHECK(state_ == STATE_NORMAL || state_ == STATE_FLUSHING_DECODER ||
    538          state_ == STATE_ERROR || state_ == STATE_END_OF_STREAM) << state_;
    539   // If Reset() was called during pending read, read callback should be fired
    540   // before the reset callback is fired.
    541   DCHECK(read_cb_.is_null());
    542   DCHECK(!reset_cb_.is_null());
    543 
    544   if (state_ != STATE_FLUSHING_DECODER) {
    545     state_ = STATE_NORMAL;
    546     active_splice_ = false;
    547     base::ResetAndReturn(&reset_cb_).Run();
    548     return;
    549   }
    550 
    551   // The resetting process will be continued in OnDecoderReinitialized().
    552   ReinitializeDecoder();
    553 }
    554 
    555 template class DecoderStream<DemuxerStream::VIDEO>;
    556 template class DecoderStream<DemuxerStream::AUDIO>;
    557 
    558 }  // namespace media
    559