Reland of Only compare sequence numbers from the same SSRC in ForwardErrorCorrection.

webrtc/modules/rtp_rtcp/source/forward_error_correction.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.
  */
 
@@ skipping 36 matching lines @@
 }
 
 // This comparator is used to compare std::unique_ptr's pointing to
 // subclasses of SortablePackets. It needs to be parametric since
 // the std::unique_ptr's are not covariant w.r.t. the types that
 // they are pointing to.
 template <typename S, typename T>
 bool ForwardErrorCorrection::SortablePacket::LessThan::operator() (
     const S& first,
     const T& second) {
+  RTC_DCHECK_EQ(first->ssrc, second->ssrc);
   return IsNewerSequenceNumber(second->seq_num, first->seq_num);
 }
 
 ForwardErrorCorrection::ReceivedPacket::ReceivedPacket() = default;
 ForwardErrorCorrection::ReceivedPacket::~ReceivedPacket() = default;
 
 ForwardErrorCorrection::RecoveredPacket::RecoveredPacket() = default;
 ForwardErrorCorrection::RecoveredPacket::~RecoveredPacket() = default;
 
 ForwardErrorCorrection::ProtectedPacket::ProtectedPacket() = default;
 ForwardErrorCorrection::ProtectedPacket::~ProtectedPacket() = default;
 
 ForwardErrorCorrection::ReceivedFecPacket::ReceivedFecPacket() = default;
 ForwardErrorCorrection::ReceivedFecPacket::~ReceivedFecPacket() = default;
 
 ForwardErrorCorrection::ForwardErrorCorrection(
     std::unique_ptr<FecHeaderReader> fec_header_reader,
-    std::unique_ptr<FecHeaderWriter> fec_header_writer)
-    : fec_header_reader_(std::move(fec_header_reader)),
+    std::unique_ptr<FecHeaderWriter> fec_header_writer,
+    uint32_t ssrc,
+    uint32_t protected_media_ssrc)
+    : ssrc_(ssrc),
+      protected_media_ssrc_(protected_media_ssrc),
+      fec_header_reader_(std::move(fec_header_reader)),
       fec_header_writer_(std::move(fec_header_writer)),
       generated_fec_packets_(fec_header_writer_->MaxFecPackets()),
       packet_mask_size_(0) {}
 
 ForwardErrorCorrection::~ForwardErrorCorrection() = default;
 
-std::unique_ptr<ForwardErrorCorrection> ForwardErrorCorrection::CreateUlpfec() {
+std::unique_ptr<ForwardErrorCorrection> ForwardErrorCorrection::CreateUlpfec(
+    uint32_t ssrc) {
   std::unique_ptr<FecHeaderReader> fec_header_reader(new UlpfecHeaderReader());
   std::unique_ptr<FecHeaderWriter> fec_header_writer(new UlpfecHeaderWriter());
   return std::unique_ptr<ForwardErrorCorrection>(new ForwardErrorCorrection(
-      std::move(fec_header_reader), std::move(fec_header_writer)));
+      std::move(fec_header_reader), std::move(fec_header_writer), ssrc, ssrc));
 }
 
