gecko-dev/dom/media/webaudio/ScriptProcessorNode.cpp

/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim:set ts=2 sw=2 sts=2 et cindent: */
/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "ScriptProcessorNode.h"
#include "mozilla/dom/ScriptProcessorNodeBinding.h"
#include "AudioBuffer.h"
#include "AudioDestinationNode.h"
#include "AudioNodeEngine.h"
#include "AudioNodeStream.h"
#include "AudioProcessingEvent.h"
#include "WebAudioUtils.h"
#include "mozilla/dom/ScriptSettings.h"
#include "mozilla/Mutex.h"
#include "mozilla/PodOperations.h"
#include "nsAutoPtr.h"
#include <deque>

namespace mozilla {
namespace dom {

// The maximum latency, in seconds, that we can live with before dropping
// buffers.
static const float MAX_LATENCY_S = 0.5;

// This class manages a queue of output buffers shared between
// the main thread and the Media Stream Graph thread.
class SharedBuffers final {
 private:
  class OutputQueue final {
   public:
    explicit OutputQueue(const char* aName) : mMutex(aName) {}

    size_t SizeOfExcludingThis(MallocSizeOf aMallocSizeOf) const {
      mMutex.AssertCurrentThreadOwns();

      size_t amount = 0;
      for (size_t i = 0; i < mBufferList.size(); i++) {
        amount += mBufferList[i].SizeOfExcludingThis(aMallocSizeOf, false);
      }

      return amount;
    }

    Mutex& Lock() const { return const_cast<OutputQueue*>(this)->mMutex; }

    size_t ReadyToConsume() const {
      // Accessed on both main thread and media graph thread.
      mMutex.AssertCurrentThreadOwns();
      return mBufferList.size();
    }

    // Produce one buffer
    AudioChunk& Produce() {
      mMutex.AssertCurrentThreadOwns();
      MOZ_ASSERT(NS_IsMainThread());
      mBufferList.push_back(AudioChunk());
      return mBufferList.back();
    }

    // Consumes one buffer.
    AudioChunk Consume() {
      mMutex.AssertCurrentThreadOwns();
      MOZ_ASSERT(!NS_IsMainThread());
      MOZ_ASSERT(ReadyToConsume() > 0);
      AudioChunk front = mBufferList.front();
      mBufferList.pop_front();
      return front;
    }

    // Empties the buffer queue.
    void Clear() {
      mMutex.AssertCurrentThreadOwns();
      mBufferList.clear();
    }

   private:
    typedef std::deque<AudioChunk> BufferList;

    // Synchronizes access to mBufferList.  Note that it's the responsibility
    // of the callers to perform the required locking, and we assert that every
    // time we access mBufferList.
    Mutex mMutex;
    // The list representing the queue.
    BufferList mBufferList;
  };

 public:
  explicit SharedBuffers(float aSampleRate)
      : mOutputQueue("SharedBuffers::outputQueue"),
        mDelaySoFar(STREAM_TIME_MAX),
        mSampleRate(aSampleRate),
        mLatency(0.0),
        mDroppingBuffers(false) {}

  size_t SizeOfIncludingThis(MallocSizeOf aMallocSizeOf) const {
    size_t amount = aMallocSizeOf(this);

    {
      MutexAutoLock lock(mOutputQueue.Lock());
      amount += mOutputQueue.SizeOfExcludingThis(aMallocSizeOf);
    }

    return amount;
  }

  // main thread

  // NotifyNodeIsConnected() may be called even when the state has not
  // changed.
  void NotifyNodeIsConnected(bool aIsConnected) {
    MOZ_ASSERT(NS_IsMainThread());
    if (!aIsConnected) {
      // Reset main thread state for FinishProducingOutputBuffer().
      mLatency = 0.0f;
      mLastEventTime = TimeStamp();
      mDroppingBuffers = false;
      // Don't flush the output buffer here because the graph thread may be
      // using it now.  The graph thread will flush when it knows it is
      // disconnected.
    }
    mNodeIsConnected = aIsConnected;
  }

