Implemented the GetSources() in native code.

webrtc/modules/rtp_rtcp/source/rtp_receiver_impl.cc

 /*
  *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #include "webrtc/modules/rtp_rtcp/source/rtp_receiver_impl.h"
 
 #include <assert.h>
 #include <math.h>
 #include <stdlib.h>
 #include <string.h>
 
+#include <set>
+#include <vector>
+
 #include "webrtc/base/logging.h"
 #include "webrtc/common_types.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_payload_registry.h"
 #include "webrtc/modules/rtp_rtcp/include/rtp_rtcp_defines.h"
 #include "webrtc/modules/rtp_rtcp/source/rtp_receiver_strategy.h"
 
 namespace webrtc {
 
+// Only return the contribuing sources in the last 10 seconds.
+static const int64_t kContributingSourcesTimeout = 10000;  // ms

[hbos, 2017/03/30 09:51:54]

Prefer ms in the name, kContributingSourcesTimeoutMs

[Zhi Huang, 2017/03/31 06:44:04]

Done.

+
 using RtpUtility::Payload;
 
 RtpReceiver* RtpReceiver::CreateVideoReceiver(
     Clock* clock,
     RtpData* incoming_payload_callback,
     RtpFeedback* incoming_messages_callback,
     RTPPayloadRegistry* rtp_payload_registry) {
   if (!incoming_payload_callback)
     incoming_payload_callback = NullObjectRtpData();
   if (!incoming_messages_callback)
@@ skipping 10 matching lines @@
     RTPPayloadRegistry* rtp_payload_registry) {
   if (!incoming_payload_callback)
     incoming_payload_callback = NullObjectRtpData();
   if (!incoming_messages_callback)
     incoming_messages_callback = NullObjectRtpFeedback();
   return new RtpReceiverImpl(
       clock, incoming_messages_callback, rtp_payload_registry,
       RTPReceiverStrategy::CreateAudioStrategy(incoming_payload_callback));
 }
 
-RtpReceiverImpl::RtpReceiverImpl(
-    Clock* clock,
-    RtpFeedback* incoming_messages_callback,
-    RTPPayloadRegistry* rtp_payload_registry,
-    RTPReceiverStrategy* rtp_media_receiver)
+RtpReceiverImpl::RtpReceiverImpl(Clock* clock,
+                                 RtpFeedback* incoming_messages_callback,
+                                 RTPPayloadRegistry* rtp_payload_registry,
+                                 RTPReceiverStrategy* rtp_media_receiver)
     : clock_(clock),
       rtp_payload_registry_(rtp_payload_registry),
       rtp_media_receiver_(rtp_media_receiver),
       cb_rtp_feedback_(incoming_messages_callback),
       last_receive_time_(0),
       last_received_payload_length_(0),
       ssrc_(0),
       num_csrcs_(0),
       current_remote_csrc_(),
       last_received_timestamp_(0),
       last_received_frame_time_ms_(-1),
-      last_received_sequence_number_(0) {
+      last_received_sequence_number_(0),
+      current_buffer_index_(0),
+      current_buffer_size_(0) {
   assert(incoming_messages_callback);
 
   memset(current_remote_csrc_, 0, sizeof(current_remote_csrc_));
 }
 
 RtpReceiverImpl::~RtpReceiverImpl() {
   for (int i = 0; i < num_csrcs_; ++i) {
     cb_rtp_feedback_->OnIncomingCSRCChanged(current_remote_csrc_[i], false);
   }
 }
@@ skipping 70 matching lines @@
     }
     LOG(LS_WARNING) << "Receiving invalid payload type.";
     return false;
   }
 
   WebRtcRTPHeader webrtc_rtp_header;
   memset(&webrtc_rtp_header, 0, sizeof(webrtc_rtp_header));
   webrtc_rtp_header.header = rtp_header;
   CheckCSRC(webrtc_rtp_header);
 
+  UpdateContributingSource();
+
   size_t payload_data_length = payload_length - rtp_header.paddingLength;
 
   bool is_first_packet_in_frame = false;
   {
     rtc::CritScope lock(&critical_section_rtp_receiver_);
     if (HaveReceivedFrame()) {
       is_first_packet_in_frame =
           last_received_sequence_number_ + 1 == rtp_header.sequenceNumber &&
           last_received_timestamp_ != rtp_header.timestamp;
     } else {
@@ skipping 23 matching lines @@
       last_received_sequence_number_ = rtp_header.sequenceNumber;
     }
   }
   return true;
 }
 
 TelephoneEventHandler* RtpReceiverImpl::GetTelephoneEventHandler() {
   return rtp_media_receiver_->GetTelephoneEventHandler();
 }
 
+const std::vector<RtpContributingSource*>&
+RtpReceiverImpl::GetContributingSources() {

[hbos, 2017/03/30 09:51:54]

What's the threading model. Should we either lock or DCHECK that we are on the
correct thread?

[Zhi Huang, 2017/03/31 06:44:04]

This method should only be called on the worker_thread_. There is a check when
the channel_proxy calls the GetContributingSources().

+  contributing_sources_.clear();

[Taylor Brandstetter, 2017/03/30 22:55:37]

If contributing_sources_ is only used in this method, it doesn't need to be a
member variable. It can just be returned by value.

[Zhi Huang, 2017/03/31 06:44:04]

If it returns the vector by value then I have to change the return type on the
whole objects chain.

Wouldn't the vector be copied everywhere (depending on the compiler's
optimization)?

[Taylor Brandstetter, 2017/03/31 22:10:50]

With C++11, the compiler should optimize away the copy operation, so it's a
pretty cheap move operation at each layer in the stack.

+  std::set<uint32_t> selected_sources_set;
+  int64_t now = clock_->TimeInMilliseconds();
+
+  for (size_t i = 1; i <= current_buffer_size_; ++i) {
+    // Iterate the buffer in reverse order.
+    size_t index =
+        (current_buffer_index_ + kContributingSourcesBufferSize - i) %
+        kContributingSourcesBufferSize;
+    RtpContributingSource& contributing_source =
+        contributing_sources_buffer_[index];
+    // Stop iterating when the contributing source object is out of date since
+    // the buffer is ordered by the timestamp.
+    if (now - contributing_source.timestamp() > kContributingSourcesTimeout) {
+      break;
+    }
+    // Return the latest timestamp for a given SSRC and skip the duplicated
+    // ones.
+    if (selected_sources_set.find(contributing_source.source()) ==
+        selected_sources_set.end()) {
+      selected_sources_set.insert(contributing_source.source());
+      contributing_sources_.push_back(&contributing_source);

[the sun, 2017/03/30 11:31:42]

IIUC, UpdateContributingSource() will be called on the network thread (which may
coincide with the libjingle worker thread right now, but that's not how it
should be), and GetContributingSources() is called on the worker thread.

This means that unless the caller guards access to contributing_sources_ with a
lock (that's unfeasible), the pointers returned here may point at completely
different data when accessed.

I think the solution would be simpler if you just returned a vector of by-value
structs, and also if you made RtpContributingSource a plain struct at this
level. If we want to present it as an interface in the API, we can wrap it
there.

[Zhi Huang, 2017/03/31 06:44:04]

I think there are two options to solve this:
1) When GetContributingSource() is called, block the network_thread from
updating the buffer and do the scanning and deduplicating stuff.
2) Block the network thread, create another copy of the buffer, unblock the
thread and do scanning and other stuff on the copied buffer without blocking.

Do you think it is worth doing 2) assuming creating another copy would be
faster?

+    }
+  }
+  // Add the contributing source using the SSRC.

[Taylor Brandstetter, 2017/03/30 22:55:37]

This doesn't properly handle the corner case of the SSRC changing. I think the
correct behavior, as the spec currently reads, would be to return an object for
the old SSRC for 10 seconds after it switches.

[Zhi Huang, 2017/03/31 06:44:04]

According to the spec:
"If the RTP packet contains CSRCs, then the RTCRtpContributingSource objects
corresponding to those CSRCs are updated. If the RTP packet contains no CSRCs,
then the RTCRtpContributingSource object corresponding to the SSRC is updated."
This is a little bit confusing.  Does it mean that the timestamp of the SSRC
source is always the last time when a packet without any CSRCs is received? What
if the packets always contain the CSRCs? 

What about updating the SSRC source no matter if the packet contains the CSRCs
or not? 
Then when the SSRC changes, the SSRC source will not be updated for 10 seconds
and when GetContributingSources() is called, just return the old SSRC and old
timestamp.  Does it make sense to you?

By saying "return an object for the old SSRC for 10 seconds after it switches",
do you mean an "additional" object or return old ssrc instead of new one?

[Taylor Brandstetter, 2017/03/31 22:10:50]

That was the topic of this issue on github:
https://github.com/w3c/webrtc-pc/issues/1091

The direction we're heading in is that the SSRC source is *always* updated. So I
think it would be safe to write this CL assuming that's what happens.
I mean an additional object. Here's an example:

Time 1.0s: Packet received with SSRC X
Time 1.5s: GetContributingSources returns [{source: X, timestamp: 1}]
Time 2.0s: Packet received with SSRC Y
Time 2.5s: GetContributingSources returns
           [{source: X, timestamp: 1},
            {source: Y, timestamp: 2}]

... Some more packets with SSRC Y are received ...

Time 11.5s: GetContributingSources returns [{source: Y, timestamp: 11}]


This is how I would interpret the standard, at least.

+  ssrc_source_.reset(new RtpContributingSource(now, ssrc_));

[hbos, 2017/03/30 09:51:54]

I thought the SSRC was a special case of CSRC, but that it still corresponded to
an RTP packet? If that isn't the case, and it corresponds to "any packet",
shouldn't it at least be set to the latest RTP packet? This sets it to "now",
which might not correspond to any packet whatsoever. Is that how it is supposed
to work?

This returns a pointer that is not valid if you call GetContributingSources
again, even if you are on the network thread and no packets are arriving. The
lifetime worries me.

[Zhi Huang, 2017/03/31 06:44:04]

I'll just use the plain old structs and return a copy of this.

+  contributing_sources_.push_back(ssrc_source_.get());
+
+  return contributing_sources_;

[hbos, 2017/03/30 09:51:54]

Since this is returning raw pointers can you document the lifetime of these
objects in the interface? E.g. only valid until the next RTP packet
GetContributingSources call, not enough that the receiver is still alive.
Callers must be on a thread that ensures that RTP packets are not arriving while
reading the result of this function.

+}
+
 bool RtpReceiverImpl::Timestamp(uint32_t* timestamp) const {
   rtc::CritScope lock(&critical_section_rtp_receiver_);
   if (!HaveReceivedFrame())
     return false;
   *timestamp = last_received_timestamp_;
   return true;
 }
 
 bool RtpReceiverImpl::LastReceivedTimeMs(int64_t* receive_time_ms) const {
   rtc::CritScope lock(&critical_section_rtp_receiver_);
@@ skipping 238 matching lines @@
     // Using CSRC 0 to signal this event, not interop safe, other
     // implementations might have CSRC 0 as a valid value.
     if (num_csrcs_diff > 0) {
       cb_rtp_feedback_->OnIncomingCSRCChanged(0, true);
     } else if (num_csrcs_diff < 0) {
       cb_rtp_feedback_->OnIncomingCSRCChanged(0, false);
     }
   }
 }
 
+void RtpReceiverImpl::UpdateContributingSource() {

[hbos, 2017/03/30 09:51:54]

Just checking: This is thread safe without holding the lock, yes?

I see critical_section_rtp_receiver_ is used elsewhere but this is read-only and
called from IncomingRtpPacket, the only place that does write, and I assume
IncomingRtpPacket is always called on the same thread, meaning there cannot be a
write happening here if I'm not mistaken.

[Zhi Huang, 2017/03/31 06:44:04]

I added a lock here because the objects need to be copied when extracting the
contributing sources in last 10 seconds.

+  int64_t now = clock_->TimeInMilliseconds();
+  for (size_t i = 0; i < num_csrcs_; ++i) {
+    RtpContributingSource contributing_source(now, current_remote_csrc_[i]);
+    contributing_sources_buffer_[current_buffer_index_] = contributing_source;

[hbos, 2017/03/30 09:51:54]

nit: = RtpContributingSource(... or std::move to move instead of copying.

[Zhi Huang, 2017/03/31 06:44:04]

Done.

+    current_buffer_index_ =
+        (current_buffer_index_ + 1) % kContributingSourcesBufferSize;
+
+    if (current_buffer_size_ < kContributingSourcesBufferSize) {
+      ++current_buffer_size_;
+    }
+  }

[Taylor Brandstetter, 2017/03/30 22:55:37]

I think your initial idea (std::list of sources sorted by timestamp, and
std::unordered_map from CSRCs to list iterators) is preferable to this
algorithm. If the average number of contributing sources in a 10-second interval
is N, and the average number in an individual packet is M:

* Update time (per RTP packet): O(M), since the map is used for O(1) lookup per
CSRC, and it's O(1) to move each updated element to the front of the list to
keep it sorted.
* "Remove 10-second old things" time: Average O(1), worst case O(N). Since the
list is sorted by timestamp, it's easy to find everything that's older than 10
seconds, and most of the time it will be "nothing".
* Retrieval time: O(N)
* Memory: O(N)

So it has about the same performance characteristics as the current algorithm,
but "getContributingSources" would be more efficient, and we wouldn't have to
choose a hard-coded size for the ring buffer, crossing our fingers that it's
large enough for everyone.

I saw a comment earlier that "map+list will do heap allocs along the way. At
least one per incoming packet...", but this isn't true. It would only need a
heap alloc when a new contributing source shows up. So the main disadvantage I
see is the lack of memory locality of an std::list.

[Zhi Huang, 2017/03/31 06:44:04]

I can change it once we have an agreement on this. Changing it won't be too hard
since I only have to change this file..

[the sun, 2017/03/31 07:03:49]

Yeah, I don't particularly like that property either, but felt it an ok
tradeoff. I guess it would be possible to use a growing std::vector instead,
using it as a ring buffer once a 10s interval has been collected. Some kind of
cap would still be needed though.
What about the remove() then push_back() sequence when updating the timestamp?
Sure, the map isn't really a problem (but note that the 10s limit needs to be
managed in UpdateContributingSource()), and we can get rid of the O(N) remove()
behavior, but there'd still be a heap alloc when appending to
contributing_sources_.

[the sun, 2017/04/01 12:23:02]

Oh, I just realized the remove/push_back can be accomplished with
std::list::splice() instead! With this twist, the map+list solution seems the
most favorable design choice here. If this is what you had in mind all along I'm
sorry I didn't see it.

+}
+
 }  // namespace webrtc