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