820102 - MediaPipeline threading and cleanup

Status: RESOLVED FIXED

(Core :: WebRTC: Audio/Video, defect)

Comment 1

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 2

[Randell Jesup [:jesup] (needinfo me)]

Comment on attachment 690521 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 690521 [details] [diff] [review]:
-----------------------------------------------------------------

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.h
@@ +175,5 @@
>    RefPtr<SrtpFlow> rtp_recv_srtp_;
>    RefPtr<SrtpFlow> rtcp_recv_srtp_;
> +
> +  // Written only on STS thread. May be read on other
> +  // threads but 

but... ?  I can't wait to read the rest!

Comment 4

[Henrik Skupin [:whimboo][⌚️UTC+2]]

Given the dupe bug 821439 and the fix here for leaking instances we should mark this bug as mlk.

Comment 5

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 6

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 7

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 8

[Eric Rescorla (:ekr)]

Comment on attachment 692709 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 692709 [details] [diff] [review]:
-----------------------------------------------------------------

I think this is getting pretty close to finished. I've run a bunch of tests and it seems to work at least as reliably as before and should fix a bunch of close type issues.

Comment 9

[Randell Jesup [:jesup] (needinfo me)]

Comment on attachment 692709 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 692709 [details] [diff] [review]:
-----------------------------------------------------------------

Primary issue is the use of stream_

::: media/mtransport/databuffer.h
@@ +17,5 @@
> +class DataBuffer {
> + public:
> +  DataBuffer() : data_(nullptr), len_(0) {}
> +  DataBuffer(const uint8_t *data, size_t len) {
> +    Assign(data, len);

I wonder how many threadsafe buffer classes we have...

We should consider nominating this for mfbt after it lands in our code.

@@ +21,5 @@
> +    Assign(data, len);
> +  }
> +
> +  void Assign(const uint8_t *data, size_t len) {
> +    data_ = new unsigned char[len];

what if len is 0?  
http://stackoverflow.com/questions/1087042/c-new-int0-will-it-allocate-memory
It's probably ok, but should be noted in a comment

@@ +29,5 @@
> +  }
> +
> +  const uint8_t *data() const { return data_; }
> +  size_t len() const { return len_; }
> +  const bool empty() const { return len_ != 0; }

We don't need it, but for mftb we'd probably want to add a clear.

::: media/webrtc/signaling/src/media/VcmSIPCCBinding.cpp
@@ +1382,5 @@
> +    }
> +
> +    CSFLogDebug(logTag, "Created video pipeline %p, conduit=%p, pc_stream=%d pc_track=%d",
> +                pipeline.get(), conduit.get(), pc_stream_id, pc_track_id);
> +        

whitespace

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp
@@ +89,1 @@
>    transport_->Detach();

I wonder how this will interact with NotifyQueuedTrackChanges/etc....  (Per the previous problems with Detach())

Aha.  only uses the conduit now, which doesn't go away.  Good.

