amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-23 03:10:04 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

format code and remove some useless code

Browse Source
This commit is contained in:
xiongguangjie 2022-06-06 20:40:04 +08:00
parent 83d75c9a72
commit c92fc8a4a8
19 changed files with 715 additions and 762 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
srt/Ack.cpp
View File

@ -74,10 +74,10 @@ bool ACKPacket::storeToData() {
std::string ACKPacket::dump() { std::string ACKPacket::dump() {
_StrPrinter printer; _StrPrinter printer;
printer << "last_ack_pkt_seq_number="<<last_ack_pkt_seq_number<<\ printer << "last_ack_pkt_seq_number=" << last_ack_pkt_seq_number << " rtt=" << rtt
" rtt="<<rtt<<" rtt_variance="<<rtt_variance<<\ << " rtt_variance=" << rtt_variance << " pkt_recv_rate=" << pkt_recv_rate
" pkt_recv_rate="<<pkt_recv_rate<<" available_buf_size="<<available_buf_size<<\ << " available_buf_size=" << available_buf_size << " estimated_link_capacity=" << estimated_link_capacity
" estimated_link_capacity="<<estimated_link_capacity<<" recv_rate="<<recv_rate; << " recv_rate=" << recv_rate;
return std::move(printer); return std::move(printer);
} }
} // namespace } // namespace SRT

10
srt/Ack.hpp
View File

@ -2,7 +2,6 @@
#define ZLMEDIAKIT_SRT_ACK_H #define ZLMEDIAKIT_SRT_ACK_H
#include "Packet.hpp" #include "Packet.hpp"
namespace SRT { namespace SRT {
/* /*
0 1 2 3 0 1 2 3
@ -33,16 +32,13 @@ namespace SRT{
Figure 13: ACK control packet Figure 13: ACK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-ack-acknowledgment https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-ack-acknowledgment
*/ */
class ACKPacket : public ControlPacket class ACKPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<ACKPacket>; using Ptr = std::shared_ptr<ACKPacket>;
ACKPacket() = default; ACKPacket() = default;
~ACKPacket() = default; ~ACKPacket() = default;
enum{ enum { ACK_CIF_SIZE = 7 * 4 };
ACK_CIF_SIZE = 7*4
};
std::string dump(); std::string dump();
///////ControlPacket override/////// ///////ControlPacket override///////
bool loadFromData(uint8_t *buf, size_t len) override; bool loadFromData(uint8_t *buf, size_t len) override;
@ -59,7 +55,6 @@ public:
uint32_t recv_rate; uint32_t recv_rate;
}; };
class ACKACKPacket : public ControlPacket { class ACKACKPacket : public ControlPacket {
public: public:
using Ptr = std::shared_ptr<ACKACKPacket>; using Ptr = std::shared_ptr<ACKACKPacket>;
@ -89,7 +84,6 @@ public:
} }
uint32_t ack_number; uint32_t ack_number;
}; };
} // namespace SRT } // namespace SRT

36
srt/Common.hpp
View File

