RTC: Refine the stat logs, limit the object cache pool

trunk/conf/full.conf

@@ -497,12 +497,30 @@ rtc_server {
     #       then system queue is 2000*4 = 8k, user can incrase reuseport to incrase the queue.
     # default: 2000
     queue_length    2000;
+    # For RTP packet and its payload cache.
+    rtp_cache {
         # Whether enable the RTP packet cache.
         # default: off
-    rtp_cache off;
+        enabled off;
+        # The cache size for rtp packet in MB, each object is about 300B..
+        # default: 128
+        pkt_size 128.0;
+        # The cache size for rtp payload in MB, each object is about 40B.
+        # default: 32
+        payload_size 32.0;
+    }
+    # For RTP shared message and the large buffer cache.
+    rtp_msg_cache {
         #Whether enable the RTP message(a large buffer) cache.
         # default: off
-    rtp_msg_cache off;
+        enabled off;
+        # The cache size for message object in MB, each object is about 40B.
+        # default: 32
+        msg_size 32.0;
+        # The cache size for message large buffer in MB, each object is about 1500B.
+        # default: 1024
+        buffer_size 1024.0;
+    }
     # 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

@@ -4902,16 +4902,31 @@ bool SrsConfig::get_rtc_server_perf_stat()
     return SRS_CONF_PERFER_TRUE(conf->arg0());
 }
 
-bool SrsConfig::get_rtc_server_rtp_cache()
+SrsConfDirective* SrsConfig::get_rtc_server_rtp_cache()
+{
+    SrsConfDirective* conf = root->get("rtc_server");
+    if (!conf) {
+        return NULL;
+    }
+
+    conf = conf->get("rtp_cache");
+    if (!conf) {
+        return NULL;
+    }
+
+    return conf;
+}
+
+bool SrsConfig::get_rtc_server_rtp_cache_enabled()
 {
     static bool DEFAULT = false;
 
-    SrsConfDirective* conf = root->get("rtc_server");
+    SrsConfDirective* conf = get_rtc_server_rtp_cache();
     if (!conf) {
         return DEFAULT;
     }
 
-    conf = conf->get("rtp_cache");
+    conf = conf->get("enabled");
     if (!conf || conf->arg0().empty()) {
         return DEFAULT;
     }
@@ -4919,16 +4934,65 @@ bool SrsConfig::get_rtc_server_rtp_cache()
     return SRS_CONF_PERFER_FALSE(conf->arg0());
 }
 
