Implement target connection bandwidth (lynckia#1736)

erizo/src/erizo/WebRtcConnection.cpp

@@ -40,6 +40,7 @@ WebRtcConnection::WebRtcConnection(std::shared_ptr<Worker> worker, std::shared_p
     remote_sdp_{std::make_shared<SdpInfo>(rtp_mappings)}, local_sdp_{std::make_shared<SdpInfo>(rtp_mappings)},
     audio_muted_{false}, video_muted_{false}, first_remote_sdp_processed_{false},
     enable_connection_quality_check_{enable_connection_quality_check}, encrypt_transport_{encrypt_transport},
+    connection_target_bw_{0},
     pipeline_{Pipeline::create()},
     pipeline_initialized_{false}, latest_mid_{0} {
   stats_ = std::make_shared<Stats>();

erizo/src/erizo/WebRtcConnection.h

@@ -160,6 +160,11 @@ class WebRtcConnection: public TransportListener, public LogContext, public Hand
 
   void setBwDistributionConfigSync(BwDistributionConfig distribution_config);
 
+  uint32_t getConnectionTargetBw() { return connection_target_bw_.load(); }
+  void setConnectionTargetBw(uint32_t target_bw) {
+    connection_target_bw_ = target_bw;
+    }
+
   inline std::string toLog() {
     return "id: " + connection_id_ + ", distributor: "
       + std::to_string(bw_distribution_config_.selected_distributor)
@@ -237,6 +242,7 @@ class WebRtcConnection: public TransportListener, public LogContext, public Hand
   ConnectionQualityCheck connection_quality_check_;
   bool enable_connection_quality_check_;
   bool encrypt_transport_;
+  std::atomic <uint32_t> connection_target_bw_;
   Pipeline::Ptr pipeline_;
   bool pipeline_initialized_;
   std::shared_ptr<HandlerManager> handler_manager_;

erizo/src/erizo/rtp/RtpPaddingManagerHandler.cpp

@@ -114,8 +114,11 @@ void RtpPaddingManagerHandler::recalculatePaddingRate() {
   bool can_send_more_bitrate = (kBitrateComparisonMargin * media_bitrate) < estimated_bandwidth;
   bool estimated_is_high_enough = estimated_bandwidth > (target_bitrate * kBitrateComparisonMargin);
   bool has_unnasigned_bitrate = false;
+  bool has_connection_target_bitrate = connection_->getConnectionTargetBw() > 0;
   if (stats_->getNode()["total"].hasChild("unnasignedBitrate")) {
-    has_unnasigned_bitrate = stats_->getNode()["total"]["unnasignedBitrate"].value() > kUnnasignedBitrateMargin;
+    has_unnasigned_bitrate =
+        stats_->getNode()["total"]["unnasignedBitrate"].value() > kUnnasignedBitrateMargin &&
+        !has_connection_target_bitrate;
   }
   if (estimated_is_high_enough || has_unnasigned_bitrate) {
     target_padding_bitrate = 0;
@@ -183,6 +186,7 @@ int64_t RtpPaddingManagerHandler::getTotalTargetBitrate() {
       }
       target_bitrate += media_stream->getTargetVideoBitrate();
     });
+  target_bitrate = std::max(target_bitrate, static_cast<int64_t>(connection_->getConnectionTargetBw()));
   stats_->getNode()["total"].insertStat("targetBitrate",
     CumulativeStat{static_cast<uint64_t>(target_bitrate)});
 

erizoAPI/WebRtcConnection.cc

@@ -169,6 +169,7 @@ NAN_MODULE_INIT(WebRtcConnection::Init) {
   Nan::SetPrototypeMethod(tpl, "removeMediaStream", removeMediaStream);
   Nan::SetPrototypeMethod(tpl, "copySdpToLocalDescription", copySdpToLocalDescription);
   Nan::SetPrototypeMethod(tpl, "setBwDistributionConfig", setBwDistributionConfig);
+  Nan::SetPrototypeMethod(tpl, "setConnectionTargetBw", setConnectionTargetBw);
   Nan::SetPrototypeMethod(tpl, "getStats", getStats);
   Nan::SetPrototypeMethod(tpl, "maybeRestartIce", maybeRestartIce);
   Nan::SetPrototypeMethod(tpl, "getDurationDistribution", getDurationDistribution);
@@ -483,6 +484,16 @@ NAN_METHOD(WebRtcConnection::setBwDistributionConfig) {
   me->setBwDistributionConfig(distrib_config);
 }
 
+NAN_METHOD(WebRtcConnection::setConnectionTargetBw) {
+  WebRtcConnection* obj = Nan::ObjectWrap::Unwrap<WebRtcConnection>(info.Holder());
+  std::shared_ptr<erizo::WebRtcConnection> me = obj->me;
+  if (!me) {
+    return;
+  }
+  int connection_target_bw = Nan::To<int>(info[0]).FromJust();
+  me->setConnectionTargetBw(connection_target_bw);
+}
+
 NAN_METHOD(WebRtcConnection::addRemoteCandidate) {
   WebRtcConnection* obj = Nan::ObjectWrap::Unwrap<WebRtcConnection>(info.Holder());
   std::shared_ptr<erizo::WebRtcConnection> me = obj->me;

erizoAPI/WebRtcConnection.h

@@ -135,6 +135,7 @@ class WebRtcConnection : public erizo::WebRtcConnectionEventListener,
     static NAN_METHOD(copySdpToLocalDescription);
 
     static NAN_METHOD(setBwDistributionConfig);
+    static NAN_METHOD(setConnectionTargetBw);
 
     static NAN_METHOD(getStats);
 

erizo_controller/erizoClient/src/Room.js

@@ -752,6 +752,7 @@ const Room = (altIo, altConnectionHelpers, altConnectionManager, specInput) => {
       spec.defaultVideoBW = response.defaultVideoBW;
       spec.maxVideoBW = response.maxVideoBW;
       that.streamPriorityStrategy = response.streamPriorityStrategy;
+      that.connectionTargetBw = response.connectionTargetBw;
 
       // 2- Retrieve list of streams
       const streamIndices = Object.keys(streams);
@@ -1057,6 +1058,16 @@ const Room = (altIo, altConnectionHelpers, altConnectionManager, specInput) => {
 
   that.setStreamPriorityStrategy = (strategyId, callback = () => { }) => {
     socket.sendMessage('setStreamPriorityStrategy', strategyId, (result) => {
+      that.streamPriorityStrategy = strategyId;
+      if (result) {
+        callback(result);
+      }
+    });
+  };
+
+  that.setConnectionTargetBandwidth = (connectionTargetBw, callback = () => { }) => {
+    socket.sendMessage('setConnectionTargetBandwidth', connectionTargetBw, (result) => {
+      that.connectionTargetBw = connectionTargetBw;
       if (result) {
         callback(result);
       }

erizo_controller/erizoController/erizoController.js

@@ -353,6 +353,7 @@ const listen = () => {
           clientId: client.id,
           singlePC: options.singlePC,
           streamPriorityStrategy: options.streamPriorityStrategy,
+          connectionTargetBw: options.connectionTargetBw,
           p2p: room.p2p,
           defaultVideoBW: global.config.erizoController.defaultVideoBW,
           maxVideoBW: global.config.erizoController.maxVideoBW,

erizo_controller/erizoController/models/Client.js

@@ -20,6 +20,8 @@ class Client extends events.EventEmitter {
     this.options = options;
     this.options.streamPriorityStrategy =
       Client.getStreamPriorityStrategy(options.streamPriorityStrategy);
+    this.options.connectionTargetBw = Number.isInteger(options.connectionTargetBw) ?
+      options.connectionTargetBw : this.options.streamPriorityStrategy.connectionTargetBw;
     this.socketEventListeners = new Map();
     this.listenToSocketEvents();
     this.user = { name: token.userName, role: token.role, permissions: {} };
@@ -46,6 +48,7 @@ class Client extends events.EventEmitter {
     this.socketEventListeners.set('getStreamStats', this.onGetStreamStats.bind(this));
     this.socketEventListeners.set('clientDisconnection', this.onClientDisconnection.bind(this));
     this.socketEventListeners.set('setStreamPriorityStrategy', this.onSetStreamPriorityStrategy.bind(this));
+    this.socketEventListeners.set('setConnectionTargetBandwidth', this.onSetConnectionTargetBandwidth.bind(this));
     this.socketEventListeners.forEach((value, key) => {
       this.channel.socketOn(key, value);
     });
@@ -267,6 +270,7 @@ class Client extends events.EventEmitter {
     options.mediaConfiguration = this.token.mediaConfiguration;
     options.singlePC = this.options.singlePC || false;
     options.streamPriorityStrategy = this.options.streamPriorityStrategy;
+    options.connectionTargetBw = this.options.connectionTargetBw;
     log.info('message: addPublisher requested, ',
       `streamId: ${id}, clientId: ${this.id}`,
       logger.objectToLog(options),
@@ -444,6 +448,7 @@ class Client extends events.EventEmitter {
         options.singlePC = this.options.singlePC || false;
         options.unifiedPlan = this.options.unifiedPlan || false;
         options.streamPriorityStrategy = this.options.streamPriorityStrategy;
+        options.connectionTargetBw = this.options.connectionTargetBw;
         stream.addAvSubscriber(this.id);
         this.room.controller.addSubscriber(this.id, options.streamId, options, (signMess) => {
           if (!this.room.streamManager.hasPublishedStream(options.streamId)
@@ -715,10 +720,17 @@ class Client extends events.EventEmitter {
   onSetStreamPriorityStrategy(strategyId, callback = () => {}) {
     this.options.streamPriorityStrategy =
       Client.getStreamPriorityStrategy(strategyId);
+    this.options.connectionTargetBw = this.options.streamPriorityStrategy.connectionTargetBw;
     this.room.amqper.broadcast('ErizoJS', { method: 'setClientStreamPriorityStrategy', args: [this.id, strategyId] });
     callback();
   }
 
+  onSetConnectionTargetBandwidth(connectionTargetBw, callback = () => {}) {
+    this.options.connectionTargetBw = connectionTargetBw;
+    this.room.amqper.broadcast('ErizoJS', { method: 'setClientConnectionTargetBandwidth', args: [this.id, connectionTargetBw] });
+    callback();
+  }
+
   onDisconnect() {
     this.stopListeningToSocketEvents();
     const timeStamp = new Date();

erizo_controller/erizoJS/erizoJSController.js

@@ -473,6 +473,13 @@ exports.ErizoJSController = (erizoJSId, threadPool, ioThreadPool) => {
     }
   };
 
+  that.setClientConnectionTargetBandwidth = (clientId, connectionTargetBw) => {
+    if (clients.has(clientId)) {
+      log.info(`message: updating connectionTargetBandwidth in client ${clientId} to ${connectionTargetBw}`);
+      clients.get(clientId).setConnectionTargetBw(connectionTargetBw);
+    }
+  };
+
   that.getStreamStats = (streamId, callbackRpc) => {
     const stats = {};
     let publisher;

erizo_controller/erizoJS/models/Client.js

@@ -23,6 +23,8 @@ class Client extends EventEmitter {
     this.ioThreadPool = ioThreadPool;
     this.singlePc = singlePc;
     this.streamPriorityStrategy = Client._getStreamPriorityStrategy(streamPriorityStrategy);
+    // The strategy connectionTargetBw is prioritized over connectionTargetBw
+    this.connectionTargetBw = options.connectionTargetBw || 0;
     this.connectionClientId = 0;
     this.options = options;
   }
@@ -61,6 +63,7 @@ class Client extends EventEmitter {
     configuration.isRemote = options.isRemote;
     configuration.encryptTransport = options.encryptTransport;
     configuration.streamPriorityStrategy = this.streamPriorityStrategy;
+    configuration.connectionTargetBw = this.connectionTargetBw;
     const connection = new RtcPeerConnection(configuration);
     connection.on('status_event', (...args) => {
       this.emit('status_event', ...args);
@@ -189,8 +192,16 @@ class Client extends EventEmitter {
     return Array.from(this.connections.values());
   }
 
+  setConnectionTargetBw(connectionTargetBw) {
+    this.connectionTargetBw = connectionTargetBw;
+    this.connections.forEach((connection) => {
+      connection.setConnectionTargetBw(connectionTargetBw);
+    });
+  }
+
   setStreamPriorityStrategy(streamPriorityStrategy) {
     this.streamPriorityStrategy = Client._getStreamPriorityStrategy(streamPriorityStrategy);
+    this.connectionTargetBw = this.streamPriorityStrategy.connectionTargetBw;
     this.connections.forEach((connection) => {
       connection.setStreamPriorityStrategy(this.streamPriorityStrategy);
     });

erizo_controller/erizoJS/models/RTCPeerConnection.js

@@ -626,6 +626,10 @@ class RTCPeerConnection extends EventEmitter {
     this.internalConnection.resetStats();
   }
 
+  setConnectionTargetBw(connectionTargetBw) {
+    this.internalConnection.setConnectionTargetBw(connectionTargetBw);
+  }
+
   setStreamPriorityStrategy(streamPriorityStrategy) {
     this.internalConnection.setStreamPriorityStrategy(streamPriorityStrategy);
   }

erizo_controller/erizoJS/models/WebRtcConnection.js

@@ -41,6 +41,7 @@ class WebRtcConnection extends EventEmitter {
     this.clientId = configuration.clientId;
     this.encryptTransport = configuration.encryptTransport;
     this.streamPriorityStrategy = configuration.streamPriorityStrategy;
+    this.connectionTargetBw = configuration.connectionTargetBw;
     //  {id: stream}
     this.mediaStreams = new Map();
     this.options = configuration.options;
@@ -352,10 +353,16 @@ class WebRtcConnection extends EventEmitter {
     this.wrtc.resetStats();
   }
 
+  setConnectionTargetBw(connectionTargetBw) {
+    this.connectionTargetBw = connectionTargetBw;
+    this.wrtc.setConnectionTargetBw(connectionTargetBw);
+  }
+
   setStreamPriorityStrategy(strategyId) {
     this.streamPriorityStrategy = strategyId;
-    this.wrtc.setBwDistributionConfig(
-      WebRtcConnection._getBwDistributionConfig(this.streamPriorityStrategy));
+    const strategy = WebRtcConnection._getBwDistributionConfig(this.streamPriorityStrategy);
+    this.setConnectionTargetBw(strategy.connectionTargetBw);
+    this.wrtc.setBwDistributionConfig(JSON.stringify(strategy));
   }
 
   copySdpInfoFromConnection(sourceConnection = {}) {
@@ -396,7 +403,7 @@ class WebRtcConnection extends EventEmitter {
       global.config.erizo.maxport,
       this.trickleIce,
       WebRtcConnection._getMediaConfiguration(this.mediaConfiguration, this.willReceivePublishers),
-      WebRtcConnection._getBwDistributionConfig(this.streamPriorityStrategy),
+      JSON.stringify(WebRtcConnection._getBwDistributionConfig(this.streamPriorityStrategy)),
       global.config.erizo.useConnectionQualityCheck,
       this.encryptTransport,
       global.config.erizo.turnserver,
@@ -409,6 +416,7 @@ class WebRtcConnection extends EventEmitter {
       const metadata = this.options.metadata || {};
       wrtc.setMetadata(JSON.stringify(metadata));
     }
+    wrtc.setConnectionTargetBw(this.connectionTargetBw);
     return wrtc;
   }
 
@@ -495,20 +503,25 @@ class WebRtcConnection extends EventEmitter {
        global.bwDistributorConfig.strategyDefinitions[strategyId];
       if (requestedStrategyDefinition.priorities) {
         const serialized = Helpers.serializeStreamPriorityStrategy(requestedStrategyDefinition);
+        const connectionTargetBw =
+          global.bwDistributorConfig.strategyDefinitions[strategyId].connectionTargetBw ?
+            global.bwDistributorConfig.strategyDefinitions[strategyId].connectionTargetBw : 0;
         if (serialized) {
           const result = {
             type: 'StreamPriority',
             strategyId,
             strategy: serialized,
+            connectionTargetBw,
           };
-          return JSON.stringify(result);
+
+          return result;
         }
       }
       log.warn(`message: Bad strategy definition. Using default distributor Config ${global.bwDistributorConfig.defaultType}`);
-      return JSON.stringify({ type: global.bwDistributorConfig.defaultType });
+      return { type: global.bwDistributorConfig.defaultType };
     }
     log.info(`message: No strategy definiton. Using default distributor Config ${global.bwDistributorConfig.defaultType}`);
-    return JSON.stringify({ type: global.bwDistributorConfig.defaultType });
+    return { type: global.bwDistributorConfig.defaultType };
   }
 }
 

erizo_controller/test/utils.js

@@ -202,6 +202,7 @@ module.exports.reset = () => {
     addRemoteCandidate: sinon.stub(),
     addMediaStream: sinon.stub().returns(Promise.resolve(true)),
     removeMediaStream: sinon.stub().returns(Promise.resolve()),
+    setConnectionTargetBw: sinon.stub(),
     getConnectionQualityLevel: sinon.stub().returns(2),
     setMetadata: sinon.stub(),
     linkSendersToSdp: sinon.stub().returns(Promise.resolve()),