@ -1,9 +1,19 @@
#ifndef ZLMEDIAKIT_SRT_COMMON_H #ifndef ZLMEDIAKIT_SRT_COMMON_H
#define ZLMEDIAKIT_SRT_COMMON_H #define ZLMEDIAKIT_SRT_COMMON_H
#if defined(_WIN32)
#include <Iphlpapi.h>
#include <winsock2.h>
#include <ws2tcpip.h>
#pragma comment(lib, "Ws2_32.lib")
#pragma comment(lib, "Iphlpapi.lib")
#else
#include <netdb.h>
#include <sys/socket.h>
#endif // defined(_WIN32)
#include <chrono> #include <chrono>
namespace SRT namespace SRT {
{
using SteadyClock = std::chrono::steady_clock; using SteadyClock = std::chrono::steady_clock;
using TimePoint = std::chrono::time_point<SteadyClock>; using TimePoint = std::chrono::time_point<SteadyClock>;
@ -45,40 +55,30 @@ inline void storeUint16LE(uint8_t *buf, uint16_t val) {
buf[1] = (val >> 8) & 0xff; buf[1] = (val >> 8) & 0xff;
} }
inline uint32_t srtVersion(int major, int minor, int patch) inline uint32_t srtVersion(int major, int minor, int patch) {
{
return patch + minor * 0x100 + major * 0x10000; return patch + minor * 0x100 + major * 0x10000;
} }
class UTicker { class UTicker {
public: public:
UTicker() { UTicker() { _created = _begin = SteadyClock::now(); }
_created = _begin = SteadyClock::now();
}
~UTicker() { ~UTicker() {}
}
/** /**
* 获取创建时间，单位微妙 * 获取创建时间，单位微妙
*/ */
int64_t elapsedTime(TimePoint now) const { int64_t elapsedTime(TimePoint now) const { return DurationCountMicroseconds(now - _begin); }
return DurationCountMicroseconds(now - _begin);
}
/** /**
* 获取上次resetTime后至今的时间，单位毫秒 * 获取上次resetTime后至今的时间，单位毫秒
*/ */
int64_t createdTime(TimePoint now) const { int64_t createdTime(TimePoint now) const { return DurationCountMicroseconds(now - _created); }
return DurationCountMicroseconds(now - _created);
}
/** /**
* 重置计时器 * 重置计时器
*/ */
void resetTime(TimePoint now) { void resetTime(TimePoint now) { _begin = now; }
_begin = now;
}
private: private:
TimePoint _begin; TimePoint _begin;

6
srt/HSExt.cpp
View File

@ -27,12 +27,11 @@ bool HSExtMessage::loadFromData(uint8_t *buf, size_t len) {
ptr += 2; ptr += 2;
return true; return true;
} }
std::string HSExtMessage::dump() { std::string HSExtMessage::dump() {
_StrPrinter printer; _StrPrinter printer;
printer << "srt version : "<<std::hex<<srt_version<<" srt flag : "<<std::hex<<srt_flag<<\ printer << "srt version : " << std::hex << srt_version << " srt flag : " << std::hex << srt_flag
" recv_tsbpd_delay="<<recv_tsbpd_delay<<" send_tsbpd_delay = "<<send_tsbpd_delay; << " recv_tsbpd_delay=" << recv_tsbpd_delay << " send_tsbpd_delay = " << send_tsbpd_delay;
return std::move(printer); return std::move(printer);
} }
bool HSExtMessage::storeToData() { bool HSExtMessage::storeToData() {
@ -85,7 +84,6 @@ bool HSExtStreamID::loadFromData(uint8_t *buf, size_t len) {
streamid.erase(streamid.find_first_of(zero), streamid.size()); streamid.erase(streamid.find_first_of(zero), streamid.size());
} }
return true; return true;
} }
bool HSExtStreamID::storeToData() { bool HSExtStreamID::storeToData() {
size_t content_size = ((streamid.length() + 4) + 3) / 4 * 4; size_t content_size = ((streamid.length() + 4) + 3) / 4 * 4;

53
srt/Packet.cpp
View File

@ -1,28 +1,11 @@
 #include "Util/MD5.h"
#if defined(_WIN32)
#include <winsock2.h>
#include <ws2tcpip.h>
#include <Iphlpapi.h>
#pragma comment (lib, "Ws2_32.lib")
#pragma comment(lib,"Iphlpapi.lib")
#else
#include <sys/socket.h>
#include <netdb.h>
#endif // defined(_WIN32)
#include <atomic>
#include "Util/logger.h" #include "Util/logger.h"
#include "Util/MD5.h" #include <atomic>
#include "Packet.hpp" #include "Packet.hpp"
namespace SRT { namespace SRT {
const size_t DataPacket::HEADER_SIZE; const size_t DataPacket::HEADER_SIZE;
const size_t ControlPacket::HEADER_SIZE; const size_t ControlPacket::HEADER_SIZE;
const size_t HandshakePacket::HS_CONTENT_MIN_SIZE; const size_t HandshakePacket::HS_CONTENT_MIN_SIZE;
@ -158,8 +141,6 @@ size_t DataPacket::payloadSize() {
return _data->size() - HEADER_SIZE; return _data->size() - HEADER_SIZE;
} }
bool ControlPacket::isControlPacket(uint8_t *buf, size_t len) { bool ControlPacket::isControlPacket(uint8_t *buf, size_t len) {
if (len < HEADER_SIZE) { if (len < HEADER_SIZE) {
WarnL << "data size" << len << " less " << HEADER_SIZE; WarnL << "data size" << len << " less " << HEADER_SIZE;
@ -302,8 +283,7 @@ bool HandshakePacket::loadExtMessage(uint8_t *buf,size_t len){
while (ptr < buf + len) { while (ptr < buf + len) {
type = loadUint16(ptr); type = loadUint16(ptr);
length = loadUint16(ptr + 2); length = loadUint16(ptr + 2);
switch (type) switch (type) {
{
case HSExt::SRT_CMD_HSREQ: case HSExt::SRT_CMD_HSREQ:
case HSExt::SRT_CMD_HSRSP: case HSExt::SRT_CMD_HSRSP:
ext = std::make_shared<HSExtMessage>(); ext = std::make_shared<HSExtMessage>();
@ -329,8 +309,7 @@ bool HandshakePacket::loadExtMessage(uint8_t *buf,size_t len){
return true; return true;
} }
bool HandshakePacket::storeExtMessage() bool HandshakePacket::storeExtMessage() {
{
uint8_t *buf = (uint8_t *)_data->data() + HEADER_SIZE + 48; uint8_t *buf = (uint8_t *)_data->data() + HEADER_SIZE + 48;
size_t len = _data->size() - HEADER_SIZE - 48; size_t len = _data->size() - HEADER_SIZE - 48;
for (auto ex : ext_list) { for (auto ex : ext_list) {
@ -399,7 +378,6 @@ bool HandshakePacket::storeToData() {
assert(encryption_field == NO_ENCRYPTION); assert(encryption_field == NO_ENCRYPTION);
return storeExtMessage(); return storeExtMessage();
} }
@ -440,25 +418,21 @@ void HandshakePacket::assignPeerIP(struct sockaddr_storage* addr){
} }
} }
} }
uint32_t HandshakePacket::generateSynCookie(struct sockaddr_storage* addr,TimePoint ts,uint32_t current_cookie, int correction ){ uint32_t HandshakePacket::generateSynCookie(
struct sockaddr_storage *addr, TimePoint ts, uint32_t current_cookie, int correction) {
static std::atomic<uint32_t> distractor { 0 }; static std::atomic<uint32_t> distractor { 0 };
uint32_t rollover = distractor.load() + 10; uint32_t rollover = distractor.load() + 10;
for (;;) for (;;) {
{
// SYN cookie // SYN cookie
char clienthost[NI_MAXHOST]; char clienthost[NI_MAXHOST];
char clientport[NI_MAXSERV]; char clientport[NI_MAXSERV];
getnameinfo((struct sockaddr*)addr, getnameinfo(
sizeof(struct sockaddr_storage), (struct sockaddr *)addr, sizeof(struct sockaddr_storage), clienthost, sizeof(clienthost), clientport,
clienthost, sizeof(clientport), NI_NUMERICHOST | NI_NUMERICSERV);
sizeof(clienthost), int64_t timestamp = (DurationCountMicroseconds(SteadyClock::now() - ts) / 60000000) + distractor.load()
clientport, + correction; // secret changes every one minute
sizeof(clientport),
NI_NUMERICHOST | NI_NUMERICSERV);
int64_t timestamp = (DurationCountMicroseconds(SteadyClock::now() - ts) / 60000000) + distractor.load() +
correction; // secret changes every one minute
std::stringstream cookiestr; std::stringstream cookiestr;
cookiestr << clienthost << ":" << clientport << ":" << timestamp; cookiestr << clienthost << ":" << clientport << ":" << timestamp;
union { union {
@ -512,8 +486,7 @@ bool NAKPacket::loadFromData(uint8_t *buf, size_t len) {
uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE; uint8_t *ptr = (uint8_t *)_data->data() + HEADER_SIZE;
uint8_t *end = (uint8_t *)_data->data() + _data->size(); uint8_t *end = (uint8_t *)_data->data() + _data->size();
LostPair lost; LostPair lost;
while (ptr<end) while (ptr < end) {
{
if ((*ptr) & 0x80) { if ((*ptr) & 0x80) {
lost.first = loadUint32(ptr) & 0x7fffffff; lost.first = loadUint32(ptr) & 0x7fffffff;
lost.second = loadUint32(ptr + 4) & 0x7fffffff; lost.second = loadUint32(ptr + 4) & 0x7fffffff;

22
srt/Packet.hpp
View File

@ -62,8 +62,6 @@ public:
uint32_t timestamp; uint32_t timestamp;
uint32_t dst_socket_id; uint32_t dst_socket_id;
TimePoint get_ts; // recv or send time
private: private:
BufferRaw::Ptr _data; BufferRaw::Ptr _data;
}; };
@ -182,15 +180,14 @@ public:
enum { HS_EXT_FILED_HSREQ = 0x00000001, HS_EXT_FILED_KMREQ = 0x00000002, HS_EXT_FILED_CONFIG = 0x00000004 }; enum { HS_EXT_FILED_HSREQ = 0x00000001, HS_EXT_FILED_KMREQ = 0x00000002, HS_EXT_FILED_CONFIG = 0x00000004 };
HandshakePacket() = default; HandshakePacket() = default;
~HandshakePacket() = default; ~HandshakePacket() = default;
static bool isHandshakePacket(uint8_t *buf, size_t len); static bool isHandshakePacket(uint8_t *buf, size_t len);
static uint32_t getHandshakeType(uint8_t *buf, size_t len); static uint32_t getHandshakeType(uint8_t *buf, size_t len);
static uint32_t getSynCookie(uint8_t *buf, size_t len); static uint32_t getSynCookie(uint8_t *buf, size_t len);
static uint32_t generateSynCookie(struct sockaddr_storage* addr,TimePoint ts,uint32_t current_cookie = 0, int correction = 0); static uint32_t
generateSynCookie(struct sockaddr_storage *addr, TimePoint ts, uint32_t current_cookie = 0, int correction = 0);
void assignPeerIP(struct sockaddr_storage *addr); void assignPeerIP(struct sockaddr_storage *addr);
///////ControlPacket override/////// ///////ControlPacket override///////
@ -209,6 +206,7 @@ public:
uint8_t peer_ip_addr[16]; uint8_t peer_ip_addr[16];
std::vector<HSExt::Ptr> ext_list; std::vector<HSExt::Ptr> ext_list;
private: private:
bool loadExtMessage(uint8_t *buf, size_t len); bool loadExtMessage(uint8_t *buf, size_t len);
bool storeExtMessage(); bool storeExtMessage();
@ -229,8 +227,7 @@ private:
Figure 12: Keep-Alive control packet Figure 12: Keep-Alive control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-keep-alive https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-keep-alive
*/ */
class KeepLivePacket : public ControlPacket class KeepLivePacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<KeepLivePacket>; using Ptr = std::shared_ptr<KeepLivePacket>;
KeepLivePacket() = default; KeepLivePacket() = default;
@ -265,8 +262,7 @@ An SRT NAK packet is formatted as follows:
Figure 14: NAK control packet Figure 14: NAK control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-nak-control-packet https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-nak-control-packet
*/ */
class NAKPacket : public ControlPacket class NAKPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<NAKPacket>; using Ptr = std::shared_ptr<NAKPacket>;
using LostPair = std::pair<uint32_t, uint32_t>; using LostPair = std::pair<uint32_t, uint32_t>;
@ -278,11 +274,11 @@ public:
bool storeToData() override; bool storeToData() override;
std::list<LostPair> lost_list; std::list<LostPair> lost_list;
private: private:
size_t getCIFSize(); size_t getCIFSize();
}; };
/* /*
0 1 2 3 0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
@ -302,8 +298,7 @@ private:
Figure 18: Drop Request control packet Figure 18: Drop Request control packet
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-message-drop-request https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-message-drop-request
*/ */
class MsgDropReqPacket : public ControlPacket class MsgDropReqPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<MsgDropReqPacket>; using Ptr = std::shared_ptr<MsgDropReqPacket>;
MsgDropReqPacket() = default; MsgDropReqPacket() = default;
@ -332,8 +327,7 @@ class MsgDropReqPacket : public ControlPacket
https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-shutdown https://haivision.github.io/srt-rfc/draft-sharabayko-srt.html#name-shutdown
*/ */
class ShutDownPacket : public ControlPacket class ShutDownPacket : public ControlPacket {
{
public: public:
using Ptr = std::shared_ptr<ShutDownPacket>; using Ptr = std::shared_ptr<ShutDownPacket>;
ShutDownPacket() = default; ShutDownPacket() = default;

19
srt/PacketQueue.cpp
View File

@ -31,14 +31,14 @@ inline bool isTSCycle(uint32_t first,uint32_t second){
PacketQueue::PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t lantency) PacketQueue::PacketQueue(uint32_t max_size, uint32_t init_seq, uint32_t lantency)
: _pkt_expected_seq(init_seq) : _pkt_expected_seq(init_seq)
, _pkt_cap(max_size) , _pkt_cap(max_size)
, _pkt_lantency(lantency) { , _pkt_lantency(lantency) {}
}
void PacketQueue::tryInsertPkt(DataPacket::Ptr pkt) { void PacketQueue::tryInsertPkt(DataPacket::Ptr pkt) {
if (_pkt_expected_seq <= pkt->packet_seq_number) { if (_pkt_expected_seq <= pkt->packet_seq_number) {
auto diff = pkt->packet_seq_number - _pkt_expected_seq; auto diff = pkt->packet_seq_number - _pkt_expected_seq;
if (diff >= (MAX_SEQ >> 1)) { if (diff >= (MAX_SEQ >> 1)) {
TraceL << "drop packet too later for cycle "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; TraceL << "drop packet too later for cycle "
<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number;
return; return;
} else { } else {
_pkt_map.emplace(pkt->packet_seq_number, pkt); _pkt_map.emplace(pkt->packet_seq_number, pkt);
@ -47,9 +47,11 @@ void PacketQueue::tryInsertPkt(DataPacket::Ptr pkt){
auto diff = _pkt_expected_seq - pkt->packet_seq_number; auto diff = _pkt_expected_seq - pkt->packet_seq_number;
if (diff >= (MAX_SEQ >> 1)) { if (diff >= (MAX_SEQ >> 1)) {
_pkt_map.emplace(pkt->packet_seq_number, pkt); _pkt_map.emplace(pkt->packet_seq_number, pkt);
TraceL<<" cycle packet "<<"expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; TraceL << " cycle packet "
<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number;
} else { } else {
//TraceL << "drop packet too later "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" << pkt->packet_seq_number; // TraceL << "drop packet too later "<< "expected seq=" << _pkt_expected_seq << " pkt seq=" <<
// pkt->packet_seq_number;
} }
} }
} }
@ -182,7 +184,9 @@ size_t PacketQueue::getExpectedSize() {
uint32_t max = _pkt_map.rbegin()->first; uint32_t max = _pkt_map.rbegin()->first;
uint32_t min = _pkt_map.begin()->first; uint32_t min = _pkt_map.begin()->first;
if ((max - min) >= (MAX_SEQ >> 1)) { if ((max - min) >= (MAX_SEQ >> 1)) {
TraceL<<"cycle "<<"expected seq "<<_pkt_expected_seq<<" min "<<min<<" max "<<max<<" size "<<_pkt_map.size(); TraceL << "cycle "
<< "expected seq " << _pkt_expected_seq << " min " << min << " max " << max << " size "
<< _pkt_map.size();
return MAX_SEQ - _pkt_expected_seq + min + 1; return MAX_SEQ - _pkt_expected_seq + min + 1;
} else { } else {
return max - _pkt_expected_seq + 1; return max - _pkt_expected_seq + 1;
@ -210,7 +214,8 @@ std::string PacketQueue::dump(){
if (_pkt_map.empty()) { if (_pkt_map.empty()) {
printer << " expected seq :" << _pkt_expected_seq; printer << " expected seq :" << _pkt_expected_seq;
} else { } else {
printer<<" expected seq :"<<_pkt_expected_seq<<" size:"<<_pkt_map.size()<<" first:"<<_pkt_map.begin()->second->packet_seq_number; printer << " expected seq :" << _pkt_expected_seq << " size:" << _pkt_map.size()
<< " first:" << _pkt_map.begin()->second->packet_seq_number;
printer << " last:" << _pkt_map.rbegin()->second->packet_seq_number; printer << " last:" << _pkt_map.rbegin()->second->packet_seq_number;
printer << " latency:" << timeLantency() / 1e3; printer << " latency:" << timeLantency() / 1e3;
} }

5
srt/PacketQueue.hpp
View File

@ -3,8 +3,8 @@
#include "Packet.hpp" #include "Packet.hpp"
#include <algorithm> #include <algorithm>
#include <list> #include <list>
#include <memory>
#include <map> #include <map>
#include <memory>
#include <tuple> #include <tuple>
#include <utility> #include <utility>
@ -31,10 +31,11 @@ public:
bool drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out); bool drop(uint32_t first, uint32_t last, std::list<DataPacket::Ptr> &out);
std::string dump(); std::string dump();
private: private:
void tryInsertPkt(DataPacket::Ptr pkt); void tryInsertPkt(DataPacket::Ptr pkt);
private:
private:
std::map<uint32_t, DataPacket::Ptr> _pkt_map; std::map<uint32_t, DataPacket::Ptr> _pkt_map;
uint32_t _pkt_expected_seq = 0; uint32_t _pkt_expected_seq = 0;

2
srt/PacketSendQueue.cpp
View File

@ -66,7 +66,7 @@ uint32_t PacketSendQueue::timeLantency() {
} else { } else {
dur = first - last; dur = first - last;
} }
if (dur > (0x01 << 31)) { if (dur > ((uint32_t)0x01 << 31)) {
TraceL << "cycle timeLantency " << dur; TraceL << "cycle timeLantency " << dur;
dur = 0xffffffff - dur; dur = 0xffffffff - dur;
} }

2
srt/PacketSendQueue.hpp
View File

@ -17,8 +17,10 @@ public:
bool drop(uint32_t num); bool drop(uint32_t num);
bool inputPacket(DataPacket::Ptr pkt); bool inputPacket(DataPacket::Ptr pkt);
std::list<DataPacket::Ptr> findPacketBySeq(uint32_t start, uint32_t end); std::list<DataPacket::Ptr> findPacketBySeq(uint32_t start, uint32_t end);
private: private:
uint32_t timeLantency(); uint32_t timeLantency();
private: private:
std::list<DataPacket::Ptr> _pkt_cache; std::list<DataPacket::Ptr> _pkt_cache;
uint32_t _pkt_cap; uint32_t _pkt_cap;

6
srt/SrtSession.cpp
View File

@ -125,10 +125,12 @@ void SrtSession::onError(const SockException &err) {
// 防止互相引用导致不释放 // 防止互相引用导致不释放
auto transport = std::move(_transport); auto transport = std::move(_transport);
getPoller()->async([transport,err] { getPoller()->async(
[transport, err] {
//延时减引用，防止使用transport对象时，销毁对象 //延时减引用，防止使用transport对象时，销毁对象
transport->onShutdown(err); transport->onShutdown(err);
}, false); },
false);
} }
void SrtSession::onManager() { void SrtSession::onManager() {

1
srt/SrtSession.hpp
View File

@ -24,7 +24,6 @@ private:
Ticker _ticker; Ticker _ticker;
struct sockaddr_storage _peer_addr; struct sockaddr_storage _peer_addr;
SrtTransport::Ptr _transport; SrtTransport::Ptr _transport;
}; };
} // namespace SRT } // namespace SRT

18
srt/SrtTransport.cpp
View File

@ -1,9 +1,9 @@
#include <stdlib.h> #include <stdlib.h>
#include "Util/onceToken.h" #include "Util/onceToken.h"
#include "SrtTransport.hpp"
#include "Packet.hpp"
#include "Ack.hpp" #include "Ack.hpp"
#include "Packet.hpp"
#include "SrtTransport.hpp"
namespace SRT { namespace SRT {
#define SRT_FIELD "srt." #define SRT_FIELD "srt."
// srt 超时时间 // srt 超时时间
@ -18,7 +18,7 @@ static std::atomic<uint32_t> s_srt_socket_id_generate{125};
SrtTransport::SrtTransport(const EventPoller::Ptr &poller) SrtTransport::SrtTransport(const EventPoller::Ptr &poller)
: _poller(poller) { : _poller(poller) {
_start_timestamp = SteadyClock::now(); _start_timestamp = SteadyClock::now();
_socket_id = s_srt_socket_id_generate.fetch_add(1);\ _socket_id = s_srt_socket_id_generate.fetch_add(1);
_pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp); _pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp);
_recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp); _recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp);
_estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp); _estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp);
@ -200,7 +200,8 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad
unregisterSelfHandshake(); unregisterSelfHandshake();
registerSelf(); registerSelf();
sendControlPacket(res, true); sendControlPacket(res, true);
TraceL<<" buf size = "<<res->max_flow_window_size<<" init seq ="<<_init_seq_number<<" lantency="<<delay; TraceL << " buf size = " << res->max_flow_window_size << " init seq =" << _init_seq_number
<< " lantency=" << delay;
_recv_buf = std::make_shared<PacketQueue>(res->max_flow_window_size, _init_seq_number, delay * 1e3); _recv_buf = std::make_shared<PacketQueue>(res->max_flow_window_size, _init_seq_number, delay * 1e3);
_send_buf = std::make_shared<PacketSendQueue>(res->max_flow_window_size, delay * 1e3); _send_buf = std::make_shared<PacketSendQueue>(res->max_flow_window_size, delay * 1e3);
_send_packet_seq_number = _init_seq_number; _send_packet_seq_number = _init_seq_number;
@ -335,7 +336,6 @@ void SrtTransport::handleDropReq(uint8_t *buf, int len, struct sockaddr_storage
_light_ack_pkt_count = 0; _light_ack_pkt_count = 0;
} }
_light_ack_pkt_count++; _light_ack_pkt_count++;
} }
void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr) { void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr) {
TraceL; TraceL;
@ -350,7 +350,6 @@ void SrtTransport::handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *
_rtt_variance = (3 * _rtt_variance + abs((long)_rtt - (long)rtt)) / 4; _rtt_variance = (3 * _rtt_variance + abs((long)_rtt - (long)rtt)) / 4;
_rtt = (7 * rtt + _rtt) / 8; _rtt = (7 * rtt + _rtt) / 8;
// TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance; // TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance;
_ack_send_timestamp.erase(pkt->ack_number); _ack_send_timestamp.erase(pkt->ack_number);
} }
@ -420,7 +419,6 @@ void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_stora
DataPacket::Ptr pkt = std::make_shared<DataPacket>(); DataPacket::Ptr pkt = std::make_shared<DataPacket>();
pkt->loadFromData(buf, len); pkt->loadFromData(buf, len);
pkt->get_ts = _now;
std::list<DataPacket::Ptr> list; std::list<DataPacket::Ptr> list;
//TraceL<<" seq="<< pkt->packet_seq_number<<" ts="<<pkt->timestamp<<" size="<<pkt->payloadSize()<<\ //TraceL<<" seq="<< pkt->packet_seq_number<<" ts="<<pkt->timestamp<<" size="<<pkt->payloadSize()<<\
//" PP="<<(int)pkt->PP<<" O="<<(int)pkt->O<<" kK="<<(int)pkt->KK<<" R="<<(int)pkt->R; //" PP="<<(int)pkt->PP<<" O="<<(int)pkt->O<<" kK="<<(int)pkt->KK<<" R="<<(int)pkt->R;
@ -435,7 +433,8 @@ void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_stora
} }
if (list.empty()) { if (list.empty()) {
//TraceL<<_recv_buf->dump()<<" nake interval:"<<nak_interval/1000<<" ticker:"<<_nak_ticker.elapsedTime(_now)/1000; // TraceL<<_recv_buf->dump()<<" nake interval:"<<nak_interval/1000<<"
// ticker:"<<_nak_ticker.elapsedTime(_now)/1000;
} }
if (_nak_ticker.elapsedTime(_now) > nak_interval) { if (_nak_ticker.elapsedTime(_now) > nak_interval) {
@ -504,7 +503,6 @@ void SrtTransport::registerSelf() {
return; return;
} }
SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this()); SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this());
} }
void SrtTransport::unregisterSelf() { void SrtTransport::unregisterSelf() {
SrtTransportManager::Instance().removeItem(std::to_string(_socket_id)); SrtTransportManager::Instance().removeItem(std::to_string(_socket_id));
@ -565,7 +563,6 @@ void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush){
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp); pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
sendDataPacket(pkt, ptr, (int)size, flush); sendDataPacket(pkt, ptr, (int)size, flush);
} }
} }
//////////// SrtTransportManager ////////////////////////// //////////// SrtTransportManager //////////////////////////
SrtTransportManager &SrtTransportManager::Instance() { SrtTransportManager &SrtTransportManager::Instance() {
@ -615,5 +612,4 @@ SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key)
return it->second.lock(); return it->second.lock();
} }
} // namespace SRT } // namespace SRT

17
srt/SrtTransport.hpp
View File

@ -1,10 +1,10 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_H #ifndef ZLMEDIAKIT_SRT_TRANSPORT_H
#define ZLMEDIAKIT_SRT_TRANSPORT_H #define ZLMEDIAKIT_SRT_TRANSPORT_H
#include <mutex> #include <atomic>
#include <chrono> #include <chrono>
#include <memory> #include <memory>
#include <atomic> #include <mutex>
#include "Network/Session.h" #include "Network/Session.h"
#include "Poller/EventPoller.h" #include "Poller/EventPoller.h"
@ -46,13 +46,12 @@ public:
void unregisterSelfHandshake(); void unregisterSelfHandshake();
void unregisterSelf(); void unregisterSelf();
protected: protected:
virtual void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {}; virtual void onHandShakeFinished(std::string &streamid, struct sockaddr_storage *addr) {};
virtual void onSRTData(DataPacket::Ptr pkt) {}; virtual void onSRTData(DataPacket::Ptr pkt) {};
virtual void onShutdown(const SockException &ex); virtual void onShutdown(const SockException &ex);
virtual bool isPusher(){ virtual bool isPusher() { return true; };
return true;
};
private: private:
void registerSelfHandshake(); void registerSelfHandshake();
@ -83,13 +82,13 @@ private:
void sendMsgDropReq(uint32_t first, uint32_t last); void sendMsgDropReq(uint32_t first, uint32_t last);
size_t getPayloadSize(); size_t getPayloadSize();
protected: protected:
void sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush = false); void sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush = false);
void sendControlPacket(ControlPacket::Ptr pkt, bool flush = true); void sendControlPacket(ControlPacket::Ptr pkt, bool flush = true);
virtual void sendPacket(Buffer::Ptr pkt, bool flush = true); virtual void sendPacket(Buffer::Ptr pkt, bool flush = true);
virtual int getLantencyMul(){ virtual int getLantencyMul() { return 4; };
return 4;
};
private: private:
//当前选中的udp链接 //当前选中的udp链接
Session::Ptr _selected_session; Session::Ptr _selected_session;
@ -137,7 +136,6 @@ private:
HandshakePacket::Ptr _handleshake_res; HandshakePacket::Ptr _handleshake_res;
ResourcePool<BufferRaw> _packet_pool; ResourcePool<BufferRaw> _packet_pool;
}; };
class SrtTransportManager { class SrtTransportManager {
@ -150,6 +148,7 @@ public:
void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr); void addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr);
void removeHandshakeItem(const std::string &key); void removeHandshakeItem(const std::string &key);
SrtTransport::Ptr getHandshakeItem(const std::string &key); SrtTransport::Ptr getHandshakeItem(const std::string &key);
private: private:
SrtTransportManager() = default; SrtTransportManager() = default;

39
srt/SrtTransportImp.cpp
View File

@ -10,16 +10,15 @@ SrtTransportImp::SrtTransportImp(const EventPoller::Ptr &poller)
SrtTransportImp::~SrtTransportImp() { SrtTransportImp::~SrtTransportImp() {
InfoP(this); InfoP(this);
uint64_t duration = _alive_ticker.createdTime() / 1000; uint64_t duration = _alive_ticker.createdTime() / 1000;
WarnP(this) << (_is_pusher ? "srt 推流器(" : "srt 播放器(") WarnP(this) << (_is_pusher ? "srt 推流器(" : "srt 播放器(") << _media_info._vhost << "/" << _media_info._app << "/"
<< _media_info._vhost << "/" << _media_info._streamid << ")断开,耗时(s):" << duration;
<< _media_info._app << "/"
<< _media_info._streamid
<< ")断开,耗时(s):" << duration;
//流量统计事件广播 //流量统计事件广播
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold); GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_total_bytes >= iFlowThreshold * 1024) { if (_total_bytes >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false, static_cast<SockInfo &>(*this)); NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastFlowReport, _media_info, _total_bytes, duration, false,
static_cast<SockInfo &>(*this));
} }
} }
@ -105,10 +104,9 @@ void SrtTransportImp::emitOnPublish() {
return; return;
} }
if (err.empty()) { if (err.empty()) {
strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(strong_self->_media_info._vhost, strong_self->_muxer = std::make_shared<MultiMediaSourceMuxer>(
strong_self->_media_info._app, strong_self->_media_info._vhost, strong_self->_media_info._app, strong_self->_media_info._streamid,
strong_self->_media_info._streamid, 0.0f, 0.0f, option);
option);
strong_self->_muxer->setMediaListener(strong_self); strong_self->_muxer->setMediaListener(strong_self);
strong_self->doCachedFunc(); strong_self->doCachedFunc();
InfoP(strong_self) << "允许 srt 推流"; InfoP(strong_self) << "允许 srt 推流";
@ -119,14 +117,15 @@ void SrtTransportImp::emitOnPublish() {
}; };
//触发推流鉴权事件 //触发推流鉴权事件
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPublish, MediaOriginType::srt_push, _media_info, invoker, static_cast<SockInfo &>(*this)); auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPublish, MediaOriginType::srt_push, _media_info, invoker,
static_cast<SockInfo &>(*this));
if (!flag) { if (!flag) {
//该事件无人监听,默认不鉴权 //该事件无人监听,默认不鉴权
invoker("", ProtocolOption()); invoker("", ProtocolOption());
} }
} }
void SrtTransportImp::emitOnPlay() { void SrtTransportImp::emitOnPlay() {
std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this()); std::weak_ptr<SrtTransportImp> weak_self = static_pointer_cast<SrtTransportImp>(shared_from_this());
Broadcast::AuthInvoker invoker = [weak_self](const string &err) { Broadcast::AuthInvoker invoker = [weak_self](const string &err) {
@ -143,7 +142,8 @@ void SrtTransportImp::emitOnPlay(){
}); });
}; };
auto flag = NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this)); auto flag = NoticeCenter::Instance().emitEvent(
Broadcast::kBroadcastMediaPlayed, _media_info, invoker, static_cast<SockInfo &>(*this));
if (!flag) { if (!flag) {
doPlay(); doPlay();
} }
@ -236,9 +236,7 @@ bool SrtTransportImp::inputFrame(const Frame::Ptr &frame) {
} }
auto frame_cached = Frame::getCacheAbleFrame(frame); auto frame_cached = Frame::getCacheAbleFrame(frame);
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, frame_cached]() { _cached_func.emplace_back([this, frame_cached]() { _muxer->inputFrame(frame_cached); });
_muxer->inputFrame(frame_cached);
});
return true; return true;
} }
@ -248,9 +246,7 @@ bool SrtTransportImp::addTrack(const Track::Ptr &track) {
} }
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this, track]() { _cached_func.emplace_back([this, track]() { _muxer->addTrack(track); });
_muxer->addTrack(track);
});
return true; return true;
} }
@ -259,9 +255,7 @@ void SrtTransportImp::addTrackCompleted() {
_muxer->addTrackCompleted(); _muxer->addTrackCompleted();
} else { } else {
lock_guard<recursive_mutex> lck(_func_mtx); lock_guard<recursive_mutex> lck(_func_mtx);
_cached_func.emplace_back([this]() { _cached_func.emplace_back([this]() { _muxer->addTrackCompleted(); });
_muxer->addTrackCompleted();
});
} }
} }
@ -278,5 +272,4 @@ int SrtTransportImp::getLantencyMul(){
return lantencyMul; return lantencyMul;
} }
} // namespace SRT } // namespace SRT