-bool SrsConfig::get_rtc_server_rtp_msg_cache()
+uint64_t SrsConfig::get_rtc_server_rtp_cache_pkt_size()
 {
-    static bool DEFAULT = false;
+    int DEFAULT = 128 * 1024 * 1024;
 
-    SrsConfDirective* conf = root->get("rtc_server");
+    SrsConfDirective* conf = get_rtc_server_rtp_cache();
     if (!conf) {
         return DEFAULT;
     }
 
+    conf = conf->get("pkt_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return (uint64_t)(1024 * ::atof(conf->arg0().c_str()));
+}
+
+uint64_t SrsConfig::get_rtc_server_rtp_cache_payload_size()
+{
+    int DEFAULT = 32 * 1024 * 1024;
+
+    SrsConfDirective* conf = get_rtc_server_rtp_cache();
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("payload_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return (uint64_t)(1024 * ::atof(conf->arg0().c_str()));
+}
+
+SrsConfDirective* SrsConfig::get_rtc_server_rtp_msg_cache()
+{
+    SrsConfDirective* conf = root->get("rtc_server");
+    if (!conf) {
+        return NULL;
+    }
+
     conf = conf->get("rtp_msg_cache");
+    if (!conf) {
+        return NULL;
+    }
+
+    return conf;
+}
+
+bool SrsConfig::get_rtc_server_rtp_msg_cache_enabled()
+{
+    static bool DEFAULT = false;
+
+    SrsConfDirective* conf = get_rtc_server_rtp_msg_cache();
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("enabled");
     if (!conf || conf->arg0().empty()) {
         return DEFAULT;
     }
@@ -4936,6 +5000,40 @@ bool SrsConfig::get_rtc_server_rtp_msg_cache()
     return SRS_CONF_PERFER_FALSE(conf->arg0());
 }
 
+uint64_t SrsConfig::get_rtc_server_rtp_msg_cache_msg_size()
+{
+    int DEFAULT = 32 * 1024 * 1024;
+
+    SrsConfDirective* conf = get_rtc_server_rtp_msg_cache();
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("msg_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return (uint64_t)(1024 * ::atof(conf->arg0().c_str()));
+}
+
+uint64_t SrsConfig::get_rtc_server_rtp_msg_cache_buffer_size()
+{
+    int DEFAULT = 1024 * 1024 * 1024;
+
+    SrsConfDirective* conf = get_rtc_server_rtp_msg_cache();
+    if (!conf) {
+        return DEFAULT;
+    }
+
+    conf = conf->get("buffer_size");
+    if (!conf || conf->arg0().empty()) {
+        return DEFAULT;
+    }
+
+    return (uint64_t)(1024 * ::atof(conf->arg0().c_str()));
+}
+
 bool SrsConfig::get_rtc_server_black_hole()
 {
     static bool DEFAULT = false;

trunk/src/app/srs_app_config.hpp

@@ -534,8 +534,19 @@ 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();
+private:
-    virtual bool get_rtc_server_rtp_msg_cache();
+    SrsConfDirective* get_rtc_server_rtp_cache();
+public:
+    virtual bool get_rtc_server_rtp_cache_enabled();
+    virtual uint64_t get_rtc_server_rtp_cache_pkt_size();
+    virtual uint64_t get_rtc_server_rtp_cache_payload_size();
+private:
+    virtual SrsConfDirective* get_rtc_server_rtp_msg_cache();
+public:
+    virtual bool get_rtc_server_rtp_msg_cache_enabled();
+    virtual uint64_t get_rtc_server_rtp_msg_cache_msg_size();
+    virtual uint64_t get_rtc_server_rtp_msg_cache_buffer_size();
+public:
     virtual bool get_rtc_server_black_hole();
     virtual std::string get_rtc_server_black_hole_addr();
 private:

trunk/src/app/srs_app_hybrid.cpp

@@ -385,15 +385,19 @@ srs_error_t SrsHybridServer::notify(int event, srs_utime_t interval, srs_utime_t
 #endif
 
     string objs_desc;
-    _srs_pps_objs_rtps->update(); _srs_pps_objs_rraw->update(); _srs_pps_objs_rfua->update(); _srs_pps_objs_rbuf->update(); _srs_pps_objs_msgs->update(); _srs_pps_objs_rothers->update();
+    _srs_pps_objs_rtps->update(); _srs_pps_objs_rraw->update(); _srs_pps_objs_rfua->update(); _srs_pps_objs_rbuf->update(); _srs_pps_objs_msgs->update(); _srs_pps_objs_rothers->update(); _srs_pps_objs_drop->update();
-    if (_srs_pps_objs_rtps->r10s() || _srs_pps_objs_rraw->r10s() || _srs_pps_objs_rfua->r10s() || _srs_pps_objs_rbuf->r10s() || _srs_pps_objs_msgs->r10s() || _srs_pps_objs_rothers->r10s()) {
+    if (_srs_pps_objs_rtps->r10s() || _srs_pps_objs_rraw->r10s() || _srs_pps_objs_rfua->r10s() || _srs_pps_objs_rbuf->r10s() || _srs_pps_objs_msgs->r10s() || _srs_pps_objs_rothers->r10s() || _srs_pps_objs_drop->r10s()) {
-        snprintf(buf, sizeof(buf), ", objs=%d,%d,%d,%d,%d,%d", _srs_pps_objs_rtps->r10s(), _srs_pps_objs_rraw->r10s(), _srs_pps_objs_rfua->r10s(), _srs_pps_objs_msgs->r10s(), _srs_pps_objs_rothers->r10s(), _srs_pps_objs_rbuf->r10s());
+        snprintf(buf, sizeof(buf), ", objs=(pkt:%d,raw:%d,fua:%d,msg:%d,oth:%d,buf:%d,drop:%d)",
+            _srs_pps_objs_rtps->r10s(), _srs_pps_objs_rraw->r10s(), _srs_pps_objs_rfua->r10s(),
+            _srs_pps_objs_msgs->r10s(), _srs_pps_objs_rothers->r10s(), _srs_pps_objs_rbuf->r10s(), _srs_pps_objs_drop->r10s());
         objs_desc = buf;
     }
 
     string cache_desc;
-    if (true) {
+    if (_srs_rtp_cache->size() || _srs_rtp_raw_cache->size() || _srs_rtp_fua_cache->size() || _srs_rtp_msg_cache_buffers->size() || _srs_rtp_msg_cache_objs->size()) {
-        snprintf(buf, sizeof(buf), ", cache=%d,%d,%d,%d,%d", _srs_rtp_cache->size(), _srs_rtp_raw_cache->size(), _srs_rtp_fua_cache->size(), _srs_rtp_msg_cache->size(), _srs_rtp_msg_cache2->size());
+        snprintf(buf, sizeof(buf), ", cache=(pkt:%d-%dw,raw:%d-%dw,fua:%d-%dw,msg:%d-%dw,buf:%d-%dw)",
+            _srs_rtp_cache->size(), _srs_rtp_cache->capacity()/10000, _srs_rtp_raw_cache->size(), _srs_rtp_raw_cache->capacity()/10000,
+            _srs_rtp_fua_cache->size(), _srs_rtp_fua_cache->capacity()/10000, _srs_rtp_msg_cache_buffers->size(), _srs_rtp_msg_cache_buffers->capacity()/10000, _srs_rtp_msg_cache_objs->size(), _srs_rtp_msg_cache_objs->capacity()/10000);
         cache_desc = buf;
     }
 

trunk/src/app/srs_app_rtc_server.cpp

@@ -279,16 +279,24 @@ srs_error_t SrsRtcServer::initialize()
         return srs_error_wrap(err, "black hole");
     }
 
-    bool rtp_cache = _srs_config->get_rtc_server_rtp_cache();
+    bool rtp_cache_enabled = _srs_config->get_rtc_server_rtp_cache_enabled();
-    _srs_rtp_cache->set_enabled(rtp_cache);
+    uint64_t rtp_cache_pkt_size = _srs_config->get_rtc_server_rtp_cache_pkt_size();
-    _srs_rtp_raw_cache->set_enabled(rtp_cache);
+    uint64_t rtp_cache_payload_size = _srs_config->get_rtc_server_rtp_cache_payload_size();
-    _srs_rtp_fua_cache->set_enabled(rtp_cache);
+    _srs_rtp_cache->set_enabled(rtp_cache_enabled, rtp_cache_pkt_size);
+    _srs_rtp_raw_cache->set_enabled(rtp_cache_enabled, rtp_cache_payload_size);
+    _srs_rtp_fua_cache->set_enabled(rtp_cache_enabled, rtp_cache_payload_size);
 
-    bool rtp_msg_cache = _srs_config->get_rtc_server_rtp_msg_cache();
+    bool rtp_msg_cache_enabled = _srs_config->get_rtc_server_rtp_msg_cache_enabled();
-    _srs_rtp_msg_cache->set_enabled(rtp_msg_cache);
+    uint64_t rtp_msg_cache_msg_size = _srs_config->get_rtc_server_rtp_msg_cache_msg_size();
-    _srs_rtp_msg_cache2->set_enabled(rtp_msg_cache);
+    uint64_t rtp_msg_cache_buffer_size = _srs_config->get_rtc_server_rtp_msg_cache_buffer_size();
+    _srs_rtp_msg_cache_buffers->set_enabled(rtp_msg_cache_enabled, rtp_msg_cache_msg_size);
+    _srs_rtp_msg_cache_objs->set_enabled(rtp_msg_cache_enabled, rtp_msg_cache_buffer_size);
 
-    srs_trace("RTC server init ok, rc=%d, rmc=%d", rtp_cache, rtp_msg_cache);
+    srs_trace("RTC server init ok, rtp-cache=(enabled:%d,pkt:%dm-%dw,payload:%dm-%dw-%dw), msg-cache=(enabled:%d,obj:%dm-%dw,buf:%dm-%dw)",
+        rtp_cache_enabled, (int)(rtp_cache_pkt_size/1024/1024), _srs_rtp_cache->capacity()/10000,
+        (int)(rtp_cache_payload_size/1024/1024), _srs_rtp_raw_cache->capacity()/10000, _srs_rtp_fua_cache->capacity()/10000,
+        rtp_msg_cache_enabled, (int)(rtp_msg_cache_msg_size/1024/1024), _srs_rtp_msg_cache_objs->capacity()/10000,
+        (int)(rtp_msg_cache_buffer_size/1024/1024), _srs_rtp_msg_cache_buffers->capacity()/10000);
 
     return err;
 }
@@ -688,49 +696,49 @@ srs_error_t SrsRtcServer::notify(int type, srs_utime_t interval, srs_utime_t tic
     string rpkts_desc;
     _srs_pps_rpkts->update(); _srs_pps_rrtps->update(); _srs_pps_rstuns->update(); _srs_pps_rrtcps->update();
     if (_srs_pps_rpkts->r10s() || _srs_pps_rrtps->r10s() || _srs_pps_rstuns->r10s() || _srs_pps_rrtcps->r10s()) {
-        snprintf(buf, sizeof(buf), ", rpkts=%d,%d,%d,%d", _srs_pps_rpkts->r10s(), _srs_pps_rrtps->r10s(), _srs_pps_rstuns->r10s(), _srs_pps_rrtcps->r10s());
+        snprintf(buf, sizeof(buf), ", rpkts=(%d,rtp:%d,stun:%d,rtcp:%d)", _srs_pps_rpkts->r10s(), _srs_pps_rrtps->r10s(), _srs_pps_rstuns->r10s(), _srs_pps_rrtcps->r10s());
         rpkts_desc = buf;
     }
 
     string spkts_desc;
     _srs_pps_spkts->update(); _srs_pps_srtps->update(); _srs_pps_sstuns->update(); _srs_pps_srtcps->update();
     if (_srs_pps_spkts->r10s() || _srs_pps_srtps->r10s() || _srs_pps_sstuns->r10s() || _srs_pps_srtcps->r10s()) {
-        snprintf(buf, sizeof(buf), ", spkts=%d,%d,%d,%d", _srs_pps_spkts->r10s(), _srs_pps_srtps->r10s(), _srs_pps_sstuns->r10s(), _srs_pps_srtcps->r10s());
+        snprintf(buf, sizeof(buf), ", spkts=(%d,rtp:%d,stun:%d,rtcp:%d)", _srs_pps_spkts->r10s(), _srs_pps_srtps->r10s(), _srs_pps_sstuns->r10s(), _srs_pps_srtcps->r10s());
         spkts_desc = buf;
     }
 
     string rtcp_desc;
     _srs_pps_pli->update(); _srs_pps_twcc->update(); _srs_pps_rr->update();
     if (_srs_pps_pli->r10s() || _srs_pps_twcc->r10s() || _srs_pps_rr->r10s()) {
-        snprintf(buf, sizeof(buf), ", rtcp=%d,%d,%d", _srs_pps_pli->r10s(), _srs_pps_twcc->r10s(), _srs_pps_rr->r10s());
+        snprintf(buf, sizeof(buf), ", rtcp=(pli:%d,twcc:%d,rr:%d)", _srs_pps_pli->r10s(), _srs_pps_twcc->r10s(), _srs_pps_rr->r10s());
         rtcp_desc = buf;
     }
 
     string snk_desc;
     _srs_pps_snack->update(); _srs_pps_snack2->update(); _srs_pps_sanack->update(); _srs_pps_svnack->update();
     if (_srs_pps_snack->r10s() || _srs_pps_sanack->r10s() || _srs_pps_svnack->r10s() || _srs_pps_snack2->r10s()) {
-        snprintf(buf, sizeof(buf), ", snk=%d,%d,%d,%d", _srs_pps_snack->r10s(), _srs_pps_sanack->r10s(), _srs_pps_svnack->r10s(), _srs_pps_snack2->r10s());
+        snprintf(buf, sizeof(buf), ", snk=(%d,a:%d,v:%d,h:%d)", _srs_pps_snack->r10s(), _srs_pps_sanack->r10s(), _srs_pps_svnack->r10s(), _srs_pps_snack2->r10s());
         snk_desc = buf;
     }
 
     string rnk_desc;
     _srs_pps_rnack->update(); _srs_pps_rnack2->update(); _srs_pps_rhnack->update(); _srs_pps_rmnack->update();
     if (_srs_pps_rnack->r10s() || _srs_pps_rnack2->r10s() || _srs_pps_rhnack->r10s() || _srs_pps_rmnack->r10s()) {
-        snprintf(buf, sizeof(buf), ", rnk=%d,%d,%d,%d", _srs_pps_rnack->r10s(), _srs_pps_rnack2->r10s(), _srs_pps_rhnack->r10s(), _srs_pps_rmnack->r10s());
+        snprintf(buf, sizeof(buf), ", rnk=(%d,%d,h:%d,m:%d)", _srs_pps_rnack->r10s(), _srs_pps_rnack2->r10s(), _srs_pps_rhnack->r10s(), _srs_pps_rmnack->r10s());
         rnk_desc = buf;
     }
 
     string drop_desc;
     SrsSnmpUdpStat* s = srs_get_udp_snmp_stat();
     if (s->rcv_buf_errors_delta || s->snd_buf_errors_delta) {
-        snprintf(buf, sizeof(buf), ", drop=%d,%d", s->rcv_buf_errors_delta, s->snd_buf_errors_delta);
+        snprintf(buf, sizeof(buf), ", drop=(r:%d,s:%d)", s->rcv_buf_errors_delta, s->snd_buf_errors_delta);
         drop_desc = buf;
     }
 
     string fid_desc;
     _srs_pps_ids->update(); _srs_pps_fids->update(); _srs_pps_fids_level0->update(); _srs_pps_addrs->update(); _srs_pps_fast_addrs->update();
     if (_srs_pps_ids->r10s(), _srs_pps_fids->r10s(), _srs_pps_fids_level0->r10s(), _srs_pps_addrs->r10s(), _srs_pps_fast_addrs->r10s()) {
-        snprintf(buf, sizeof(buf), ", fid=%d,%d,%d,%d,%d", _srs_pps_ids->r10s(), _srs_pps_fids->r10s(), _srs_pps_fids_level0->r10s(), _srs_pps_addrs->r10s(), _srs_pps_fast_addrs->r10s());
+        snprintf(buf, sizeof(buf), ", fid=(id:%d,fid:%d,ffid:%d,addr:%d,faddr:%d)", _srs_pps_ids->r10s(), _srs_pps_fids->r10s(), _srs_pps_fids_level0->r10s(), _srs_pps_addrs->r10s(), _srs_pps_fast_addrs->r10s());
         fid_desc = buf;
     }
 

trunk/src/kernel/srs_kernel_flv.cpp

@@ -361,7 +361,7 @@ SrsSharedPtrMessage* SrsSharedPtrMessage::copy()
 {
     srs_assert(ptr);
     
-    SrsSharedPtrMessage* copy = _srs_rtp_msg_cache2->allocate();
+    SrsSharedPtrMessage* copy = _srs_rtp_msg_cache_objs->allocate();
 
     // We got an object from cache, the ptr might exists, so unwrap it.
     copy->unwrap();

trunk/src/kernel/srs_kernel_rtc_rtp.cpp

@@ -41,6 +41,7 @@ SrsPps* _srs_pps_objs_rraw = new SrsPps();
 SrsPps* _srs_pps_objs_rfua = new SrsPps();
 SrsPps* _srs_pps_objs_rbuf = new SrsPps();
 SrsPps* _srs_pps_objs_rothers = new SrsPps();
+SrsPps* _srs_pps_objs_drop = new SrsPps();
 
 /* @see https://tools.ietf.org/html/rfc1889#section-5.1
   0                   1                   2                   3
@@ -851,10 +852,10 @@ void SrsRtpPacket2::reuse_shared_msg()
 
     // We only recycle the RTC UDP packet messages.
     if (shared_msg->payload && shared_msg->size == kRtpPacketSize && shared_msg->count() == 0) {
-        _srs_rtp_msg_cache->recycle(shared_msg);
+        _srs_rtp_msg_cache_buffers->recycle(shared_msg);
     } else {
         shared_msg->unwrap();
-        _srs_rtp_msg_cache2->recycle(shared_msg);
+        _srs_rtp_msg_cache_objs->recycle(shared_msg);
     }
 
     shared_msg = NULL;
@@ -885,7 +886,7 @@ char* SrsRtpPacket2::wrap(int size)
     // Create a large enough message, with under-layer buffer.
     while (true) {
         srs_freep(shared_msg);
-        shared_msg = _srs_rtp_msg_cache->allocate();
+        shared_msg = _srs_rtp_msg_cache_buffers->allocate();
 
         // If got a cached message(which has payload), but it's too small,
         // we free it and allocate a larger one.
@@ -1050,8 +1051,8 @@ SrsRtpObjectCacheManager<SrsRtpPacket2>* _srs_rtp_cache = new SrsRtpObjectCacheM
 SrsRtpObjectCacheManager<SrsRtpRawPayload>* _srs_rtp_raw_cache = new SrsRtpObjectCacheManager<SrsRtpRawPayload>();
 SrsRtpObjectCacheManager<SrsRtpFUAPayload2>* _srs_rtp_fua_cache = new SrsRtpObjectCacheManager<SrsRtpFUAPayload2>();
 
-SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache = new SrsRtpObjectCacheManager<SrsSharedPtrMessage>();
+SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache_buffers = new SrsRtpObjectCacheManager<SrsSharedPtrMessage>();
-SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache2 = new SrsRtpObjectCacheManager<SrsSharedPtrMessage>();
+SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache_objs = new SrsRtpObjectCacheManager<SrsSharedPtrMessage>();
 
 SrsRtpRawPayload::SrsRtpRawPayload()
 {

trunk/src/kernel/srs_kernel_rtc_rtp.hpp

@@ -333,6 +333,10 @@ public:
     virtual srs_error_t decode(SrsBuffer* buf);
 };
 
+// For object cache manager to stat the object dropped.
+#include <srs_kernel_kbps.hpp>
+extern SrsPps* _srs_pps_objs_drop;
+
 // The RTP packet or message cache manager.
 template<typename T>
 class SrsRtpObjectCacheManager
@@ -340,9 +344,11 @@ class SrsRtpObjectCacheManager
 private:
     bool enabled_;
     std::list<T*> cache_objs_;
+    size_t capacity_;
 public:
     SrsRtpObjectCacheManager() {
         enabled_ = false;
+        capacity_ = 0;
     }
     virtual ~SrsRtpObjectCacheManager() {
         typedef typename std::list<T*>::iterator iterator;
@@ -353,8 +359,9 @@ public:
     }
 public:
     // Enable or disable cache.
-    void set_enabled(bool v) {
+    void set_enabled(bool v, uint64_t memory) {
         enabled_ = v;
+        capacity_ = (size_t)(memory / sizeof(T));
     }
     bool enabled() {
         return enabled_;
@@ -362,6 +369,9 @@ public:
     int size() {
         return (int)cache_objs_.size();
     }
+    int capacity() {
+        return (int)capacity_;
+    }
     // Try to allocate from cache, create new object if no cache.
     T* allocate() {
         while (true) {
@@ -389,12 +399,20 @@ public:
             return;
         }
 
-        // TODO: FIXME: Directly free to keep low memory?
+        // If disabled, drop the object.
         if (!enabled_) {
             srs_freep(p);
             return;
         }
 
+        // If exceed the capacity, drop the object.
+        if (cache_objs_.size() > capacity_) {
+            ++_srs_pps_objs_drop->sugar;
+
+            srs_freep(p);
+            return;
+        }
+
         // Recycle it.
         cache_objs_.push_back(p);
     }
@@ -527,7 +545,7 @@ extern SrsRtpObjectCacheManager<SrsRtpRawPayload>* _srs_rtp_raw_cache;
 extern SrsRtpObjectCacheManager<SrsRtpFUAPayload2>* _srs_rtp_fua_cache;
 
 // For RTP packet shared messages cache.
-extern SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache;
+extern SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache_buffers;
-extern SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache2;
+extern SrsRtpObjectCacheManager<SrsSharedPtrMessage>* _srs_rtp_msg_cache_objs;
 
 #endif