diff --git a/src/Http/HlsPlayer.cpp b/src/Http/HlsPlayer.cpp index 1de45cf8..c6509c0c 100644 --- a/src/Http/HlsPlayer.cpp +++ b/src/Http/HlsPlayer.cpp @@ -71,6 +71,11 @@ void HlsPlayer::teardown() { void HlsPlayer::fetchSegment() { if (_ts_list.empty()) { + // 如果是点播文件，播放列表为空代表文件播放结束，关闭播放器: #2628 + if(!HlsParser::isLive()){ + teardown(); + return; + } //播放列表为空，那么立即重新下载m3u8文件 _timer.reset(); fetchIndexFile(); @@ -121,18 +126,21 @@ void HlsPlayer::fetchSegment() { WarnL << "Download ts segment " << url << " failed:" << err; if (err.getErrCode() == Err_timeout) { strong_self->_timeout_multiple = MAX(strong_self->_timeout_multiple + 1, MAX_TIMEOUT_MULTIPLE); - }else{ - strong_self->_timeout_multiple = MAX(strong_self->_timeout_multiple -1 , MIN_TIMEOUT_MULTIPLE); + } else { + strong_self->_timeout_multiple = MAX(strong_self->_timeout_multiple - 1, MIN_TIMEOUT_MULTIPLE); } } - //提前半秒下载好 - auto delay = duration - ticker.elapsedTime() / 1000.0f - 0.5; - if (delay <= 0) { - //延时最小10ms - delay = 10; + // 提前0.5秒下载好，支持点播文件控制下载速度: #2628 + auto delay = duration - 0.5 - ticker.elapsedTime() / 1000.0f; + if (delay > 2.0) { + // 提前1秒下载 + delay -= 1.0; + } else if (delay <= 0) { + // 延时最小10ms + delay = 0.01; } - //延时下载下一个切片 - strong_self->_timer_ts.reset(new Timer(delay / 1000.0f, [weak_self]() { + // 延时下载下一个切片 + strong_self->_timer_ts.reset(new Timer(delay, [weak_self]() { auto strong_self = weak_self.lock(); if (strong_self) { strong_self->fetchSegment(); diff --git a/tests/test_bench_forward.cpp b/tests/test_bench_forward.cpp new file mode 100644 index 00000000..634d7b8e --- /dev/null +++ b/tests/test_bench_forward.cpp @@ -0,0 +1,267 @@ +/* + * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved. + * + * This file is part of ZLMediaKit(https://github.com/xia-chu/ZLMediaKit). + * + * Use of this source code is governed by MIT license that can be found in the + * LICENSE file in the root of the source tree. All contributing project authors + * may be found in the AUTHORS file in the root of the source tree. + */ + +#include +#include +#include +#include +#include "Util/logger.h" +#include "Util/onceToken.h" +#include "Util/CMD.h" +#include "Util/File.h" +#include "Common/config.h" +#include "Common/Parser.h" +#include "Rtsp/Rtsp.h" +#include "Thread/WorkThreadPool.h" +#include "Pusher/MediaPusher.h" +#include "Player/PlayerProxy.h" + +using namespace std; +using namespace toolkit; +using namespace mediakit; + +class CMD_main : public CMD { +public: + CMD_main() { + _parser.reset(new OptionParser(nullptr)); + + (*_parser) << Option('l',/*该选项简称，如果是\x00则说明无简称*/ + "level",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + to_string(LTrace).data(),/*该选项默认值*/ + false,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "日志等级,LTrace~LError(0~4)",/*该选项说明文字*/ + nullptr); + + + (*_parser) << Option('t',/*该选项简称，如果是\x00则说明无简称*/ + "threads",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + to_string(thread::hardware_concurrency()).data(),/*该选项默认值*/ + false,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "启动事件触发线程数",/*该选项说明文字*/ + nullptr); + + (*_parser) << Option('i',/*该选项简称，如果是\x00则说明无简称*/ + "inputs",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + "/tmp/inputs.txt",/*该选项默认值*/ + false,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "拉流地址配置文件，支持rtmp、rtsp, hls，多个地址以 \"换行符\" 分割",/*该选项说明文字*/ + nullptr); + + (*_parser) << Option('o',/*该选项简称，如果是\x00则说明无简称*/ + "outputs",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + "/tmp/outputs.txt",/*该选项默认值*/ + false,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "推流地址配置文件，支持rtmp、rtsp，多个地址以 \"换行符\" 分割",/*该选项说明文字*/ + nullptr); + + (*_parser) << Option('d',/*该选项简称，如果是\x00则说明无简称*/ + "delay",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + "50",/*该选项默认值*/ + true,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "启动拉流代理间隔,单位毫秒",/*该选项说明文字*/ + nullptr); + + (*_parser) << Option('m',/*该选项简称，如果是\x00则说明无简称*/ + "merge",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + "300",/*该选项默认值*/ + true,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "合并写毫秒,合并写能提高性能",/*该选项说明文字*/ + nullptr); + + (*_parser) << Option('T',/*该选项简称，如果是\x00则说明无简称*/ + "rtp",/*该选项全称,每个选项必须有全称；不得为null或空字符串*/ + Option::ArgRequired,/*该选项后面必须跟值*/ + to_string((int) (Rtsp::RTP_TCP)).data(),/*该选项默认值*/ + true,/*该选项是否必须赋值，如果没有默认值且为ArgRequired时用户必须提供该参数否则将抛异常*/ + "rtsp拉流方式,支持tcp/udp/multicast:0/1/2",/*该选项说明文字*/ + nullptr); + + } + + ~CMD_main() override {} + + const char *description() const override { + return "主程序命令参数"; + } +}; + + +//此程序为zlm的转推性能测试工具，用于测试拉流代理转推性能 +int main(int argc, char *argv[]) { + CMD_main cmd_main; + try { + cmd_main.operator()(argc, argv); + } catch (ExitException &) { + return 0; + } catch (std::exception &ex) { + cout << ex.what() << endl; + return -1; + } + + int threads = cmd_main["threads"]; + LogLevel logLevel = (LogLevel) cmd_main["level"].as(); + logLevel = MIN(MAX(logLevel, LTrace), LError); + auto in_urls = cmd_main["inputs"]; + auto out_urls = cmd_main["outputs"]; + auto rtp_type = cmd_main["rtp"].as(); + auto delay_ms = cmd_main["delay"].as(); + auto merge_ms = cmd_main["merge"].as(); + + //设置日志 + Logger::Instance().add(std::make_shared("ConsoleChannel", logLevel)); + //启动异步日志线程 + Logger::Instance().setWriter(std::make_shared()); + + //设置线程数 + EventPollerPool::setPoolSize(threads); + WorkThreadPool::setPoolSize(threads); + + //设置合并写 + mINI::Instance()[General::kMergeWriteMS] = merge_ms; + + + std::vector input_urls; + std::vector output_urls; + + auto parse_urls = [&]() { + // 获取输入源列表 + auto inputs = ::split(toolkit::File::loadFile(in_urls.c_str()), "\n"); + for(auto &url : inputs){ + if(url.empty() || url.find("://") == std::string::npos) { + continue; + } + auto input_url = ::trim(url); + input_urls.emplace_back(input_url); + } + // 获取输出源列表 + auto outputs = ::split(toolkit::File::loadFile(out_urls.c_str()), "\n"); + for(auto &url : outputs){ + if(url.empty() || url.find("://") == std::string::npos){ + continue; + } + auto output_url = ::trim(url); + output_urls.emplace_back(output_url); + } + + if(input_urls.empty() || input_urls.size() != output_urls.size()){ + return -1; + } + + for(size_t i = 0; i < input_urls.size(); i++){ + InfoL << "拉流地址: " << input_urls[i] << ",推流地址：" << output_urls[i]; + } + return 0; + }; + + if (0 != parse_urls()){ + cout << "请检查inputs和outputs文件是否正确！" << endl; + return -1; + } + + //推流器map + recursive_mutex mtx; + unordered_map proxy_map; + unordered_map pusher_map; + + auto add_pusher = [&](const MediaSource::Ptr &src, const string &url, int index) { + auto pusher = std::make_shared(src); + pusher->setOnCreateSocket([](const EventPoller::Ptr &poller) { + //socket关闭互斥锁，提高性能 + return std::make_shared(poller, false); + }); + //设置推流失败监听 + pusher->setOnPublished([&mtx, &pusher_map, index](const SockException &ex) { + if (ex) { + //推流失败，移除之 + lock_guard lck(mtx); + pusher_map.erase(index); + } + }); + //设置推流中途断开监听 + pusher->setOnShutdown([&mtx, &pusher_map, index](const SockException &ex) { + //推流中途失败，移除之 + lock_guard lck(mtx); + pusher_map.erase(index); + }); + //设置rtsp推流方式(在rtsp推流时有效) + (*pusher)[Client::kRtpType] = rtp_type; + pusher->publish(url); + //保持对象不销毁 + lock_guard lck(mtx); + pusher_map.emplace(index, std::move(pusher)); + //休眠后再启动下一个推流，防止短时间海量链接 + if (delay_ms > 0) { + usleep(1000 * delay_ms); + } + }; + + // 添加转推任务 + for(size_t i = 0; i < input_urls.size(); i++) { + //休眠一秒打印 + sleep(1); + auto schema = findSubString(output_urls[i].data(), nullptr, "://"); + if (schema != RTSP_SCHEMA && schema != RTMP_SCHEMA) { + cout << "推流协议只支持rtsp或rtmp！" << endl; + return -1; + } + ProtocolOption option; + option.enable_ts = false; + option.enable_fmp4 = false; + option.enable_hls = false; + option.enable_mp4 = false; + option.modify_stamp = (int)ProtocolOption::kModifyStampRelative; + //添加拉流代理 + auto proxy = std::make_shared(DEFAULT_VHOST, "app", std::to_string(i), option, -1, nullptr, 1); + //开始拉流代理 + proxy->play(input_urls[i]); + proxy_map.emplace(i, std::move(proxy)); + } + + // 设置退出信号 + static bool exit_flag = false; + signal(SIGINT, [](int) { exit_flag = true; }); + while (!exit_flag) { + //休眠一秒打印 + sleep(1); + + size_t alive_pusher = 0; + { + lock_guard lck(mtx); + alive_pusher = pusher_map.size(); + } + InfoL << "在线转推器个数:" << alive_pusher; + + auto find_pusher = [&](int index){ + lock_guard lck(mtx); + auto it = pusher_map.find(index); + if (it == pusher_map.end()){ + return false; + } + return true; + }; + for(size_t i = 0; i < input_urls.size(); i++) { + if (!find_pusher(i)){ + auto input_url = input_urls[i]; + auto src = MediaSource::find(RTMP_SCHEMA, DEFAULT_VHOST, "app", std::to_string(i), false); + if (src != nullptr){ + add_pusher(src,output_urls[i],i); + } + } + } + } + + return 0; +} \ No newline at end of file