  void FinishProducingOutputBuffer(const AudioChunk& aBuffer) {
    MOZ_ASSERT(NS_IsMainThread());

    if (!mNodeIsConnected) {
      // The output buffer is not used, and mLastEventTime will not be
      // initialized until the node is re-connected.
      return;
    }

    TimeStamp now = TimeStamp::Now();

    if (mLastEventTime.IsNull()) {
      mLastEventTime = now;
    } else {
      // When main thread blocking has built up enough so
      // |mLatency > MAX_LATENCY_S|, frame dropping starts. It continues until
      // the output buffer is completely empty, at which point the accumulated
      // latency is also reset to 0.
      // It could happen that the output queue becomes empty before the input
      // node has fully caught up. In this case there will be events where
      // |(now - mLastEventTime)| is very short, making mLatency negative.
      // As this happens and the size of |mLatency| becomes greater than
      // MAX_LATENCY_S, frame dropping starts again to maintain an as short
      // output queue as possible.
      float latency = (now - mLastEventTime).ToSeconds();
      float bufferDuration = aBuffer.mDuration / mSampleRate;
      mLatency += latency - bufferDuration;
      mLastEventTime = now;
      if (fabs(mLatency) > MAX_LATENCY_S) {
        mDroppingBuffers = true;
      }
    }

    MutexAutoLock lock(mOutputQueue.Lock());
    if (mDroppingBuffers) {
      if (mOutputQueue.ReadyToConsume()) {
        return;
      }
      mDroppingBuffers = false;
      mLatency = 0;
    }

    for (uint32_t offset = 0; offset < aBuffer.mDuration;
         offset += WEBAUDIO_BLOCK_SIZE) {
      AudioChunk& chunk = mOutputQueue.Produce();
      chunk = aBuffer;
      chunk.SliceTo(offset, offset + WEBAUDIO_BLOCK_SIZE);
    }
  }

  // graph thread

  AudioChunk GetOutputBuffer() {
    MOZ_ASSERT(!NS_IsMainThread());
    AudioChunk buffer;

    {
      MutexAutoLock lock(mOutputQueue.Lock());
      if (mOutputQueue.ReadyToConsume() > 0) {
        if (mDelaySoFar == STREAM_TIME_MAX) {
          mDelaySoFar = 0;
        }
        buffer = mOutputQueue.Consume();
      } else {
        // If we're out of buffers to consume, just output silence
        buffer.SetNull(WEBAUDIO_BLOCK_SIZE);
        if (mDelaySoFar != STREAM_TIME_MAX) {
          // Remember the delay that we just hit
          mDelaySoFar += WEBAUDIO_BLOCK_SIZE;
        }
      }
    }

    return buffer;
  }

  StreamTime DelaySoFar() const {
    MOZ_ASSERT(!NS_IsMainThread());
    return mDelaySoFar == STREAM_TIME_MAX ? 0 : mDelaySoFar;
  }

  void Flush() {
    MOZ_ASSERT(!NS_IsMainThread());
    mDelaySoFar = STREAM_TIME_MAX;
    {
      MutexAutoLock lock(mOutputQueue.Lock());
      mOutputQueue.Clear();
    }
  }

 private:
  OutputQueue mOutputQueue;
  // How much delay we've seen so far.  This measures the amount of delay
  // caused by the main thread lagging behind in producing output buffers.
  // STREAM_TIME_MAX means that we have not received our first buffer yet.
  // Graph thread only.
  StreamTime mDelaySoFar;
  // The samplerate of the context.
  const float mSampleRate;
  // The remaining members are main thread only.
  // This is the latency caused by the buffering. If this grows too high, we
  // will drop buffers until it is acceptable.
  float mLatency;
  // This is the time at which we last produced a buffer, to detect if the main
  // thread has been blocked.
  TimeStamp mLastEventTime;
  // True if we should be dropping buffers.
  bool mDroppingBuffers;
  // True iff the AudioNode has at least one input or output connected.
  bool mNodeIsConnected;
};

class ScriptProcessorNodeEngine final : public AudioNodeEngine {
 public:
  ScriptProcessorNodeEngine(ScriptProcessorNode* aNode,
                            AudioDestinationNode* aDestination,
                            uint32_t aBufferSize,
                            uint32_t aNumberOfInputChannels)
      : AudioNodeEngine(aNode),
        mDestination(aDestination->Stream()),
        mSharedBuffers(new SharedBuffers(mDestination->SampleRate())),
        mBufferSize(aBufferSize),
        mInputChannelCount(aNumberOfInputChannels),
        mInputWriteIndex(0) {}