-std::unique_ptr<ForwardErrorCorrection>
-ForwardErrorCorrection::CreateFlexfec() {
+std::unique_ptr<ForwardErrorCorrection> ForwardErrorCorrection::CreateFlexfec(
+    uint32_t ssrc,
+    uint32_t protected_media_ssrc) {
   std::unique_ptr<FecHeaderReader> fec_header_reader(new FlexfecHeaderReader());
   std::unique_ptr<FecHeaderWriter> fec_header_writer(new FlexfecHeaderWriter());
   return std::unique_ptr<ForwardErrorCorrection>(new ForwardErrorCorrection(
-      std::move(fec_header_reader), std::move(fec_header_writer)));
+      std::move(fec_header_reader), std::move(fec_header_writer), ssrc,
+      protected_media_ssrc));
 }
 
 int ForwardErrorCorrection::EncodeFec(const PacketList& media_packets,
                                       uint8_t protection_factor,
                                       int num_important_packets,
                                       bool use_unequal_protection,
                                       FecMaskType fec_mask_type,
                                       std::list<Packet*>* fec_packets) {
   const size_t num_media_packets = media_packets.size();
 
@@ skipping 223 matching lines @@
 void ForwardErrorCorrection::ResetState(
     RecoveredPacketList* recovered_packets) {
   // Free the memory for any existing recovered packets, if the caller hasn't.
   recovered_packets->clear();
   received_fec_packets_.clear();
 }
 
 void ForwardErrorCorrection::InsertMediaPacket(
     RecoveredPacketList* recovered_packets,
     ReceivedPacket* received_packet) {
+  RTC_DCHECK_EQ(received_packet->ssrc, protected_media_ssrc_);
+
   // Search for duplicate packets.
   for (const auto& recovered_packet : *recovered_packets) {
-    if (received_packet->seq_num == recovered_packet->seq_num) {
+    RTC_DCHECK_EQ(recovered_packet->ssrc, received_packet->ssrc);
+    if (recovered_packet->seq_num == received_packet->seq_num) {
       // Duplicate packet, no need to add to list.
       // Delete duplicate media packet data.
       received_packet->pkt = nullptr;
       return;
     }
   }
+
   std::unique_ptr<RecoveredPacket> recovered_packet(new RecoveredPacket());
   // This "recovered packet" was not recovered using parity packets.
   recovered_packet->was_recovered = false;
   // This media packet has already been passed on.
   recovered_packet->returned = true;
+  recovered_packet->ssrc = received_packet->ssrc;
   recovered_packet->seq_num = received_packet->seq_num;
   recovered_packet->pkt = received_packet->pkt;
   recovered_packet->pkt->length = received_packet->pkt->length;
   // TODO(holmer): Consider replacing this with a binary search for the right
   // position, and then just insert the new packet. Would get rid of the sort.
   RecoveredPacket* recovered_packet_ptr = recovered_packet.get();
   recovered_packets->push_back(std::move(recovered_packet));
   recovered_packets->sort(SortablePacket::LessThan());
   UpdateCoveringFecPackets(*recovered_packet_ptr);
 }
 
 void ForwardErrorCorrection::UpdateCoveringFecPackets(
     const RecoveredPacket& packet) {
   for (auto& fec_packet : received_fec_packets_) {
     // Is this FEC packet protecting the media packet |packet|?
     auto protected_it = std::lower_bound(fec_packet->protected_packets.begin(),
                                          fec_packet->protected_packets.end(),
                                          &packet, SortablePacket::LessThan());
     if (protected_it != fec_packet->protected_packets.end() &&
         (*protected_it)->seq_num == packet.seq_num) {
       // Found an FEC packet which is protecting |packet|.
       (*protected_it)->pkt = packet.pkt;
     }
   }
 }
 
 void ForwardErrorCorrection::InsertFecPacket(
     const RecoveredPacketList& recovered_packets,
     ReceivedPacket* received_packet) {
+  RTC_DCHECK_EQ(received_packet->ssrc, ssrc_);
+
   // Check for duplicate.
   for (const auto& existing_fec_packet : received_fec_packets_) {
-    if (received_packet->seq_num == existing_fec_packet->seq_num) {
+    RTC_DCHECK_EQ(existing_fec_packet->ssrc, received_packet->ssrc);
+    if (existing_fec_packet->seq_num == received_packet->seq_num) {
       // Delete duplicate FEC packet data.
       received_packet->pkt = nullptr;
       return;
     }
   }
+
   std::unique_ptr<ReceivedFecPacket> fec_packet(new ReceivedFecPacket());
   fec_packet->pkt = received_packet->pkt;
+  fec_packet->ssrc = received_packet->ssrc;
   fec_packet->seq_num = received_packet->seq_num;
-  fec_packet->ssrc = received_packet->ssrc;
   // Parse ULPFEC/FlexFEC header specific info.
   bool ret = fec_header_reader_->ReadFecHeader(fec_packet.get());
   if (!ret) {
     return;
   }
+
+  // TODO(brandtr): Update here when we support multistream protection.
+  if (fec_packet->protected_ssrc != protected_media_ssrc_) {
+    LOG(LS_INFO)
+        << "Received FEC packet is protecting an unknown media SSRC; dropping.";
+    return;
+  }
+
   // Parse packet mask from header and represent as protected packets.
   for (uint16_t byte_idx = 0; byte_idx < fec_packet->packet_mask_size;
        ++byte_idx) {
     uint8_t packet_mask =
         fec_packet->pkt->data[fec_packet->packet_mask_offset + byte_idx];
     for (uint16_t bit_idx = 0; bit_idx < 8; ++bit_idx) {
       if (packet_mask & (1 << (7 - bit_idx))) {
         std::unique_ptr<ProtectedPacket> protected_packet(
             new ProtectedPacket());
         // This wraps naturally with the sequence number.
+        protected_packet->ssrc = protected_media_ssrc_;
         protected_packet->seq_num = static_cast<uint16_t>(
             fec_packet->seq_num_base + (byte_idx << 3) + bit_idx);
         protected_packet->pkt = nullptr;
         fec_packet->protected_packets.push_back(std::move(protected_packet));
       }
     }
   }
+
   if (fec_packet->protected_packets.empty()) {
     // All-zero packet mask; we can discard this FEC packet.
     LOG(LS_WARNING) << "Received FEC packet has an all-zero packet mask.";
   } else {
     AssignRecoveredPackets(recovered_packets, fec_packet.get());
     // TODO(holmer): Consider replacing this with a binary search for the right
     // position, and then just insert the new packet. Would get rid of the sort.
-    //
-    // For correct decoding, |received_fec_packets_| does not necessarily
-    // need to be sorted by sequence number (see decoding algorithm in
-    // AttemptRecover()). By keeping it sorted we try to recover the
-    // oldest lost packets first, however.
     received_fec_packets_.push_back(std::move(fec_packet));
     received_fec_packets_.sort(SortablePacket::LessThan());
     const size_t max_fec_packets = fec_header_reader_->MaxFecPackets();
     if (received_fec_packets_.size() > max_fec_packets) {
       received_fec_packets_.pop_front();
     }
     RTC_DCHECK_LE(received_fec_packets_.size(), max_fec_packets);
   }
 }
 
@@ skipping 23 matching lines @@
     }
   }
 }
 
 void ForwardErrorCorrection::InsertPackets(
     ReceivedPacketList* received_packets,
     RecoveredPacketList* recovered_packets) {
   while (!received_packets->empty()) {
     ReceivedPacket* received_packet = received_packets->front().get();
 
-    // Check for discarding oldest FEC packet, to avoid wrong FEC decoding from
-    // sequence number wrap-around. Detection of old FEC packet is based on
-    // sequence number difference of received packet and oldest packet in FEC
-    // packet list.
+    // Discard old FEC packets such that the sequence numbers in
+    // |received_fec_packets_| span at most 1/2 of the sequence number space.
+    // This is important for keeping |received_fec_packets_| sorted, and may
+    // also reduce the possibility of incorrect decoding due to sequence number
+    // wrap-around.
     // TODO(marpan/holmer): We should be able to improve detection/discarding of
     // old FEC packets based on timestamp information or better sequence number
     // thresholding (e.g., to distinguish between wrap-around and reordering).
-    if (!received_fec_packets_.empty()) {
-      uint16_t seq_num_diff =
-          abs(static_cast<int>(received_packet->seq_num) -
-              static_cast<int>(received_fec_packets_.front()->seq_num));
-      if (seq_num_diff > 0x3fff) {
-        received_fec_packets_.pop_front();
+    if (!received_fec_packets_.empty() &&
+        received_packet->ssrc == received_fec_packets_.front()->ssrc) {
+      // It only makes sense to detect wrap-around when |received_packet|
+      // and |front_received_fec_packet| belong to the same sequence number
+      // space, i.e., the same SSRC. This happens when |received_packet|
+      // is a FEC packet, or if |received_packet| is a media packet and
+      // RED+ULPFEC is used.
+      auto it = received_fec_packets_.begin();
+      while (it != received_fec_packets_.end()) {
+        uint16_t seq_num_diff = abs(static_cast<int>(received_packet->seq_num) -
+                                    static_cast<int>((*it)->seq_num));
+        if (seq_num_diff > 0x3fff) {
+          it = received_fec_packets_.erase(it);
+        } else {
+          // No need to keep iterating, since |received_fec_packets_| is sorted.
+          break;
+        }
       }
     }
 
     if (received_packet->is_fec) {
       InsertFecPacket(*recovered_packets, received_packet);
     } else {
       InsertMediaPacket(recovered_packets, received_packet);
     }
     // Delete the received packet "wrapper".
     received_packets->pop_front();
@@ skipping 49 matching lines @@
     LOG(LS_WARNING) << "The recovered packet had a length larger than a "
                     << "typical IP packet, and is thus dropped.";
     return false;
   }
   // Set the SN field.
   ByteWriter<uint16_t>::WriteBigEndian(&recovered_packet->pkt->data[2],
                                        recovered_packet->seq_num);
   // Set the SSRC field.
   ByteWriter<uint32_t>::WriteBigEndian(&recovered_packet->pkt->data[8],
                                        fec_packet.protected_ssrc);
+  recovered_packet->ssrc = fec_packet.protected_ssrc;
   return true;
 }
 
 void ForwardErrorCorrection::XorHeaders(const Packet& src, Packet* dst) {
   // XOR the first 2 bytes of the header: V, P, X, CC, M, PT fields.
   dst->data[0] ^= src.data[0];
   dst->data[1] ^= src.data[1];
 
   // XOR the length recovery field.
   uint8_t src_payload_length_network_order[2];
@@ skipping 115 matching lines @@
   return (packet[2] << 8) + packet[3];
 }
 
 uint32_t ForwardErrorCorrection::ParseSsrc(uint8_t* packet) {
   return (packet[8] << 24) + (packet[9] << 16) + (packet[10] << 8) + packet[11];
 }
 
 int ForwardErrorCorrection::DecodeFec(
     ReceivedPacketList* received_packets,
     RecoveredPacketList* recovered_packets) {
-  // TODO(marpan/ajm): can we check for multiple ULP headers, and return an
-  // error?
+  RTC_DCHECK(received_packets);
+  RTC_DCHECK(recovered_packets);
+
   const size_t max_media_packets = fec_header_reader_->MaxMediaPackets();
   if (recovered_packets->size() == max_media_packets) {
-    const unsigned int seq_num_diff =
-        abs(static_cast<int>(received_packets->front()->seq_num) -
-            static_cast<int>(recovered_packets->back()->seq_num));
-    if (seq_num_diff > max_media_packets) {
-      // A big gap in sequence numbers. The old recovered packets
-      // are now useless, so it's safe to do a reset.
-      ResetState(recovered_packets);
+    const RecoveredPacket* back_recovered_packet =
+        recovered_packets->back().get();
+    for (const auto& received_packet : *received_packets) {
+      if (received_packet->ssrc == back_recovered_packet->ssrc) {
+        const unsigned int seq_num_diff =
+            abs(static_cast<int>(received_packet->seq_num) -
+                static_cast<int>(back_recovered_packet->seq_num));
+        if (seq_num_diff > max_media_packets) {
+          // A big gap in sequence numbers. The old recovered packets
+          // are now useless, so it's safe to do a reset.
+          LOG(LS_INFO) << "Big gap in media/ULPFEC sequence numbers. No need "
+                          "to keep the old packets in the FEC buffers, thus "
+                          "resetting them.";
+          ResetState(recovered_packets);
+          break;
+        }
+      }
     }
   }
+
   InsertPackets(received_packets, recovered_packets);
   AttemptRecovery(recovered_packets);
+
   return 0;
 }
 
 size_t ForwardErrorCorrection::MaxPacketOverhead() const {
   return fec_header_writer_->MaxPacketOverhead();
 }
 
 FecHeaderReader::FecHeaderReader(size_t max_media_packets,
                                  size_t max_fec_packets)
     : max_media_packets_(max_media_packets),
@@ skipping 24 matching lines @@
 
 size_t FecHeaderWriter::MaxFecPackets() const {
   return max_fec_packets_;
 }
 
 size_t FecHeaderWriter::MaxPacketOverhead() const {
   return max_packet_overhead_;
 }
 
 }  // namespace webrtc