izackp/Apollo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Properly catch exceptions in stream broadcast handlers to prevent unhandled exception crash/termination.

Browse Source
This commit is contained in:
Mathias Tillman
2022-04-07 08:20:47 +02:00
parent 0044ec1d52
commit a9b9d1bd09
1 changed files with 77 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
sunshine/stream.cpp
View File

@@ -963,63 +963,69 @@ void videoBroadcastThread(udp::socket &sock) {
fec_blocks[0] = payload; fec_blocks[0] = payload;
} }
auto blockIndex = 0; try {
std::for_each(fec_blocks_begin, fec_blocks_end, [&](std::string_view &current_payload) { auto blockIndex = 0;
auto packets = (current_payload.size() + (blocksize - 1)) / blocksize; std::for_each(fec_blocks_begin, fec_blocks_end, [&](std::string_view &current_payload) {
auto packets = (current_payload.size() + (blocksize - 1)) / blocksize;
for(int x = 0; x < packets; ++x) { for(int x = 0; x < packets; ++x) {
auto *inspect = (video_packet_raw_t *)&current_payload[x * blocksize]; auto *inspect = (video_packet_raw_t *)&current_payload[x * blocksize];
inspect->packet.frameIndex = packet->pts; inspect->packet.frameIndex = packet->pts;
inspect->packet.streamPacketIndex = ((uint32_t)lowseq + x) << 8; inspect->packet.streamPacketIndex = ((uint32_t)lowseq + x) << 8;
// Match multiFecFlags with Moonlight // Match multiFecFlags with Moonlight
inspect->packet.multiFecFlags = 0x10; inspect->packet.multiFecFlags = 0x10;
inspect->packet.multiFecBlocks = (blockIndex << 4) | lastBlockIndex; inspect->packet.multiFecBlocks = (blockIndex << 4) | lastBlockIndex;
if(x == 0) { if(x == 0) {
inspect->packet.flags |= FLAG_SOF; inspect->packet.flags |= FLAG_SOF;
}
if(x == packets - 1) {
inspect->packet.flags |= FLAG_EOF;
}
} }
if(x == packets - 1) { auto shards = fec::encode(current_payload, blocksize, fecPercentage, session->config.minRequiredFecPackets);
inspect->packet.flags |= FLAG_EOF;
// set FEC info now that we know for sure what our percentage will be for this frame
for(auto x = 0; x < shards.size(); ++x) {
auto *inspect = (video_packet_raw_t *)shards.data(x);
inspect->packet.fecInfo =
(x << 12 |
shards.data_shards << 22 |
shards.percentage << 4);
inspect->rtp.header = 0x80 | FLAG_EXTENSION;
inspect->rtp.sequenceNumber = util::endian::big<uint16_t>(lowseq + x);
inspect->packet.multiFecBlocks = (blockIndex << 4) | lastBlockIndex;
inspect->packet.frameIndex = packet->pts;
} }
}
auto shards = fec::encode(current_payload, blocksize, fecPercentage, session->config.minRequiredFecPackets); for(auto x = 0; x < shards.size(); ++x) {
sock.send_to(asio::buffer(shards[x]), session->video.peer);
}
// set FEC info now that we know for sure what our percentage will be for this frame if(packet->flags & AV_PKT_FLAG_KEY) {
for(auto x = 0; x < shards.size(); ++x) { BOOST_LOG(verbose) << "Key Frame ["sv << packet->pts << "] :: send ["sv << shards.size() << "] shards..."sv;
auto *inspect = (video_packet_raw_t *)shards.data(x); }
else {
BOOST_LOG(verbose) << "Frame ["sv << packet->pts << "] :: send ["sv << shards.size() << "] shards..."sv << std::endl;
}
inspect->packet.fecInfo = ++blockIndex;
(x << 12 | lowseq += shards.size();
shards.data_shards << 22 | });
shards.percentage << 4);
inspect->rtp.header = 0x80 | FLAG_EXTENSION; session->video.lowseq = lowseq;
inspect->rtp.sequenceNumber = util::endian::big<uint16_t>(lowseq + x); }
catch(const std::exception &e) {
inspect->packet.multiFecBlocks = (blockIndex << 4) | lastBlockIndex; BOOST_LOG(error) << "Broadcast video failed "sv << e.what();
inspect->packet.frameIndex = packet->pts; std::this_thread::sleep_for(100ms);
} }
for(auto x = 0; x < shards.size(); ++x) {
sock.send_to(asio::buffer(shards[x]), session->video.peer);
}
if(packet->flags & AV_PKT_FLAG_KEY) {
BOOST_LOG(verbose) << "Key Frame ["sv << packet->pts << "] :: send ["sv << shards.size() << "] shards..."sv;
}
else {
BOOST_LOG(verbose) << "Frame ["sv << packet->pts << "] :: send ["sv << shards.size() << "] shards..."sv << std::endl;
}
++blockIndex;
lowseq += shards.size();
});
session->video.lowseq = lowseq;
} }
shutdown_event->raise(true); shutdown_event->raise(true);
@@ -1077,29 +1083,35 @@ void audioBroadcastThread(udp::socket &sock) {
auto &shards_p = session->audio.shards_p; auto &shards_p = session->audio.shards_p;
std::copy_n(audio_packet->payload(), bytes, shards_p[sequenceNumber % RTPA_DATA_SHARDS]); std::copy_n(audio_packet->payload(), bytes, shards_p[sequenceNumber % RTPA_DATA_SHARDS]);
sock.send_to(asio::buffer((char *)audio_packet.get(), sizeof(audio_packet_raw_t) + bytes), session->audio.peer); try {
sock.send_to(asio::buffer((char *)audio_packet.get(), sizeof(audio_packet_raw_t) + bytes), session->audio.peer);
BOOST_LOG(verbose) << "Audio ["sv << sequenceNumber << "] :: send..."sv; BOOST_LOG(verbose) << "Audio ["sv << sequenceNumber << "] :: send..."sv;
auto &fec_packet = session->audio.fec_packet; auto &fec_packet = session->audio.fec_packet;
// initialize the FEC header at the beginning of the FEC block // initialize the FEC header at the beginning of the FEC block
if(sequenceNumber % RTPA_DATA_SHARDS == 0) { if(sequenceNumber % RTPA_DATA_SHARDS == 0) {
fec_packet->fecHeader.baseSequenceNumber = util::endian::big(sequenceNumber); fec_packet->fecHeader.baseSequenceNumber = util::endian::big(sequenceNumber);
fec_packet->fecHeader.baseTimestamp = util::endian::big(timestamp); fec_packet->fecHeader.baseTimestamp = util::endian::big(timestamp);
}
// generate parity shards at the end of the FEC block
if((sequenceNumber + 1) % RTPA_DATA_SHARDS == 0) {
reed_solomon_encode(rs.get(), shards_p.begin(), RTPA_TOTAL_SHARDS, bytes);
for(auto x = 0; x < RTPA_FEC_SHARDS; ++x) {
fec_packet->rtp.sequenceNumber = util::endian::big<std::uint16_t>(sequenceNumber + x + 1);
fec_packet->fecHeader.fecShardIndex = x;
memcpy(fec_packet->payload(), shards_p[RTPA_DATA_SHARDS + x], bytes);
sock.send_to(asio::buffer((char *)fec_packet.get(), sizeof(audio_fec_packet_raw_t) + bytes), session->audio.peer);
BOOST_LOG(verbose) << "Audio FEC ["sv << (sequenceNumber & ~(RTPA_DATA_SHARDS - 1)) << ' ' << x << "] :: send..."sv;
} }
// generate parity shards at the end of the FEC block
if((sequenceNumber + 1) % RTPA_DATA_SHARDS == 0) {
reed_solomon_encode(rs.get(), shards_p.begin(), RTPA_TOTAL_SHARDS, bytes);
for(auto x = 0; x < RTPA_FEC_SHARDS; ++x) {
fec_packet->rtp.sequenceNumber = util::endian::big<std::uint16_t>(sequenceNumber + x + 1);
fec_packet->fecHeader.fecShardIndex = x;
memcpy(fec_packet->payload(), shards_p[RTPA_DATA_SHARDS + x], bytes);
sock.send_to(asio::buffer((char *)fec_packet.get(), sizeof(audio_fec_packet_raw_t) + bytes), session->audio.peer);
BOOST_LOG(verbose) << "Audio FEC ["sv << (sequenceNumber & ~(RTPA_DATA_SHARDS - 1)) << ' ' << x << "] :: send..."sv;
}
}
}
catch(const std::exception &e) {
BOOST_LOG(error) << "Broadcast audio failed "sv << e.what();
std::this_thread::sleep_for(100ms);
} }
} }