Merge branch 'master' of https://gitee.com/xia-chu/ZLMediaKit into dev

3rdpart/ZLToolKit

@@ -1 +1 @@
-Subproject commit 13916d41e81e1b302f02eb1682b815df5e241c3d
+Subproject commit d5de1d4f0c86139fe8f0b6d956e45dee3b151783

src/Common/MediaSink.cpp

@@ -11,10 +11,13 @@
 #include "MediaSink.h"
 
 //最多等待未初始化的Track 10秒，超时之后会忽略未初始化的Track
-#define MAX_WAIT_MS_READY 10000
+static size_t constexpr kMaxWaitReadyMS= 10000;
 
 //如果直播流只有单Track，最多等待3秒，超时后未收到其他Track的数据，则认为是单Track
-#define MAX_WAIT_MS_ADD_TRACK 3000
+static size_t constexpr kMaxAddTrackMS = 3000;
+
+//如果track未就绪，我们先缓存帧数据，但是有最大个数限制(100帧时大约4秒)，防止内存溢出
+static size_t constexpr kMaxUnreadyFrame = 100;
 
 namespace mediakit{
 
@@ -37,8 +40,14 @@ void MediaSink::addTrack(const Track::Ptr &track_in) {
         if (_all_track_ready) {
             onTrackFrame(frame);
         } else {
+            auto &frame_unread = _frame_unread[frame->getCodecId()];
+            if (frame_unread.size() > kMaxUnreadyFrame) {
+                //未就绪的的track，不能缓存太多的帧，否则可能内存溢出
+                frame_unread.clear();
+                WarnL << "cached frame of unready track(" << frame->getCodecName() << ") is too much, now cleared";
+            }
             //还有Track未就绪，先缓存之
-            _frame_unread[frame->getCodecId()].emplace_back(Frame::getCacheAbleFrame(frame));
+            frame_unread.emplace_back(Frame::getCacheAbleFrame(frame));
         }
     }));
 }
