From 1833780655023465bd8512678ef80bc45e756aab Mon Sep 17 00:00:00 2001 From: winlin Date: Fri, 26 Feb 2021 16:21:59 +0800 Subject: [PATCH] RTC: Support RTP packet cache manager --- trunk/conf/full.conf | 3 + trunk/src/app/srs_app_config.cpp | 19 ++- trunk/src/app/srs_app_config.hpp | 1 + trunk/src/app/srs_app_rtc_server.cpp | 5 +- trunk/src/kernel/srs_kernel_rtc_rtp.cpp | 172 +++++++++++++++++++++--- trunk/src/kernel/srs_kernel_rtc_rtp.hpp | 49 ++++++- 6 files changed, 229 insertions(+), 20 deletions(-) diff --git a/trunk/conf/full.conf b/trunk/conf/full.conf index 37c3b3cb1..aa9acd709 100644 --- a/trunk/conf/full.conf +++ b/trunk/conf/full.conf @@ -497,6 +497,9 @@ rtc_server { # then system queue is 2000*4 = 8k, user can incrase reuseport to incrase the queue. # default: 2000 queue_length 2000; + # Whether enable the RTP packet cache. + # default: off + rtp_cache off; # The black-hole to copy packet to, for debugging. # For example, when debugging Chrome publish stream, the received packets are encrypted cipher, # we can set the publisher black-hole, SRS will copy the plaintext packets to black-hole, and diff --git a/trunk/src/app/srs_app_config.cpp b/trunk/src/app/srs_app_config.cpp index 0df692464..b605ef8b1 100644 --- a/trunk/src/app/srs_app_config.cpp +++ b/trunk/src/app/srs_app_config.cpp @@ -3651,7 +3651,7 @@ srs_error_t SrsConfig::check_normal_config() string n = conf->at(i)->name; if (n != "enabled" && n != "listen" && n != "dir" && n != "candidate" && n != "ecdsa" && n != "encrypt" && n != "reuseport" && n != "merge_nalus" && n != "perf_stat" && n != "black_hole" - && n != "ip_family") { + && n != "ip_family" && n != "rtp_cache") { return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal rtc_server.%s", n.c_str()); } } @@ -4902,6 +4902,23 @@ bool SrsConfig::get_rtc_server_perf_stat() return SRS_CONF_PERFER_TRUE(conf->arg0()); } +bool SrsConfig::get_rtc_server_rtp_cache() +{ + static bool DEFAULT = false; + + SrsConfDirective* conf = root->get("rtc_server"); + if (!conf) { + return DEFAULT; + } + + conf = conf->get("rtp_cache"); + if (!conf || conf->arg0().empty()) { + return DEFAULT; + } + + return SRS_CONF_PERFER_FALSE(conf->arg0()); +} + bool SrsConfig::get_rtc_server_black_hole() { static bool DEFAULT = false; diff --git a/trunk/src/app/srs_app_config.hpp b/trunk/src/app/srs_app_config.hpp index 08cb4f13b..d3ead7ddb 100644 --- a/trunk/src/app/srs_app_config.hpp +++ b/trunk/src/app/srs_app_config.hpp @@ -534,6 +534,7 @@ public: virtual int get_rtc_server_reuseport(); virtual bool get_rtc_server_merge_nalus(); virtual bool get_rtc_server_perf_stat(); + virtual bool get_rtc_server_rtp_cache(); virtual bool get_rtc_server_black_hole(); virtual std::string get_rtc_server_black_hole_addr(); private: diff --git a/trunk/src/app/srs_app_rtc_server.cpp b/trunk/src/app/srs_app_rtc_server.cpp index 2315e3d10..d4996de4c 100644 --- a/trunk/src/app/srs_app_rtc_server.cpp +++ b/trunk/src/app/srs_app_rtc_server.cpp @@ -283,7 +283,10 @@ srs_error_t SrsRtcServer::initialize() return srs_error_wrap(err, "black hole"); } - srs_trace("RTC server init ok"); + bool rtp_cache = _srs_config->get_rtc_server_rtp_cache(); + _srs_rtp_cache->set_enabled(rtp_cache); + + srs_trace("RTC server init ok, rtp_cache=%d", rtp_cache); return err; } diff --git a/trunk/src/kernel/srs_kernel_rtc_rtp.cpp b/trunk/src/kernel/srs_kernel_rtc_rtp.cpp index 0fa2dc689..352fed02e 100644 --- a/trunk/src/kernel/srs_kernel_rtc_rtp.cpp +++ b/trunk/src/kernel/srs_kernel_rtc_rtp.cpp @@ -162,13 +162,18 @@ int32_t srs_seq_distance(uint16_t value, uint16_t pre_value) SrsRtpExtensionTypes::SrsRtpExtensionTypes() { - memset(ids_, kRtpExtensionNone, sizeof(ids_)); + reset(); } SrsRtpExtensionTypes::~SrsRtpExtensionTypes() { } +void SrsRtpExtensionTypes::reset() +{ + memset(ids_, kRtpExtensionNone, sizeof(ids_)); +} + bool SrsRtpExtensionTypes::register_by_uri(int id, std::string uri) { for (int i = 0; i < (int)(sizeof(kExtensions)/sizeof(kExtensions[0])); ++i) { @@ -201,14 +206,22 @@ SrsRtpExtensionType SrsRtpExtensionTypes::get_type(int id) const -SrsRtpExtensionTwcc::SrsRtpExtensionTwcc(): has_twcc_(false), id_(0), sn_(0) +SrsRtpExtensionTwcc::SrsRtpExtensionTwcc() { + reset(); } SrsRtpExtensionTwcc::~SrsRtpExtensionTwcc() { } +void SrsRtpExtensionTwcc::reset() +{ + has_twcc_ = false; + id_ = 0; + sn_ = 0; +} + bool SrsRtpExtensionTwcc::has_twcc_ext() { return has_twcc_; @@ -286,10 +299,22 @@ void SrsRtpExtensionTwcc::set_sn(uint16_t sn) has_twcc_ = true; } -SrsRtpExtensionOneByte::SrsRtpExtensionOneByte() : has_ext_(false), id_(0), value_(0) +SrsRtpExtensionOneByte::SrsRtpExtensionOneByte() +{ + reset(); +} + +SrsRtpExtensionOneByte::~SrsRtpExtensionOneByte() { } +void SrsRtpExtensionOneByte::reset() +{ + has_ext_ = false; + id_ = 0; + value_ = 0; +} + void SrsRtpExtensionOneByte::set_id(int id) { id_ = id; @@ -338,14 +363,26 @@ srs_error_t SrsRtpExtensionOneByte::encode(SrsBuffer* buf) return err; } -SrsRtpExtensions::SrsRtpExtensions() : has_ext_(false) +SrsRtpExtensions::SrsRtpExtensions() { + reset(); } SrsRtpExtensions::~SrsRtpExtensions() { } +void SrsRtpExtensions::reset() +{ + if (has_ext_) { + types_.reset(); + twcc_.reset(); + audio_level_.reset(); + } + + has_ext_ = false; +} + srs_error_t SrsRtpExtensions::decode(SrsBuffer* buf) { srs_error_t err = srs_success; @@ -524,6 +561,15 @@ srs_error_t SrsRtpExtensions::set_audio_level(int id, uint8_t level) } SrsRtpHeader::SrsRtpHeader() +{ + reset(); +} + +SrsRtpHeader::~SrsRtpHeader() +{ +} + +void SrsRtpHeader::reset() { padding_length = 0; cc = 0; @@ -532,11 +578,10 @@ SrsRtpHeader::SrsRtpHeader() sequence = 0; timestamp = 0; ssrc = 0; + memset(csrc, 0, sizeof(csrc)); ignore_padding_ = false; -} -SrsRtpHeader::~SrsRtpHeader() -{ + extensions_.reset(); } srs_error_t SrsRtpHeader::decode(SrsBuffer* buf) @@ -764,13 +809,10 @@ ISrsRtpPacketDecodeHandler::~ISrsRtpPacketDecodeHandler() SrsRtpPacket2::SrsRtpPacket2() { payload = NULL; - decode_handler = NULL; - - nalu_type = SrsAvcNaluTypeReserved; shared_msg = NULL; cache_buffer_ = NULL; - frame_type = SrsFrameTypeReserved; - cached_payload_size = 0; + + reset(); ++_srs_pps_objs_rtps->sugar; } @@ -782,10 +824,33 @@ SrsRtpPacket2::~SrsRtpPacket2() srs_freep(cache_buffer_); } +void SrsRtpPacket2::reset() +{ + nalu_type = SrsAvcNaluTypeReserved; + frame_type = SrsFrameTypeReserved; + cached_payload_size = 0; + decode_handler = NULL; + + header.reset(); + + // We do not cache the payload, it will be created even + // we cache it, because it only stores pointers to the shared message, + // and it's different for each packet. + srs_freep(payload); + + // We should reset the cached buffer. + if (cache_buffer_) { + cache_buffer_->skip(-1 * cache_buffer_->pos()); + } +} + char* SrsRtpPacket2::wrap(int size) { // If the buffer is large enough, reuse it. if (shared_msg && shared_msg->size >= size) { + // The size maybe changed, so we MUST reset it. + cache_buffer_->set_size(size); + return shared_msg->payload; } @@ -798,6 +863,7 @@ char* SrsRtpPacket2::wrap(int size) char* buf = new char[nb_buffer]; shared_msg->wrap(buf, nb_buffer); + // The size of buffer must equal to the actual size. srs_freep(cache_buffer_); cache_buffer_ = new SrsBuffer(buf, size); @@ -827,6 +893,18 @@ SrsBuffer* SrsRtpPacket2::cache_buffer() const return cache_buffer_; } +bool SrsRtpPacket2::try_recycle() +{ + // When recycling, and there is references about he shared buffer, we must free + // the shared message(may not free the buffer) to stop reuse the shared message. + if (shared_msg && shared_msg->count() > 0) { + srs_freep(shared_msg); + } + + // OK, allow to recycle this object. + return true; +} + void SrsRtpPacket2::set_padding(int size) { header.set_padding(size); @@ -855,14 +933,13 @@ bool SrsRtpPacket2::is_audio() SrsRtpPacket2* SrsRtpPacket2::copy() { - SrsRtpPacket2* cp = new SrsRtpPacket2(); + SrsRtpPacket2* cp = _srs_rtp_cache->allocate(); cp->header = header; cp->payload = payload? payload->copy():NULL; cp->nalu_type = nalu_type; - cp->shared_msg = shared_msg? shared_msg->copy():NULL; - cp->cache_buffer_ = cache_buffer_? cache_buffer_->copy():NULL; + cp->wrap(shared_msg); // Wrap the shared message and buffer. cp->frame_type = frame_type; cp->cached_payload_size = cached_payload_size; @@ -946,6 +1023,71 @@ srs_error_t SrsRtpPacket2::decode(SrsBuffer* buf) return err; } +SrsRtpPacketCacheManager::SrsRtpPacketCacheManager() +{ + enabled_ = false; +} + +SrsRtpPacketCacheManager::~SrsRtpPacketCacheManager() +{ + list::iterator it; + for (it = cache_pkts_.begin(); it != cache_pkts_.end(); ++it) { + SrsRtpPacket2* pkt = *it; + srs_freep(pkt); + } +} + +void SrsRtpPacketCacheManager::set_enabled(bool v) +{ + enabled_ = v; +} + +bool SrsRtpPacketCacheManager::enabled() +{ + return enabled_; +} + +SrsRtpPacket2* SrsRtpPacketCacheManager::allocate() +{ + if (!enabled_ || cache_pkts_.empty()) { + return new SrsRtpPacket2(); + } + + SrsRtpPacket2* pkt = cache_pkts_.back(); + cache_pkts_.pop_back(); + + // We MUST reset it to reuse it. + pkt->reset(); + + return pkt; +} + +void SrsRtpPacketCacheManager::recycle(SrsRtpPacket2* p) +{ + // The p may be NULL, because srs_freep(NULL) is ok. + if (!p) { + return; + } + + // TODO: FIXME: Directly free to keep low memory? + if (!enabled_) { + srs_freep(p); + return; + } + + // If there is any reference about the message, we should free the + // shared message then recycle it(or free it). + if (!p->try_recycle()) { + srs_freep(p); + return; + } + + // Recycle it. + cache_pkts_.push_back(p); +} + +SrsRtpPacketCacheManager* _srs_rtp_cache = new SrsRtpPacketCacheManager(); + SrsRtpRawPayload::SrsRtpRawPayload() { payload = NULL; diff --git a/trunk/src/kernel/srs_kernel_rtc_rtp.hpp b/trunk/src/kernel/srs_kernel_rtc_rtp.hpp index 920e91588..4174f68bb 100644 --- a/trunk/src/kernel/srs_kernel_rtc_rtp.hpp +++ b/trunk/src/kernel/srs_kernel_rtc_rtp.hpp @@ -30,6 +30,7 @@ #include #include +#include class SrsRtpPacket2; @@ -120,6 +121,9 @@ public: public: SrsRtpExtensionTypes(); virtual ~SrsRtpExtensionTypes(); +public: + // Reset the object to reuse it. + void reset(); private: bool register_id(int id, SrsRtpExtensionType type, std::string uri); private: @@ -135,6 +139,9 @@ class SrsRtpExtensionTwcc// : public ISrsCodec public: SrsRtpExtensionTwcc(); virtual ~SrsRtpExtensionTwcc(); +public: + // Reset the object to reuse it. + void reset(); bool has_twcc_ext(); uint8_t get_id(); @@ -157,7 +164,10 @@ class SrsRtpExtensionOneByte// : public ISrsCodec uint8_t value_; public: SrsRtpExtensionOneByte(); - virtual ~SrsRtpExtensionOneByte() {} + virtual ~SrsRtpExtensionOneByte(); +public: + // Reset the object to reuse it. + void reset(); bool exists() { return has_ext_; } int get_id() { return id_; } @@ -182,14 +192,16 @@ private: public: SrsRtpExtensions(); virtual ~SrsRtpExtensions(); - +public: + // Reset the object to reuse it. + void reset(); +public: bool exists(); void set_types_(const SrsRtpExtensionTypes* types); srs_error_t get_twcc_sequence_number(uint16_t& twcc_sn); srs_error_t set_twcc_sequence_number(uint8_t id, uint16_t sn); srs_error_t get_audio_level(uint8_t& level); srs_error_t set_audio_level(int id, uint8_t level); - // ISrsCodec public: virtual srs_error_t decode(SrsBuffer* buf); @@ -217,6 +229,9 @@ private: public: SrsRtpHeader(); virtual ~SrsRtpHeader(); +public: + // Reset the object to reuse it. + void reset(); public: virtual srs_error_t decode(SrsBuffer* buf); private: @@ -262,6 +277,7 @@ public: virtual void on_before_decode_payload(SrsRtpPacket2* pkt, SrsBuffer* buf, ISrsRtpPayloader** ppayload) = 0; }; +// The RTP packet with cached shared message. class SrsRtpPacket2 { // RTP packet fields. @@ -289,6 +305,8 @@ public: SrsRtpPacket2(); virtual ~SrsRtpPacket2(); public: + // Reset the object to reuse it. + void reset(); // Wrap buffer to shared_message, which is managed by us. char* wrap(int size); char* wrap(char* data, int size); @@ -296,6 +314,9 @@ public: char* wrap(SrsSharedPtrMessage* msg); // Get the cache buffer which binds to the shared message. SrsBuffer* cache_buffer() const; + // Try to start recycle, return whether it's reusable. + // @remark If not reusable, user should free it directly. + bool try_recycle(); public: // Set the padding of RTP packet. void set_padding(int size); @@ -316,6 +337,28 @@ public: virtual srs_error_t decode(SrsBuffer* buf); }; +// The RTP packet cache manager. +class SrsRtpPacketCacheManager +{ +private: + bool enabled_; + std::list cache_pkts_; +public: + SrsRtpPacketCacheManager(); + virtual ~SrsRtpPacketCacheManager(); +public: + // Enable or disable cache. + void set_enabled(bool v); + bool enabled(); + // Try to allocate from cache, create new packet if no cache. + SrsRtpPacket2* allocate(); + // Recycle the packet to cache. + // @remark User can directly free the packet. + void recycle(SrsRtpPacket2* p); +}; + +extern SrsRtpPacketCacheManager* _srs_rtp_cache; + // Single payload data. class SrsRtpRawPayload : public ISrsRtpPayloader {