1 // Copyright 2023 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.
5 #include "media/filters/manifest_demuxer.h"
9 #include "base/logging.h"
10 #include "base/memory/scoped_refptr.h"
11 #include "base/task/bind_post_task.h"
12 #include "base/task/sequenced_task_runner.h"
13 #include "base/time/time.h"
14 #include "media/base/container_names.h"
15 #include "media/base/demuxer_stream.h"
16 #include "media/base/media_log.h"
17 #include "media/base/media_track.h"
18 #include "media/base/pipeline_status.h"
19 #include "media/formats/hls/audio_rendition.h"
20 #include "media/formats/hls/media_playlist.h"
21 #include "media/formats/hls/multivariant_playlist.h"
22 #include "media/formats/hls/types.h"
23 #include "media/formats/hls/variant_stream.h"
24 #include "third_party/abseil-cpp/absl/types/optional.h"
28 ManifestDemuxer::ManifestDemuxerStream::~ManifestDemuxerStream() = default;
30 ManifestDemuxer::ManifestDemuxerStream::ManifestDemuxerStream(
31 DemuxerStream* stream,
33 : read_cb_(std::move(cb)), stream_(stream) {}
35 void ManifestDemuxer::ManifestDemuxerStream::Read(uint32_t count,
36 DemuxerStream::ReadCB cb) {
37 stream_->Read(count, base::BindOnce(read_cb_, std::move(cb)));
41 ManifestDemuxer::ManifestDemuxerStream::audio_decoder_config() {
42 return stream_->audio_decoder_config();
46 ManifestDemuxer::ManifestDemuxerStream::video_decoder_config() {
47 return stream_->video_decoder_config();
50 DemuxerStream::Type ManifestDemuxer::ManifestDemuxerStream::type() const {
51 return stream_->type();
54 StreamLiveness ManifestDemuxer::ManifestDemuxerStream::liveness() const {
55 return stream_->liveness();
58 void ManifestDemuxer::ManifestDemuxerStream::EnableBitstreamConverter() {
59 stream_->EnableBitstreamConverter();
62 bool ManifestDemuxer::ManifestDemuxerStream::SupportsConfigChanges() {
63 return stream_->SupportsConfigChanges();
66 ManifestDemuxer::~ManifestDemuxer() {
67 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
70 chunk_demuxer_.reset();
73 ManifestDemuxer::ManifestDemuxer(
74 scoped_refptr<base::SequencedTaskRunner> media_task_runner,
75 base::RepeatingCallback<void(base::TimeDelta)> request_seek,
76 std::unique_ptr<ManifestDemuxer::Engine> impl,
78 : request_seek_(std::move(request_seek)),
79 media_log_(media_log->Clone()),
80 media_task_runner_(std::move(media_task_runner)),
81 impl_(std::move(impl)) {}
83 std::vector<DemuxerStream*> ManifestDemuxer::GetAllStreams() {
84 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
86 // For each stream that ChunkDemuxer returns, we need to wrap it so that we
87 // can grab the timestamp. Chunk demuxer's streams live forever, so ours
88 // might as well also live forever, even if that leaks a small amount of
90 // TODO(crbug/1266991): Rearchitect the demuxer stream ownership model to
91 // prevent long-lived streams from potentially leaking memory.
92 std::vector<DemuxerStream*> streams;
93 for (DemuxerStream* chunk_demuxer_stream : chunk_demuxer_->GetAllStreams()) {
94 auto it = streams_.find(chunk_demuxer_stream);
95 if (it != streams_.end()) {
96 streams.push_back(it->second.get());
99 auto wrapper = std::make_unique<ManifestDemuxerStream>(
100 chunk_demuxer_stream,
101 base::BindRepeating(&ManifestDemuxer::OnDemuxerStreamRead,
102 weak_factory_.GetWeakPtr()));
103 streams.push_back(wrapper.get());
104 streams_[chunk_demuxer_stream] = std::move(wrapper);
109 std::string ManifestDemuxer::GetDisplayName() const {
110 return impl_->GetName();
113 DemuxerType ManifestDemuxer::GetDemuxerType() const {
114 return DemuxerType::kManifestDemuxer;
117 void ManifestDemuxer::Initialize(DemuxerHost* host,
118 PipelineStatusCallback status_cb) {
119 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
120 DCHECK(!pending_init_);
123 pending_init_ = std::move(status_cb);
124 chunk_demuxer_ = std::make_unique<ChunkDemuxer>(
125 base::BindOnce(&ManifestDemuxer::OnChunkDemuxerOpened,
126 weak_factory_.GetWeakPtr()),
127 base::BindRepeating(&ManifestDemuxer::OnProgress,
128 weak_factory_.GetWeakPtr()),
129 base::BindRepeating(&ManifestDemuxer::OnEncryptedMediaData,
130 weak_factory_.GetWeakPtr()),
133 chunk_demuxer_->Initialize(
134 host, base::BindOnce(&ManifestDemuxer::OnChunkDemuxerInitialized,
135 weak_factory_.GetWeakPtr()));
137 impl_->Initialize(this, base::BindOnce(&ManifestDemuxer::OnEngineInitialized,
138 weak_factory_.GetWeakPtr()));
141 void ManifestDemuxer::AbortPendingReads() {
142 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
143 chunk_demuxer_->AbortPendingReads();
144 impl_->AbortPendingReads();
147 void ManifestDemuxer::StartWaitingForSeek(base::TimeDelta seek_time) {
148 if (!media_task_runner_->RunsTasksInCurrentSequence()) {
149 media_task_runner_->PostTask(
150 FROM_HERE, base::BindOnce(&ManifestDemuxer::StartWaitingForSeek,
151 weak_factory_.GetWeakPtr(), seek_time));
154 media_time_ = seek_time;
155 chunk_demuxer_->StartWaitingForSeek(seek_time);
156 impl_->StartWaitingForSeek();
159 void ManifestDemuxer::CancelPendingSeek(base::TimeDelta seek_time) {
160 if (!media_task_runner_->RunsTasksInCurrentSequence()) {
161 media_task_runner_->PostTask(
162 FROM_HERE, base::BindOnce(&ManifestDemuxer::CancelPendingSeek,
163 weak_factory_.GetWeakPtr(), seek_time));
167 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
169 // Since cancellation happens from the main thread, it's possible for a
170 // function order to look like:
171 // Main Thread | Media Thread
172 // --------------------------------------|-----------------------------------
173 // CancelPendingSeek |
174 // | | CompletePendingSeek
175 // `----------------------------|-> CancelPendingSeek
176 // so `pending_seek_` might already be called. If `pending_seek_` is still
177 // pending, then canceling the chunk demuxer pending seek should execute
178 // its callback immediately with a success status, and we'd just then be left
179 // waiting for the engine to finish.
180 // TODO(crbug/1266991): Make the engine cancelable as well.
183 chunk_demuxer_->CancelPendingSeek(seek_time);
187 void ManifestDemuxer::Seek(base::TimeDelta time,
188 PipelineStatusCallback status_cb) {
189 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
190 CHECK(!pending_seek_);
192 pending_seek_ = std::move(status_cb);
195 // Seeks and periodic updates are considered to be events. No two events may
196 // be running at the same time. Seeks still need to happen however, so a seek
197 // should be stored for later.
198 if (has_pending_event_) {
205 void ManifestDemuxer::SeekInternal() {
206 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
208 has_pending_event_ = true;
210 // Cancel any outstanding events, we don't want them interrupting us.
211 cancelable_next_event_.Cancel();
213 impl_->Seek(media_time_, base::BindOnce(&ManifestDemuxer::OnEngineSeeked,
214 weak_factory_.GetWeakPtr()));
217 bool ManifestDemuxer::IsSeekable() const {
218 return impl_->IsSeekable();
221 void ManifestDemuxer::Stop() {
222 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
223 cancelable_next_event_.Cancel();
225 chunk_demuxer_->Stop();
228 base::TimeDelta ManifestDemuxer::GetStartTime() const {
229 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
230 // TODO(crbug/1266991): Support time remapping for streams that start > 0.
231 return base::TimeDelta();
234 base::Time ManifestDemuxer::GetTimelineOffset() const {
235 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
236 // TODO(crbug/1266991): Implement this with the value of the
237 // EXT-X-PROGRAM-DATETIME tag.
238 // TODO(crbug/1266991): Moderate that tag with respect to any underlying
239 // streams' nonzero timeline offsets that the wrapped ChunkDemuxer may have?
240 // And should wrapped ChunkDemuxer's enforcement that any specified (non-null)
241 // offset across multiple ChunkDemuxer::OnSourceInitDone() match be relaxed if
242 // its wrapped by an HLS demuxer which might ignore those offsets?
243 return chunk_demuxer_->GetTimelineOffset();
246 int64_t ManifestDemuxer::GetMemoryUsage() const {
247 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
248 // TODO(crbug/1266991): Consider other potential significant memory usage
249 // here of the player impl.
250 int64_t demuxer_usage = chunk_demuxer_ ? chunk_demuxer_->GetMemoryUsage() : 0;
251 int64_t impl_usage = impl_ ? impl_->GetMemoryUsage() : 0;
252 return demuxer_usage + impl_usage;
255 absl::optional<container_names::MediaContainerName>
256 ManifestDemuxer::GetContainerForMetrics() const {
257 // TODO(crbug/1266991): Consider how this is used. HLS can involve multiple
258 // stream types (mp2t, mp4, etc). Refactor to report something useful.
259 return absl::nullopt;
262 void ManifestDemuxer::OnEnabledAudioTracksChanged(
263 const std::vector<MediaTrack::Id>& track_ids,
264 base::TimeDelta curr_time,
265 TrackChangeCB change_completed_cb) {
266 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
267 chunk_demuxer_->OnEnabledAudioTracksChanged(track_ids, curr_time,
268 std::move(change_completed_cb));
271 void ManifestDemuxer::OnSelectedVideoTrackChanged(
272 const std::vector<MediaTrack::Id>& track_ids,
273 base::TimeDelta curr_time,
274 TrackChangeCB change_completed_cb) {
275 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
276 chunk_demuxer_->OnSelectedVideoTrackChanged(track_ids, curr_time,
277 std::move(change_completed_cb));
280 void ManifestDemuxer::SetPlaybackRate(double rate) {
281 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
282 bool rate_increase = rate > current_playback_rate_;
283 current_playback_rate_ = rate;
284 if (has_pending_event_ || pending_seek_) {
288 if (rate_increase || (rate == 0 && !IsSeekable())) {
289 // If the playback rate increased, or it was a pause of live content,
290 // cancel the next event and set a new one.
291 cancelable_next_event_.Cancel();
296 bool ManifestDemuxer::AddRole(base::StringPiece role,
297 std::string container,
299 CHECK(chunk_demuxer_);
300 if (ChunkDemuxer::kOk !=
301 chunk_demuxer_->AddId(std::string(role), container, codec)) {
304 chunk_demuxer_->SetParseWarningCallback(
306 base::BindRepeating(&ManifestDemuxer::OnChunkDemuxerParseWarning,
307 weak_factory_.GetWeakPtr(), std::string(role)));
308 chunk_demuxer_->SetTracksWatcher(
310 base::BindRepeating(&ManifestDemuxer::OnChunkDemuxerTracksChanged,
311 weak_factory_.GetWeakPtr(), std::string(role)));
315 void ManifestDemuxer::RemoveRole(base::StringPiece role) {
316 chunk_demuxer_->RemoveId(std::string(role));
319 void ManifestDemuxer::SetSequenceMode(base::StringPiece role,
320 bool sequence_mode) {
321 CHECK(chunk_demuxer_);
322 return chunk_demuxer_->SetSequenceMode(std::string(role), sequence_mode);
325 void ManifestDemuxer::SetDuration(double duration) {
326 CHECK(chunk_demuxer_);
327 return chunk_demuxer_->SetDuration(duration);
330 Ranges<base::TimeDelta> ManifestDemuxer::GetBufferedRanges(
331 base::StringPiece role) {
332 CHECK(chunk_demuxer_);
333 return chunk_demuxer_->GetBufferedRanges(std::string(role));
336 void ManifestDemuxer::Remove(base::StringPiece role,
337 base::TimeDelta start,
338 base::TimeDelta end) {
339 chunk_demuxer_->Remove(std::string(role), start, end);
342 void ManifestDemuxer::RemoveAndReset(base::StringPiece role,
343 base::TimeDelta start,
345 base::TimeDelta* offset) {
346 CHECK(chunk_demuxer_);
347 Remove(role, start, end);
348 chunk_demuxer_->ResetParserState(std::string(role), start, end, offset);
349 chunk_demuxer_->AbortPendingReads();
352 void ManifestDemuxer::SetGroupStartIfParsingAndSequenceMode(
353 base::StringPiece role,
354 base::TimeDelta start) {
355 CHECK(chunk_demuxer_);
356 if (!chunk_demuxer_->IsParsingMediaSegment(std::string(role))) {
357 chunk_demuxer_->SetGroupStartTimestampIfInSequenceMode(std::string(role),
362 void ManifestDemuxer::EvictCodedFrames(base::StringPiece role,
363 base::TimeDelta time,
365 CHECK(chunk_demuxer_);
366 if (!chunk_demuxer_->EvictCodedFrames(std::string(role), time, data_size)) {
367 MEDIA_LOG(ERROR, media_log_) << "EvictCodedFrames(" << role << ") failed.";
371 bool ManifestDemuxer::AppendAndParseData(base::StringPiece role,
372 base::TimeDelta start,
374 base::TimeDelta* offset,
377 CHECK(chunk_demuxer_);
378 if (!chunk_demuxer_->AppendToParseBuffer(std::string(role), data,
383 switch (chunk_demuxer_->RunSegmentParserLoop(std::string(role), start, end,
385 case StreamParser::ParseStatus::kSuccess:
387 case StreamParser::ParseStatus::kSuccessHasMoreData:
388 break; // Keep parsing.
395 void ManifestDemuxer::OnError(PipelineStatus error) {
396 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
397 cancelable_next_event_.Cancel();
398 weak_factory_.InvalidateWeakPtrs();
401 std::move(pending_init_).Run(std::move(error).AddHere());
406 std::move(pending_seek_).Run(std::move(error).AddHere());
410 host_->OnDemuxerError(std::move(error).AddHere());
414 void ManifestDemuxer::RequestSeek(base::TimeDelta time) {
415 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
416 request_seek_.Run(time);
419 void ManifestDemuxer::SetGroupStartTimestamp(base::StringPiece role,
420 base::TimeDelta time) {
421 chunk_demuxer_->SetGroupStartTimestampIfInSequenceMode(std::string(role),
425 void ManifestDemuxer::SetEndOfStream() {
426 chunk_demuxer_->MarkEndOfStream(PIPELINE_OK);
429 void ManifestDemuxer::UnsetEndOfStream() {
430 chunk_demuxer_->UnmarkEndOfStream();
433 ChunkDemuxer* ManifestDemuxer::GetChunkDemuxerForTesting() {
434 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
435 return chunk_demuxer_.get();
438 void ManifestDemuxer::OnChunkDemuxerOpened() {
439 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
440 demuxer_opened_ = true;
441 MaybeCompleteInitialize();
444 void ManifestDemuxer::OnProgress() {}
446 void ManifestDemuxer::OnEngineInitialized(PipelineStatus status) {
447 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
448 if (!status.is_ok()) {
449 OnError(std::move(status).AddHere());
452 engine_impl_ready_ = true;
453 MaybeCompleteInitialize();
456 void ManifestDemuxer::MaybeCompleteInitialize() {
457 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
458 if (!demuxer_opened_ || !engine_impl_ready_) {
465 void ManifestDemuxer::TriggerEvent() {
466 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
467 auto queue_next_event = base::BindOnce(
468 &ManifestDemuxer::OnEngineEventFinished, weak_factory_.GetWeakPtr());
469 TriggerEventWithTime(std::move(queue_next_event), media_time_);
472 void ManifestDemuxer::TriggerEventWithTime(DelayCallback cb,
473 base::TimeDelta current_time) {
474 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
475 has_pending_event_ = true;
476 impl_->OnTimeUpdate(current_time, current_playback_rate_,
477 base::BindPostTask(media_task_runner_, std::move(cb)));
480 void ManifestDemuxer::OnEngineEventFinished(base::TimeDelta delay) {
481 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
483 // There should always be an outstanding player event when this method is
484 // called. Player events get set in the Trigger methods, which happen as part
485 // of the time-based player loop, and as part of the seek process.
486 CHECK(has_pending_event_);
487 has_pending_event_ = false;
489 // If there is a pending seek, execute it now. Seeking always calls
490 // |OnEngineEventFinished| when it is finished.
496 if (delay != kNoTimestamp) {
497 // Schedule an event to take place again after a delay.
498 cancelable_next_event_.Reset(base::BindOnce(&ManifestDemuxer::TriggerEvent,
499 weak_factory_.GetWeakPtr()));
500 media_task_runner_->PostDelayedTask(
501 FROM_HERE, cancelable_next_event_.callback(), delay);
505 void ManifestDemuxer::OnChunkDemuxerInitialized(PipelineStatus init_status) {
506 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
508 if (!pending_init_) {
509 OnError(std::move(init_status));
512 std::move(pending_init_).Run(std::move(init_status));
515 void ManifestDemuxer::OnEngineSeeked(SeekResponse seek_status) {
516 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
517 CHECK(pending_seek_);
518 if (!seek_status.has_value()) {
519 std::move(pending_seek_).Run(std::move(seek_status).error().AddHere());
523 chunk_demuxer_->Seek(media_time_,
524 base::BindOnce(&ManifestDemuxer::OnChunkDemuxerSeeked,
525 weak_factory_.GetWeakPtr()));
527 if (std::move(seek_status).value() == SeekState::kNeedsData) {
528 // Buffers need to be refilled, or ChunkDemuxer::Seek will never complete.
529 can_complete_seek_ = false;
530 TriggerEventWithTime(base::BindOnce(&ManifestDemuxer::OnSeekBuffered,
531 weak_factory_.GetWeakPtr()),
536 void ManifestDemuxer::OnSeekBuffered(base::TimeDelta delay_time) {
537 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
539 if (!pending_seek_) {
540 // ChunkDemuxer::Seek replied with an error, and has already reset the flag.
541 CHECK(can_complete_seek_);
545 if (!can_complete_seek_) {
546 // ChunkDemuxer::Seek has not yet replied. Set the flag to true and exit.
547 can_complete_seek_ = true;
551 // Finish seeking and schedule a new event ASAP to continue.
552 std::move(pending_seek_).Run(OkStatus());
553 OnEngineEventFinished(base::Seconds(0));
556 void ManifestDemuxer::OnChunkDemuxerSeeked(PipelineStatus seek_status) {
557 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
558 CHECK(pending_seek_);
559 if (!seek_status.is_ok()) {
560 can_complete_seek_ = true;
561 std::move(pending_seek_).Run(std::move(seek_status));
565 if (!can_complete_seek_) {
566 // The engine should reply shortly after finishing the event which
567 // repopulated ChunkDemuxer's buffers. Reset the flag to allow that reply
568 // to finish the seek process.
569 can_complete_seek_ = true;
573 // Finish seeking and schedule a new event ASAP to continue.
574 std::move(pending_seek_).Run(std::move(seek_status));
575 OnEngineEventFinished(base::Seconds(0));
578 void ManifestDemuxer::OnChunkDemuxerParseWarning(
580 SourceBufferParseWarning warning) {
581 MEDIA_LOG(WARNING, media_log_)
582 << "ParseWarning (" << role << "): " << static_cast<int>(warning);
585 void ManifestDemuxer::OnChunkDemuxerTracksChanged(
587 std::unique_ptr<MediaTracks> tracks) {
588 MEDIA_LOG(WARNING, media_log_) << "TracksChanged for role: " << role;
591 void ManifestDemuxer::OnEncryptedMediaData(EmeInitDataType type,
592 const std::vector<uint8_t>& data) {
593 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
594 // TODO(crbug/1266991): This will be required for iOS support in the future.
598 void ManifestDemuxer::OnDemuxerStreamRead(
599 DemuxerStream::ReadCB wrapped_read_cb,
600 DemuxerStream::Status status,
601 DemuxerStream::DecoderBufferVector buffers) {
602 DCHECK(media_task_runner_->RunsTasksInCurrentSequence());
603 if (status == DemuxerStream::Status::kOk) {
604 // The entire vector must be checked as timestamps are often out of order.
605 for (const auto& buffer : buffers) {
606 if (!buffer->end_of_stream() && buffer->timestamp() > media_time_) {
607 media_time_ = buffer->timestamp();
612 std::move(wrapped_read_cb).Run(status, std::move(buffers));