RTC: Cache the large buffer allocation

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

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;

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:

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

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:

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

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.

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<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);
-}
+SrsRtpObjectCacheManager<SrsRtpPacket2>* _srs_rtp_cache = new SrsRtpObjectCacheManager<SrsRtpPacket2>();
 
-SrsRtpPacketCacheManager* _srs_rtp_cache = new SrsRtpPacketCacheManager();
+SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache = new SrsRtpObjectCacheManager<SrsSharedPtrMessage>();
 
 SrsRtpRawPayload::SrsRtpRawPayload()
 {

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<typename T>
+class SrsRtpObjectCacheManager
 {
 private:
     bool enabled_;
-    std::list<SrsRtpPacket2*> cache_pkts_;
+    std::list<T*> cache_objs_;
 public:
-    SrsRtpPacketCacheManager();
-    virtual ~SrsRtpPacketCacheManager();
+    SrsRtpObjectCacheManager() {
+        enabled_ = false;
+    }
+    virtual ~SrsRtpObjectCacheManager() {
+        typedef typename std::list<T*>::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<SrsRtpPacket2>* _srs_rtp_cache;
+
+// For RTP packet shared messages cache.
+extern SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache;
 
 // Single payload data.
 class SrsRtpRawPayload : public ISrsRtpPayloader