Perf: Avoid RTP packet copy for player NACK.

trunk/src/app/srs_app_rtc_conn.cpp

@@ -448,17 +448,6 @@ srs_error_t SrsRtcPlayStream::initialize(SrsRequest* req, std::map<uint32_t, Srs
     nack_enabled_ = _srs_config->get_rtc_nack_enabled(req->vhost);
     srs_trace("RTC player nack=%d", nack_enabled_);
 
-    // Apply configs for all tracks.
-    for (map<uint32_t, SrsRtcAudioSendTrack*>::iterator it = audio_tracks_.begin(); it != audio_tracks_.end(); ++it) {
-        SrsRtcAudioSendTrack* track = it->second;
-        track->set_nack_enabled(nack_enabled_);
-    }
-
-    for (map<uint32_t, SrsRtcVideoSendTrack*>::iterator it = video_tracks_.begin(); it != video_tracks_.end(); ++it) {
-        SrsRtcVideoSendTrack* track = it->second;
-        track->set_nack_enabled(nack_enabled_);
-    }
-
     // Update stat for session.
     session_->stat_->nn_subscribers++;
 
@@ -582,6 +571,7 @@ srs_error_t SrsRtcPlayStream::cycle()
         }
 
         // Send-out the RTP packet and do cleanup
+        // @remark Note that the pkt might be set to NULL.
         if ((err = send_packet(pkt)) != srs_success) {
             uint32_t nn = 0;
             if (epp->can_print(err, &nn)) {
@@ -590,11 +580,13 @@ srs_error_t SrsRtcPlayStream::cycle()
             srs_freep(err);
         }
 
+        // Release the packet to cache.
+        // @remark Note that the pkt might be set to NULL.
         _srs_rtp_cache->recycle(pkt);
     }
 }
 
-srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket2* pkt)
+srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket2*& pkt)
 {
     srs_error_t err = srs_success;
 
@@ -605,9 +597,11 @@ srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket2* pkt)
     }
 
     // For audio, we transcoded AAC to opus in extra payloads.
+    SrsRtcAudioSendTrack* audio_track = NULL;
+    SrsRtcVideoSendTrack* video_track = NULL;
     if (pkt->is_audio()) {
         // TODO: FIXME: Any simple solution?
-        SrsRtcAudioSendTrack* audio_track = audio_tracks_[pkt->header.get_ssrc()];
+        audio_track = audio_tracks_[pkt->header.get_ssrc()];
 
         if ((err = audio_track->on_rtp(pkt)) != srs_success) {
             return srs_error_wrap(err, "audio track, SSRC=%u, SEQ=%u", pkt->header.get_ssrc(), pkt->header.get_sequence());
@@ -616,16 +610,26 @@ srs_error_t SrsRtcPlayStream::send_packet(SrsRtpPacket2* pkt)
         // TODO: FIXME: Padding audio to the max payload in RTP packets.
     } else {
         // TODO: FIXME: Any simple solution?
-        SrsRtcVideoSendTrack* video_track = video_tracks_[pkt->header.get_ssrc()];
+        video_track = video_tracks_[pkt->header.get_ssrc()];
 
         if ((err = video_track->on_rtp(pkt)) != srs_success) {
             return srs_error_wrap(err, "video track, SSRC=%u, SEQ=%u", pkt->header.get_ssrc(), pkt->header.get_sequence());
         }
     }
 
-    // Detail log, should disable it in release version.
-    srs_info("RTC: Update PT=%u, SSRC=%#x, Time=%u, %u bytes", pkt->header.get_payload_type(), pkt->header.get_ssrc(),
-        pkt->header.get_timestamp(), pkt->nb_bytes());
+    // For NACK to handle packet.
+    // @remark Note that the pkt might be set to NULL.
+    if (nack_enabled_) {
+        if (audio_track) {
+            if ((err = audio_track->on_nack(&pkt)) != srs_success) {
+                return srs_error_wrap(err, "on nack");
+            }
+        } else if (video_track) {
+            if ((err = video_track->on_nack(&pkt)) != srs_success) {
+                return srs_error_wrap(err, "on nack");
+            }
+        }
+    }
 
     return err;
 }
