Bug 884365: Deliver gUM data directly to PeerConnection to avoid delay buildup and resampling r=roc

content/media/DOMMediaStream.h

@@ -39,6 +39,8 @@ class AudioStreamTrack;
 class VideoStreamTrack;
 }
 
+class MediaStreamDirectListener;
+
 /**
  * DOM wrapper for MediaStreams.
  */
@@ -74,6 +76,14 @@ public:
 
   MediaStream* GetStream() const { return mStream; }
 
+  /**
+   * Overridden in DOMLocalMediaStreams to allow getUserMedia to pass
+   * data directly to RTCPeerConnection without going through graph queuing.
+   * Returns a bool to let us know if direct data will be delivered.
+   */
+  virtual bool AddDirectListener(MediaStreamDirectListener *aListener) { return false; }
+  virtual void RemoveDirectListener(MediaStreamDirectListener *aListener) {}
+
   bool IsFinished();
   /**
    * Returns a principal indicating who may access this stream. The stream contents

content/media/MediaStreamGraph.cpp

@@ -1976,7 +1976,7 @@ SourceMediaStream::AddTrack(TrackID aID, TrackRate aRate, TrackTicks aStart,
 }
 
 bool
-SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment)
+SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment, MediaSegment *aRawSegment)
 {
   MutexAutoLock lock(mMutex);
   // ::EndAllTrackAndFinished() can end these before the sources notice
@@ -1990,7 +1990,9 @@ SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment)
       // Indirect listeners (via subsequent TrackUnion nodes) are synced to
       // playout time, and so can be delayed by buffering.
 
-      track->mData->AppendFrom(aSegment);
+      // Must notify first, since AppendFrom() will empty out aSegment
+      NotifyDirectConsumers(track, aRawSegment ? aRawSegment : aSegment);
+      track->mData->AppendFrom(aSegment); // note: aSegment is now dead
       appended = true;
     } else {
       aSegment->Clear();
@@ -2002,6 +2004,35 @@ SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment)
   return appended;
 }
 
+void
+SourceMediaStream::NotifyDirectConsumers(TrackData *aTrack,
+                                         MediaSegment *aSegment)
+{
+  // Call with mMutex locked
+  MOZ_ASSERT(aTrack);
+
+  for (uint32_t j = 0; j < mDirectListeners.Length(); ++j) {
+    MediaStreamDirectListener* l = mDirectListeners[j];
+    TrackTicks offset = 0; // FIX! need a separate TrackTicks.... or the end of the internal buffer
+    l->NotifyRealtimeData(static_cast<MediaStreamGraph*>(GraphImpl()), aTrack->mID, aTrack->mRate,
+                          offset, aTrack->mCommands, *aSegment);
+  }
+}
+
+void
+SourceMediaStream::AddDirectListener(MediaStreamDirectListener* aListener)
+{
+  MutexAutoLock lock(mMutex);
+  mDirectListeners.AppendElement(aListener);
+}
+
+void
+SourceMediaStream::RemoveDirectListener(MediaStreamDirectListener* aListener)
+{
+  MutexAutoLock lock(mMutex);
+  mDirectListeners.RemoveElement(aListener);
+}
+
 bool
 SourceMediaStream::HaveEnoughBuffered(TrackID aID)
 {

content/media/MediaStreamGraph.h

@@ -169,6 +169,30 @@ public:
                                         const MediaSegment& aQueuedMedia) {}
 };
 
+/**
+ * This is a base class for media graph thread listener direct callbacks
+ * from within AppendToTrack().  Note that your regular listener will
+ * still get NotifyQueuedTrackChanges() callbacks from the MSG thread, so
+ * you must be careful to ignore them if AddDirectListener was successful.
+ */
+class MediaStreamDirectListener : public MediaStreamListener
+{
+public:
+  virtual ~MediaStreamDirectListener() {}
+
+  /*
+   * This will be called on any MediaStreamDirectListener added to a
+   * a SourceMediaStream when AppendToTrack() is called.  The MediaSegment
+   * will be the RawSegment (unresampled) if available in AppendToTrack().
+   * Note that NotifyQueuedTrackChanges() calls will also still occur.
+   */
+  virtual void NotifyRealtimeData(MediaStreamGraph* aGraph, TrackID aID,
+                                  TrackRate aTrackRate,
+                                  TrackTicks aTrackOffset,
+                                  uint32_t aTrackEvents,
+                                  const MediaSegment& aMedia) {}
+};
+
 /**
  * This is a base class for main-thread listener callbacks.
  * This callback is invoked on the main thread when the main-thread-visible
@@ -599,6 +623,10 @@ public:
    * it is still possible for a NotifyPull to occur.
    */
   void SetPullEnabled(bool aEnabled);
