Handler to filter packets from higher quality layers

erizo/src/erizo/MediaDefinitions.h

@@ -25,19 +25,20 @@ struct dataPacket {
   dataPacket() = default;
 
   dataPacket(int comp_, const char *data_, int length_, packetType type_, uint64_t received_time_ms_) :
-    comp{comp_}, length{length_}, type{type_}, received_time_ms{received_time_ms_}, is_keyframe{false} {
+    comp{comp_}, length{length_}, type{type_}, received_time_ms{received_time_ms_}, is_keyframe{false},
+    ending_of_layer_frame{false} {
       memcpy(data, data_, length_);
   }
 
   dataPacket(int comp_, const char *data_, int length_, packetType type_) :
     comp{comp_}, length{length_}, type{type_}, received_time_ms{ClockUtils::timePointToMs(clock::now())},
-    is_keyframe{false} {
+    is_keyframe{false}, ending_of_layer_frame{false} {
       memcpy(data, data_, length_);
   }
 
   dataPacket(int comp_, const unsigned char *data_, int length_) :
     comp{comp_}, length{length_}, type{VIDEO_PACKET}, received_time_ms{ClockUtils::timePointToMs(clock::now())},
-    is_keyframe{false} {
+    is_keyframe{false}, ending_of_layer_frame{false} {
       memcpy(data, data_, length_);
   }
 
@@ -65,6 +66,7 @@ struct dataPacket {
   std::vector<int> compatible_spatial_layers;
   std::vector<int> compatible_temporal_layers;
   bool is_keyframe;  // Note: It can be just a keyframe first packet in VP8
+  bool ending_of_layer_frame;
 };
 
 class Monitor {

erizo/src/erizo/WebRtcConnection.cpp

@@ -27,6 +27,8 @@
 #include "rtp/SRPacketHandler.h"
 #include "rtp/SenderBandwidthEstimationHandler.h"
 #include "rtp/LayerDetectorHandler.h"
+#include "rtp/QualityFilterHandler.h"
+#include "rtp/QualityManager.h"
 
 namespace erizo {
 DEFINE_LOGGER(WebRtcConnection, "WebRtcConnection");
@@ -48,6 +50,7 @@ WebRtcConnection::WebRtcConnection(std::shared_ptr<Worker> worker, const std::st
   source_fb_sink_ = this;
   sink_fb_source_ = this;
   stats_ = std::make_shared<Stats>();
+  quality_manager_ = std::make_shared<QualityManager>();
   globalState_ = CONN_INITIAL;
 
   rtcp_processor_ = std::make_shared<RtcpForwarder>(static_cast<MediaSink*>(this), static_cast<MediaSource*>(this));
@@ -233,8 +236,6 @@ bool WebRtcConnection::setRemoteSdp(const std::string &sdp) {
     }
   }
 
-
-
   initializePipeline();
 
   return true;
@@ -244,13 +245,15 @@ void WebRtcConnection::initializePipeline() {
   pipeline_->addService(shared_from_this());
   pipeline_->addService(rtcp_processor_);
   pipeline_->addService(stats_);
+  pipeline_->addService(quality_manager_);
 
   pipeline_->addFront(PacketReader(this));
 
   pipeline_->addFront(LayerDetectorHandler());
   pipeline_->addFront(RtcpProcessorHandler());
   pipeline_->addFront(IncomingStatsHandler());
   pipeline_->addFront(FecReceiverHandler());
+  pipeline_->addFront(QualityFilterHandler());
   pipeline_->addFront(RtpAudioMuteHandler());
   pipeline_->addFront(RtpSlideShowHandler());
   pipeline_->addFront(BandwidthEstimationHandler());
@@ -811,4 +814,11 @@ void WebRtcConnection::sendPacket(std::shared_ptr<dataPacket> p) {
   pipeline_->write(p);
 }
 
+void WebRtcConnection::setQualityLayer(int spatial_layer, int temporal_layer) {
+  asyncTask([spatial_layer, temporal_layer] (std::shared_ptr<WebRtcConnection> connection) {
+    connection->quality_manager_->setSpatialLayer(spatial_layer);
+    connection->quality_manager_->setTemporalLayer(temporal_layer);
+  });
+}
+
 }  // namespace erizo

erizo/src/erizo/WebRtcConnection.h

@@ -19,6 +19,7 @@
 #include "lib/Clock.h"
 #include "pipeline/Handler.h"
 #include "pipeline/Service.h"
+#include "rtp/QualityManager.h"
 
 namespace erizo {
 
@@ -105,6 +106,9 @@ class WebRtcConnection: public MediaSink, public MediaSource, public FeedbackSin
    * @return the size of the data sent
    */
   int sendPLI() override;
+
+  void setQualityLayer(int spatial_layer, int temporal_layer);
+
   /**
    * Sets the Event Listener for this WebRtcConnection
    */
@@ -211,6 +215,7 @@ class WebRtcConnection: public MediaSink, public MediaSource, public FeedbackSin
   std::shared_ptr<Transport> videoTransport_, audioTransport_;
 
   std::shared_ptr<Stats> stats_;
+  std::shared_ptr<QualityManager> quality_manager_;
   WebRTCEvent globalState_;
 
   boost::mutex updateStateMutex_;  // , slideShowMutex_;

erizo/src/erizo/rtp/LayerDetectorHandler.cpp

@@ -85,7 +85,7 @@ void LayerDetectorHandler::parseLayerInfoFromVP9(std::shared_ptr<dataPacket> pac
   int spatial_layer = payload->spatialID;
 
   packet->compatible_spatial_layers = {};
-  for (int i = 0; i <= spatial_layer; i++) {
+  for (int i = 5; i >= spatial_layer; i--) {
     packet->compatible_spatial_layers.push_back(i);
   }
 
@@ -110,6 +110,8 @@ void LayerDetectorHandler::parseLayerInfoFromVP9(std::shared_ptr<dataPacket> pac
   } else {
     packet->is_keyframe = false;
   }
+
+  packet->ending_of_layer_frame = payload->endingOfLayerFrame;
   delete payload;
 }
 

erizo/src/erizo/rtp/QualityFilterHandler.cpp

@@ -0,0 +1,141 @@
+#include "rtp/QualityFilterHandler.h"
+
+#include "./WebRtcConnection.h"
+#include "lib/ClockUtils.h"
+#include "rtp/RtpUtils.h"
+
+namespace erizo {
+
+DEFINE_LOGGER(QualityFilterHandler, "rtp.QualityFilterHandler");
+
+QualityFilterHandler::QualityFilterHandler()
+  : connection_{nullptr}, enabled_{true}, initialized_{false},
+  receiving_multiple_ssrc_{false}, target_spatial_layer_{0}, target_temporal_layer_{0},
+  video_sink_ssrc_{0}, video_source_ssrc_{0}, last_ssrc_received_{0},
+  max_video_bw_{0} {}
+
+void QualityFilterHandler::enable() {
+  enabled_ = true;
+}
+
+void QualityFilterHandler::disable() {
+  enabled_ = false;
+}
+
+void QualityFilterHandler::handleFeedbackPackets(std::shared_ptr<dataPacket> packet) {
+  RtpUtils::forEachRRBlock(packet, [this](RtcpHeader *chead) {
+    RtpUtils::updateREMB(chead, max_video_bw_);
+
+    RtpUtils::forEachNack(chead, [this, chead](uint16_t seq_num, uint16_t plb) {
+      SequenceNumber result = translator_.reverse(seq_num);
+      if (result.type == SequenceNumberType::Valid) {
+        chead->setNackPid(result.input);
+      }
+    });
+  });
+}
+
+void QualityFilterHandler::read(Context *ctx, std::shared_ptr<dataPacket> packet) {
+  if (enabled_) {
+    handleFeedbackPackets(packet);  // TODO(javier) remove this line when RTCP termination is enabled
+
+    // TODO(javier): Handle RRs and NACKs and translate Sequence Numbers?
+  }
+
+  ctx->fireRead(packet);
+}
+
+void QualityFilterHandler::checkLayers() {
+  int new_spatial_layer = quality_manager_->getSpatialLayer();
+  if (new_spatial_layer != target_spatial_layer_) {
+    sendPLI();
+    target_spatial_layer_ = new_spatial_layer;
+  }
+  int new_temporal_layer = quality_manager_->getTemporalLayer();
+  target_temporal_layer_ = new_temporal_layer;
+}
+
+void QualityFilterHandler::checkSSRCChange(uint32_t ssrc) {
+  if (last_ssrc_received_ != ssrc) {
+    translator_.reset();
+  }
+  last_ssrc_received_ = ssrc;
+}
+
+void QualityFilterHandler::sendPLI() {
+  getContext()->fireRead(RtpUtils::createPLI(video_sink_ssrc_, video_source_ssrc_));
+}
+
+void QualityFilterHandler::write(Context *ctx, std::shared_ptr<dataPacket> packet) {
+  RtcpHeader *chead = reinterpret_cast<RtcpHeader*>(packet->data);
+  if (!chead->isRtcp() && enabled_ && packet->type == VIDEO_PACKET) {
+    RtpHeader *rtp_header = reinterpret_cast<RtpHeader*>(packet->data);
+
+    checkLayers();
+
+    uint32_t ssrc = rtp_header->getSSRC();
+    uint16_t sequence_number = rtp_header->getSeqNumber();
+
+    if (ssrc != last_ssrc_received_) {
+      receiving_multiple_ssrc_ = true;
+    }
+
+    if (!packet->belongsToSpatialLayer(target_spatial_layer_)) {
+      if (ssrc == video_sink_ssrc_) {
+        translator_.get(sequence_number, true);
+      }
+      return;
+    }
+
+    checkSSRCChange(ssrc);
+    rtp_header->setSSRC(video_sink_ssrc_);
+
+    if (!packet->belongsToTemporalLayer(target_temporal_layer_)) {
+      translator_.get(sequence_number, true);
+      return;
+    }
+
+    SequenceNumber sequence_number_info = translator_.get(sequence_number, false);
+    if (sequence_number_info.type != SequenceNumberType::Valid) {
+      return;
+    }
+
+    if (packet->compatible_spatial_layers.back() == target_spatial_layer_ && packet->ending_of_layer_frame) {
+      rtp_header->setMarker(1);
+    }
+
+    rtp_header->setSeqNumber(sequence_number_info.output);
+  }
+
+  // TODO(javier): Handle SRs and translate Sequence Numbers?
+
+  ctx->fireWrite(packet);
+}
+
+void QualityFilterHandler::notifyUpdate() {
+  auto pipeline = getContext()->getPipelineShared();
+  if (!pipeline) {
+    return;
+  }
+
+  auto processor = pipeline->getService<RtcpProcessor>();
+
+  if (processor) {
+    max_video_bw_ = processor->getMaxVideoBW();
+  }
+
+  if (initialized_) {
+    return;
+  }
+
+  connection_ = pipeline->getService<WebRtcConnection>().get();
+  if (!connection_) {
+    return;
+  }
+
+  quality_manager_ = pipeline->getService<QualityManager>();
+
+  video_sink_ssrc_ = connection_->getVideoSinkSSRC();
+  video_source_ssrc_ = connection_->getVideoSourceSSRC();
+}
+}  // namespace erizo

erizo/src/erizo/rtp/QualityFilterHandler.h

@@ -0,0 +1,58 @@
+#ifndef ERIZO_SRC_ERIZO_RTP_QUALITYFILTERHANDLER_H_
+#define ERIZO_SRC_ERIZO_RTP_QUALITYFILTERHANDLER_H_
+
+#include <memory>
+#include <string>
+#include <random>
+#include <map>
+
+#include "./logger.h"
+#include "pipeline/Handler.h"
+#include "rtp/SequenceNumberTranslator.h"
+#include "rtp/QualityManager.h"
+
+namespace erizo {
+
+class WebRtcConnection;
+
+class QualityFilterHandler: public Handler, public std::enable_shared_from_this<QualityFilterHandler> {
+  DECLARE_LOGGER();
+
+
+ public:
+  QualityFilterHandler();
+
+  void enable() override;
+  void disable() override;
+
+  std::string getName() override {
+     return "quality_filter";
+  }
+
+  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 sendPLI();
+  void checkLayers();
+  void handleFeedbackPackets(std::shared_ptr<dataPacket> packet);
+  void checkSSRCChange(uint32_t ssrc);
+
+ private:
+  std::shared_ptr<QualityManager> quality_manager_;
+  SequenceNumberTranslator translator_;
+  WebRtcConnection *connection_;
+  bool enabled_;
+  bool initialized_;
+  bool receiving_multiple_ssrc_;
+  int target_spatial_layer_;
+  int target_temporal_layer_;
+  uint32_t video_sink_ssrc_;
+  uint32_t video_source_ssrc_;
+  uint32_t last_ssrc_received_;
+  uint32_t max_video_bw_;
+};
+}  // namespace erizo
+
+#endif  // ERIZO_SRC_ERIZO_RTP_QUALITYFILTERHANDLER_H_

erizo/src/erizo/rtp/QualityManager.cpp

@@ -0,0 +1,10 @@
+#include "rtp/QualityManager.h"
+
+namespace erizo {
+
+DEFINE_LOGGER(QualityManager, "rtp.QualityManager");
+
+QualityManager::QualityManager()
+  : spatial_layer_{0}, temporal_layer_{0} {}
+
+}  // namespace erizo

erizo/src/erizo/rtp/QualityManager.h

@@ -0,0 +1,27 @@
+#ifndef ERIZO_SRC_ERIZO_RTP_QUALITYMANAGER_H_
+#define ERIZO_SRC_ERIZO_RTP_QUALITYMANAGER_H_
+
+#include "./logger.h"
+#include "pipeline/Service.h"
+
+namespace erizo {
+
+class QualityManager: public Service, public std::enable_shared_from_this<QualityManager> {
+  DECLARE_LOGGER();
+
+ public:
+  QualityManager();
+
+  int getSpatialLayer() const { return spatial_layer_; }
+  int getTemporalLayer() const { return temporal_layer_; }
+
+  void setSpatialLayer(int spatial_layer) { spatial_layer_ = spatial_layer; }
+  void setTemporalLayer(int temporal_layer) { temporal_layer_ = temporal_layer; }
+
+ private:
+  int spatial_layer_;
+  int temporal_layer_;
+};
+}  // namespace erizo
+
+#endif  // ERIZO_SRC_ERIZO_RTP_QUALITYMANAGER_H_

erizo/src/erizo/rtp/RtcpAggregator.cpp

@@ -42,10 +42,6 @@ void RtcpAggregator::addSourceSsrc(uint32_t ssrc) {
   }
 }
 
-void RtcpAggregator::setMaxVideoBW(uint32_t bandwidth) {
-  this->maxVideoBw_ = bandwidth;
-}
-
 void RtcpAggregator::setPublisherBW(uint32_t bandwidth) {
   defaultVideoBw_ = (bandwidth*1.2) > maxVideoBw_? maxVideoBw_:(bandwidth*1.2);
 }

erizo/src/erizo/rtp/RtcpAggregator.h

@@ -23,7 +23,6 @@ class RtcpAggregator: public RtcpProcessor{
   RtcpAggregator(MediaSink* msink, MediaSource* msource, uint32_t maxVideoBw = 300000);
   virtual ~RtcpAggregator() {}
   void addSourceSsrc(uint32_t ssrc);
-  void setMaxVideoBW(uint32_t bandwidth);
   void setPublisherBW(uint32_t bandwidth);
   void analyzeSr(RtcpHeader* chead);
   int analyzeFeedback(char* buf, int len);

erizo/src/erizo/rtp/RtcpForwarder.cpp

@@ -33,10 +33,6 @@ void RtcpForwarder::addSourceSsrc(uint32_t ssrc) {
   }
 }
 
-void RtcpForwarder::setMaxVideoBW(uint32_t bandwidth) {
-  this->maxVideoBw_ = bandwidth;
-}
-
 void RtcpForwarder::setPublisherBW(uint32_t bandwidth) {
 }