For #307, zero copy for RTP STAP packet

trunk/src/app/srs_app_rtc.hpp

@@ -48,11 +48,7 @@ const int kRtpPacketSize        = 1500;
 const uint8_t kOpusPayloadType  = 111;
 const uint8_t kH264PayloadType  = 102;
 
-// H.264 nalu header type mask.
-const uint8_t kNalTypeMask      = 0x1F;
-
 // @see: https://tools.ietf.org/html/rfc6184#section-5.2
-const uint8_t kStapA            = 24;
 const uint8_t kFuA              = 28;
 
 // @see: https://tools.ietf.org/html/rfc6184#section-5.8

trunk/src/app/srs_app_rtc_conn.cpp

@@ -398,7 +398,7 @@ srs_error_t SrsDtlsSession::protect_rtp2(char* buf, int* pnn_buf, SrsRtpPacket2*
     if ((err = pkt->encode(&stream)) != srs_success) {
         return srs_error_wrap(err, "encode packet");
     }
-    
+
     *pnn_buf = stream.pos();
 
     if (srtp_protect(srtp_send, buf, pnn_buf) != 0) {
@@ -654,9 +654,18 @@ srs_error_t SrsRtcSenderThread::send_messages(
 
                 // Well, for each IDR, we append a SPS/PPS before it, which is packaged in STAP-A.
                 if (msg->has_idr()) {
-                    if ((err = packet_stap_a(source, msg, rtp_packets)) != srs_success) {
+                    SrsRtpPacket2* packet = NULL;
+                    if ((err = packet_stap_a(source, msg, &packet)) != srs_success) {
                         return srs_error_wrap(err, "packet stap-a");
                     }
+
+                    err = send_message2(msg, is_video, is_audio, packet, skt);
+                    srs_freep(packet);
+                    if (err != srs_success) {
+                        return srs_error_wrap(err, "send message");
+                    }
+
+                    *pnn_rtp_pkts += 1;
                 }
 
                 if (sample->size <= kRtpMaxPayloadSize) {
@@ -778,8 +787,10 @@ srs_error_t SrsRtcSenderThread::packet_opus(SrsSample* sample, SrsRtpPacket2** p
     packet->rtp_header.set_ssrc(audio_ssrc);
     packet->rtp_header.set_payload_type(audio_payload_type);
 
-    packet->payload = sample->bytes;
-    packet->nn_payload = sample->size;
+    SrsRtpRawPayload* raw = new SrsRtpRawPayload();
+    raw->payload = sample->bytes;
+    raw->nn_payload = sample->size;
+    packet->payload = raw;
 
     // TODO: FIXME: Why 960? Need Refactoring?
     audio_timestamp += 960;
@@ -852,7 +863,7 @@ srs_error_t SrsRtcSenderThread::packet_single_nalu(SrsSharedPtrMessage* shared_f
     return err;
 }
 
-srs_error_t SrsRtcSenderThread::packet_stap_a(SrsSource* source, SrsSharedPtrMessage* shared_frame, vector<SrsRtpSharedPacket*>& rtp_packets)
+srs_error_t SrsRtcSenderThread::packet_stap_a(SrsSource* source, SrsSharedPtrMessage* shared_frame, SrsRtpPacket2** ppacket)
 {
     srs_error_t err = srs_success;
 
@@ -872,32 +883,34 @@ srs_error_t SrsRtcSenderThread::packet_stap_a(SrsSource* source, SrsSharedPtrMes
         return srs_error_new(ERROR_RTC_RTP_MUXER, "sps/pps empty");
     }
 
-    uint8_t header = sps[0];
-    uint8_t nal_type = header & kNalTypeMask;
+    srs_verbose("rtp stap-a nalu, size=%u, seq=%u, timestamp=%lu",
+        (sps.size() + pps.size()), video_sequence, (shared_frame->timestamp * 90));
 
-    char buf[kRtpPacketSize];
-    SrsBuffer* stream = new SrsBuffer(buf, kRtpPacketSize);
-    SrsAutoFree(SrsBuffer, stream);
+    SrsRtpPacket2* packet = new SrsRtpPacket2();
+    packet->rtp_header.set_marker(false);
+    packet->rtp_header.set_timestamp(shared_frame->timestamp * 90);
+    packet->rtp_header.set_sequence(video_sequence++);
+    packet->rtp_header.set_ssrc(video_ssrc);
+    packet->rtp_header.set_payload_type(video_payload_type);
 
-    // stap-a header
-    uint8_t stap_a_header = kStapA;
-    stap_a_header |= (nal_type & (~kNalTypeMask));
-    stream->write_1bytes(stap_a_header);
+    SrsRtpSTAPPayload* stap = new SrsRtpSTAPPayload();
 
-    stream->write_2bytes(sps.size());
-    stream->write_bytes((char*)sps.data(), sps.size());
+    uint8_t header = sps[0];
+    uint8_t nal_type = header & kNalTypeMask;
 
-    stream->write_2bytes(pps.size());
-    stream->write_bytes((char*)pps.data(), pps.size());
+    stap->nri = (SrsAvcNaluType)nal_type;
+    stap->nn_nalus = 2;
+    stap->nalus = new SrsSample[stap->nn_nalus];
 
-    srs_verbose("rtp stap-a nalu, size=%u, seq=%u, timestamp=%lu", (sps.size() + pps.size()), video_sequence, (shared_frame->timestamp * 90));
+    stap->nalus[0].bytes = (char*)&sps[0];
+    stap->nalus[0].size = (int)sps.size();
 
-    SrsRtpSharedPacket* packet = new SrsRtpSharedPacket();
-    if ((err = packet->create((shared_frame->timestamp * 90), video_sequence++, kVideoSSRC, kH264PayloadType, stream->data(), stream->pos())) != srs_success) {
-        return srs_error_wrap(err, "rtp packet encode");
-    }
+    stap->nalus[1].bytes = (char*)&pps[0];
+    stap->nalus[1].size = (int)pps.size();
 
-    rtp_packets.push_back(packet);
+    packet->payload = stap;
+
+    *ppacket = packet;
 
     return err;
 }

trunk/src/app/srs_app_rtc_conn.hpp

@@ -160,7 +160,7 @@ private:
 private:
     srs_error_t packet_fu_a(SrsSharedPtrMessage* shared_frame, SrsSample* sample, std::vector<SrsRtpSharedPacket*>& rtp_packets);
     srs_error_t packet_single_nalu(SrsSharedPtrMessage* shared_frame, SrsSample* sample, std::vector<SrsRtpSharedPacket*>& rtp_packets);
-    srs_error_t packet_stap_a(SrsSource* source, SrsSharedPtrMessage* shared_frame, std::vector<SrsRtpSharedPacket*>& rtp_packets);
+    srs_error_t packet_stap_a(SrsSource* source, SrsSharedPtrMessage* shared_frame, SrsRtpPacket2** ppacket);
 };
 
 class SrsRtcSession

trunk/src/kernel/srs_kernel_buffer.cpp

@@ -29,6 +29,14 @@ using namespace std;
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_utility.hpp>
 
+ISrsEncoder::ISrsEncoder()
+{
+}
+
+ISrsEncoder::~ISrsEncoder()
+{
+}
+
 ISrsCodec::ISrsCodec()
 {
 }

trunk/src/kernel/srs_kernel_buffer.hpp

@@ -31,6 +31,24 @@
 
 class SrsBuffer;
 
+// Encoder.
+class ISrsEncoder
+{
+public:
+    ISrsEncoder();
+    virtual ~ISrsEncoder();
+public:
+    /**
+     * get the number of bytes to code to.
+     */
+    // TODO: FIXME: change to uint64_t.
+    virtual int nb_bytes() = 0;
+    /**
+     * encode object to bytes in SrsBuffer.
+     */
+    virtual srs_error_t encode(SrsBuffer* buf) = 0;
+};
+
 /**
  * the srs codec, to code and decode object with bytes:
  *      code: to encode/serialize object to bytes in buffer,
@@ -56,21 +74,11 @@ class SrsBuffer;
  * @remark protocol or amf0 or json should implements this interface.
  */
 // TODO: FIXME: protocol, amf0, json should implements it.
-class ISrsCodec
+class ISrsCodec : public ISrsEncoder
 {
 public:
     ISrsCodec();
     virtual ~ISrsCodec();
-public:
-    /**
-     * get the number of bytes to code to.
-     */
-    // TODO: FIXME: change to uint64_t.
-    virtual int nb_bytes() = 0;
-    /**
-     * encode object to bytes in SrsBuffer.
-     */
-    virtual srs_error_t encode(SrsBuffer* buf) = 0;
 public:
     /**
      * decode object from bytes in SrsBuffer.

trunk/src/kernel/srs_kernel_codec.cpp

@@ -32,6 +32,7 @@ using namespace std;
 #include <srs_kernel_buffer.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_core_autofree.hpp>
+#include <srs_kernel_rtp.hpp>
 
 string srs_video_codec_id2str(SrsVideoCodecId codec)
 {
@@ -375,9 +376,6 @@ srs_error_t SrsSample::parse_bframe()
 {
     srs_error_t err = srs_success;
 
-    // H.264 nalu header type mask.
-    static uint8_t kNalTypeMask      = 0x1F;
-
     uint8_t header = bytes[0];
     SrsAvcNaluType nal_type = (SrsAvcNaluType)(header & kNalTypeMask);
 

trunk/src/kernel/srs_kernel_rtp.cpp

@@ -32,6 +32,8 @@ using namespace std;
 #include <srs_kernel_buffer.hpp>
 #include <srs_kernel_utility.hpp>
 
+// @see: https://tools.ietf.org/html/rfc6184#section-5.2
+const uint8_t kStapA            = 24;
 
 SrsRtpHeader::SrsRtpHeader()
 {
@@ -148,32 +150,106 @@ void SrsRtpHeader::set_ssrc(uint32_t ssrc)
 SrsRtpPacket2::SrsRtpPacket2()
 {
     payload = NULL;
-    nn_payload = 0;
 }
 
 SrsRtpPacket2::~SrsRtpPacket2()
 {
+    srs_freep(payload);
 }
 
-srs_error_t SrsRtpPacket2::encode(SrsBuffer* stream)
+srs_error_t SrsRtpPacket2::encode(SrsBuffer* buf)
 {
     srs_error_t err = srs_success;
 
-    if ((err = rtp_header.encode(stream)) != srs_success) {
+    if ((err = rtp_header.encode(buf)) != srs_success) {
         return srs_error_wrap(err, "rtp header");
     }
 
+    if (payload && (err = payload->encode(buf)) != srs_success) {
+        return srs_error_wrap(err, "encode payload");
+    }
+
+    return err;
+}
+
+SrsRtpRawPayload::SrsRtpRawPayload()
+{
+    payload = NULL;
+    nn_payload = 0;
+}
+
+SrsRtpRawPayload::~SrsRtpRawPayload()
+{
+}
+
+int SrsRtpRawPayload::nb_bytes()
+{
+    return nn_payload;
+}
+
+srs_error_t SrsRtpRawPayload::encode(SrsBuffer* buf)
+{
     if (nn_payload <= 0) {
-        return 0;
+        return srs_success;
     }
 
-    if (!stream->require(nn_payload)) {
+    if (!buf->require(nn_payload)) {
         return srs_error_new(ERROR_RTC_RTP_MUXER, "requires %d bytes", nn_payload);
     }
 
-    stream->write_bytes(payload, nn_payload);
+    buf->write_bytes(payload, nn_payload);
 
-    return err;
+    return srs_success;
+}
+
+SrsRtpSTAPPayload::SrsRtpSTAPPayload()
+{
+    nri = (SrsAvcNaluType)0;
+    nalus = NULL;
+    nn_nalus = 0;
+}
+
+SrsRtpSTAPPayload::~SrsRtpSTAPPayload()
+{
+    srs_freepa(nalus);
+}
+
+int SrsRtpSTAPPayload::nb_bytes()
+{
+    int size = 1;
+
+    for (int i = 0; i < nn_nalus; i++) {
+        SrsSample* p = nalus + i;
+        size += 2 + p->size;
+    }
+
+    return size;
+}
+
+srs_error_t SrsRtpSTAPPayload::encode(SrsBuffer* buf)
+{
+    if (!buf->require(1)) {
+        return srs_error_new(ERROR_RTC_RTP_MUXER, "requires %d bytes", 1);
+    }
+
+    // STAP header, RTP payload format for aggregation packets
+    // @see https://tools.ietf.org/html/rfc6184#section-5.7
+    uint8_t v = kStapA;
+    v |= (nri & (~kNalTypeMask));
+    buf->write_1bytes(v);
+
+    // NALUs.
+    for (int i = 0; i < nn_nalus; i++) {
+        SrsSample* p = nalus + i;
+        if (!buf->require(2 + p->size)) {
+            return srs_error_new(ERROR_RTC_RTP_MUXER, "requires %d bytes", 2 + p->size);
+        }
+
+        buf->write_2bytes(p->size);
+        buf->write_bytes(p->bytes, p->size);
+    }
+
+    return srs_success;
 }
 
 SrsRtpSharedPacket::SrsRtpSharedPacketPayload::SrsRtpSharedPacketPayload()

trunk/src/kernel/srs_kernel_rtp.hpp

@@ -26,11 +26,17 @@
 
 #include <srs_core.hpp>
 
+#include <srs_kernel_buffer.hpp>
+#include <srs_kernel_codec.hpp>
+
 #include <string>
 
 const int kRtpHeaderFixedSize = 12;
 const uint8_t kRtpMarker = 0x80;
 
+// H.264 nalu header type mask.
+const uint8_t kNalTypeMask      = 0x1F;
+
 class SrsBuffer;
 
 class SrsRtpHeader
@@ -74,14 +80,45 @@ class SrsRtpPacket2
 {
 public:
     SrsRtpHeader rtp_header;
+    ISrsEncoder* payload;
+public:
+    SrsRtpPacket2();
+    virtual ~SrsRtpPacket2();
+public:
+    virtual srs_error_t encode(SrsBuffer* buf);
+};
+
+class SrsRtpRawPayload : public ISrsEncoder
+{
+public:
     // @remark We only refer to the memory, user must free it.
     char* payload;
     int nn_payload;
 public:
-    SrsRtpPacket2();
-    virtual ~SrsRtpPacket2();
+    SrsRtpRawPayload();
+    virtual ~SrsRtpRawPayload();
+// interface ISrsEncoder
+public:
+    virtual int nb_bytes();
+    virtual srs_error_t encode(SrsBuffer* buf);
+};
+
+class SrsRtpSTAPPayload : public ISrsEncoder
+{
+public:
+    // The NRI in NALU type.
+    SrsAvcNaluType nri;
+    // The NALU samples.
+    // @remark We only refer to the memory, user must free its bytes.
+    SrsSample* nalus;
+    int nn_nalus;
+public:
+    SrsRtpSTAPPayload();
+    virtual ~SrsRtpSTAPPayload();
+// interface ISrsEncoder
 public:
-    virtual srs_error_t encode(SrsBuffer* stream);
+    virtual int nb_bytes();
+    virtual srs_error_t encode(SrsBuffer* buf);
 };
 
 class SrsRtpSharedPacket