+
+  void AddDirectListener(MediaStreamDirectListener* aListener);
+  void RemoveDirectListener(MediaStreamDirectListener* aListener);
+
   /**
    * Add a new track to the stream starting at the given base time (which
    * must be greater than or equal to the last time passed to
@@ -613,7 +641,7 @@ public:
    * Returns false if the data was not appended because no such track exists
    * or the stream was already finished.
    */
-  bool AppendToTrack(TrackID aID, MediaSegment* aSegment);
+  bool AppendToTrack(TrackID aID, MediaSegment* aSegment, MediaSegment *aRawSegment = nullptr);
   /**
    * Returns true if the buffer currently has enough data.
    * Returns false if there isn't enough data or if no such track exists.
@@ -715,6 +743,15 @@ protected:
     return nullptr;
   }
 
+  /**
+   * Notify direct consumers of new data to one of the stream tracks.
+   * The data doesn't have to be resampled (though it may be).  This is called
+   * from AppendToTrack on the thread providing the data, and will call
+   * the Listeners on this thread.
+   */
+  void NotifyDirectConsumers(TrackData *aTrack,
+                             MediaSegment *aSegment);
+
   // Media stream graph thread only
   MediaStreamListener::Consumption mLastConsumptionState;
 
@@ -724,6 +761,7 @@ protected:
   // protected by mMutex
   StreamTime mUpdateKnownTracksTime;
   nsTArray<TrackData> mUpdateTracks;
+  nsTArray<nsRefPtr<MediaStreamDirectListener> > mDirectListeners;
   bool mPullEnabled;
   bool mUpdateFinished;
   bool mDestroyed;

dom/media/MediaManager.cpp

@@ -295,6 +295,23 @@ public:
     }
   }
 
+  // Allow getUserMedia to pass input data directly to PeerConnection/MediaPipeline
+  virtual bool AddDirectListener(MediaStreamDirectListener *aListener) MOZ_OVERRIDE
+  {
+    if (mSourceStream) {
+      mSourceStream->AddDirectListener(aListener);
+      return true; // application should ignore NotifyQueuedTrackData
+    }
+    return false;
+  }
+
+  virtual void RemoveDirectListener(MediaStreamDirectListener *aListener) MOZ_OVERRIDE
+  {
+    if (mSourceStream) {
+      mSourceStream->RemoveDirectListener(aListener);
+    }
+  }
+
   // The actual MediaStream is a TrackUnionStream. But these resources need to be
   // explicitly destroyed too.
   nsRefPtr<SourceMediaStream> mSourceStream;

media/webrtc/signaling/src/media/VcmSIPCCBinding.cpp

@@ -2088,7 +2088,7 @@ static int vcmTxStartICE_m(cc_mcapid_t mcap_id,
             pc.impl()->GetHandle(),
             pc.impl()->GetMainThread().get(),
             pc.impl()->GetSTSThread(),
-            stream->GetMediaStream()->GetStream(),
+            stream->GetMediaStream(),
             pc_track_id,
             conduit, rtp_flow, rtcp_flow);
 
@@ -2127,7 +2127,7 @@ static int vcmTxStartICE_m(cc_mcapid_t mcap_id,
             pc.impl()->GetHandle(),
             pc.impl()->GetMainThread().get(),
             pc.impl()->GetSTSThread(),
-            stream->GetMediaStream()->GetStream(),
+            stream->GetMediaStream(),
             pc_track_id,
             conduit, rtp_flow, rtcp_flow);
 

media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp

@@ -530,6 +530,13 @@ nsresult MediaPipelineTransmit::Init() {
 
   stream_->AddListener(listener_);
 
+  // Is this a gUM mediastream?  If so, also register the Listener directly with
+  // the SourceMediaStream that's attached to the TrackUnion so we can get direct
+  // unqueued (and not resampled) data
+  if (domstream_->AddDirectListener(listener_)) {
+    listener_->direct_connect_ = true;
+  }
+
   return MediaPipeline::Init();
 }
 
@@ -647,6 +654,18 @@ nsresult MediaPipeline::PipelineTransport::SendRtcpPacket_s(
                                out_len);
 }
 