@@ +122,5 @@
>    return res;
>  }
>  
>  nsresult MediaPipeline::TransportReadyInt(TransportFlow *flow) {
> +  MOZ_ASSERT(!description_.empty());  // trap an error we have been seeing

Bug #?

@@ +315,5 @@
>    ++rtp_packets_sent_;
> +
> +  if (!(rtp_packets_sent_ % 100)) {
> +    MOZ_MTLOG(MP_LOG_INFO, "RTP sent packet count for " << description_
> +              << " Pipeline " << static_cast<void *>(this) 

trailing ws here and elsewhere

@@ +541,5 @@
>    NS_ENSURE_TRUE(pipeline_->rtp_transport_, NS_ERROR_NULL_POINTER);
>  
>    // libsrtp enciphers in place, so we need a new, big enough
>    // buffer.
>    // XXX. allocates and deletes one buffer per packet sent.

File a bug for reducing memory churn in the future and put number here

@@ +817,5 @@
>                           0, samples_length, AUDIO_FORMAT_S16);
>  
>      char buf[32];
> +    PR_snprintf(buf, 32, "%p", source_);
> +    source_->AppendToTrack(1,  // TODO([e-mail]): Track ID

We have a bug (or bugs) on adding multiple streams of a type; put the bug # here.

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.h
@@ +138,3 @@
>      void Detach() { pipeline_ = NULL; }
>      MediaPipeline *pipeline() const { return pipeline_; }
> +    

trailing whitespace in a few places

@@ +169,5 @@
>    void PacketReceived(TransportLayer *layer, const unsigned char *data,
>                        size_t len);
>  
> +  Direction direction_;
> +  nsDOMMediaStream* stream_;  // A pointer to the stream we are servicing.

It's generally ok to access a MediaStream from other threads, and lifetime is not handled by addref/release.  However, nsDOMMediaStreams can't be addrefed/released from other threads, so I wonder why this needs to be a nsDOMMediaStream - and if it does, much care needs to be made about thread access and it should be noted here as this code deals with a lot of threads.  If it's only Addrefed/Released from MainThread (and deletion of this object happens there), then it should be an nsRefPtr<>.

The other fields here are well-documented as to thread access; this should be too.

@@ +243,5 @@
> +    stream_->GetStream()->RemoveListener(listener_);
> +    // Remove our reference so that when the MediaStreamGraph
> +    // releases the listener, it will be destroyed.
> +    listener_ = nullptr;
> +    stream_ = nullptr;

Assert it's on MainThread

@@ +327,5 @@
>      stream_->GetStream()->RemoveListener(listener_);
> +    // Remove our reference so that when the MediaStreamGraph
> +    // releases the listener, it will be destroyed.
> +    listener_ = nullptr;
> +    stream_ = nullptr;

Assert mainthread

@@ +381,5 @@
> +  // Called on the main thread.
> +  virtual void DetachMediaStream() {
> +    conduit_ = nullptr;  // Force synchronous destruction so we
> +                         // stop generating video.
> +    stream_ = nullptr;

Assert mainthread

Comment 10

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 11

[Randell Jesup [:jesup] (needinfo me)]

Comment on attachment 692773 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 692773 [details] [diff] [review]:
-----------------------------------------------------------------

::: media/webrtc/signaling/src/media/VcmSIPCCBinding.cpp
@@ +1382,5 @@
> +    }
> +
> +    CSFLogDebug(logTag, "Created video pipeline %p, conduit=%p, pc_stream=%d pc_track=%d",
> +                pipeline.get(), conduit.get(), pc_stream_id, pc_track_id);
> +        

trailing ws

@@ +1959,5 @@
> +            pc.impl()->GetMainThread().get(),
> +            pc.impl()->GetSTSThread(),
> +            stream->GetMediaStream()->GetStream(),
> +            conduit, rtp_flow, rtcp_flow);
> +    

ws

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp
@@ +812,5 @@
>      segment.AppendFrames(samples.forget(), samples_length,
>                           0, samples_length, AUDIO_FORMAT_S16);
>  
>      char buf[32];
> +    PR_snprintf(buf, 32, "%p", source_);

Um....
a) sizeof(buf) not 32 please
b) why is this here....?  Looks like leftover cruft.

::: media/webrtc/signaling/test/mediapipeline_unittest.cpp
@@ +137,5 @@
>          ConfigureSendMediaCodec(&audio_config_);
>      EXPECT_EQ(mozilla::kMediaConduitNoError, err);
>  
> +    std::string test_pc("PC");
> +    

ws

Comment 12

[Timothy B. Terriberry (:derf)]

Comment on attachment 692773 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 692773 [details] [diff] [review]:
-----------------------------------------------------------------

r=me with a few nits, but there's one question I couldn't answer.

::: media/mtransport/databuffer.h
@@ +35,5 @@
> +  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataBuffer)
> +
> +private:
> +  ScopedDeleteArray<uint8_t> data_;
> +  int32_t len_;

Why is len_ int32_t when Assign() takes and len() returns size_t? That seems like it could be a problem on 64-bit systems.

::: media/webrtc/signaling/src/media/VcmSIPCCBinding.cpp
@@ +1340,5 @@
> +
> +    CSFLogDebug(logTag, "Created audio pipeline %p, conduit=%p, pc_stream=%d pc_track=%d",
> +                pipeline.get(), conduit.get(), pc_stream_id, pc_track_id);
> +
> +    // Now we have all the pieces, create the pipeline

