/* Copyright (C) 2020 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org EDI output, UDP and TCP transports and their configuration */ /* This file is part of the ODR-mmbTools. This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this program. If not, see . */ #include "Transport.h" #include #include using namespace std; namespace edi { void configuration_t::print() const { etiLog.level(info) << "EDI Output"; etiLog.level(info) << " verbose " << verbose; for (auto edi_dest : destinations) { if (auto udp_dest = dynamic_pointer_cast(edi_dest)) { etiLog.level(info) << " UDP to " << udp_dest->dest_addr << ":" << udp_dest->dest_port; if (not udp_dest->source_addr.empty()) { etiLog.level(info) << " source " << udp_dest->source_addr; etiLog.level(info) << " ttl " << udp_dest->ttl; } etiLog.level(info) << " source port " << udp_dest->source_port; } else if (auto tcp_dest = dynamic_pointer_cast(edi_dest)) { etiLog.level(info) << " TCP listening on port " << tcp_dest->listen_port; etiLog.level(info) << " max frames queued " << tcp_dest->max_frames_queued; } else if (auto tcp_dest = dynamic_pointer_cast(edi_dest)) { etiLog.level(info) << " TCP client connecting to " << tcp_dest->dest_addr << ":" << tcp_dest->dest_port; etiLog.level(info) << " max frames queued " << tcp_dest->max_frames_queued; } else { throw logic_error("EDI destination not implemented"); } } if (interleaver_enabled()) { etiLog.level(info) << " interleave " << latency_frames * 24 << " ms"; } } Sender::Sender(const configuration_t& conf) : m_conf(conf), edi_pft(m_conf) { if (m_conf.verbose) { etiLog.log(info, "Setup EDI Output"); } for (const auto& edi_dest : m_conf.destinations) { if (const auto udp_dest = dynamic_pointer_cast(edi_dest)) { auto udp_socket = std::make_shared(udp_dest->source_port); if (not udp_dest->source_addr.empty()) { udp_socket->setMulticastSource(udp_dest->source_addr.c_str()); udp_socket->setMulticastTTL(udp_dest->ttl); } udp_sockets.emplace(udp_dest.get(), udp_socket); } else if (auto tcp_dest = dynamic_pointer_cast(edi_dest)) { auto dispatcher = make_shared(tcp_dest->max_frames_queued); dispatcher->start(tcp_dest->listen_port, "0.0.0.0"); tcp_dispatchers.emplace(tcp_dest.get(), dispatcher); } else if (auto tcp_dest = dynamic_pointer_cast(edi_dest)) { auto tcp_send_client = make_shared(tcp_dest->dest_addr, tcp_dest->dest_port); tcp_senders.emplace(tcp_dest.get(), tcp_send_client); } else { throw logic_error("EDI destination not implemented"); } } if (m_conf.interleaver_enabled()) { edi_interleaver.SetLatency(m_conf.latency_frames); } if (m_conf.dump) { edi_debug_file.open("./edi.debug"); } if (m_conf.verbose) { etiLog.log(info, "EDI output set up"); } } void Sender::write(const TagPacket& tagpacket) { // Assemble into one AF Packet edi::AFPacket af_packet = edi_afPacketiser.Assemble(tagpacket); if (m_conf.enable_pft) { // Apply PFT layer to AF Packet (Reed Solomon FEC and Fragmentation) vector edi_fragments = edi_pft.Assemble(af_packet); if (m_conf.verbose) { fprintf(stderr, "EDI Output: Number of PFT fragment before interleaver %zu\n", edi_fragments.size()); } if (m_conf.interleaver_enabled()) { edi_fragments = edi_interleaver.Interleave(edi_fragments); } if (m_conf.verbose) { fprintf(stderr, "EDI Output: Number of PFT fragments %zu\n", edi_fragments.size()); } /* Spread out the transmission of all fragments over 25% of the 24ms AF packet duration * to reduce the risk of losing a burst of fragments because of congestion. * * 25% was chosen so that other outputs still have time to do their thing. */ auto inter_fragment_wait_time = std::chrono::microseconds(0); if (edi_fragments.size() > 1) { inter_fragment_wait_time = std::chrono::microseconds(llrint(0.25 * 24000.0 / edi_fragments.size())); } // Send over ethernet for (auto& edi_frag : edi_fragments) { if (m_conf.dump) { ostream_iterator debug_iterator(edi_debug_file); copy(edi_frag.begin(), edi_frag.end(), debug_iterator); } for (auto& dest : m_conf.destinations) { if (const auto& udp_dest = dynamic_pointer_cast(dest)) { Socket::InetAddress addr; addr.resolveUdpDestination(udp_dest->dest_addr, udp_dest->dest_port); udp_sockets.at(udp_dest.get())->send(edi_frag, addr); } else if (auto tcp_dest = dynamic_pointer_cast(dest)) { tcp_dispatchers.at(tcp_dest.get())->write(edi_frag); } else if (auto tcp_dest = dynamic_pointer_cast(dest)) { tcp_senders.at(tcp_dest.get())->sendall(edi_frag); } else { throw logic_error("EDI destination not implemented"); } } std::this_thread::sleep_for(inter_fragment_wait_time); } } else { // Send over ethernet if (m_conf.dump) { ostream_iterator debug_iterator(edi_debug_file); copy(af_packet.begin(), af_packet.end(), debug_iterator); } for (auto& dest : m_conf.destinations) { if (const auto& udp_dest = dynamic_pointer_cast(dest)) { Socket::InetAddress addr; addr.resolveUdpDestination(udp_dest->dest_addr, udp_dest->dest_port); if (af_packet.size() > 1400 and not m_udp_fragmentation_warning_printed) { fprintf(stderr, "EDI Output: AF packet larger than 1400," " consider using PFT to avoid UP fragmentation.\n"); m_udp_fragmentation_warning_printed = true; } udp_sockets.at(udp_dest.get())->send(af_packet, addr); } else if (auto tcp_dest = dynamic_pointer_cast(dest)) { tcp_dispatchers.at(tcp_dest.get())->write(af_packet); } else if (auto tcp_dest = dynamic_pointer_cast(dest)) { tcp_senders.at(tcp_dest.get())->sendall(af_packet); } else { throw logic_error("EDI destination not implemented"); } } } } }