@@ -84,7 +93,7 @@ void MediaSink::checkTrackIfReady(const Track::Ptr &track){
     }
 
     if(!_all_track_ready){
-        if(_ticker.elapsedTime() > MAX_WAIT_MS_READY){
+        if(_ticker.elapsedTime() > kMaxWaitReadyMS){
             //如果超过规定时间，那么不再等待并忽略未准备好的Track
             emitAllTrackReady();
             return;
@@ -101,7 +110,7 @@ void MediaSink::checkTrackIfReady(const Track::Ptr &track){
             return;
         }
 
-        if(_track_map.size() == 1 && _ticker.elapsedTime() > MAX_WAIT_MS_ADD_TRACK){
+        if(_track_map.size() == 1 && _ticker.elapsedTime() > kMaxAddTrackMS){
             //如果只有一个Track，那么在该Track添加后，我们最多还等待若干时间(可能后面还会添加Track)
             emitAllTrackReady();
             return;

src/Extension/Frame.cpp

@@ -180,4 +180,102 @@ const char *CodecInfo::getCodecName() {
 TrackType CodecInfo::getTrackType() {
     return mediakit::getTrackType(getCodecId());
 }
+
+static size_t constexpr kMaxFrameCacheSize = 100;
+
+bool FrameMerger::willFlush(const Frame::Ptr &frame) const{
+    if (_frameCached.empty()) {
+        return false;
+    }
+    switch (_type) {
+        case none : {
+            //frame不是完整的帧，我们合并为一帧
+            bool new_frame = false;
+            switch (frame->getCodecId()) {
+                case CodecH264:
+                case CodecH265: {
+                    //如果是新的一帧，前面的缓存需要输出
+                    new_frame = frame->prefixSize();
+                    break;
+                }
+                default: break;
+            }
+            return new_frame || _frameCached.back()->dts() != frame->dts() || _frameCached.size() > kMaxFrameCacheSize;
+        }
+
+        case mp4_nal_size:
+        case h264_prefix: {
+            if (_frameCached.back()->dts() != frame->dts()) {
+                //时间戳变化了
+                return true;
+            }
+            if (frame->getCodecId() == CodecH264 &&
+                H264_TYPE(frame->data()[frame->prefixSize()]) == H264Frame::NAL_B_P) {
+                //如果是264的b/p帧，那么也刷新输出
+                return true;
+            }
+            return _frameCached.size() > kMaxFrameCacheSize;
+        }
+        default: /*不可达*/ assert(0); return true;
+    }
+}
+
+void FrameMerger::doMerge(BufferLikeString &merged, const Frame::Ptr &frame) const{
+    switch (_type) {
+        case none : {
+            merged.append(frame->data(), frame->size());
+            break;
+        }
+        case h264_prefix: {
+            if (frame->prefixSize()) {
+                merged.append(frame->data(), frame->size());
+            } else {
+                merged.append("\x00\x00\x00\x01", 4);
+                merged.append(frame->data(), frame->size());
+            }
+            break;
+        }
+        case mp4_nal_size: {
+            uint32_t nalu_size = (uint32_t) (frame->size() - frame->prefixSize());
+            nalu_size = htonl(nalu_size);
+            merged.append((char *) &nalu_size, 4);
+            merged.append(frame->data() + frame->prefixSize(), frame->size() - frame->prefixSize());
+            break;
+        }
+        default: /*不可达*/ assert(0); break;
+    }
+}
+
+void FrameMerger::inputFrame(const Frame::Ptr &frame, const onOutput &cb) {
+    if (willFlush(frame)) {
+        Frame::Ptr back = _frameCached.back();
+        Buffer::Ptr merged_frame = back;
+        bool have_idr = back->keyFrame();
+
+        if (_frameCached.size() != 1 || _type == mp4_nal_size) {
+            //在MP4模式下，一帧数据也需要在前添加nalu_size
+            BufferLikeString merged;
+            merged.reserve(back->size() + 1024);
+            _frameCached.for_each([&](const Frame::Ptr &frame) {
+                doMerge(merged, frame);
+                if (frame->keyFrame()) {
+                    have_idr = true;
+                }
+            });
+            merged_frame = std::make_shared<BufferOffset<BufferLikeString> >(std::move(merged));
+        }
+        cb(back->dts(), back->pts(), merged_frame, have_idr);
+        _frameCached.clear();
+    }
+    _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
+}
+
+FrameMerger::FrameMerger(int type) {
+    _type = type;
+}
+
+void FrameMerger::clear() {
+    _frameCached.clear();
+}
+
 }//namespace mediakit

src/Extension/Frame.h

@@ -448,5 +448,32 @@ private:
     Buffer::Ptr _buf;
 };
 
+/**
+ * 合并一些时间戳相同的frame
+ */
+class FrameMerger {
+public:
+    using onOutput = function<void(uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool have_idr)>;
+    enum {
+        none = 0,
+        h264_prefix,
+        mp4_nal_size,
+    };
+
+    FrameMerger(int type);
+    ~FrameMerger() = default;
+
+    void clear();
+    void inputFrame(const Frame::Ptr &frame, const onOutput &cb);
+
+private:
+    bool willFlush(const Frame::Ptr &frame) const;
+    void doMerge(BufferLikeString &buffer, const Frame::Ptr &frame) const;
+
+private:
+    int _type;
+    List<Frame::Ptr> _frameCached;
+};
+
 }//namespace mediakit
 #endif //ZLMEDIAKIT_FRAME_H

src/Record/MP4.cpp

@@ -80,14 +80,6 @@ int mp4_writer_write(mp4_writer_t* mp4, int track, const void* data, size_t byte
     }
 }
 
-int mp4_writer_write_l(mp4_writer_t* mp4, int track, const void* data, size_t bytes, int64_t pts, int64_t dts, int flags, int add_nalu_size){
-    if (mp4->is_fmp4) {
-        return fmp4_writer_write_l(mp4->u.fmp4, track, data, bytes, pts, dts, flags, add_nalu_size);
-    } else {
-        return mov_writer_write_l(mp4->u.mov, track, data, bytes, pts, dts, flags, add_nalu_size);
-    }
-}
-
 int mp4_writer_save_segment(mp4_writer_t* mp4){
     if (mp4->is_fmp4) {
         return fmp4_writer_save_segment(mp4->u.fmp4);

src/Record/MP4Muxer.cpp

@@ -60,7 +60,7 @@ void MP4MuxerInterface::resetTracks() {
     _started = false;
     _have_video = false;
     _mov_writter = nullptr;
-    _frameCached.clear();
+    _frame_merger.clear();
     _codec_to_trackid.clear();
 }
 
@@ -92,47 +92,22 @@ void MP4MuxerInterface::inputFrame(const Frame::Ptr &frame) {
                 break;
             }
         }
+
         case CodecH265: {
             //这里的代码逻辑是让SPS、PPS、IDR这些时间戳相同的帧打包到一起当做一个帧处理，
-            if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
+            _frame_merger.inputFrame(frame, [&](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool have_idr) {
-                Frame::Ptr back = _frameCached.back();
+                track_info.stamp.revise(dts, pts, dts_out, pts_out);
-                //求相对时间戳
+                mp4_writer_write(_mov_writter.get(),
-                track_info.stamp.revise(back->dts(), back->pts(), dts_out, pts_out);
+                                 track_info.track_id,
-
+                                 buffer->data(),
-                if (_frameCached.size() != 1) {
+                                 buffer->size(),
-                    //缓存中有多帧，需要按照mp4格式合并一起
+                                 pts_out,
-                    BufferLikeString merged;
+                                 dts_out,
-                    merged.reserve(back->size() + 1024);
+                                 have_idr ? MOV_AV_FLAG_KEYFREAME : 0);
-                    _frameCached.for_each([&](const Frame::Ptr &frame) {
+            });
-                        uint32_t nalu_size = (uint32_t)(frame->size() - frame->prefixSize());
-                        nalu_size = htonl(nalu_size);
-                        merged.append((char *) &nalu_size, 4);
-                        merged.append(frame->data() + frame->prefixSize(), frame->size() - frame->prefixSize());
-                    });
-                    mp4_writer_write(_mov_writter.get(),
-                                       track_info.track_id,
-                                       merged.data(),
-                                       merged.size(),
-                                       pts_out,
-                                       dts_out,
-                                       back->keyFrame() ? MOV_AV_FLAG_KEYFREAME : 0);
-                } else {
-                    //缓存中只有一帧视频
-                    mp4_writer_write_l(_mov_writter.get(),
-                                       track_info.track_id,
-                                       back->data() + back->prefixSize(),
-                                       back->size() - back->prefixSize(),
-                                       pts_out,
-                                       dts_out,
-                                       back->keyFrame() ? MOV_AV_FLAG_KEYFREAME : 0,
-                                       1/*需要生成头4个字节的MP4格式start code*/);
-                }
-                _frameCached.clear();
-            }
-            //缓存帧，时间戳相同的帧合并一起写入mp4
-            _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
-        }
             break;
+        }
+
         default: {
             track_info.stamp.revise(frame->dts(), frame->pts(), dts_out, pts_out);
             mp4_writer_write(_mov_writter.get(),
@@ -142,8 +117,9 @@ void MP4MuxerInterface::inputFrame(const Frame::Ptr &frame) {
                              pts_out,
                              dts_out,
                              frame->keyFrame() ? MOV_AV_FLAG_KEYFREAME : 0);
-        }
             break;
+        }
+
     }
 }
 