@@ -1158,6 +1162,7 @@ srs_error_t SrsRtcPublishStream::on_rtp_plaintext(char* plaintext, int nb_plaint
     err = do_on_rtp_plaintext(pkt, &buf);
 
     // Release the packet to cache.
+    // @remark Note that the pkt might be set to NULL.
     _srs_rtp_cache->recycle(pkt);
 
     return err;

trunk/src/app/srs_app_rtc_conn.hpp

@@ -252,7 +252,7 @@ class SrsRtcPlayStream : virtual public ISrsCoroutineHandler, virtual public ISr
 public:
     virtual srs_error_t cycle();
 private:
-    srs_error_t send_packet(SrsRtpPacket2* pkt);
+    srs_error_t send_packet(SrsRtpPacket2*& pkt);
 public:
     // Directly set the status of track, generally for init to set the default value.
     void set_all_tracks_status(bool status);

trunk/src/app/srs_app_rtc_source.cpp

@@ -1983,7 +1983,6 @@ SrsRtcSendTrack::SrsRtcSendTrack(SrsRtcConnection* session, SrsRtcTrackDescripti
     session_ = session;
     track_desc_ = track_desc->copy();
     statistic_ = new SrsRtcTrackStatistic();
-    nack_enabled_ = false;
 
     if (is_audio) {
         rtp_queue_ = new SrsRtpRingBuffer(100);
@@ -2002,11 +2001,6 @@ SrsRtcSendTrack::~SrsRtcSendTrack()
     srs_freep(nack_epp);
 }
 
-void SrsRtcSendTrack::set_nack_enabled(bool v)
-{
-    nack_enabled_ = v;
-}
-
 bool SrsRtcSendTrack::has_ssrc(uint32_t ssrc)
 {
     return track_desc_->has_ssrc(ssrc);
@@ -2055,6 +2049,21 @@ std::string SrsRtcSendTrack::get_track_id()
     return track_desc_->id_;
 }
 
+srs_error_t SrsRtcSendTrack::on_nack(SrsRtpPacket2** ppkt)
+{
+    srs_error_t err = srs_success;
+
+    SrsRtpPacket2* pkt = *ppkt;
+    uint16_t seq = pkt->header.get_sequence();
+
+    // insert into video_queue and audio_queue
+    // We directly use the pkt, never copy it, so we should set the pkt to NULL.
+    rtp_queue_->set(seq, pkt);
+    *ppkt = NULL;
+
+    return err;
+}
+
 srs_error_t SrsRtcSendTrack::on_recv_nack(const vector<uint16_t>& lost_seqs)
 {
     srs_error_t err = srs_success;
@@ -2117,12 +2126,6 @@ srs_error_t SrsRtcAudioSendTrack::on_rtp(SrsRtpPacket2* pkt)
         // TODO: FIXME: Should update PT for RTX.
     }
 
-    // Put rtp packet to NACK/ARQ queue
-    if (nack_enabled_) {
-        SrsRtpPacket2* nack = pkt->copy();
-        rtp_queue_->set(nack->header.get_sequence(), nack);
-    }
-
     // Update stats.
     session_->stat_->nn_out_audios++;
 
@@ -2177,12 +2180,6 @@ srs_error_t SrsRtcVideoSendTrack::on_rtp(SrsRtpPacket2* pkt)
         // TODO: FIXME: Should update PT for RTX.
     }
 
-    // Put rtp packet to NACK/ARQ queue
-    if (nack_enabled_) {
-        SrsRtpPacket2* nack = pkt->copy();
-        rtp_queue_->set(nack->header.get_sequence(), nack);
-    }
-
     // Update stats.
     session_->stat_->nn_out_videos++;
 

trunk/src/app/srs_app_rtc_source.hpp

@@ -579,21 +579,22 @@ class SrsRtcSendTrack
     SrsRtcConnection* session_;
     // NACK ARQ ring buffer.
     SrsRtpRingBuffer* rtp_queue_;
-    // Whether enabled nack.
-    bool nack_enabled_;
 private:
     // The pithy print for special stage.
     SrsErrorPithyPrint* nack_epp;
 public:
     SrsRtcSendTrack(SrsRtcConnection* session, SrsRtcTrackDescription* track_desc, bool is_audio);
     virtual ~SrsRtcSendTrack();
 public:
-    void set_nack_enabled(bool v);
     bool has_ssrc(uint32_t ssrc);
     SrsRtpPacket2* fetch_rtp_packet(uint16_t seq);
     bool set_track_status(bool active);
     bool get_track_status();
     std::string get_track_id();
+public:
+    // Note that we can set the pkt to NULL to avoid copy, for example, if the NACK cache the pkt and
+    // set to NULL, nack nerver copy it but set the pkt to NULL.
+    srs_error_t on_nack(SrsRtpPacket2** ppkt);
 public:
     virtual srs_error_t on_rtp(SrsRtpPacket2* pkt) = 0;
     virtual srs_error_t on_rtcp(SrsRtpPacket2* pkt) = 0;