Merge remote-tracking branch 'origin/master' into dev

3rdpart/ZLToolKit

@@ -1 +1 @@
-Subproject commit 6a8bdfc80db3930d2ba8067d06db467dad8baddf
+Subproject commit e5535a7164f55eb9062213f40ddc68c0294e6f57

CMakeLists.txt

@@ -370,7 +370,7 @@ if(ENABLE_JEMALLOC_STATIC)
     set(ENABLE_JEMALLOC_STAT OFF)
   endif ()
   include(Jemalloc)
-  include_directories(SYSTEM ${DEP_ROOT_DIR}/${JEMALLOC_NAME}/include/jemalloc)
+  include_directories(SYSTEM ${DEP_ROOT_DIR}/${JEMALLOC_NAME}/include)
   link_directories(${DEP_ROOT_DIR}/${JEMALLOC_NAME}/lib)
   # 用于影响后续查找过程
   # Used to affect subsequent lookup process

conf/config.ini

@@ -135,6 +135,8 @@ segDur=2
 #m3u8索引中,hls保留切片个数(实际保留切片个数大2~3个)
 #如果设置为0，则不删除切片，而是保存为点播
 segNum=3
+#HLS切片延迟个数，大于0将生成hls_delay.m3u8文件，0则不生成
+segDelay=0
 #HLS切片从m3u8文件中移除后，继续保留在磁盘上的个数
 segRetain=5
 #是否广播 hls切片(ts/fmp4)完成通知(on_record_ts)

src/Common/config.cpp