+// Called if we're attached with AddDirectListener()
+void MediaPipelineTransmit::PipelineListener::
+NotifyRealtimeData(MediaStreamGraph* graph, TrackID tid,
+                   TrackRate rate,
+                   TrackTicks offset,
+                   uint32_t events,
+                   const MediaSegment& media) {
+  MOZ_MTLOG(ML_DEBUG, "MediaPipeline::NotifyRealtimeData()");
+
+  NewData(graph, tid, rate, offset, events, media);
+}
+
 void MediaPipelineTransmit::PipelineListener::
 NotifyQueuedTrackChanges(MediaStreamGraph* graph, TrackID tid,
                          TrackRate rate,
@@ -655,6 +674,18 @@ NotifyQueuedTrackChanges(MediaStreamGraph* graph, TrackID tid,
                          const MediaSegment& queued_media) {
   MOZ_MTLOG(ML_DEBUG, "MediaPipeline::NotifyQueuedTrackChanges()");
 
+  // ignore non-direct data if we're also getting direct data
+  if (!direct_connect_) {
+    NewData(graph, tid, rate, offset, events, queued_media);
+  }
+}
+
+void MediaPipelineTransmit::PipelineListener::
+NewData(MediaStreamGraph* graph, TrackID tid,
+        TrackRate rate,
+        TrackTicks offset,
+        uint32_t events,
+        const MediaSegment& media) {
   if (!active_) {
     MOZ_MTLOG(ML_DEBUG, "Discarding packets because transport not ready");
     return;
@@ -663,13 +694,13 @@ NotifyQueuedTrackChanges(MediaStreamGraph* graph, TrackID tid,
   // TODO(ekr@rtfm.com): For now assume that we have only one
   // track type and it's destined for us
   // See bug 784517
-  if (queued_media.GetType() == MediaSegment::AUDIO) {
+  if (media.GetType() == MediaSegment::AUDIO) {
     if (conduit_->type() != MediaSessionConduit::AUDIO) {
       // Ignore data in case we have a muxed stream
       return;
     }
     AudioSegment* audio = const_cast<AudioSegment *>(
-        static_cast<const AudioSegment *>(&queued_media));
+        static_cast<const AudioSegment *>(&media));
 
     AudioSegment::ChunkIterator iter(*audio);
     while(!iter.IsEnded()) {
@@ -677,14 +708,14 @@ NotifyQueuedTrackChanges(MediaStreamGraph* graph, TrackID tid,
                         rate, *iter);
       iter.Next();
     }
-  } else if (queued_media.GetType() == MediaSegment::VIDEO) {
+  } else if (media.GetType() == MediaSegment::VIDEO) {
 #ifdef MOZILLA_INTERNAL_API
     if (conduit_->type() != MediaSessionConduit::VIDEO) {
       // Ignore data in case we have a muxed stream
       return;
     }
     VideoSegment* video = const_cast<VideoSegment *>(
-        static_cast<const VideoSegment *>(&queued_media));
+        static_cast<const VideoSegment *>(&media));
 
     VideoSegment::ChunkIterator iter(*video);
     while(!iter.IsEnded()) {

media/webrtc/signaling/src/mediapipeline/MediaPipeline.h

@@ -304,15 +304,17 @@ class MediaPipelineTransmit : public MediaPipeline {
   MediaPipelineTransmit(const std::string& pc,
                         nsCOMPtr<nsIEventTarget> main_thread,
                         nsCOMPtr<nsIEventTarget> sts_thread,
-                        MediaStream *stream,
+                        DOMMediaStream *domstream,
                         TrackID track_id,
                         RefPtr<MediaSessionConduit> conduit,
                         RefPtr<TransportFlow> rtp_transport,
                         RefPtr<TransportFlow> rtcp_transport) :
       MediaPipeline(pc, TRANSMIT, main_thread, sts_thread,
-                    stream, track_id, conduit, rtp_transport,
+                    domstream->GetStream(), track_id, conduit, rtp_transport,
                     rtcp_transport),
-      listener_(new PipelineListener(conduit)) {}
+      listener_(new PipelineListener(conduit)),
+      domstream_(domstream)
+  {}
 
   // Initialize (stuff here may fail)
   virtual nsresult Init();
@@ -320,6 +322,8 @@ class MediaPipelineTransmit : public MediaPipeline {
   // Called on the main thread.
   virtual void DetachMediaStream() {
     ASSERT_ON_THREAD(main_thread_);
+    domstream_->RemoveDirectListener(listener_);
+    domstream_ = nullptr;
     stream_->RemoveListener(listener_);
     // Let the listener be destroyed with the pipeline (or later).
     stream_ = nullptr;
@@ -329,11 +333,13 @@ class MediaPipelineTransmit : public MediaPipeline {
   virtual nsresult TransportReady_s(TransportFlow *flow);
 
   // Separate class to allow ref counting
-  class PipelineListener : public MediaStreamListener {
+  class PipelineListener : public MediaStreamDirectListener {
+   friend class MediaPipelineTransmit;
    public:
     PipelineListener(const RefPtr<MediaSessionConduit>& conduit)
       : conduit_(conduit),
         active_(false),
+        direct_connect_(false),
         last_img_(-1),
         samples_10ms_buffer_(nullptr),
         buffer_current_(0),
@@ -364,7 +370,20 @@ class MediaPipelineTransmit : public MediaPipeline {
                                           const MediaSegment& queued_media) MOZ_OVERRIDE;
     virtual void NotifyPull(MediaStreamGraph* aGraph, StreamTime aDesiredTime) MOZ_OVERRIDE {}
 
+    // Implement MediaStreamDirectListener
+    virtual void NotifyRealtimeData(MediaStreamGraph* graph, TrackID tid,
+                                    TrackRate rate,
+                                    TrackTicks offset,
+                                    uint32_t events,
+                                    const MediaSegment& media) MOZ_OVERRIDE;
+
    private:
+    void NewData(MediaStreamGraph* graph, TrackID tid,
+                 TrackRate rate,
+                 TrackTicks offset,
+                 uint32_t events,
+                 const MediaSegment& media);
+
     virtual void ProcessAudioChunk(AudioSessionConduit *conduit,
 				   TrackRate rate, AudioChunk& chunk);
 #ifdef MOZILLA_INTERNAL_API
@@ -373,6 +392,7 @@ class MediaPipelineTransmit : public MediaPipeline {
 #endif
     RefPtr<MediaSessionConduit> conduit_;
     volatile bool active_;
+    bool direct_connect_;
 
     int32_t last_img_; // serial number of last Image
 
@@ -388,6 +408,7 @@ class MediaPipelineTransmit : public MediaPipeline {
 
  private:
   RefPtr<PipelineListener> listener_;
+  DOMMediaStream *domstream_;
 };
 
 

media/webrtc/signaling/test/FakeMediaStreams.h

@@ -49,6 +49,18 @@ class Fake_MediaStreamListener
   NS_INLINE_DECL_THREADSAFE_REFCOUNTING(Fake_MediaStreamListener)
 };
 
+class Fake_MediaStreamDirectListener : public Fake_MediaStreamListener
+{
+ public:
+  virtual ~Fake_MediaStreamDirectListener() {}
+
+  virtual void NotifyRealtimeData(mozilla::MediaStreamGraph* graph, mozilla::TrackID tid,
+                                  mozilla::TrackRate rate,
+                                  mozilla::TrackTicks offset,
+                                  uint32_t events,
+                                  const mozilla::MediaSegment& media) = 0;
+};
+
 // Note: only one listener supported
 class Fake_MediaStream {
  public:
@@ -112,6 +124,11 @@ class Fake_SourceMediaStream : public Fake_MediaStream {
                 mozilla::MediaSegment* aSegment) {}
   void EndTrack(mozilla::TrackID aID) {}
 
+  bool AppendToTrack(mozilla::TrackID aID, mozilla::MediaSegment* aSegment,
+                     mozilla::MediaSegment *aRawSegment) {
+    return AppendToTrack(aID, aSegment);
+  }
+
   bool AppendToTrack(mozilla::TrackID aID, mozilla::MediaSegment* aSegment) {
     bool nonZeroSample = false;
     MOZ_ASSERT(aSegment);
@@ -154,6 +171,9 @@ class Fake_SourceMediaStream : public Fake_MediaStream {
   void SetPullEnabled(bool aEnabled) {
     mPullEnabled = aEnabled;
   }
+  void AddDirectListener(Fake_MediaStreamListener* aListener) {}
+  void RemoveDirectListener(Fake_MediaStreamListener* aListener) {}
+
   //Don't pull anymore data,if mStop is true.
   void StopStream() {
    mStop = true;
@@ -204,6 +224,11 @@ public:
     return ds.forget();
   }
 
+  virtual void Stop() {} // Really DOMLocalMediaStream
+
+  virtual bool AddDirectListener(Fake_MediaStreamListener *aListener) { return false; }
+  virtual void RemoveDirectListener(Fake_MediaStreamListener *aListener) {}
+
   Fake_MediaStream *GetStream() { return mMediaStream; }
 
   // Hints to tell the SDP generator about whether this
@@ -275,7 +300,9 @@ namespace mozilla {
 typedef Fake_MediaStream MediaStream;
 typedef Fake_SourceMediaStream SourceMediaStream;
 typedef Fake_MediaStreamListener MediaStreamListener;
+typedef Fake_MediaStreamDirectListener MediaStreamDirectListener;
 typedef Fake_DOMMediaStream DOMMediaStream;
+typedef Fake_DOMMediaStream DOMLocalMediaStream;
 }
 
 #endif

media/webrtc/signaling/test/mediapipeline_unittest.cpp

@@ -195,7 +195,7 @@ class TestAgentSend : public TestAgent {
         test_pc,
         NULL,
         test_utils->sts_target(),
-        audio_->GetStream(),
+        audio_,
         1,
         audio_conduit_,
         audio_rtp_transport_.flow_,