13
srt/SrtTransportImp.hpp
View File

@ -1,11 +1,10 @@
#ifndef ZLMEDIAKIT_SRT_TRANSPORT_IMP_H #ifndef ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#define ZLMEDIAKIT_SRT_TRANSPORT_IMP_H #define ZLMEDIAKIT_SRT_TRANSPORT_IMP_H
#include <mutex>
#include "Common/MultiMediaSourceMuxer.h" #include "Common/MultiMediaSourceMuxer.h"
#include "Rtp/Decoder.h" #include "Rtp/Decoder.h"
#include "TS/TSMediaSource.h"
#include "SrtTransport.hpp" #include "SrtTransport.hpp"
#include "TS/TSMediaSource.h"
#include <mutex>
namespace SRT { namespace SRT {
using namespace toolkit; using namespace toolkit;
@ -23,9 +22,7 @@ public:
SrtTransport::inputSockData(buf, len, addr); SrtTransport::inputSockData(buf, len, addr);
_total_bytes += len; _total_bytes += len;
} }
void onSendTSData(const Buffer::Ptr &buffer, bool flush){ void onSendTSData(const Buffer::Ptr &buffer, bool flush) { SrtTransport::onSendTSData(buffer, flush); }
SrtTransport::onSendTSData(buffer,flush);
}
/// SockInfo override /// SockInfo override
std::string get_local_ip() override; std::string get_local_ip() override;
uint16_t get_local_port() override; uint16_t get_local_port() override;
@ -45,9 +42,7 @@ protected:
SrtTransport::sendPacket(pkt, flush); SrtTransport::sendPacket(pkt, flush);
}; };
bool isPusher() override{ bool isPusher() override { return _is_pusher; }
return _is_pusher;
}
///////MediaSourceEvent override/////// ///////MediaSourceEvent override///////
// 关闭 // 关闭

