ZLMediaKit/srt/SrtTransport.cpp

824 lines
30 KiB
C++
Raw Normal View History

#include "Util/onceToken.h"
#include <iterator>
#include <stdlib.h>
2022-06-03 13:25:32 +08:00
2022-06-07 09:52:20 +08:00
#include "Ack.hpp"
#include "Packet.hpp"
#include "SrtTransport.hpp"
2022-06-03 13:25:32 +08:00
namespace SRT {
#define SRT_FIELD "srt."
2022-06-07 09:52:20 +08:00
// srt 超时时间
const std::string kTimeOutSec = SRT_FIELD "timeoutSec";
// srt 单端口udp服务器
const std::string kPort = SRT_FIELD "port";
const std::string kLatencyMul = SRT_FIELD "latencyMul";
2022-06-11 22:51:11 +08:00
const std::string kPktBufSize = SRT_FIELD "pktBufSize";
2022-06-03 13:25:32 +08:00
2022-06-07 09:52:20 +08:00
static std::atomic<uint32_t> s_srt_socket_id_generate { 125 };
2022-06-03 13:25:32 +08:00
//////////// SrtTransport //////////////////////////
SrtTransport::SrtTransport(const EventPoller::Ptr &poller)
: _poller(poller) {
2022-06-07 09:52:20 +08:00
_start_timestamp = SteadyClock::now();
_socket_id = s_srt_socket_id_generate.fetch_add(1);
_pkt_recv_rate_context = std::make_shared<PacketRecvRateContext>(_start_timestamp);
//_recv_rate_context = std::make_shared<RecvRateContext>(_start_timestamp);
2022-06-07 09:52:20 +08:00
_estimated_link_capacity_context = std::make_shared<EstimatedLinkCapacityContext>(_start_timestamp);
}
2022-06-03 13:25:32 +08:00
2022-06-07 09:52:20 +08:00
SrtTransport::~SrtTransport() {
TraceL << " ";
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
const EventPoller::Ptr &SrtTransport::getPoller() const {
return _poller;
}
void SrtTransport::setSession(Session::Ptr session) {
_history_sessions.emplace(session.get(), session);
if (_selected_session) {
InfoL << "srt network changed: " << _selected_session->get_peer_ip() << ":"
<< _selected_session->get_peer_port() << " -> " << session->get_peer_ip() << ":"
<< session->get_peer_port() << ", id:" << _selected_session->getIdentifier();
}
_selected_session = session;
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
const Session::Ptr &SrtTransport::getSession() const {
return _selected_session;
}
2022-06-07 09:52:20 +08:00
void SrtTransport::switchToOtherTransport(uint8_t *buf, int len, uint32_t socketid, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
BufferRaw::Ptr tmp = BufferRaw::create();
struct sockaddr_storage tmp_addr = *addr;
2022-06-07 09:52:20 +08:00
tmp->assign((char *)buf, len);
2022-06-03 13:25:32 +08:00
auto trans = SrtTransportManager::Instance().getItem(std::to_string(socketid));
2022-06-07 09:52:20 +08:00
if (trans) {
trans->getPoller()->async([tmp, tmp_addr, trans] {
trans->inputSockData((uint8_t *)tmp->data(), tmp->size(), (struct sockaddr_storage *)&tmp_addr);
2022-06-03 13:25:32 +08:00
});
}
}
2022-09-20 00:39:42 +08:00
void SrtTransport::createTimerForCheckAlive(){
std::weak_ptr<SrtTransport> weak_self = std::static_pointer_cast<SrtTransport>(shared_from_this());
auto timeoutSec = getTimeOutSec();
_timer = std::make_shared<Timer>(
timeoutSec/ 2,
[weak_self,timeoutSec]() {
auto strong_self = weak_self.lock();
if (!strong_self) {
return false;
}
if (strong_self->_alive_ticker.elapsedTime() > timeoutSec * 1000) {
strong_self->onShutdown(SockException(Err_timeout, "接收srt数据超时"));
}
return true;
},
getPoller());
}
2022-06-03 13:25:32 +08:00
void SrtTransport::inputSockData(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-09-20 00:39:42 +08:00
_alive_ticker.resetTime();
if(!_timer){
createTimerForCheckAlive();
}
2022-06-07 09:52:20 +08:00
using srt_control_handler = void (SrtTransport::*)(uint8_t * buf, int len, struct sockaddr_storage *addr);
2022-06-03 13:25:32 +08:00
static std::unordered_map<uint16_t, srt_control_handler> s_control_functions;
static onceToken token([]() {
s_control_functions.emplace(ControlPacket::HANDSHAKE, &SrtTransport::handleHandshake);
s_control_functions.emplace(ControlPacket::KEEPALIVE, &SrtTransport::handleKeeplive);
s_control_functions.emplace(ControlPacket::ACK, &SrtTransport::handleACK);
s_control_functions.emplace(ControlPacket::NAK, &SrtTransport::handleNAK);
s_control_functions.emplace(ControlPacket::CONGESTIONWARNING, &SrtTransport::handleCongestionWarning);
s_control_functions.emplace(ControlPacket::SHUTDOWN, &SrtTransport::handleShutDown);
s_control_functions.emplace(ControlPacket::ACKACK, &SrtTransport::handleACKACK);
s_control_functions.emplace(ControlPacket::DROPREQ, &SrtTransport::handleDropReq);
s_control_functions.emplace(ControlPacket::PEERERROR, &SrtTransport::handlePeerError);
s_control_functions.emplace(ControlPacket::USERDEFINEDTYPE, &SrtTransport::handleUserDefinedType);
});
_now = SteadyClock::now();
2022-06-03 13:25:32 +08:00
// 处理srt数据
if (DataPacket::isDataPacket(buf, len)) {
2022-06-07 09:52:20 +08:00
uint32_t socketId = DataPacket::getSocketID(buf, len);
if (socketId == _socket_id) {
if(_handleshake_timer){
_handleshake_timer.reset();
}
2022-09-22 01:07:10 +08:00
_pkt_recv_rate_context->inputPacket(_now,len+UDP_HDR_SIZE);
//_recv_rate_context->inputPacket(_now, len);
2022-06-03 13:25:32 +08:00
handleDataPacket(buf, len, addr);
checkAndSendAckNak();
2022-06-07 09:52:20 +08:00
} else {
2022-06-22 18:41:51 +08:00
WarnL<<"DataPacket switch to other transport: "<<socketId;
2022-06-07 09:52:20 +08:00
switchToOtherTransport(buf, len, socketId, addr);
2022-06-03 13:25:32 +08:00
}
} else {
if (ControlPacket::isControlPacket(buf, len)) {
2022-06-07 09:52:20 +08:00
uint32_t socketId = ControlPacket::getSocketID(buf, len);
uint16_t type = ControlPacket::getControlType(buf, len);
if (type != ControlPacket::HANDSHAKE && socketId != _socket_id && _socket_id != 0) {
2022-06-03 13:25:32 +08:00
// socket id not same
2022-06-22 18:41:51 +08:00
WarnL<<"ControlPacket: "<< (int)type <<" switch to other transport: "<<socketId;
2022-06-07 09:52:20 +08:00
switchToOtherTransport(buf, len, socketId, addr);
2022-06-03 13:25:32 +08:00
return;
}
//_pkt_recv_rate_context->inputPacket(_now,len);
//_estimated_link_capacity_context->inputPacket(_now);
//_recv_rate_context->inputPacket(_now, len);
2022-06-03 13:25:32 +08:00
auto it = s_control_functions.find(type);
if (it == s_control_functions.end()) {
2022-06-07 09:52:20 +08:00
WarnL << " not support type ignore" << ControlPacket::getControlType(buf, len);
2022-06-03 13:25:32 +08:00
return;
2022-06-07 09:52:20 +08:00
} else {
(this->*(it->second))(buf, len, addr);
2022-06-03 13:25:32 +08:00
}
if(_is_handleshake_finished && isPusher()){
checkAndSendAckNak();
}
2022-06-03 13:25:32 +08:00
} else {
// not reach
WarnL << "not reach this";
}
}
}
2022-06-03 13:25:32 +08:00
void SrtTransport::handleHandshakeInduction(HandshakePacket &pkt, struct sockaddr_storage *addr) {
// Induction Phase
if (_handleshake_res) {
if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_INDUCTION){
if(pkt.srt_socket_id == _handleshake_res->dst_socket_id){
TraceL << getIdentifier() <<" Induction repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
2022-08-27 13:25:23 +08:00
sendControlPacket(_handleshake_res, true);
}else{
TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
onShutdown(SockException(Err_other, "client new connection"));
}
return;
}else if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_CONCLUSION){
if(_handleshake_res->dst_socket_id != pkt.srt_socket_id){
TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
onShutdown(SockException(Err_other, "client new connection"));
2022-08-27 13:25:23 +08:00
}
return;
2022-08-27 12:46:35 +08:00
}else{
WarnL<<"not reach this";
2022-08-27 12:46:35 +08:00
}
2022-06-03 13:25:32 +08:00
return;
2022-08-27 12:46:35 +08:00
}else{
TraceL << getIdentifier() <<" Induction from "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
_induction_ts = _now;
_start_timestamp = _now;
2022-06-03 13:25:32 +08:00
_init_seq_number = pkt.initial_packet_sequence_number;
_max_window_size = pkt.max_flow_window_size;
_mtu = pkt.mtu;
_last_pkt_seq = _init_seq_number - 1;
_estimated_link_capacity_context->setLastSeq(_last_pkt_seq);
2022-06-03 13:25:32 +08:00
_peer_socket_id = pkt.srt_socket_id;
HandshakePacket::Ptr res = std::make_shared<HandshakePacket>();
res->dst_socket_id = _peer_socket_id;
res->timestamp = DurationCountMicroseconds(_start_timestamp.time_since_epoch());
res->mtu = _mtu;
res->max_flow_window_size = _max_window_size;
res->initial_packet_sequence_number = _init_seq_number;
res->version = 5;
res->encryption_field = HandshakePacket::NO_ENCRYPTION;
res->extension_field = 0x4A17;
res->handshake_type = HandshakePacket::HS_TYPE_INDUCTION;
res->srt_socket_id = _peer_socket_id;
res->syn_cookie = HandshakePacket::generateSynCookie(addr, _start_timestamp);
_sync_cookie = res->syn_cookie;
memcpy(res->peer_ip_addr, pkt.peer_ip_addr, sizeof(pkt.peer_ip_addr) * sizeof(pkt.peer_ip_addr[0]));
_handleshake_res = res;
res->storeToData();
registerSelfHandshake();
sendControlPacket(res, true);
_handleshake_timer = std::make_shared<Timer>(0.2,[this]()->bool{
sendControlPacket(_handleshake_res, true);
return true;
},getPoller());
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockaddr_storage *addr) {
2022-06-07 09:52:20 +08:00
if (!_handleshake_res) {
ErrorL << "must Induction Phase for handleshake ";
2022-06-03 13:25:32 +08:00
return;
}
if (_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_INDUCTION) {
// first
HSExtMessage::Ptr req;
HSExtStreamID::Ptr sid;
2022-06-04 13:30:59 +08:00
uint32_t srt_flag = 0xbf;
2022-06-07 09:52:20 +08:00
uint16_t delay = DurationCountMicroseconds(_now - _induction_ts) * getLatencyMul() / 1000;
2022-06-11 23:50:47 +08:00
if (delay <= 120) {
delay = 120;
}
2022-07-06 21:40:24 +08:00
for (auto& ext : pkt.ext_list) {
2022-06-07 09:52:20 +08:00
// TraceL << getIdentifier() << " ext " << ext->dump();
2022-06-03 13:25:32 +08:00
if (!req) {
req = std::dynamic_pointer_cast<HSExtMessage>(ext);
}
2022-06-07 09:52:20 +08:00
if (!sid) {
2022-06-03 13:25:32 +08:00
sid = std::dynamic_pointer_cast<HSExtStreamID>(ext);
}
}
2022-06-07 09:52:20 +08:00
if (sid) {
2022-06-03 13:25:32 +08:00
_stream_id = sid->streamid;
}
2022-06-07 09:52:20 +08:00
if (req) {
2022-06-11 23:11:14 +08:00
if (req->srt_flag != srt_flag) {
2022-06-18 13:01:04 +08:00
WarnL << " flag " << req->srt_flag;
2022-06-11 23:11:14 +08:00
}
2022-06-18 13:01:04 +08:00
srt_flag = req->srt_flag;
delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay;
2022-06-04 13:30:59 +08:00
}
TraceL << getIdentifier() << " CONCLUSION Phase from"<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);;
2022-06-03 13:25:32 +08:00
HandshakePacket::Ptr res = std::make_shared<HandshakePacket>();
res->dst_socket_id = _peer_socket_id;
res->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
2022-06-03 13:25:32 +08:00
res->mtu = _mtu;
res->max_flow_window_size = _max_window_size;
res->initial_packet_sequence_number = _init_seq_number;
res->version = 5;
res->encryption_field = HandshakePacket::NO_ENCRYPTION;
res->extension_field = HandshakePacket::HS_EXT_FILED_HSREQ;
res->handshake_type = HandshakePacket::HS_TYPE_CONCLUSION;
res->srt_socket_id = _socket_id;
res->syn_cookie = 0;
res->assignPeerIP(addr);
HSExtMessage::Ptr ext = std::make_shared<HSExtMessage>();
ext->extension_type = HSExt::SRT_CMD_HSRSP;
ext->srt_version = srtVersion(1, 5, 0);
2022-06-04 13:30:59 +08:00
ext->srt_flag = srt_flag;
ext->recv_tsbpd_delay = ext->send_tsbpd_delay = delay;
2022-06-03 13:25:32 +08:00
res->ext_list.push_back(std::move(ext));
res->storeToData();
_handleshake_res = res;
unregisterSelfHandshake();
registerSelf();
sendControlPacket(res, true);
TraceL << " buf size = " << res->max_flow_window_size << " init seq =" << _init_seq_number
<< " latency=" << delay;
2022-06-18 13:01:04 +08:00
_recv_buf = std::make_shared<PacketRecvQueue>(getPktBufSize(), _init_seq_number, delay * 1e3,srt_flag);
_send_buf = std::make_shared<PacketSendQueue>(getPktBufSize(), delay * 1e3,srt_flag);
_send_packet_seq_number = _init_seq_number;
_buf_delay = delay;
2022-06-07 09:52:20 +08:00
onHandShakeFinished(_stream_id, addr);
2022-08-27 15:18:32 +08:00
if(!isPusher()){
_handleshake_timer.reset();
}
2022-06-03 13:25:32 +08:00
} else {
if(_handleshake_res->handshake_type == HandshakePacket::HS_TYPE_CONCLUSION){
if(_handleshake_res->dst_socket_id != pkt.srt_socket_id){
TraceL << getIdentifier() <<" new connection fron client "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
onShutdown(SockException(Err_other, "client new connection"));
}else{
TraceL << getIdentifier() <<" CONCLUSION repeate "<<SockUtil::inet_ntoa((struct sockaddr *)addr) << ":" << SockUtil::inet_port((struct sockaddr *)addr);
2022-08-27 13:25:23 +08:00
sendControlPacket(_handleshake_res, true);
}
2022-08-27 12:46:35 +08:00
}else{
WarnL<<"not reach this";
2022-08-27 12:46:35 +08:00
}
return;
2022-08-27 12:46:35 +08:00
2022-06-03 13:25:32 +08:00
}
//_last_ack_pkt_seq_num = _init_seq_number;
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleHandshake(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
HandshakePacket pkt;
2022-06-22 18:26:37 +08:00
if(!pkt.loadFromData(buf, len)){
WarnL<<"is not vaild HandshakePacket";
return;
}
2022-06-07 09:52:20 +08:00
if (pkt.handshake_type == HandshakePacket::HS_TYPE_INDUCTION) {
handleHandshakeInduction(pkt, addr);
} else if (pkt.handshake_type == HandshakePacket::HS_TYPE_CONCLUSION) {
handleHandshakeConclusion(pkt, addr);
} else {
WarnL << " not support handshake type = " << pkt.handshake_type;
2022-06-22 18:11:20 +08:00
WarnL <<pkt.dump();
2022-06-03 13:25:32 +08:00
}
_ack_ticker.resetTime(_now);
_nak_ticker.resetTime(_now);
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleKeeplive(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
sendKeepLivePacket();
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::sendKeepLivePacket() {
KeepLivePacket::Ptr pkt = std::make_shared<KeepLivePacket>();
pkt->dst_socket_id = _peer_socket_id;
2022-06-07 09:52:20 +08:00
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData();
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
}
void SrtTransport::handleACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
2022-06-03 13:25:32 +08:00
ACKPacket ack;
2022-06-07 09:52:20 +08:00
if (!ack.loadFromData(buf, len)) {
2022-06-04 11:23:18 +08:00
return;
}
2022-06-03 13:25:32 +08:00
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
pkt->dst_socket_id = _peer_socket_id;
2022-06-07 09:52:20 +08:00
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
2022-06-03 13:25:32 +08:00
pkt->ack_number = ack.ack_number;
pkt->storeToData();
_send_buf->drop(ack.last_ack_pkt_seq_number);
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
// TraceL<<"ack number "<<ack.ack_number;
}
2022-06-07 09:52:20 +08:00
void SrtTransport::sendMsgDropReq(uint32_t first, uint32_t last) {
2022-06-04 11:23:18 +08:00
MsgDropReqPacket::Ptr pkt = std::make_shared<MsgDropReqPacket>();
pkt->dst_socket_id = _peer_socket_id;
2022-06-07 09:52:20 +08:00
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
2022-06-04 11:23:18 +08:00
pkt->first_pkt_seq_num = first;
pkt->last_pkt_seq_num = last;
pkt->storeToData();
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleNAK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
NAKPacket pkt;
2022-06-07 09:52:20 +08:00
pkt.loadFromData(buf, len);
bool empty = false;
bool flush = false;
2022-07-06 21:40:24 +08:00
for (auto& it : pkt.lost_list) {
2022-06-07 09:52:20 +08:00
if (pkt.lost_list.back() == it) {
flush = true;
}
empty = true;
2022-06-07 09:52:20 +08:00
auto re_list = _send_buf->findPacketBySeq(it.first, it.second - 1);
2022-07-06 21:40:24 +08:00
for (auto& pkt : re_list) {
pkt->R = 1;
pkt->storeToHeader();
2022-06-07 09:52:20 +08:00
sendPacket(pkt, flush);
empty = false;
}
2022-06-07 09:52:20 +08:00
if (empty) {
sendMsgDropReq(it.first, it.second - 1);
}
}
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleCongestionWarning(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
TraceL;
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleShutDown(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
TraceL;
onShutdown(SockException(Err_shutdown, "peer close connection"));
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleDropReq(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
MsgDropReqPacket pkt;
2022-06-07 09:52:20 +08:00
pkt.loadFromData(buf, len);
std::list<DataPacket::Ptr> list;
2022-06-07 09:52:20 +08:00
// TraceL<<"drop "<<pkt.first_pkt_seq_num<<" last "<<pkt.last_pkt_seq_num;
_recv_buf->drop(pkt.first_pkt_seq_num, pkt.last_pkt_seq_num, list);
//checkAndSendAckNak();
2022-06-07 09:52:20 +08:00
if (list.empty()) {
return;
}
uint32_t max_seq = 0;
2022-07-06 21:40:24 +08:00
for (auto& data : list) {
max_seq = data->packet_seq_number;
if (_last_pkt_seq + 1 != data->packet_seq_number) {
TraceL << "pkt lost " << _last_pkt_seq + 1 << "->" << data->packet_seq_number;
}
_last_pkt_seq = data->packet_seq_number;
onSRTData(std::move(data));
}
2022-06-11 23:04:24 +08:00
/*
_recv_nack.drop(max_seq);
auto lost = _recv_buf->getLostSeq();
_recv_nack.update(_now, lost);
lost.clear();
_recv_nack.getLostList(_now, _rtt, _rtt_variance, lost);
if (!lost.empty()) {
sendNAKPacket(lost);
// TraceL << "check lost send nack";
}
2022-06-11 23:04:24 +08:00
*/
}
void SrtTransport::checkAndSendAckNak(){
2022-06-07 09:52:20 +08:00
auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if (nak_interval <= 20 * 1000) {
nak_interval = 20 * 1000;
}
2022-06-07 09:52:20 +08:00
if (_nak_ticker.elapsedTime(_now) > nak_interval) {
auto lost = _recv_buf->getLostSeq();
2022-06-07 09:52:20 +08:00
if (!lost.empty()) {
sendNAKPacket(lost);
}
_nak_ticker.resetTime(_now);
}
2022-06-07 09:52:20 +08:00
if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
_light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now);
2022-06-07 09:52:20 +08:00
// send a ack per 10 ms for receiver
sendACKPacket();
2022-06-07 09:52:20 +08:00
} else {
if (_light_ack_pkt_count >= 64) {
// for high bitrate stream send light ack
2022-06-07 09:52:20 +08:00
// TODO
sendLightACKPacket();
2022-06-07 09:52:20 +08:00
TraceL << "send light ack";
}
_light_ack_pkt_count = 0;
}
_light_ack_pkt_count++;
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleUserDefinedType(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
TraceL;
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleACKACK(uint8_t *buf, int len, struct sockaddr_storage *addr) {
// TraceL;
2022-06-03 13:25:32 +08:00
ACKACKPacket::Ptr pkt = std::make_shared<ACKACKPacket>();
2022-06-07 09:52:20 +08:00
pkt->loadFromData(buf, len);
2022-06-03 13:25:32 +08:00
if(_ack_send_timestamp.find(pkt->ack_number)!=_ack_send_timestamp.end()){
uint32_t rtt = DurationCountMicroseconds(_now - _ack_send_timestamp[pkt->ack_number]);
_rtt_variance = (3 * _rtt_variance + abs((long)_rtt - (long)rtt)) / 4;
_rtt = (7 * rtt + _rtt) / 8;
// TraceL<<" rtt:"<<_rtt<<" rtt variance:"<<_rtt_variance;
_ack_send_timestamp.erase(pkt->ack_number);
2022-06-03 13:25:32 +08:00
if(_last_recv_ackack_seq_num < pkt->ack_number){
_last_recv_ackack_seq_num = pkt->ack_number;
}else{
if((_last_recv_ackack_seq_num-pkt->ack_number)>(MAX_TS>>1)){
_last_recv_ackack_seq_num = pkt->ack_number;
}
}
if(_ack_send_timestamp.size()>1000){
// clear data
for(auto it = _ack_send_timestamp.begin(); it != _ack_send_timestamp.end();){
if(DurationCountMicroseconds(_now-it->second)>5e6){
// 超过五秒没有ackack 丢弃
it = _ack_send_timestamp.erase(it);
}else{
it++;
}
}
}
}
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handlePeerError(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
TraceL;
}
void SrtTransport::sendACKPacket() {
uint32_t recv_rate = 0;
2022-06-07 09:52:20 +08:00
ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
2022-06-03 13:25:32 +08:00
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
2022-06-03 13:25:32 +08:00
pkt->ack_number = ++_ack_number_count;
pkt->last_ack_pkt_seq_number = _recv_buf->getExpectedSeq();
pkt->rtt = _rtt;
pkt->rtt_variance = _rtt_variance;
pkt->available_buf_size = _recv_buf->getAvailableBufferSize();
pkt->pkt_recv_rate = _pkt_recv_rate_context->getPacketRecvRate(recv_rate);
pkt->estimated_link_capacity = _estimated_link_capacity_context->getEstimatedLinkCapacity();
pkt->recv_rate = recv_rate;
2022-09-22 12:03:53 +08:00
if(0){
TraceL<<pkt->pkt_recv_rate<<" pkt/s "<<recv_rate<<" byte/s "<<pkt->estimated_link_capacity<<" pkt/s (cap) "<<pkt->available_buf_size<<" available buf";
//TraceL<<_pkt_recv_rate_context->dump();
//TraceL<<"recv estimated:";
//TraceL<< _pkt_recv_rate_context->dump();
//TraceL<<"recv queue:";
//TraceL<<_recv_buf->dump();
}
if(pkt->available_buf_size<2){
pkt->available_buf_size = 2;
2022-09-22 01:07:10 +08:00
}
2022-06-03 13:25:32 +08:00
pkt->storeToData();
_ack_send_timestamp[pkt->ack_number] = _now;
//_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
// TraceL<<"send ack "<<pkt->dump();
2022-06-18 13:01:04 +08:00
// TraceL<<_recv_buf->dump();
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
void SrtTransport::sendLightACKPacket() {
2022-06-07 09:52:20 +08:00
ACKPacket::Ptr pkt = std::make_shared<ACKPacket>();
2022-06-03 13:25:32 +08:00
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
2022-06-03 13:25:32 +08:00
pkt->ack_number = 0;
pkt->last_ack_pkt_seq_number = _recv_buf->getExpectedSeq();
pkt->rtt = 0;
pkt->rtt_variance = 0;
pkt->available_buf_size = 0;
pkt->pkt_recv_rate = 0;
pkt->estimated_link_capacity = 0;
pkt->recv_rate = 0;
pkt->storeToData();
//_last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number;
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
TraceL << "send ack " << pkt->dump();
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::sendNAKPacket(std::list<PacketQueue::LostPair> &lost_list) {
2022-06-03 13:25:32 +08:00
NAKPacket::Ptr pkt = std::make_shared<NAKPacket>();
std::list<PacketQueue::LostPair> tmp;
auto size = NAKPacket::getCIFSize(lost_list);
size_t paylaod_size = getPayloadSize();
if (size > paylaod_size) {
WarnL << "loss report cif size " << size;
size_t num = paylaod_size / 8;
size_t msgNum = (lost_list.size() + num - 1) / num;
decltype(lost_list.begin()) cur, next;
for (size_t i = 0; i < msgNum; ++i) {
cur = lost_list.begin();
std::advance(cur, i * num);
if (i == msgNum - 1) {
next = lost_list.end();
} else {
next = lost_list.begin();
std::advance(next, (i + 1) * num);
}
tmp.assign(cur, next);
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->lost_list = tmp;
pkt->storeToData();
sendControlPacket(pkt, true);
}
2022-06-03 13:25:32 +08:00
} else {
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->lost_list = lost_list;
pkt->storeToData();
sendControlPacket(pkt, true);
}
2022-06-03 13:25:32 +08:00
2022-06-07 09:52:20 +08:00
// TraceL<<"send NAK "<<pkt->dump();
2022-06-03 13:25:32 +08:00
}
2022-06-03 21:17:01 +08:00
2022-06-07 09:52:20 +08:00
void SrtTransport::sendShutDown() {
2022-06-03 21:17:01 +08:00
ShutDownPacket::Ptr pkt = std::make_shared<ShutDownPacket>();
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(_now - _start_timestamp);
pkt->storeToData();
2022-06-07 09:52:20 +08:00
sendControlPacket(pkt, true);
2022-06-03 21:17:01 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::handleDataPacket(uint8_t *buf, int len, struct sockaddr_storage *addr) {
2022-06-03 13:25:32 +08:00
DataPacket::Ptr pkt = std::make_shared<DataPacket>();
2022-06-07 09:52:20 +08:00
pkt->loadFromData(buf, len);
_estimated_link_capacity_context->inputPacket(_now,pkt);
std::list<DataPacket::Ptr> list;
2022-06-03 13:25:32 +08:00
//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;
2022-06-07 09:52:20 +08:00
_recv_buf->inputPacket(pkt, list);
if (list.empty()) {
// when no data ok send nack to sender immediately
} else {
uint32_t last_seq;
2022-07-06 21:40:24 +08:00
for (auto& data : list) {
last_seq = data->packet_seq_number;
if (_last_pkt_seq + 1 != data->packet_seq_number) {
TraceL << "pkt lost " << _last_pkt_seq + 1 << "->" << data->packet_seq_number;
}
_last_pkt_seq = data->packet_seq_number;
onSRTData(std::move(data));
}
2022-06-11 23:04:24 +08:00
//_recv_nack.drop(last_seq);
}
/*
auto lost = _recv_buf->getLostSeq();
_recv_nack.update(_now, lost);
lost.clear();
_recv_nack.getLostList(_now, _rtt, _rtt_variance, lost);
if (!lost.empty()) {
// TraceL << "check lost send nack immediately";
sendNAKPacket(lost);
2022-06-03 17:49:43 +08:00
}
2022-06-11 23:04:24 +08:00
*/
/*
2022-06-07 09:52:20 +08:00
auto nak_interval = (_rtt + _rtt_variance * 4) / 2;
if (nak_interval <= 20 * 1000) {
nak_interval = 20 * 1000;
}
2022-06-05 12:13:28 +08:00
2022-06-07 09:52:20 +08:00
if (_nak_ticker.elapsedTime(_now) > nak_interval) {
// Periodic NAK reports
2022-06-03 13:25:32 +08:00
auto lost = _recv_buf->getLostSeq();
2022-06-07 09:52:20 +08:00
if (!lost.empty()) {
sendNAKPacket(lost);
// TraceL<<"send NAK";
} else {
// TraceL<<"lost is empty";
2022-06-03 13:25:32 +08:00
}
_nak_ticker.resetTime(_now);
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
if (_ack_ticker.elapsedTime(_now) > 10 * 1000) {
2022-06-03 17:49:43 +08:00
_light_ack_pkt_count = 0;
_ack_ticker.resetTime(_now);
2022-06-07 09:52:20 +08:00
// send a ack per 10 ms for receiver
2022-06-03 17:49:43 +08:00
sendACKPacket();
2022-06-07 09:52:20 +08:00
} else {
if (_light_ack_pkt_count >= 64) {
2022-06-03 17:49:43 +08:00
// for high bitrate stream send light ack
2022-06-07 09:52:20 +08:00
// TODO
2022-06-03 17:49:43 +08:00
sendLightACKPacket();
2022-06-07 09:52:20 +08:00
TraceL << "send light ack";
2022-06-03 17:49:43 +08:00
}
_light_ack_pkt_count = 0;
2022-06-03 13:25:32 +08:00
}
2022-06-03 17:49:43 +08:00
_light_ack_pkt_count++;
*/
2022-06-07 09:52:20 +08:00
// bufCheckInterval();
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::sendDataPacket(DataPacket::Ptr pkt, char *buf, int len, bool flush) {
pkt->storeToData((uint8_t *)buf, len);
sendPacket(pkt, flush);
_send_buf->inputPacket(pkt);
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::sendControlPacket(ControlPacket::Ptr pkt, bool flush) {
sendPacket(pkt, flush);
}
void SrtTransport::sendPacket(Buffer::Ptr pkt, bool flush) {
if (_selected_session) {
auto tmp = _packet_pool.obtain2();
tmp->assign(pkt->data(), pkt->size());
_selected_session->setSendFlushFlag(flush);
_selected_session->send(std::move(tmp));
} else {
WarnL << "not reach this";
2022-06-03 13:25:32 +08:00
}
}
2022-06-07 09:52:20 +08:00
std::string SrtTransport::getIdentifier() {
2022-06-03 13:25:32 +08:00
return _selected_session ? _selected_session->getIdentifier() : "";
}
2022-06-07 09:52:20 +08:00
void SrtTransport::registerSelfHandshake() {
SrtTransportManager::Instance().addHandshakeItem(std::to_string(_sync_cookie), shared_from_this());
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::unregisterSelfHandshake() {
if (_sync_cookie == 0) {
2022-06-03 13:25:32 +08:00
return;
}
SrtTransportManager::Instance().removeHandshakeItem(std::to_string(_sync_cookie));
}
void SrtTransport::registerSelf() {
2022-06-07 09:52:20 +08:00
if (_socket_id == 0) {
2022-06-03 13:25:32 +08:00
return;
}
2022-06-07 09:52:20 +08:00
SrtTransportManager::Instance().addItem(std::to_string(_socket_id), shared_from_this());
2022-06-03 13:25:32 +08:00
}
2022-06-07 09:52:20 +08:00
void SrtTransport::unregisterSelf() {
2022-06-03 13:25:32 +08:00
SrtTransportManager::Instance().removeItem(std::to_string(_socket_id));
}
2022-06-07 09:52:20 +08:00
void SrtTransport::onShutdown(const SockException &ex) {
2022-06-03 21:17:01 +08:00
sendShutDown();
2022-06-03 13:25:32 +08:00
WarnL << ex.what();
unregisterSelfHandshake();
unregisterSelf();
for (auto &pr : _history_sessions) {
auto session = pr.second.lock();
if (session) {
session->shutdown(ex);
}
}
}
2022-06-07 09:52:20 +08:00
size_t SrtTransport::getPayloadSize() {
size_t ret = (_mtu - 28 - 16) / 188 * 188;
return ret;
}
2022-06-07 09:52:20 +08:00
void SrtTransport::onSendTSData(const Buffer::Ptr &buffer, bool flush) {
// TraceL;
DataPacket::Ptr pkt;
size_t payloadSize = getPayloadSize();
2022-06-07 09:52:20 +08:00
size_t size = buffer->size();
char *ptr = buffer->data();
char *end = buffer->data() + size;
2022-06-07 09:52:20 +08:00
while (ptr < end && size >= payloadSize) {
pkt = std::make_shared<DataPacket>();
pkt->f = 0;
2022-06-07 09:52:20 +08:00
pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3;
pkt->O = 0;
pkt->KK = 0;
pkt->R = 0;
pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
2022-06-07 09:52:20 +08:00
sendDataPacket(pkt, ptr, (int)payloadSize, flush);
ptr += payloadSize;
size -= payloadSize;
}
2022-06-07 09:52:20 +08:00
if (size > 0 && ptr < end) {
pkt = std::make_shared<DataPacket>();
pkt->f = 0;
2022-06-07 09:52:20 +08:00
pkt->packet_seq_number = _send_packet_seq_number & 0x7fffffff;
_send_packet_seq_number = (_send_packet_seq_number + 1) & 0x7fffffff;
pkt->PP = 3;
pkt->O = 0;
pkt->KK = 0;
pkt->R = 0;
pkt->msg_number = _send_msg_number++;
pkt->dst_socket_id = _peer_socket_id;
pkt->timestamp = DurationCountMicroseconds(SteadyClock::now() - _start_timestamp);
2022-06-07 09:52:20 +08:00
sendDataPacket(pkt, ptr, (int)size, flush);
}
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
//////////// SrtTransportManager //////////////////////////
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
SrtTransportManager &SrtTransportManager::Instance() {
static SrtTransportManager s_instance;
return s_instance;
}
void SrtTransportManager::addItem(const std::string &key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_mtx);
_map[key] = ptr;
}
SrtTransport::Ptr SrtTransportManager::getItem(const std::string &key) {
if (key.empty()) {
return nullptr;
}
std::lock_guard<std::mutex> lck(_mtx);
auto it = _map.find(key);
if (it == _map.end()) {
return nullptr;
}
return it->second.lock();
}
void SrtTransportManager::removeItem(const std::string &key) {
std::lock_guard<std::mutex> lck(_mtx);
_map.erase(key);
}
void SrtTransportManager::addHandshakeItem(const std::string &key, const SrtTransport::Ptr &ptr) {
std::lock_guard<std::mutex> lck(_handshake_mtx);
_handshake_map[key] = ptr;
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
void SrtTransportManager::removeHandshakeItem(const std::string &key) {
2022-06-07 09:52:20 +08:00
std::lock_guard<std::mutex> lck(_handshake_mtx);
2022-06-03 13:25:32 +08:00
_handshake_map.erase(key);
}
2022-06-07 09:52:20 +08:00
2022-06-03 13:25:32 +08:00
SrtTransport::Ptr SrtTransportManager::getHandshakeItem(const std::string &key) {
if (key.empty()) {
return nullptr;
}
std::lock_guard<std::mutex> lck(_handshake_mtx);
auto it = _handshake_map.find(key);
if (it == _handshake_map.end()) {
return nullptr;
}
return it->second.lock();
}
} // namespace SRT