RTC: Support RTP packet cache manager

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

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;

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:

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;
 }

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,8 +299,20 @@ 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)
@@ -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<SrsRtpPacket2*>::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;

trunk/src/kernel/srs_kernel_rtc_rtp.hpp

@@ -30,6 +30,7 @@
 #include <srs_kernel_codec.hpp>
 
 #include <string>
+#include <list>
 
 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<SrsRtpPacket2*> 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
 {