src/Record/MP4Muxer.h

@@ -72,8 +72,8 @@ private:
         int track_id = -1;
         Stamp stamp;
     };
-    List<Frame::Ptr> _frameCached;
     unordered_map<int, track_info> _codec_to_trackid;
+    FrameMerger _frame_merger{FrameMerger::mp4_nal_size};
 };
 
 class MP4Muxer : public MP4MuxerInterface{

src/Record/TsMuxer.cpp

@@ -103,32 +103,14 @@ void TsMuxer::inputFrame(const Frame::Ptr &frame) {
 
         case CodecH265: {
             //这里的代码逻辑是让SPS、PPS、IDR这些时间戳相同的帧打包到一起当做一个帧处理，
-            if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
+            _frame_merger.inputFrame(frame, [&](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool have_idr){
-                Frame::Ptr back = _frameCached.back();
+                track_info.stamp.revise(dts, pts, dts_out, pts_out);
-                Buffer::Ptr merged_frame = back;
-                if (_frameCached.size() != 1) {
-                    BufferLikeString merged;
-                    merged.reserve(back->size() + 1024);
-                    _frameCached.for_each([&](const Frame::Ptr &frame) {
-                        if (frame->prefixSize()) {
-                            merged.append(frame->data(), frame->size());
-                        } else {
-                            merged.append("\x00\x00\x00\x01", 4);
-                            merged.append(frame->data(), frame->size());
-                        }
-                        if (frame->keyFrame()) {
-                            _is_idr_fast_packet = true;
-                        }
-                    });
-                    merged_frame = std::make_shared<BufferOffset<BufferLikeString> >(std::move(merged));
-                }
-                track_info.stamp.revise(back->dts(), back->pts(), dts_out, pts_out);
                 //取视频时间戳为TS的时间戳
-                _timestamp = (uint32_t)dts_out;
+                _timestamp = (uint32_t) dts_out;
-                mpeg_ts_write(_context, track_info.track_id, back->keyFrame() ? 0x0001 : 0, pts_out * 90LL,dts_out * 90LL, merged_frame->data(), merged_frame->size());
+                _is_idr_fast_packet = have_idr;
-                _frameCached.clear();
+                mpeg_ts_write(_context, track_info.track_id, have_idr ? 0x0001 : 0,
-            }
+                              pts_out * 90LL, dts_out * 90LL, buffer->data(), buffer->size());
-            _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
+            });
             break;
         }
 
@@ -141,11 +123,12 @@ void TsMuxer::inputFrame(const Frame::Ptr &frame) {
 
         default: {
             track_info.stamp.revise(frame->dts(), frame->pts(), dts_out, pts_out);
-            if(!_have_video){
+            if (!_have_video) {
                 //没有视频时，才以音频时间戳为TS的时间戳
-                _timestamp = (uint32_t)dts_out;
+                _timestamp = (uint32_t) dts_out;
             }
-            mpeg_ts_write(_context, track_info.track_id, frame->keyFrame() ? 0x0001 : 0, pts_out * 90LL, dts_out * 90LL, frame->data(), frame->size());
+            mpeg_ts_write(_context, track_info.track_id, frame->keyFrame() ? 0x0001 : 0,
+                          pts_out * 90LL, dts_out * 90LL, frame->data(), frame->size());
             break;
         }
     }
@@ -187,6 +170,7 @@ void TsMuxer::uninit() {
         _context = nullptr;
     }
     _codec_to_trackid.clear();
+    _frame_merger.clear();
 }
 
 }//namespace mediakit

