michael@0: /* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*-*/
michael@0: /* This Source Code Form is subject to the terms of the Mozilla Public
michael@0:  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
michael@0:  * You can obtain one at http://mozilla.org/MPL/2.0/. */
michael@0: 
michael@0: #include "MediaStreamGraphImpl.h"
michael@0: #include "mozilla/LinkedList.h"
michael@0: #include "mozilla/MathAlgorithms.h"
michael@0: #include "mozilla/unused.h"
michael@0: 
michael@0: #include "AudioSegment.h"
michael@0: #include "VideoSegment.h"
michael@0: #include "nsContentUtils.h"
michael@0: #include "nsIAppShell.h"
michael@0: #include "nsIObserver.h"
michael@0: #include "nsPrintfCString.h"
michael@0: #include "nsServiceManagerUtils.h"
michael@0: #include "nsWidgetsCID.h"
michael@0: #include "prerror.h"
michael@0: #include "prlog.h"
michael@0: #include "mozilla/Attributes.h"
michael@0: #include "TrackUnionStream.h"
michael@0: #include "ImageContainer.h"
michael@0: #include "AudioChannelService.h"
michael@0: #include "AudioNodeEngine.h"
michael@0: #include "AudioNodeStream.h"
michael@0: #include "AudioNodeExternalInputStream.h"
michael@0: #include <algorithm>
michael@0: #include "DOMMediaStream.h"
michael@0: #include "GeckoProfiler.h"
michael@0: #include "mozilla/unused.h"
michael@0: #include "speex/speex_resampler.h"
michael@0: #ifdef MOZ_WEBRTC
michael@0: #include "AudioOutputObserver.h"
michael@0: #endif
michael@0: 
michael@0: using namespace mozilla::layers;
michael@0: using namespace mozilla::dom;
michael@0: using namespace mozilla::gfx;
michael@0: 
michael@0: namespace mozilla {
michael@0: 
michael@0: #ifdef PR_LOGGING
michael@0: PRLogModuleInfo* gMediaStreamGraphLog;
michael@0: #define STREAM_LOG(type, msg) PR_LOG(gMediaStreamGraphLog, type, msg)
michael@0: #else
michael@0: #define STREAM_LOG(type, msg)
michael@0: #endif
michael@0: 
michael@0: /**
michael@0:  * The singleton graph instance.
michael@0:  */
michael@0: static MediaStreamGraphImpl* gGraph;
michael@0: 
michael@0: MediaStreamGraphImpl::~MediaStreamGraphImpl()
michael@0: {
michael@0:   NS_ASSERTION(IsEmpty(),
michael@0:                "All streams should have been destroyed by messages from the main thread");
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("MediaStreamGraph %p destroyed", this));
michael@0: }
michael@0: 
michael@0: 
michael@0: StreamTime
michael@0: MediaStreamGraphImpl::GetDesiredBufferEnd(MediaStream* aStream)
michael@0: {
michael@0:   StreamTime current = mCurrentTime - aStream->mBufferStartTime;
michael@0:   // When waking up media decoders, we need a longer safety margin, as it can
michael@0:   // take more time to get new samples. A factor of two seem to work.
michael@0:   return current +
michael@0:       2 * MillisecondsToMediaTime(std::max(AUDIO_TARGET_MS, VIDEO_TARGET_MS));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::FinishStream(MediaStream* aStream)
michael@0: {
michael@0:   if (aStream->mFinished)
michael@0:     return;
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("MediaStream %p will finish", aStream));
michael@0:   aStream->mFinished = true;
michael@0:   aStream->mBuffer.AdvanceKnownTracksTime(STREAM_TIME_MAX);
michael@0:   // Force at least one more iteration of the control loop, since we rely
michael@0:   // on UpdateCurrentTime to notify our listeners once the stream end
michael@0:   // has been reached.
michael@0:   EnsureNextIteration();
michael@0: 
michael@0:   SetStreamOrderDirty();
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::AddStream(MediaStream* aStream)
michael@0: {
michael@0:   aStream->mBufferStartTime = mCurrentTime;
michael@0:   *mStreams.AppendElement() = already_AddRefed<MediaStream>(aStream);
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("Adding media stream %p to the graph", aStream));
michael@0: 
michael@0:   SetStreamOrderDirty();
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::RemoveStream(MediaStream* aStream)
michael@0: {
michael@0:   // Remove references in mStreamUpdates before we allow aStream to die.
michael@0:   // Pending updates are not needed (since the main thread has already given
michael@0:   // up the stream) so we will just drop them.
michael@0:   {
michael@0:     MonitorAutoLock lock(mMonitor);
michael@0:     for (uint32_t i = 0; i < mStreamUpdates.Length(); ++i) {
michael@0:       if (mStreamUpdates[i].mStream == aStream) {
michael@0:         mStreamUpdates[i].mStream = nullptr;
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // Ensure that mMixer is updated when necessary.
michael@0:   SetStreamOrderDirty();
michael@0: 
michael@0:   // This unrefs the stream, probably destroying it
michael@0:   mStreams.RemoveElement(aStream);
michael@0: 
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("Removing media stream %p from the graph", aStream));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::UpdateConsumptionState(SourceMediaStream* aStream)
michael@0: {
michael@0:   MediaStreamListener::Consumption state =
michael@0:       aStream->mIsConsumed ? MediaStreamListener::CONSUMED
michael@0:       : MediaStreamListener::NOT_CONSUMED;
michael@0:   if (state != aStream->mLastConsumptionState) {
michael@0:     aStream->mLastConsumptionState = state;
michael@0:     for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
michael@0:       MediaStreamListener* l = aStream->mListeners[j];
michael@0:       l->NotifyConsumptionChanged(this, state);
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ExtractPendingInput(SourceMediaStream* aStream,
michael@0:                                           GraphTime aDesiredUpToTime,
michael@0:                                           bool* aEnsureNextIteration)
michael@0: {
michael@0:   bool finished;
michael@0:   {
michael@0:     MutexAutoLock lock(aStream->mMutex);
michael@0:     if (aStream->mPullEnabled && !aStream->mFinished &&
michael@0:         !aStream->mListeners.IsEmpty()) {
michael@0:       // Compute how much stream time we'll need assuming we don't block
michael@0:       // the stream at all between mBlockingDecisionsMadeUntilTime and
michael@0:       // aDesiredUpToTime.
michael@0:       StreamTime t =
michael@0:         GraphTimeToStreamTime(aStream, mStateComputedTime) +
michael@0:         (aDesiredUpToTime - mStateComputedTime);
michael@0:       STREAM_LOG(PR_LOG_DEBUG+1, ("Calling NotifyPull aStream=%p t=%f current end=%f", aStream,
michael@0:                                   MediaTimeToSeconds(t),
michael@0:                                   MediaTimeToSeconds(aStream->mBuffer.GetEnd())));
michael@0:       if (t > aStream->mBuffer.GetEnd()) {
michael@0:         *aEnsureNextIteration = true;
michael@0: #ifdef DEBUG
michael@0:         if (aStream->mListeners.Length() == 0) {
michael@0:           STREAM_LOG(PR_LOG_ERROR, ("No listeners in NotifyPull aStream=%p desired=%f current end=%f",
michael@0:                                     aStream, MediaTimeToSeconds(t),
michael@0:                                     MediaTimeToSeconds(aStream->mBuffer.GetEnd())));
michael@0:           aStream->DumpTrackInfo();
michael@0:         }
michael@0: #endif
michael@0:         for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
michael@0:           MediaStreamListener* l = aStream->mListeners[j];
michael@0:           {
michael@0:             MutexAutoUnlock unlock(aStream->mMutex);
michael@0:             l->NotifyPull(this, t);
michael@0:           }
michael@0:         }
michael@0:       }
michael@0:     }
michael@0:     finished = aStream->mUpdateFinished;
michael@0:     for (int32_t i = aStream->mUpdateTracks.Length() - 1; i >= 0; --i) {
michael@0:       SourceMediaStream::TrackData* data = &aStream->mUpdateTracks[i];
michael@0:       aStream->ApplyTrackDisabling(data->mID, data->mData);
michael@0:       for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
michael@0:         MediaStreamListener* l = aStream->mListeners[j];
michael@0:         TrackTicks offset = (data->mCommands & SourceMediaStream::TRACK_CREATE)
michael@0:             ? data->mStart : aStream->mBuffer.FindTrack(data->mID)->GetSegment()->GetDuration();
michael@0:         l->NotifyQueuedTrackChanges(this, data->mID, data->mOutputRate,
michael@0:                                     offset, data->mCommands, *data->mData);
michael@0:       }
michael@0:       if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
michael@0:         MediaSegment* segment = data->mData.forget();
michael@0:         STREAM_LOG(PR_LOG_DEBUG, ("SourceMediaStream %p creating track %d, rate %d, start %lld, initial end %lld",
michael@0:                                   aStream, data->mID, data->mOutputRate, int64_t(data->mStart),
michael@0:                                   int64_t(segment->GetDuration())));
michael@0: 
michael@0:         aStream->mBuffer.AddTrack(data->mID, data->mOutputRate, data->mStart, segment);
michael@0:         // The track has taken ownership of data->mData, so let's replace
michael@0:         // data->mData with an empty clone.
michael@0:         data->mData = segment->CreateEmptyClone();
michael@0:         data->mCommands &= ~SourceMediaStream::TRACK_CREATE;
michael@0:       } else if (data->mData->GetDuration() > 0) {
michael@0:         MediaSegment* dest = aStream->mBuffer.FindTrack(data->mID)->GetSegment();
michael@0:         STREAM_LOG(PR_LOG_DEBUG+1, ("SourceMediaStream %p track %d, advancing end from %lld to %lld",
michael@0:                                     aStream, data->mID,
michael@0:                                     int64_t(dest->GetDuration()),
michael@0:                                     int64_t(dest->GetDuration() + data->mData->GetDuration())));
michael@0:         dest->AppendFrom(data->mData);
michael@0:       }
michael@0:       if (data->mCommands & SourceMediaStream::TRACK_END) {
michael@0:         aStream->mBuffer.FindTrack(data->mID)->SetEnded();
michael@0:         aStream->mUpdateTracks.RemoveElementAt(i);
michael@0:       }
michael@0:     }
michael@0:     if (!aStream->mFinished) {
michael@0:       aStream->mBuffer.AdvanceKnownTracksTime(aStream->mUpdateKnownTracksTime);
michael@0:     }
michael@0:   }
michael@0:   if (aStream->mBuffer.GetEnd() > 0) {
michael@0:     aStream->mHasCurrentData = true;
michael@0:   }
michael@0:   if (finished) {
michael@0:     FinishStream(aStream);
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::UpdateBufferSufficiencyState(SourceMediaStream* aStream)
michael@0: {
michael@0:   StreamTime desiredEnd = GetDesiredBufferEnd(aStream);
michael@0:   nsTArray<SourceMediaStream::ThreadAndRunnable> runnables;
michael@0: 
michael@0:   {
michael@0:     MutexAutoLock lock(aStream->mMutex);
michael@0:     for (uint32_t i = 0; i < aStream->mUpdateTracks.Length(); ++i) {
michael@0:       SourceMediaStream::TrackData* data = &aStream->mUpdateTracks[i];
michael@0:       if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
michael@0:         // This track hasn't been created yet, so we have no sufficiency
michael@0:         // data. The track will be created in the next iteration of the
michael@0:         // control loop and then we'll fire insufficiency notifications
michael@0:         // if necessary.
michael@0:         continue;
michael@0:       }
michael@0:       if (data->mCommands & SourceMediaStream::TRACK_END) {
michael@0:         // This track will end, so no point in firing not-enough-data
michael@0:         // callbacks.
michael@0:         continue;
michael@0:       }
michael@0:       StreamBuffer::Track* track = aStream->mBuffer.FindTrack(data->mID);
michael@0:       // Note that track->IsEnded() must be false, otherwise we would have
michael@0:       // removed the track from mUpdateTracks already.
michael@0:       NS_ASSERTION(!track->IsEnded(), "What is this track doing here?");
michael@0:       data->mHaveEnough = track->GetEndTimeRoundDown() >= desiredEnd;
michael@0:       if (!data->mHaveEnough) {
michael@0:         runnables.MoveElementsFrom(data->mDispatchWhenNotEnough);
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   for (uint32_t i = 0; i < runnables.Length(); ++i) {
michael@0:     runnables[i].mTarget->Dispatch(runnables[i].mRunnable, 0);
michael@0:   }
michael@0: }
michael@0: 
michael@0: StreamTime
michael@0: MediaStreamGraphImpl::GraphTimeToStreamTime(MediaStream* aStream,
michael@0:                                             GraphTime aTime)
michael@0: {
michael@0:   NS_ASSERTION(aTime <= mStateComputedTime,
michael@0:                "Don't ask about times where we haven't made blocking decisions yet");
michael@0:   if (aTime <= mCurrentTime) {
michael@0:     return std::max<StreamTime>(0, aTime - aStream->mBufferStartTime);
michael@0:   }
michael@0:   GraphTime t = mCurrentTime;
michael@0:   StreamTime s = t - aStream->mBufferStartTime;
michael@0:   while (t < aTime) {
michael@0:     GraphTime end;
michael@0:     if (!aStream->mBlocked.GetAt(t, &end)) {
michael@0:       s += std::min(aTime, end) - t;
michael@0:     }
michael@0:     t = end;
michael@0:   }
michael@0:   return std::max<StreamTime>(0, s);
michael@0: }
michael@0: 
michael@0: StreamTime
michael@0: MediaStreamGraphImpl::GraphTimeToStreamTimeOptimistic(MediaStream* aStream,
michael@0:                                                       GraphTime aTime)
michael@0: {
michael@0:   GraphTime computedUpToTime = std::min(mStateComputedTime, aTime);
michael@0:   StreamTime s = GraphTimeToStreamTime(aStream, computedUpToTime);
michael@0:   return s + (aTime - computedUpToTime);
michael@0: }
michael@0: 
michael@0: GraphTime
michael@0: MediaStreamGraphImpl::StreamTimeToGraphTime(MediaStream* aStream,
michael@0:                                             StreamTime aTime, uint32_t aFlags)
michael@0: {
michael@0:   if (aTime >= STREAM_TIME_MAX) {
michael@0:     return GRAPH_TIME_MAX;
michael@0:   }
michael@0:   MediaTime bufferElapsedToCurrentTime = mCurrentTime - aStream->mBufferStartTime;
michael@0:   if (aTime < bufferElapsedToCurrentTime ||
michael@0:       (aTime == bufferElapsedToCurrentTime && !(aFlags & INCLUDE_TRAILING_BLOCKED_INTERVAL))) {
michael@0:     return aTime + aStream->mBufferStartTime;
michael@0:   }
michael@0: 
michael@0:   MediaTime streamAmount = aTime - bufferElapsedToCurrentTime;
michael@0:   NS_ASSERTION(streamAmount >= 0, "Can't answer queries before current time");
michael@0: 
michael@0:   GraphTime t = mCurrentTime;
michael@0:   while (t < GRAPH_TIME_MAX) {
michael@0:     if (!(aFlags & INCLUDE_TRAILING_BLOCKED_INTERVAL) && streamAmount == 0) {
michael@0:       return t;
michael@0:     }
michael@0:     bool blocked;
michael@0:     GraphTime end;
michael@0:     if (t < mStateComputedTime) {
michael@0:       blocked = aStream->mBlocked.GetAt(t, &end);
michael@0:       end = std::min(end, mStateComputedTime);
michael@0:     } else {
michael@0:       blocked = false;
michael@0:       end = GRAPH_TIME_MAX;
michael@0:     }
michael@0:     if (blocked) {
michael@0:       t = end;
michael@0:     } else {
michael@0:       if (streamAmount == 0) {
michael@0:         // No more stream time to consume at time t, so we're done.
michael@0:         break;
michael@0:       }
michael@0:       MediaTime consume = std::min(end - t, streamAmount);
michael@0:       streamAmount -= consume;
michael@0:       t += consume;
michael@0:     }
michael@0:   }
michael@0:   return t;
michael@0: }
michael@0: 
michael@0: GraphTime
michael@0: MediaStreamGraphImpl::GetAudioPosition(MediaStream* aStream)
michael@0: {
michael@0:   if (aStream->mAudioOutputStreams.IsEmpty()) {
michael@0:     return mCurrentTime;
michael@0:   }
michael@0:   int64_t positionInFrames = aStream->mAudioOutputStreams[0].mStream->GetPositionInFrames();
michael@0:   if (positionInFrames < 0) {
michael@0:     return mCurrentTime;
michael@0:   }
michael@0:   return aStream->mAudioOutputStreams[0].mAudioPlaybackStartTime +
michael@0:       TicksToTimeRoundDown(mSampleRate,
michael@0:                            positionInFrames);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::UpdateCurrentTime()
michael@0: {
michael@0:   GraphTime prevCurrentTime, nextCurrentTime;
michael@0:   if (mRealtime) {
michael@0:     TimeStamp now = TimeStamp::Now();
michael@0:     prevCurrentTime = mCurrentTime;
michael@0:     nextCurrentTime =
michael@0:       SecondsToMediaTime((now - mCurrentTimeStamp).ToSeconds()) + mCurrentTime;
michael@0: 
michael@0:     mCurrentTimeStamp = now;
michael@0:     STREAM_LOG(PR_LOG_DEBUG+1, ("Updating current time to %f (real %f, mStateComputedTime %f)",
michael@0:                MediaTimeToSeconds(nextCurrentTime),
michael@0:                (now - mInitialTimeStamp).ToSeconds(),
michael@0:                MediaTimeToSeconds(mStateComputedTime)));
michael@0:   } else {
michael@0:     prevCurrentTime = mCurrentTime;
michael@0:     nextCurrentTime = mCurrentTime + MillisecondsToMediaTime(MEDIA_GRAPH_TARGET_PERIOD_MS);
michael@0:     STREAM_LOG(PR_LOG_DEBUG+1, ("Updating offline current time to %f (mStateComputedTime %f)",
michael@0:                MediaTimeToSeconds(nextCurrentTime),
michael@0:                MediaTimeToSeconds(mStateComputedTime)));
michael@0:   }
michael@0: 
michael@0:   if (mStateComputedTime < nextCurrentTime) {
michael@0:     STREAM_LOG(PR_LOG_WARNING, ("Media graph global underrun detected"));
michael@0:     nextCurrentTime = mStateComputedTime;
michael@0:   }
michael@0: 
michael@0:   if (prevCurrentTime >= nextCurrentTime) {
michael@0:     NS_ASSERTION(prevCurrentTime == nextCurrentTime, "Time can't go backwards!");
michael@0:     // This could happen due to low clock resolution, maybe?
michael@0:     STREAM_LOG(PR_LOG_DEBUG, ("Time did not advance"));
michael@0:     // There's not much left to do here, but the code below that notifies
michael@0:     // listeners that streams have ended still needs to run.
michael@0:   }
michael@0: 
michael@0:   nsTArray<MediaStream*> streamsReadyToFinish;
michael@0:   nsAutoTArray<bool,800> streamHasOutput;
michael@0:   streamHasOutput.SetLength(mStreams.Length());
michael@0:   for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:     MediaStream* stream = mStreams[i];
michael@0: 
michael@0:     // Calculate blocked time and fire Blocked/Unblocked events
michael@0:     GraphTime blockedTime = 0;
michael@0:     GraphTime t = prevCurrentTime;
michael@0:     // include |nextCurrentTime| to ensure NotifyBlockingChanged() is called
michael@0:     // before NotifyFinished() when |nextCurrentTime == stream end time|
michael@0:     while (t <= nextCurrentTime) {
michael@0:       GraphTime end;
michael@0:       bool blocked = stream->mBlocked.GetAt(t, &end);
michael@0:       if (blocked) {
michael@0:         blockedTime += std::min(end, nextCurrentTime) - t;
michael@0:       }
michael@0:       if (blocked != stream->mNotifiedBlocked) {
michael@0:         for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
michael@0:           MediaStreamListener* l = stream->mListeners[j];
michael@0:           l->NotifyBlockingChanged(this,
michael@0:               blocked ? MediaStreamListener::BLOCKED : MediaStreamListener::UNBLOCKED);
michael@0:         }
michael@0:         stream->mNotifiedBlocked = blocked;
michael@0:       }
michael@0:       t = end;
michael@0:     }
michael@0: 
michael@0:     stream->AdvanceTimeVaryingValuesToCurrentTime(nextCurrentTime, blockedTime);
michael@0:     // Advance mBlocked last so that implementations of
michael@0:     // AdvanceTimeVaryingValuesToCurrentTime can rely on the value of mBlocked.
michael@0:     stream->mBlocked.AdvanceCurrentTime(nextCurrentTime);
michael@0: 
michael@0:     streamHasOutput[i] = blockedTime < nextCurrentTime - prevCurrentTime;
michael@0:     // Make this an assertion when bug 957832 is fixed.
michael@0:     NS_WARN_IF_FALSE(!streamHasOutput[i] || !stream->mNotifiedFinished,
michael@0:       "Shouldn't have already notified of finish *and* have output!");
michael@0: 
michael@0:     if (stream->mFinished && !stream->mNotifiedFinished) {
michael@0:       streamsReadyToFinish.AppendElement(stream);
michael@0:     }
michael@0:     STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p bufferStartTime=%f blockedTime=%f",
michael@0:                                 stream, MediaTimeToSeconds(stream->mBufferStartTime),
michael@0:                                 MediaTimeToSeconds(blockedTime)));
michael@0:   }
michael@0: 
michael@0:   mCurrentTime = nextCurrentTime;
michael@0: 
michael@0:   // Do these after setting mCurrentTime so that StreamTimeToGraphTime works properly.
michael@0:   for (uint32_t i = 0; i < streamHasOutput.Length(); ++i) {
michael@0:     if (!streamHasOutput[i]) {
michael@0:       continue;
michael@0:     }
michael@0:     MediaStream* stream = mStreams[i];
michael@0:     for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
michael@0:       MediaStreamListener* l = stream->mListeners[j];
michael@0:       l->NotifyOutput(this, mCurrentTime);
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   for (uint32_t i = 0; i < streamsReadyToFinish.Length(); ++i) {
michael@0:     MediaStream* stream = streamsReadyToFinish[i];
michael@0:     // The stream is fully finished when all of its track data has been played
michael@0:     // out.
michael@0:     if (mCurrentTime >=
michael@0:           stream->StreamTimeToGraphTime(stream->GetStreamBuffer().GetAllTracksEnd()))  {
michael@0:       NS_WARN_IF_FALSE(stream->mNotifiedBlocked,
michael@0:         "Should've notified blocked=true for a fully finished stream");
michael@0:       stream->mNotifiedFinished = true;
michael@0:       stream->mLastPlayedVideoFrame.SetNull();
michael@0:       SetStreamOrderDirty();
michael@0:       for (uint32_t j = 0; j < stream->mListeners.Length(); ++j) {
michael@0:         MediaStreamListener* l = stream->mListeners[j];
michael@0:         l->NotifyFinished(this);
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: bool
michael@0: MediaStreamGraphImpl::WillUnderrun(MediaStream* aStream, GraphTime aTime,
michael@0:                                    GraphTime aEndBlockingDecisions, GraphTime* aEnd)
michael@0: {
michael@0:   // Finished streams can't underrun. ProcessedMediaStreams also can't cause
michael@0:   // underrun currently, since we'll always be able to produce data for them
michael@0:   // unless they block on some other stream.
michael@0:   if (aStream->mFinished || aStream->AsProcessedStream()) {
michael@0:     return false;
michael@0:   }
michael@0:   GraphTime bufferEnd =
michael@0:     StreamTimeToGraphTime(aStream, aStream->GetBufferEnd(),
michael@0:                           INCLUDE_TRAILING_BLOCKED_INTERVAL);
michael@0: #ifdef DEBUG
michael@0:   if (bufferEnd < mCurrentTime) {
michael@0:     STREAM_LOG(PR_LOG_ERROR, ("MediaStream %p underrun, "
michael@0:                               "bufferEnd %f < mCurrentTime %f (%lld < %lld), Streamtime %lld",
michael@0:                               aStream, MediaTimeToSeconds(bufferEnd), MediaTimeToSeconds(mCurrentTime),
michael@0:                               bufferEnd, mCurrentTime, aStream->GetBufferEnd()));
michael@0:     aStream->DumpTrackInfo();
michael@0:     NS_ASSERTION(bufferEnd >= mCurrentTime, "Buffer underran");
michael@0:   }
michael@0: #endif
michael@0:   // We should block after bufferEnd.
michael@0:   if (bufferEnd <= aTime) {
michael@0:     STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p will block due to data underrun, "
michael@0:                                 "bufferEnd %f",
michael@0:                                 aStream, MediaTimeToSeconds(bufferEnd)));
michael@0:     return true;
michael@0:   }
michael@0:   // We should keep blocking if we're currently blocked and we don't have
michael@0:   // data all the way through to aEndBlockingDecisions. If we don't have
michael@0:   // data all the way through to aEndBlockingDecisions, we'll block soon,
michael@0:   // but we might as well remain unblocked and play the data we've got while
michael@0:   // we can.
michael@0:   if (bufferEnd <= aEndBlockingDecisions && aStream->mBlocked.GetBefore(aTime)) {
michael@0:     STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p will block due to speculative data underrun, "
michael@0:                                 "bufferEnd %f",
michael@0:                                 aStream, MediaTimeToSeconds(bufferEnd)));
michael@0:     return true;
michael@0:   }
michael@0:   // Reconsider decisions at bufferEnd
michael@0:   *aEnd = std::min(*aEnd, bufferEnd);
michael@0:   return false;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::MarkConsumed(MediaStream* aStream)
michael@0: {
michael@0:   if (aStream->mIsConsumed) {
michael@0:     return;
michael@0:   }
michael@0:   aStream->mIsConsumed = true;
michael@0: 
michael@0:   ProcessedMediaStream* ps = aStream->AsProcessedStream();
michael@0:   if (!ps) {
michael@0:     return;
michael@0:   }
michael@0:   // Mark all the inputs to this stream as consumed
michael@0:   for (uint32_t i = 0; i < ps->mInputs.Length(); ++i) {
michael@0:     MarkConsumed(ps->mInputs[i]->mSource);
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::UpdateStreamOrderForStream(mozilla::LinkedList<MediaStream>* aStack,
michael@0:                                                  already_AddRefed<MediaStream> aStream)
michael@0: {
michael@0:   nsRefPtr<MediaStream> stream = aStream;
michael@0:   NS_ASSERTION(!stream->mHasBeenOrdered, "stream should not have already been ordered");
michael@0:   if (stream->mIsOnOrderingStack) {
michael@0:     MediaStream* iter = aStack->getLast();
michael@0:     AudioNodeStream* ns = stream->AsAudioNodeStream();
michael@0:     bool delayNodePresent = ns ? ns->Engine()->AsDelayNodeEngine() != nullptr : false;
michael@0:     bool cycleFound = false;
michael@0:     if (iter) {
michael@0:       do {
michael@0:         cycleFound = true;
michael@0:         iter->AsProcessedStream()->mInCycle = true;
michael@0:         AudioNodeStream* ns = iter->AsAudioNodeStream();
michael@0:         if (ns && ns->Engine()->AsDelayNodeEngine()) {
michael@0:           delayNodePresent = true;
michael@0:         }
michael@0:         iter = iter->getPrevious();
michael@0:       } while (iter && iter != stream);
michael@0:     }
michael@0:     if (cycleFound && !delayNodePresent) {
michael@0:       // If we have detected a cycle, the previous loop should exit with stream
michael@0:       // == iter, or the node is connected to itself. Go back in the cycle and
michael@0:       // mute all nodes we find, or just mute the node itself.
michael@0:       if (!iter) {
michael@0:         // The node is connected to itself.
michael@0:         // There can't be a non-AudioNodeStream here, because only AudioNodes
michael@0:         // can be self-connected.
michael@0:         iter = aStack->getLast();
michael@0:         MOZ_ASSERT(iter->AsAudioNodeStream());
michael@0:         iter->AsAudioNodeStream()->Mute();
michael@0:       } else {
michael@0:         MOZ_ASSERT(iter);
michael@0:         do {
michael@0:           AudioNodeStream* nodeStream = iter->AsAudioNodeStream();
michael@0:           if (nodeStream) {
michael@0:             nodeStream->Mute();
michael@0:           }
michael@0:         } while((iter = iter->getNext()));
michael@0:       }
michael@0:     }
michael@0:     return;
michael@0:   }
michael@0:   ProcessedMediaStream* ps = stream->AsProcessedStream();
michael@0:   if (ps) {
michael@0:     aStack->insertBack(stream);
michael@0:     stream->mIsOnOrderingStack = true;
michael@0:     for (uint32_t i = 0; i < ps->mInputs.Length(); ++i) {
michael@0:       MediaStream* source = ps->mInputs[i]->mSource;
michael@0:       if (!source->mHasBeenOrdered) {
michael@0:         nsRefPtr<MediaStream> s = source;
michael@0:         UpdateStreamOrderForStream(aStack, s.forget());
michael@0:       }
michael@0:     }
michael@0:     aStack->popLast();
michael@0:     stream->mIsOnOrderingStack = false;
michael@0:   }
michael@0: 
michael@0:   stream->mHasBeenOrdered = true;
michael@0:   *mStreams.AppendElement() = stream.forget();
michael@0: }
michael@0: 
michael@0: static void AudioMixerCallback(AudioDataValue* aMixedBuffer,
michael@0:                                AudioSampleFormat aFormat,
michael@0:                                uint32_t aChannels,
michael@0:                                uint32_t aFrames,
michael@0:                                uint32_t aSampleRate)
michael@0: {
michael@0:   // Need an api to register mixer callbacks, bug 989921
michael@0: #ifdef MOZ_WEBRTC
michael@0:   if (aFrames > 0 && aChannels > 0) {
michael@0:     // XXX need Observer base class and registration API
michael@0:     if (gFarendObserver) {
michael@0:       gFarendObserver->InsertFarEnd(aMixedBuffer, aFrames, false,
michael@0:                                     aSampleRate, aChannels, aFormat);
michael@0:     }
michael@0:   }
michael@0: #endif
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::UpdateStreamOrder()
michael@0: {
michael@0:   mOldStreams.SwapElements(mStreams);
michael@0:   mStreams.ClearAndRetainStorage();
michael@0:   bool shouldMix = false;
michael@0:   for (uint32_t i = 0; i < mOldStreams.Length(); ++i) {
michael@0:     MediaStream* stream = mOldStreams[i];
michael@0:     stream->mHasBeenOrdered = false;
michael@0:     stream->mIsConsumed = false;
michael@0:     stream->mIsOnOrderingStack = false;
michael@0:     stream->mInBlockingSet = false;
michael@0:     if (stream->AsSourceStream() &&
michael@0:         stream->AsSourceStream()->NeedsMixing()) {
michael@0:       shouldMix = true;
michael@0:     }
michael@0:     ProcessedMediaStream* ps = stream->AsProcessedStream();
michael@0:     if (ps) {
michael@0:       ps->mInCycle = false;
michael@0:       AudioNodeStream* ns = ps->AsAudioNodeStream();
michael@0:       if (ns) {
michael@0:         ns->Unmute();
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   if (!mMixer && shouldMix) {
michael@0:     mMixer = new AudioMixer(AudioMixerCallback);
michael@0:   } else if (mMixer && !shouldMix) {
michael@0:     mMixer = nullptr;
michael@0:   }
michael@0: 
michael@0:   mozilla::LinkedList<MediaStream> stack;
michael@0:   for (uint32_t i = 0; i < mOldStreams.Length(); ++i) {
michael@0:     nsRefPtr<MediaStream>& s = mOldStreams[i];
michael@0:     if (s->IsIntrinsicallyConsumed()) {
michael@0:       MarkConsumed(s);
michael@0:     }
michael@0:     if (!s->mHasBeenOrdered) {
michael@0:       UpdateStreamOrderForStream(&stack, s.forget());
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::RecomputeBlocking(GraphTime aEndBlockingDecisions)
michael@0: {
michael@0:   bool blockingDecisionsWillChange = false;
michael@0: 
michael@0:   STREAM_LOG(PR_LOG_DEBUG+1, ("Media graph %p computing blocking for time %f",
michael@0:                               this, MediaTimeToSeconds(mStateComputedTime)));
michael@0:   for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:     MediaStream* stream = mStreams[i];
michael@0:     if (!stream->mInBlockingSet) {
michael@0:       // Compute a partition of the streams containing 'stream' such that we can
michael@0:       // compute the blocking status of each subset independently.
michael@0:       nsAutoTArray<MediaStream*,10> streamSet;
michael@0:       AddBlockingRelatedStreamsToSet(&streamSet, stream);
michael@0: 
michael@0:       GraphTime end;
michael@0:       for (GraphTime t = mStateComputedTime;
michael@0:            t < aEndBlockingDecisions; t = end) {
michael@0:         end = GRAPH_TIME_MAX;
michael@0:         RecomputeBlockingAt(streamSet, t, aEndBlockingDecisions, &end);
michael@0:         if (end < GRAPH_TIME_MAX) {
michael@0:           blockingDecisionsWillChange = true;
michael@0:         }
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     GraphTime end;
michael@0:     stream->mBlocked.GetAt(mCurrentTime, &end);
michael@0:     if (end < GRAPH_TIME_MAX) {
michael@0:       blockingDecisionsWillChange = true;
michael@0:     }
michael@0:   }
michael@0:   STREAM_LOG(PR_LOG_DEBUG+1, ("Media graph %p computed blocking for interval %f to %f",
michael@0:                               this, MediaTimeToSeconds(mStateComputedTime),
michael@0:                               MediaTimeToSeconds(aEndBlockingDecisions)));
michael@0:   mStateComputedTime = aEndBlockingDecisions;
michael@0: 
michael@0:   if (blockingDecisionsWillChange) {
michael@0:     // Make sure we wake up to notify listeners about these changes.
michael@0:     EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::AddBlockingRelatedStreamsToSet(nsTArray<MediaStream*>* aStreams,
michael@0:                                                      MediaStream* aStream)
michael@0: {
michael@0:   if (aStream->mInBlockingSet)
michael@0:     return;
michael@0:   aStream->mInBlockingSet = true;
michael@0:   aStreams->AppendElement(aStream);
michael@0:   for (uint32_t i = 0; i < aStream->mConsumers.Length(); ++i) {
michael@0:     MediaInputPort* port = aStream->mConsumers[i];
michael@0:     if (port->mFlags & (MediaInputPort::FLAG_BLOCK_INPUT | MediaInputPort::FLAG_BLOCK_OUTPUT)) {
michael@0:       AddBlockingRelatedStreamsToSet(aStreams, port->mDest);
michael@0:     }
michael@0:   }
michael@0:   ProcessedMediaStream* ps = aStream->AsProcessedStream();
michael@0:   if (ps) {
michael@0:     for (uint32_t i = 0; i < ps->mInputs.Length(); ++i) {
michael@0:       MediaInputPort* port = ps->mInputs[i];
michael@0:       if (port->mFlags & (MediaInputPort::FLAG_BLOCK_INPUT | MediaInputPort::FLAG_BLOCK_OUTPUT)) {
michael@0:         AddBlockingRelatedStreamsToSet(aStreams, port->mSource);
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::MarkStreamBlocking(MediaStream* aStream)
michael@0: {
michael@0:   if (aStream->mBlockInThisPhase)
michael@0:     return;
michael@0:   aStream->mBlockInThisPhase = true;
michael@0:   for (uint32_t i = 0; i < aStream->mConsumers.Length(); ++i) {
michael@0:     MediaInputPort* port = aStream->mConsumers[i];
michael@0:     if (port->mFlags & MediaInputPort::FLAG_BLOCK_OUTPUT) {
michael@0:       MarkStreamBlocking(port->mDest);
michael@0:     }
michael@0:   }
michael@0:   ProcessedMediaStream* ps = aStream->AsProcessedStream();
michael@0:   if (ps) {
michael@0:     for (uint32_t i = 0; i < ps->mInputs.Length(); ++i) {
michael@0:       MediaInputPort* port = ps->mInputs[i];
michael@0:       if (port->mFlags & MediaInputPort::FLAG_BLOCK_INPUT) {
michael@0:         MarkStreamBlocking(port->mSource);
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::RecomputeBlockingAt(const nsTArray<MediaStream*>& aStreams,
michael@0:                                           GraphTime aTime,
michael@0:                                           GraphTime aEndBlockingDecisions,
michael@0:                                           GraphTime* aEnd)
michael@0: {
michael@0:   for (uint32_t i = 0; i < aStreams.Length(); ++i) {
michael@0:     MediaStream* stream = aStreams[i];
michael@0:     stream->mBlockInThisPhase = false;
michael@0:   }
michael@0: 
michael@0:   for (uint32_t i = 0; i < aStreams.Length(); ++i) {
michael@0:     MediaStream* stream = aStreams[i];
michael@0: 
michael@0:     if (stream->mFinished) {
michael@0:       GraphTime endTime = StreamTimeToGraphTime(stream,
michael@0:          stream->GetStreamBuffer().GetAllTracksEnd());
michael@0:       if (endTime <= aTime) {
michael@0:         STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p is blocked due to being finished", stream));
michael@0:         // We'll block indefinitely
michael@0:         MarkStreamBlocking(stream);
michael@0:         *aEnd = std::min(*aEnd, aEndBlockingDecisions);
michael@0:         continue;
michael@0:       } else {
michael@0:         STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p is finished, but not blocked yet (end at %f, with blocking at %f)",
michael@0:                                     stream, MediaTimeToSeconds(stream->GetBufferEnd()),
michael@0:                                     MediaTimeToSeconds(endTime)));
michael@0:         *aEnd = std::min(*aEnd, endTime);
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     GraphTime end;
michael@0:     bool explicitBlock = stream->mExplicitBlockerCount.GetAt(aTime, &end) > 0;
michael@0:     *aEnd = std::min(*aEnd, end);
michael@0:     if (explicitBlock) {
michael@0:       STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p is blocked due to explicit blocker", stream));
michael@0:       MarkStreamBlocking(stream);
michael@0:       continue;
michael@0:     }
michael@0: 
michael@0:     bool underrun = WillUnderrun(stream, aTime, aEndBlockingDecisions, aEnd);
michael@0:     if (underrun) {
michael@0:       // We'll block indefinitely
michael@0:       MarkStreamBlocking(stream);
michael@0:       *aEnd = std::min(*aEnd, aEndBlockingDecisions);
michael@0:       continue;
michael@0:     }
michael@0:   }
michael@0:   NS_ASSERTION(*aEnd > aTime, "Failed to advance!");
michael@0: 
michael@0:   for (uint32_t i = 0; i < aStreams.Length(); ++i) {
michael@0:     MediaStream* stream = aStreams[i];
michael@0:     stream->mBlocked.SetAtAndAfter(aTime, stream->mBlockInThisPhase);
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::NotifyHasCurrentData(MediaStream* aStream)
michael@0: {
michael@0:   if (!aStream->mNotifiedHasCurrentData && aStream->mHasCurrentData) {
michael@0:     for (uint32_t j = 0; j < aStream->mListeners.Length(); ++j) {
michael@0:       MediaStreamListener* l = aStream->mListeners[j];
michael@0:       l->NotifyHasCurrentData(this);
michael@0:     }
michael@0:     aStream->mNotifiedHasCurrentData = true;
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::CreateOrDestroyAudioStreams(GraphTime aAudioOutputStartTime,
michael@0:                                                   MediaStream* aStream)
michael@0: {
michael@0:   MOZ_ASSERT(mRealtime, "Should only attempt to create audio streams in real-time mode");
michael@0: 
michael@0:   nsAutoTArray<bool,2> audioOutputStreamsFound;
michael@0:   for (uint32_t i = 0; i < aStream->mAudioOutputStreams.Length(); ++i) {
michael@0:     audioOutputStreamsFound.AppendElement(false);
michael@0:   }
michael@0: 
michael@0:   if (!aStream->mAudioOutputs.IsEmpty()) {
michael@0:     for (StreamBuffer::TrackIter tracks(aStream->GetStreamBuffer(), MediaSegment::AUDIO);
michael@0:          !tracks.IsEnded(); tracks.Next()) {
michael@0:       uint32_t i;
michael@0:       for (i = 0; i < audioOutputStreamsFound.Length(); ++i) {
michael@0:         if (aStream->mAudioOutputStreams[i].mTrackID == tracks->GetID()) {
michael@0:           break;
michael@0:         }
michael@0:       }
michael@0:       if (i < audioOutputStreamsFound.Length()) {
michael@0:         audioOutputStreamsFound[i] = true;
michael@0:       } else {
michael@0:         // No output stream created for this track yet. Check if it's time to
michael@0:         // create one.
michael@0:         GraphTime startTime =
michael@0:           StreamTimeToGraphTime(aStream, tracks->GetStartTimeRoundDown(),
michael@0:                                 INCLUDE_TRAILING_BLOCKED_INTERVAL);
michael@0:         if (startTime >= mStateComputedTime) {
michael@0:           // The stream wants to play audio, but nothing will play for the forseeable
michael@0:           // future, so don't create the stream.
michael@0:           continue;
michael@0:         }
michael@0: 
michael@0:         // Allocating a AudioStream would be slow, so we finish the Init async
michael@0:         MediaStream::AudioOutputStream* audioOutputStream =
michael@0:           aStream->mAudioOutputStreams.AppendElement();
michael@0:         audioOutputStream->mAudioPlaybackStartTime = aAudioOutputStartTime;
michael@0:         audioOutputStream->mBlockedAudioTime = 0;
michael@0:         audioOutputStream->mLastTickWritten = 0;
michael@0:         audioOutputStream->mStream = new AudioStream();
michael@0:         // XXX for now, allocate stereo output. But we need to fix this to
michael@0:         // match the system's ideal channel configuration.
michael@0:         // NOTE: we presume this is either fast or async-under-the-covers
michael@0:         audioOutputStream->mStream->Init(2, mSampleRate,
michael@0:                                          aStream->mAudioChannelType,
michael@0:                                          AudioStream::LowLatency);
michael@0:         audioOutputStream->mTrackID = tracks->GetID();
michael@0: 
michael@0:         LogLatency(AsyncLatencyLogger::AudioStreamCreate,
michael@0:                    reinterpret_cast<uint64_t>(aStream),
michael@0:                    reinterpret_cast<int64_t>(audioOutputStream->mStream.get()));
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   for (int32_t i = audioOutputStreamsFound.Length() - 1; i >= 0; --i) {
michael@0:     if (!audioOutputStreamsFound[i]) {
michael@0:       aStream->mAudioOutputStreams[i].mStream->Shutdown();
michael@0:       aStream->mAudioOutputStreams.RemoveElementAt(i);
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: TrackTicks
michael@0: MediaStreamGraphImpl::PlayAudio(MediaStream* aStream,
michael@0:                                 GraphTime aFrom, GraphTime aTo)
michael@0: {
michael@0:   MOZ_ASSERT(mRealtime, "Should only attempt to play audio in realtime mode");
michael@0: 
michael@0:   TrackTicks ticksWritten = 0;
michael@0:   // We compute the number of needed ticks by converting a difference of graph
michael@0:   // time rather than by substracting two converted stream time to ensure that
michael@0:   // the rounding between {Graph,Stream}Time and track ticks is not dependant
michael@0:   // on the absolute value of the {Graph,Stream}Time, and so that number of
michael@0:   // ticks to play is the same for each cycle.
michael@0:   TrackTicks ticksNeeded = TimeToTicksRoundDown(mSampleRate, aTo) - TimeToTicksRoundDown(mSampleRate, aFrom);
michael@0: 
michael@0:   if (aStream->mAudioOutputStreams.IsEmpty()) {
michael@0:     return 0;
michael@0:   }
michael@0: 
michael@0:   // When we're playing multiple copies of this stream at the same time, they're
michael@0:   // perfectly correlated so adding volumes is the right thing to do.
michael@0:   float volume = 0.0f;
michael@0:   for (uint32_t i = 0; i < aStream->mAudioOutputs.Length(); ++i) {
michael@0:     volume += aStream->mAudioOutputs[i].mVolume;
michael@0:   }
michael@0: 
michael@0:   for (uint32_t i = 0; i < aStream->mAudioOutputStreams.Length(); ++i) {
michael@0:     MediaStream::AudioOutputStream& audioOutput = aStream->mAudioOutputStreams[i];
michael@0:     StreamBuffer::Track* track = aStream->mBuffer.FindTrack(audioOutput.mTrackID);
michael@0:     AudioSegment* audio = track->Get<AudioSegment>();
michael@0:     AudioSegment output;
michael@0:     MOZ_ASSERT(track->GetRate() == mSampleRate);
michael@0: 
michael@0:     // offset and audioOutput.mLastTickWritten can differ by at most one sample,
michael@0:     // because of the rounding issue. We track that to ensure we don't skip a
michael@0:     // sample. One sample may be played twice, but this should not happen
michael@0:     // again during an unblocked sequence of track samples.
michael@0:     TrackTicks offset = track->TimeToTicksRoundDown(GraphTimeToStreamTime(aStream, aFrom));
michael@0:     if (audioOutput.mLastTickWritten &&
michael@0:         audioOutput.mLastTickWritten != offset) {
michael@0:       // If there is a global underrun of the MSG, this property won't hold, and
michael@0:       // we reset the sample count tracking.
michael@0:       if (offset - audioOutput.mLastTickWritten == 1) {
michael@0:         offset = audioOutput.mLastTickWritten;
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     // We don't update aStream->mBufferStartTime here to account for
michael@0:     // time spent blocked. Instead, we'll update it in UpdateCurrentTime after the
michael@0:     // blocked period has completed. But we do need to make sure we play from the
michael@0:     // right offsets in the stream buffer, even if we've already written silence for
michael@0:     // some amount of blocked time after the current time.
michael@0:     GraphTime t = aFrom;
michael@0:     while (ticksNeeded) {
michael@0:       GraphTime end;
michael@0:       bool blocked = aStream->mBlocked.GetAt(t, &end);
michael@0:       end = std::min(end, aTo);
michael@0: 
michael@0:       // Check how many ticks of sound we can provide if we are blocked some
michael@0:       // time in the middle of this cycle.
michael@0:       TrackTicks toWrite = 0;
michael@0:       if (end >= aTo) {
michael@0:         toWrite = ticksNeeded;
michael@0:       } else {
michael@0:         toWrite = TimeToTicksRoundDown(mSampleRate, end - aFrom);
michael@0:       }
michael@0:       ticksNeeded -= toWrite;
michael@0: 
michael@0:       if (blocked) {
michael@0:         output.InsertNullDataAtStart(toWrite);
michael@0:         STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p writing %ld blocking-silence samples for %f to %f (%ld to %ld)\n",
michael@0:                                     aStream, toWrite, MediaTimeToSeconds(t), MediaTimeToSeconds(end),
michael@0:                                     offset, offset + toWrite));
michael@0:       } else {
michael@0:         TrackTicks endTicksNeeded = offset + toWrite;
michael@0:         TrackTicks endTicksAvailable = audio->GetDuration();
michael@0:         STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p writing %ld samples for %f to %f (samples %ld to %ld)\n",
michael@0:                                      aStream, toWrite, MediaTimeToSeconds(t), MediaTimeToSeconds(end),
michael@0:                                      offset, endTicksNeeded));
michael@0: 
michael@0:         if (endTicksNeeded <= endTicksAvailable) {
michael@0:           output.AppendSlice(*audio, offset, endTicksNeeded);
michael@0:           offset = endTicksNeeded;
michael@0:         } else {
michael@0:           MOZ_ASSERT(track->IsEnded(), "Not enough data, and track not ended.");
michael@0:           // If we are at the end of the track, maybe write the remaining
michael@0:           // samples, and pad with/output silence.
michael@0:           if (endTicksNeeded > endTicksAvailable &&
michael@0:               offset < endTicksAvailable) {
michael@0:             output.AppendSlice(*audio, offset, endTicksAvailable);
michael@0:             toWrite -= endTicksAvailable - offset;
michael@0:             offset = endTicksAvailable;
michael@0:           }
michael@0:           output.AppendNullData(toWrite);
michael@0:         }
michael@0:         output.ApplyVolume(volume);
michael@0:       }
michael@0:       t = end;
michael@0:     }
michael@0:     audioOutput.mLastTickWritten = offset;
michael@0: 
michael@0:     // Need unique id for stream & track - and we want it to match the inserter
michael@0:     output.WriteTo(LATENCY_STREAM_ID(aStream, track->GetID()),
michael@0:                    audioOutput.mStream, mMixer);
michael@0:   }
michael@0:   return ticksWritten;
michael@0: }
michael@0: 
michael@0: static void
michael@0: SetImageToBlackPixel(PlanarYCbCrImage* aImage)
michael@0: {
michael@0:   uint8_t blackPixel[] = { 0x10, 0x80, 0x80 };
michael@0: 
michael@0:   PlanarYCbCrData data;
michael@0:   data.mYChannel = blackPixel;
michael@0:   data.mCbChannel = blackPixel + 1;
michael@0:   data.mCrChannel = blackPixel + 2;
michael@0:   data.mYStride = data.mCbCrStride = 1;
michael@0:   data.mPicSize = data.mYSize = data.mCbCrSize = IntSize(1, 1);
michael@0:   aImage->SetData(data);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::PlayVideo(MediaStream* aStream)
michael@0: {
michael@0:   MOZ_ASSERT(mRealtime, "Should only attempt to play video in realtime mode");
michael@0: 
michael@0:   if (aStream->mVideoOutputs.IsEmpty())
michael@0:     return;
michael@0: 
michael@0:   // Display the next frame a bit early. This is better than letting the current
michael@0:   // frame be displayed for too long.
michael@0:   GraphTime framePosition = mCurrentTime + MEDIA_GRAPH_TARGET_PERIOD_MS;
michael@0:   NS_ASSERTION(framePosition >= aStream->mBufferStartTime, "frame position before buffer?");
michael@0:   StreamTime frameBufferTime = GraphTimeToStreamTime(aStream, framePosition);
michael@0: 
michael@0:   TrackTicks start;
michael@0:   const VideoFrame* frame = nullptr;
michael@0:   StreamBuffer::Track* track;
michael@0:   for (StreamBuffer::TrackIter tracks(aStream->GetStreamBuffer(), MediaSegment::VIDEO);
michael@0:        !tracks.IsEnded(); tracks.Next()) {
michael@0:     VideoSegment* segment = tracks->Get<VideoSegment>();
michael@0:     TrackTicks thisStart;
michael@0:     const VideoFrame* thisFrame =
michael@0:       segment->GetFrameAt(tracks->TimeToTicksRoundDown(frameBufferTime), &thisStart);
michael@0:     if (thisFrame && thisFrame->GetImage()) {
michael@0:       start = thisStart;
michael@0:       frame = thisFrame;
michael@0:       track = tracks.get();
michael@0:     }
michael@0:   }
michael@0:   if (!frame || *frame == aStream->mLastPlayedVideoFrame)
michael@0:     return;
michael@0: 
michael@0:   STREAM_LOG(PR_LOG_DEBUG+1, ("MediaStream %p writing video frame %p (%dx%d)",
michael@0:                               aStream, frame->GetImage(), frame->GetIntrinsicSize().width,
michael@0:                               frame->GetIntrinsicSize().height));
michael@0:   GraphTime startTime = StreamTimeToGraphTime(aStream,
michael@0:       track->TicksToTimeRoundDown(start), INCLUDE_TRAILING_BLOCKED_INTERVAL);
michael@0:   TimeStamp targetTime = mCurrentTimeStamp +
michael@0:       TimeDuration::FromMilliseconds(double(startTime - mCurrentTime));
michael@0:   for (uint32_t i = 0; i < aStream->mVideoOutputs.Length(); ++i) {
michael@0:     VideoFrameContainer* output = aStream->mVideoOutputs[i];
michael@0: 
michael@0:     if (frame->GetForceBlack()) {
michael@0:       nsRefPtr<Image> image =
michael@0:         output->GetImageContainer()->CreateImage(ImageFormat::PLANAR_YCBCR);
michael@0:       if (image) {
michael@0:         // Sets the image to a single black pixel, which will be scaled to fill
michael@0:         // the rendered size.
michael@0:         SetImageToBlackPixel(static_cast<PlanarYCbCrImage*>(image.get()));
michael@0:       }
michael@0:       output->SetCurrentFrame(frame->GetIntrinsicSize(), image,
michael@0:                               targetTime);
michael@0:     } else {
michael@0:       output->SetCurrentFrame(frame->GetIntrinsicSize(), frame->GetImage(),
michael@0:                               targetTime);
michael@0:     }
michael@0: 
michael@0:     nsCOMPtr<nsIRunnable> event =
michael@0:       NS_NewRunnableMethod(output, &VideoFrameContainer::Invalidate);
michael@0:     NS_DispatchToMainThread(event, NS_DISPATCH_NORMAL);
michael@0:   }
michael@0:   if (!aStream->mNotifiedFinished) {
michael@0:     aStream->mLastPlayedVideoFrame = *frame;
michael@0:   }
michael@0: }
michael@0: 
michael@0: bool
michael@0: MediaStreamGraphImpl::ShouldUpdateMainThread()
michael@0: {
michael@0:   if (mRealtime) {
michael@0:     return true;
michael@0:   }
michael@0: 
michael@0:   TimeStamp now = TimeStamp::Now();
michael@0:   if ((now - mLastMainThreadUpdate).ToMilliseconds() > MEDIA_GRAPH_TARGET_PERIOD_MS) {
michael@0:     mLastMainThreadUpdate = now;
michael@0:     return true;
michael@0:   }
michael@0:   return false;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::PrepareUpdatesToMainThreadState(bool aFinalUpdate)
michael@0: {
michael@0:   mMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   // We don't want to frequently update the main thread about timing update
michael@0:   // when we are not running in realtime.
michael@0:   if (aFinalUpdate || ShouldUpdateMainThread()) {
michael@0:     mStreamUpdates.SetCapacity(mStreamUpdates.Length() + mStreams.Length());
michael@0:     for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:       MediaStream* stream = mStreams[i];
michael@0:       if (!stream->MainThreadNeedsUpdates()) {
michael@0:         continue;
michael@0:       }
michael@0:       StreamUpdate* update = mStreamUpdates.AppendElement();
michael@0:       update->mGraphUpdateIndex = stream->mGraphUpdateIndices.GetAt(mCurrentTime);
michael@0:       update->mStream = stream;
michael@0:       update->mNextMainThreadCurrentTime =
michael@0:         GraphTimeToStreamTime(stream, mCurrentTime);
michael@0:       update->mNextMainThreadFinished = stream->mNotifiedFinished;
michael@0:     }
michael@0:     if (!mPendingUpdateRunnables.IsEmpty()) {
michael@0:       mUpdateRunnables.MoveElementsFrom(mPendingUpdateRunnables);
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // Don't send the message to the main thread if it's not going to have
michael@0:   // any work to do.
michael@0:   if (aFinalUpdate ||
michael@0:       !mUpdateRunnables.IsEmpty() ||
michael@0:       !mStreamUpdates.IsEmpty()) {
michael@0:     EnsureStableStateEventPosted();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::EnsureImmediateWakeUpLocked(MonitorAutoLock& aLock)
michael@0: {
michael@0:   if (mWaitState == WAITSTATE_WAITING_FOR_NEXT_ITERATION ||
michael@0:       mWaitState == WAITSTATE_WAITING_INDEFINITELY) {
michael@0:     mWaitState = WAITSTATE_WAKING_UP;
michael@0:     aLock.Notify();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::EnsureNextIteration()
michael@0: {
michael@0:   MonitorAutoLock lock(mMonitor);
michael@0:   EnsureNextIterationLocked(lock);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::EnsureNextIterationLocked(MonitorAutoLock& aLock)
michael@0: {
michael@0:   if (mNeedAnotherIteration)
michael@0:     return;
michael@0:   mNeedAnotherIteration = true;
michael@0:   if (mWaitState == WAITSTATE_WAITING_INDEFINITELY) {
michael@0:     mWaitState = WAITSTATE_WAKING_UP;
michael@0:     aLock.Notify();
michael@0:   }
michael@0: }
michael@0: 
michael@0: /**
michael@0:  * Returns smallest value of t such that
michael@0:  * TimeToTicksRoundUp(aSampleRate, t) is a multiple of WEBAUDIO_BLOCK_SIZE
michael@0:  * and floor(TimeToTicksRoundUp(aSampleRate, t)/WEBAUDIO_BLOCK_SIZE) >
michael@0:  * floor(TimeToTicksRoundUp(aSampleRate, aTime)/WEBAUDIO_BLOCK_SIZE).
michael@0:  */
michael@0: static GraphTime
michael@0: RoundUpToNextAudioBlock(TrackRate aSampleRate, GraphTime aTime)
michael@0: {
michael@0:   TrackTicks ticks = TimeToTicksRoundUp(aSampleRate, aTime);
michael@0:   uint64_t block = ticks >> WEBAUDIO_BLOCK_SIZE_BITS;
michael@0:   uint64_t nextBlock = block + 1;
michael@0:   TrackTicks nextTicks = nextBlock << WEBAUDIO_BLOCK_SIZE_BITS;
michael@0:   // Find the smallest time t such that TimeToTicksRoundUp(aSampleRate,t) == nextTicks
michael@0:   // That's the smallest integer t such that
michael@0:   //   t*aSampleRate > ((nextTicks - 1) << MEDIA_TIME_FRAC_BITS)
michael@0:   // Both sides are integers, so this is equivalent to
michael@0:   //   t*aSampleRate >= ((nextTicks - 1) << MEDIA_TIME_FRAC_BITS) + 1
michael@0:   //   t >= (((nextTicks - 1) << MEDIA_TIME_FRAC_BITS) + 1)/aSampleRate
michael@0:   //   t = ceil((((nextTicks - 1) << MEDIA_TIME_FRAC_BITS) + 1)/aSampleRate)
michael@0:   // Using integer division, that's
michael@0:   //   t = (((nextTicks - 1) << MEDIA_TIME_FRAC_BITS) + 1 + aSampleRate - 1)/aSampleRate
michael@0:   //     = ((nextTicks - 1) << MEDIA_TIME_FRAC_BITS)/aSampleRate + 1
michael@0:   return ((nextTicks - 1) << MEDIA_TIME_FRAC_BITS)/aSampleRate + 1;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ProduceDataForStreamsBlockByBlock(uint32_t aStreamIndex,
michael@0:                                                         TrackRate aSampleRate,
michael@0:                                                         GraphTime aFrom,
michael@0:                                                         GraphTime aTo)
michael@0: {
michael@0:   GraphTime t = aFrom;
michael@0:   while (t < aTo) {
michael@0:     GraphTime next = RoundUpToNextAudioBlock(aSampleRate, t);
michael@0:     for (uint32_t i = aStreamIndex; i < mStreams.Length(); ++i) {
michael@0:       ProcessedMediaStream* ps = mStreams[i]->AsProcessedStream();
michael@0:       if (ps) {
michael@0:         ps->ProcessInput(t, next, (next == aTo) ? ProcessedMediaStream::ALLOW_FINISH : 0);
michael@0:       }
michael@0:     }
michael@0:     t = next;
michael@0:   }
michael@0:   NS_ASSERTION(t == aTo, "Something went wrong with rounding to block boundaries");
michael@0: }
michael@0: 
michael@0: bool
michael@0: MediaStreamGraphImpl::AllFinishedStreamsNotified()
michael@0: {
michael@0:   for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:     MediaStream* s = mStreams[i];
michael@0:     if (s->mFinished && !s->mNotifiedFinished) {
michael@0:       return false;
michael@0:     }
michael@0:   }
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::PauseAllAudioOutputs()
michael@0: {
michael@0:   for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:     MediaStream* s = mStreams[i];
michael@0:     for (uint32_t j = 0; j < s->mAudioOutputStreams.Length(); ++j) {
michael@0:       s->mAudioOutputStreams[j].mStream->Pause();
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ResumeAllAudioOutputs()
michael@0: {
michael@0:   for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:     MediaStream* s = mStreams[i];
michael@0:     for (uint32_t j = 0; j < s->mAudioOutputStreams.Length(); ++j) {
michael@0:       s->mAudioOutputStreams[j].mStream->Resume();
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: struct AutoProfilerUnregisterThread
michael@0: {
michael@0:   // The empty ctor is used to silence a pre-4.8.0 GCC unused variable warning.
michael@0:   AutoProfilerUnregisterThread()
michael@0:   {
michael@0:   }
michael@0: 
michael@0:   ~AutoProfilerUnregisterThread()
michael@0:   {
michael@0:     profiler_unregister_thread();
michael@0:   }
michael@0: };
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::RunThread()
michael@0: {
michael@0:   nsTArray<MessageBlock> messageQueue;
michael@0:   {
michael@0:     MonitorAutoLock lock(mMonitor);
michael@0:     messageQueue.SwapElements(mMessageQueue);
michael@0:   }
michael@0:   NS_ASSERTION(!messageQueue.IsEmpty(),
michael@0:                "Shouldn't have started a graph with empty message queue!");
michael@0: 
michael@0:   uint32_t ticksProcessed = 0;
michael@0:   AutoProfilerUnregisterThread autoUnregister;
michael@0: 
michael@0:   for (;;) {
michael@0:     // Check if a memory report has been requested.
michael@0:     {
michael@0:       MonitorAutoLock lock(mMemoryReportMonitor);
michael@0:       if (mNeedsMemoryReport) {
michael@0:         mNeedsMemoryReport = false;
michael@0: 
michael@0:         for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:           AudioNodeStream* stream = mStreams[i]->AsAudioNodeStream();
michael@0:           if (stream) {
michael@0:             AudioNodeSizes usage;
michael@0:             stream->SizeOfAudioNodesIncludingThis(MallocSizeOf, usage);
michael@0:             mAudioStreamSizes.AppendElement(usage);
michael@0:           }
michael@0:         }
michael@0: 
michael@0:         lock.Notify();
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     // Update mCurrentTime to the min of the playing audio times, or using the
michael@0:     // wall-clock time change if no audio is playing.
michael@0:     UpdateCurrentTime();
michael@0: 
michael@0:     // Calculate independent action times for each batch of messages (each
michael@0:     // batch corresponding to an event loop task). This isolates the performance
michael@0:     // of different scripts to some extent.
michael@0:     for (uint32_t i = 0; i < messageQueue.Length(); ++i) {
michael@0:       mProcessingGraphUpdateIndex = messageQueue[i].mGraphUpdateIndex;
michael@0:       nsTArray<nsAutoPtr<ControlMessage> >& messages = messageQueue[i].mMessages;
michael@0: 
michael@0:       for (uint32_t j = 0; j < messages.Length(); ++j) {
michael@0:         messages[j]->Run();
michael@0:       }
michael@0:     }
michael@0:     messageQueue.Clear();
michael@0: 
michael@0:     if (mStreamOrderDirty) {
michael@0:       UpdateStreamOrder();
michael@0:     }
michael@0: 
michael@0:     GraphTime endBlockingDecisions =
michael@0:       RoundUpToNextAudioBlock(mSampleRate, mCurrentTime + MillisecondsToMediaTime(AUDIO_TARGET_MS));
michael@0:     bool ensureNextIteration = false;
michael@0: 
michael@0:     // Grab pending stream input.
michael@0:     for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:       SourceMediaStream* is = mStreams[i]->AsSourceStream();
michael@0:       if (is) {
michael@0:         UpdateConsumptionState(is);
michael@0:         ExtractPendingInput(is, endBlockingDecisions, &ensureNextIteration);
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     // The loop is woken up so soon that mCurrentTime barely advances and we
michael@0:     // end up having endBlockingDecisions == mStateComputedTime.
michael@0:     // Since stream blocking is computed in the interval of
michael@0:     // [mStateComputedTime, endBlockingDecisions), it won't be computed at all.
michael@0:     // We should ensure next iteration so that pending blocking changes will be
michael@0:     // computed in next loop.
michael@0:     if (endBlockingDecisions == mStateComputedTime) {
michael@0:       ensureNextIteration = true;
michael@0:     }
michael@0: 
michael@0:     // Figure out which streams are blocked and when.
michael@0:     GraphTime prevComputedTime = mStateComputedTime;
michael@0:     RecomputeBlocking(endBlockingDecisions);
michael@0: 
michael@0:     // Play stream contents.
michael@0:     bool allBlockedForever = true;
michael@0:     // True when we've done ProcessInput for all processed streams.
michael@0:     bool doneAllProducing = false;
michael@0:     // This is the number of frame that are written to the AudioStreams, for
michael@0:     // this cycle.
michael@0:     TrackTicks ticksPlayed = 0;
michael@0:     // Figure out what each stream wants to do
michael@0:     for (uint32_t i = 0; i < mStreams.Length(); ++i) {
michael@0:       MediaStream* stream = mStreams[i];
michael@0:       if (!doneAllProducing) {
michael@0:         ProcessedMediaStream* ps = stream->AsProcessedStream();
michael@0:         if (ps) {
michael@0:           AudioNodeStream* n = stream->AsAudioNodeStream();
michael@0:           if (n) {
michael@0: #ifdef DEBUG
michael@0:             // Verify that the sampling rate for all of the following streams is the same
michael@0:             for (uint32_t j = i + 1; j < mStreams.Length(); ++j) {
michael@0:               AudioNodeStream* nextStream = mStreams[j]->AsAudioNodeStream();
michael@0:               if (nextStream) {
michael@0:                 MOZ_ASSERT(n->SampleRate() == nextStream->SampleRate(),
michael@0:                            "All AudioNodeStreams in the graph must have the same sampling rate");
michael@0:               }
michael@0:             }
michael@0: #endif
michael@0:             // Since an AudioNodeStream is present, go ahead and
michael@0:             // produce audio block by block for all the rest of the streams.
michael@0:             ProduceDataForStreamsBlockByBlock(i, n->SampleRate(), prevComputedTime, mStateComputedTime);
michael@0:             ticksProcessed += TimeToTicksRoundDown(n->SampleRate(), mStateComputedTime - prevComputedTime);
michael@0:             doneAllProducing = true;
michael@0:           } else {
michael@0:             ps->ProcessInput(prevComputedTime, mStateComputedTime,
michael@0:                              ProcessedMediaStream::ALLOW_FINISH);
michael@0:             NS_WARN_IF_FALSE(stream->mBuffer.GetEnd() >=
michael@0:                              GraphTimeToStreamTime(stream, mStateComputedTime),
michael@0:                              "Stream did not produce enough data");
michael@0:           }
michael@0:         }
michael@0:       }
michael@0:       NotifyHasCurrentData(stream);
michael@0:       if (mRealtime) {
michael@0:         // Only playback audio and video in real-time mode
michael@0:         CreateOrDestroyAudioStreams(prevComputedTime, stream);
michael@0:         TrackTicks ticksPlayedForThisStream = PlayAudio(stream, prevComputedTime, mStateComputedTime);
michael@0:         if (!ticksPlayed) {
michael@0:           ticksPlayed = ticksPlayedForThisStream;
michael@0:         } else {
michael@0:           MOZ_ASSERT(!ticksPlayedForThisStream || ticksPlayedForThisStream == ticksPlayed,
michael@0:               "Each stream should have the same number of frame.");
michael@0:         }
michael@0:         PlayVideo(stream);
michael@0:       }
michael@0:       SourceMediaStream* is = stream->AsSourceStream();
michael@0:       if (is) {
michael@0:         UpdateBufferSufficiencyState(is);
michael@0:       }
michael@0:       GraphTime end;
michael@0:       if (!stream->mBlocked.GetAt(mCurrentTime, &end) || end < GRAPH_TIME_MAX) {
michael@0:         allBlockedForever = false;
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     if (mMixer) {
michael@0:       mMixer->FinishMixing();
michael@0:     }
michael@0: 
michael@0:     if (ensureNextIteration || !allBlockedForever) {
michael@0:       EnsureNextIteration();
michael@0:     }
michael@0: 
michael@0:     // Send updates to the main thread and wait for the next control loop
michael@0:     // iteration.
michael@0:     {
michael@0:       MonitorAutoLock lock(mMonitor);
michael@0:       bool finalUpdate = mForceShutDown ||
michael@0:         (mCurrentTime >= mEndTime && AllFinishedStreamsNotified()) ||
michael@0:         (IsEmpty() && mMessageQueue.IsEmpty());
michael@0:       PrepareUpdatesToMainThreadState(finalUpdate);
michael@0:       if (finalUpdate) {
michael@0:         // Enter shutdown mode. The stable-state handler will detect this
michael@0:         // and complete shutdown. Destroy any streams immediately.
michael@0:         STREAM_LOG(PR_LOG_DEBUG, ("MediaStreamGraph %p waiting for main thread cleanup", this));
michael@0:         // We'll shut down this graph object if it does not get restarted.
michael@0:         mLifecycleState = LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP;
michael@0:         // No need to Destroy streams here. The main-thread owner of each
michael@0:         // stream is responsible for calling Destroy on them.
michael@0:         return;
michael@0:       }
michael@0: 
michael@0:       // No need to wait in non-realtime mode, just churn through the input as soon
michael@0:       // as possible.
michael@0:       if (mRealtime) {
michael@0:         PRIntervalTime timeout = PR_INTERVAL_NO_TIMEOUT;
michael@0:         TimeStamp now = TimeStamp::Now();
michael@0:         bool pausedOutputs = false;
michael@0:         if (mNeedAnotherIteration) {
michael@0:           int64_t timeoutMS = MEDIA_GRAPH_TARGET_PERIOD_MS -
michael@0:             int64_t((now - mCurrentTimeStamp).ToMilliseconds());
michael@0:           // Make sure timeoutMS doesn't overflow 32 bits by waking up at
michael@0:           // least once a minute, if we need to wake up at all
michael@0:           timeoutMS = std::max<int64_t>(0, std::min<int64_t>(timeoutMS, 60*1000));
michael@0:           timeout = PR_MillisecondsToInterval(uint32_t(timeoutMS));
michael@0:           STREAM_LOG(PR_LOG_DEBUG+1, ("Waiting for next iteration; at %f, timeout=%f",
michael@0:                                      (now - mInitialTimeStamp).ToSeconds(), timeoutMS/1000.0));
michael@0:           mWaitState = WAITSTATE_WAITING_FOR_NEXT_ITERATION;
michael@0:         } else {
michael@0:           mWaitState = WAITSTATE_WAITING_INDEFINITELY;
michael@0:           PauseAllAudioOutputs();
michael@0:           pausedOutputs = true;
michael@0:         }
michael@0:         if (timeout > 0) {
michael@0:           mMonitor.Wait(timeout);
michael@0:           STREAM_LOG(PR_LOG_DEBUG+1, ("Resuming after timeout; at %f, elapsed=%f",
michael@0:                                      (TimeStamp::Now() - mInitialTimeStamp).ToSeconds(),
michael@0:                                      (TimeStamp::Now() - now).ToSeconds()));
michael@0:         }
michael@0:         if (pausedOutputs) {
michael@0:           ResumeAllAudioOutputs();
michael@0:         }
michael@0:       }
michael@0:       mWaitState = WAITSTATE_RUNNING;
michael@0:       mNeedAnotherIteration = false;
michael@0:       messageQueue.SwapElements(mMessageQueue);
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ApplyStreamUpdate(StreamUpdate* aUpdate)
michael@0: {
michael@0:   mMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   MediaStream* stream = aUpdate->mStream;
michael@0:   if (!stream)
michael@0:     return;
michael@0:   stream->mMainThreadCurrentTime = aUpdate->mNextMainThreadCurrentTime;
michael@0:   stream->mMainThreadFinished = aUpdate->mNextMainThreadFinished;
michael@0: 
michael@0:   if (stream->mWrapper) {
michael@0:     stream->mWrapper->NotifyStreamStateChanged();
michael@0:   }
michael@0:   for (int32_t i = stream->mMainThreadListeners.Length() - 1; i >= 0; --i) {
michael@0:     stream->mMainThreadListeners[i]->NotifyMainThreadStateChanged();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ShutdownThreads()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Must be called on main thread");
michael@0:   // mGraph's thread is not running so it's OK to do whatever here
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("Stopping threads for MediaStreamGraph %p", this));
michael@0: 
michael@0:   if (mThread) {
michael@0:     mThread->Shutdown();
michael@0:     mThread = nullptr;
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::ForceShutDown()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Must be called on main thread");
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("MediaStreamGraph %p ForceShutdown", this));
michael@0:   {
michael@0:     MonitorAutoLock lock(mMonitor);
michael@0:     mForceShutDown = true;
michael@0:     EnsureImmediateWakeUpLocked(lock);
michael@0:   }
michael@0: }
michael@0: 
michael@0: namespace {
michael@0: 
michael@0: class MediaStreamGraphInitThreadRunnable : public nsRunnable {
michael@0: public:
michael@0:   explicit MediaStreamGraphInitThreadRunnable(MediaStreamGraphImpl* aGraph)
michael@0:     : mGraph(aGraph)
michael@0:   {
michael@0:   }
michael@0:   NS_IMETHOD Run()
michael@0:   {
michael@0:     char aLocal;
michael@0:     profiler_register_thread("MediaStreamGraph", &aLocal);
michael@0:     mGraph->RunThread();
michael@0:     return NS_OK;
michael@0:   }
michael@0: private:
michael@0:   MediaStreamGraphImpl* mGraph;
michael@0: };
michael@0: 
michael@0: class MediaStreamGraphThreadRunnable : public nsRunnable {
michael@0: public:
michael@0:   explicit MediaStreamGraphThreadRunnable(MediaStreamGraphImpl* aGraph)
michael@0:     : mGraph(aGraph)
michael@0:   {
michael@0:   }
michael@0:   NS_IMETHOD Run()
michael@0:   {
michael@0:     mGraph->RunThread();
michael@0:     return NS_OK;
michael@0:   }
michael@0: private:
michael@0:   MediaStreamGraphImpl* mGraph;
michael@0: };
michael@0: 
michael@0: class MediaStreamGraphShutDownRunnable : public nsRunnable {
michael@0: public:
michael@0:   MediaStreamGraphShutDownRunnable(MediaStreamGraphImpl* aGraph) : mGraph(aGraph) {}
michael@0:   NS_IMETHOD Run()
michael@0:   {
michael@0:     NS_ASSERTION(mGraph->mDetectedNotRunning,
michael@0:                  "We should know the graph thread control loop isn't running!");
michael@0: 
michael@0:     mGraph->ShutdownThreads();
michael@0: 
michael@0:     // mGraph's thread is not running so it's OK to do whatever here
michael@0:     if (mGraph->IsEmpty()) {
michael@0:       // mGraph is no longer needed, so delete it.
michael@0:       mGraph->Destroy();
michael@0:     } else {
michael@0:       // The graph is not empty.  We must be in a forced shutdown, or a
michael@0:       // non-realtime graph that has finished processing.  Some later
michael@0:       // AppendMessage will detect that the manager has been emptied, and
michael@0:       // delete it.
michael@0:       NS_ASSERTION(mGraph->mForceShutDown || !mGraph->mRealtime,
michael@0:                    "Not in forced shutdown?");
michael@0:       for (uint32_t i = 0; i < mGraph->mStreams.Length(); ++i) {
michael@0:         DOMMediaStream* s = mGraph->mStreams[i]->GetWrapper();
michael@0:         if (s) {
michael@0:           s->NotifyMediaStreamGraphShutdown();
michael@0:         }
michael@0:       }
michael@0: 
michael@0:       mGraph->mLifecycleState =
michael@0:         MediaStreamGraphImpl::LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION;
michael@0:     }
michael@0:     return NS_OK;
michael@0:   }
michael@0: private:
michael@0:   MediaStreamGraphImpl* mGraph;
michael@0: };
michael@0: 
michael@0: class MediaStreamGraphStableStateRunnable : public nsRunnable {
michael@0: public:
michael@0:   explicit MediaStreamGraphStableStateRunnable(MediaStreamGraphImpl* aGraph)
michael@0:     : mGraph(aGraph)
michael@0:   {
michael@0:   }
michael@0:   NS_IMETHOD Run()
michael@0:   {
michael@0:     if (mGraph) {
michael@0:       mGraph->RunInStableState();
michael@0:     }
michael@0:     return NS_OK;
michael@0:   }
michael@0: private:
michael@0:   MediaStreamGraphImpl* mGraph;
michael@0: };
michael@0: 
michael@0: /*
michael@0:  * Control messages forwarded from main thread to graph manager thread
michael@0:  */
michael@0: class CreateMessage : public ControlMessage {
michael@0: public:
michael@0:   CreateMessage(MediaStream* aStream) : ControlMessage(aStream) {}
michael@0:   virtual void Run() MOZ_OVERRIDE
michael@0:   {
michael@0:     mStream->GraphImpl()->AddStream(mStream);
michael@0:     mStream->Init();
michael@0:   }
michael@0:   virtual void RunDuringShutdown() MOZ_OVERRIDE
michael@0:   {
michael@0:     // Make sure to run this message during shutdown too, to make sure
michael@0:     // that we balance the number of streams registered with the graph
michael@0:     // as they're destroyed during shutdown.
michael@0:     Run();
michael@0:   }
michael@0: };
michael@0: 
michael@0: class MediaStreamGraphShutdownObserver MOZ_FINAL : public nsIObserver
michael@0: {
michael@0: public:
michael@0:   NS_DECL_ISUPPORTS
michael@0:   NS_DECL_NSIOBSERVER
michael@0: };
michael@0: 
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::RunInStableState()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Must be called on main thread");
michael@0: 
michael@0:   nsTArray<nsCOMPtr<nsIRunnable> > runnables;
michael@0:   // When we're doing a forced shutdown, pending control messages may be
michael@0:   // run on the main thread via RunDuringShutdown. Those messages must
michael@0:   // run without the graph monitor being held. So, we collect them here.
michael@0:   nsTArray<nsAutoPtr<ControlMessage> > controlMessagesToRunDuringShutdown;
michael@0: 
michael@0:   {
michael@0:     MonitorAutoLock lock(mMonitor);
michael@0:     mPostedRunInStableStateEvent = false;
michael@0: 
michael@0:     runnables.SwapElements(mUpdateRunnables);
michael@0:     for (uint32_t i = 0; i < mStreamUpdates.Length(); ++i) {
michael@0:       StreamUpdate* update = &mStreamUpdates[i];
michael@0:       if (update->mStream) {
michael@0:         ApplyStreamUpdate(update);
michael@0:       }
michael@0:     }
michael@0:     mStreamUpdates.Clear();
michael@0: 
michael@0:     // Don't start the thread for a non-realtime graph until it has been
michael@0:     // explicitly started by StartNonRealtimeProcessing.
michael@0:     if (mLifecycleState == LIFECYCLE_THREAD_NOT_STARTED &&
michael@0:         (mRealtime || mNonRealtimeProcessing)) {
michael@0:       mLifecycleState = LIFECYCLE_RUNNING;
michael@0:       // Start the thread now. We couldn't start it earlier because
michael@0:       // the graph might exit immediately on finding it has no streams. The
michael@0:       // first message for a new graph must create a stream.
michael@0:       nsCOMPtr<nsIRunnable> event = new MediaStreamGraphInitThreadRunnable(this);
michael@0:       NS_NewNamedThread("MediaStreamGrph", getter_AddRefs(mThread), event);
michael@0:     }
michael@0: 
michael@0:     if (mCurrentTaskMessageQueue.IsEmpty()) {
michael@0:       if (mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP && IsEmpty()) {
michael@0:         // Complete shutdown. First, ensure that this graph is no longer used.
michael@0:         // A new graph graph will be created if one is needed.
michael@0:         STREAM_LOG(PR_LOG_DEBUG, ("Disconnecting MediaStreamGraph %p", this));
michael@0:         if (this == gGraph) {
michael@0:           // null out gGraph if that's the graph being shut down
michael@0:           gGraph = nullptr;
michael@0:         }
michael@0:         // Asynchronously clean up old graph. We don't want to do this
michael@0:         // synchronously because it spins the event loop waiting for threads
michael@0:         // to shut down, and we don't want to do that in a stable state handler.
michael@0:         mLifecycleState = LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
michael@0:         nsCOMPtr<nsIRunnable> event = new MediaStreamGraphShutDownRunnable(this);
michael@0:         NS_DispatchToMainThread(event);
michael@0:       }
michael@0:     } else {
michael@0:       if (mLifecycleState <= LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
michael@0:         MessageBlock* block = mMessageQueue.AppendElement();
michael@0:         block->mMessages.SwapElements(mCurrentTaskMessageQueue);
michael@0:         block->mGraphUpdateIndex = mNextGraphUpdateIndex;
michael@0:         ++mNextGraphUpdateIndex;
michael@0:         EnsureNextIterationLocked(lock);
michael@0:       }
michael@0: 
michael@0:       // If the MediaStreamGraph has more messages going to it, try to revive
michael@0:       // it to process those messages. Don't do this if we're in a forced
michael@0:       // shutdown or it's a non-realtime graph that has already terminated
michael@0:       // processing.
michael@0:       if (mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP &&
michael@0:           mRealtime && !mForceShutDown) {
michael@0:         mLifecycleState = LIFECYCLE_RUNNING;
michael@0:         // Revive the MediaStreamGraph since we have more messages going to it.
michael@0:         // Note that we need to put messages into its queue before reviving it,
michael@0:         // or it might exit immediately.
michael@0:         nsCOMPtr<nsIRunnable> event = new MediaStreamGraphThreadRunnable(this);
michael@0:         mThread->Dispatch(event, 0);
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     if ((mForceShutDown || !mRealtime) &&
michael@0:         mLifecycleState == LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
michael@0:       // Defer calls to RunDuringShutdown() to happen while mMonitor is not held.
michael@0:       for (uint32_t i = 0; i < mMessageQueue.Length(); ++i) {
michael@0:         MessageBlock& mb = mMessageQueue[i];
michael@0:         controlMessagesToRunDuringShutdown.MoveElementsFrom(mb.mMessages);
michael@0:       }
michael@0:       mMessageQueue.Clear();
michael@0:       MOZ_ASSERT(mCurrentTaskMessageQueue.IsEmpty());
michael@0:       // Stop MediaStreamGraph threads. Do not clear gGraph since
michael@0:       // we have outstanding DOM objects that may need it.
michael@0:       mLifecycleState = LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
michael@0:       nsCOMPtr<nsIRunnable> event = new MediaStreamGraphShutDownRunnable(this);
michael@0:       NS_DispatchToMainThread(event);
michael@0:     }
michael@0: 
michael@0:     mDetectedNotRunning = mLifecycleState > LIFECYCLE_RUNNING;
michael@0:   }
michael@0: 
michael@0:   // Make sure we get a new current time in the next event loop task
michael@0:   mPostedRunInStableState = false;
michael@0: 
michael@0:   for (uint32_t i = 0; i < runnables.Length(); ++i) {
michael@0:     runnables[i]->Run();
michael@0:   }
michael@0:   for (uint32_t i = 0; i < controlMessagesToRunDuringShutdown.Length(); ++i) {
michael@0:     controlMessagesToRunDuringShutdown[i]->RunDuringShutdown();
michael@0:   }
michael@0: 
michael@0: #ifdef DEBUG
michael@0:   mCanRunMessagesSynchronously = mDetectedNotRunning &&
michael@0:     mLifecycleState >= LIFECYCLE_WAITING_FOR_THREAD_SHUTDOWN;
michael@0: #endif
michael@0: }
michael@0: 
michael@0: static NS_DEFINE_CID(kAppShellCID, NS_APPSHELL_CID);
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::EnsureRunInStableState()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "main thread only");
michael@0: 
michael@0:   if (mPostedRunInStableState)
michael@0:     return;
michael@0:   mPostedRunInStableState = true;
michael@0:   nsCOMPtr<nsIRunnable> event = new MediaStreamGraphStableStateRunnable(this);
michael@0:   nsCOMPtr<nsIAppShell> appShell = do_GetService(kAppShellCID);
michael@0:   if (appShell) {
michael@0:     appShell->RunInStableState(event);
michael@0:   } else {
michael@0:     NS_ERROR("Appshell already destroyed?");
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::EnsureStableStateEventPosted()
michael@0: {
michael@0:   mMonitor.AssertCurrentThreadOwns();
michael@0: 
michael@0:   if (mPostedRunInStableStateEvent)
michael@0:     return;
michael@0:   mPostedRunInStableStateEvent = true;
michael@0:   nsCOMPtr<nsIRunnable> event = new MediaStreamGraphStableStateRunnable(this);
michael@0:   NS_DispatchToMainThread(event);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::AppendMessage(ControlMessage* aMessage)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "main thread only");
michael@0:   NS_ASSERTION(!aMessage->GetStream() ||
michael@0:                !aMessage->GetStream()->IsDestroyed(),
michael@0:                "Stream already destroyed");
michael@0: 
michael@0:   if (mDetectedNotRunning &&
michael@0:       mLifecycleState > LIFECYCLE_WAITING_FOR_MAIN_THREAD_CLEANUP) {
michael@0:     // The graph control loop is not running and main thread cleanup has
michael@0:     // happened. From now on we can't append messages to mCurrentTaskMessageQueue,
michael@0:     // because that will never be processed again, so just RunDuringShutdown
michael@0:     // this message.
michael@0:     // This should only happen during forced shutdown, or after a non-realtime
michael@0:     // graph has finished processing.
michael@0: #ifdef DEBUG
michael@0:     MOZ_ASSERT(mCanRunMessagesSynchronously);
michael@0:     mCanRunMessagesSynchronously = false;
michael@0: #endif
michael@0:     aMessage->RunDuringShutdown();
michael@0: #ifdef DEBUG
michael@0:     mCanRunMessagesSynchronously = true;
michael@0: #endif
michael@0:     delete aMessage;
michael@0:     if (IsEmpty() &&
michael@0:         mLifecycleState >= LIFECYCLE_WAITING_FOR_STREAM_DESTRUCTION) {
michael@0:       if (gGraph == this) {
michael@0:         gGraph = nullptr;
michael@0:       }
michael@0:       Destroy();
michael@0:     }
michael@0:     return;
michael@0:   }
michael@0: 
michael@0:   mCurrentTaskMessageQueue.AppendElement(aMessage);
michael@0:   EnsureRunInStableState();
michael@0: }
michael@0: 
michael@0: MediaStream::MediaStream(DOMMediaStream* aWrapper)
michael@0:   : mBufferStartTime(0)
michael@0:   , mExplicitBlockerCount(0)
michael@0:   , mBlocked(false)
michael@0:   , mGraphUpdateIndices(0)
michael@0:   , mFinished(false)
michael@0:   , mNotifiedFinished(false)
michael@0:   , mNotifiedBlocked(false)
michael@0:   , mHasCurrentData(false)
michael@0:   , mNotifiedHasCurrentData(false)
michael@0:   , mWrapper(aWrapper)
michael@0:   , mMainThreadCurrentTime(0)
michael@0:   , mMainThreadFinished(false)
michael@0:   , mMainThreadDestroyed(false)
michael@0:   , mGraph(nullptr)
michael@0:   , mAudioChannelType(dom::AudioChannel::Normal)
michael@0: {
michael@0:   MOZ_COUNT_CTOR(MediaStream);
michael@0:   // aWrapper should not already be connected to a MediaStream! It needs
michael@0:   // to be hooked up to this stream, and since this stream is only just
michael@0:   // being created now, aWrapper must not be connected to anything.
michael@0:   NS_ASSERTION(!aWrapper || !aWrapper->GetStream(),
michael@0:                "Wrapper already has another media stream hooked up to it!");
michael@0: }
michael@0: 
michael@0: size_t
michael@0: MediaStream::SizeOfExcludingThis(MallocSizeOf aMallocSizeOf) const
michael@0: {
michael@0:   size_t amount = 0;
michael@0: 
michael@0:   // Not owned:
michael@0:   // - mGraph - Not reported here
michael@0:   // - mConsumers - elements
michael@0:   // Future:
michael@0:   // - mWrapper
michael@0:   // - mVideoOutputs - elements
michael@0:   // - mLastPlayedVideoFrame
michael@0:   // - mListeners - elements
michael@0:   // - mAudioOutputStreams - elements
michael@0: 
michael@0:   amount += mBuffer.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mAudioOutputs.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mVideoOutputs.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mExplicitBlockerCount.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mListeners.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mMainThreadListeners.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mDisabledTrackIDs.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mBlocked.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mGraphUpdateIndices.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mConsumers.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   amount += mAudioOutputStreams.SizeOfExcludingThis(aMallocSizeOf);
michael@0:   for (size_t i = 0; i < mAudioOutputStreams.Length(); i++) {
michael@0:     amount += mAudioOutputStreams[i].SizeOfExcludingThis(aMallocSizeOf);
michael@0:   }
michael@0: 
michael@0:   return amount;
michael@0: }
michael@0: 
michael@0: size_t
michael@0: MediaStream::SizeOfIncludingThis(MallocSizeOf aMallocSizeOf) const
michael@0: {
michael@0:   return aMallocSizeOf(this) + SizeOfExcludingThis(aMallocSizeOf);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::Init()
michael@0: {
michael@0:   MediaStreamGraphImpl* graph = GraphImpl();
michael@0:   mBlocked.SetAtAndAfter(graph->mCurrentTime, true);
michael@0:   mExplicitBlockerCount.SetAtAndAfter(graph->mCurrentTime, true);
michael@0:   mExplicitBlockerCount.SetAtAndAfter(graph->mStateComputedTime, false);
michael@0: }
michael@0: 
michael@0: MediaStreamGraphImpl*
michael@0: MediaStream::GraphImpl()
michael@0: {
michael@0:   return mGraph;
michael@0: }
michael@0: 
michael@0: MediaStreamGraph*
michael@0: MediaStream::Graph()
michael@0: {
michael@0:   return mGraph;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetGraphImpl(MediaStreamGraphImpl* aGraph)
michael@0: {
michael@0:   MOZ_ASSERT(!mGraph, "Should only be called once");
michael@0:   mGraph = aGraph;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetGraphImpl(MediaStreamGraph* aGraph)
michael@0: {
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(aGraph);
michael@0:   SetGraphImpl(graph);
michael@0: }
michael@0: 
michael@0: StreamTime
michael@0: MediaStream::GraphTimeToStreamTime(GraphTime aTime)
michael@0: {
michael@0:   return GraphImpl()->GraphTimeToStreamTime(this, aTime);
michael@0: }
michael@0: 
michael@0: StreamTime
michael@0: MediaStream::GraphTimeToStreamTimeOptimistic(GraphTime aTime)
michael@0: {
michael@0:   return GraphImpl()->GraphTimeToStreamTimeOptimistic(this, aTime);
michael@0: }
michael@0: 
michael@0: GraphTime
michael@0: MediaStream::StreamTimeToGraphTime(StreamTime aTime)
michael@0: {
michael@0:   return GraphImpl()->StreamTimeToGraphTime(this, aTime, 0);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::FinishOnGraphThread()
michael@0: {
michael@0:   GraphImpl()->FinishStream(this);
michael@0: }
michael@0: 
michael@0: int64_t
michael@0: MediaStream::GetProcessingGraphUpdateIndex()
michael@0: {
michael@0:   return GraphImpl()->GetProcessingGraphUpdateIndex();
michael@0: }
michael@0: 
michael@0: StreamBuffer::Track*
michael@0: MediaStream::EnsureTrack(TrackID aTrackId, TrackRate aSampleRate)
michael@0: {
michael@0:   StreamBuffer::Track* track = mBuffer.FindTrack(aTrackId);
michael@0:   if (!track) {
michael@0:     nsAutoPtr<MediaSegment> segment(new AudioSegment());
michael@0:     for (uint32_t j = 0; j < mListeners.Length(); ++j) {
michael@0:       MediaStreamListener* l = mListeners[j];
michael@0:       l->NotifyQueuedTrackChanges(Graph(), aTrackId,
michael@0:                                   GraphImpl()->AudioSampleRate(), 0,
michael@0:                                   MediaStreamListener::TRACK_EVENT_CREATED,
michael@0:                                   *segment);
michael@0:     }
michael@0:     track = &mBuffer.AddTrack(aTrackId, aSampleRate, 0, segment.forget());
michael@0:   }
michael@0:   return track;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveAllListenersImpl()
michael@0: {
michael@0:   for (int32_t i = mListeners.Length() - 1; i >= 0; --i) {
michael@0:     nsRefPtr<MediaStreamListener> listener = mListeners[i].forget();
michael@0:     listener->NotifyRemoved(GraphImpl());
michael@0:   }
michael@0:   mListeners.Clear();
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::DestroyImpl()
michael@0: {
michael@0:   for (int32_t i = mConsumers.Length() - 1; i >= 0; --i) {
michael@0:     mConsumers[i]->Disconnect();
michael@0:   }
michael@0:   for (uint32_t i = 0; i < mAudioOutputStreams.Length(); ++i) {
michael@0:     mAudioOutputStreams[i].mStream->Shutdown();
michael@0:   }
michael@0:   mAudioOutputStreams.Clear();
michael@0:   mGraph = nullptr;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::Destroy()
michael@0: {
michael@0:   // Keep this stream alive until we leave this method
michael@0:   nsRefPtr<MediaStream> kungFuDeathGrip = this;
michael@0: 
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream) : ControlMessage(aStream) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->RemoveAllListenersImpl();
michael@0:       auto graph = mStream->GraphImpl();
michael@0:       mStream->DestroyImpl();
michael@0:       graph->RemoveStream(mStream);
michael@0:     }
michael@0:     virtual void RunDuringShutdown()
michael@0:     { Run(); }
michael@0:   };
michael@0:   mWrapper = nullptr;
michael@0:   GraphImpl()->AppendMessage(new Message(this));
michael@0:   // Message::RunDuringShutdown may have removed this stream from the graph,
michael@0:   // but our kungFuDeathGrip above will have kept this stream alive if
michael@0:   // necessary.
michael@0:   mMainThreadDestroyed = true;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::AddAudioOutput(void* aKey)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, void* aKey) : ControlMessage(aStream), mKey(aKey) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->AddAudioOutputImpl(mKey);
michael@0:     }
michael@0:     void* mKey;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aKey));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetAudioOutputVolumeImpl(void* aKey, float aVolume)
michael@0: {
michael@0:   for (uint32_t i = 0; i < mAudioOutputs.Length(); ++i) {
michael@0:     if (mAudioOutputs[i].mKey == aKey) {
michael@0:       mAudioOutputs[i].mVolume = aVolume;
michael@0:       return;
michael@0:     }
michael@0:   }
michael@0:   NS_ERROR("Audio output key not found");
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetAudioOutputVolume(void* aKey, float aVolume)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, void* aKey, float aVolume) :
michael@0:       ControlMessage(aStream), mKey(aKey), mVolume(aVolume) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->SetAudioOutputVolumeImpl(mKey, mVolume);
michael@0:     }
michael@0:     void* mKey;
michael@0:     float mVolume;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aKey, aVolume));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveAudioOutputImpl(void* aKey)
michael@0: {
michael@0:   for (uint32_t i = 0; i < mAudioOutputs.Length(); ++i) {
michael@0:     if (mAudioOutputs[i].mKey == aKey) {
michael@0:       mAudioOutputs.RemoveElementAt(i);
michael@0:       return;
michael@0:     }
michael@0:   }
michael@0:   NS_ERROR("Audio output key not found");
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveAudioOutput(void* aKey)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, void* aKey) :
michael@0:       ControlMessage(aStream), mKey(aKey) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->RemoveAudioOutputImpl(mKey);
michael@0:     }
michael@0:     void* mKey;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aKey));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::AddVideoOutput(VideoFrameContainer* aContainer)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, VideoFrameContainer* aContainer) :
michael@0:       ControlMessage(aStream), mContainer(aContainer) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->AddVideoOutputImpl(mContainer.forget());
michael@0:     }
michael@0:     nsRefPtr<VideoFrameContainer> mContainer;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aContainer));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveVideoOutput(VideoFrameContainer* aContainer)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, VideoFrameContainer* aContainer) :
michael@0:       ControlMessage(aStream), mContainer(aContainer) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->RemoveVideoOutputImpl(mContainer);
michael@0:     }
michael@0:     nsRefPtr<VideoFrameContainer> mContainer;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aContainer));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::ChangeExplicitBlockerCount(int32_t aDelta)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, int32_t aDelta) :
michael@0:       ControlMessage(aStream), mDelta(aDelta) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->ChangeExplicitBlockerCountImpl(
michael@0:           mStream->GraphImpl()->mStateComputedTime, mDelta);
michael@0:     }
michael@0:     int32_t mDelta;
michael@0:   };
michael@0: 
michael@0:   // This can happen if this method has been called asynchronously, and the
michael@0:   // stream has been destroyed since then.
michael@0:   if (mMainThreadDestroyed) {
michael@0:     return;
michael@0:   }
michael@0:   GraphImpl()->AppendMessage(new Message(this, aDelta));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::AddListenerImpl(already_AddRefed<MediaStreamListener> aListener)
michael@0: {
michael@0:   MediaStreamListener* listener = *mListeners.AppendElement() = aListener;
michael@0:   listener->NotifyBlockingChanged(GraphImpl(),
michael@0:     mNotifiedBlocked ? MediaStreamListener::BLOCKED : MediaStreamListener::UNBLOCKED);
michael@0:   if (mNotifiedFinished) {
michael@0:     listener->NotifyFinished(GraphImpl());
michael@0:   }
michael@0:   if (mNotifiedHasCurrentData) {
michael@0:     listener->NotifyHasCurrentData(GraphImpl());
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::AddListener(MediaStreamListener* aListener)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, MediaStreamListener* aListener) :
michael@0:       ControlMessage(aStream), mListener(aListener) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->AddListenerImpl(mListener.forget());
michael@0:     }
michael@0:     nsRefPtr<MediaStreamListener> mListener;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aListener));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveListenerImpl(MediaStreamListener* aListener)
michael@0: {
michael@0:   // wouldn't need this if we could do it in the opposite order
michael@0:   nsRefPtr<MediaStreamListener> listener(aListener);
michael@0:   mListeners.RemoveElement(aListener);
michael@0:   listener->NotifyRemoved(GraphImpl());
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RemoveListener(MediaStreamListener* aListener)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, MediaStreamListener* aListener) :
michael@0:       ControlMessage(aStream), mListener(aListener) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->RemoveListenerImpl(mListener);
michael@0:     }
michael@0:     nsRefPtr<MediaStreamListener> mListener;
michael@0:   };
michael@0:   // If the stream is destroyed the Listeners have or will be
michael@0:   // removed.
michael@0:   if (!IsDestroyed()) {
michael@0:     GraphImpl()->AppendMessage(new Message(this, aListener));
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::RunAfterPendingUpdates(nsRefPtr<nsIRunnable> aRunnable)
michael@0: {
michael@0:   MOZ_ASSERT(NS_IsMainThread());
michael@0:   MediaStreamGraphImpl* graph = GraphImpl();
michael@0: 
michael@0:   // Special case when a non-realtime graph has not started, to ensure the
michael@0:   // runnable will run in finite time.
michael@0:   if (!(graph->mRealtime || graph->mNonRealtimeProcessing)) {
michael@0:     aRunnable->Run();
michael@0:   }
michael@0: 
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     explicit Message(MediaStream* aStream,
michael@0:                      already_AddRefed<nsIRunnable> aRunnable)
michael@0:       : ControlMessage(aStream)
michael@0:       , mRunnable(aRunnable) {}
michael@0:     virtual void Run() MOZ_OVERRIDE
michael@0:     {
michael@0:       mStream->Graph()->
michael@0:         DispatchToMainThreadAfterStreamStateUpdate(mRunnable.forget());
michael@0:     }
michael@0:     virtual void RunDuringShutdown() MOZ_OVERRIDE
michael@0:     {
michael@0:       // Don't run mRunnable now as it may call AppendMessage() which would
michael@0:       // assume that there are no remaining controlMessagesToRunDuringShutdown.
michael@0:       MOZ_ASSERT(NS_IsMainThread());
michael@0:       NS_DispatchToCurrentThread(mRunnable);
michael@0:     }
michael@0:   private:
michael@0:     nsRefPtr<nsIRunnable> mRunnable;
michael@0:   };
michael@0: 
michael@0:   graph->AppendMessage(new Message(this, aRunnable.forget()));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetTrackEnabledImpl(TrackID aTrackID, bool aEnabled)
michael@0: {
michael@0:   if (aEnabled) {
michael@0:     mDisabledTrackIDs.RemoveElement(aTrackID);
michael@0:   } else {
michael@0:     if (!mDisabledTrackIDs.Contains(aTrackID)) {
michael@0:       mDisabledTrackIDs.AppendElement(aTrackID);
michael@0:     }
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::SetTrackEnabled(TrackID aTrackID, bool aEnabled)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaStream* aStream, TrackID aTrackID, bool aEnabled) :
michael@0:       ControlMessage(aStream), mTrackID(aTrackID), mEnabled(aEnabled) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->SetTrackEnabledImpl(mTrackID, mEnabled);
michael@0:     }
michael@0:     TrackID mTrackID;
michael@0:     bool mEnabled;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aTrackID, aEnabled));
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStream::ApplyTrackDisabling(TrackID aTrackID, MediaSegment* aSegment, MediaSegment* aRawSegment)
michael@0: {
michael@0:   // mMutex must be owned here if this is a SourceMediaStream
michael@0:   if (!mDisabledTrackIDs.Contains(aTrackID)) {
michael@0:     return;
michael@0:   }
michael@0:   aSegment->ReplaceWithDisabled();
michael@0:   if (aRawSegment) {
michael@0:     aRawSegment->ReplaceWithDisabled();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::DestroyImpl()
michael@0: {
michael@0:   // Hold mMutex while mGraph is reset so that other threads holding mMutex
michael@0:   // can null-check know that the graph will not destroyed.
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   MediaStream::DestroyImpl();
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::SetPullEnabled(bool aEnabled)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   mPullEnabled = aEnabled;
michael@0:   if (mPullEnabled && GraphImpl()) {
michael@0:     GraphImpl()->EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::AddTrack(TrackID aID, TrackRate aRate, TrackTicks aStart,
michael@0:                             MediaSegment* aSegment)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   TrackData* data = mUpdateTracks.AppendElement();
michael@0:   data->mID = aID;
michael@0:   data->mInputRate = aRate;
michael@0:   // We resample all audio input tracks to the sample rate of the audio mixer.
michael@0:   data->mOutputRate = aSegment->GetType() == MediaSegment::AUDIO ?
michael@0:                       GraphImpl()->AudioSampleRate() : aRate;
michael@0:   data->mStart = aStart;
michael@0:   data->mCommands = TRACK_CREATE;
michael@0:   data->mData = aSegment;
michael@0:   data->mHaveEnough = false;
michael@0:   if (auto graph = GraphImpl()) {
michael@0:     graph->EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::ResampleAudioToGraphSampleRate(TrackData* aTrackData, MediaSegment* aSegment)
michael@0: {
michael@0:   if (aSegment->GetType() != MediaSegment::AUDIO ||
michael@0:       aTrackData->mInputRate == GraphImpl()->AudioSampleRate()) {
michael@0:     return;
michael@0:   }
michael@0:   AudioSegment* segment = static_cast<AudioSegment*>(aSegment);
michael@0:   if (!aTrackData->mResampler) {
michael@0:     int channels = segment->ChannelCount();
michael@0: 
michael@0:     // If this segment is just silence, we delay instanciating the resampler.
michael@0:     if (channels) {
michael@0:       SpeexResamplerState* state = speex_resampler_init(channels,
michael@0:                                                         aTrackData->mInputRate,
michael@0:                                                         GraphImpl()->AudioSampleRate(),
michael@0:                                                         SPEEX_RESAMPLER_QUALITY_DEFAULT,
michael@0:                                                         nullptr);
michael@0:       if (!state) {
michael@0:         return;
michael@0:       }
michael@0:       aTrackData->mResampler.own(state);
michael@0:     }
michael@0:   }
michael@0:   segment->ResampleChunks(aTrackData->mResampler);
michael@0: }
michael@0: 
michael@0: bool
michael@0: SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment, MediaSegment *aRawSegment)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   // ::EndAllTrackAndFinished() can end these before the sources notice
michael@0:   bool appended = false;
michael@0:   auto graph = GraphImpl();
michael@0:   if (!mFinished && graph) {
michael@0:     TrackData *track = FindDataForTrack(aID);
michael@0:     if (track) {
michael@0:       // Data goes into mData, and on the next iteration of the MSG moves
michael@0:       // into the track's segment after NotifyQueuedTrackChanges().  This adds
michael@0:       // 0-10ms of delay before data gets to direct listeners.
michael@0:       // Indirect listeners (via subsequent TrackUnion nodes) are synced to
michael@0:       // playout time, and so can be delayed by buffering.
michael@0: 
michael@0:       // Apply track disabling before notifying any consumers directly
michael@0:       // or inserting into the graph
michael@0:       ApplyTrackDisabling(aID, aSegment, aRawSegment);
michael@0: 
michael@0:       ResampleAudioToGraphSampleRate(track, aSegment);
michael@0: 
michael@0:       // Must notify first, since AppendFrom() will empty out aSegment
michael@0:       NotifyDirectConsumers(track, aRawSegment ? aRawSegment : aSegment);
michael@0:       track->mData->AppendFrom(aSegment); // note: aSegment is now dead
michael@0:       appended = true;
michael@0:       graph->EnsureNextIteration();
michael@0:     } else {
michael@0:       aSegment->Clear();
michael@0:     }
michael@0:   }
michael@0:   return appended;
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::NotifyDirectConsumers(TrackData *aTrack,
michael@0:                                          MediaSegment *aSegment)
michael@0: {
michael@0:   // Call with mMutex locked
michael@0:   MOZ_ASSERT(aTrack);
michael@0: 
michael@0:   for (uint32_t j = 0; j < mDirectListeners.Length(); ++j) {
michael@0:     MediaStreamDirectListener* l = mDirectListeners[j];
michael@0:     TrackTicks offset = 0; // FIX! need a separate TrackTicks.... or the end of the internal buffer
michael@0:     l->NotifyRealtimeData(static_cast<MediaStreamGraph*>(GraphImpl()), aTrack->mID, aTrack->mOutputRate,
michael@0:                           offset, aTrack->mCommands, *aSegment);
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::AddDirectListener(MediaStreamDirectListener* aListener)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   mDirectListeners.AppendElement(aListener);
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::RemoveDirectListener(MediaStreamDirectListener* aListener)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   mDirectListeners.RemoveElement(aListener);
michael@0: }
michael@0: 
michael@0: bool
michael@0: SourceMediaStream::HaveEnoughBuffered(TrackID aID)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   TrackData *track = FindDataForTrack(aID);
michael@0:   if (track) {
michael@0:     return track->mHaveEnough;
michael@0:   }
michael@0:   return false;
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::DispatchWhenNotEnoughBuffered(TrackID aID,
michael@0:     nsIEventTarget* aSignalThread, nsIRunnable* aSignalRunnable)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   TrackData* data = FindDataForTrack(aID);
michael@0:   if (!data) {
michael@0:     aSignalThread->Dispatch(aSignalRunnable, 0);
michael@0:     return;
michael@0:   }
michael@0: 
michael@0:   if (data->mHaveEnough) {
michael@0:     if (data->mDispatchWhenNotEnough.IsEmpty()) {
michael@0:       data->mDispatchWhenNotEnough.AppendElement()->Init(aSignalThread, aSignalRunnable);
michael@0:     }
michael@0:   } else {
michael@0:     aSignalThread->Dispatch(aSignalRunnable, 0);
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::EndTrack(TrackID aID)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   // ::EndAllTrackAndFinished() can end these before the sources call this
michael@0:   if (!mFinished) {
michael@0:     TrackData *track = FindDataForTrack(aID);
michael@0:     if (track) {
michael@0:       track->mCommands |= TRACK_END;
michael@0:     }
michael@0:   }
michael@0:   if (auto graph = GraphImpl()) {
michael@0:     graph->EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::AdvanceKnownTracksTime(StreamTime aKnownTime)
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   MOZ_ASSERT(aKnownTime >= mUpdateKnownTracksTime);
michael@0:   mUpdateKnownTracksTime = aKnownTime;
michael@0:   if (auto graph = GraphImpl()) {
michael@0:     graph->EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::FinishWithLockHeld()
michael@0: {
michael@0:   mMutex.AssertCurrentThreadOwns();
michael@0:   mUpdateFinished = true;
michael@0:   if (auto graph = GraphImpl()) {
michael@0:     graph->EnsureNextIteration();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::EndAllTrackAndFinish()
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   for (uint32_t i = 0; i < mUpdateTracks.Length(); ++i) {
michael@0:     SourceMediaStream::TrackData* data = &mUpdateTracks[i];
michael@0:     data->mCommands |= TRACK_END;
michael@0:   }
michael@0:   FinishWithLockHeld();
michael@0:   // we will call NotifyFinished() to let GetUserMedia know
michael@0: }
michael@0: 
michael@0: TrackTicks
michael@0: SourceMediaStream::GetBufferedTicks(TrackID aID)
michael@0: {
michael@0:   StreamBuffer::Track* track  = mBuffer.FindTrack(aID);
michael@0:   if (track) {
michael@0:     MediaSegment* segment = track->GetSegment();
michael@0:     if (segment) {
michael@0:       return segment->GetDuration() -
michael@0:         track->TimeToTicksRoundDown(
michael@0:           GraphTimeToStreamTime(GraphImpl()->mStateComputedTime));
michael@0:     }
michael@0:   }
michael@0:   return 0;
michael@0: }
michael@0: 
michael@0: void
michael@0: SourceMediaStream::RegisterForAudioMixing()
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   mNeedsMixing = true;
michael@0: }
michael@0: 
michael@0: bool
michael@0: SourceMediaStream::NeedsMixing()
michael@0: {
michael@0:   MutexAutoLock lock(mMutex);
michael@0:   return mNeedsMixing;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaInputPort::Init()
michael@0: {
michael@0:   STREAM_LOG(PR_LOG_DEBUG, ("Adding MediaInputPort %p (from %p to %p) to the graph",
michael@0:              this, mSource, mDest));
michael@0:   mSource->AddConsumer(this);
michael@0:   mDest->AddInput(this);
michael@0:   // mPortCount decremented via MediaInputPort::Destroy's message
michael@0:   ++mDest->GraphImpl()->mPortCount;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaInputPort::Disconnect()
michael@0: {
michael@0:   NS_ASSERTION(!mSource == !mDest,
michael@0:                "mSource must either both be null or both non-null");
michael@0:   if (!mSource)
michael@0:     return;
michael@0: 
michael@0:   mSource->RemoveConsumer(this);
michael@0:   mSource = nullptr;
michael@0:   mDest->RemoveInput(this);
michael@0:   mDest = nullptr;
michael@0: 
michael@0:   GraphImpl()->SetStreamOrderDirty();
michael@0: }
michael@0: 
michael@0: MediaInputPort::InputInterval
michael@0: MediaInputPort::GetNextInputInterval(GraphTime aTime)
michael@0: {
michael@0:   InputInterval result = { GRAPH_TIME_MAX, GRAPH_TIME_MAX, false };
michael@0:   GraphTime t = aTime;
michael@0:   GraphTime end;
michael@0:   for (;;) {
michael@0:     if (!mDest->mBlocked.GetAt(t, &end))
michael@0:       break;
michael@0:     if (end == GRAPH_TIME_MAX)
michael@0:       return result;
michael@0:     t = end;
michael@0:   }
michael@0:   result.mStart = t;
michael@0:   GraphTime sourceEnd;
michael@0:   result.mInputIsBlocked = mSource->mBlocked.GetAt(t, &sourceEnd);
michael@0:   result.mEnd = std::min(end, sourceEnd);
michael@0:   return result;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaInputPort::Destroy()
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaInputPort* aPort)
michael@0:       : ControlMessage(nullptr), mPort(aPort) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mPort->Disconnect();
michael@0:       --mPort->GraphImpl()->mPortCount;
michael@0:       mPort->SetGraphImpl(nullptr);
michael@0:       NS_RELEASE(mPort);
michael@0:     }
michael@0:     virtual void RunDuringShutdown()
michael@0:     {
michael@0:       Run();
michael@0:     }
michael@0:     MediaInputPort* mPort;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this));
michael@0: }
michael@0: 
michael@0: MediaStreamGraphImpl*
michael@0: MediaInputPort::GraphImpl()
michael@0: {
michael@0:   return mGraph;
michael@0: }
michael@0: 
michael@0: MediaStreamGraph*
michael@0: MediaInputPort::Graph()
michael@0: {
michael@0:   return mGraph;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaInputPort::SetGraphImpl(MediaStreamGraphImpl* aGraph)
michael@0: {
michael@0:   MOZ_ASSERT(!mGraph || !aGraph, "Should only be set once");
michael@0:   mGraph = aGraph;
michael@0: }
michael@0: 
michael@0: already_AddRefed<MediaInputPort>
michael@0: ProcessedMediaStream::AllocateInputPort(MediaStream* aStream, uint32_t aFlags,
michael@0:                                         uint16_t aInputNumber, uint16_t aOutputNumber)
michael@0: {
michael@0:   // This method creates two references to the MediaInputPort: one for
michael@0:   // the main thread, and one for the MediaStreamGraph.
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(MediaInputPort* aPort)
michael@0:       : ControlMessage(aPort->GetDestination()),
michael@0:         mPort(aPort) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mPort->Init();
michael@0:       // The graph holds its reference implicitly
michael@0:       mPort->GraphImpl()->SetStreamOrderDirty();
michael@0:       unused << mPort.forget();
michael@0:     }
michael@0:     virtual void RunDuringShutdown()
michael@0:     {
michael@0:       Run();
michael@0:     }
michael@0:     nsRefPtr<MediaInputPort> mPort;
michael@0:   };
michael@0:   nsRefPtr<MediaInputPort> port = new MediaInputPort(aStream, this, aFlags,
michael@0:                                                      aInputNumber, aOutputNumber);
michael@0:   port->SetGraphImpl(GraphImpl());
michael@0:   GraphImpl()->AppendMessage(new Message(port));
michael@0:   return port.forget();
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessedMediaStream::Finish()
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(ProcessedMediaStream* aStream)
michael@0:       : ControlMessage(aStream) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       mStream->GraphImpl()->FinishStream(mStream);
michael@0:     }
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this));
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessedMediaStream::SetAutofinish(bool aAutofinish)
michael@0: {
michael@0:   class Message : public ControlMessage {
michael@0:   public:
michael@0:     Message(ProcessedMediaStream* aStream, bool aAutofinish)
michael@0:       : ControlMessage(aStream), mAutofinish(aAutofinish) {}
michael@0:     virtual void Run()
michael@0:     {
michael@0:       static_cast<ProcessedMediaStream*>(mStream)->SetAutofinishImpl(mAutofinish);
michael@0:     }
michael@0:     bool mAutofinish;
michael@0:   };
michael@0:   GraphImpl()->AppendMessage(new Message(this, aAutofinish));
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessedMediaStream::DestroyImpl()
michael@0: {
michael@0:   for (int32_t i = mInputs.Length() - 1; i >= 0; --i) {
michael@0:     mInputs[i]->Disconnect();
michael@0:   }
michael@0:   MediaStream::DestroyImpl();
michael@0:   // The stream order is only important if there are connections, in which
michael@0:   // case MediaInputPort::Disconnect() called SetStreamOrderDirty().
michael@0:   // MediaStreamGraphImpl::RemoveStream() will also call
michael@0:   // SetStreamOrderDirty(), for other reasons.
michael@0: }
michael@0: 
michael@0: /**
michael@0:  * We make the initial mCurrentTime nonzero so that zero times can have
michael@0:  * special meaning if necessary.
michael@0:  */
michael@0: static const int32_t INITIAL_CURRENT_TIME = 1;
michael@0: 
michael@0: MediaStreamGraphImpl::MediaStreamGraphImpl(bool aRealtime, TrackRate aSampleRate)
michael@0:   : mCurrentTime(INITIAL_CURRENT_TIME)
michael@0:   , mStateComputedTime(INITIAL_CURRENT_TIME)
michael@0:   , mProcessingGraphUpdateIndex(0)
michael@0:   , mPortCount(0)
michael@0:   , mMonitor("MediaStreamGraphImpl")
michael@0:   , mLifecycleState(LIFECYCLE_THREAD_NOT_STARTED)
michael@0:   , mWaitState(WAITSTATE_RUNNING)
michael@0:   , mEndTime(GRAPH_TIME_MAX)
michael@0:   , mSampleRate(aSampleRate)
michael@0:   , mNeedAnotherIteration(false)
michael@0:   , mForceShutDown(false)
michael@0:   , mPostedRunInStableStateEvent(false)
michael@0:   , mDetectedNotRunning(false)
michael@0:   , mPostedRunInStableState(false)
michael@0:   , mRealtime(aRealtime)
michael@0:   , mNonRealtimeProcessing(false)
michael@0:   , mStreamOrderDirty(false)
michael@0:   , mLatencyLog(AsyncLatencyLogger::Get())
michael@0:   , mMixer(nullptr)
michael@0:   , mMemoryReportMonitor("MSGIMemory")
michael@0:   , mSelfRef(MOZ_THIS_IN_INITIALIZER_LIST())
michael@0:   , mAudioStreamSizes()
michael@0:   , mNeedsMemoryReport(false)
michael@0: #ifdef DEBUG
michael@0:   , mCanRunMessagesSynchronously(false)
michael@0: #endif
michael@0: {
michael@0: #ifdef PR_LOGGING
michael@0:   if (!gMediaStreamGraphLog) {
michael@0:     gMediaStreamGraphLog = PR_NewLogModule("MediaStreamGraph");
michael@0:   }
michael@0: #endif
michael@0: 
michael@0:   mCurrentTimeStamp = mInitialTimeStamp = mLastMainThreadUpdate = TimeStamp::Now();
michael@0: 
michael@0:   RegisterWeakMemoryReporter(this);
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraphImpl::Destroy()
michael@0: {
michael@0:   // First unregister from memory reporting.
michael@0:   UnregisterWeakMemoryReporter(this);
michael@0: 
michael@0:   // Clear the self reference which will destroy this instance.
michael@0:   mSelfRef = nullptr;
michael@0: }
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(MediaStreamGraphShutdownObserver, nsIObserver)
michael@0: 
michael@0: static bool gShutdownObserverRegistered = false;
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: MediaStreamGraphShutdownObserver::Observe(nsISupports *aSubject,
michael@0:                                           const char *aTopic,
michael@0:                                           const char16_t *aData)
michael@0: {
michael@0:   if (strcmp(aTopic, NS_XPCOM_SHUTDOWN_OBSERVER_ID) == 0) {
michael@0:     if (gGraph) {
michael@0:       gGraph->ForceShutDown();
michael@0:     }
michael@0:     nsContentUtils::UnregisterShutdownObserver(this);
michael@0:     gShutdownObserverRegistered = false;
michael@0:   }
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: MediaStreamGraph*
michael@0: MediaStreamGraph::GetInstance()
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Main thread only");
michael@0: 
michael@0:   if (!gGraph) {
michael@0:     if (!gShutdownObserverRegistered) {
michael@0:       gShutdownObserverRegistered = true;
michael@0:       nsContentUtils::RegisterShutdownObserver(new MediaStreamGraphShutdownObserver());
michael@0:     }
michael@0: 
michael@0:     AudioStream::InitPreferredSampleRate();
michael@0: 
michael@0:     gGraph = new MediaStreamGraphImpl(true, AudioStream::PreferredSampleRate());
michael@0: 
michael@0:     STREAM_LOG(PR_LOG_DEBUG, ("Starting up MediaStreamGraph %p", gGraph));
michael@0:   }
michael@0: 
michael@0:   return gGraph;
michael@0: }
michael@0: 
michael@0: MediaStreamGraph*
michael@0: MediaStreamGraph::CreateNonRealtimeInstance(TrackRate aSampleRate)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Main thread only");
michael@0: 
michael@0:   MediaStreamGraphImpl* graph = new MediaStreamGraphImpl(false, aSampleRate);
michael@0: 
michael@0:   return graph;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraph::DestroyNonRealtimeInstance(MediaStreamGraph* aGraph)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "Main thread only");
michael@0:   MOZ_ASSERT(aGraph->IsNonRealtime(), "Should not destroy the global graph here");
michael@0: 
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(aGraph);
michael@0:   if (graph->mForceShutDown)
michael@0:     return; // already done
michael@0: 
michael@0:   if (!graph->mNonRealtimeProcessing) {
michael@0:     // Start the graph, but don't produce anything
michael@0:     graph->StartNonRealtimeProcessing(1, 0);
michael@0:   }
michael@0:   graph->ForceShutDown();
michael@0: }
michael@0: 
michael@0: NS_IMPL_ISUPPORTS(MediaStreamGraphImpl, nsIMemoryReporter)
michael@0: 
michael@0: struct ArrayClearer
michael@0: {
michael@0:   ArrayClearer(nsTArray<AudioNodeSizes>& aArray) : mArray(aArray) {}
michael@0:   ~ArrayClearer() { mArray.Clear(); }
michael@0:   nsTArray<AudioNodeSizes>& mArray;
michael@0: };
michael@0: 
michael@0: NS_IMETHODIMP
michael@0: MediaStreamGraphImpl::CollectReports(nsIHandleReportCallback* aHandleReport,
michael@0:                                      nsISupports* aData)
michael@0: {
michael@0:   // Clears out the report array after we're done with it.
michael@0:   ArrayClearer reportCleanup(mAudioStreamSizes);
michael@0: 
michael@0:   {
michael@0:     MonitorAutoLock memoryReportLock(mMemoryReportMonitor);
michael@0:     mNeedsMemoryReport = true;
michael@0: 
michael@0:     {
michael@0:       // Wake up the MSG thread.
michael@0:       MonitorAutoLock monitorLock(mMonitor);
michael@0:       EnsureImmediateWakeUpLocked(monitorLock);
michael@0:     }
michael@0: 
michael@0:     // Wait for the report to complete.
michael@0:     nsresult rv;
michael@0:     while ((rv = memoryReportLock.Wait()) != NS_OK) {
michael@0:       if (PR_GetError() != PR_PENDING_INTERRUPT_ERROR) {
michael@0:         return rv;
michael@0:       }
michael@0:     }
michael@0:   }
michael@0: 
michael@0: #define REPORT(_path, _amount, _desc)                                       \
michael@0:   do {                                                                      \
michael@0:     nsresult rv;                                                            \
michael@0:     rv = aHandleReport->Callback(EmptyCString(), _path,                     \
michael@0:                                  KIND_HEAP, UNITS_BYTES, _amount,           \
michael@0:                                  NS_LITERAL_CSTRING(_desc), aData);         \
michael@0:     NS_ENSURE_SUCCESS(rv, rv);                                              \
michael@0:   } while (0)
michael@0: 
michael@0:   for (size_t i = 0; i < mAudioStreamSizes.Length(); i++) {
michael@0:     const AudioNodeSizes& usage = mAudioStreamSizes[i];
michael@0:     const char* const nodeType =  usage.mNodeType.get();
michael@0: 
michael@0:     nsPrintfCString domNodePath("explicit/webaudio/audio-node/%s/dom-nodes",
michael@0:                                   nodeType);
michael@0:     REPORT(domNodePath, usage.mDomNode,
michael@0:            "Memory used by AudioNode DOM objects (Web Audio).");
michael@0: 
michael@0:     nsPrintfCString enginePath("explicit/webaudio/audio-node/%s/engine-objects",
michael@0:                                 nodeType);
michael@0:     REPORT(enginePath, usage.mEngine,
michael@0:            "Memory used by AudioNode engine objects (Web Audio).");
michael@0: 
michael@0:     nsPrintfCString streamPath("explicit/webaudio/audio-node/%s/stream-objects",
michael@0:                                 nodeType);
michael@0:     REPORT(streamPath, usage.mStream,
michael@0:            "Memory used by AudioNode stream objects (Web Audio).");
michael@0: 
michael@0:   }
michael@0: 
michael@0: #undef REPORT
michael@0: 
michael@0:   return NS_OK;
michael@0: }
michael@0: 
michael@0: SourceMediaStream*
michael@0: MediaStreamGraph::CreateSourceStream(DOMMediaStream* aWrapper)
michael@0: {
michael@0:   SourceMediaStream* stream = new SourceMediaStream(aWrapper);
michael@0:   NS_ADDREF(stream);
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
michael@0:   stream->SetGraphImpl(graph);
michael@0:   graph->AppendMessage(new CreateMessage(stream));
michael@0:   return stream;
michael@0: }
michael@0: 
michael@0: ProcessedMediaStream*
michael@0: MediaStreamGraph::CreateTrackUnionStream(DOMMediaStream* aWrapper)
michael@0: {
michael@0:   TrackUnionStream* stream = new TrackUnionStream(aWrapper);
michael@0:   NS_ADDREF(stream);
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
michael@0:   stream->SetGraphImpl(graph);
michael@0:   graph->AppendMessage(new CreateMessage(stream));
michael@0:   return stream;
michael@0: }
michael@0: 
michael@0: AudioNodeExternalInputStream*
michael@0: MediaStreamGraph::CreateAudioNodeExternalInputStream(AudioNodeEngine* aEngine, TrackRate aSampleRate)
michael@0: {
michael@0:   MOZ_ASSERT(NS_IsMainThread());
michael@0:   if (!aSampleRate) {
michael@0:     aSampleRate = aEngine->NodeMainThread()->Context()->SampleRate();
michael@0:   }
michael@0:   AudioNodeExternalInputStream* stream = new AudioNodeExternalInputStream(aEngine, aSampleRate);
michael@0:   NS_ADDREF(stream);
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
michael@0:   stream->SetGraphImpl(graph);
michael@0:   graph->AppendMessage(new CreateMessage(stream));
michael@0:   return stream;
michael@0: }
michael@0: 
michael@0: AudioNodeStream*
michael@0: MediaStreamGraph::CreateAudioNodeStream(AudioNodeEngine* aEngine,
michael@0:                                         AudioNodeStreamKind aKind,
michael@0:                                         TrackRate aSampleRate)
michael@0: {
michael@0:   MOZ_ASSERT(NS_IsMainThread());
michael@0:   if (!aSampleRate) {
michael@0:     aSampleRate = aEngine->NodeMainThread()->Context()->SampleRate();
michael@0:   }
michael@0:   AudioNodeStream* stream = new AudioNodeStream(aEngine, aKind, aSampleRate);
michael@0:   NS_ADDREF(stream);
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
michael@0:   stream->SetGraphImpl(graph);
michael@0:   if (aEngine->HasNode()) {
michael@0:     stream->SetChannelMixingParametersImpl(aEngine->NodeMainThread()->ChannelCount(),
michael@0:                                            aEngine->NodeMainThread()->ChannelCountModeValue(),
michael@0:                                            aEngine->NodeMainThread()->ChannelInterpretationValue());
michael@0:   }
michael@0:   graph->AppendMessage(new CreateMessage(stream));
michael@0:   return stream;
michael@0: }
michael@0: 
michael@0: bool
michael@0: MediaStreamGraph::IsNonRealtime() const
michael@0: {
michael@0:   return this != gGraph;
michael@0: }
michael@0: 
michael@0: void
michael@0: MediaStreamGraph::StartNonRealtimeProcessing(TrackRate aRate, uint32_t aTicksToProcess)
michael@0: {
michael@0:   NS_ASSERTION(NS_IsMainThread(), "main thread only");
michael@0: 
michael@0:   MediaStreamGraphImpl* graph = static_cast<MediaStreamGraphImpl*>(this);
michael@0:   NS_ASSERTION(!graph->mRealtime, "non-realtime only");
michael@0: 
michael@0:   if (graph->mNonRealtimeProcessing)
michael@0:     return;
michael@0:   graph->mEndTime = graph->mCurrentTime + TicksToTimeRoundUp(aRate, aTicksToProcess);
michael@0:   graph->mNonRealtimeProcessing = true;
michael@0:   graph->EnsureRunInStableState();
michael@0: }
michael@0: 
michael@0: void
michael@0: ProcessedMediaStream::AddInput(MediaInputPort* aPort)
michael@0: {
michael@0:   mInputs.AppendElement(aPort);
michael@0:   GraphImpl()->SetStreamOrderDirty();
michael@0: }
michael@0: 
michael@0: }