gecko-dev/dom/media/MediaStreamGraph.cpp
Andreas Pehrson 71cacec3d5 Bug 1410829 - Signal listeners removed on shutdown so they can clean up. r=padenot
When shutting down we shut modules down in the order of
[media, gfx, cycleCollector].

At the same time we rely on destructors to clean up resources for MediaStreams
and MediaStreamTracks, but these objects may be held until cycleCollector
shutdown. Gfx resources are not allowed to be released after gfx shutdown, which
is where we this approach hits a wall.

This patch will signal them through the three available listener types to clean
up during media shutdown.

MozReview-Commit-ID: FwsG3ukV29P

--HG--
extra : rebase_source : 554ec29a43b7551b3b5570577b0559285e36d4fd
2017-10-26 16:13:38 +02:00

4219 lines
135 KiB
C++

/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*-*/
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this file,
* You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "MediaStreamGraphImpl.h"
#include "mozilla/MathAlgorithms.h"
#include "mozilla/Unused.h"
#include "AudioSegment.h"
#include "VideoSegment.h"
#include "nsContentUtils.h"
#include "nsIObserver.h"
#include "nsPrintfCString.h"
#include "nsServiceManagerUtils.h"
#include "prerror.h"
#include "mozilla/Logging.h"
#include "mozilla/Attributes.h"
#include "TrackUnionStream.h"
#include "ImageContainer.h"
#include "AudioCaptureStream.h"
#include "AudioNodeStream.h"
#include "AudioNodeExternalInputStream.h"
#include "MediaStreamListener.h"
#include "MediaStreamVideoSink.h"
#include "mozilla/dom/BaseAudioContextBinding.h"
#include "mozilla/media/MediaUtils.h"
#include <algorithm>
#include "GeckoProfiler.h"
#include "VideoFrameContainer.h"
#include "mozilla/AbstractThread.h"
#include "mozilla/Unused.h"
#ifdef MOZ_WEBRTC
#include "AudioOutputObserver.h"
#endif
#include "mtransport/runnable_utils.h"
#include "webaudio/blink/DenormalDisabler.h"
#include "webaudio/blink/HRTFDatabaseLoader.h"
using namespace mozilla::layers;
using namespace mozilla::dom;
using namespace mozilla::gfx;
using namespace mozilla::media;
namespace mozilla {
LazyLogModule gMediaStreamGraphLog("MediaStreamGraph");
#ifdef LOG
#undef LOG
#endif // LOG
#define LOG(type, msg) MOZ_LOG(gMediaStreamGraphLog, type, msg)
enum SourceMediaStream::TrackCommands : uint32_t {
TRACK_CREATE = TrackEventCommand::TRACK_EVENT_CREATED,
TRACK_END = TrackEventCommand::TRACK_EVENT_ENDED,
TRACK_UNUSED = TrackEventCommand::TRACK_EVENT_UNUSED,
};
/**
* A hash table containing the graph instances, one per document.
*/
static nsDataHashtable<nsUint32HashKey, MediaStreamGraphImpl*> gGraphs;
MediaStreamGraphImpl::~MediaStreamGraphImpl()
{
NS_ASSERTION(IsEmpty(),
"All streams should have been destroyed by messages from the main thread");
LOG(LogLevel::Debug, ("MediaStreamGraph %p destroyed", this));
LOG(LogLevel::Debug, ("MediaStreamGraphImpl::~MediaStreamGraphImpl"));
}
void
MediaStreamGraphImpl::FinishStream(MediaStream* aStream)
{
if (aStream->mFinished)
return;
LOG(LogLevel::Debug, ("MediaStream %p will finish", aStream));
#ifdef DEBUG
for (StreamTracks::TrackIter track(aStream->mTracks);
!track.IsEnded(); track.Next()) {
if (!track->IsEnded()) {
LOG(LogLevel::Error,
("MediaStream %p will finish, but track %d has not ended.",
aStream,
track->GetID()));
NS_ASSERTION(false, "Finished stream cannot contain live track");
}
}
#endif
aStream->mFinished = true;
aStream->mTracks.AdvanceKnownTracksTime(STREAM_TIME_MAX);
SetStreamOrderDirty();
}
void
MediaStreamGraphImpl::AddStreamGraphThread(MediaStream* aStream)
{
aStream->mTracksStartTime = mProcessedTime;
if (aStream->AsSourceStream()) {
SourceMediaStream* source = aStream->AsSourceStream();
TimeStamp currentTimeStamp = CurrentDriver()->GetCurrentTimeStamp();
TimeStamp processedTimeStamp = currentTimeStamp +
TimeDuration::FromSeconds(MediaTimeToSeconds(mProcessedTime - IterationEnd()));
source->SetStreamTracksStartTimeStamp(processedTimeStamp);
}
if (aStream->IsSuspended()) {
mSuspendedStreams.AppendElement(aStream);
LOG(LogLevel::Debug,
("Adding media stream %p to the graph, in the suspended stream array",
aStream));
} else {
mStreams.AppendElement(aStream);
LOG(LogLevel::Debug,
("Adding media stream %p to graph %p, count %zu",
aStream,
this,
mStreams.Length()));
LOG(LogLevel::Debug,
("Adding media stream %p to graph %p, count %zu",
aStream,
this,
mStreams.Length()));
}
SetStreamOrderDirty();
}
void
MediaStreamGraphImpl::RemoveStreamGraphThread(MediaStream* aStream)
{
// Remove references in mStreamUpdates before we allow aStream to die.
// Pending updates are not needed (since the main thread has already given
// up the stream) so we will just drop them.
{
MonitorAutoLock lock(mMonitor);
for (uint32_t i = 0; i < mStreamUpdates.Length(); ++i) {
if (mStreamUpdates[i].mStream == aStream) {
mStreamUpdates[i].mStream = nullptr;
}
}
}
// Ensure that mFirstCycleBreaker and mMixer are updated when necessary.
SetStreamOrderDirty();
if (aStream->IsSuspended()) {
mSuspendedStreams.RemoveElement(aStream);
} else {
mStreams.RemoveElement(aStream);
}
LOG(LogLevel::Debug,
("Removed media stream %p from graph %p, count %zu",
aStream,
this,
mStreams.Length()));
LOG(LogLevel::Debug,
("Removed media stream %p from graph %p, count %zu",
aStream,
this,
mStreams.Length()));
NS_RELEASE(aStream); // probably destroying it
}
void
MediaStreamGraphImpl::ExtractPendingInput(SourceMediaStream* aStream,
GraphTime aDesiredUpToTime,
bool* aEnsureNextIteration)
{
bool finished;
{
MutexAutoLock lock(aStream->mMutex);
if (aStream->mPullEnabled && !aStream->mFinished &&
!aStream->mListeners.IsEmpty()) {
// Compute how much stream time we'll need assuming we don't block
// the stream at all.
StreamTime t = aStream->GraphTimeToStreamTime(aDesiredUpToTime);
LOG(LogLevel::Verbose,
("Calling NotifyPull aStream=%p t=%f current end=%f",
aStream,
MediaTimeToSeconds(t),
MediaTimeToSeconds(aStream->mTracks.GetEnd())));
if (t > aStream->mTracks.GetEnd()) {
*aEnsureNextIteration = true;
#ifdef DEBUG
if (aStream->mListeners.Length() == 0) {
LOG(
LogLevel::Error,
("No listeners in NotifyPull aStream=%p desired=%f current end=%f",
aStream,
MediaTimeToSeconds(t),
MediaTimeToSeconds(aStream->mTracks.GetEnd())));
aStream->DumpTrackInfo();
}
#endif
for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
MediaStreamListener* l = aStream->mListeners[j];
{
MutexAutoUnlock unlock(aStream->mMutex);
l->NotifyPull(this, t);
}
}
}
}
finished = aStream->mUpdateFinished;
bool shouldNotifyTrackCreated = false;
for (int32_t i = aStream->mUpdateTracks.Length() - 1; i >= 0; --i) {
SourceMediaStream::TrackData* data = &aStream->mUpdateTracks[i];
aStream->ApplyTrackDisabling(data->mID, data->mData);
// Dealing with NotifyQueuedTrackChanges and NotifyQueuedAudioData part.
// The logic is different from the manipulating of aStream->mTracks part.
// So it is not combined with the manipulating of aStream->mTracks part.
StreamTime offset =
(data->mCommands & SourceMediaStream::TRACK_CREATE)
? data->mStart
: aStream->mTracks.FindTrack(data->mID)->GetSegment()->GetDuration();
// Audio case.
if (data->mData->GetType() == MediaSegment::AUDIO) {
if (data->mCommands) {
MOZ_ASSERT(!(data->mCommands & SourceMediaStream::TRACK_UNUSED));
for (MediaStreamListener* l : aStream->mListeners) {
if (data->mCommands & SourceMediaStream::TRACK_END) {
l->NotifyQueuedAudioData(this, data->mID,
offset, *(static_cast<AudioSegment*>(data->mData.get())));
}
l->NotifyQueuedTrackChanges(this, data->mID,
offset, static_cast<TrackEventCommand>(data->mCommands), *data->mData);
if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
l->NotifyQueuedAudioData(this, data->mID,
offset, *(static_cast<AudioSegment*>(data->mData.get())));
}
}
} else {
for (MediaStreamListener* l : aStream->mListeners) {
l->NotifyQueuedAudioData(this, data->mID,
offset, *(static_cast<AudioSegment*>(data->mData.get())));
}
}
}
// Video case.
if (data->mData->GetType() == MediaSegment::VIDEO) {
if (data->mCommands) {
MOZ_ASSERT(!(data->mCommands & SourceMediaStream::TRACK_UNUSED));
for (MediaStreamListener* l : aStream->mListeners) {
l->NotifyQueuedTrackChanges(this, data->mID,
offset, static_cast<TrackEventCommand>(data->mCommands), *data->mData);
}
}
}
for (TrackBound<MediaStreamTrackListener>& b : aStream->mTrackListeners) {
if (b.mTrackID != data->mID) {
continue;
}
b.mListener->NotifyQueuedChanges(this, offset, *data->mData);
if (data->mCommands & SourceMediaStream::TRACK_END) {
b.mListener->NotifyEnded();
}
}
if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
MediaSegment* segment = data->mData.forget();
LOG(LogLevel::Debug,
("SourceMediaStream %p creating track %d, start %" PRId64
", initial end %" PRId64,
aStream,
data->mID,
int64_t(data->mStart),
int64_t(segment->GetDuration())));
data->mEndOfFlushedData += segment->GetDuration();
aStream->mTracks.AddTrack(data->mID, data->mStart, segment);
// The track has taken ownership of data->mData, so let's replace
// data->mData with an empty clone.
data->mData = segment->CreateEmptyClone();
data->mCommands &= ~SourceMediaStream::TRACK_CREATE;
shouldNotifyTrackCreated = true;
} else if (data->mData->GetDuration() > 0) {
MediaSegment* dest = aStream->mTracks.FindTrack(data->mID)->GetSegment();
LOG(LogLevel::Verbose,
("SourceMediaStream %p track %d, advancing end from %" PRId64
" to %" PRId64,
aStream,
data->mID,
int64_t(dest->GetDuration()),
int64_t(dest->GetDuration() + data->mData->GetDuration())));
data->mEndOfFlushedData += data->mData->GetDuration();
dest->AppendFrom(data->mData);
}
if (data->mCommands & SourceMediaStream::TRACK_END) {
aStream->mTracks.FindTrack(data->mID)->SetEnded();
aStream->mUpdateTracks.RemoveElementAt(i);
}
}
if (shouldNotifyTrackCreated) {
for (MediaStreamListener* l : aStream->mListeners) {
l->NotifyFinishedTrackCreation(this);
}
}
if (!aStream->mFinished) {
aStream->mTracks.AdvanceKnownTracksTime(aStream->mUpdateKnownTracksTime);
}
}
if (aStream->mTracks.GetEnd() > 0) {
aStream->mHasCurrentData = true;
}
if (finished) {
FinishStream(aStream);
}
}
StreamTime
MediaStreamGraphImpl::GraphTimeToStreamTimeWithBlocking(MediaStream* aStream,
GraphTime aTime)
{
MOZ_ASSERT(aTime <= mStateComputedTime,
"Don't ask about times where we haven't made blocking decisions yet");
return std::max<StreamTime>(0,
std::min(aTime, aStream->mStartBlocking) - aStream->mTracksStartTime);
}
GraphTime
MediaStreamGraphImpl::IterationEnd() const
{
return CurrentDriver()->IterationEnd();
}
void
MediaStreamGraphImpl::UpdateCurrentTimeForStreams(GraphTime aPrevCurrentTime)
{
for (MediaStream* stream : AllStreams()) {
bool isAnyBlocked = stream->mStartBlocking < mStateComputedTime;
bool isAnyUnblocked = stream->mStartBlocking > aPrevCurrentTime;
// Calculate blocked time and fire Blocked/Unblocked events
GraphTime blockedTime = mStateComputedTime - stream->mStartBlocking;
NS_ASSERTION(blockedTime >= 0, "Error in blocking time");
stream->AdvanceTimeVaryingValuesToCurrentTime(mStateComputedTime,
blockedTime);
LOG(LogLevel::Verbose,
("MediaStream %p bufferStartTime=%f blockedTime=%f",
stream,
MediaTimeToSeconds(stream->mTracksStartTime),
MediaTimeToSeconds(blockedTime)));
stream->mStartBlocking = mStateComputedTime;
if (isAnyUnblocked && stream->mNotifiedBlocked) {
for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
MediaStreamListener* l = stream->mListeners[j];
l->NotifyBlockingChanged(this, MediaStreamListener::UNBLOCKED);
}
stream->mNotifiedBlocked = false;
}
if (isAnyBlocked && !stream->mNotifiedBlocked) {
for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
MediaStreamListener* l = stream->mListeners[j];
l->NotifyBlockingChanged(this, MediaStreamListener::BLOCKED);
}
stream->mNotifiedBlocked = true;
}
if (isAnyUnblocked) {
NS_ASSERTION(!stream->mNotifiedFinished,
"Shouldn't have already notified of finish *and* have output!");
for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
MediaStreamListener* l = stream->mListeners[j];
l->NotifyOutput(this, mProcessedTime);
}
}
// The stream is fully finished when all of its track data has been played
// out.
if (stream->mFinished && !stream->mNotifiedFinished &&
mProcessedTime >=
stream->StreamTimeToGraphTime(stream->GetStreamTracks().GetAllTracksEnd())) {
stream->mNotifiedFinished = true;
SetStreamOrderDirty();
for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
MediaStreamListener* l = stream->mListeners[j];
l->NotifyEvent(this, MediaStreamGraphEvent::EVENT_FINISHED);
}
}
}
}
template<typename C, typename Chunk>
void
MediaStreamGraphImpl::ProcessChunkMetadataForInterval(MediaStream* aStream,
TrackID aTrackID,
C& aSegment,
StreamTime aStart,
StreamTime aEnd)
{
MOZ_ASSERT(aStream);
MOZ_ASSERT(IsTrackIDExplicit(aTrackID));
StreamTime offset = 0;
for (typename C::ConstChunkIterator chunk(aSegment);
!chunk.IsEnded(); chunk.Next()) {
if (offset >= aEnd) {
break;
}
offset += chunk->GetDuration();
if (chunk->IsNull() || offset < aStart) {
continue;
}
PrincipalHandle principalHandle = chunk->GetPrincipalHandle();
if (principalHandle != aSegment.GetLastPrincipalHandle()) {
aSegment.SetLastPrincipalHandle(principalHandle);
LOG(LogLevel::Debug,
("MediaStream %p track %d, principalHandle "
"changed in %sChunk with duration %lld",
aStream,
aTrackID,
aSegment.GetType() == MediaSegment::AUDIO ? "Audio" : "Video",
(long long)chunk->GetDuration()));
for (const TrackBound<MediaStreamTrackListener>& listener :
aStream->mTrackListeners) {
if (listener.mTrackID == aTrackID) {
listener.mListener->NotifyPrincipalHandleChanged(this, principalHandle);
}
}
}
}
}
void
MediaStreamGraphImpl::ProcessChunkMetadata(GraphTime aPrevCurrentTime)
{
for (MediaStream* stream : AllStreams()) {
StreamTime iterationStart = stream->GraphTimeToStreamTime(aPrevCurrentTime);
StreamTime iterationEnd = stream->GraphTimeToStreamTime(mProcessedTime);
for (StreamTracks::TrackIter tracks(stream->mTracks);
!tracks.IsEnded(); tracks.Next()) {
MediaSegment* segment = tracks->GetSegment();
if (!segment) {
continue;
}
if (tracks->GetType() == MediaSegment::AUDIO) {
AudioSegment* audio = static_cast<AudioSegment*>(segment);
ProcessChunkMetadataForInterval<AudioSegment, AudioChunk>(
stream, tracks->GetID(), *audio, iterationStart, iterationEnd);
} else if (tracks->GetType() == MediaSegment::VIDEO) {
VideoSegment* video = static_cast<VideoSegment*>(segment);
ProcessChunkMetadataForInterval<VideoSegment, VideoChunk>(
stream, tracks->GetID(), *video, iterationStart, iterationEnd);
} else {
MOZ_CRASH("Unknown track type");
}
}
}
}
GraphTime
MediaStreamGraphImpl::WillUnderrun(MediaStream* aStream,
GraphTime aEndBlockingDecisions)
{
// Finished streams can't underrun. ProcessedMediaStreams also can't cause
// underrun currently, since we'll always be able to produce data for them
// unless they block on some other stream.
if (aStream->mFinished || aStream->AsProcessedStream()) {
return aEndBlockingDecisions;
}
// This stream isn't finished or suspended. We don't need to call
// StreamTimeToGraphTime since an underrun is the only thing that can block
// it.
GraphTime bufferEnd = aStream->GetTracksEnd() + aStream->mTracksStartTime;
#ifdef DEBUG
if (bufferEnd < mProcessedTime) {
LOG(LogLevel::Error,
("MediaStream %p underrun, "
"bufferEnd %f < mProcessedTime %f (%" PRId64 " < %" PRId64
"), Streamtime %" PRId64,
aStream,
MediaTimeToSeconds(bufferEnd),
MediaTimeToSeconds(mProcessedTime),
bufferEnd,
mProcessedTime,
aStream->GetTracksEnd()));
aStream->DumpTrackInfo();
NS_ASSERTION(bufferEnd >= mProcessedTime, "Buffer underran");
}
#endif
return std::min(bufferEnd, aEndBlockingDecisions);
}
namespace {
// Value of mCycleMarker for unvisited streams in cycle detection.
const uint32_t NOT_VISITED = UINT32_MAX;
// Value of mCycleMarker for ordered streams in muted cycles.
const uint32_t IN_MUTED_CYCLE = 1;
} // namespace
bool
MediaStreamGraphImpl::AudioTrackPresent(bool& aNeedsAEC)
{
AssertOnGraphThreadOrNotRunning();
bool audioTrackPresent = false;
for (uint32_t i = 0; i < mStreams.Length() && audioTrackPresent == false; ++i) {
MediaStream* stream = mStreams[i];
SourceMediaStream* source = stream->AsSourceStream();
#ifdef MOZ_WEBRTC
if (source && source->NeedsMixing()) {
aNeedsAEC = true;
}
#endif
// If this is a AudioNodeStream, force a AudioCallbackDriver.
if (stream->AsAudioNodeStream()) {
audioTrackPresent = true;
} else {
for (StreamTracks::TrackIter tracks(stream->GetStreamTracks(), MediaSegment::AUDIO);
!tracks.IsEnded(); tracks.Next()) {
audioTrackPresent = true;
}
}
if (source) {
audioTrackPresent = source->HasPendingAudioTrack();
}
}
// XXX For some reason, there are race conditions when starting an audio input where
// we find no active audio tracks. In any case, if we have an active audio input we
// should not allow a switch back to a SystemClockDriver
if (!audioTrackPresent && mInputDeviceUsers.Count() != 0) {
NS_WARNING("No audio tracks, but full-duplex audio is enabled!!!!!");
audioTrackPresent = true;
#ifdef MOZ_WEBRTC
aNeedsAEC = true;
#endif
}
return audioTrackPresent;
}
void
MediaStreamGraphImpl::UpdateStreamOrder()
{
bool shouldAEC = false;
bool audioTrackPresent = AudioTrackPresent(shouldAEC);
// Note that this looks for any audio streams, input or output, and switches to a
// SystemClockDriver if there are none. However, if another is already pending, let that
// switch happen.
if (!audioTrackPresent && mRealtime &&
CurrentDriver()->AsAudioCallbackDriver()) {
MonitorAutoLock mon(mMonitor);
if (CurrentDriver()->AsAudioCallbackDriver()->IsStarted() &&
!(CurrentDriver()->Switching())) {
if (mLifecycleState == LIFECYCLE_RUNNING) {
SystemClockDriver* driver = new SystemClockDriver(this);
CurrentDriver()->SwitchAtNextIteration(driver);
}
}
}
bool switching = false;
{
MonitorAutoLock mon(mMonitor);
switching = CurrentDriver()->Switching();
}
if (audioTrackPresent && mRealtime &&
!CurrentDriver()->AsAudioCallbackDriver() &&
!switching) {
MonitorAutoLock mon(mMonitor);
if (mLifecycleState == LIFECYCLE_RUNNING) {
AudioCallbackDriver* driver = new AudioCallbackDriver(this);
CurrentDriver()->SwitchAtNextIteration(driver);
}
}
if (!mStreamOrderDirty) {
return;
}
mStreamOrderDirty = false;
// The algorithm for finding cycles is based on Tim Leslie's iterative
// implementation [1][2] of Pearce's variant [3] of Tarjan's strongly
// connected components (SCC) algorithm. There are variations (a) to
// distinguish whether streams in SCCs of size 1 are in a cycle and (b) to
// re-run the algorithm over SCCs with breaks at DelayNodes.
//
// [1] http://www.timl.id.au/?p=327
// [2] https://github.com/scipy/scipy/blob/e2c502fca/scipy/sparse/csgraph/_traversal.pyx#L582
// [3] http://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.102.1707
//
// There are two stacks. One for the depth-first search (DFS),
mozilla::LinkedList<MediaStream> dfsStack;
// and another for streams popped from the DFS stack, but still being
// considered as part of SCCs involving streams on the stack.
mozilla::LinkedList<MediaStream> sccStack;
// An index into mStreams for the next stream found with no unsatisfied
// upstream dependencies.
uint32_t orderedStreamCount = 0;
for (uint32_t i = 0; i < mStreams.Length(); ++i) {
MediaStream* s = mStreams[i];
ProcessedMediaStream* ps = s->AsProcessedStream();
if (ps) {
// The dfsStack initially contains a list of all processed streams in
// unchanged order.
dfsStack.insertBack(s);
ps->mCycleMarker = NOT_VISITED;
} else {
// SourceMediaStreams have no inputs and so can be ordered now.
mStreams[orderedStreamCount] = s;
++orderedStreamCount;
}
}
// mNextStackMarker corresponds to "index" in Tarjan's algorithm. It is a
// counter to label mCycleMarker on the next visited stream in the DFS
// uniquely in the set of visited streams that are still being considered.
//
// In this implementation, the counter descends so that the values are
// strictly greater than the values that mCycleMarker takes when the stream
// has been ordered (0 or IN_MUTED_CYCLE).
//
// Each new stream labelled, as the DFS searches upstream, receives a value
// less than those used for all other streams being considered.
uint32_t nextStackMarker = NOT_VISITED - 1;
// Reset list of DelayNodes in cycles stored at the tail of mStreams.
mFirstCycleBreaker = mStreams.Length();
// Rearrange dfsStack order as required to DFS upstream and pop streams
// in processing order to place in mStreams.
while (auto ps = static_cast<ProcessedMediaStream*>(dfsStack.getFirst())) {
const auto& inputs = ps->mInputs;
MOZ_ASSERT(ps->AsProcessedStream());
if (ps->mCycleMarker == NOT_VISITED) {
// Record the position on the visited stack, so that any searches
// finding this stream again know how much of the stack is in the cycle.
ps->mCycleMarker = nextStackMarker;
--nextStackMarker;
// Not-visited input streams should be processed first.
// SourceMediaStreams have already been ordered.
for (uint32_t i = inputs.Length(); i--; ) {
if (inputs[i]->mSource->IsSuspended()) {
continue;
}
auto input = inputs[i]->mSource->AsProcessedStream();
if (input && input->mCycleMarker == NOT_VISITED) {
// It can be that this stream has an input which is from a suspended
// AudioContext.
if (input->isInList()) {
input->remove();
dfsStack.insertFront(input);
}
}
}
continue;
}
// Returning from DFS. Pop from dfsStack.
ps->remove();
// cycleStackMarker keeps track of the highest marker value on any
// upstream stream, if any, found receiving input, directly or indirectly,
// from the visited stack (and so from |ps|, making a cycle). In a
// variation from Tarjan's SCC algorithm, this does not include |ps|
// unless it is part of the cycle.
uint32_t cycleStackMarker = 0;
for (uint32_t i = inputs.Length(); i--; ) {
if (inputs[i]->mSource->IsSuspended()) {
continue;
}
auto input = inputs[i]->mSource->AsProcessedStream();
if (input) {
cycleStackMarker = std::max(cycleStackMarker, input->mCycleMarker);
}
}
if (cycleStackMarker <= IN_MUTED_CYCLE) {
// All inputs have been ordered and their stack markers have been removed.
// This stream is not part of a cycle. It can be processed next.
ps->mCycleMarker = 0;
mStreams[orderedStreamCount] = ps;
++orderedStreamCount;
continue;
}
// A cycle has been found. Record this stream for ordering when all
// streams in this SCC have been popped from the DFS stack.
sccStack.insertFront(ps);
if (cycleStackMarker > ps->mCycleMarker) {
// Cycles have been found that involve streams that remain on the stack.
// Leave mCycleMarker indicating the most downstream (last) stream on
// the stack known to be part of this SCC. In this way, any searches on
// other paths that find |ps| will know (without having to traverse from
// this stream again) that they are part of this SCC (i.e. part of an
// intersecting cycle).
ps->mCycleMarker = cycleStackMarker;
continue;
}
// |ps| is the root of an SCC involving no other streams on dfsStack, the
// complete SCC has been recorded, and streams in this SCC are part of at
// least one cycle.
MOZ_ASSERT(cycleStackMarker == ps->mCycleMarker);
// If there are DelayNodes in this SCC, then they may break the cycles.
bool haveDelayNode = false;
auto next = sccStack.getFirst();
// Streams in this SCC are identified by mCycleMarker <= cycleStackMarker.
// (There may be other streams later in sccStack from other incompletely
// searched SCCs, involving streams still on dfsStack.)
//
// DelayNodes in cycles must behave differently from those not in cycles,
// so all DelayNodes in the SCC must be identified.
while (next && static_cast<ProcessedMediaStream*>(next)->
mCycleMarker <= cycleStackMarker) {
auto ns = next->AsAudioNodeStream();
// Get next before perhaps removing from list below.
next = next->getNext();
if (ns && ns->Engine()->AsDelayNodeEngine()) {
haveDelayNode = true;
// DelayNodes break cycles by producing their output in a
// preprocessing phase; they do not need to be ordered before their
// consumers. Order them at the tail of mStreams so that they can be
// handled specially. Do so now, so that DFS ignores them.
ns->remove();
ns->mCycleMarker = 0;
--mFirstCycleBreaker;
mStreams[mFirstCycleBreaker] = ns;
}
}
auto after_scc = next;
while ((next = sccStack.getFirst()) != after_scc) {
next->remove();
auto removed = static_cast<ProcessedMediaStream*>(next);
if (haveDelayNode) {
// Return streams to the DFS stack again (to order and detect cycles
// without delayNodes). Any of these streams that are still inputs
// for streams on the visited stack must be returned to the front of
// the stack to be ordered before their dependents. We know that none
// of these streams need input from streams on the visited stack, so
// they can all be searched and ordered before the current stack head
// is popped.
removed->mCycleMarker = NOT_VISITED;
dfsStack.insertFront(removed);
} else {
// Streams in cycles without any DelayNodes must be muted, and so do
// not need input and can be ordered now. They must be ordered before
// their consumers so that their muted output is available.
removed->mCycleMarker = IN_MUTED_CYCLE;
mStreams[orderedStreamCount] = removed;
++orderedStreamCount;
}
}
}
MOZ_ASSERT(orderedStreamCount == mFirstCycleBreaker);
}
void
MediaStreamGraphImpl::NotifyHasCurrentData(MediaStream* aStream)
{
if (!aStream->mNotifiedHasCurrentData && aStream->mHasCurrentData) {
for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
MediaStreamListener* l = aStream->mListeners[j];
l->NotifyHasCurrentData(this);
}
aStream->mNotifiedHasCurrentData = true;
}
}
void
MediaStreamGraphImpl::CreateOrDestroyAudioStreams(MediaStream* aStream)
{
MOZ_ASSERT(mRealtime, "Should only attempt to create audio streams in real-time mode");
if (aStream->mAudioOutputs.IsEmpty()) {
aStream->mAudioOutputStreams.Clear();
return;
}
if (!aStream->GetStreamTracks().GetAndResetTracksDirty() &&
!aStream->mAudioOutputStreams.IsEmpty()) {
return;
}
LOG(LogLevel::Debug,
("Updating AudioOutputStreams for MediaStream %p", aStream));
AutoTArray<bool,2> audioOutputStreamsFound;
for (uint32_t i = 0; i < aStream->mAudioOutputStreams.Length(); ++i) {
audioOutputStreamsFound.AppendElement(false);
}
for (StreamTracks::TrackIter tracks(aStream->GetStreamTracks(), MediaSegment::AUDIO);
!tracks.IsEnded(); tracks.Next()) {
uint32_t i;
for (i = 0; i < audioOutputStreamsFound.Length(); ++i) {
if (aStream->mAudioOutputStreams[i].mTrackID == tracks->GetID()) {
break;
}
}
if (i < audioOutputStreamsFound.Length()) {
audioOutputStreamsFound[i] = true;
} else {
MediaStream::AudioOutputStream* audioOutputStream =
aStream->mAudioOutputStreams.AppendElement();
audioOutputStream->mAudioPlaybackStartTime = mProcessedTime;
audioOutputStream->mBlockedAudioTime = 0;
audioOutputStream->mLastTickWritten = 0;
audioOutputStream->mTrackID = tracks->GetID();
bool switching = false;
{
MonitorAutoLock lock(mMonitor);
switching = CurrentDriver()->Switching();
}
if (!CurrentDriver()->AsAudioCallbackDriver() &&
!switching) {
MonitorAutoLock mon(mMonitor);
if (mLifecycleState == LIFECYCLE_RUNNING) {
AudioCallbackDriver* driver = new AudioCallbackDriver(this);
CurrentDriver()->SwitchAtNextIteration(driver);
}
}
}
}
for (int32_t i = audioOutputStreamsFound.Length() - 1; i >= 0; --i) {
if (!audioOutputStreamsFound[i]) {
aStream->mAudioOutputStreams.RemoveElementAt(i);
}
}
}
StreamTime
MediaStreamGraphImpl::PlayAudio(MediaStream* aStream)
{
MOZ_ASSERT(mRealtime, "Should only attempt to play audio in realtime mode");
float volume = 0.0f;
for (uint32_t i = 0; i < aStream->mAudioOutputs.Length(); ++i) {
volume += aStream->mAudioOutputs[i].mVolume;
}
StreamTime ticksWritten = 0;
for (uint32_t i = 0; i < aStream->mAudioOutputStreams.Length(); ++i) {
ticksWritten = 0;
MediaStream::AudioOutputStream& audioOutput = aStream->mAudioOutputStreams[i];
StreamTracks::Track* track = aStream->mTracks.FindTrack(audioOutput.mTrackID);
AudioSegment* audio = track->Get<AudioSegment>();
AudioSegment output;
StreamTime offset = aStream->GraphTimeToStreamTime(mProcessedTime);
// We don't update aStream->mTracksStartTime here to account for time spent
// blocked. Instead, we'll update it in UpdateCurrentTimeForStreams after
// the blocked period has completed. But we do need to make sure we play
// from the right offsets in the stream buffer, even if we've already
// written silence for some amount of blocked time after the current time.
GraphTime t = mProcessedTime;
while (t < mStateComputedTime) {
bool blocked = t >= aStream->mStartBlocking;
GraphTime end = blocked ? mStateComputedTime : aStream->mStartBlocking;
NS_ASSERTION(end <= mStateComputedTime, "mStartBlocking is wrong!");
// Check how many ticks of sound we can provide if we are blocked some
// time in the middle of this cycle.
StreamTime toWrite = end - t;
if (blocked) {
output.InsertNullDataAtStart(toWrite);
ticksWritten += toWrite;
LOG(LogLevel::Verbose,
("MediaStream %p writing %" PRId64 " blocking-silence samples for "
"%f to %f (%" PRId64 " to %" PRId64 ")",
aStream,
toWrite,
MediaTimeToSeconds(t),
MediaTimeToSeconds(end),
offset,
offset + toWrite));
} else {
StreamTime endTicksNeeded = offset + toWrite;
StreamTime endTicksAvailable = audio->GetDuration();
if (endTicksNeeded <= endTicksAvailable) {
LOG(LogLevel::Verbose,
("MediaStream %p writing %" PRId64 " samples for %f to %f "
"(samples %" PRId64 " to %" PRId64 ")",
aStream,
toWrite,
MediaTimeToSeconds(t),
MediaTimeToSeconds(end),
offset,
endTicksNeeded));
output.AppendSlice(*audio, offset, endTicksNeeded);
ticksWritten += toWrite;
offset = endTicksNeeded;
} else {
// MOZ_ASSERT(track->IsEnded(), "Not enough data, and track not ended.");
// If we are at the end of the track, maybe write the remaining
// samples, and pad with/output silence.
if (endTicksNeeded > endTicksAvailable &&
offset < endTicksAvailable) {
output.AppendSlice(*audio, offset, endTicksAvailable);
LOG(LogLevel::Verbose,
("MediaStream %p writing %" PRId64 " samples for %f to %f "
"(samples %" PRId64 " to %" PRId64 ")",
aStream,
toWrite,
MediaTimeToSeconds(t),
MediaTimeToSeconds(end),
offset,
endTicksNeeded));
uint32_t available = endTicksAvailable - offset;
ticksWritten += available;
toWrite -= available;
offset = endTicksAvailable;
}
output.AppendNullData(toWrite);
LOG(LogLevel::Verbose,
("MediaStream %p writing %" PRId64 " padding slsamples for %f to "
"%f (samples %" PRId64 " to %" PRId64 ")",
aStream,
toWrite,
MediaTimeToSeconds(t),
MediaTimeToSeconds(end),
offset,
endTicksNeeded));
ticksWritten += toWrite;
}
output.ApplyVolume(volume);
}
t = end;
}
audioOutput.mLastTickWritten = offset;
// Need unique id for stream & track - and we want it to match the inserter
output.WriteTo(LATENCY_STREAM_ID(aStream, track->GetID()),
mMixer,
CurrentDriver()->AsAudioCallbackDriver()->OutputChannelCount(),
mSampleRate);
}
return ticksWritten;
}
void
MediaStreamGraphImpl::OpenAudioInputImpl(int aID,
AudioDataListener *aListener)
{
// Bug 1238038 Need support for multiple mics at once
if (mInputDeviceUsers.Count() > 0 &&
!mInputDeviceUsers.Get(aListener, nullptr)) {
NS_ASSERTION(false, "Input from multiple mics not yet supported; bug 1238038");
// Need to support separate input-only AudioCallback drivers; they'll
// call us back on "other" threads. We will need to echo-cancel them, though.
return;
}
mInputWanted = true;
// Add to count of users for this ID.
// XXX Since we can't rely on IDs staying valid (ugh), use the listener as
// a stand-in for the ID. Fix as part of support for multiple-captures
// (Bug 1238038)
uint32_t count = 0;
mInputDeviceUsers.Get(aListener, &count); // ok if this fails
count++;
mInputDeviceUsers.Put(aListener, count); // creates a new entry in the hash if needed
if (count == 1) { // first open for this listener
// aID is a cubeb_devid, and we assume that opaque ptr is valid until
// we close cubeb.
mInputDeviceID = aID;
mAudioInputs.AppendElement(aListener); // always monitor speaker data
// Switch Drivers since we're adding input (to input-only or full-duplex)
MonitorAutoLock mon(mMonitor);
if (mLifecycleState == LIFECYCLE_RUNNING) {
AudioCallbackDriver* driver = new AudioCallbackDriver(this);
driver->SetMicrophoneActive(true);
LOG(
LogLevel::Debug,
("OpenAudioInput: starting new AudioCallbackDriver(input) %p", driver));
LOG(
LogLevel::Debug,
("OpenAudioInput: starting new AudioCallbackDriver(input) %p", driver));
driver->SetInputListener(aListener);
CurrentDriver()->SwitchAtNextIteration(driver);
} else {
LOG(LogLevel::Error, ("OpenAudioInput in shutdown!"));
LOG(LogLevel::Debug, ("OpenAudioInput in shutdown!"));
NS_ASSERTION(false, "Can't open cubeb inputs in shutdown");
}
}
}
nsresult
MediaStreamGraphImpl::OpenAudioInput(int aID,
AudioDataListener *aListener)
{
// So, so, so annoying. Can't AppendMessage except on Mainthread
if (!NS_IsMainThread()) {
RefPtr<nsIRunnable> runnable =
WrapRunnable(this,
&MediaStreamGraphImpl::OpenAudioInput,
aID,
RefPtr<AudioDataListener>(aListener));
mAbstractMainThread->Dispatch(runnable.forget());
return NS_OK;
}
class Message : public ControlMessage {
public:
Message(MediaStreamGraphImpl *aGraph, int aID,
AudioDataListener *aListener) :
ControlMessage(nullptr), mGraph(aGraph), mID(aID), mListener(aListener) {}
void Run() override
{
mGraph->OpenAudioInputImpl(mID, mListener);
}
MediaStreamGraphImpl *mGraph;
int mID;
RefPtr<AudioDataListener> mListener;
};
// XXX Check not destroyed!
this->AppendMessage(MakeUnique<Message>(this, aID, aListener));
return NS_OK;
}
void
MediaStreamGraphImpl::CloseAudioInputImpl(AudioDataListener *aListener)
{
uint32_t count;
DebugOnly<bool> result = mInputDeviceUsers.Get(aListener, &count);
MOZ_ASSERT(result);
if (--count > 0) {
mInputDeviceUsers.Put(aListener, count);
return; // still in use
}
mInputDeviceUsers.Remove(aListener);
mInputDeviceID = -1;
mInputWanted = false;
AudioCallbackDriver *driver = CurrentDriver()->AsAudioCallbackDriver();
if (driver) {
driver->RemoveInputListener(aListener);
}
mAudioInputs.RemoveElement(aListener);
// Switch Drivers since we're adding or removing an input (to nothing/system or output only)
bool shouldAEC = false;
bool audioTrackPresent = AudioTrackPresent(shouldAEC);
MonitorAutoLock mon(mMonitor);
if (mLifecycleState == LIFECYCLE_RUNNING) {
GraphDriver* driver;
if (audioTrackPresent) {
// We still have audio output
LOG(LogLevel::Debug, ("CloseInput: output present (AudioCallback)"));
driver = new AudioCallbackDriver(this);
CurrentDriver()->SwitchAtNextIteration(driver);
} else if (CurrentDriver()->AsAudioCallbackDriver()) {
LOG(LogLevel::Debug,
("CloseInput: no output present (SystemClockCallback)"));
driver = new SystemClockDriver(this);
CurrentDriver()->SwitchAtNextIteration(driver);
} // else SystemClockDriver->SystemClockDriver, no switch
}
}
void
MediaStreamGraphImpl::CloseAudioInput(AudioDataListener *aListener)
{
// So, so, so annoying. Can't AppendMessage except on Mainthread
if (!NS_IsMainThread()) {
RefPtr<nsIRunnable> runnable =
WrapRunnable(this,
&MediaStreamGraphImpl::CloseAudioInput,
RefPtr<AudioDataListener>(aListener));
mAbstractMainThread->Dispatch(runnable.forget());
return;
}
class Message : public ControlMessage {
public:
Message(MediaStreamGraphImpl *aGraph, AudioDataListener *aListener) :
ControlMessage(nullptr), mGraph(aGraph), mListener(aListener) {}
void Run() override
{
mGraph->CloseAudioInputImpl(mListener);
}
MediaStreamGraphImpl *mGraph;
RefPtr<AudioDataListener> mListener;
};
this->AppendMessage(MakeUnique<Message>(this, aListener));
}
// All AudioInput listeners get the same speaker data (at least for now).
void
MediaStreamGraph::NotifyOutputData(AudioDataValue* aBuffer, size_t aFrames,
TrackRate aRate, uint32_t aChannels)
{
for (auto& listener : mAudioInputs) {
listener->NotifyOutputData(this, aBuffer, aFrames, aRate, aChannels);
}
}
bool
MediaStreamGraph::OnGraphThreadOrNotRunning() const
{
// either we're on the right thread (and calling CurrentDriver() is safe),
// or we're going to fail the assert anyway, so don't cross-check
// via CurrentDriver().
MediaStreamGraphImpl const * graph =
static_cast<MediaStreamGraphImpl const *>(this);
// if all the safety checks fail, assert we own the monitor
return graph->mDetectedNotRunning ?
NS_IsMainThread() : graph->mDriver->OnThread();
}
bool
MediaStreamGraphImpl::ShouldUpdateMainThread()
{
if (mRealtime) {
return true;
}
TimeStamp now = TimeStamp::Now();
if ((now - mLastMainThreadUpdate).ToMilliseconds() > CurrentDriver()->IterationDuration()) {
mLastMainThreadUpdate = now;
return true;
}
return false;
}
void
MediaStreamGraphImpl::PrepareUpdatesToMainThreadState(bool aFinalUpdate)
{
mMonitor.AssertCurrentThreadOwns();
// We don't want to frequently update the main thread about timing update
// when we are not running in realtime.
if (aFinalUpdate || ShouldUpdateMainThread()) {
// Strip updates that will be obsoleted below, so as to keep the length of
// mStreamUpdates sane.
size_t keptUpdateCount = 0;
for (size_t i = 0; i < mStreamUpdates.Length(); ++i) {
MediaStream* stream = mStreamUpdates[i].mStream;
// RemoveStreamGraphThread() clears mStream in updates for
// streams that are removed from the graph.
MOZ_ASSERT(!stream || stream->GraphImpl() == this);
if (!stream || stream->MainThreadNeedsUpdates()) {
// Discard this update as it has either been cleared when the stream
// was destroyed or there will be a newer update below.
continue;
}
if (keptUpdateCount != i) {
mStreamUpdates[keptUpdateCount] = Move(mStreamUpdates[i]);
MOZ_ASSERT(!mStreamUpdates[i].mStream);
}
++keptUpdateCount;
}
mStreamUpdates.TruncateLength(keptUpdateCount);
mStreamUpdates.SetCapacity(mStreamUpdates.Length() + mStreams.Length() +
mSuspendedStreams.Length());
for (MediaStream* stream : AllStreams()) {
if (!stream->MainThreadNeedsUpdates()) {
continue;
}
StreamUpdate* update = mStreamUpdates.AppendElement();
update->mStream = stream;
// No blocking to worry about here, since we've passed
// UpdateCurrentTimeForStreams.
update->mNextMainThreadCurrentTime =
stream->GraphTimeToStreamTime(mProcessedTime);
update->mNextMainThreadFinished = stream->mNotifiedFinished;
}
if (!mPendingUpdateRunnables.IsEmpty()) {
mUpdateRunnables.AppendElements(Move(mPendingUpdateRunnables));
}
}
// Don't send the message to the main thread if it's not going to have
// any work to do.
if (aFinalUpdate ||
!mUpdateRunnables.IsEmpty() ||
!mStreamUpdates.IsEmpty()) {
EnsureStableStateEventPosted();
}
}
GraphTime
MediaStreamGraphImpl::RoundUpToNextAudioBlock(GraphTime aTime)
{
StreamTime ticks = aTime;
uint64_t block = ticks >> WEBAUDIO_BLOCK_SIZE_BITS;
uint64_t nextBlock = block + 1;
StreamTime nextTicks = nextBlock << WEBAUDIO_BLOCK_SIZE_BITS;
return nextTicks;
}
void
MediaStreamGraphImpl::ProduceDataForStreamsBlockByBlock(uint32_t aStreamIndex,
TrackRate aSampleRate)
{
MOZ_ASSERT(aStreamIndex <= mFirstCycleBreaker,
"Cycle breaker is not AudioNodeStream?");
GraphTime t = mProcessedTime;
while (t < mStateComputedTime) {
GraphTime next = RoundUpToNextAudioBlock(t);
for (uint32_t i = mFirstCycleBreaker; i < mStreams.Length(); ++i) {
auto ns = static_cast<AudioNodeStream*>(mStreams[i]);
MOZ_ASSERT(ns->AsAudioNodeStream());
ns->ProduceOutputBeforeInput(t);
}
for (uint32_t i = aStreamIndex; i < mStreams.Length(); ++i) {
ProcessedMediaStream* ps = mStreams[i]->AsProcessedStream();
if (ps) {
ps->ProcessInput(t, next,
(next == mStateComputedTime) ? ProcessedMediaStream::ALLOW_FINISH : 0);
}
}
t = next;
}
NS_ASSERTION(t == mStateComputedTime,
"Something went wrong with rounding to block boundaries");
}
bool
MediaStreamGraphImpl::AllFinishedStreamsNotified()
{
for (MediaStream* stream : AllStreams()) {
if (stream->mFinished && !stream->mNotifiedFinished) {
return false;
}
}
return true;
}
void
MediaStreamGraphImpl::RunMessageAfterProcessing(UniquePtr<ControlMessage> aMessage)
{
MOZ_ASSERT(CurrentDriver()->OnThread());
if (mFrontMessageQueue.IsEmpty()) {
mFrontMessageQueue.AppendElement();
}
// Only one block is used for messages from the graph thread.
MOZ_ASSERT(mFrontMessageQueue.Length() == 1);
mFrontMessageQueue[0].mMessages.AppendElement(Move(aMessage));
}
void
MediaStreamGraphImpl::RunMessagesInQueue()
{
// Calculate independent action times for each batch of messages (each
// batch corresponding to an event loop task). This isolates the performance
// of different scripts to some extent.
for (uint32_t i = 0; i < mFrontMessageQueue.Length(); ++i) {
nsTArray<UniquePtr<ControlMessage>>& messages = mFrontMessageQueue[i].mMessages;
for (uint32_t j = 0; j < messages.Length(); ++j) {
messages[j]->Run();
}
}
mFrontMessageQueue.Clear();
}
void
MediaStreamGraphImpl::UpdateGraph(GraphTime aEndBlockingDecisions)
{
MOZ_ASSERT(aEndBlockingDecisions >= mProcessedTime);
// The next state computed time can be the same as the previous: it
// means the driver would be have been blocking indefinitly, but the graph has
// been woken up right after having been to sleep.
MOZ_ASSERT(aEndBlockingDecisions >= mStateComputedTime);
UpdateStreamOrder();
bool ensureNextIteration = false;
// Grab pending stream input and compute blocking time
for (MediaStream* stream : mStreams) {
if (SourceMediaStream* is = stream->AsSourceStream()) {
ExtractPendingInput(is, aEndBlockingDecisions, &ensureNextIteration);
}
if (stream->mFinished) {
// The stream's not suspended, and since it's finished, underruns won't
// stop it playing out. So there's no blocking other than what we impose
// here.
GraphTime endTime = stream->GetStreamTracks().GetAllTracksEnd() +
stream->mTracksStartTime;
if (endTime <= mStateComputedTime) {
LOG(LogLevel::Verbose,
("MediaStream %p is blocked due to being finished", stream));
stream->mStartBlocking = mStateComputedTime;
} else {
LOG(LogLevel::Verbose,
("MediaStream %p is finished, but not blocked yet (end at %f, with "
"blocking at %f)",
stream,
MediaTimeToSeconds(stream->GetTracksEnd()),
MediaTimeToSeconds(endTime)));
// Data can't be added to a finished stream, so underruns are irrelevant.
stream->mStartBlocking = std::min(endTime, aEndBlockingDecisions);
}
} else {
stream->mStartBlocking = WillUnderrun(stream, aEndBlockingDecisions);
}
}
for (MediaStream* stream : mSuspendedStreams) {
stream->mStartBlocking = mStateComputedTime;
}
// The loop is woken up so soon that IterationEnd() barely advances and we
// end up having aEndBlockingDecision == mStateComputedTime.
// Since stream blocking is computed in the interval of
// [mStateComputedTime, aEndBlockingDecision), it won't be computed at all.
// We should ensure next iteration so that pending blocking changes will be
// computed in next loop.
if (ensureNextIteration ||
aEndBlockingDecisions == mStateComputedTime) {
EnsureNextIteration();
}
}
void
MediaStreamGraphImpl::Process()
{
// Play stream contents.
bool allBlockedForever = true;
// True when we've done ProcessInput for all processed streams.
bool doneAllProducing = false;
// This is the number of frame that are written to the AudioStreams, for
// this cycle.
StreamTime ticksPlayed = 0;
mMixer.StartMixing();
// Figure out what each stream wants to do
for (uint32_t i = 0; i < mStreams.Length(); ++i) {
MediaStream* stream = mStreams[i];
if (!doneAllProducing) {
ProcessedMediaStream* ps = stream->AsProcessedStream();
if (ps) {
AudioNodeStream* n = stream->AsAudioNodeStream();
if (n) {
#ifdef DEBUG
// Verify that the sampling rate for all of the following streams is the same
for (uint32_t j = i + 1; j < mStreams.Length(); ++j) {
AudioNodeStream* nextStream = mStreams[j]->AsAudioNodeStream();
if (nextStream) {
MOZ_ASSERT(n->SampleRate() == nextStream->SampleRate(),
"All AudioNodeStreams in the graph must have the same sampling rate");
}
}
#endif
// Since an AudioNodeStream is present, go ahead and
// produce audio block by block for all the rest of the streams.
ProduceDataForStreamsBlockByBlock(i, n->SampleRate());
doneAllProducing = true;
} else {
ps->ProcessInput(mProcessedTime, mStateComputedTime,
ProcessedMediaStream::ALLOW_FINISH);
NS_ASSERTION(stream->mTracks.GetEnd() >=
GraphTimeToStreamTimeWithBlocking(stream, mStateComputedTime),
"Stream did not produce enough data");
}
}
}
NotifyHasCurrentData(stream);
// Only playback audio and video in real-time mode
if (mRealtime) {
CreateOrDestroyAudioStreams(stream);
if (CurrentDriver()->AsAudioCallbackDriver()) {
StreamTime ticksPlayedForThisStream = PlayAudio(stream);
if (!ticksPlayed) {
ticksPlayed = ticksPlayedForThisStream;
} else {
MOZ_ASSERT(!ticksPlayedForThisStream || ticksPlayedForThisStream == ticksPlayed,
"Each stream should have the same number of frame.");
}
}
}
if (stream->mStartBlocking > mProcessedTime) {
allBlockedForever = false;
}
}
if (CurrentDriver()->AsAudioCallbackDriver()) {
if (!ticksPlayed) {
// Nothing was played, so the mixer doesn't know how many frames were
// processed. We still tell it so AudioCallbackDriver knows how much has
// been processed. (bug 1406027)
mMixer.Mix(nullptr,
CurrentDriver()->AsAudioCallbackDriver()->OutputChannelCount(),
mStateComputedTime - mProcessedTime,
mSampleRate);
}
mMixer.FinishMixing();
}
if (!allBlockedForever) {
EnsureNextIteration();
}
}
bool
MediaStreamGraphImpl::UpdateMainThreadState()
{
MonitorAutoLock lock(mMonitor);
bool finalUpdate = mForceShutDown ||
(mProcessedTime >= mEndTime && AllFinishedStreamsNotified()) ||
(IsEmpty() && mBackMessageQueue.IsEmpty());
PrepareUpdatesToMainThreadState(finalUpdate);
if (finalUpdate) {
// Enter shutdown mode. The stable-state handler will detect this
// and complete shutdown. Destroy any streams immediately.
LOG(LogLevel::Debug,
("MediaStreamGraph %p waiting for main thread cleanup", this));
// We'll shut down this graph object if it does not get restarted.
mLifecycleState = LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP;
// No need to Destroy streams here. The main-thread owner of each
// stream is responsible for calling Destroy on them.
return false;
}
CurrentDriver()->WaitForNextIteration();
SwapMessageQueues();
return true;
}
bool
MediaStreamGraphImpl::OneIteration(GraphTime aStateEnd)
{
WebCore::DenormalDisabler disabler;
// Process graph message from the main thread for this iteration.
RunMessagesInQueue();
GraphTime stateEnd = std::min(aStateEnd, mEndTime);
UpdateGraph(stateEnd);
mStateComputedTime = stateEnd;
Process();
GraphTime oldProcessedTime = mProcessedTime;
mProcessedTime = stateEnd;
UpdateCurrentTimeForStreams(oldProcessedTime);
ProcessChunkMetadata(oldProcessedTime);
// Process graph messages queued from RunMessageAfterProcessing() on this
// thread during the iteration.
RunMessagesInQueue();
return UpdateMainThreadState();
}
void
MediaStreamGraphImpl::ApplyStreamUpdate(StreamUpdate* aUpdate)
{
mMonitor.AssertCurrentThreadOwns();
MediaStream* stream = aUpdate->mStream;
if (!stream)
return;
stream->mMainThreadCurrentTime = aUpdate->mNextMainThreadCurrentTime;
stream->mMainThreadFinished = aUpdate->mNextMainThreadFinished;
if (stream->ShouldNotifyStreamFinished()) {
stream->NotifyMainThreadListeners();
}
}
void
MediaStreamGraphImpl::ForceShutDown(media::ShutdownTicket* aShutdownTicket)
{
NS_ASSERTION(NS_IsMainThread(), "Must be called on main thread");
LOG(LogLevel::Debug, ("MediaStreamGraph %p ForceShutdown", this));
MonitorAutoLock lock(mMonitor);
if (aShutdownTicket) {
MOZ_ASSERT(!mForceShutdownTicket);
// Avoid waiting forever for a graph to shut down
// synchronously. Reports are that some 3rd-party audio drivers
// occasionally hang in shutdown (both for us and Chrome).
NS_NewTimerWithCallback(getter_AddRefs(mShutdownTimer),
this,
MediaStreamGraph::AUDIO_CALLBACK_DRIVER_SHUTDOWN_TIMEOUT,
nsITimer::TYPE_ONE_SHOT);
}
mForceShutDown = true;
mForceShutdownTicket = aShutdownTicket;
if (mLifecycleState == LIFECYCLE_THREAD_NOT_STARTED) {
// We *could* have just sent this a message to start up, so don't
// yank the rug out from under it. Tell it to startup and let it
// shut down.
RefPtr<GraphDriver> driver = CurrentDriver();
MonitorAutoUnlock unlock(mMonitor);
driver->Start();
}
EnsureNextIterationLocked();
}
NS_IMETHODIMP
MediaStreamGraphImpl::Notify(nsITimer* aTimer)
{
MonitorAutoLock lock(mMonitor);
NS_ASSERTION(!mForceShutdownTicket, "MediaStreamGraph took too long to shut down!");
// Sigh, graph took too long to shut down. Stop blocking system
// shutdown and hope all is well.
mForceShutdownTicket = nullptr;
return NS_OK;
}
NS_IMETHODIMP
MediaStreamGraphImpl::GetName(nsACString& aName)
{
aName.AssignLiteral("MediaStreamGraphImpl");
return NS_OK;
}
/* static */ StaticRefPtr<nsIAsyncShutdownBlocker> gMediaStreamGraphShutdownBlocker;
namespace {
class MediaStreamGraphShutDownRunnable : public Runnable {
public:
explicit MediaStreamGraphShutDownRunnable(MediaStreamGraphImpl* aGraph)
: Runnable("MediaStreamGraphShutDownRunnable")
, mGraph(aGraph)
{}
NS_IMETHOD Run()
{
NS_ASSERTION(mGraph->mDetectedNotRunning,
"We should know the graph thread control loop isn't running!");
LOG(LogLevel::Debug, ("Shutting down graph %p", mGraph.get()));
// We've asserted the graph isn't running. Use mDriver instead of CurrentDriver
// to avoid thread-safety checks
#if 0 // AudioCallbackDrivers are released asynchronously anyways
// XXX a better test would be have setting mDetectedNotRunning make sure
// any current callback has finished and block future ones -- or just
// handle it all in Shutdown()!
if (mGraph->mDriver->AsAudioCallbackDriver()) {
MOZ_ASSERT(!mGraph->mDriver->AsAudioCallbackDriver()->InCallback());
}
#endif
mGraph->mDriver->Shutdown(); // This will wait until it's shutdown since
// we'll start tearing down the graph after this
// Release the driver now so that an AudioCallbackDriver will release its
// SharedThreadPool reference. Each SharedThreadPool reference must be
// released before SharedThreadPool::SpinUntilEmpty() runs on
// xpcom-shutdown-threads. Don't wait for GC/CC to release references to
// objects owning streams, or for expiration of mGraph->mShutdownTimer,
// which won't otherwise release its reference on the graph until
// nsTimerImpl::Shutdown(), which runs after xpcom-shutdown-threads.
mGraph->mDriver = nullptr;
// Safe to access these without the monitor since the graph isn't running.
// We may be one of several graphs. Drop ticket to eventually unblock shutdown.
if (mGraph->mShutdownTimer && !mGraph->mForceShutdownTicket) {
MOZ_ASSERT(false,
"AudioCallbackDriver took too long to shut down and we let shutdown"
" continue - freezing and leaking");
// The timer fired, so we may be deeper in shutdown now. Block any further
// teardown and just leak, for safety.
return NS_OK;
}
mGraph->mForceShutdownTicket = nullptr;
// We can't block past the final LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION
// stage, since completion of that stage requires all streams to be freed,
// which requires shutdown to proceed.
// mGraph's thread is not running so it's OK to do whatever here
if (mGraph->IsEmpty()) {
// mGraph is no longer needed, so delete it.
mGraph->Destroy();
} else {
// The graph is not empty. We must be in a forced shutdown, or a
// non-realtime graph that has finished processing. Some later
// AppendMessage will detect that the graph has been emptied, and
// delete it.
NS_ASSERTION(mGraph->mForceShutDown || !mGraph->mRealtime,
"Not in forced shutdown?");
for (MediaStream* stream : mGraph->AllStreams()) {
// Clean up all MediaSegments since we cannot release Images too
// late during shutdown. Also notify listeners that they were removed
// so they can clean up any gfx resources.
if (SourceMediaStream* source = stream->AsSourceStream()) {
// Finishing a SourceStream prevents new data from being appended.
source->Finish();
}
stream->GetStreamTracks().Clear();
stream->RemoveAllListenersImpl();
}
mGraph->mLifecycleState =
MediaStreamGraphImpl::LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION;
}
return NS_OK;
}
private:
RefPtr<MediaStreamGraphImpl> mGraph;
};
class MediaStreamGraphStableStateRunnable : public Runnable {
public:
explicit MediaStreamGraphStableStateRunnable(MediaStreamGraphImpl* aGraph,
bool aSourceIsMSG)
: Runnable("MediaStreamGraphStableStateRunnable")
, mGraph(aGraph)
, mSourceIsMSG(aSourceIsMSG)
{
}
NS_IMETHOD Run() override
{
if (mGraph) {
mGraph->RunInStableState(mSourceIsMSG);
}
return NS_OK;
}
private:
RefPtr<MediaStreamGraphImpl> mGraph;
bool mSourceIsMSG;
};
/*
* Control messages forwarded from main thread to graph manager thread
*/
class CreateMessage : public ControlMessage {
public:
explicit CreateMessage(MediaStream* aStream) : ControlMessage(aStream) {}
void Run() override
{
mStream->GraphImpl()->AddStreamGraphThread(mStream);
}
void RunDuringShutdown() override
{
// Make sure to run this message during shutdown too, to make sure
// that we balance the number of streams registered with the graph
// as they're destroyed during shutdown.
Run();
}
};
} // namespace
void
MediaStreamGraphImpl::RunInStableState(bool aSourceIsMSG)
{
NS_ASSERTION(NS_IsMainThread(), "Must be called on main thread");
nsTArray<nsCOMPtr<nsIRunnable> > runnables;
// When we're doing a forced shutdown, pending control messages may be
// run on the main thread via RunDuringShutdown. Those messages must
// run without the graph monitor being held. So, we collect them here.
nsTArray<UniquePtr<ControlMessage>> controlMessagesToRunDuringShutdown;
{
MonitorAutoLock lock(mMonitor);
if (aSourceIsMSG) {
MOZ_ASSERT(mPostedRunInStableStateEvent);
mPostedRunInStableStateEvent = false;
}
// This should be kept in sync with the LifecycleState enum in
// MediaStreamGraphImpl.h
const char* LifecycleState_str[] = {
"LIFECYCLE_THREAD_NOT_STARTED",
"LIFECYCLE_RUNNING",
"LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP",
"LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN",
"LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION"
};
if (mLifecycleState != LIFECYCLE_RUNNING) {
LOG(LogLevel::Debug,
("Running %p in stable state. Current state: %s",
this,
LifecycleState_str[mLifecycleState]));
}
runnables.SwapElements(mUpdateRunnables);
for (uint32_t i = 0; i < mStreamUpdates.Length(); ++i) {
StreamUpdate* update = &mStreamUpdates[i];
if (update->mStream) {
ApplyStreamUpdate(update);
}
}
mStreamUpdates.Clear();
if (mCurrentTaskMessageQueue.IsEmpty()) {
if (mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP && IsEmpty()) {
// Complete shutdown. First, ensure that this graph is no longer used.
// A new graph graph will be created if one is needed.
// Asynchronously clean up old graph. We don't want to do this
// synchronously because it spins the event loop waiting for threads
// to shut down, and we don't want to do that in a stable state handler.
mLifecycleState = LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
LOG(LogLevel::Debug,
("Sending MediaStreamGraphShutDownRunnable %p", this));
nsCOMPtr<nsIRunnable> event = new MediaStreamGraphShutDownRunnable(this );
mAbstractMainThread->Dispatch(event.forget());
LOG(LogLevel::Debug, ("Disconnecting MediaStreamGraph %p", this));
// Find the graph in the hash table and remove it.
for (auto iter = gGraphs.Iter(); !iter.Done(); iter.Next()) {
if (iter.UserData() == this) {
iter.Remove();
break;
}
}
}
} else {
if (mLifecycleState <= LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
MessageBlock* block = mBackMessageQueue.AppendElement();
block->mMessages.SwapElements(mCurrentTaskMessageQueue);
EnsureNextIterationLocked();
}
// If the MediaStreamGraph has more messages going to it, try to revive
// it to process those messages. Don't do this if we're in a forced
// shutdown or it's a non-realtime graph that has already terminated
// processing.
if (mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP &&
mRealtime && !mForceShutDown) {
mLifecycleState = LIFECYCLE_RUNNING;
// Revive the MediaStreamGraph since we have more messages going to it.
// Note that we need to put messages into its queue before reviving it,
// or it might exit immediately.
{
LOG(LogLevel::Debug,
("Reviving a graph (%p) ! %s",
this,
CurrentDriver()->AsAudioCallbackDriver() ? "AudioDriver"
: "SystemDriver"));
RefPtr<GraphDriver> driver = CurrentDriver();
MonitorAutoUnlock unlock(mMonitor);
driver->Revive();
}
}
}
// Don't start the thread for a non-realtime graph until it has been
// explicitly started by StartNonRealtimeProcessing.
if (mLifecycleState == LIFECYCLE_THREAD_NOT_STARTED &&
(mRealtime || mNonRealtimeProcessing)) {
mLifecycleState = LIFECYCLE_RUNNING;
// Start the thread now. We couldn't start it earlier because
// the graph might exit immediately on finding it has no streams. The
// first message for a new graph must create a stream.
{
// We should exit the monitor for now, because starting a stream might
// take locks, and we don't want to deadlock.
LOG(LogLevel::Debug,
("Starting a graph (%p) ! %s",
this,
CurrentDriver()->AsAudioCallbackDriver() ? "AudioDriver"
: "SystemDriver"));
RefPtr<GraphDriver> driver = CurrentDriver();
MonitorAutoUnlock unlock(mMonitor);
driver->Start();
// It's not safe to Shutdown() a thread from StableState, and
// releasing this may shutdown a SystemClockDriver thread.
// Proxy the release to outside of StableState.
NS_ReleaseOnMainThreadSystemGroup(
"MediaStreamGraphImpl::CurrentDriver", driver.forget(),
true); // always proxy
}
}
if ((mForceShutDown || !mRealtime) &&
mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
// Defer calls to RunDuringShutdown() to happen while mMonitor is not held.
for (uint32_t i = 0; i < mBackMessageQueue.Length(); ++i) {
MessageBlock& mb = mBackMessageQueue[i];
controlMessagesToRunDuringShutdown.AppendElements(Move(mb.mMessages));
}
mBackMessageQueue.Clear();
MOZ_ASSERT(mCurrentTaskMessageQueue.IsEmpty());
// Stop MediaStreamGraph threads. Do not clear gGraph since
// we have outstanding DOM objects that may need it.
mLifecycleState = LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
nsCOMPtr<nsIRunnable> event = new MediaStreamGraphShutDownRunnable(this);
mAbstractMainThread->Dispatch(event.forget());
}
mDetectedNotRunning = mLifecycleState > LIFECYCLE_RUNNING;
}
// Make sure we get a new current time in the next event loop task
if (!aSourceIsMSG) {
MOZ_ASSERT(mPostedRunInStableState);
mPostedRunInStableState = false;
}
for (uint32_t i = 0; i < controlMessagesToRunDuringShutdown.Length(); ++i) {
controlMessagesToRunDuringShutdown[i]->RunDuringShutdown();
}
#ifdef DEBUG
mCanRunMessagesSynchronously = mDetectedNotRunning &&
mLifecycleState >= LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
#endif
for (uint32_t i = 0; i < runnables.Length(); ++i) {
runnables[i]->Run();
}
}
void
MediaStreamGraphImpl::EnsureRunInStableState()
{
NS_ASSERTION(NS_IsMainThread(), "main thread only");
if (mPostedRunInStableState)
return;
mPostedRunInStableState = true;
nsCOMPtr<nsIRunnable> event = new MediaStreamGraphStableStateRunnable(this, false);
nsContentUtils::RunInStableState(event.forget());
}
void
MediaStreamGraphImpl::EnsureStableStateEventPosted()
{
mMonitor.AssertCurrentThreadOwns();
if (mPostedRunInStableStateEvent)
return;
mPostedRunInStableStateEvent = true;
nsCOMPtr<nsIRunnable> event = new MediaStreamGraphStableStateRunnable(this, true);
mAbstractMainThread->Dispatch(event.forget());
}
void
MediaStreamGraphImpl::AppendMessage(UniquePtr<ControlMessage> aMessage)
{
MOZ_ASSERT(NS_IsMainThread(), "main thread only");
MOZ_ASSERT(!aMessage->GetStream() ||
!aMessage->GetStream()->IsDestroyed(),
"Stream already destroyed");
if (mDetectedNotRunning &&
mLifecycleState > LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
// The graph control loop is not running and main thread cleanup has
// happened. From now on we can't append messages to mCurrentTaskMessageQueue,
// because that will never be processed again, so just RunDuringShutdown
// this message.
// This should only happen during forced shutdown, or after a non-realtime
// graph has finished processing.
#ifdef DEBUG
MOZ_ASSERT(mCanRunMessagesSynchronously);
mCanRunMessagesSynchronously = false;
#endif
aMessage->RunDuringShutdown();
#ifdef DEBUG
mCanRunMessagesSynchronously = true;
#endif
if (IsEmpty() &&
mLifecycleState >= LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION) {
// Find the graph in the hash table and remove it.
for (auto iter = gGraphs.Iter(); !iter.Done(); iter.Next()) {
if (iter.UserData() == this) {
iter.Remove();
break;
}
}
Destroy();
}
return;
}
mCurrentTaskMessageQueue.AppendElement(Move(aMessage));
EnsureRunInStableState();
}
void
MediaStreamGraphImpl::Dispatch(already_AddRefed<nsIRunnable>&& aRunnable)
{
mAbstractMainThread->Dispatch(Move(aRunnable));
}
MediaStream::MediaStream()
: mTracksStartTime(0)
, mStartBlocking(GRAPH_TIME_MAX)
, mSuspendedCount(0)
, mFinished(false)
, mNotifiedFinished(false)
, mNotifiedBlocked(false)
, mHasCurrentData(false)
, mNotifiedHasCurrentData(false)
, mMainThreadCurrentTime(0)
, mMainThreadFinished(false)
, mFinishedNotificationSent(false)
, mMainThreadDestroyed(false)
, mNrOfMainThreadUsers(0)
, mGraph(nullptr)
{
MOZ_COUNT_CTOR(MediaStream);
}
MediaStream::~MediaStream()
{
MOZ_COUNT_DTOR(MediaStream);
NS_ASSERTION(mMainThreadDestroyed, "Should have been destroyed already");
NS_ASSERTION(mMainThreadListeners.IsEmpty(),
"All main thread listeners should have been removed");
}
size_t
MediaStream::SizeOfExcludingThis(MallocSizeOf aMallocSizeOf) const
{
size_t amount = 0;
// Not owned:
// - mGraph - Not reported here
// - mConsumers - elements
// Future:
// - mVideoOutputs - elements
// - mLastPlayedVideoFrame
// - mListeners - elements
// - mAudioOutputStream - elements
amount += mTracks.SizeOfExcludingThis(aMallocSizeOf);
amount += mAudioOutputs.ShallowSizeOfExcludingThis(aMallocSizeOf);
amount += mVideoOutputs.ShallowSizeOfExcludingThis(aMallocSizeOf);
amount += mListeners.ShallowSizeOfExcludingThis(aMallocSizeOf);
amount += mMainThreadListeners.ShallowSizeOfExcludingThis(aMallocSizeOf);
amount += mDisabledTracks.ShallowSizeOfExcludingThis(aMallocSizeOf);
amount += mConsumers.ShallowSizeOfExcludingThis(aMallocSizeOf);
return amount;
}
size_t
MediaStream::SizeOfIncludingThis(MallocSizeOf aMallocSizeOf) const
{
return aMallocSizeOf(this) + SizeOfExcludingThis(aMallocSizeOf);
}
void
MediaStream::IncrementSuspendCount()
{
++mSuspendedCount;
if (mSuspendedCount == 1) {
for (uint32_t i = 0; i < mConsumers.Length(); ++i) {
mConsumers[i]->Suspended();
}
}
}
void
MediaStream::DecrementSuspendCount()
{
NS_ASSERTION(mSuspendedCount > 0, "Suspend count underrun");
--mSuspendedCount;
if (mSuspendedCount == 0) {
for (uint32_t i = 0; i < mConsumers.Length(); ++i) {
mConsumers[i]->Resumed();
}
}
}
MediaStreamGraphImpl*
MediaStream::GraphImpl()
{
return mGraph;
}
MediaStreamGraph*
MediaStream::Graph()
{
return mGraph;
}
void
MediaStream::SetGraphImpl(MediaStreamGraphImpl* aGraph)
{
MOZ_ASSERT(!mGraph, "Should only be called once");
mGraph = aGraph;
mTracks.InitGraphRate(aGraph->GraphRate());
}
void
MediaStream::SetGraphImpl(MediaStreamGraph* aGraph)
{
MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(aGraph);
SetGraphImpl(graph);
}
StreamTime
MediaStream::GraphTimeToStreamTime(GraphTime aTime)
{
NS_ASSERTION(mStartBlocking == GraphImpl()->mStateComputedTime ||
aTime <= mStartBlocking,
"Incorrectly ignoring blocking!");
return aTime - mTracksStartTime;
}
GraphTime
MediaStream::StreamTimeToGraphTime(StreamTime aTime)
{
NS_ASSERTION(mStartBlocking == GraphImpl()->mStateComputedTime ||
aTime + mTracksStartTime <= mStartBlocking,
"Incorrectly ignoring blocking!");
return aTime + mTracksStartTime;
}
StreamTime
MediaStream::GraphTimeToStreamTimeWithBlocking(GraphTime aTime)
{
return GraphImpl()->GraphTimeToStreamTimeWithBlocking(this, aTime);
}
void
MediaStream::FinishOnGraphThread()
{
GraphImpl()->FinishStream(this);
}
StreamTracks::Track*
MediaStream::FindTrack(TrackID aID)
{
return mTracks.FindTrack(aID);
}
StreamTracks::Track*
MediaStream::EnsureTrack(TrackID aTrackId)
{
StreamTracks::Track* track = mTracks.FindTrack(aTrackId);
if (!track) {
nsAutoPtr<MediaSegment> segment(new AudioSegment());
for (uint32_t j = 0; j < mListeners.Length(); ++j) {
MediaStreamListener* l = mListeners[j];
l->NotifyQueuedTrackChanges(Graph(), aTrackId, 0,
TrackEventCommand::TRACK_EVENT_CREATED,
*segment);
// TODO If we ever need to ensure several tracks at once, we will have to
// change this.
l->NotifyFinishedTrackCreation(Graph());
}
track = &mTracks.AddTrack(aTrackId, 0, segment.forget());
}
return track;
}
void
MediaStream::RemoveAllListenersImpl()
{
GraphImpl()->AssertOnGraphThreadOrNotRunning();
auto streamListeners(mListeners);
for (auto& l : streamListeners) {
l->NotifyEvent(GraphImpl(), MediaStreamGraphEvent::EVENT_REMOVED);
}
mListeners.Clear();
auto trackListeners(mTrackListeners);
for (auto& l : trackListeners) {
l.mListener->NotifyRemoved();
}
mTrackListeners.Clear();
if (SourceMediaStream* source = AsSourceStream()) {
source->RemoveAllDirectListeners();
}
}
void
MediaStream::DestroyImpl()
{
for (int32_t i = mConsumers.Length() - 1; i >= 0; --i) {
mConsumers[i]->Disconnect();
}
mGraph = nullptr;
}
void
MediaStream::Destroy()
{
NS_ASSERTION(mNrOfMainThreadUsers == 0,
"Do not mix Destroy() and RegisterUser()/UnregisterUser()");
// Keep this stream alive until we leave this method
RefPtr<MediaStream> kungFuDeathGrip = this;
class Message : public ControlMessage {
public:
explicit Message(MediaStream* aStream) : ControlMessage(aStream) {}
void Run() override
{
mStream->RemoveAllListenersImpl();
auto graph = mStream->GraphImpl();
mStream->DestroyImpl();
graph->RemoveStreamGraphThread(mStream);
}
void RunDuringShutdown() override
{ Run(); }
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this));
// Message::RunDuringShutdown may have removed this stream from the graph,
// but our kungFuDeathGrip above will have kept this stream alive if
// necessary.
mMainThreadDestroyed = true;
}
void
MediaStream::RegisterUser()
{
MOZ_ASSERT(NS_IsMainThread());
++mNrOfMainThreadUsers;
}
void
MediaStream::UnregisterUser()
{
MOZ_ASSERT(NS_IsMainThread());
--mNrOfMainThreadUsers;
NS_ASSERTION(mNrOfMainThreadUsers >= 0, "Double-removal of main thread user");
NS_ASSERTION(!IsDestroyed(), "Do not mix Destroy() and RegisterUser()/UnregisterUser()");
if (mNrOfMainThreadUsers == 0) {
Destroy();
}
}
void
MediaStream::AddAudioOutput(void* aKey)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, void* aKey) : ControlMessage(aStream), mKey(aKey) {}
void Run() override
{
mStream->AddAudioOutputImpl(mKey);
}
void* mKey;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aKey));
}
void
MediaStream::SetAudioOutputVolumeImpl(void* aKey, float aVolume)
{
for (uint32_t i = 0; i < mAudioOutputs.Length(); ++i) {
if (mAudioOutputs[i].mKey == aKey) {
mAudioOutputs[i].mVolume = aVolume;
return;
}
}
NS_ERROR("Audio output key not found");
}
void
MediaStream::SetAudioOutputVolume(void* aKey, float aVolume)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, void* aKey, float aVolume) :
ControlMessage(aStream), mKey(aKey), mVolume(aVolume) {}
void Run() override
{
mStream->SetAudioOutputVolumeImpl(mKey, mVolume);
}
void* mKey;
float mVolume;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aKey, aVolume));
}
void
MediaStream::AddAudioOutputImpl(void* aKey)
{
LOG(LogLevel::Info,
("MediaStream %p Adding AudioOutput for key %p", this, aKey));
mAudioOutputs.AppendElement(AudioOutput(aKey));
}
void
MediaStream::RemoveAudioOutputImpl(void* aKey)
{
LOG(LogLevel::Info,
("MediaStream %p Removing AudioOutput for key %p", this, aKey));
for (uint32_t i = 0; i < mAudioOutputs.Length(); ++i) {
if (mAudioOutputs[i].mKey == aKey) {
mAudioOutputs.RemoveElementAt(i);
return;
}
}
NS_ERROR("Audio output key not found");
}
void
MediaStream::RemoveAudioOutput(void* aKey)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, void* aKey) :
ControlMessage(aStream), mKey(aKey) {}
void Run() override
{
mStream->RemoveAudioOutputImpl(mKey);
}
void* mKey;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aKey));
}
void
MediaStream::AddVideoOutputImpl(already_AddRefed<MediaStreamVideoSink> aSink,
TrackID aID)
{
RefPtr<MediaStreamVideoSink> sink = aSink;
LOG(LogLevel::Info,
("MediaStream %p Adding MediaStreamVideoSink %p as output",
this,
sink.get()));
MOZ_ASSERT(aID != TRACK_NONE);
for (auto entry : mVideoOutputs) {
if (entry.mListener == sink &&
(entry.mTrackID == TRACK_ANY || entry.mTrackID == aID)) {
return;
}
}
TrackBound<MediaStreamVideoSink>* l = mVideoOutputs.AppendElement();
l->mListener = sink;
l->mTrackID = aID;
AddDirectTrackListenerImpl(sink.forget(), aID);
}
void
MediaStream::RemoveVideoOutputImpl(MediaStreamVideoSink* aSink,
TrackID aID)
{
LOG(
LogLevel::Info,
("MediaStream %p Removing MediaStreamVideoSink %p as output", this, aSink));
MOZ_ASSERT(aID != TRACK_NONE);
// Ensure that any frames currently queued for playback by the compositor
// are removed.
aSink->ClearFrames();
for (size_t i = 0; i < mVideoOutputs.Length(); ++i) {
if (mVideoOutputs[i].mListener == aSink &&
(mVideoOutputs[i].mTrackID == TRACK_ANY ||
mVideoOutputs[i].mTrackID == aID)) {
mVideoOutputs.RemoveElementAt(i);
}
}
RemoveDirectTrackListenerImpl(aSink, aID);
}
void
MediaStream::AddVideoOutput(MediaStreamVideoSink* aSink, TrackID aID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamVideoSink* aSink, TrackID aID) :
ControlMessage(aStream), mSink(aSink), mID(aID) {}
void Run() override
{
mStream->AddVideoOutputImpl(mSink.forget(), mID);
}
RefPtr<MediaStreamVideoSink> mSink;
TrackID mID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aSink, aID));
}
void
MediaStream::RemoveVideoOutput(MediaStreamVideoSink* aSink, TrackID aID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamVideoSink* aSink, TrackID aID) :
ControlMessage(aStream), mSink(aSink), mID(aID) {}
void Run() override
{
mStream->RemoveVideoOutputImpl(mSink, mID);
}
RefPtr<MediaStreamVideoSink> mSink;
TrackID mID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aSink, aID));
}
void
MediaStream::Suspend()
{
class Message : public ControlMessage {
public:
explicit Message(MediaStream* aStream) :
ControlMessage(aStream) {}
void Run() override
{
mStream->GraphImpl()->IncrementSuspendCount(mStream);
}
};
// This can happen if this method has been called asynchronously, and the
// stream has been destroyed since then.
if (mMainThreadDestroyed) {
return;
}
GraphImpl()->AppendMessage(MakeUnique<Message>(this));
}
void
MediaStream::Resume()
{
class Message : public ControlMessage {
public:
explicit Message(MediaStream* aStream) :
ControlMessage(aStream) {}
void Run() override
{
mStream->GraphImpl()->DecrementSuspendCount(mStream);
}
};
// This can happen if this method has been called asynchronously, and the
// stream has been destroyed since then.
if (mMainThreadDestroyed) {
return;
}
GraphImpl()->AppendMessage(MakeUnique<Message>(this));
}
void
MediaStream::AddListenerImpl(already_AddRefed<MediaStreamListener> aListener)
{
MediaStreamListener* listener = *mListeners.AppendElement() = aListener;
listener->NotifyBlockingChanged(GraphImpl(),
mNotifiedBlocked ? MediaStreamListener::BLOCKED : MediaStreamListener::UNBLOCKED);
for (StreamTracks::TrackIter it(mTracks); !it.IsEnded(); it.Next()) {
MediaStream* inputStream = nullptr;
TrackID inputTrackID = TRACK_INVALID;
if (ProcessedMediaStream* ps = AsProcessedStream()) {
// The only ProcessedMediaStream where we should have listeners is
// TrackUnionStream - it's what's used as owned stream in DOMMediaStream,
// the only main-thread exposed stream type.
// TrackUnionStream guarantees that each of its tracks has an input track.
// Other types do not implement GetInputStreamFor() and will return null.
inputStream = ps->GetInputStreamFor(it->GetID());
MOZ_ASSERT(inputStream);
inputTrackID = ps->GetInputTrackIDFor(it->GetID());
MOZ_ASSERT(IsTrackIDExplicit(inputTrackID));
}
uint32_t flags = TrackEventCommand::TRACK_EVENT_CREATED;
if (it->IsEnded()) {
flags |= TrackEventCommand::TRACK_EVENT_ENDED;
}
nsAutoPtr<MediaSegment> segment(it->GetSegment()->CreateEmptyClone());
listener->NotifyQueuedTrackChanges(Graph(), it->GetID(), it->GetEnd(),
static_cast<TrackEventCommand>(flags), *segment,
inputStream, inputTrackID);
}
if (mNotifiedFinished) {
listener->NotifyEvent(GraphImpl(), MediaStreamGraphEvent::EVENT_FINISHED);
}
if (mNotifiedHasCurrentData) {
listener->NotifyHasCurrentData(GraphImpl());
}
}
void
MediaStream::AddListener(MediaStreamListener* aListener)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamListener* aListener) :
ControlMessage(aStream), mListener(aListener) {}
void Run() override
{
mStream->AddListenerImpl(mListener.forget());
}
RefPtr<MediaStreamListener> mListener;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener));
}
void
MediaStream::RemoveListenerImpl(MediaStreamListener* aListener)
{
// wouldn't need this if we could do it in the opposite order
RefPtr<MediaStreamListener> listener(aListener);
mListeners.RemoveElement(aListener);
listener->NotifyEvent(GraphImpl(), MediaStreamGraphEvent::EVENT_REMOVED);
}
void
MediaStream::RemoveListener(MediaStreamListener* aListener)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamListener* aListener) :
ControlMessage(aStream), mListener(aListener) {}
void Run() override
{
mStream->RemoveListenerImpl(mListener);
}
RefPtr<MediaStreamListener> mListener;
};
// If the stream is destroyed the Listeners have or will be
// removed.
if (!IsDestroyed()) {
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener));
}
}
void
MediaStream::AddTrackListenerImpl(already_AddRefed<MediaStreamTrackListener> aListener,
TrackID aTrackID)
{
TrackBound<MediaStreamTrackListener>* l = mTrackListeners.AppendElement();
l->mListener = aListener;
l->mTrackID = aTrackID;
StreamTracks::Track* track = FindTrack(aTrackID);
if (!track) {
return;
}
PrincipalHandle lastPrincipalHandle =
track->GetSegment()->GetLastPrincipalHandle();
l->mListener->NotifyPrincipalHandleChanged(Graph(), lastPrincipalHandle);
if (track->IsEnded() &&
track->GetEnd() <= GraphTimeToStreamTime(GraphImpl()->mStateComputedTime)) {
l->mListener->NotifyEnded();
}
}
void
MediaStream::AddTrackListener(MediaStreamTrackListener* aListener,
TrackID aTrackID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamTrackListener* aListener,
TrackID aTrackID) :
ControlMessage(aStream), mListener(aListener), mTrackID(aTrackID) {}
void Run() override
{
mStream->AddTrackListenerImpl(mListener.forget(), mTrackID);
}
RefPtr<MediaStreamTrackListener> mListener;
TrackID mTrackID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener, aTrackID));
}
void
MediaStream::RemoveTrackListenerImpl(MediaStreamTrackListener* aListener,
TrackID aTrackID)
{
for (size_t i = 0; i < mTrackListeners.Length(); ++i) {
if (mTrackListeners[i].mListener == aListener &&
mTrackListeners[i].mTrackID == aTrackID) {
mTrackListeners[i].mListener->NotifyRemoved();
mTrackListeners.RemoveElementAt(i);
return;
}
}
}
void
MediaStream::RemoveTrackListener(MediaStreamTrackListener* aListener,
TrackID aTrackID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, MediaStreamTrackListener* aListener,
TrackID aTrackID) :
ControlMessage(aStream), mListener(aListener), mTrackID(aTrackID) {}
void Run() override
{
mStream->RemoveTrackListenerImpl(mListener, mTrackID);
}
void RunDuringShutdown() override
{
// During shutdown we still want the listener's NotifyRemoved to be
// called, since not doing that might block shutdown of other modules.
Run();
}
RefPtr<MediaStreamTrackListener> mListener;
TrackID mTrackID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener, aTrackID));
}
void
MediaStream::AddDirectTrackListenerImpl(already_AddRefed<DirectMediaStreamTrackListener> aListener,
TrackID aTrackID)
{
// Base implementation, for streams that don't support direct track listeners.
RefPtr<DirectMediaStreamTrackListener> listener = aListener;
listener->NotifyDirectListenerInstalled(
DirectMediaStreamTrackListener::InstallationResult::STREAM_NOT_SUPPORTED);
}
void
MediaStream::AddDirectTrackListener(DirectMediaStreamTrackListener* aListener,
TrackID aTrackID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, DirectMediaStreamTrackListener* aListener,
TrackID aTrackID) :
ControlMessage(aStream), mListener(aListener), mTrackID(aTrackID) {}
void Run() override
{
mStream->AddDirectTrackListenerImpl(mListener.forget(), mTrackID);
}
RefPtr<DirectMediaStreamTrackListener> mListener;
TrackID mTrackID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener, aTrackID));
}
void
MediaStream::RemoveDirectTrackListenerImpl(DirectMediaStreamTrackListener* aListener,
TrackID aTrackID)
{
// Base implementation, the listener was never added so nothing to do.
RefPtr<DirectMediaStreamTrackListener> listener = aListener;
}
void
MediaStream::RemoveDirectTrackListener(DirectMediaStreamTrackListener* aListener,
TrackID aTrackID)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, DirectMediaStreamTrackListener* aListener,
TrackID aTrackID) :
ControlMessage(aStream), mListener(aListener), mTrackID(aTrackID) {}
void Run() override
{
mStream->RemoveDirectTrackListenerImpl(mListener, mTrackID);
}
void RunDuringShutdown() override
{
// During shutdown we still want the listener's
// NotifyDirectListenerUninstalled to be called, since not doing that
// might block shutdown of other modules.
Run();
}
RefPtr<DirectMediaStreamTrackListener> mListener;
TrackID mTrackID;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aListener, aTrackID));
}
void
MediaStream::RunAfterPendingUpdates(already_AddRefed<nsIRunnable> aRunnable)
{
MOZ_ASSERT(NS_IsMainThread());
MediaStreamGraphImpl* graph = GraphImpl();
nsCOMPtr<nsIRunnable> runnable(aRunnable);
// Special case when a non-realtime graph has not started, to ensure the
// runnable will run in finite time.
if (!(graph->mRealtime || graph->mNonRealtimeProcessing)) {
runnable->Run();
return;
}
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, already_AddRefed<nsIRunnable> aRunnable)
: ControlMessage(aStream)
, mRunnable(aRunnable)
{}
void Run() override
{
mStream->Graph()->DispatchToMainThreadAfterStreamStateUpdate(
mRunnable.forget());
}
void RunDuringShutdown() override
{
// Don't run mRunnable now as it may call AppendMessage() which would
// assume that there are no remaining controlMessagesToRunDuringShutdown.
MOZ_ASSERT(NS_IsMainThread());
mStream->GraphImpl()->Dispatch(mRunnable.forget());
}
private:
nsCOMPtr<nsIRunnable> mRunnable;
};
graph->AppendMessage(MakeUnique<Message>(this, runnable.forget()));
}
void
MediaStream::SetTrackEnabledImpl(TrackID aTrackID, DisabledTrackMode aMode)
{
if (aMode == DisabledTrackMode::ENABLED) {
for (int32_t i = mDisabledTracks.Length() - 1; i >= 0; --i) {
if (aTrackID == mDisabledTracks[i].mTrackID) {
mDisabledTracks.RemoveElementAt(i);
return;
}
}
} else {
for (const DisabledTrack& t : mDisabledTracks) {
if (aTrackID == t.mTrackID) {
NS_ERROR("Changing disabled track mode for a track is not allowed");
return;
}
}
mDisabledTracks.AppendElement(Move(DisabledTrack(aTrackID, aMode)));
}
}
DisabledTrackMode
MediaStream::GetDisabledTrackMode(TrackID aTrackID)
{
for (const DisabledTrack& t : mDisabledTracks) {
if (t.mTrackID == aTrackID) {
return t.mMode;
}
}
return DisabledTrackMode::ENABLED;
}
void
MediaStream::SetTrackEnabled(TrackID aTrackID, DisabledTrackMode aMode)
{
class Message : public ControlMessage {
public:
Message(MediaStream* aStream, TrackID aTrackID, DisabledTrackMode aMode) :
ControlMessage(aStream),
mTrackID(aTrackID),
mMode(aMode) {}
void Run() override
{
mStream->SetTrackEnabledImpl(mTrackID, mMode);
}
TrackID mTrackID;
DisabledTrackMode mMode;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aTrackID, aMode));
}
void
MediaStream::ApplyTrackDisabling(TrackID aTrackID, MediaSegment* aSegment, MediaSegment* aRawSegment)
{
DisabledTrackMode mode = GetDisabledTrackMode(aTrackID);
if (mode == DisabledTrackMode::ENABLED) {
return;
}
if (mode == DisabledTrackMode::SILENCE_BLACK) {
aSegment->ReplaceWithDisabled();
if (aRawSegment) {
aRawSegment->ReplaceWithDisabled();
}
} else if (mode == DisabledTrackMode::SILENCE_FREEZE) {
aSegment->ReplaceWithNull();
if (aRawSegment) {
aRawSegment->ReplaceWithNull();
}
} else {
MOZ_CRASH("Unsupported mode");
}
}
void
MediaStream::AddMainThreadListener(MainThreadMediaStreamListener* aListener)
{
MOZ_ASSERT(NS_IsMainThread());
MOZ_ASSERT(aListener);
MOZ_ASSERT(!mMainThreadListeners.Contains(aListener));
mMainThreadListeners.AppendElement(aListener);
// If it is not yet time to send the notification, then finish here.
if (!mFinishedNotificationSent) {
return;
}
class NotifyRunnable final : public Runnable
{
public:
explicit NotifyRunnable(MediaStream* aStream)
: Runnable("MediaStream::NotifyRunnable")
, mStream(aStream)
{}
NS_IMETHOD Run() override
{
MOZ_ASSERT(NS_IsMainThread());
mStream->NotifyMainThreadListeners();
return NS_OK;
}
private:
~NotifyRunnable() {}
RefPtr<MediaStream> mStream;
};
nsCOMPtr<nsIRunnable> runnable = new NotifyRunnable(this);
GraphImpl()->Dispatch(runnable.forget());
}
SourceMediaStream::SourceMediaStream()
: MediaStream()
, mMutex("mozilla::media::SourceMediaStream")
, mUpdateKnownTracksTime(0)
, mPullEnabled(false)
, mUpdateFinished(false)
, mNeedsMixing(false)
{
}
nsresult
SourceMediaStream::OpenAudioInput(int aID,
AudioDataListener *aListener)
{
if (GraphImpl()) {
mInputListener = aListener;
return GraphImpl()->OpenAudioInput(aID, aListener);
}
return NS_ERROR_FAILURE;
}
void
SourceMediaStream::CloseAudioInput()
{
// Destroy() may have run already and cleared this
if (GraphImpl() && mInputListener) {
GraphImpl()->CloseAudioInput(mInputListener);
}
mInputListener = nullptr;
}
void
SourceMediaStream::DestroyImpl()
{
CloseAudioInput();
GraphImpl()->AssertOnGraphThreadOrNotRunning();
for (int32_t i = mConsumers.Length() - 1; i >= 0; --i) {
// Disconnect before we come under mMutex's lock since it can call back
// through RemoveDirectTrackListenerImpl() and deadlock.
mConsumers[i]->Disconnect();
}
// Hold mMutex while mGraph is reset so that other threads holding mMutex
// can null-check know that the graph will not destroyed.
MutexAutoLock lock(mMutex);
MediaStream::DestroyImpl();
}
void
SourceMediaStream::SetPullEnabled(bool aEnabled)
{
MutexAutoLock lock(mMutex);
mPullEnabled = aEnabled;
if (mPullEnabled && GraphImpl()) {
GraphImpl()->EnsureNextIteration();
}
}
void
SourceMediaStream::AddTrackInternal(TrackID aID, TrackRate aRate, StreamTime aStart,
MediaSegment* aSegment, uint32_t aFlags)
{
MutexAutoLock lock(mMutex);
nsTArray<TrackData> *track_data = (aFlags & ADDTRACK_QUEUED) ?
&mPendingTracks : &mUpdateTracks;
TrackData* data = track_data->AppendElement();
LOG(LogLevel::Debug,
("AddTrackInternal: %lu/%lu",
(long)mPendingTracks.Length(),
(long)mUpdateTracks.Length()));
data->mID = aID;
data->mInputRate = aRate;
data->mResamplerChannelCount = 0;
data->mStart = aStart;
data->mEndOfFlushedData = aStart;
data->mCommands = TRACK_CREATE;
data->mData = aSegment;
ResampleAudioToGraphSampleRate(data, aSegment);
if (!(aFlags & ADDTRACK_QUEUED) && GraphImpl()) {
GraphImpl()->EnsureNextIteration();
}
}
void
SourceMediaStream::AddAudioTrack(TrackID aID, TrackRate aRate, StreamTime aStart,
AudioSegment* aSegment, uint32_t aFlags)
{
AddTrackInternal(aID, aRate, aStart, aSegment, aFlags);
}
void
SourceMediaStream::FinishAddTracks()
{
MutexAutoLock lock(mMutex);
mUpdateTracks.AppendElements(Move(mPendingTracks));
LOG(LogLevel::Debug,
("FinishAddTracks: %lu/%lu",
(long)mPendingTracks.Length(),
(long)mUpdateTracks.Length()));
if (GraphImpl()) {
GraphImpl()->EnsureNextIteration();
}
}
void
SourceMediaStream::ResampleAudioToGraphSampleRate(TrackData* aTrackData, MediaSegment* aSegment)
{
if (aSegment->GetType() != MediaSegment::AUDIO ||
aTrackData->mInputRate == GraphImpl()->GraphRate()) {
return;
}
AudioSegment* segment = static_cast<AudioSegment*>(aSegment);
int channels = segment->ChannelCount();
// If this segment is just silence, we delay instanciating the resampler. We
// also need to recreate the resampler if the channel count changes.
if (channels && aTrackData->mResamplerChannelCount != channels) {
SpeexResamplerState* state = speex_resampler_init(channels,
aTrackData->mInputRate,
GraphImpl()->GraphRate(),
SPEEX_RESAMPLER_QUALITY_MIN,
nullptr);
if (!state) {
return;
}
aTrackData->mResampler.own(state);
aTrackData->mResamplerChannelCount = channels;
}
segment->ResampleChunks(aTrackData->mResampler, aTrackData->mInputRate, GraphImpl()->GraphRate());
}
void
SourceMediaStream::AdvanceTimeVaryingValuesToCurrentTime(GraphTime aCurrentTime,
GraphTime aBlockedTime)
{
MutexAutoLock lock(mMutex);
mTracksStartTime += aBlockedTime;
mStreamTracksStartTimeStamp += TimeDuration::FromSeconds(GraphImpl()->MediaTimeToSeconds(aBlockedTime));
mTracks.ForgetUpTo(aCurrentTime - mTracksStartTime);
}
bool
SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment, MediaSegment *aRawSegment)
{
MutexAutoLock lock(mMutex);
// ::EndAllTrackAndFinished() can end these before the sources notice
bool appended = false;
auto graph = GraphImpl();
if (!mFinished && graph) {
TrackData *track = FindDataForTrack(aID);
if (track) {
// Data goes into mData, and on the next iteration of the MSG moves
// into the track's segment after NotifyQueuedTrackChanges(). This adds
// 0-10ms of delay before data gets to direct listeners.
// Indirect listeners (via subsequent TrackUnion nodes) are synced to
// playout time, and so can be delayed by buffering.
// Apply track disabling before notifying any consumers directly
// or inserting into the graph
ApplyTrackDisabling(aID, aSegment, aRawSegment);
ResampleAudioToGraphSampleRate(track, aSegment);
// Must notify first, since AppendFrom() will empty out aSegment
NotifyDirectConsumers(track, aRawSegment ? aRawSegment : aSegment);
track->mData->AppendFrom(aSegment); // note: aSegment is now dead
appended = true;
GraphImpl()->EnsureNextIteration();
} else {
aSegment->Clear();
}
}
return appended;
}
void
SourceMediaStream::NotifyDirectConsumers(TrackData *aTrack,
MediaSegment *aSegment)
{
mMutex.AssertCurrentThreadOwns();
MOZ_ASSERT(aTrack);
for (const TrackBound<DirectMediaStreamTrackListener>& source
: mDirectTrackListeners) {
if (aTrack->mID != source.mTrackID) {
continue;
}
StreamTime offset = 0; // FIX! need a separate StreamTime.... or the end of the internal buffer
source.mListener->NotifyRealtimeTrackDataAndApplyTrackDisabling(Graph(), offset, *aSegment);
}
}
// These handle notifying all the listeners of an event
void
SourceMediaStream::NotifyListenersEventImpl(MediaStreamGraphEvent aEvent)
{
for (uint32_t j = 0; j < mListeners.Length(); ++j) {
MediaStreamListener* l = mListeners[j];
l->NotifyEvent(GraphImpl(), aEvent);
}
}
void
SourceMediaStream::NotifyListenersEvent(MediaStreamGraphEvent aNewEvent)
{
class Message : public ControlMessage {
public:
Message(SourceMediaStream* aStream, MediaStreamGraphEvent aEvent) :
ControlMessage(aStream), mEvent(aEvent) {}
void Run() override
{
mStream->AsSourceStream()->NotifyListenersEventImpl(mEvent);
}
MediaStreamGraphEvent mEvent;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aNewEvent));
}
void
SourceMediaStream::AddDirectTrackListenerImpl(already_AddRefed<DirectMediaStreamTrackListener> aListener,
TrackID aTrackID)
{
MOZ_ASSERT(IsTrackIDExplicit(aTrackID));
MutexAutoLock lock(mMutex);
RefPtr<DirectMediaStreamTrackListener> listener = aListener;
LOG(LogLevel::Debug,
("Adding direct track listener %p bound to track %d to source stream %p",
listener.get(),
aTrackID,
this));
StreamTracks::Track* track = FindTrack(aTrackID);
if (!track) {
LOG(LogLevel::Warning,
("Couldn't find source track for direct track listener %p",
listener.get()));
listener->NotifyDirectListenerInstalled(
DirectMediaStreamTrackListener::InstallationResult::TRACK_NOT_FOUND_AT_SOURCE);
return;
}
bool isAudio = track->GetType() == MediaSegment::AUDIO;
bool isVideo = track->GetType() == MediaSegment::VIDEO;
if (!isAudio && !isVideo) {
LOG(
LogLevel::Warning,
("Source track for direct track listener %p is unknown", listener.get()));
MOZ_ASSERT(false);
return;
}
for (auto entry : mDirectTrackListeners) {
if (entry.mListener == listener &&
(entry.mTrackID == TRACK_ANY || entry.mTrackID == aTrackID)) {
listener->NotifyDirectListenerInstalled(
DirectMediaStreamTrackListener::InstallationResult::ALREADY_EXISTS);
return;
}
}
TrackBound<DirectMediaStreamTrackListener>* sourceListener =
mDirectTrackListeners.AppendElement();
sourceListener->mListener = listener;
sourceListener->mTrackID = aTrackID;
LOG(LogLevel::Debug, ("Added direct track listener %p", listener.get()));
listener->NotifyDirectListenerInstalled(
DirectMediaStreamTrackListener::InstallationResult::SUCCESS);
// Pass buffered data to the listener
AudioSegment bufferedAudio;
VideoSegment bufferedVideo;
MediaSegment& bufferedData =
isAudio ? static_cast<MediaSegment&>(bufferedAudio)
: static_cast<MediaSegment&>(bufferedVideo);
MediaSegment& trackSegment = *track->GetSegment();
if (mTracks.GetForgottenDuration() < trackSegment.GetDuration()) {
bufferedData.AppendSlice(trackSegment,
mTracks.GetForgottenDuration(),
trackSegment.GetDuration());
}
if (TrackData* updateData = FindDataForTrack(aTrackID)) {
bufferedData.AppendSlice(*updateData->mData, 0, updateData->mData->GetDuration());
}
if (bufferedData.GetDuration() != 0) {
listener->NotifyRealtimeTrackData(Graph(), 0, bufferedData);
}
}
void
SourceMediaStream::RemoveDirectTrackListenerImpl(DirectMediaStreamTrackListener* aListener,
TrackID aTrackID)
{
MutexAutoLock lock(mMutex);
for (int32_t i = mDirectTrackListeners.Length() - 1; i >= 0; --i) {
const TrackBound<DirectMediaStreamTrackListener>& source =
mDirectTrackListeners[i];
if (source.mListener == aListener && source.mTrackID == aTrackID) {
aListener->NotifyDirectListenerUninstalled();
mDirectTrackListeners.RemoveElementAt(i);
}
}
}
StreamTime
SourceMediaStream::GetEndOfAppendedData(TrackID aID)
{
MutexAutoLock lock(mMutex);
TrackData *track = FindDataForTrack(aID);
if (track) {
return track->mEndOfFlushedData + track->mData->GetDuration();
}
NS_ERROR("Track not found");
return 0;
}
void
SourceMediaStream::EndTrack(TrackID aID)
{
MutexAutoLock lock(mMutex);
TrackData *track = FindDataForTrack(aID);
if (track) {
track->mCommands |= TrackEventCommand::TRACK_EVENT_ENDED;
}
if (auto graph = GraphImpl()) {
graph->EnsureNextIteration();
}
}
void
SourceMediaStream::AdvanceKnownTracksTime(StreamTime aKnownTime)
{
MutexAutoLock lock(mMutex);
MOZ_ASSERT(aKnownTime >= mUpdateKnownTracksTime);
mUpdateKnownTracksTime = aKnownTime;
if (auto graph = GraphImpl()) {
graph->EnsureNextIteration();
}
}
void
SourceMediaStream::FinishWithLockHeld()
{
mMutex.AssertCurrentThreadOwns();
mUpdateFinished = true;
if (auto graph = GraphImpl()) {
graph->EnsureNextIteration();
}
}
void
SourceMediaStream::SetTrackEnabledImpl(TrackID aTrackID, DisabledTrackMode aMode)
{
{
MutexAutoLock lock(mMutex);
for (TrackBound<DirectMediaStreamTrackListener>& l: mDirectTrackListeners) {
if (l.mTrackID != aTrackID) {
continue;
}
DisabledTrackMode oldMode = GetDisabledTrackMode(aTrackID);
bool oldEnabled = oldMode == DisabledTrackMode::ENABLED;
if (!oldEnabled && aMode == DisabledTrackMode::ENABLED) {
LOG(LogLevel::Debug,
("SourceMediaStream %p track %d setting "
"direct listener enabled",
this,
aTrackID));
l.mListener->DecreaseDisabled(oldMode);
} else if (oldEnabled && aMode != DisabledTrackMode::ENABLED) {
LOG(LogLevel::Debug,
("SourceMediaStream %p track %d setting "
"direct listener disabled",
this,
aTrackID));
l.mListener->IncreaseDisabled(aMode);
}
}
}
MediaStream::SetTrackEnabledImpl(aTrackID, aMode);
}
void
SourceMediaStream::EndAllTrackAndFinish()
{
MutexAutoLock lock(mMutex);
for (uint32_t i = 0; i < mUpdateTracks.Length(); ++i) {
SourceMediaStream::TrackData* data = &mUpdateTracks[i];
data->mCommands |= TrackEventCommand::TRACK_EVENT_ENDED;
}
mPendingTracks.Clear();
FinishWithLockHeld();
// we will call NotifyEvent() to let GetUserMedia know
}
void
SourceMediaStream::RemoveAllDirectListeners()
{
GraphImpl()->AssertOnGraphThreadOrNotRunning();
auto directListeners(mDirectTrackListeners);
for (auto& l : directListeners) {
l.mListener->NotifyDirectListenerUninstalled();
}
mDirectTrackListeners.Clear();
}
SourceMediaStream::~SourceMediaStream()
{
}
void
SourceMediaStream::RegisterForAudioMixing()
{
MutexAutoLock lock(mMutex);
mNeedsMixing = true;
}
bool
SourceMediaStream::NeedsMixing()
{
MutexAutoLock lock(mMutex);
return mNeedsMixing;
}
bool
SourceMediaStream::HasPendingAudioTrack()
{
MutexAutoLock lock(mMutex);
bool audioTrackPresent = false;
for (auto& data : mPendingTracks) {
if (data.mData->GetType() == MediaSegment::AUDIO) {
audioTrackPresent = true;
break;
}
}
return audioTrackPresent;
}
bool
SourceMediaStream::OpenNewAudioCallbackDriver(AudioDataListener * aListener)
{
MOZ_ASSERT(GraphImpl()->mLifecycleState ==
MediaStreamGraphImpl::LifecycleState::LIFECYCLE_RUNNING);
AudioCallbackDriver* nextDriver = new AudioCallbackDriver(GraphImpl());
nextDriver->SetInputListener(aListener);
{
MonitorAutoLock lock(GraphImpl()->GetMonitor());
GraphImpl()->CurrentDriver()->SwitchAtNextIteration(nextDriver);
}
return true;
}
void
MediaInputPort::Init()
{
LOG(LogLevel::Debug,
("Adding MediaInputPort %p (from %p to %p) to the graph",
this,
mSource,
mDest));
mSource->AddConsumer(this);
mDest->AddInput(this);
// mPortCount decremented via MediaInputPort::Destroy's message
++mDest->GraphImpl()->mPortCount;
}
void
MediaInputPort::Disconnect()
{
GraphImpl()->AssertOnGraphThreadOrNotRunning();
NS_ASSERTION(!mSource == !mDest,
"mSource must either both be null or both non-null");
if (!mSource)
return;
mSource->RemoveConsumer(this);
mDest->RemoveInput(this);
mSource = nullptr;
mDest = nullptr;
GraphImpl()->SetStreamOrderDirty();
}
MediaInputPort::InputInterval
MediaInputPort::GetNextInputInterval(GraphTime aTime)
{
InputInterval result = { GRAPH_TIME_MAX, GRAPH_TIME_MAX, false };
if (aTime >= mDest->mStartBlocking) {
return result;
}
result.mStart = aTime;
result.mEnd = mDest->mStartBlocking;
result.mInputIsBlocked = aTime >= mSource->mStartBlocking;
if (!result.mInputIsBlocked) {
result.mEnd = std::min(result.mEnd, mSource->mStartBlocking);
}
return result;
}
void
MediaInputPort::Suspended()
{
mDest->InputSuspended(this);
}
void
MediaInputPort::Resumed()
{
mDest->InputResumed(this);
}
void
MediaInputPort::Destroy()
{
class Message : public ControlMessage {
public:
explicit Message(MediaInputPort* aPort)
: ControlMessage(nullptr), mPort(aPort) {}
void Run() override
{
mPort->Disconnect();
--mPort->GraphImpl()->mPortCount;
mPort->SetGraphImpl(nullptr);
NS_RELEASE(mPort);
}
void RunDuringShutdown() override
{
Run();
}
MediaInputPort* mPort;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this));
}
MediaStreamGraphImpl*
MediaInputPort::GraphImpl()
{
return mGraph;
}
MediaStreamGraph*
MediaInputPort::Graph()
{
return mGraph;
}
void
MediaInputPort::SetGraphImpl(MediaStreamGraphImpl* aGraph)
{
MOZ_ASSERT(!mGraph || !aGraph, "Should only be set once");
mGraph = aGraph;
}
void
MediaInputPort::BlockSourceTrackIdImpl(TrackID aTrackId, BlockingMode aBlockingMode)
{
mBlockedTracks.AppendElement(Pair<TrackID, BlockingMode>(aTrackId, aBlockingMode));
}
already_AddRefed<Pledge<bool>>
MediaInputPort::BlockSourceTrackId(TrackID aTrackId, BlockingMode aBlockingMode)
{
class Message : public ControlMessage {
public:
Message(MediaInputPort* aPort,
TrackID aTrackId,
BlockingMode aBlockingMode,
already_AddRefed<nsIRunnable> aRunnable)
: ControlMessage(aPort->GetDestination())
, mPort(aPort)
, mTrackId(aTrackId)
, mBlockingMode(aBlockingMode)
, mRunnable(aRunnable)
{
}
void Run() override
{
mPort->BlockSourceTrackIdImpl(mTrackId, mBlockingMode);
if (mRunnable) {
mStream->Graph()->DispatchToMainThreadAfterStreamStateUpdate(
mRunnable.forget());
}
}
void RunDuringShutdown() override
{
Run();
}
RefPtr<MediaInputPort> mPort;
TrackID mTrackId;
BlockingMode mBlockingMode;
nsCOMPtr<nsIRunnable> mRunnable;
};
MOZ_ASSERT(IsTrackIDExplicit(aTrackId),
"Only explicit TrackID is allowed");
auto pledge = MakeRefPtr<Pledge<bool>>();
nsCOMPtr<nsIRunnable> runnable = NewRunnableFrom([pledge]() {
MOZ_ASSERT(NS_IsMainThread());
pledge->Resolve(true);
return NS_OK;
});
GraphImpl()->AppendMessage(
MakeUnique<Message>(this, aTrackId, aBlockingMode, runnable.forget()));
return pledge.forget();
}
already_AddRefed<MediaInputPort>
ProcessedMediaStream::AllocateInputPort(MediaStream* aStream, TrackID aTrackID,
TrackID aDestTrackID,
uint16_t aInputNumber, uint16_t aOutputNumber,
nsTArray<TrackID>* aBlockedTracks)
{
// This method creates two references to the MediaInputPort: one for
// the main thread, and one for the MediaStreamGraph.
class Message : public ControlMessage {
public:
explicit Message(MediaInputPort* aPort)
: ControlMessage(aPort->GetDestination()),
mPort(aPort) {}
void Run() override
{
mPort->Init();
// The graph holds its reference implicitly
mPort->GraphImpl()->SetStreamOrderDirty();
Unused << mPort.forget();
}
void RunDuringShutdown() override
{
Run();
}
RefPtr<MediaInputPort> mPort;
};
MOZ_ASSERT(aStream->GraphImpl() == GraphImpl());
MOZ_ASSERT(aTrackID == TRACK_ANY || IsTrackIDExplicit(aTrackID),
"Only TRACK_ANY and explicit ID are allowed for source track");
MOZ_ASSERT(aDestTrackID == TRACK_ANY || IsTrackIDExplicit(aDestTrackID),
"Only TRACK_ANY and explicit ID are allowed for destination track");
MOZ_ASSERT(aTrackID != TRACK_ANY || aDestTrackID == TRACK_ANY,
"Generic MediaInputPort cannot produce a single destination track");
RefPtr<MediaInputPort> port = new MediaInputPort(
aStream, aTrackID, this, aDestTrackID, aInputNumber, aOutputNumber);
if (aBlockedTracks) {
for (TrackID trackID : *aBlockedTracks) {
port->BlockSourceTrackIdImpl(trackID, BlockingMode::CREATION);
}
}
port->SetGraphImpl(GraphImpl());
GraphImpl()->AppendMessage(MakeUnique<Message>(port));
return port.forget();
}
void
ProcessedMediaStream::Finish()
{
class Message : public ControlMessage {
public:
explicit Message(ProcessedMediaStream* aStream)
: ControlMessage(aStream) {}
void Run() override
{
mStream->GraphImpl()->FinishStream(mStream);
}
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this));
}
void
ProcessedMediaStream::SetAutofinish(bool aAutofinish)
{
class Message : public ControlMessage {
public:
Message(ProcessedMediaStream* aStream, bool aAutofinish)
: ControlMessage(aStream), mAutofinish(aAutofinish) {}
void Run() override
{
static_cast<ProcessedMediaStream*>(mStream)->SetAutofinishImpl(mAutofinish);
}
bool mAutofinish;
};
GraphImpl()->AppendMessage(MakeUnique<Message>(this, aAutofinish));
}
void
ProcessedMediaStream::DestroyImpl()
{
for (int32_t i = mInputs.Length() - 1; i >= 0; --i) {
mInputs[i]->Disconnect();
}
for (int32_t i = mSuspendedInputs.Length() - 1; i >= 0; --i) {
mSuspendedInputs[i]->Disconnect();
}
MediaStream::DestroyImpl();
// The stream order is only important if there are connections, in which
// case MediaInputPort::Disconnect() called SetStreamOrderDirty().
// MediaStreamGraphImpl::RemoveStreamGraphThread() will also call
// SetStreamOrderDirty(), for other reasons.
}
MediaStreamGraphImpl::MediaStreamGraphImpl(GraphDriverType aDriverRequested,
TrackRate aSampleRate,
AbstractThread* aMainThread)
: MediaStreamGraph(aSampleRate)
, mPortCount(0)
, mInputWanted(false)
, mInputDeviceID(-1)
, mOutputWanted(true)
, mOutputDeviceID(-1)
, mNeedAnotherIteration(false)
, mGraphDriverAsleep(false)
, mMonitor("MediaStreamGraphImpl")
, mLifecycleState(LIFECYCLE_THREAD_NOT_STARTED)
, mEndTime(GRAPH_TIME_MAX)
, mForceShutDown(false)
, mPostedRunInStableStateEvent(false)
, mDetectedNotRunning(false)
, mPostedRunInStableState(false)
, mRealtime(aDriverRequested != OFFLINE_THREAD_DRIVER)
, mNonRealtimeProcessing(false)
, mStreamOrderDirty(false)
, mLatencyLog(AsyncLatencyLogger::Get())
, mAbstractMainThread(aMainThread)
#ifdef MOZ_WEBRTC
, mFarendObserverRef(nullptr)
#endif
, mSelfRef(this)
, mOutputChannels(std::min<uint32_t>(8, CubebUtils::MaxNumberOfChannels()))
#ifdef DEBUG
, mCanRunMessagesSynchronously(false)
#endif
{
if (mRealtime) {
if (aDriverRequested == AUDIO_THREAD_DRIVER) {
AudioCallbackDriver* driver = new AudioCallbackDriver(this);
mDriver = driver;
} else {
mDriver = new SystemClockDriver(this);
}
} else {
mDriver = new OfflineClockDriver(this, MEDIA_GRAPH_TARGET_PERIOD_MS);
}
mLastMainThreadUpdate = TimeStamp::Now();
RegisterWeakAsyncMemoryReporter(this);
}
AbstractThread*
MediaStreamGraph::AbstractMainThread()
{
MOZ_ASSERT(static_cast<MediaStreamGraphImpl*>(this)->mAbstractMainThread);
return static_cast<MediaStreamGraphImpl*>(this)->mAbstractMainThread;
}
void
MediaStreamGraphImpl::Destroy()
{
// First unregister from memory reporting.
UnregisterWeakMemoryReporter(this);
// Clear the self reference which will destroy this instance.
mSelfRef = nullptr;
}
static
uint32_t WindowToHash(nsPIDOMWindowInner* aWindow)
{
uint32_t hashkey = 0;
hashkey = AddToHash(hashkey, aWindow);
return hashkey;
}
MediaStreamGraph*
MediaStreamGraph::GetInstance(MediaStreamGraph::GraphDriverType aGraphDriverRequested,
nsPIDOMWindowInner* aWindow)
{
NS_ASSERTION(NS_IsMainThread(), "Main thread only");
MediaStreamGraphImpl* graph = nullptr;
// We hash the nsPIDOMWindowInner to form a key to the gloabl
// MediaStreamGraph hashtable. Effectively, this means there is a graph per
// document.
uint32_t hashkey = WindowToHash(aWindow);
if (!gGraphs.Get(hashkey, &graph)) {
if (!gMediaStreamGraphShutdownBlocker) {
class Blocker : public media::ShutdownBlocker
{
public:
Blocker()
: media::ShutdownBlocker(NS_LITERAL_STRING(
"MediaStreamGraph shutdown: blocking on msg thread"))
{}
NS_IMETHOD
BlockShutdown(nsIAsyncShutdownClient* aProfileBeforeChange) override
{
// Distribute the global async shutdown blocker in a ticket. If there
// are zero graphs then shutdown is unblocked when we go out of scope.
auto ticket = MakeRefPtr<media::ShutdownTicket>(
gMediaStreamGraphShutdownBlocker.get());
gMediaStreamGraphShutdownBlocker = nullptr;
for (auto iter = gGraphs.Iter(); !iter.Done(); iter.Next()) {
iter.UserData()->ForceShutDown(ticket);
}
return NS_OK;
}
};
gMediaStreamGraphShutdownBlocker = new Blocker();
nsCOMPtr<nsIAsyncShutdownClient> barrier = media::GetShutdownBarrier();
nsresult rv = barrier->
AddBlocker(gMediaStreamGraphShutdownBlocker,
NS_LITERAL_STRING(__FILE__), __LINE__,
NS_LITERAL_STRING("MediaStreamGraph shutdown"));
MOZ_RELEASE_ASSERT(NS_SUCCEEDED(rv));
}
AbstractThread* mainThread;
if (aWindow) {
nsCOMPtr<nsIGlobalObject> parentObject = do_QueryInterface(aWindow);
mainThread = parentObject->AbstractMainThreadFor(TaskCategory::Other);
} else {
// Uncommon case, only for some old configuration of webspeech.
mainThread = AbstractThread::MainThread();
}
graph = new MediaStreamGraphImpl(aGraphDriverRequested,
CubebUtils::PreferredSampleRate(),
mainThread);
gGraphs.Put(hashkey, graph);
LOG(LogLevel::Debug,
("Starting up MediaStreamGraph %p for window %p", graph, aWindow));
}
return graph;
}
MediaStreamGraph*
MediaStreamGraph::CreateNonRealtimeInstance(TrackRate aSampleRate,
nsPIDOMWindowInner* aWindow)
{
NS_ASSERTION(NS_IsMainThread(), "Main thread only");
nsCOMPtr<nsIGlobalObject> parentObject = do_QueryInterface(aWindow);
MediaStreamGraphImpl* graph = new MediaStreamGraphImpl(
OFFLINE_THREAD_DRIVER,
aSampleRate,
parentObject->AbstractMainThreadFor(TaskCategory::Other));
LOG(LogLevel::Debug, ("Starting up Offline MediaStreamGraph %p", graph));
return graph;
}
void
MediaStreamGraph::DestroyNonRealtimeInstance(MediaStreamGraph* aGraph)
{
NS_ASSERTION(NS_IsMainThread(), "Main thread only");
MOZ_ASSERT(aGraph->IsNonRealtime(), "Should not destroy the global graph here");
MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(aGraph);
if (!graph->mNonRealtimeProcessing) {
// Start the graph, but don't produce anything
graph->StartNonRealtimeProcessing(0);
}
graph->ForceShutDown(nullptr);
}
NS_IMPL_ISUPPORTS(MediaStreamGraphImpl, nsIMemoryReporter, nsITimerCallback,
nsINamed)
NS_IMETHODIMP
MediaStreamGraphImpl::CollectReports(nsIHandleReportCallback* aHandleReport,
nsISupports* aData, bool aAnonymize)
{
if (mLifecycleState >= LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN) {
// Shutting down, nothing to report.
FinishCollectReports(aHandleReport, aData, nsTArray<AudioNodeSizes>());
return NS_OK;
}
class Message final : public ControlMessage {
public:
Message(MediaStreamGraphImpl *aGraph,
nsIHandleReportCallback* aHandleReport,
nsISupports *aHandlerData)
: ControlMessage(nullptr)
, mGraph(aGraph)
, mHandleReport(aHandleReport)
, mHandlerData(aHandlerData) {}
void Run() override
{
mGraph->CollectSizesForMemoryReport(mHandleReport.forget(),
mHandlerData.forget());
}
void RunDuringShutdown() override
{
// Run this message during shutdown too, so that endReports is called.
Run();
}
MediaStreamGraphImpl *mGraph;
// nsMemoryReporterManager keeps the callback and data alive only if it
// does not time out.
nsCOMPtr<nsIHandleReportCallback> mHandleReport;
nsCOMPtr<nsISupports> mHandlerData;
};
// When a non-realtime graph has not started, there is no thread yet, so
// collect sizes on this thread.
if (!(mRealtime || mNonRealtimeProcessing)) {
CollectSizesForMemoryReport(do_AddRef(aHandleReport), do_AddRef(aData));
return NS_OK;
}
AppendMessage(MakeUnique<Message>(this, aHandleReport, aData));
return NS_OK;
}
void
MediaStreamGraphImpl::CollectSizesForMemoryReport(
already_AddRefed<nsIHandleReportCallback> aHandleReport,
already_AddRefed<nsISupports> aHandlerData)
{
class FinishCollectRunnable final : public Runnable
{
public:
explicit FinishCollectRunnable(
already_AddRefed<nsIHandleReportCallback> aHandleReport,
already_AddRefed<nsISupports> aHandlerData)
: mozilla::Runnable("FinishCollectRunnable")
, mHandleReport(aHandleReport)
, mHandlerData(aHandlerData)
{}
NS_IMETHOD Run() override
{
MediaStreamGraphImpl::FinishCollectReports(mHandleReport, mHandlerData,
Move(mAudioStreamSizes));
return NS_OK;
}
nsTArray<AudioNodeSizes> mAudioStreamSizes;
private:
~FinishCollectRunnable() {}
// Avoiding nsCOMPtr because NSCAP_ASSERT_NO_QUERY_NEEDED in its
// constructor modifies the ref-count, which cannot be done off main
// thread.
RefPtr<nsIHandleReportCallback> mHandleReport;
RefPtr<nsISupports> mHandlerData;
};
RefPtr<FinishCollectRunnable> runnable =
new FinishCollectRunnable(Move(aHandleReport), Move(aHandlerData));
auto audioStreamSizes = &runnable->mAudioStreamSizes;
for (MediaStream* s : AllStreams()) {
AudioNodeStream* stream = s->AsAudioNodeStream();
if (stream) {
AudioNodeSizes* usage = audioStreamSizes->AppendElement();
stream->SizeOfAudioNodesIncludingThis(MallocSizeOf, *usage);
}
}
mAbstractMainThread->Dispatch(runnable.forget());
}
void
MediaStreamGraphImpl::
FinishCollectReports(nsIHandleReportCallback* aHandleReport, nsISupports* aData,
const nsTArray<AudioNodeSizes>& aAudioStreamSizes)
{
MOZ_ASSERT(NS_IsMainThread());
nsCOMPtr<nsIMemoryReporterManager> manager =
do_GetService("@mozilla.org/memory-reporter-manager;1");
if (!manager)
return;
#define REPORT(_path, _amount, _desc) \
aHandleReport->Callback(EmptyCString(), _path, KIND_HEAP, UNITS_BYTES, \
_amount, NS_LITERAL_CSTRING(_desc), aData);
for (size_t i = 0; i < aAudioStreamSizes.Length(); i++) {
const AudioNodeSizes& usage = aAudioStreamSizes[i];
const char* const nodeType =
usage.mNodeType ? usage.mNodeType : "<unknown>";
nsPrintfCString enginePath("explicit/webaudio/audio-node/%s/engine-objects",
nodeType);
REPORT(enginePath, usage.mEngine,
"Memory used by AudioNode engine objects (Web Audio).");
nsPrintfCString streamPath("explicit/webaudio/audio-node/%s/stream-objects",
nodeType);
REPORT(streamPath, usage.mStream,
"Memory used by AudioNode stream objects (Web Audio).");
}
size_t hrtfLoaders = WebCore::HRTFDatabaseLoader::sizeOfLoaders(MallocSizeOf);
if (hrtfLoaders) {
REPORT(NS_LITERAL_CSTRING(
"explicit/webaudio/audio-node/PannerNode/hrtf-databases"),
hrtfLoaders,
"Memory used by PannerNode databases (Web Audio).");
}
#undef REPORT
manager->EndReport();
}
SourceMediaStream*
MediaStreamGraph::CreateSourceStream()
{
SourceMediaStream* stream = new SourceMediaStream();
AddStream(stream);
return stream;
}
ProcessedMediaStream*
MediaStreamGraph::CreateTrackUnionStream()
{
TrackUnionStream* stream = new TrackUnionStream();
AddStream(stream);
return stream;
}
ProcessedMediaStream*
MediaStreamGraph::CreateAudioCaptureStream(TrackID aTrackId)
{
AudioCaptureStream* stream = new AudioCaptureStream(aTrackId);
AddStream(stream);
return stream;
}
void
MediaStreamGraph::AddStream(MediaStream* aStream)
{
NS_ADDREF(aStream);
MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
aStream->SetGraphImpl(graph);
graph->AppendMessage(MakeUnique<CreateMessage>(aStream));
}
class GraphStartedRunnable final : public Runnable
{
public:
GraphStartedRunnable(AudioNodeStream* aStream, MediaStreamGraph* aGraph)
: Runnable("GraphStartedRunnable")
, mStream(aStream)
, mGraph(aGraph)
{ }
NS_IMETHOD Run() override {
mGraph->NotifyWhenGraphStarted(mStream);
return NS_OK;
}
private:
RefPtr<AudioNodeStream> mStream;
MediaStreamGraph* mGraph;
};
void
MediaStreamGraph::NotifyWhenGraphStarted(AudioNodeStream* aStream)
{
MOZ_ASSERT(NS_IsMainThread());
class GraphStartedNotificationControlMessage : public ControlMessage
{
public:
explicit GraphStartedNotificationControlMessage(AudioNodeStream* aStream)
: ControlMessage(aStream)
{
}
void Run() override
{
// This runs on the graph thread, so when this runs, and the current
// driver is an AudioCallbackDriver, we know the audio hardware is
// started. If not, we are going to switch soon, keep reposting this
// ControlMessage.
MediaStreamGraphImpl* graphImpl = mStream->GraphImpl();
if (graphImpl->CurrentDriver()->AsAudioCallbackDriver()) {
nsCOMPtr<nsIRunnable> event = new dom::StateChangeTask(
mStream->AsAudioNodeStream(), nullptr, AudioContextState::Running);
graphImpl->Dispatch(event.forget());
} else {
nsCOMPtr<nsIRunnable> event = new GraphStartedRunnable(
mStream->AsAudioNodeStream(), mStream->Graph());
graphImpl->Dispatch(event.forget());
}
}
void RunDuringShutdown() override
{
}
};
if (!aStream->IsDestroyed()) {
MediaStreamGraphImpl* graphImpl = static_cast<MediaStreamGraphImpl*>(this);
graphImpl->AppendMessage(MakeUnique<GraphStartedNotificationControlMessage>(aStream));
}
}
void
MediaStreamGraphImpl::IncrementSuspendCount(MediaStream* aStream)
{
if (!aStream->IsSuspended()) {
MOZ_ASSERT(mStreams.Contains(aStream));
mStreams.RemoveElement(aStream);
mSuspendedStreams.AppendElement(aStream);
SetStreamOrderDirty();
}
aStream->IncrementSuspendCount();
}
void
MediaStreamGraphImpl::DecrementSuspendCount(MediaStream* aStream)
{
bool wasSuspended = aStream->IsSuspended();
aStream->DecrementSuspendCount();
if (wasSuspended && !aStream->IsSuspended()) {
MOZ_ASSERT(mSuspendedStreams.Contains(aStream));
mSuspendedStreams.RemoveElement(aStream);
mStreams.AppendElement(aStream);
ProcessedMediaStream* ps = aStream->AsProcessedStream();
if (ps) {
ps->mCycleMarker = NOT_VISITED;
}
SetStreamOrderDirty();
}
}
void
MediaStreamGraphImpl::SuspendOrResumeStreams(AudioContextOperation aAudioContextOperation,
const nsTArray<MediaStream*>& aStreamSet)
{
// For our purpose, Suspend and Close are equivalent: we want to remove the
// streams from the set of streams that are going to be processed.
for (MediaStream* stream : aStreamSet) {
if (aAudioContextOperation == AudioContextOperation::Resume) {
DecrementSuspendCount(stream);
} else {
IncrementSuspendCount(stream);
}
}
LOG(LogLevel::Debug,
("Moving streams between suspended and running"
"state: mStreams: %zu, mSuspendedStreams: %zu",
mStreams.Length(),
mSuspendedStreams.Length()));
#ifdef DEBUG
// The intersection of the two arrays should be null.
for (uint32_t i = 0; i < mStreams.Length(); i++) {
for (uint32_t j = 0; j < mSuspendedStreams.Length(); j++) {
MOZ_ASSERT(
mStreams[i] != mSuspendedStreams[j],
"The suspended stream set and running stream set are not disjoint.");
}
}
#endif
}
void
MediaStreamGraphImpl::AudioContextOperationCompleted(MediaStream* aStream,
void* aPromise,
AudioContextOperation aOperation)
{
// This can be called from the thread created to do cubeb operation, or the
// MSG thread. The pointers passed back here are refcounted, so are still
// alive.
MonitorAutoLock lock(mMonitor);
AudioContextState state;
switch (aOperation) {
case AudioContextOperation::Suspend:
state = AudioContextState::Suspended;
break;
case AudioContextOperation::Resume:
state = AudioContextState::Running;
break;
case AudioContextOperation::Close:
state = AudioContextState::Closed;
break;
default: MOZ_CRASH("Not handled.");
}
nsCOMPtr<nsIRunnable> event = new dom::StateChangeTask(
aStream->AsAudioNodeStream(), aPromise, state);
mAbstractMainThread->Dispatch(event.forget());
}
void
MediaStreamGraphImpl::ApplyAudioContextOperationImpl(
MediaStream* aDestinationStream, const nsTArray<MediaStream*>& aStreams,
AudioContextOperation aOperation, void* aPromise)
{
MOZ_ASSERT(CurrentDriver()->OnThread());
SuspendOrResumeStreams(aOperation, aStreams);
bool switching = false;
GraphDriver* nextDriver = nullptr;
{
MonitorAutoLock lock(mMonitor);
switching = CurrentDriver()->Switching();
if (switching) {
nextDriver = CurrentDriver()->NextDriver();
}
}
// If we have suspended the last AudioContext, and we don't have other
// streams that have audio, this graph will automatically switch to a
// SystemCallbackDriver, because it can't find a MediaStream that has an audio
// track. When resuming, force switching to an AudioCallbackDriver (if we're
// not already switching). It would have happened at the next iteration
// anyways, but doing this now save some time.
if (aOperation == AudioContextOperation::Resume) {
if (!CurrentDriver()->AsAudioCallbackDriver()) {
AudioCallbackDriver* driver;
if (switching) {
MOZ_ASSERT(nextDriver->AsAudioCallbackDriver());
driver = nextDriver->AsAudioCallbackDriver();
} else {
driver = new AudioCallbackDriver(this);
MonitorAutoLock lock(mMonitor);
CurrentDriver()->SwitchAtNextIteration(driver);
}
driver->EnqueueStreamAndPromiseForOperation(aDestinationStream,
aPromise, aOperation);
} else {
// We are resuming a context, but we are already using an
// AudioCallbackDriver, we can resolve the promise now.
AudioContextOperationCompleted(aDestinationStream, aPromise, aOperation);
}
}
// Close, suspend: check if we are going to switch to a
// SystemAudioCallbackDriver, and pass the promise to the AudioCallbackDriver
// if that's the case, so it can notify the content.
// This is the same logic as in UpdateStreamOrder, but it's simpler to have it
// here as well so we don't have to store the Promise(s) on the Graph.
if (aOperation != AudioContextOperation::Resume) {
bool shouldAEC = false;
bool audioTrackPresent = AudioTrackPresent(shouldAEC);
if (!audioTrackPresent && CurrentDriver()->AsAudioCallbackDriver()) {
CurrentDriver()->AsAudioCallbackDriver()->
EnqueueStreamAndPromiseForOperation(aDestinationStream, aPromise,
aOperation);
SystemClockDriver* driver;
if (nextDriver) {
MOZ_ASSERT(!nextDriver->AsAudioCallbackDriver());
} else {
driver = new SystemClockDriver(this);
MonitorAutoLock lock(mMonitor);
CurrentDriver()->SwitchAtNextIteration(driver);
}
// We are closing or suspending an AudioContext, but we just got resumed.
// Queue the operation on the next driver so that the ordering is
// preserved.
} else if (!audioTrackPresent && switching) {
MOZ_ASSERT(nextDriver->AsAudioCallbackDriver() ||
nextDriver->AsSystemClockDriver()->IsFallback());
if (nextDriver->AsAudioCallbackDriver()) {
nextDriver->AsAudioCallbackDriver()->
EnqueueStreamAndPromiseForOperation(aDestinationStream, aPromise,
aOperation);
} else {
// If this is not an AudioCallbackDriver, this means we failed opening an
// AudioCallbackDriver in the past, and we're constantly trying to re-open
// an new audio stream, but are running this graph that has an audio track
// off a SystemClockDriver for now to keep things moving. This is the
// case where we're trying to switch an an system driver (because suspend
// or close have been called on an AudioContext, or we've closed the
// page), but we're already running one. We can just resolve the promise
// now: we're already running off a system thread.
AudioContextOperationCompleted(aDestinationStream, aPromise, aOperation);
}
} else {
// We are closing or suspending an AudioContext, but something else is
// using the audio stream, we can resolve the promise now.
AudioContextOperationCompleted(aDestinationStream, aPromise, aOperation);
}
}
}
void
MediaStreamGraph::ApplyAudioContextOperation(MediaStream* aDestinationStream,
const nsTArray<MediaStream*>& aStreams,
AudioContextOperation aOperation,
void* aPromise)
{
class AudioContextOperationControlMessage : public ControlMessage
{
public:
AudioContextOperationControlMessage(MediaStream* aDestinationStream,
const nsTArray<MediaStream*>& aStreams,
AudioContextOperation aOperation,
void* aPromise)
: ControlMessage(aDestinationStream)
, mStreams(aStreams)
, mAudioContextOperation(aOperation)
, mPromise(aPromise)
{
}
void Run() override
{
mStream->GraphImpl()->ApplyAudioContextOperationImpl(mStream,
mStreams, mAudioContextOperation, mPromise);
}
void RunDuringShutdown() override
{
MOZ_ASSERT(mAudioContextOperation == AudioContextOperation::Close,
"We should be reviving the graph?");
}
private:
// We don't need strong references here for the same reason ControlMessage
// doesn't.
nsTArray<MediaStream*> mStreams;
AudioContextOperation mAudioContextOperation;
void* mPromise;
};
MediaStreamGraphImpl* graphImpl = static_cast<MediaStreamGraphImpl*>(this);
graphImpl->AppendMessage(
MakeUnique<AudioContextOperationControlMessage>(aDestinationStream, aStreams,
aOperation, aPromise));
}
bool
MediaStreamGraph::IsNonRealtime() const
{
return !static_cast<const MediaStreamGraphImpl*>(this)->mRealtime;
}
void
MediaStreamGraph::StartNonRealtimeProcessing(uint32_t aTicksToProcess)
{
NS_ASSERTION(NS_IsMainThread(), "main thread only");
MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
NS_ASSERTION(!graph->mRealtime, "non-realtime only");
if (graph->mNonRealtimeProcessing)
return;
graph->mEndTime =
graph->RoundUpToNextAudioBlock(graph->mStateComputedTime +
aTicksToProcess - 1);
graph->mNonRealtimeProcessing = true;
graph->EnsureRunInStableState();
}
void
ProcessedMediaStream::AddInput(MediaInputPort* aPort)
{
MediaStream* s = aPort->GetSource();
if (!s->IsSuspended()) {
mInputs.AppendElement(aPort);
} else {
mSuspendedInputs.AppendElement(aPort);
}
GraphImpl()->SetStreamOrderDirty();
}
void
ProcessedMediaStream::InputSuspended(MediaInputPort* aPort)
{
GraphImpl()->AssertOnGraphThreadOrNotRunning();
mInputs.RemoveElement(aPort);
mSuspendedInputs.AppendElement(aPort);
GraphImpl()->SetStreamOrderDirty();
}
void
ProcessedMediaStream::InputResumed(MediaInputPort* aPort)
{
GraphImpl()->AssertOnGraphThreadOrNotRunning();
mSuspendedInputs.RemoveElement(aPort);
mInputs.AppendElement(aPort);
GraphImpl()->SetStreamOrderDirty();
}
void
MediaStreamGraph::RegisterCaptureStreamForWindow(
uint64_t aWindowId, ProcessedMediaStream* aCaptureStream)
{
MOZ_ASSERT(NS_IsMainThread());
MediaStreamGraphImpl* graphImpl = static_cast<MediaStreamGraphImpl*>(this);
graphImpl->RegisterCaptureStreamForWindow(aWindowId, aCaptureStream);
}
void
MediaStreamGraphImpl::RegisterCaptureStreamForWindow(
uint64_t aWindowId, ProcessedMediaStream* aCaptureStream)
{
MOZ_ASSERT(NS_IsMainThread());
WindowAndStream winAndStream;
winAndStream.mWindowId = aWindowId;
winAndStream.mCaptureStreamSink = aCaptureStream;
mWindowCaptureStreams.AppendElement(winAndStream);
}
void
MediaStreamGraph::UnregisterCaptureStreamForWindow(uint64_t aWindowId)
{
MOZ_ASSERT(NS_IsMainThread());
MediaStreamGraphImpl* graphImpl = static_cast<MediaStreamGraphImpl*>(this);
graphImpl->UnregisterCaptureStreamForWindow(aWindowId);
}
void
MediaStreamGraphImpl::UnregisterCaptureStreamForWindow(uint64_t aWindowId)
{
MOZ_ASSERT(NS_IsMainThread());
for (int32_t i = mWindowCaptureStreams.Length() - 1; i >= 0; i--) {
if (mWindowCaptureStreams[i].mWindowId == aWindowId) {
mWindowCaptureStreams.RemoveElementAt(i);
}
}
}
already_AddRefed<MediaInputPort>
MediaStreamGraph::ConnectToCaptureStream(uint64_t aWindowId,
MediaStream* aMediaStream)
{
return aMediaStream->GraphImpl()->ConnectToCaptureStream(aWindowId,
aMediaStream);
}
already_AddRefed<MediaInputPort>
MediaStreamGraphImpl::ConnectToCaptureStream(uint64_t aWindowId,
MediaStream* aMediaStream)
{
MOZ_ASSERT(NS_IsMainThread());
for (uint32_t i = 0; i < mWindowCaptureStreams.Length(); i++) {
if (mWindowCaptureStreams[i].mWindowId == aWindowId) {
ProcessedMediaStream* sink = mWindowCaptureStreams[i].mCaptureStreamSink;
return sink->AllocateInputPort(aMediaStream);
}
}
return nullptr;
}
void
MediaStreamGraph::DispatchToMainThreadAfterStreamStateUpdate(
already_AddRefed<nsIRunnable> aRunnable)
{
AssertOnGraphThreadOrNotRunning();
*mPendingUpdateRunnables.AppendElement() =
AbstractMainThread()->CreateDirectTaskDrainer(Move(aRunnable));
}
} // namespace mozilla