From 4d0863468af1ea580540aabc7f71fa746625b754 Mon Sep 17 00:00:00 2001 From: winlin Date: Fri, 26 Feb 2021 19:46:52 +0800 Subject: [PATCH] RTC: Cache the large buffer allocation --- 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_conn.cpp | 10 +- trunk/src/app/srs_app_rtc_conn.hpp | 2 +- trunk/src/app/srs_app_rtc_server.cpp | 5 +- trunk/src/kernel/srs_kernel_flv.hpp | 2 + trunk/src/kernel/srs_kernel_rtc_rtp.cpp | 145 +++++++----------------- trunk/src/kernel/srs_kernel_rtc_rtp.hpp | 81 +++++++++---- 9 files changed, 135 insertions(+), 133 deletions(-) diff --git a/trunk/conf/full.conf b/trunk/conf/full.conf index b850003e5..19bd6f7e1 100644 --- a/trunk/conf/full.conf +++ b/trunk/conf/full.conf @@ -500,6 +500,9 @@ rtc_server { # Whether enable the RTP packet cache. # default: off rtp_cache off; + #Whether enable the RTP message(a large buffer) cache. + # default: off + rtp_msg_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 52fadae34..49ff10fea 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 != "rtp_cache") { + && n != "ip_family" && n != "rtp_cache" && n != "rtp_msg_cache") { return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal rtc_server.%s", n.c_str()); } } @@ -4919,6 +4919,23 @@ bool SrsConfig::get_rtc_server_rtp_cache() return SRS_CONF_PERFER_FALSE(conf->arg0()); } +bool SrsConfig::get_rtc_server_rtp_msg_cache() +{ + static bool DEFAULT = false; + + SrsConfDirective* conf = root->get("rtc_server"); + if (!conf) { + return DEFAULT; + } + + conf = conf->get("rtp_msg_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 a6cebc7c3..5ba3e36f9 100644 --- a/trunk/src/app/srs_app_config.hpp +++ b/trunk/src/app/srs_app_config.hpp @@ -535,6 +535,7 @@ public: 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_rtp_msg_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_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp index 290fbe40f..2ac360d5d 100644 --- a/trunk/src/app/srs_app_rtc_conn.cpp +++ b/trunk/src/app/srs_app_rtc_conn.cpp @@ -1193,11 +1193,11 @@ srs_error_t SrsRtcPublishStream::on_rtp_plaintext(char* plaintext, int nb_plaint SrsRtpPacket2* pkt = _srs_rtp_cache->allocate(); // Copy the packet body. - pkt->wrap(plaintext, nb_plaintext); - srs_assert(pkt->cache_buffer()->pos() == 0); + char* p = pkt->wrap(plaintext, nb_plaintext); // Handle the packet. - err = do_on_rtp_plaintext(pkt); + SrsBuffer buf(p, nb_plaintext); + err = do_on_rtp_plaintext(pkt, &buf); // Release the packet to cache. _srs_rtp_cache->recycle(pkt); @@ -1205,14 +1205,14 @@ srs_error_t SrsRtcPublishStream::on_rtp_plaintext(char* plaintext, int nb_plaint return err; } -srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket2* pkt) +srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket2* pkt, SrsBuffer* buf) { srs_error_t err = srs_success; pkt->set_decode_handler(this); pkt->set_extension_types(&extension_types_); - if ((err = pkt->decode(pkt->cache_buffer())) != srs_success) { + if ((err = pkt->decode(buf)) != srs_success) { return srs_error_wrap(err, "decode rtp packet"); } diff --git a/trunk/src/app/srs_app_rtc_conn.hpp b/trunk/src/app/srs_app_rtc_conn.hpp index 3483862cb..644c718f9 100644 --- a/trunk/src/app/srs_app_rtc_conn.hpp +++ b/trunk/src/app/srs_app_rtc_conn.hpp @@ -357,7 +357,7 @@ private: // @remark We copy the plaintext, user should free it. srs_error_t on_rtp_plaintext(char* plaintext, int nb_plaintext); private: - srs_error_t do_on_rtp_plaintext(SrsRtpPacket2* pkt); + srs_error_t do_on_rtp_plaintext(SrsRtpPacket2* pkt, SrsBuffer* buf); public: srs_error_t check_send_nacks(); public: diff --git a/trunk/src/app/srs_app_rtc_server.cpp b/trunk/src/app/srs_app_rtc_server.cpp index 40bc4dae9..36b4b53ee 100644 --- a/trunk/src/app/srs_app_rtc_server.cpp +++ b/trunk/src/app/srs_app_rtc_server.cpp @@ -287,7 +287,10 @@ srs_error_t SrsRtcServer::initialize() 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); + bool rtp_msg_cache = _srs_config->get_rtc_server_rtp_msg_cache(); + _srs_rtp_msg_cache->set_enabled(rtp_msg_cache); + + srs_trace("RTC server init ok, rc=%d, rmc=%d", rtp_cache, rtp_msg_cache); return err; } diff --git a/trunk/src/kernel/srs_kernel_flv.hpp b/trunk/src/kernel/srs_kernel_flv.hpp index f3be5931b..ba53e92db 100644 --- a/trunk/src/kernel/srs_kernel_flv.hpp +++ b/trunk/src/kernel/srs_kernel_flv.hpp @@ -312,6 +312,8 @@ public: SrsSharedPtrMessage(); virtual ~SrsSharedPtrMessage(); public: + // For object cache to reset and reuse it. + bool reset() { return true; } // Create shared ptr message, // copy header, manage the payload of msg, // set the payload to NULL to prevent double free. diff --git a/trunk/src/kernel/srs_kernel_rtc_rtp.cpp b/trunk/src/kernel/srs_kernel_rtc_rtp.cpp index cbbe7741a..60c076a9d 100644 --- a/trunk/src/kernel/srs_kernel_rtc_rtp.cpp +++ b/trunk/src/kernel/srs_kernel_rtc_rtp.cpp @@ -811,7 +811,6 @@ SrsRtpPacket2::SrsRtpPacket2() { payload = NULL; shared_msg = NULL; - cache_buffer_ = NULL; reset(); @@ -821,11 +820,17 @@ SrsRtpPacket2::SrsRtpPacket2() SrsRtpPacket2::~SrsRtpPacket2() { srs_freep(payload); - srs_freep(shared_msg); - srs_freep(cache_buffer_); + + // Recyle the real owner of message, no other reference object. + if (shared_msg && shared_msg->count() == 0) { + _srs_rtp_msg_cache->recycle(shared_msg); + shared_msg = NULL; + } else { + srs_freep(shared_msg); + } } -void SrsRtpPacket2::reset() +bool SrsRtpPacket2::reset() { nalu_type = SrsAvcNaluTypeReserved; frame_type = SrsFrameTypeReserved; @@ -839,38 +844,49 @@ void SrsRtpPacket2::reset() // 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()); + // Recyle the real owner of message, no other reference object. + if (shared_msg && shared_msg->count() == 0) { + _srs_rtp_msg_cache->recycle(shared_msg); + shared_msg = NULL; + } else { + srs_freep(shared_msg); } + + return true; } 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; } - // Create buffer if empty or not large enough. - srs_freep(shared_msg); - shared_msg = new SrsSharedPtrMessage(); + // Create a large enough message, with under-layer buffer. + while (true) { + srs_freep(shared_msg); + shared_msg = _srs_rtp_msg_cache->allocate(); - // For RTC, we use larger under-layer buffer for each packet. - int nb_buffer = srs_max(size, kRtpPacketSize); - char* buf = new char[nb_buffer]; - shared_msg->wrap(buf, nb_buffer); + // If got a cached message(which has payload), but it's too small, + // we free it and allocate a larger one. + if (shared_msg->payload && shared_msg->size < size) { + continue; + } - // The size of buffer must equal to the actual size. - srs_freep(cache_buffer_); - cache_buffer_ = new SrsBuffer(buf, size); + // Create under-layer buffer for new message + if (!shared_msg->payload) { + // For RTC, we use larger under-layer buffer for each packet. + int nb_buffer = srs_max(size, kRtpPacketSize); + char* buf = new char[nb_buffer]; + shared_msg->wrap(buf, nb_buffer); - ++_srs_pps_objs_rbuf->sugar; + ++_srs_pps_objs_rbuf->sugar; + } - return buf; + break; + } + + return shared_msg->payload; } char* SrsRtpPacket2::wrap(char* data, int size) @@ -885,29 +901,9 @@ char* SrsRtpPacket2::wrap(SrsSharedPtrMessage* msg) srs_freep(shared_msg); shared_msg = msg->copy(); - srs_freep(cache_buffer_); - cache_buffer_ = new SrsBuffer(msg->payload, msg->size); - return msg->payload; } -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); @@ -1026,70 +1022,9 @@ 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); -} +SrsRtpObjectCacheManager* _srs_rtp_cache = new SrsRtpObjectCacheManager(); -SrsRtpPacketCacheManager* _srs_rtp_cache = new SrsRtpPacketCacheManager(); +SrsRtpObjectCacheManager* _srs_rtp_msg_cache = new SrsRtpObjectCacheManager(); SrsRtpRawPayload::SrsRtpRawPayload() { diff --git a/trunk/src/kernel/srs_kernel_rtc_rtp.hpp b/trunk/src/kernel/srs_kernel_rtc_rtp.hpp index 4174f68bb..d8995a39a 100644 --- a/trunk/src/kernel/srs_kernel_rtc_rtp.hpp +++ b/trunk/src/kernel/srs_kernel_rtc_rtp.hpp @@ -285,8 +285,6 @@ public: SrsRtpHeader header; ISrsRtpPayloader* payload; private: - // The buffer bind to the shared message. - SrsBuffer* cache_buffer_; // The original shared message, all RTP packets can refer to its data. SrsSharedPtrMessage* shared_msg; // Helper fields. @@ -306,17 +304,12 @@ public: virtual ~SrsRtpPacket2(); public: // Reset the object to reuse it. - void reset(); + virtual bool reset(); // Wrap buffer to shared_message, which is managed by us. char* wrap(int size); char* wrap(char* data, int size); // Wrap the shared message, we copy it. 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); @@ -337,27 +330,75 @@ public: virtual srs_error_t decode(SrsBuffer* buf); }; -// The RTP packet cache manager. -class SrsRtpPacketCacheManager +// The RTP packet or message cache manager. +template +class SrsRtpObjectCacheManager { private: bool enabled_; - std::list cache_pkts_; + std::list cache_objs_; public: - SrsRtpPacketCacheManager(); - virtual ~SrsRtpPacketCacheManager(); + SrsRtpObjectCacheManager() { + enabled_ = false; + } + virtual ~SrsRtpObjectCacheManager() { + typedef typename std::list::iterator iterator; + for (iterator it = cache_objs_.begin(); it != cache_objs_.end(); ++it) { + T* obj = *it; + srs_freep(obj); + } + } 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. + void set_enabled(bool v) { + enabled_ = v; + } + bool enabled() { + return enabled_; + } + // Try to allocate from cache, create new object if no cache. + T* allocate() { + while (true) { + if (!enabled_ || cache_objs_.empty()) { + return new T(); + } + + T* obj = cache_objs_.back(); + cache_objs_.pop_back(); + + // If reset the object fail, drop the cached object. + if (!obj->reset()) { + srs_freep(obj); + continue; + } + + return obj; + } + } + // Recycle the object to cache. // @remark User can directly free the packet. - void recycle(SrsRtpPacket2* p); + void recycle(T* 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; + } + + // Recycle it. + cache_objs_.push_back(p); + } }; -extern SrsRtpPacketCacheManager* _srs_rtp_cache; +// For RTP packets cache. +extern SrsRtpObjectCacheManager* _srs_rtp_cache; + +// For RTP packet shared messages cache. +extern SrsRtpObjectCacheManager* _srs_rtp_msg_cache; // Single payload data. class SrsRtpRawPayload : public ISrsRtpPayloader