src/Record/TsMuxer.h

@@ -59,6 +59,8 @@ private:
     void stampSync();
 
 private:
+    bool _have_video = false;
+    bool _is_idr_fast_packet = false;
     void *_context = nullptr;
     char _tsbuf[188];
     uint32_t _timestamp = 0;
@@ -67,9 +69,7 @@ private:
         Stamp stamp;
     };
     unordered_map<int, track_info> _codec_to_trackid;
-    List<Frame::Ptr> _frameCached;
+    FrameMerger _frame_merger{FrameMerger::h264_prefix};
-    bool _is_idr_fast_packet = false;
-    bool _have_video = false;
 };
 
 }//namespace mediakit

src/Rtp/Decoder.cpp

@@ -100,34 +100,6 @@ static const char *getCodecName(int codec_id) {
     }
 }
 
-void FrameMerger::inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb){
-    bool flush = false;
-    switch (frame->getCodecId()) {
-        case CodecH264:
-        case CodecH265:{
-            //如果是新的一帧，前面的缓存需要输出
-            flush = frame->prefixSize();
-            break;
-        }
-        default: break;
-    }
-    if (!_frameCached.empty() && (flush || _frameCached.back()->dts() != frame->dts())) {
-        Frame::Ptr back = _frameCached.back();
-        Buffer::Ptr merged_frame = back;
-        if(_frameCached.size() != 1){
-            BufferLikeString merged;
-            merged.reserve(back->size() + 1024);
-            _frameCached.for_each([&](const Frame::Ptr &frame){
-                merged.append(frame->data(),frame->size());
-            });
-            merged_frame = std::make_shared<BufferOffset<BufferLikeString> >(std::move(merged));
-        }
-        cb(back->dts(),back->pts(),merged_frame);
-        _frameCached.clear();
-    }
-    _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
-}
-
 void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t bytes, int finish){
     switch (codecid) {
         case PSI_STREAM_H264: {
@@ -188,7 +160,7 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
     switch (codecid) {
         case PSI_STREAM_H264: {
             auto frame = std::make_shared<H264FrameNoCacheAble>((char *) data, bytes, (uint32_t)dts, (uint32_t)pts, prefixSize((char *) data, bytes));
-            _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
+            _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool) {
                 onFrame(std::make_shared<FrameWrapper<H264FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
             });
             break;
@@ -196,7 +168,7 @@ void DecoderImp::onDecode(int stream,int codecid,int flags,int64_t pts,int64_t d
 
         case PSI_STREAM_H265: {
             auto frame = std::make_shared<H265FrameNoCacheAble>((char *) data, bytes, (uint32_t)dts, (uint32_t)pts, prefixSize((char *) data, bytes));
-            _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer) {
+            _merger.inputFrame(frame,[this](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool) {
                 onFrame(std::make_shared<FrameWrapper<H265FrameNoCacheAble> >(buffer, dts, pts, prefixSize(buffer->data(), buffer->size()), 0));
             });
             break;

src/Rtp/Decoder.h

@@ -34,18 +34,6 @@ protected:
     virtual ~Decoder() = default;
 };
 
-/**
- * 合并一些时间戳相同的frame
- */
-class FrameMerger {
-public:
-    FrameMerger() = default;
-    ~FrameMerger() = default;
-    void inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb);
-private:
-    List<Frame::Ptr> _frameCached;
-};
-
 class DecoderImp{
 public:
     typedef enum {
@@ -71,7 +59,7 @@ private:
 private:
     Decoder::Ptr _decoder;
     MediaSinkInterface *_sink;
-    FrameMerger _merger;
+    FrameMerger _merger{FrameMerger::none};
     Ticker _last_unsported_print;
 };
 

src/Rtp/PSEncoder.cpp

@@ -123,33 +123,18 @@ void PSEncoder::inputFrame(const Frame::Ptr &frame) {
                 break;
             }
         }
+
         case CodecH265: {
             //这里的代码逻辑是让SPS、PPS、IDR这些时间戳相同的帧打包到一起当做一个帧处理，
-            if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
+            _frame_merger.inputFrame(frame, [&](uint32_t dts, uint32_t pts, const Buffer::Ptr &buffer, bool have_idr) {
-                Frame::Ptr back = _frameCached.back();
+                track_info.stamp.revise(dts, pts, dts_out, pts_out);
-                Buffer::Ptr merged_frame = back;
+                //取视频时间戳为TS的时间戳
-                if (_frameCached.size() != 1) {
+                _timestamp = (uint32_t) dts_out;
-                    BufferLikeString merged;
+                ps_muxer_input(_muxer.get(), track_info.track_id, have_idr ? 0x0001 : 0,
-                    merged.reserve(back->size() + 1024);
+                               pts_out * 90LL, dts_out * 90LL, buffer->data(), buffer->size());
-                    _frameCached.for_each([&](const Frame::Ptr &frame) {
+            });
-                        if (frame->prefixSize()) {
-                            merged.append(frame->data(), frame->size());
-                        } else {
-                            merged.append("\x00\x00\x00\x01", 4);
-                            merged.append(frame->data(), frame->size());
-                        }
-                    });
-                    merged_frame = std::make_shared<BufferOffset<BufferLikeString> >(std::move(merged));
-                }
-                track_info.stamp.revise(back->dts(), back->pts(), dts_out, pts_out);
-                _timestamp = (uint32_t)dts_out;
-                ps_muxer_input(_muxer.get(), track_info.track_id, back->keyFrame() ? 0x0001 : 0, pts_out * 90LL,
-                               dts_out * 90LL, merged_frame->data(), merged_frame->size());
-                _frameCached.clear();
-            }
-            _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
-        }
             break;
+        }
 
         case CodecAAC: {
             if (frame->prefixSize() == 0) {
@@ -160,11 +145,11 @@ void PSEncoder::inputFrame(const Frame::Ptr &frame) {
 
         default: {
             track_info.stamp.revise(frame->dts(), frame->pts(), dts_out, pts_out);
-            _timestamp = (uint32_t)dts_out;
+            _timestamp = (uint32_t) dts_out;
             ps_muxer_input(_muxer.get(), track_info.track_id, frame->keyFrame() ? 0x0001 : 0, pts_out * 90LL,
                            dts_out * 90LL, frame->data(), frame->size());
-        }
             break;
+        }
     }
 }
 

src/Rtp/PSEncoder.h

@@ -61,9 +61,9 @@ private:
 private:
     uint32_t _timestamp = 0;
     BufferRaw::Ptr _buffer;
-    List<Frame::Ptr> _frameCached;
     std::shared_ptr<struct ps_muxer_t> _muxer;
     unordered_map<int, track_info> _codec_to_trackid;
+    FrameMerger _frame_merger{FrameMerger::h264_prefix};
 };
 
 class PSEncoderImp : public PSEncoder{

tests/test_player.cpp

@@ -23,41 +23,6 @@ using namespace std;
 using namespace toolkit;
 using namespace mediakit;
 
-
-/**
- * 合并一些时间戳相同的frame
- */
-class FrameMerger {
-public:
-    FrameMerger() = default;
-    virtual ~FrameMerger() = default;
-
-    void inputFrame(const Frame::Ptr &frame,const function<void(uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer)> &cb){
-        if (!_frameCached.empty() && _frameCached.back()->dts() != frame->dts()) {
-            Frame::Ptr back = _frameCached.back();
-            Buffer::Ptr merged_frame = back;
-            if(_frameCached.size() != 1){
-                string merged;
-                _frameCached.for_each([&](const Frame::Ptr &frame){
-                    if(frame->prefixSize()){
-                        merged.append(frame->data(),frame->size());
-                    } else{
-                        merged.append("\x00\x00\x00\x01",4);
-                        merged.append(frame->data(),frame->size());
-                    }
-                });
-                merged_frame = std::make_shared<BufferString>(std::move(merged));
-            }
-            cb(back->dts(),back->pts(),merged_frame);
-            _frameCached.clear();
-        }
-        _frameCached.emplace_back(Frame::getCacheAbleFrame(frame));
-    }
-private:
-    List<Frame::Ptr> _frameCached;
-};
-
-
 #ifdef WIN32
 #include <TCHAR.h>
 
@@ -129,7 +94,7 @@ int main(int argc, char *argv[]) {
                     displayer.set<YuvDisplayer>(nullptr,url);
                 }
                 if(!merger){
-                    merger.set<FrameMerger>();
+                    merger.set<FrameMerger>(FrameMerger::h264_prefix);
                 }
                 merger.get<FrameMerger>().inputFrame(frame,[&](uint32_t dts,uint32_t pts,const Buffer::Ptr &buffer){
                     AVFrame *pFrame = nullptr;