@@ -307,6 +307,7 @@ namespace Hls {
 const string kSegmentDuration = HLS_FIELD "segDur";
 const string kSegmentNum = HLS_FIELD "segNum";
 const string kSegmentKeep = HLS_FIELD "segKeep";
+const string kSegmentDelay = HLS_FIELD "segDelay";
 const string kSegmentRetain = HLS_FIELD "segRetain";
 const string kFileBufSize = HLS_FIELD "fileBufSize";
 const string kBroadcastRecordTs = HLS_FIELD "broadcastRecordTs";
@@ -317,6 +318,7 @@ static onceToken token([]() {
     mINI::Instance()[kSegmentDuration] = 2;
     mINI::Instance()[kSegmentNum] = 3;
     mINI::Instance()[kSegmentKeep] = false;
+    mINI::Instance()[kSegmentDelay] = 0;
     mINI::Instance()[kSegmentRetain] = 5;
     mINI::Instance()[kFileBufSize] = 64 * 1024;
     mINI::Instance()[kBroadcastRecordTs] = false;

src/Common/config.h

@@ -349,6 +349,8 @@ extern const std::string kSegmentDuration;
 extern const std::string kSegmentNum;
 // 如果设置为0，则不保留切片，设置为1则一直保留切片
 extern const std::string kSegmentKeep;
+// HLS切片延迟个数，大于0将生成hls_delay.m3u8文件，0则不生成
+extern const std::string kSegmentDelay;
 // HLS切片从m3u8文件中移除后，继续保留在磁盘上的个数
 extern const std::string kSegmentRetain;
 // HLS文件写缓存大小

src/Extension/Frame.cpp

@@ -69,12 +69,16 @@ CodecId getCodecByMovId(int object_id) {
     if (object_id == MOV_OBJECT_NONE) {
         return CodecInvalid;
     }
-    switch (object_id) {
+
-#define XX(name, type, value, str, mpeg_id, mp4_id) case mp4_id : return name;
+#define XX(name, type, value, str, mpeg_id, mp4_id) { mp4_id, name },
-        CODEC_MAP(XX)
+    static map<int, CodecId> s_map = { CODEC_MAP(XX) };
 #undef XX
-        default : WarnL << "Unsupported mov: " << object_id; return CodecInvalid;
+    auto it = s_map.find(object_id);
+    if (it == s_map.end()) {
+        WarnL << "Unsupported mov: " << object_id;
+        return CodecInvalid;
     }
+    return it->second;
 }
 #endif
 
@@ -89,17 +93,20 @@ int getMpegIdByCodec(CodecId codec) {
 }
 
 CodecId getCodecByMpegId(int mpeg_id) {
-    if (mpeg_id == PSI_STREAM_RESERVED) {
+    if (mpeg_id == PSI_STREAM_RESERVED || mpeg_id == 0xBD) {
+        // 海康的 PS 流中会有0xBD 的包
         return CodecInvalid;
     }
-    switch (mpeg_id) {
+
-#define XX(name, type, value, str, mpeg_id, mp4_id) case mpeg_id : return name;
+#define XX(name, type, value, str, mpeg_id, mp4_id) { mpeg_id, name },
-        CODEC_MAP(XX)
+    static map<int, CodecId> s_map = { CODEC_MAP(XX) };
 #undef XX
-        // 海康的 PS 流中会有0xBD 的包
+    auto it = s_map.find(mpeg_id);
-        case 0xBD: return CodecInvalid;
+    if (it == s_map.end()) {
-        default : WarnL << "Unsupported mpeg: " << mpeg_id; return CodecInvalid;
+        WarnL << "Unsupported mpeg: " << mpeg_id;
+        return CodecInvalid;
     }
+    return it->second;
 }
 
 #endif

src/Extension/Track.h

@@ -129,8 +129,8 @@ public:
         _fps = fps;
     }
 
-    int getVideoHeight() const override { return _width; }
+    int getVideoWidth() const override { return _width; }
-    int getVideoWidth() const override { return _height; }
+    int getVideoHeight() const override { return _height; }
     float getVideoFps() const override { return _fps; }
     bool ready() const override { return true; }
 

src/Record/HlsMaker.cpp

@@ -24,15 +24,40 @@ HlsMaker::HlsMaker(bool is_fmp4, float seg_duration, uint32_t seg_number, bool s
     _seg_keep = seg_keep;
 }
 
-void HlsMaker::makeIndexFile(bool eof) {
+void HlsMaker::makeIndexFile(bool include_delay, bool eof) {
+    GET_CONFIG(uint32_t, segDelay, Hls::kSegmentDelay);
+    GET_CONFIG(uint32_t, segRetain, Hls::kSegmentRetain);
+    std::deque<std::tuple<int, std::string>> temp(_seg_dur_list);
+    if (!include_delay) {
+        while (temp.size() > _seg_number) {
+            temp.pop_front();
+        }
+    }
     int maxSegmentDuration = 0;
-    for (auto &tp : _seg_dur_list) {
+    for (auto &tp : temp) {
         int dur = std::get<0>(tp);
         if (dur > maxSegmentDuration) {
             maxSegmentDuration = dur;
         }
     }
-    auto index_seq = _seg_number ? (_file_index > _seg_number ? _file_index - _seg_number : 0LL) : 0LL;
+    uint64_t index_seq;
+    if (_seg_number) {
+        if (include_delay) {
+            if (_file_index > _seg_number + segDelay) {
+                index_seq = _file_index - _seg_number - segDelay;
+            } else {
+                index_seq = 0LL;
+            }
+        } else {
+            if (_file_index > _seg_number) {
+                index_seq = _file_index - _seg_number;
+            } else {
+                index_seq = 0LL;
+            }
+        }
+    } else {
+        index_seq = 0LL;
+    }
 
     string index_str;
     index_str.reserve(2048);
@@ -50,7 +75,7 @@ void HlsMaker::makeIndexFile(bool eof) {
     }
 
     stringstream ss;
-    for (auto &tp : _seg_dur_list) {
+    for (auto &tp : temp) {
         ss << "#EXTINF:" << std::setprecision(3) << std::get<0>(tp) / 1000.0 << ",\n" << std::get<1>(tp) << "\n";
     }
     index_str += ss.str();
@@ -58,7 +83,7 @@ void HlsMaker::makeIndexFile(bool eof) {
     if (eof) {
         index_str += "#EXT-X-ENDLIST\n";
     }
-    onWriteHls(index_str);
+    onWriteHls(index_str, include_delay);
 }
 
 void HlsMaker::inputInitSegment(const char *data, size_t len) {
@@ -91,12 +116,13 @@ void HlsMaker::inputData(const char *data, size_t len, uint64_t timestamp, bool
 }
 
 void HlsMaker::delOldSegment() {
+    GET_CONFIG(uint32_t, segDelay, Hls::kSegmentDelay);
     if (_seg_number == 0) {
         //如果设置为保留0个切片，则认为是保存为点播
         return;
     }
     //在hls m3u8索引文件中,我们保存的切片个数跟_seg_number相关设置一致
-    if (_file_index > _seg_number) {
+    if (_file_index > _seg_number + segDelay) {
         _seg_dur_list.pop_front();
     }
     //如果设置为一直保存，就不删除
@@ -105,8 +131,8 @@ void HlsMaker::delOldSegment() {
     }
     GET_CONFIG(uint32_t, segRetain, Hls::kSegmentRetain);
     //但是实际保存的切片个数比m3u8所述多若干个,这样做的目的是防止播放器在切片删除前能下载完毕
-    if (_file_index > _seg_number + segRetain) {
+    if (_file_index > _seg_number + segDelay + segRetain) {
-        onDelSegment(_file_index - _seg_number - segRetain - 1);
+        onDelSegment(_file_index - _seg_number - segDelay - segRetain - 1);
     }
 }
 
@@ -125,6 +151,7 @@ void HlsMaker::addNewSegment(uint64_t stamp) {
 }
 
 void HlsMaker::flushLastSegment(bool eof){
+    GET_CONFIG(uint32_t, segDelay, Hls::kSegmentDelay);
     if (_last_file_name.empty()) {
         //不存在上个切片
         return;
@@ -139,7 +166,11 @@ void HlsMaker::flushLastSegment(bool eof){
     //先flush ts切片，否则可能存在ts文件未写入完毕就被访问的情况
     onFlushLastSegment(seg_dur);
     //然后写m3u8文件
-    makeIndexFile(eof);
+    makeIndexFile(false, eof);
+    //写入切片延迟的m3u8文件
+    if (segDelay) {
+        makeIndexFile(true, eof);
+    }
 }
 
 bool HlsMaker::isLive() const {

src/Record/HlsMaker.h

@@ -96,7 +96,7 @@ protected:
     /**
      * 写m3u8文件回调
      */
-    virtual void onWriteHls(const std::string &data) = 0;
+    virtual void onWriteHls(const std::string &data, bool include_delay) = 0;
 
     /**
      * 上一个 ts 切片写入完成, 可在这里进行通知处理
@@ -115,7 +115,7 @@ private:
      * 生成m3u8文件
      * @param eof true代表点播
      */
-    void makeIndexFile(bool eof = false);
+    void makeIndexFile(bool include_delay, bool eof = false);
 
     /**
      * 删除旧的ts切片

src/Record/HlsMakerImp.cpp

@@ -21,11 +21,20 @@ using namespace toolkit;
 
 namespace mediakit {
 
+std::string getDelayPath(const std::string& originalPath) {
+    std::size_t pos = originalPath.find(".m3u8");
+    if (pos != std::string::npos) {
+        return originalPath.substr(0, pos) + "_delay.m3u8";
+    }
+    return originalPath;
+}
+
 HlsMakerImp::HlsMakerImp(bool is_fmp4, const string &m3u8_file, const string &params, uint32_t bufSize, float seg_duration,
                          uint32_t seg_number, bool seg_keep) : HlsMaker(is_fmp4, seg_duration, seg_number, seg_keep) {
     _poller = EventPollerPool::Instance().getPoller();
     _path_prefix = m3u8_file.substr(0, m3u8_file.rfind('/'));
     _path_hls = m3u8_file;
+    _path_hls_delay = getDelayPath(m3u8_file);
     _params = params;
     _buf_size = bufSize;
     _file_buf.reset(new char[bufSize], [](char *ptr) { delete[] ptr; });
@@ -62,6 +71,7 @@ void HlsMakerImp::clearCache(bool immediately, bool eof) {
     {
         std::list<std::string> lst;
         lst.emplace_back(_path_hls);
+        lst.emplace_back(_path_hls_delay);
         if (!_path_init.empty()) {
             lst.emplace_back(_path_init);
         }
@@ -146,16 +156,17 @@ void HlsMakerImp::onWriteSegment(const char *data, size_t len) {
     }
 }
 
-void HlsMakerImp::onWriteHls(const std::string &data) {
+void HlsMakerImp::onWriteHls(const std::string &data, bool include_delay) {
-    auto hls = makeFile(_path_hls);
+    auto path = include_delay ? _path_hls_delay : _path_hls;
+    auto hls = makeFile(path);
     if (hls) {
         fwrite(data.data(), data.size(), 1, hls.get());
         hls.reset();
-        if (_media_src) {
+        if (_media_src && !include_delay) {
             _media_src->setIndexFile(data);
         }
     } else {
-        WarnL << "Create hls file failed," << _path_hls << " " << get_uv_errmsg();
+        WarnL << "Create hls file failed," << path << " " << get_uv_errmsg();
     }
 }
 

src/Record/HlsMakerImp.h

@@ -49,7 +49,7 @@ protected:
     void onDelSegment(uint64_t index) override;
     void onWriteInitSegment(const char *data, size_t len) override;
     void onWriteSegment(const char *data, size_t len) override;
-    void onWriteHls(const std::string &data) override;
+    void onWriteHls(const std::string &data, bool include_delay) override;
     void onFlushLastSegment(uint64_t duration_ms) override;
 
 private:
@@ -60,6 +60,7 @@ private:
     int _buf_size;
     std::string _params;
     std::string _path_hls;
+    std::string _path_hls_delay;
     std::string _path_init;
     std::string _path_prefix;
     RecordInfo _info;

src/Rtcp/RtcpFCI.cpp

@@ -210,7 +210,7 @@ string FCI_NACK::dumpString() const {
 }
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-
+#pragma pack(push, 1)
 class RunLengthChunk {
 public:
     static size_t constexpr kSize = 2;
@@ -241,6 +241,7 @@ public:
     // 打印本对象
     string dumpString() const;
 };
+#pragma pack(pop)
 
 RunLengthChunk::RunLengthChunk(SymbolStatus status, uint16_t run_length) {
     type = 0;
@@ -261,7 +262,7 @@ string RunLengthChunk::dumpString() const {
 }
 
 ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-
+#pragma pack(push, 1)
 class StatusVecChunk {
 public:
     static size_t constexpr kSize = 2;
@@ -292,6 +293,7 @@ public:
     // 打印本对象
     string dumpString() const;
 };
+#pragma pack(pop)
 
 StatusVecChunk::StatusVecChunk(bool symbol_bit, const vector<SymbolStatus> &status) {
     CHECK(status.size() << symbol_bit <= 14);

src/Rtcp/RtcpFCI.h

@@ -14,6 +14,7 @@
 #include "Rtcp.h"
 
 namespace mediakit {
+#pragma pack(push, 1)
 
 /////////////////////////////////////////// PSFB ////////////////////////////////////////////////////
 
@@ -375,6 +376,6 @@ private:
     // feedback packet count,反馈包号,本包是第几个transport-cc包，每次加1                          |
     uint8_t fb_pkt_count;
 };
-
+#pragma pack(pop)
 } // namespace mediakit
 #endif // ZLMEDIAKIT_RTCPFCI_H

tests/test_bench_forward.cpp

@@ -180,7 +180,7 @@ int main(int argc, char *argv[]) {
         auto pusher = std::make_shared<MediaPusher>(src);
         pusher->setOnCreateSocket([](const EventPoller::Ptr &poller) {
             //socket关闭互斥锁，提高性能	
-            return std::make_shared<Socket>(poller, false);
+            return Socket::createSocket(poller, false);
         });
         //设置推流失败监听	
         pusher->setOnPublished([&mtx, &pusher_map, index](const SockException &ex) {

tests/test_bench_pull.cpp

@@ -123,7 +123,7 @@ int main(int argc, char *argv[]) {
         auto tag = player.get();
         player->setOnCreateSocket([](const EventPoller::Ptr &poller) {
             //socket关闭互斥锁，提高性能
-            return std::make_shared<Socket>(poller, false);
+            return Socket::createSocket(poller, false);
         });
         //设置播放失败监听
         player->setOnPlayResult([&mtx, &player_map, tag](const SockException &ex) {

tests/test_bench_push.cpp

@@ -166,7 +166,7 @@ int main(int argc, char *argv[]) {
         auto tag = pusher.get();
         pusher->setOnCreateSocket([](const EventPoller::Ptr &poller) {
             //socket关闭互斥锁，提高性能
-            return std::make_shared<Socket>(poller, false);
+            return Socket::createSocket(poller, false);
         });
         //设置推流失败监听
         pusher->setOnPublished([&mtx, &pusher_map, tag](const SockException &ex) {

tests/test_rtp.cpp

@@ -10,11 +10,8 @@
 
 #include <map>
 #include <iostream>
-#include "Util/MD5.h"
-#include "Util/File.h"
-#include "Util/logger.h"
-#include "Util/SSLBox.h"
 #include "Util/util.h"
+#include "Util/logger.h"
 #include "Network/TcpServer.h"
 #include "Common/config.h"
 #include "Rtsp/RtspSession.h"
@@ -30,66 +27,73 @@ static semaphore sem;
 
 #if defined(ENABLE_RTPPROXY)
 static bool loadFile(const char *path, const EventPoller::Ptr &poller) {
-    FILE *fp = fopen(path, "rb");
+    std::shared_ptr<FILE> fp(fopen(path, "rb"), [](FILE *fp) {
+        sem.post();
+        if (fp) {
+            fclose(fp);
+        }
+    });
     if (!fp) {
         WarnL << "open file failed:" << path;
         return false;
     }
 
-    uint64_t timeStamp_last = 0;
-    uint16_t len;
-    char rtp[0xFFFF];
     struct sockaddr_storage addr;
     memset(&addr, 0, sizeof(addr));
     addr.ss_family = AF_INET;
     auto sock = Socket::createSocket(poller);
-    size_t total_size = 0;
+    auto process = RtpSelector::Instance().getProcess("test", true);
-    RtpProcess::Ptr process;
+
-    uint32_t ssrc = 0;
+    uint64_t stamp_last = 0;
+    auto total_size = std::make_shared<size_t>(0);
+    auto do_read = [fp, total_size, sock, addr, process, stamp_last]() mutable -> int {
+        uint16_t len;
+        char rtp[0xFFFF];
         while (true) {
-        if (2 != fread(&len, 1, 2, fp)) {
+            if (2 != fread(&len, 1, 2, fp.get())) {
-            WarnL;
+                WarnL << "Read rtp size failed";
-            break;
+                // 重新播放
+                fseek(fp.get(), 0, SEEK_SET);
+                return 1;
             }
             len = ntohs(len);
             if (len < 12 || len > sizeof(rtp)) {
-            WarnL << len;
+                WarnL << "Invalid rtp size: " << len;
-            break;
+                return 0;
             }
 
-        if (len != fread(rtp, 1, len, fp)) {
+            if (len != fread(rtp, 1, len, fp.get())) {
-            WarnL;
+                WarnL << "Read rtp data failed";
-            break;
+                return 0;
             }
-        total_size += len;
+            (*total_size) += len;
-        uint64_t timeStamp = 0;
+            uint64_t stamp = 0;
-
-        if (!process) {
-            if (!RtpSelector::getSSRC(rtp, len, ssrc)) {
-                WarnL << "get ssrc from rtp failed:" << len;
-                return false;
-            }
-            process = RtpSelector::Instance().getProcess(printSSRC(ssrc), true);
-        }
-        if (process) {
             try {
-                process->inputRtp(true, sock, rtp, len, (struct sockaddr *)&addr, &timeStamp);
+                process->inputRtp(true, sock, rtp, len, (struct sockaddr *)&addr, &stamp);
-            } catch (...) {
+            } catch (std::exception &ex) {
-                RtpSelector::Instance().delProcess(printSSRC(ssrc), process.get());
+                WarnL << "Input rtp failed: " << ex.what();
-                throw;
+                return 0;
-            }
             }
 
-        auto diff = timeStamp - timeStamp_last;
+            auto diff = stamp - stamp_last;
-        if (diff > 0 && diff < 500) {
+            if (diff < 0 || diff > 500) {
-            usleep(diff * 1000);
+                diff = 1;
-        } else {
-            usleep(1 * 1000);
             }
-        timeStamp_last = timeStamp;
+            if (diff) {
+                stamp_last = stamp;
+                return diff;
             }
-    WarnL << total_size / 1024 << "KB";
+        }
-    fclose(fp);
+    };
+    poller->doDelayTask(1, [do_read, total_size, process]() mutable {
+        auto ret = do_read();
+        if (!ret) {
+            WarnL << *total_size / 1024 << "KB";
+            RtpSelector::Instance().delProcess("test", process.get());
+        }
+        return ret;
+    });
+
     return true;
 }
 #endif // #if defined(ENABLE_RTPPROXY)
@@ -111,20 +115,13 @@ int main(int argc,char *argv[]) {
     // mINI::Instance()[RtpProxy::kDumpDir] = "/Users/xzl/Desktop/";
 
     if (argc == 2) {
-        auto poller = EventPollerPool::Instance().getPoller();
+        loadFile(argv[1], EventPollerPool::Instance().getPoller());
-        poller->async_first([poller,argv](){
-            loadFile(argv[1],poller);
-            sem.post();
-        });
         sem.wait();
-        sleep(1);
+    } else {
-    }
-    else
         ErrorL << "parameter error.";
+    }
 #else
     ErrorL << "please ENABLE_RTPPROXY and then test";
 #endif // #if defined(ENABLE_RTPPROXY)
     return 0;
 }
-
-

webrtc/SctpAssociation.cpp

@@ -677,9 +677,9 @@ namespace RTC
                         if (notification->sn_header.sn_length > 0)
                         {
                             static const size_t BufferSize{ 1024 };
-                            static char buffer[BufferSize];
+                            thread_local static char buffer[BufferSize];
 
-                            uint32_t len = notification->sn_header.sn_length;
+                            uint32_t len = notification->sn_assoc_change.sac_length - sizeof(struct sctp_assoc_change);
 
                             for (uint32_t i{ 0 }; i < len; ++i)
                             {
@@ -745,9 +745,9 @@ namespace RTC
                         if (notification->sn_header.sn_length > 0)
                         {
                             static const size_t BufferSize{ 1024 };
-                            static char buffer[BufferSize];
+                            thread_local static char buffer[BufferSize];
 
-                            uint32_t len = notification->sn_header.sn_length;
+                            uint32_t len = notification->sn_assoc_change.sac_length - sizeof(struct sctp_assoc_change);
 
                             for (uint32_t i{ 0 }; i < len; ++i)
                             {
@@ -786,7 +786,7 @@ namespace RTC
             case SCTP_REMOTE_ERROR:
             {
                 static const size_t BufferSize{ 1024 };
-                static char buffer[BufferSize];
+                thread_local static char buffer[BufferSize];
 
                 uint32_t len = notification->sn_remote_error.sre_length - sizeof(struct sctp_remote_error);
 
@@ -822,7 +822,7 @@ namespace RTC
             case SCTP_SEND_FAILED_EVENT:
             {
                 static const size_t BufferSize{ 1024 };
-                static char buffer[BufferSize];
+                thread_local static char buffer[BufferSize];
 
                 uint32_t len =
                   notification->sn_send_failed_event.ssfe_length - sizeof(struct sctp_send_failed_event);

webrtc/WebRtcTransport.cpp

@@ -59,8 +59,8 @@ static onceToken token([]() {
     mINI::Instance()[kTimeOutSec] = 15;
     mINI::Instance()[kExternIP] = "";
     mINI::Instance()[kRembBitRate] = 0;
-    mINI::Instance()[kPort] = 0;
+    mINI::Instance()[kPort] = 8000;
-    mINI::Instance()[kTcpPort] = 0;
+    mINI::Instance()[kTcpPort] = 8000;
 
     mINI::Instance()[kStartBitrate] = 0;
     mINI::Instance()[kMaxBitrate] = 0;