  SharedBuffers* GetSharedBuffers() const { return mSharedBuffers; }

  enum {
    IS_CONNECTED,
  };

  void SetInt32Parameter(uint32_t aIndex, int32_t aParam) override {
    switch (aIndex) {
      case IS_CONNECTED:
        mIsConnected = aParam;
        break;
      default:
        NS_ERROR("Bad Int32Parameter");
    }  // End index switch.
  }

  void ProcessBlock(AudioNodeStream* aStream, GraphTime aFrom,
                    const AudioBlock& aInput, AudioBlock* aOutput,
                    bool* aFinished) override {
    // This node is not connected to anything. Per spec, we don't fire the
    // onaudioprocess event. We also want to clear out the input and output
    // buffer queue, and output a null buffer.
    if (!mIsConnected) {
      aOutput->SetNull(WEBAUDIO_BLOCK_SIZE);
      mSharedBuffers->Flush();
      mInputWriteIndex = 0;
      return;
    }

    // The input buffer is allocated lazily when non-null input is received.
    if (!aInput.IsNull() && !mInputBuffer) {
      mInputBuffer = ThreadSharedFloatArrayBufferList::Create(
          mInputChannelCount, mBufferSize, fallible);
      if (mInputBuffer && mInputWriteIndex) {
        // Zero leading for null chunks that were skipped.
        for (uint32_t i = 0; i < mInputChannelCount; ++i) {
          float* channelData = mInputBuffer->GetDataForWrite(i);
          PodZero(channelData, mInputWriteIndex);
        }
      }
    }

    // First, record our input buffer, if its allocation succeeded.
    uint32_t inputChannelCount = mInputBuffer ? mInputBuffer->GetChannels() : 0;
    for (uint32_t i = 0; i < inputChannelCount; ++i) {
      float* writeData = mInputBuffer->GetDataForWrite(i) + mInputWriteIndex;
      if (aInput.IsNull()) {
        PodZero(writeData, aInput.GetDuration());
      } else {
        MOZ_ASSERT(aInput.GetDuration() == WEBAUDIO_BLOCK_SIZE, "sanity check");
        MOZ_ASSERT(aInput.ChannelCount() == inputChannelCount);
        AudioBlockCopyChannelWithScale(
            static_cast<const float*>(aInput.mChannelData[i]), aInput.mVolume,
            writeData);
      }
    }
    mInputWriteIndex += aInput.GetDuration();

    // Now, see if we have data to output
    // Note that we need to do this before sending the buffer to the main
    // thread so that our delay time is updated.
    *aOutput = mSharedBuffers->GetOutputBuffer();

    if (mInputWriteIndex >= mBufferSize) {
      SendBuffersToMainThread(aStream, aFrom);
      mInputWriteIndex -= mBufferSize;
    }
  }

  bool IsActive() const override {
    // Could return false when !mIsConnected after all output chunks produced
    // by main thread events calling
    // SharedBuffers::FinishProducingOutputBuffer() have been processed.
    return true;
  }

  size_t SizeOfExcludingThis(MallocSizeOf aMallocSizeOf) const override {
    // Not owned:
    // - mDestination (probably)
    size_t amount = AudioNodeEngine::SizeOfExcludingThis(aMallocSizeOf);
    amount += mSharedBuffers->SizeOfIncludingThis(aMallocSizeOf);
    if (mInputBuffer) {
      amount += mInputBuffer->SizeOfIncludingThis(aMallocSizeOf);
    }

    return amount;
  }

  size_t SizeOfIncludingThis(MallocSizeOf aMallocSizeOf) const override {
    return aMallocSizeOf(this) + SizeOfExcludingThis(aMallocSizeOf);
  }