1
srt/Statistic.cpp
View File

@ -65,7 +65,6 @@ uint32_t EstimatedLinkCapacityContext::getEstimatedLinkCapacity() {
double dur = tmp[0] / 1e6; double dur = tmp[0] / 1e6;
return (uint32_t)(1.0 / dur); return (uint32_t)(1.0 / dur);
} }
void RecvRateContext::inputPacket(TimePoint &ts, size_t size) { void RecvRateContext::inputPacket(TimePoint &ts, size_t size) {

15
srt/Statistic.hpp
View File

@ -8,22 +8,25 @@
namespace SRT { namespace SRT {
class PacketRecvRateContext { class PacketRecvRateContext {
public: public:
PacketRecvRateContext(TimePoint start):_start(start){}; PacketRecvRateContext(TimePoint start)
: _start(start) {};
~PacketRecvRateContext() = default; ~PacketRecvRateContext() = default;
void inputPacket(TimePoint &ts); void inputPacket(TimePoint &ts);
uint32_t getPacketRecvRate(); uint32_t getPacketRecvRate();
private: private:
std::map<int64_t, int64_t> _pkt_map; std::map<int64_t, int64_t> _pkt_map;
TimePoint _start; TimePoint _start;
}; };
class EstimatedLinkCapacityContext { class EstimatedLinkCapacityContext {
public: public:
EstimatedLinkCapacityContext(TimePoint start):_start(start){}; EstimatedLinkCapacityContext(TimePoint start)
: _start(start) {};
~EstimatedLinkCapacityContext() = default; ~EstimatedLinkCapacityContext() = default;
void inputPacket(TimePoint &ts); void inputPacket(TimePoint &ts);
uint32_t getEstimatedLinkCapacity(); uint32_t getEstimatedLinkCapacity();
private: private:
std::map<int64_t, int64_t> _pkt_map; std::map<int64_t, int64_t> _pkt_map;
TimePoint _start; TimePoint _start;
@ -31,16 +34,16 @@ private:
class RecvRateContext { class RecvRateContext {
public: public:
RecvRateContext(TimePoint start):_start(start){}; RecvRateContext(TimePoint start)
: _start(start) {};
~RecvRateContext() = default; ~RecvRateContext() = default;
void inputPacket(TimePoint &ts, size_t size); void inputPacket(TimePoint &ts, size_t size);
uint32_t getRecvRate(); uint32_t getRecvRate();
private: private:
std::map<int64_t, size_t> _pkt_map; std::map<int64_t, size_t> _pkt_map;
TimePoint _start; TimePoint _start;
}; };
} // namespace SRT } // namespace SRT
#endif // ZLMEDIAKIT_SRT_STATISTIC_H #endif // ZLMEDIAKIT_SRT_STATISTIC_H