We just created the pipeline up above, didn't we?

@@ +1387,1 @@
>      // Now we have all the pieces, create the pipeline

I don't think this comment is accurate any longer.

@@ +1973,1 @@
>      // Now we have all the pieces, create the pipeline

And I assume this is here just because of C&P.

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp
@@ +84,5 @@
>  
>  
>  // Disconnect us from the transport so that we can cleanly destruct
>  // the pipeline on the main thread.
> +void MediaPipeline::DetachTransport_s() {

Can we assert we're on the STS thread here?

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.h
@@ +42,5 @@
> +//
> +// GSM:
> +//   * Assembles the pipeline
> +// SocketTransportService
> +//   * Receives notification that the ICE and DTLS have completed

"the" ICE?

@@ +93,4 @@
>    }
>  
>    virtual ~MediaPipeline() {
> +    Shutdown();

This seems pretty fragile for a refcounted class... I see on IRC we already screwed up once and were releasing the last reference from the wrong thread.

@@ +97,5 @@
> +  }
> +
> +  void Shutdown() {
> +    // Verify we are on the main thread.
> +    if (main_thread_) {

Didn't you just add an ASSERT_ON_THREAD macro to do all this?

@@ +171,5 @@
>                        size_t len);
>  
> +  Direction direction_;
> +  RefPtr<MediaStream> stream_;  // A pointer to the stream we are servicing.
> +  		      		// Used on STS and MediaStreamGraph threads.

Also read/written on the main thread, which is the part we have to be very careful about. It would be nice to document why this is safe here.

@@ +172,5 @@
>  
> +  Direction direction_;
> +  RefPtr<MediaStream> stream_;  // A pointer to the stream we are servicing.
> +  		      		// Used on STS and MediaStreamGraph threads.
> +  RefPtr<MediaSessionConduit> conduit_;  // Our conduit. Used on the STS.

Also read/written on the main thread.

@@ +242,5 @@
>  
> +  // Called on the main thread.
> +  virtual void DetachMediaStream() {
> +    ASSERT_ON_THREAD(main_thread_);
> +    stream_->RemoveListener(listener_);

So, I spent quite a while tracing through things, and I can't prove to myself that this RemoveListener() call can't get made before the AddListener() call dispatched in Init() actually gets run on the main thread (this applies for the other listeners, too).

Comment 13

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 14

[Eric Rescorla (:ekr)]

Comment on attachment 692773 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 692773 [details] [diff] [review]:
-----------------------------------------------------------------

::: media/mtransport/databuffer.h
@@ +35,5 @@
> +  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(DataBuffer)
> +
> +private:
> +  ScopedDeleteArray<uint8_t> data_;
> +  int32_t len_;

Copy-and-paste bug. I will fix.

::: media/webrtc/signaling/src/media/VcmSIPCCBinding.cpp
@@ +1340,5 @@
> +
> +    CSFLogDebug(logTag, "Created audio pipeline %p, conduit=%p, pc_stream=%d pc_track=%d",
> +                pipeline.get(), conduit.get(), pc_stream_id, pc_track_id);
> +
> +    // Now we have all the pieces, create the pipeline

Yeah, I'll move this comment and the ones below.

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.cpp
@@ +84,5 @@
>  
>  
>  // Disconnect us from the transport so that we can cleanly destruct
>  // the pipeline on the main thread.
> +void MediaPipeline::DetachTransport_s() {

My convention is not to assert on private members with thread suffixes. I can if you think it's important.

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.h
@@ +42,5 @@
> +//
> +// GSM:
> +//   * Assembles the pipeline
> +// SocketTransportService
> +//   * Receives notification that the ICE and DTLS have completed

Fair enough.

@@ +93,4 @@
>    }
>  
>    virtual ~MediaPipeline() {
> +    Shutdown();

Need a test for whether Shutdown has already been called. Sometimes we destroy the pipeline in the MSG thread after it has already been called in the main thread.

Proposed resolution: have Shutdown not explicitly called but have the dtor ASSERT that Shutdown was called
(though probably in a non-thread-safe fashion, since it's just a double check)

@@ +97,5 @@
> +  }
> +
> +  void Shutdown() {
> +    // Verify we are on the main thread.
> +    if (main_thread_) {

Yeah, the usual thing where I got tired of typing it but then didn't go back and refactor.

@@ +242,5 @@
>  
> +  // Called on the main thread.
> +  virtual void DetachMediaStream() {
> +    ASSERT_ON_THREAD(main_thread_);
> +    stream_->RemoveListener(listener_);

I see your point. Is there a callback for when AddListener() has succeeded?

Comment 15

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 16

[Eric Rescorla (:ekr)]

derf:

Rethinking the DetachMediaStream issue.

I'm not sure it matters. MSG allegedly guarantees order of operations.

The "normal path" is.

1. Call AddListener()
2. Listener is added and media starts to flow.
3. RemoveListener()
4. Media continues to flow.
5. Listener is removed and object is deleted.

In the case you suggest, we get the order 1, 3, 2, 4, 5

This seems safe.

Note that this does not apply to the sequence of AddListener() in MediaPipelineTransmit::TransportReady() because that happens on the STS thread, so in principle those might be able to happen in the opposite order, though it's complicated b/c DetachTransport() also happens on the STS thread.

Comment 17

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 18

[Timothy B. Terriberry (:derf)]

(In reply to Eric Rescorla (:ekr) from comment #16)
> In the case you suggest, we get the order 1, 3, 2, 4, 5

I think you misunderstood... Init() gets run from one of the SIPCC threads and queues a task to call AddListener() on the main thread. However, there may already be a task queued on the main thread that will, e.g., shut down all PCs, which then releases the MediaPipeline, whose destructor calls Shutdown(), which calls RemoveListener() on the main thread _without_ queuing a task, i.e., before the deferred AddListener(). So the actual order is 3, 1, ?, ?, ?. The order gets screwed up before things ever get to the MSG.

Comment 19

[Timothy B. Terriberry (:derf)]

(In reply to Eric Rescorla (:ekr) from comment #14)
> My convention is not to assert on private members with thread suffixes. I
> can if you think it's important.

Yeah, our convention is generally the opposite.

Comment 20

[Eric Rescorla (:ekr)]

Attachment: MediaPipeline threading and cleanup

Comment 21

[Timothy B. Terriberry (:derf)]

Comment on attachment 694498 [details] [diff] [review]
MediaPipeline threading and cleanup

Review of attachment 694498 [details] [diff] [review]:
-----------------------------------------------------------------

I only reviewed the interdiffs from the last patch, but LGTM.

::: media/webrtc/signaling/src/mediapipeline/MediaPipeline.h
@@ +166,5 @@
>                        size_t len);
>  
> +  Direction direction_;
> +  RefPtr<MediaStream> stream_;  // A pointer to the stream we are servicing.
> +                // Written on the main thread.

The indentation here is messed up.

@@ +169,5 @@
> +  RefPtr<MediaStream> stream_;  // A pointer to the stream we are servicing.
> +                // Written on the main thread.
> +  		      		// Used on STS and MediaStreamGraph threads.
> +  RefPtr<MediaSessionConduit> conduit_;  // Our conduit. Used on the STS
> +                                         // and main threads.

conduit_ is also written to from the main thread (since "used" seems to be "read" here).

@@ +271,5 @@
> +    virtual void ProcessVideoChunk(VideoSessionConduit *conduit,
> +				   TrackRate rate, VideoChunk& chunk);
> +#endif
> +    RefPtr<MediaSessionConduit> conduit_;
> +    bool active_;

Make this volatile. It doesn't really solve any of the problems, but it at least serves as a warning to others.

Comment 22

[Ed Morley [:emorley]]

Backed out for assertions:
https://tbpl.mozilla.org/?tree=Mozilla-Inbound&rev=9587e39f9a50

https://hg.mozilla.org/integration/mozilla-inbound/rev/0134b2193fe5

Comment 23

[:Ms2ger (he/him; ⌚ UTC+1/+2)]

https://hg.mozilla.org/mozilla-central/rev/97c8a4726fcd