Add a handler to control PLI sending rate (#788)

erizo/src/erizo/WebRtcConnection.cpp

@@ -29,6 +29,7 @@
 #include "rtp/LayerDetectorHandler.h"
 #include "rtp/QualityFilterHandler.h"
 #include "rtp/QualityManager.h"
+#include "rtp/PliPacerHandler.h"
 
 namespace erizo {
 DEFINE_LOGGER(WebRtcConnection, "WebRtcConnection");
@@ -256,6 +257,7 @@ void WebRtcConnection::initializePipeline() {
   pipeline_->addFront(QualityFilterHandler());
   pipeline_->addFront(RtpAudioMuteHandler());
   pipeline_->addFront(RtpSlideShowHandler());
+  pipeline_->addFront(PliPacerHandler());
   pipeline_->addFront(BandwidthEstimationHandler());
   pipeline_->addFront(RtcpFeedbackGenerationHandler());
   pipeline_->addFront(RtpRetransmissionHandler());

erizo/src/erizo/rtp/PliPacerHandler.cpp

@@ -0,0 +1,87 @@
+#include "rtp/PliPacerHandler.h"
+
+#include "rtp/RtpUtils.h"
+#include "./MediaDefinitions.h"
+#include "./WebRtcConnection.h"
+
+namespace erizo {
+
+DEFINE_LOGGER(PliPacerHandler, "rtp.PliPacerHandler");
+
+constexpr duration PliPacerHandler::kMinPLIPeriod;
+constexpr duration PliPacerHandler::kKeyframeTimeout;
+
+PliPacerHandler::PliPacerHandler(std::shared_ptr<erizo::Clock> the_clock)
+    : enabled_{true}, connection_{nullptr}, clock_{the_clock}, time_last_keyframe_{clock_->now()},
+      waiting_for_keyframe_{false}, scheduled_pli_{-1},
+      video_sink_ssrc_{0}, video_source_ssrc_{0}, fir_seq_number_{0} {}
+
+void PliPacerHandler::enable() {
+  enabled_ = true;
+}
+
+void PliPacerHandler::disable() {
+  enabled_ = false;
+}
+
+void PliPacerHandler::notifyUpdate() {
+  auto pipeline = getContext()->getPipelineShared();
+  if (pipeline && !connection_) {
+    connection_ = pipeline->getService<WebRtcConnection>().get();
+    video_sink_ssrc_ = connection_->getVideoSinkSSRC();
+    video_source_ssrc_ = connection_->getVideoSourceSSRC();
+  }
+}
+
+void PliPacerHandler::read(Context *ctx, std::shared_ptr<dataPacket> packet) {
+  if (enabled_ && packet->is_keyframe) {
+    time_last_keyframe_ = clock_->now();
+    waiting_for_keyframe_ = false;
+    connection_->getWorker()->unschedule(scheduled_pli_);
+    scheduled_pli_ = -1;
+  }
+  ctx->fireRead(packet);
+}
+
+void PliPacerHandler::sendPLI() {
+  getContext()->fireWrite(RtpUtils::createPLI(video_source_ssrc_, video_sink_ssrc_));
+  scheduleNextPLI();
+}
+
+void PliPacerHandler::sendFIR() {
+  ELOG_WARN("%s message: Timed out waiting for a keyframe", connection_->toLog());
+  getContext()->fireWrite(RtpUtils::createFIR(video_source_ssrc_, video_sink_ssrc_, fir_seq_number_++));
+  getContext()->fireWrite(RtpUtils::createFIR(video_source_ssrc_, video_sink_ssrc_, fir_seq_number_++));
+  getContext()->fireWrite(RtpUtils::createFIR(video_source_ssrc_, video_sink_ssrc_, fir_seq_number_++));
+  waiting_for_keyframe_ = false;
+  scheduled_pli_ = -1;
+}
+
+void PliPacerHandler::scheduleNextPLI() {
+  if (!waiting_for_keyframe_ || !enabled_) {
+    return;
+  }
+  std::weak_ptr<PliPacerHandler> weak_this = shared_from_this();
+  scheduled_pli_ = connection_->getWorker()->scheduleFromNow([weak_this] {
+    if (auto this_ptr = weak_this.lock()) {
+      if (this_ptr->clock_->now() - this_ptr->time_last_keyframe_ >= kKeyframeTimeout) {
+        this_ptr->sendFIR();
+        return;
+      }
+      this_ptr->sendPLI();
+    }
+  }, kMinPLIPeriod);
+}
+
+void PliPacerHandler::write(Context *ctx, std::shared_ptr<dataPacket> packet) {
+  if (enabled_ && RtpUtils::isPLI(packet)) {
+    if (waiting_for_keyframe_) {
+      return;
+    }
+    waiting_for_keyframe_ = true;
+    scheduleNextPLI();
+  }
+  ctx->fireWrite(packet);
+}
+
+}  // namespace erizo

erizo/src/erizo/rtp/PliPacerHandler.h

@@ -0,0 +1,54 @@
+#ifndef ERIZO_SRC_ERIZO_RTP_PLIPACERHANDLER_H_
+#define ERIZO_SRC_ERIZO_RTP_PLIPACERHANDLER_H_
+
+#include <string>
+
+#include "./logger.h"
+#include "pipeline/Handler.h"
+#include "lib/Clock.h"
+
+namespace erizo {
+
+class WebRtcConnection;
+
+class PliPacerHandler: public Handler, public std::enable_shared_from_this<PliPacerHandler> {
+  DECLARE_LOGGER();
+
+ public:
+  static constexpr duration kMinPLIPeriod = std::chrono::milliseconds(200);
+  static constexpr duration kKeyframeTimeout = std::chrono::seconds(4);
+
+ public:
+  explicit PliPacerHandler(std::shared_ptr<erizo::Clock> the_clock = std::make_shared<SteadyClock>());
+
+  void enable() override;
+  void disable() override;
+
+  std::string getName() override {
+    return "pli-pacer";
+  }
+
+  void read(Context *ctx, std::shared_ptr<dataPacket> packet) override;
+  void write(Context *ctx, std::shared_ptr<dataPacket> packet) override;
+  void notifyUpdate() override;
+
+ private:
+  void scheduleNextPLI();
+  void sendPLI();
+  void sendFIR();
+
+ private:
+  bool enabled_;
+  WebRtcConnection* connection_;
+  std::shared_ptr<erizo::Clock> clock_;
+  time_point time_last_keyframe_;
+  bool waiting_for_keyframe_;
+  int scheduled_pli_;
+  uint32_t video_sink_ssrc_;
+  uint32_t video_source_ssrc_;
+  uint8_t fir_seq_number_;
+};
+
+}  // namespace erizo
+
+#endif  // ERIZO_SRC_ERIZO_RTP_PLIPACERHANDLER_H_

erizo/src/erizo/rtp/RtpHeaders.h

@@ -1,3 +1,4 @@
+
 /*
  * RtpHeaders.h
  */
@@ -350,6 +351,13 @@ class RtcpHeader {
       uint32_t ssrcsource;
       uint32_t fci;
     } pli;
+
+    struct fir_t {
+      uint32_t ssrcsource;
+      uint32_t mediasource;
+      uint32_t seqnumber:8;
+      uint32_t reserved:24;
+    } fir;
   } report;
 
   inline RtcpHeader() : blockcount(0), padding(0), version(2), packettype(0), length(0), ssrc(0) {
@@ -511,6 +519,12 @@ class RtcpHeader {
   inline void setFCI(uint32_t fci) {
     report.pli.fci = htonl(fci);
   }
+  inline void setFIRSourceSSRC(uint32_t ssrc) {
+    report.fir.mediasource = htonl(ssrc);
+  }
+  inline void setFIRSequenceNumber(uint8_t seq_number) {
+    report.fir.seqnumber = seq_number;
+  }
 };
 
 

erizo/src/erizo/rtp/RtpUtils.cpp

@@ -35,10 +35,32 @@ void RtpUtils::forEachNack(RtcpHeader *chead, std::function<void(uint16_t, uint1
   }
 }
 
+bool RtpUtils::isPLI(std::shared_ptr<dataPacket> packet) {
+  bool is_pli = false;
+  forEachRRBlock(packet, [&is_pli] (RtcpHeader *header) {
+    if (header->getPacketType() == RTCP_PS_Feedback_PT &&
+        header->getBlockCount() == RTCP_PLI_FMT) {
+          is_pli = true;
+        }
+  });
+  return is_pli;
+}
+
+bool RtpUtils::isFIR(std::shared_ptr<dataPacket> packet) {
+  bool is_fir = false;
+  forEachRRBlock(packet, [&is_fir] (RtcpHeader *header) {
+    if (header->getPacketType() == RTCP_PS_Feedback_PT &&
+        header->getBlockCount() == RTCP_FIR_FMT) {
+          is_fir = true;
+        }
+  });
+  return is_fir;
+}
+
 std::shared_ptr<dataPacket> RtpUtils::createPLI(uint32_t source_ssrc, uint32_t sink_ssrc) {
   RtcpHeader pli;
   pli.setPacketType(RTCP_PS_Feedback_PT);
-  pli.setBlockCount(1);
+  pli.setBlockCount(RTCP_PLI_FMT);
   pli.setSSRC(sink_ssrc);
   pli.setSourceSSRC(source_ssrc);
   pli.setLength(2);
@@ -47,6 +69,21 @@ std::shared_ptr<dataPacket> RtpUtils::createPLI(uint32_t source_ssrc, uint32_t s
   return std::make_shared<dataPacket>(0, buf, len, VIDEO_PACKET);
 }
 
+std::shared_ptr<dataPacket> RtpUtils::createFIR(uint32_t source_ssrc, uint32_t sink_ssrc, uint8_t seq_number) {
+  RtcpHeader fir;
+  fir.setPacketType(RTCP_PS_Feedback_PT);
+  fir.setBlockCount(RTCP_FIR_FMT);
+  fir.setSSRC(sink_ssrc);
+  fir.setSourceSSRC(source_ssrc);
+  fir.setLength(4);
+  fir.setFIRSourceSSRC(source_ssrc);
+  fir.setFIRSequenceNumber(seq_number);
+  char *buf = reinterpret_cast<char*>(&fir);
+  int len = (fir.getLength() + 1) * 4;
+  return std::make_shared<dataPacket>(0, buf, len, VIDEO_PACKET);
+}
+
+
 int RtpUtils::getPaddingLength(std::shared_ptr<dataPacket> packet) {
   RtpHeader *rtp_header = reinterpret_cast<RtpHeader*>(packet->data);
   if (rtp_header->hasPadding()) {

erizo/src/erizo/rtp/RtpUtils.h

@@ -19,10 +19,16 @@ class RtpUtils {
 
   static void updateREMB(RtcpHeader *chead, uint bitrate);
 
+  static bool isPLI(std::shared_ptr<dataPacket> packet);
+
+  static bool isFIR(std::shared_ptr<dataPacket> packet);
+
   static void forEachNack(RtcpHeader *chead, std::function<void(uint16_t, uint16_t)> f);
 
   static std::shared_ptr<dataPacket> createPLI(uint32_t source_ssrc, uint32_t sink_ssrc);
 
+  static std::shared_ptr<dataPacket> createFIR(uint32_t source_ssrc, uint32_t sink_ssrc, uint8_t seq_number);
+
   static int getPaddingLength(std::shared_ptr<dataPacket> packet);
 };