 private:
  void SendBuffersToMainThread(AudioNodeStream* aStream, GraphTime aFrom) {
    MOZ_ASSERT(!NS_IsMainThread());

    // we now have a full input buffer ready to be sent to the main thread.
    StreamTime playbackTick = mDestination->GraphTimeToStreamTime(aFrom);
    // Add the duration of the current sample
    playbackTick += WEBAUDIO_BLOCK_SIZE;
    // Add the delay caused by the main thread
    playbackTick += mSharedBuffers->DelaySoFar();
    // Compute the playback time in the coordinate system of the destination
    double playbackTime = mDestination->StreamTimeToSeconds(playbackTick);

    class Command final : public Runnable {
     public:
      Command(AudioNodeStream* aStream,
              already_AddRefed<ThreadSharedFloatArrayBufferList> aInputBuffer,
              double aPlaybackTime)
          : mozilla::Runnable("Command"),
            mStream(aStream),
            mInputBuffer(aInputBuffer),
            mPlaybackTime(aPlaybackTime) {}

      NS_IMETHOD Run() override {
        auto engine =
            static_cast<ScriptProcessorNodeEngine*>(mStream->Engine());
        AudioChunk output;
        output.SetNull(engine->mBufferSize);
        {
          auto node =
              static_cast<ScriptProcessorNode*>(engine->NodeMainThread());
          if (!node) {
            return NS_OK;
          }

          if (node->HasListenersFor(nsGkAtoms::onaudioprocess)) {
            DispatchAudioProcessEvent(node, &output);
          }
          // The node may have been destroyed during event dispatch.
        }

        // Append it to our output buffer queue
        engine->GetSharedBuffers()->FinishProducingOutputBuffer(output);

        return NS_OK;
      }

      // Sets up |output| iff buffers are set in event handlers.
      void DispatchAudioProcessEvent(ScriptProcessorNode* aNode,
                                     AudioChunk* aOutput) {
        AudioContext* context = aNode->Context();
        if (!context) {
          return;
        }

        AutoJSAPI jsapi;
        if (NS_WARN_IF(!jsapi.Init(aNode->GetOwner()))) {
          return;
        }
        JSContext* cx = jsapi.cx();
        uint32_t inputChannelCount = aNode->ChannelCount();

        // Create the input buffer
        RefPtr<AudioBuffer> inputBuffer;
        if (mInputBuffer) {
          ErrorResult rv;
          inputBuffer = AudioBuffer::Create(
              context->GetOwner(), inputChannelCount, aNode->BufferSize(),
              context->SampleRate(), mInputBuffer.forget(), rv);
          if (rv.Failed()) {
            rv.SuppressException();
            return;
          }
        }

        // Ask content to produce data in the output buffer
        // Note that we always avoid creating the output buffer here, and we try
        // to avoid creating the input buffer as well.  The AudioProcessingEvent
        // class knows how to lazily create them if needed once the script tries
        // to access them.  Otherwise, we may be able to get away without
        // creating them!
        RefPtr<AudioProcessingEvent> event =
            new AudioProcessingEvent(aNode, nullptr, nullptr);
        event->InitEvent(inputBuffer, inputChannelCount, mPlaybackTime);
        aNode->DispatchTrustedEvent(event);

        // Steal the output buffers if they have been set.
        // Don't create a buffer if it hasn't been used to return output;
        // FinishProducingOutputBuffer() will optimize output = null.
        // GetThreadSharedChannelsForRate() may also return null after OOM.
        if (event->HasOutputBuffer()) {
          ErrorResult rv;
          AudioBuffer* buffer = event->GetOutputBuffer(rv);
          // HasOutputBuffer() returning true means that GetOutputBuffer()
          // will not fail.
          MOZ_ASSERT(!rv.Failed());
          *aOutput = buffer->GetThreadSharedChannelsForRate(cx);
          MOZ_ASSERT(aOutput->IsNull() ||
                         aOutput->mBufferFormat == AUDIO_FORMAT_FLOAT32,
                     "AudioBuffers initialized from JS have float data");
        }
      }

     private:
      RefPtr<AudioNodeStream> mStream;
      RefPtr<ThreadSharedFloatArrayBufferList> mInputBuffer;
      double mPlaybackTime;
    };

    RefPtr<Command> command =
        new Command(aStream, mInputBuffer.forget(), playbackTime);
    mAbstractMainThread->Dispatch(command.forget());
  }

