diff --git a/CHANGELOG.md b/CHANGELOG.md index 15c7680ea2..d95e7db752 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -62,7 +62,6 @@ * Downgrade OpenSSL to version 1.1.0h (27 Mar 2018) until issue [#257](https://github.com/versatica/mediasoup/issues/257) is fixed. - ### 2.6.2 * C++: Remove all `Destroy()` class methods and no longer do `delete this`. diff --git a/lib/Consumer.js b/lib/Consumer.js index a04ff60a8e..c4b6796183 100644 --- a/lib/Consumer.js +++ b/lib/Consumer.js @@ -29,6 +29,7 @@ class Consumer extends EnhancedEventEmitter // - .kind // - .peer (just if this Consumer belongs to a Peer) // - .transport + // - .transportCallback // - .rtpParameters // - .source this._data = data; @@ -222,6 +223,14 @@ class Consumer extends EnhancedEventEmitter // Remove notification subscriptions. this._channel.removeAllListeners(this._internal.consumerId); + // Remove previously registered listener for transport's @close so that closed Consumer object can be picked up by GC + // while a Transport object it used to reffer to still stays alive + if (this._data.transport && this._data.transportCallback) + this._data.transport.removeListener('@close', this._data.transportCallback); + + this._data.transportCallback = undefined; + this._data.transport = undefined; + if (notifyChannel) { this._channel.request('consumer.close', this._internal) @@ -338,13 +347,13 @@ class Consumer extends EnhancedEventEmitter this.safeEmit('handled'); - transport.on('@close', () => + this._data.transportCallback = () => { this._internal.transportId = undefined; this._data.transport = undefined; this.safeEmit('unhandled'); - }); + }; }) .catch((error) => { @@ -709,7 +718,6 @@ class Consumer extends EnhancedEventEmitter if (typeof interval !== 'number') { - logger.error('enableStats() | invalid interval provided'); return; } diff --git a/lib/Peer.js b/lib/Peer.js index b356b47bac..5b9d14a0dc 100644 --- a/lib/Peer.js +++ b/lib/Peer.js @@ -753,10 +753,18 @@ class Peer extends EnhancedEventEmitter transportId : undefined }; - const consumerRtpParameters = + let consumerRtpParameters = ortc.getConsumerRtpParameters( producer.consumableRtpParameters, this.rtpCapabilities); + consumerRtpParameters.codecs.forEach((codec) => { + if (codec.rtcpFeedback !== undefined && Array.isArray(codec.rtcpFeedback)) + { + let rtcpFeedback = codec.rtcpFeedback.filter((val) => val.type !== 'ffmpeg-proxy'); + codec.rtcpFeedback = rtcpFeedback; + } + }); + const data = { kind : producer.kind, @@ -790,7 +798,8 @@ class Peer extends EnhancedEventEmitter { logger.error('"router.createConsumer" request failed: %s', String(error)); - throw error; + // remove throw as it's causing uncaught promise rejections + // throw error; }); this.safeEmit('newconsumer', consumer); diff --git a/lib/Producer.js b/lib/Producer.js index 4d1113e517..8e41f9bd41 100644 --- a/lib/Producer.js +++ b/lib/Producer.js @@ -28,6 +28,7 @@ class Producer extends EnhancedEventEmitter // - .kind // - .peer // - .transport + // -.transportCallback // - .rtpParameters // - .consumableRtpParameters this._data = data; @@ -184,6 +185,13 @@ class Producer extends EnhancedEventEmitter this.emit('@notify', 'producerClosed', data); } + // Remove previously registered listener for transport's @close so that closed Producer object can be picked up by GC + // while a Transport object it used to reffer to still stays alive + if (this.transport && !this.transport.closed) + this.transport.removeListener('@close', this._data.transportCallback); + this._data.transportCallback = undefined; + this._data.transport = undefined; + this.emit('@close'); this.safeEmit('close', 'local', appData); @@ -214,6 +222,11 @@ class Producer extends EnhancedEventEmitter clearInterval(this._statsInterval); } + if (this.transport && !this.transport.closed) + this.transport.removeListener('@close', this._data.transportCallback); + this._data.transportCallback = undefined; + this._data.transport = undefined; + this.emit('@close'); this.safeEmit('close', 'remote', appData); @@ -589,11 +602,12 @@ class Producer extends EnhancedEventEmitter { // On closure, the worker Transport closes all its Producers and the client // side gets producer.on('unhandled'). - this.transport.on('@close', () => + this._data.transportCallback = () => { if (!this._closed) this.close(undefined, false); - }); + }; + this.transport.on('@close', this._data.transportCallback); } _handleWorkerNotifications() diff --git a/lib/Room.js b/lib/Room.js index df5f307f21..18b8429fc3 100644 --- a/lib/Room.js +++ b/lib/Room.js @@ -12,7 +12,7 @@ const logger = new Logger('Room'); class Room extends EnhancedEventEmitter { - constructor(internal, data, channel) + constructor(internal, data, channel, getCpuPercentage) { super(logger); @@ -22,9 +22,12 @@ class Room extends EnhancedEventEmitter this._closed = false; // Internal data. + // - .workerId // - .routerId this._internal = internal; + this._getCpuPercentage = getCpuPercentage; + // Room data. // - .rtpCapabilities this._data = data; @@ -48,6 +51,16 @@ class Room extends EnhancedEventEmitter return this._internal.routerId; } + get workerId() + { + return this._internal.workerId; + } + + get cpuPercentage() + { + return this._getCpuPercentage(); + } + get closed() { return this._closed; @@ -227,7 +240,7 @@ class Room extends EnhancedEventEmitter createRtpStreamer(producer, options) { - logger.debug('createRtpStreamer()'); + logger.debug('"createRtpStreamer()" sendOldNack is %s producer.kind is %s', options.sendOldNack, producer.kind); if (!this._producers.has(producer.id)) return Promise.reject(new Error('Producer not found')); @@ -257,10 +270,22 @@ class Room extends EnhancedEventEmitter }; // TODO: The app should provide his own RTP capabilities. - const consumerRtpParameters = + let consumerRtpParameters = ortc.getConsumerRtpParameters( producer.consumableRtpParameters, this.rtpCapabilities); + // Remove ffmpeg-proxy from capabilities if sendOldNack is not present + if (options.sendOldNack !== true) + { + consumerRtpParameters.codecs.forEach((codec) => { + if (codec.rtcpFeedback !== undefined && Array.isArray(codec.rtcpFeedback)) + { + let rtcpFeedback = codec.rtcpFeedback.filter((val) => val.type !== 'ffmpeg-proxy'); + codec.rtcpFeedback = rtcpFeedback; + } + }); + } + const data = { kind : producer.kind, diff --git a/lib/Server.js b/lib/Server.js index 08cb3c2ef5..30fe5d8721 100644 --- a/lib/Server.js +++ b/lib/Server.js @@ -56,6 +56,9 @@ class Server extends EnhancedEventEmitter // Remove numWorkers. delete options.numWorkers; + const workerOptions = options.worker; + delete options.worker; + // Normalize some options. if (!check.array(options.logTags)) @@ -142,7 +145,7 @@ class Server extends EnhancedEventEmitter // Create a Worker instance (do it in a separate method to avoid creating // a callback function within a loop). - this._addWorker(new Worker(workerId, workerParameters)); + this._addWorker(new Worker(workerId, workerParameters, workerOptions)); } } @@ -161,6 +164,11 @@ class Server extends EnhancedEventEmitter return this._workers.size; } + get workers() + { + return Array.from(this._workers); + } + /** * Close the Server. */ @@ -257,22 +265,49 @@ class Server extends EnhancedEventEmitter // This may throw. const rtpCapabilities = ortc.generateRoomRtpCapabilities(mediaCodecs); - if (typeof workerIdx === 'number') - { - if (workerIdx < 0 || workerIdx >= this._workers.size) - throw new Error('requested worker idx out of range'); - // Update latest worker idx. - this._latestWorkerIdx = workerIdx; + const array = Array.from(this._workers); + + let lowestCpu; + let lowestCpuIndex; + const zeroCpus = []; + for (let i = array.length - 1; i >= 0; i--) { + const worker = array[i]; + const wcpu = worker.cpuPercentage; + if (wcpu) { + if (!lowestCpu || wcpu < lowestCpu) { + lowestCpu = wcpu; + lowestCpuIndex = i; + } + } else { + zeroCpus.push(worker); + } } - else - { - // Increase latest worker idx. - this._latestWorkerIdx++; - // Move to 0 if out of range. - if (this._latestWorkerIdx === this._workers.size) - this._latestWorkerIdx = 0; + if (zeroCpus.length) { + if (zeroCpus.length === array.length) { + if (typeof workerIdx === 'number') + { + if (workerIdx < 0 || workerIdx >= this._workers.size) + throw new Error('requested worker idx out of range'); + + // Update latest worker idx. + this._latestWorkerIdx = workerIdx; + } + else + { + // Increase latest worker idx. + this._latestWorkerIdx++; + + // Move to 0 if out of range. + if (this._latestWorkerIdx === this._workers.size) + this._latestWorkerIdx = 0; + } + } else { + this._latestWorkerIdx = array.indexOf(zeroCpus[Math.floor(Math.random() * Math.floor(zeroCpus.length))]); + } + } else { + this._latestWorkerIdx = lowestCpuIndex } const worker = Array.from(this._workers)[this._latestWorkerIdx]; diff --git a/lib/Worker.js b/lib/Worker.js index 64c436c56d..993bab9d4d 100644 --- a/lib/Worker.js +++ b/lib/Worker.js @@ -1,5 +1,6 @@ const path = require('path'); const spawn = require('child_process').spawn; +const fs = require('fs'); const process = require('process'); const EventEmitter = require('events').EventEmitter; const Logger = require('./Logger'); @@ -11,6 +12,18 @@ const CHANNEL_FD = 3; const logger = new Logger('Worker'); +const getCpuUsage = function(pid, cb) +{ + fs.readFile(`/proc/${ pid }/stat`, function(err, data) + { + const elems = data.toString().split(' '); + const utime = parseInt(elems[13]); + const stime = parseInt(elems[14]); + + cb(utime + stime); + }); +}; + let workerPath; // If env MEDIASOUP_WORKER_PATH is given, use it. @@ -22,13 +35,13 @@ if (process.env.MEDIASOUP_WORKER_PATH) else if (process.env.MEDIASOUP_BUILDTYPE === 'Debug') { workerPath = - path.join(__dirname, '..', 'worker', 'out', 'Debug', 'mediasoup-worker'); + path.join(__dirname, '..', 'worker', 'out', 'Debug', 'mediasoup-worker'); } // Otherwise use 'Release'. else { workerPath = - path.join(__dirname, '..', 'worker', 'out', 'Release', 'mediasoup-worker'); + path.join(__dirname, '..', 'worker', 'out', 'Release', 'mediasoup-worker'); } // Set environtment variable for the worker. @@ -36,7 +49,7 @@ process.env.MEDIASOUP_CHANNEL_FD = String(CHANNEL_FD); class Worker extends EventEmitter { - constructor(id, parameters) + constructor(id, parameters, options = {}) { super(); this.setMaxListeners(Infinity); @@ -45,18 +58,18 @@ class Worker extends EventEmitter const spawnArgs = [ id ].concat(parameters); const spawnOptions = - { - detached : false, - - /* - * fd 0 (stdin) : Just ignore it. - * fd 1 (stdout) : Pipe it for 3rd libraries in the worker. - * that log their own stuff. - * fd 2 (stderr) : Same as stdout. - * fd 3 (channel) : Channel fd. - */ - stdio : [ 'ignore', 'pipe', 'pipe', 'pipe' ] - }; + { + detached : false, + + /* + * fd 0 (stdin) : Just ignore it. + * fd 1 (stdout) : Pipe it for 3rd libraries in the worker. + * that log their own stuff. + * fd 2 (stderr) : Same as stdout. + * fd 3 (channel) : Channel fd. + */ + stdio : [ 'ignore', 'pipe', 'pipe', 'pipe' ] + }; // Closed flag. this._closed = false; @@ -100,6 +113,7 @@ class Worker extends EventEmitter logger.error( 'child process exited [id:%s, code:%s, signal:%s]', id, code, signal); + clearInterval(this._cpuInterval); this._child = null; this.close(); }); @@ -108,9 +122,36 @@ class Worker extends EventEmitter { logger.error('child process error [id:%s, error:%s]', id, error); + clearInterval(this._cpuInterval); this._child = null; this.close(); }); + + this._cpuPercentage = null; + if (options.monitorCpu) { + this._cpuInterval = setInterval(() => { + if (this._child) { + getCpuUsage(this._child.pid, (startTime) => { + setTimeout(() => { + if (this._child) { + getCpuUsage(this._child.pid, (endTime) => { + const delta = endTime - startTime; + const percentage = 100 * (delta / 10000); + this._cpuPercentage = percentage; + }); + } + }, 1000); + }); + } + }, 1000); + } + + this._id = id; + } + + get cpuPercentage() + { + return this._cpuPercentage; } close() @@ -127,8 +168,9 @@ class Worker extends EventEmitter // Kill mediasoup-worker process. if (this._child) { - // Remove event listeners but leave a fake 'error' hander - // to avoid propagation. + // Remove event listeners but leave a fake 'error' hander + // to avoid propagation. + clearInterval(this._cpuInterval); this._child.removeAllListeners('exit'); this._child.removeAllListeners('error'); this._child.on('error', () => null); @@ -183,20 +225,23 @@ class Worker extends EventEmitter } /** - * Create a Room instance. - * - * @return {Room} - */ + * Create a Room instance. + * + * @return {Room} + */ Room(data) { logger.debug('Room()'); const internal = { - routerId : utils.randomNumber() + workerId : this._id, + routerId : utils.randomNumber(), }; - const room = new Room(internal, data, this._channel); + const room = new Room(internal, data, this._channel, () => { + return this._cpuPercentage; + }); // Store the Room instance and remove it when closed. this._rooms.add(room); diff --git a/lib/supportedRtpCapabilities.js b/lib/supportedRtpCapabilities.js index f8844013f6..e58dbd3107 100644 --- a/lib/supportedRtpCapabilities.js +++ b/lib/supportedRtpCapabilities.js @@ -154,7 +154,9 @@ const supportedRtpCapabilities = { type: 'nack' }, { type: 'nack', parameter: 'pli' }, { type: 'ccm', parameter: 'fir' }, - { type: 'goog-remb' } + { type: 'goog-remb' }, + { type: 'ffmpeg-proxy' }, + { type: 'ffmpeg-proxy', parameter: 'yes' } ] }, { @@ -167,8 +169,10 @@ const supportedRtpCapabilities = { type: 'nack' }, { type: 'nack', parameter: 'pli' }, { type: 'ccm', parameter: 'fir' }, - { type: 'goog-remb' } - ] + { type: 'goog-remb' }, + { type: 'ffmpeg-proxy' }, + { type: 'ffmpeg-proxy', parameter: 'yes' } + ] }, { kind : 'video', @@ -184,7 +188,9 @@ const supportedRtpCapabilities = { type: 'nack' }, { type: 'nack', parameter: 'pli' }, { type: 'ccm', parameter: 'fir' }, - { type: 'goog-remb' } + { type: 'goog-remb' }, + { type: 'ffmpeg-proxy' }, + { type: 'ffmpeg-proxy', parameter: 'yes' } ] }, { @@ -201,7 +207,9 @@ const supportedRtpCapabilities = { type: 'nack' }, { type: 'nack', parameter: 'pli' }, { type: 'ccm', parameter: 'fir' }, - { type: 'goog-remb' } + { type: 'goog-remb' }, + { type: 'ffmpeg-proxy' }, + { type: 'ffmpeg-proxy', parameter: 'yes' } ] }, { @@ -214,7 +222,9 @@ const supportedRtpCapabilities = { type: 'nack' }, { type: 'nack', parameter: 'pli' }, { type: 'ccm', parameter: 'fir' }, - { type: 'goog-remb' } + { type: 'goog-remb' }, + { type: 'ffmpeg-proxy' }, + { type: 'ffmpeg-proxy', parameter: 'yes' } ] } ], diff --git a/package.json b/package.json index 57f4c2fa6d..bf2a24c66f 100644 --- a/package.json +++ b/package.json @@ -1,5 +1,5 @@ { - "name": "mediasoup", + "name": "@livelyvideo/mediasoup", "version": "2.6.14", "description": "Cutting Edge WebRTC Video Conferencing", "author": "Iñaki Baz Castillo (https://inakibaz.me)", @@ -8,9 +8,14 @@ ], "homepage": "https://mediasoup.org", "license": "ISC", + "publishConfig": { + "always-auth": true, + "registry": "https://npm.livelyvideo.tv" + }, "repository": { "type": "git", - "url": "https://github.com/versatica/mediasoup.git" + "private": true, + "url": "https://github.com/LivelyVideo/mediasoup.git" }, "keywords": [ "webrtc", diff --git a/worker/common.gypi b/worker/common.gypi index 15f55c3cc5..90fe74927b 100644 --- a/worker/common.gypi +++ b/worker/common.gypi @@ -21,7 +21,7 @@ { 'Release': { - 'cflags': [ '-g' ] + 'cflags': [ '-g', '-O3' ] }, 'Debug': { diff --git a/worker/include/RTC/NackGenerator.hpp b/worker/include/RTC/NackGenerator.hpp index 37bbbc28cf..31f70ee15c 100644 --- a/worker/include/RTC/NackGenerator.hpp +++ b/worker/include/RTC/NackGenerator.hpp @@ -13,6 +13,14 @@ namespace RTC { class NackGenerator : public Timer::Listener { + public: + enum class NACKedPacket + { + NOT_FOUND = 0, + FOUND = 1, + TOO_OLD = 2 + }; + public: class Listener { @@ -43,7 +51,7 @@ namespace RTC explicit NackGenerator(Listener* listener); ~NackGenerator() override; - bool ReceivePacket(RTC::RtpPacket* packet); + NACKedPacket ReceivePacket(RTC::RtpPacket* packet); size_t GetNackListLength() const; void Reset(); diff --git a/worker/include/RTC/Producer.hpp b/worker/include/RTC/Producer.hpp index ccbdeaa900..0eb013ca3d 100644 --- a/worker/include/RTC/Producer.hpp +++ b/worker/include/RTC/Producer.hpp @@ -32,10 +32,11 @@ namespace RTC private: struct HeaderExtensionIds { - uint8_t ssrcAudioLevel{ 0 }; // 0 means no ssrc-audio-level id. - uint8_t absSendTime{ 0 }; // 0 means no abs-send-time id. - uint8_t mid{ 0 }; // 0 means no MID id. - uint8_t rid{ 0 }; // 0 means no RID id. + uint8_t ssrcAudioLevel{ 0 }; // 0 means no ssrc-audio-level id. + uint8_t absSendTime{ 0 }; // 0 means no abs-send-time id. + uint8_t mid{ 0 }; // 0 means no MID id. + uint8_t rid{ 0 }; // 0 means no RID id. + uint8_t videoOrientation{ 0 }; // 0 means no video-orientation id. }; private: diff --git a/worker/include/RTC/RtpPacket.hpp b/worker/include/RTC/RtpPacket.hpp index ba5153de95..380b686845 100644 --- a/worker/include/RTC/RtpPacket.hpp +++ b/worker/include/RTC/RtpPacket.hpp @@ -112,6 +112,7 @@ namespace RTC uint8_t* GetExtension(RTC::RtpHeaderExtensionUri::Type uri, uint8_t* len) const; bool ReadAudioLevel(uint8_t* volume, bool* voice) const; bool ReadAbsSendTime(uint32_t* time) const; + bool ReadVideoOrientation(uint16_t* data) const; bool ReadMid(const uint8_t** data, size_t* len) const; bool ReadRid(const uint8_t** data, size_t* len) const; uint8_t* GetPayload() const; @@ -276,18 +277,18 @@ namespace RTC { *len = 0; - if (this->extensionMap.find(uri) == this->extensionMap.end()) + if (this->extensionMap.find(uri) == this->extensionMap.end()) { return nullptr; + } uint8_t id = this->extensionMap.at(uri); - - if (HasOneByteExtensions()) + if (HasOneByteExtensions()) { - if (this->oneByteExtensions.find(id) == this->oneByteExtensions.end()) + if (this->oneByteExtensions.find(id) == this->oneByteExtensions.end()){ return nullptr; + } *len = this->oneByteExtensions.at(id)->len + 1; - return this->oneByteExtensions.at(id)->value; } else if (HasTwoBytesExtensions()) @@ -337,6 +338,54 @@ namespace RTC return true; } + inline bool RtpPacket::ReadVideoOrientation(uint16_t* rotation) const + { + uint8_t extenLen; + uint8_t cvoByte; + uint8_t* extenValue; + + extenValue = GetExtension(RTC::RtpHeaderExtensionUri::Type::VIDEO_ORIENTATION, &extenLen); + if (!extenValue || extenLen != 1) { + return false; + } + /** + Coordination of Video Orientation + Bit# 7 6 5 4 3 2 1 0(LSB) + Definition 0 0 0 0 C F R1 R0 + + C = Camera + 0: Front-facing camera, facing the user. If camera direction is unknown, this is the default value used. + 1: Back-facing camera, facing away from the user. + + F = Flip + 0: No flip operation. If the sending client flips value is unknown, this is the default value used. + 1: Horizontal flip operation + + R1 R0 = Rotation + 0 0 = 0 rotation + 0 1 = 90 CCW (aka: 270 CW) + 1 0 = 180 CCW (aka: 180 CW) + 1 1 = 270 CCW (aka: 90 CW) + **/ + cvoByte = Utils::Byte::Get1Byte(extenValue, 0); + int rValue = (cvoByte & 0x03); + switch (rValue) { + case 3: // NOTE: using counter clockwise values + *rotation = 270; + break; + case 2: + *rotation = 180; + break; + case 1: + *rotation = 90; + break; + default: + *rotation = 0; + } + + return true; + } + inline bool RtpPacket::ReadMid(const uint8_t** data, size_t* len) const { uint8_t extenLen; diff --git a/worker/include/RTC/RtpStream.hpp b/worker/include/RTC/RtpStream.hpp index 8f6936a677..b8f873c783 100644 --- a/worker/include/RTC/RtpStream.hpp +++ b/worker/include/RTC/RtpStream.hpp @@ -23,6 +23,7 @@ namespace RTC uint32_t clockRate{ 0 }; bool useNack{ false }; bool usePli{ false }; + bool sendOldNack{ false }; }; public: @@ -43,6 +44,7 @@ namespace RTC const std::string& GetId() const; void RestartStatusCheckTimer(); void StopStatusCheckTimer(); + bool SendOldNack() const {return params.sendOldNack;} protected: bool UpdateSeq(RTC::RtpPacket* packet); @@ -68,6 +70,8 @@ namespace RTC size_t pliCount{ 0 }; size_t nackCount{ 0 }; size_t sliCount{ 0 }; + uint16_t videoOrientation{ 0 }; + bool cvoReceived{ false }; RtpDataCounter transmissionCounter; RtpDataCounter retransmissionCounter; diff --git a/worker/include/RTC/RtpStreamSend.hpp b/worker/include/RTC/RtpStreamSend.hpp index e8e59fcb21..ce3599c477 100644 --- a/worker/include/RTC/RtpStreamSend.hpp +++ b/worker/include/RTC/RtpStreamSend.hpp @@ -11,18 +11,17 @@ namespace RTC { class RtpStreamSend : public RtpStream { - private: + public: struct StorageItem { - uint8_t store[RTC::MtuSize]; - }; - - private: - struct BufferItem - { - uint16_t seq{ 0 }; // RTP seq. - uint64_t resentAtTime{ 0 }; + // Cloned packet. RTC::RtpPacket* packet{ nullptr }; + // Memory to hold the cloned packet (with extra space for RTX encoding). + uint8_t store[RTC::MtuSize + 100]; + // Last time this packet was resent. + uint64_t resentAtTime{ 0 }; + // Number of times this packet was resent. + uint8_t sentTimes{ 0 }; }; public: @@ -43,16 +42,21 @@ namespace RTC private: void StorePacket(RTC::RtpPacket* packet); + void ResetStorageItem(StorageItem* storageItem); + void UpdateBufferStartIdx(); /* Pure virtual methods inherited from RtpStream. */ protected: void CheckStatus() override; private: - // Passed by argument. + std::vector buffer; + uint16_t bufferStartIdx{ 0 }; + size_t bufferSize{ 0 }; std::vector storage; - using Buffer = std::list; - Buffer buffer; + + RTC::RtpDataCounter transmissionCounter; + // Stats. float rtt{ 0 }; diff --git a/worker/include/RTC/Transport.hpp b/worker/include/RTC/Transport.hpp index 9b0b3f898b..ad6772bd25 100644 --- a/worker/include/RTC/Transport.hpp +++ b/worker/include/RTC/Transport.hpp @@ -44,9 +44,10 @@ namespace RTC // maps them to the corresponding ids in the room). struct HeaderExtensionIds { - uint8_t absSendTime{ 0 }; // 0 means no abs-send-time id. - uint8_t mid{ 0 }; // 0 means no MID id. - uint8_t rid{ 0 }; // 0 means no RID id. + uint8_t absSendTime{ 0 }; // 0 means no abs-send-time id. + uint8_t mid{ 0 }; // 0 means no MID id. + uint8_t rid{ 0 }; // 0 means no RID id. + uint8_t videoOrientation{ 0 }; // 0 means no video-orientation id. }; public: diff --git a/worker/mediasoup-worker.gyp b/worker/mediasoup-worker.gyp index 11915e4a4d..86e5edfe3c 100644 --- a/worker/mediasoup-worker.gyp +++ b/worker/mediasoup-worker.gyp @@ -213,6 +213,8 @@ }], [ 'OS == "linux"', { + 'cflags': [ '-O3' ], + 'ldflags': [ '-O3' ], 'defines': [ '_POSIX_C_SOURCE=200112', @@ -221,8 +223,8 @@ }], [ 'OS == "linux" and mediasoup_asan == "true"', { - 'cflags': [ '-fsanitize=address' ], - 'ldflags': [ '-fsanitize=address' ] + 'cflags': [ '-fsanitize=address', '-O3' ], + 'ldflags': [ '-fsanitize=address', '-O3' ] }], [ 'OS in "linux freebsd"', { diff --git a/worker/src/RTC/Consumer.cpp b/worker/src/RTC/Consumer.cpp index e54229e787..ec24f64bd7 100644 --- a/worker/src/RTC/Consumer.cpp +++ b/worker/src/RTC/Consumer.cpp @@ -768,6 +768,7 @@ namespace RTC auto& codec = this->rtpParameters.GetCodecForEncoding(encoding); bool useNack{ false }; bool usePli{ false }; + bool sendOldNack{ false }; for (auto& fb : codec.rtcpFeedback) { @@ -783,6 +784,12 @@ namespace RTC usePli = true; } + if (!sendOldNack && fb.type == "ffmpeg-proxy" && fb.parameter == "yes") + { + MS_DEBUG_TAG(rtcp, "Ffmpeg proxy uses sendOldNack"); + + sendOldNack = true; + } } // Create stream params. @@ -794,12 +801,11 @@ namespace RTC params.clockRate = codec.clockRate; params.useNack = useNack; params.usePli = usePli; + params.sendOldNack = sendOldNack; // Create a RtpStreamSend for sending a single media stream. - if (useNack) - this->rtpStream = new RTC::RtpStreamSend(params, 1500); - else - this->rtpStream = new RTC::RtpStreamSend(params, 0); + size_t bufferSize = params.useNack ? 600 : 0; + this->rtpStream = new RTC::RtpStreamSend(params, bufferSize); if (encoding.hasRtx && encoding.rtx.ssrc != 0u) { diff --git a/worker/src/RTC/NackGenerator.cpp b/worker/src/RTC/NackGenerator.cpp index f9a19cc052..ee039ccf92 100644 --- a/worker/src/RTC/NackGenerator.cpp +++ b/worker/src/RTC/NackGenerator.cpp @@ -34,8 +34,8 @@ namespace RTC delete this->timer; } - // Returns true if this is a found nacked packet. False otherwise. - bool NackGenerator::ReceivePacket(RTC::RtpPacket* packet) + // Returns 0 if this is a found nacked packet. 1 if not found, 2 if pkt is already NACK'ed or out of order. + NackGenerator::NACKedPacket NackGenerator::ReceivePacket(RTC::RtpPacket* packet) { MS_TRACE(); @@ -50,12 +50,12 @@ namespace RTC if (isKeyFrame) this->keyFrameList.insert(seq); - return false; + return NackGenerator::NACKedPacket::NOT_FOUND; } // Obviously never nacked, so ignore. if (seq == this->lastSeq) - return false; + return NackGenerator::NACKedPacket::NOT_FOUND; // May be an out of order packet, or already handled retransmitted packet, // or a retransmitted packet. @@ -74,7 +74,7 @@ namespace RTC this->nackList.erase(it); - return true; + return NackGenerator::NACKedPacket::FOUND; } // Out of order packet or already handled NACKed packet. @@ -84,7 +84,7 @@ namespace RTC packet->GetSsrc(), packet->GetSequenceNumber()); - return false; + return NackGenerator::NACKedPacket::TOO_OLD; // this makes nice with the ffmpeg proxy } // If we are here it means that we may have lost some packets so seq @@ -106,14 +106,14 @@ namespace RTC { this->lastSeq++; - return false; + return NackGenerator::NACKedPacket::NOT_FOUND; } AddPacketsToNackList(this->lastSeq + 1, seq); // NackGenerator instance may have been reseted. if (!this->started) - return false; + return NackGenerator::NACKedPacket::NOT_FOUND; this->lastSeq = seq; @@ -125,7 +125,7 @@ namespace RTC MayRunTimer(); - return false; + return NackGenerator::NACKedPacket::NOT_FOUND; } void NackGenerator::CleanOldNackItems(uint16_t seq) diff --git a/worker/src/RTC/PlainRtpTransport.cpp b/worker/src/RTC/PlainRtpTransport.cpp index fee6e408da..7dc9e00266 100644 --- a/worker/src/RTC/PlainRtpTransport.cpp +++ b/worker/src/RTC/PlainRtpTransport.cpp @@ -283,6 +283,11 @@ namespace RTC packet->AddExtensionMapping( RtpHeaderExtensionUri::Type::ABS_SEND_TIME, this->headerExtensionIds.absSendTime); } + if (this->headerExtensionIds.videoOrientation != 0u) + { + packet->AddExtensionMapping( + RtpHeaderExtensionUri::Type::VIDEO_ORIENTATION, this->headerExtensionIds.videoOrientation); + } if (this->headerExtensionIds.mid != 0u) { packet->AddExtensionMapping(RtpHeaderExtensionUri::Type::MID, this->headerExtensionIds.mid); diff --git a/worker/src/RTC/Producer.cpp b/worker/src/RTC/Producer.cpp index 6c9dc7e9b2..e1c0e72de5 100644 --- a/worker/src/RTC/Producer.cpp +++ b/worker/src/RTC/Producer.cpp @@ -75,6 +75,7 @@ namespace RTC static const Json::StaticString JsonStringRtpStreamInfos{ "rtpStreamInfos" }; static const Json::StaticString JsonStringRtpStream{ "rtpStream" }; static const Json::StaticString JsonStringRid{ "rid" }; + static const Json::StaticString JsonStringVideoOrientation{ "videoOrientation" }; static const Json::StaticString JsonStringProfile{ "profile" }; static const Json::StaticString JsonStringNone{ "none" }; static const Json::StaticString JsonStringDefault{ "default" }; @@ -154,6 +155,9 @@ namespace RTC if (this->headerExtensionIds.rid != 0u) jsonHeaderExtensionIds[JsonStringRid] = this->headerExtensionIds.rid; + if (this->headerExtensionIds.videoOrientation != 0u) + jsonHeaderExtensionIds[JsonStringVideoOrientation] = this->headerExtensionIds.videoOrientation; + json[JsonStringHeaderExtensionIds] = jsonHeaderExtensionIds; json[JsonStringPaused] = this->paused; @@ -382,6 +386,7 @@ namespace RTC uint8_t absSendTimeId{ 0 }; uint8_t midId{ 0 }; uint8_t ridId{ 0 }; + uint8_t videoOrientationId{ 0 }; for (auto& exten : this->rtpParameters.headerExtensions) { @@ -408,6 +413,17 @@ namespace RTC this->transportHeaderExtensionIds.absSendTime = exten.id; } + if ((videoOrientationId == 0u) && exten.type == RTC::RtpHeaderExtensionUri::Type::VIDEO_ORIENTATION) + { + if (idMapping.find(exten.id) != idMapping.end()) + videoOrientationId = idMapping[exten.id]; + else + videoOrientationId = exten.id; + + this->headerExtensionIds.videoOrientation = videoOrientationId; + this->transportHeaderExtensionIds.videoOrientation = exten.id; + } + if ((midId == 0u) && exten.type == RTC::RtpHeaderExtensionUri::Type::MID) { if (idMapping.find(exten.id) != idMapping.end()) @@ -515,6 +531,7 @@ namespace RTC bool useNack{ false }; bool usePli{ false }; bool useRemb{ false }; + bool sendOldNack{ false }; for (auto& fb : codec.rtcpFeedback) { @@ -536,6 +553,12 @@ namespace RTC useRemb = true; } + if (!sendOldNack && fb.type == "ffmpeg-proxy" && fb.parameter == "yes") + { + MS_DEBUG_2TAGS(rtcp, rtx, "Ffmpeg proxy uses sendOldNack"); + + sendOldNack = true; + } } // Create stream params. @@ -547,6 +570,14 @@ namespace RTC params.clockRate = codec.clockRate; params.useNack = useNack; params.usePli = usePli; + params.sendOldNack = sendOldNack; + + if (params.sendOldNack) { + MS_DEBUG_2TAGS(rtcp, rtx, "Producer of %s creating new RtpStreamRecv: useNack=%s, sendOldNack=%s", + (this->kind == RTC::Media::Kind::VIDEO)? "video" : "audio", + params.useNack ? "true" : "false", + params.sendOldNack ? "true" : "false" ); + } // Create a RtpStreamRecv for receiving a media stream. auto* rtpStream = new RTC::RtpStreamRecv(this, params); @@ -620,6 +651,12 @@ namespace RTC packet->AddExtensionMapping( RtpHeaderExtensionUri::Type::RTP_STREAM_ID, this->headerExtensionIds.rid); } + + if (this->headerExtensionIds.videoOrientation != 0u) + { + packet->AddExtensionMapping( + RtpHeaderExtensionUri::Type::VIDEO_ORIENTATION, this->headerExtensionIds.videoOrientation); + } } void Producer::ActivateStream(RTC::RtpStreamRecv* rtpStream) diff --git a/worker/src/RTC/RtpPacket.cpp b/worker/src/RTC/RtpPacket.cpp index 543c35c9b6..d9ed587a52 100644 --- a/worker/src/RTC/RtpPacket.cpp +++ b/worker/src/RTC/RtpPacket.cpp @@ -534,7 +534,6 @@ namespace RTC { uint8_t id = (*ptr & 0xF0) >> 4; size_t len = static_cast(*ptr & 0x0F) + 1; - if (ptr + 1 + len > extensionEnd) { MS_WARN_TAG( diff --git a/worker/src/RTC/RtpStream.cpp b/worker/src/RTC/RtpStream.cpp index 74a904a01e..21c873b695 100644 --- a/worker/src/RTC/RtpStream.cpp +++ b/worker/src/RTC/RtpStream.cpp @@ -63,6 +63,8 @@ namespace RTC static const Json::StaticString JsonStringBitRate{ "bitrate" }; static const Json::StaticString JsonStringPacketsLost{ "packetsLost" }; static const Json::StaticString JsonStringFractionLost{ "fractionLost" }; + static const Json::StaticString JsonStringVideoOrientation{ "videoOrientation" }; + static const Json::StaticString JsonStringCvoReceived{ "cvoReceived" }; static const Json::StaticString JsonStringPacketsDiscarded{ "packetsDiscarded" }; static const Json::StaticString JsonStringPacketsRepaired{ "packetsRepaired" }; static const Json::StaticString JsonStringFirCount{ "firCount" }; @@ -73,17 +75,19 @@ namespace RTC Json::Value json(Json::objectValue); uint64_t now = DepLibUV::GetTime(); - json[JsonStringId] = this->rtpStreamId; - json[JsonStringTimestamp] = Json::UInt64{ now }; - json[JsonStringSsrc] = Json::UInt{ this->params.ssrc }; - json[JsonStringMediaType] = RtpCodecMimeType::type2String[this->params.mimeType.type]; - json[JsonStringKind] = RtpCodecMimeType::type2String[this->params.mimeType.type]; - json[JsonStringMimeType] = this->params.mimeType.ToString(); - json[JsonStringPacketCount] = static_cast(this->transmissionCounter.GetPacketCount()); - json[JsonStringByteCount] = static_cast(this->transmissionCounter.GetBytes()); - json[JsonStringBitRate] = Json::UInt{ this->transmissionCounter.GetRate(now) }; + json[JsonStringId] = this->rtpStreamId; + json[JsonStringTimestamp] = Json::UInt64{ now }; + json[JsonStringSsrc] = Json::UInt{ this->params.ssrc }; + json[JsonStringMediaType] = RtpCodecMimeType::type2String[this->params.mimeType.type]; + json[JsonStringKind] = RtpCodecMimeType::type2String[this->params.mimeType.type]; + json[JsonStringMimeType] = this->params.mimeType.ToString(); + json[JsonStringPacketCount] = static_cast(this->transmissionCounter.GetPacketCount()); + json[JsonStringByteCount] = static_cast(this->transmissionCounter.GetBytes()); + json[JsonStringBitRate] = Json::UInt{ this->transmissionCounter.GetRate(now) }; json[JsonStringPacketsLost] = Json::UInt{ this->packetsLost }; json[JsonStringFractionLost] = Json::UInt{ this->fractionLost }; + json[JsonStringVideoOrientation] = Json::UInt{ this->videoOrientation }; + json[JsonStringCvoReceived] = this->cvoReceived; json[JsonStringPacketsDiscarded] = static_cast(this->packetsDiscarded); json[JsonStringPacketsRepaired] = static_cast(this->packetsRepaired); json[JsonStringFirCount] = static_cast(this->firCount); @@ -112,6 +116,10 @@ namespace RTC this->maxPacketMs = DepLibUV::GetTime(); } + if (packet->ReadVideoOrientation(&this->videoOrientation)) { + this->cvoReceived = true; + } + // If not a valid packet ignore it. if (!UpdateSeq(packet)) { @@ -244,12 +252,12 @@ namespace RTC Json::Value json(Json::objectValue); - json[JsonStringSsrc] = Json::UInt{ this->ssrc }; - json[JsonStringPayloadType] = Json::UInt{ this->payloadType }; - json[JsonStringMimeType] = this->mimeType.ToString(); - json[JsonStringClockRate] = Json::UInt{ this->clockRate }; - json[JsonStringUseNack] = this->useNack; - json[JsonStringUsePli] = this->usePli; + json[JsonStringSsrc] = Json::UInt{ this->ssrc }; + json[JsonStringPayloadType] = Json::UInt{ this->payloadType }; + json[JsonStringMimeType] = this->mimeType.ToString(); + json[JsonStringClockRate] = Json::UInt{ this->clockRate }; + json[JsonStringUseNack] = this->useNack; + json[JsonStringUsePli] = this->usePli; return json; } diff --git a/worker/src/RTC/RtpStreamRecv.cpp b/worker/src/RTC/RtpStreamRecv.cpp index 0e4c71f722..be822ddc4a 100644 --- a/worker/src/RTC/RtpStreamRecv.cpp +++ b/worker/src/RTC/RtpStreamRecv.cpp @@ -134,8 +134,20 @@ namespace RTC // Pass the packet to the NackGenerator and return true just if this was a // NACKed packet. - if (this->params.useNack) - return this->nackGenerator->ReceivePacket(packet); + if (this->params.useNack) { + RTC::NackGenerator::NACKedPacket ret = this->nackGenerator->ReceivePacket(packet); + switch(ret) { + case RTC::NackGenerator::NACKedPacket::NOT_FOUND: + return false; + case RTC::NackGenerator::NACKedPacket::FOUND: + return true; + case RTC::NackGenerator::NACKedPacket::TOO_OLD: + MS_DEBUG_TAG( + rtx, + "RtpStreamRecv::ReceiveRtxPacket this->params.sendOldNack=%s", (this->params.sendOldNack ? "true" : "false")); + return this->params.sendOldNack; + } + } return false; } diff --git a/worker/src/RTC/RtpStreamSend.cpp b/worker/src/RTC/RtpStreamSend.cpp index 42462a9109..baf77f1840 100644 --- a/worker/src/RTC/RtpStreamSend.cpp +++ b/worker/src/RTC/RtpStreamSend.cpp @@ -10,7 +10,9 @@ namespace RTC { /* Static. */ - + // 17: 16 bit mask + the initial sequence number. + static constexpr size_t MaxRequestedPackets{ 17 }; + static std::vector RetransmissionContainer(MaxRequestedPackets + 1); // Don't retransmit packets older than this (ms). static constexpr uint32_t MaxRetransmissionDelay{ 2000 }; static constexpr uint32_t DefaultRtt{ 100 }; @@ -18,7 +20,8 @@ namespace RTC /* Instance methods. */ RtpStreamSend::RtpStreamSend(RTC::RtpStream::Params& params, size_t bufferSize) - : RtpStream::RtpStream(params), storage(bufferSize) + : RtpStream::RtpStream(params), buffer(bufferSize > 0 ? 65536 : 0, nullptr), + storage(bufferSize) { MS_TRACE(); } @@ -57,6 +60,9 @@ namespace RTC if (!this->storage.empty()) StorePacket(packet); + // Increase transmission counter. + this->transmissionCounter.Update(packet); + return true; } @@ -100,47 +106,13 @@ namespace RTC { MS_TRACE(); - // 17: 16 bit mask + the initial sequence number. - static constexpr size_t MaxRequestedPackets{ 17 }; - // Ensure the container's first element is 0. container[0] = nullptr; // If NACK is not supported, exit. if (!this->params.useNack) { - MS_WARN_TAG(rtx, "NACK not negotiated"); - - return; - } - - // If the buffer is empty just return. - if (this->buffer.empty()) - return; - - uint16_t firstSeq = seq; - uint16_t lastSeq = firstSeq + MaxRequestedPackets - 1; - - // Number of requested packets cannot be greater than the container size - 1. - MS_ASSERT(container.size() - 1 >= MaxRequestedPackets, "RtpPacket container is too small"); - - auto bufferIt = this->buffer.begin(); - auto bufferItReverse = this->buffer.rbegin(); - uint16_t bufferFirstSeq = (*bufferIt).seq; - uint16_t bufferLastSeq = (*bufferItReverse).seq; - - // Requested packet range not found. - if ( - SeqManager::IsSeqHigherThan(firstSeq, bufferLastSeq) || - SeqManager::IsSeqLowerThan(lastSeq, bufferFirstSeq)) - { - MS_WARN_TAG( - rtx, - "requested packet range not in the buffer [seq:%" PRIu16 ", bufferFirstseq:%" PRIu16 - ", bufferLastseq:%" PRIu16 "]", - seq, - bufferFirstSeq, - bufferLastSeq); + MS_WARN_TAG(rtx, "NACK not supported"); return; } @@ -165,73 +137,72 @@ namespace RTC if (requested) { - for (; bufferIt != this->buffer.end(); ++bufferIt) + auto* storageItem = this->buffer[seq]; + RTC::RtpPacket* packet{ nullptr }; + uint32_t diffMs; + + // Calculate how the elapsed time between the max timestampt seen and + // the requested packet's timestampt (in ms). + if (storageItem) { - auto currentSeq = (*bufferIt).seq; + packet = storageItem->packet; + + uint32_t diffTs = this->maxPacketTs - packet->GetTimestamp(); + diffMs = diffTs * 1000 / this->params.clockRate; + } - // Found. - if (currentSeq == seq) + if (!storageItem) + { + // Do nothing. + } + // Don't resend the packet if older than MaxRetransmissionDelay ms. + else if (diffMs > MaxRetransmissionDelay) + { + if (!tooOldPacketFound) { - auto currentPacket = (*bufferIt).packet; - // Calculate how the elapsed time between the max timestampt seen and - // the requested packet's timestampt (in ms). - uint32_t diffTs = this->maxPacketTs - currentPacket->GetTimestamp(); - uint32_t diffMs = diffTs * 1000 / this->params.clockRate; - - // Just provide the packet if no older than MaxRetransmissionDelay ms. - if (diffMs > MaxRetransmissionDelay) - { - if (!tooOldPacketFound) - { - // TODO: May we ask for a key frame in this case? - - MS_WARN_TAG( - rtx, - "ignoring retransmission for too old packet " - "[seq:%" PRIu16 ", max age:%" PRIu32 "ms, packet age:%" PRIu32 "ms]", - currentPacket->GetSequenceNumber(), - MaxRetransmissionDelay, - diffMs); - - tooOldPacketFound = true; - } - - break; - } - - // Don't resent the packet if it was resent in the last RTT ms. - auto resentAtTime = (*bufferIt).resentAtTime; - - if ((resentAtTime != 0u) && now - resentAtTime <= static_cast(rtt)) - { - MS_DEBUG_TAG( - rtx, - "ignoring retransmission for a packet already resent in the last RTT ms " - "[seq:%" PRIu16 ", rtt:%" PRIu32 "]", - currentPacket->GetSequenceNumber(), - rtt); - - break; - } - - // Store the packet in the container and then increment its index. - container[containerIdx++] = currentPacket; - - // Save when this packet was resent. - (*bufferIt).resentAtTime = now; - - sent = true; - if (isFirstPacket) - firstPacketSent = true; - - break; + MS_WARN_TAG( + rtx, + "ignoring retransmission for too old packet " + "[seq:%" PRIu16 ", max age:%" PRIu32 "ms, packet age:%" PRIu32 "ms]", + packet->GetSequenceNumber(), + MaxRetransmissionDelay, + diffMs); + + tooOldPacketFound = true; } + } + // Don't resent the packet if it was resent in the last RTT ms. + // clang-format off + else if ( + storageItem->resentAtTime != 0u && + now - storageItem->resentAtTime <= static_cast(rtt) + ) + // clang-format on + { + MS_DEBUG_TAG( + rtx, + "ignoring retransmission for a packet already resent in the last RTT ms " + "[seq:%" PRIu16 ", rtt:%" PRIu32 "]", + packet->GetSequenceNumber(), + rtt); + } + // Store the packet in the container and then increment its index. + else { + container[containerIdx++] = packet; + + // Save when this packet was resent. + storageItem->resentAtTime = now; + + // Increase the number of times this packet was sent. + storageItem->sentTimes++; + + sent = true; - // It can not be after this packet. - if (SeqManager::IsSeqHigherThan(currentSeq, seq)) - break; + if (isFirstPacket) + firstPacketSent = true; } } + requested = (bitmask & 1) != 0; bitmask >>= 1; @@ -297,14 +268,65 @@ namespace RTC { MS_TRACE(); - // Delete cloned packets. - for (auto& bufferItem : this->buffer) + if (this->storage.empty()) + return; + + for (uint32_t idx{ 0 }; idx < this->buffer.size(); ++idx) { - delete bufferItem.packet; + auto* storageItem = this->buffer[idx]; + + if (!storageItem) + { + // TODO + MS_ASSERT(this->buffer[idx] == nullptr, "key should be NULL"); + + continue; + } + + // Reset (free RTP packet) the storage item. + ResetStorageItem(storageItem); + + // Unfill the buffer item. + this->buffer[idx] = nullptr; } - // Clear list. - this->buffer.clear(); + // Reset buffer. + this->bufferStartIdx = 0; + this->bufferSize = 0; + } + + inline void RtpStreamSend::ResetStorageItem(StorageItem* storageItem) + { + MS_TRACE(); + + MS_ASSERT(storageItem, "storageItem cannot be nullptr"); + + delete storageItem->packet; + + storageItem->packet = nullptr; + storageItem->resentAtTime = 0; + storageItem->sentTimes = 0; + } + + /** + * Iterates the buffer starting from the current start idx + 1 until next + * used one. It takes into account that the buffer is circular. + */ + inline void RtpStreamSend::UpdateBufferStartIdx() + { + uint16_t seq = this->bufferStartIdx + 1; + + for (uint32_t idx{ 0 }; idx < this->buffer.size(); ++idx, ++seq) + { + auto* storageItem = this->buffer[seq]; + + if (storageItem) + { + this->bufferStartIdx = seq; + + break; + } + } } inline void RtpStreamSend::StorePacket(RTC::RtpPacket* packet) @@ -323,75 +345,68 @@ namespace RTC return; } - // Sum the packet seq number and the number of 16 bits cycles. - auto packetSeq = packet->GetSequenceNumber(); - BufferItem bufferItem; - - bufferItem.seq = packetSeq; + auto seq = packet->GetSequenceNumber(); + auto* storageItem = this->buffer[seq]; - // If empty do it easy. - if (this->buffer.empty()) + // Buffer is empty. + if (this->bufferSize == 0) { - auto store = this->storage[0].store; + // Take the first storage position. + storageItem = std::addressof(this->storage[0]); + this->buffer[seq] = storageItem; - bufferItem.packet = packet->Clone(store); - this->buffer.push_back(bufferItem); - - return; + // Increase buffer size and set start index. + this->bufferSize++; + this->bufferStartIdx = seq; } - - // Otherwise, do the stuff. - - Buffer::iterator newBufferIt; - uint8_t* store{ nullptr }; - - // Iterate the buffer in reverse order and find the proper place to store the - // packet. - auto bufferItReverse = this->buffer.rbegin(); - for (; bufferItReverse != this->buffer.rend(); ++bufferItReverse) + // The buffer item is already used. Check whether we should replace its + // storage with the new packet or just ignore it (if duplicated packet). + else if (storageItem) { - auto currentSeq = (*bufferItReverse).seq; - - if (SeqManager::IsSeqHigherThan(packetSeq, currentSeq)) - { - // Get a forward iterator pointing to the same element. - auto it = bufferItReverse.base(); - - newBufferIt = this->buffer.insert(it, bufferItem); - - break; - } + auto* storedPacket = storageItem->packet; - // Packet is already stored. - if (packetSeq == currentSeq) + if (packet->GetTimestamp() == storedPacket->GetTimestamp()) return; - } - // The packet was older than anything in the buffer, store it at the beginning. - if (bufferItReverse == this->buffer.rend()) - newBufferIt = this->buffer.insert(this->buffer.begin(), bufferItem); + // Reset the storage item. + ResetStorageItem(storageItem); - // If the buffer is not full use the next free storage item. - if (this->buffer.size() - 1 < this->storage.size()) + // If this was the item referenced by the buffer start index, move it to + // the next one. + if (this->bufferStartIdx == seq) + UpdateBufferStartIdx(); + } + // Buffer not yet full, add an entry. + else if (this->bufferSize < this->storage.size()) { - store = this->storage[this->buffer.size() - 1].store; + // Take the next storage position. + storageItem = std::addressof(this->storage[this->bufferSize]); + this->buffer[seq] = storageItem; + + // Increase buffer size. + this->bufferSize++; } - // Otherwise remove the first packet of the buffer and replace its storage area. + // Buffer full, remove oldest entry and add new one. else { - auto& firstBufferItem = *(this->buffer.begin()); - auto firstPacket = firstBufferItem.packet; - - // Store points to the store used by the first packet. - store = const_cast(firstPacket->GetData()); - // Free the first packet. - delete firstPacket; - // Remove the first element in the list. - this->buffer.pop_front(); + auto* firstStorageItem = this->buffer[this->bufferStartIdx]; + + // Reset the first storage item. + ResetStorageItem(firstStorageItem); + + // Unfill the buffer start item. + this->buffer[this->bufferStartIdx] = nullptr; + + // Move the buffer start index. + UpdateBufferStartIdx(); + + // Take the freed storage item. + storageItem = firstStorageItem; + this->buffer[seq] = storageItem; } - // Update the new buffer item so it points to the cloned packed. - (*newBufferIt).packet = packet->Clone(store); + // Clone the packet into the retrieved storage item. + storageItem->packet = packet->Clone(storageItem->store); } void RtpStreamSend::SetRtx(uint8_t payloadType, uint32_t ssrc) diff --git a/worker/src/RTC/Transport.cpp b/worker/src/RTC/Transport.cpp index b7fc5cd129..edd284287a 100644 --- a/worker/src/RTC/Transport.cpp +++ b/worker/src/RTC/Transport.cpp @@ -167,6 +167,9 @@ namespace RTC if (producer->GetTransportHeaderExtensionIds().rid != 0u) this->headerExtensionIds.rid = producer->GetTransportHeaderExtensionIds().rid; + + if (producer->GetTransportHeaderExtensionIds().videoOrientation != 0u) + this->headerExtensionIds.videoOrientation = producer->GetTransportHeaderExtensionIds().videoOrientation; } void Transport::HandleConsumer(RTC::Consumer* consumer) diff --git a/worker/src/RTC/WebRtcTransport.cpp b/worker/src/RTC/WebRtcTransport.cpp index d7bcbf3a2a..9fa1202d33 100644 --- a/worker/src/RTC/WebRtcTransport.cpp +++ b/worker/src/RTC/WebRtcTransport.cpp @@ -247,6 +247,7 @@ namespace RTC static const Json::StaticString JsonStringFailed{ "failed" }; static const Json::StaticString JsonStringHeaderExtensionIds{ "headerExtensionIds" }; static const Json::StaticString JsonStringAbsSendTime{ "absSendTime" }; + static const Json::StaticString JsonStringVideoOrientation{ "videoOrientation" }; static const Json::StaticString JsonStringMid{ "mid" }; static const Json::StaticString JsonStringRid{ "rid" }; static const Json::StaticString JsonStringRtpListener{ "rtpListener" }; @@ -338,6 +339,9 @@ namespace RTC if (this->headerExtensionIds.absSendTime != 0u) jsonHeaderExtensionIds[JsonStringAbsSendTime] = this->headerExtensionIds.absSendTime; + if (this->headerExtensionIds.videoOrientation != 0u) + jsonHeaderExtensionIds[JsonStringVideoOrientation] = this->headerExtensionIds.videoOrientation; + if (this->headerExtensionIds.mid != 0u) jsonHeaderExtensionIds[JsonStringMid] = this->headerExtensionIds.mid; @@ -857,6 +861,11 @@ namespace RTC packet->AddExtensionMapping( RtpHeaderExtensionUri::Type::ABS_SEND_TIME, this->headerExtensionIds.absSendTime); } + if (this->headerExtensionIds.videoOrientation != 0u) + { + packet->AddExtensionMapping( + RtpHeaderExtensionUri::Type::VIDEO_ORIENTATION, this->headerExtensionIds.videoOrientation); + } if (this->headerExtensionIds.mid != 0u) { packet->AddExtensionMapping(RtpHeaderExtensionUri::Type::MID, this->headerExtensionIds.mid); diff --git a/worker/test/src/RTC/TestRtpStreamSend.cpp b/worker/test/src/RTC/TestRtpStreamSend.cpp index 580e97c075..66b906622c 100644 --- a/worker/test/src/RTC/TestRtpStreamSend.cpp +++ b/worker/test/src/RTC/TestRtpStreamSend.cpp @@ -79,13 +79,16 @@ SCENARIO("NACK and RTP packets retransmission", "[rtp][rtcp]") params.useNack = true; // Create a RtpStreamSend. - RtpStreamSend* stream = new RtpStreamSend(params, 200); + RtpStreamSend* stream = new RtpStreamSend(params, 4); - // Receive all the packets in order into the stream. + // Receive all the packets (some of them not in order and/or duplicated). stream->ReceivePacket(packet1); + stream->ReceivePacket(packet3); stream->ReceivePacket(packet2); stream->ReceivePacket(packet3); stream->ReceivePacket(packet4); + stream->ReceivePacket(packet4); + stream->ReceivePacket(packet5); stream->ReceivePacket(packet5); // Create a NACK item that request for all the packets. @@ -101,30 +104,22 @@ SCENARIO("NACK and RTP packets retransmission", "[rtp][rtcp]") auto rtxPacket2 = rtpRetransmissionContainer[1]; auto rtxPacket3 = rtpRetransmissionContainer[2]; auto rtxPacket4 = rtpRetransmissionContainer[3]; - auto rtxPacket5 = rtpRetransmissionContainer[4]; - auto rtxPacket6 = rtpRetransmissionContainer[5]; REQUIRE(rtxPacket1); - REQUIRE(rtxPacket1->GetSequenceNumber() == packet1->GetSequenceNumber()); - REQUIRE(rtxPacket1->GetTimestamp() == packet1->GetTimestamp()); + REQUIRE(rtxPacket1->GetSequenceNumber() == packet2->GetSequenceNumber()); + REQUIRE(rtxPacket1->GetTimestamp() == packet2->GetTimestamp()); REQUIRE(rtxPacket2); - REQUIRE(rtxPacket2->GetSequenceNumber() == packet2->GetSequenceNumber()); - REQUIRE(rtxPacket2->GetTimestamp() == packet2->GetTimestamp()); + REQUIRE(rtxPacket2->GetSequenceNumber() == packet3->GetSequenceNumber()); + REQUIRE(rtxPacket2->GetTimestamp() == packet3->GetTimestamp()); REQUIRE(rtxPacket3); - REQUIRE(rtxPacket3->GetSequenceNumber() == packet3->GetSequenceNumber()); - REQUIRE(rtxPacket3->GetTimestamp() == packet3->GetTimestamp()); + REQUIRE(rtxPacket3->GetSequenceNumber() == packet4->GetSequenceNumber()); + REQUIRE(rtxPacket3->GetTimestamp() == packet4->GetTimestamp()); REQUIRE(rtxPacket4); - REQUIRE(rtxPacket4->GetSequenceNumber() == packet4->GetSequenceNumber()); - REQUIRE(rtxPacket4->GetTimestamp() == packet4->GetTimestamp()); - - REQUIRE(rtxPacket5); - REQUIRE(rtxPacket5->GetSequenceNumber() == packet5->GetSequenceNumber()); - REQUIRE(rtxPacket5->GetTimestamp() == packet5->GetTimestamp()); - - REQUIRE(rtxPacket6 == nullptr); + REQUIRE(rtxPacket4->GetSequenceNumber() == packet5->GetSequenceNumber()); + REQUIRE(rtxPacket4->GetTimestamp() == packet5->GetTimestamp()); // Clean stuff. delete packet1;