  friend class ScriptProcessorNode;

  RefPtr<AudioNodeStream> mDestination;
  nsAutoPtr<SharedBuffers> mSharedBuffers;
  RefPtr<ThreadSharedFloatArrayBufferList> mInputBuffer;
  const uint32_t mBufferSize;
  const uint32_t mInputChannelCount;
  // The write index into the current input buffer
  uint32_t mInputWriteIndex;
  bool mIsConnected = false;
};

ScriptProcessorNode::ScriptProcessorNode(AudioContext* aContext,
                                         uint32_t aBufferSize,
                                         uint32_t aNumberOfInputChannels,
                                         uint32_t aNumberOfOutputChannels)
    : AudioNode(aContext, aNumberOfInputChannels,
                mozilla::dom::ChannelCountMode::Explicit,
                mozilla::dom::ChannelInterpretation::Speakers),
      mBufferSize(aBufferSize ? aBufferSize
                              :  // respect what the web developer requested
                      4096)      // choose our own buffer size -- 4KB for now
      ,
      mNumberOfOutputChannels(aNumberOfOutputChannels) {
  MOZ_ASSERT(BufferSize() % WEBAUDIO_BLOCK_SIZE == 0, "Invalid buffer size");
  ScriptProcessorNodeEngine* engine = new ScriptProcessorNodeEngine(
      this, aContext->Destination(), BufferSize(), aNumberOfInputChannels);
  mStream = AudioNodeStream::Create(
      aContext, engine, AudioNodeStream::NO_STREAM_FLAGS, aContext->Graph());
}

ScriptProcessorNode::~ScriptProcessorNode() {}

size_t ScriptProcessorNode::SizeOfExcludingThis(
    MallocSizeOf aMallocSizeOf) const {
  size_t amount = AudioNode::SizeOfExcludingThis(aMallocSizeOf);
  return amount;
}

size_t ScriptProcessorNode::SizeOfIncludingThis(
    MallocSizeOf aMallocSizeOf) const {
  return aMallocSizeOf(this) + SizeOfExcludingThis(aMallocSizeOf);
}

void ScriptProcessorNode::EventListenerAdded(nsAtom* aType) {
  AudioNode::EventListenerAdded(aType);
  if (aType == nsGkAtoms::onaudioprocess) {
    UpdateConnectedStatus();
  }
}

void ScriptProcessorNode::EventListenerRemoved(nsAtom* aType) {
  AudioNode::EventListenerRemoved(aType);
  if (aType == nsGkAtoms::onaudioprocess) {
    UpdateConnectedStatus();
  }
}

JSObject* ScriptProcessorNode::WrapObject(JSContext* aCx,
                                          JS::Handle<JSObject*> aGivenProto) {
  return ScriptProcessorNode_Binding::Wrap(aCx, this, aGivenProto);
}

void ScriptProcessorNode::UpdateConnectedStatus() {
  bool isConnected =
      mHasPhantomInput || !(OutputNodes().IsEmpty() &&
                            OutputParams().IsEmpty() && InputNodes().IsEmpty());

  // Events are queued even when there is no listener because a listener
  // may be added while events are in the queue.
  SendInt32ParameterToStream(ScriptProcessorNodeEngine::IS_CONNECTED,
                             isConnected);

  if (isConnected && HasListenersFor(nsGkAtoms::onaudioprocess)) {
    MarkActive();
  } else {
    MarkInactive();
  }

  auto engine = static_cast<ScriptProcessorNodeEngine*>(mStream->Engine());
  engine->GetSharedBuffers()->NotifyNodeIsConnected(isConnected);
}

}  // namespace dom
}  // namespace mozilla