diff options
Diffstat (limited to 'src')
-rw-r--r-- | src/ConfigParser.cpp | 11 | ||||
-rw-r--r-- | src/DabMultiplexer.cpp | 230 | ||||
-rw-r--r-- | src/DabMultiplexer.h | 89 | ||||
-rw-r--r-- | src/DabMux.cpp | 52 | ||||
-rw-r--r-- | src/ManagementServer.cpp | 75 | ||||
-rw-r--r-- | src/ManagementServer.h | 20 | ||||
-rw-r--r-- | src/fig/FIG.h | 10 | ||||
-rw-r--r-- | src/fig/FIG0_10.cpp | 11 | ||||
-rw-r--r-- | src/fig/FIG0structs.h | 18 | ||||
-rw-r--r-- | src/fig/FIGCarousel.cpp | 7 | ||||
-rw-r--r-- | src/fig/FIGCarousel.h | 4 | ||||
-rw-r--r-- | src/utils.cpp | 25 | ||||
-rw-r--r-- | src/utils.h | 6 | ||||
-rw-r--r-- | src/zmq2edi/EDISender.cpp | 390 | ||||
-rw-r--r-- | src/zmq2edi/EDISender.h | 91 | ||||
-rw-r--r-- | src/zmq2edi/zmq2edi.cpp | 419 |
16 files changed, 1215 insertions, 243 deletions
diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp index 74e627b..7d166b6 100644 --- a/src/ConfigParser.cpp +++ b/src/ConfigParser.cpp @@ -36,16 +36,13 @@ # include "config.h" #endif -#include "dabOutput/dabOutput.h" #include "utils.h" -#include "DabMux.h" -#include "ManagementServer.h" #include "input/Edi.h" #include "input/Prbs.h" #include "input/Zmq.h" #include "input/File.h" #include "input/Udp.h" -#include "Eti.h" +#include "fig/FIG0structs.h" #include <boost/property_tree/ptree.hpp> #include <boost/algorithm/string/classification.hpp> #include <boost/algorithm/string/split.hpp> @@ -63,6 +60,12 @@ using namespace std; using boost::property_tree::ptree; using boost::property_tree::ptree_error; +constexpr uint16_t DEFAULT_DATA_BITRATE = 384; +constexpr uint16_t DEFAULT_PACKET_BITRATE = 32; + +constexpr uint32_t DEFAULT_SERVICE_ID = 50; + + static void setup_subchannel_from_ptree(shared_ptr<DabSubchannel>& subchan, const ptree &pt, std::shared_ptr<dabEnsemble> ensemble, diff --git a/src/DabMultiplexer.cpp b/src/DabMultiplexer.cpp index bd1c909..52f053a 100644 --- a/src/DabMultiplexer.cpp +++ b/src/DabMultiplexer.cpp @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2024 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li */ /* @@ -23,11 +23,14 @@ along with ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. */ +#include <cmath> #include <set> #include <memory> #include "DabMultiplexer.h" #include "ConfigParser.h" -#include "fig/FIG.h" +#include "ManagementServer.h" +#include "crc.h" +#include "utils.h" using namespace std; @@ -44,16 +47,101 @@ static vector<string> split_pipe_separated_string(const std::string& s) return components; } -DabMultiplexer::DabMultiplexer( - boost::property_tree::ptree pt) : +uint64_t MuxTime::init(uint32_t tist_at_fct0_ms, double tist_offset) +{ + // Things we must guarantee, up to granularity of 24ms: + // Difference between current time and EDI time = tist_offset + // TIST of frame 0 = tist_at_fct0_ms + // In order to achieve the second, we calculate the initial + // counter value so that FCT0 corresponds to the desired TIST. + // + // Changing the tist_offset at runtime will throw off the TIST@FCT0 value + m_tist_offset_ms = std::lround(tist_offset * 1000); + + using Sec = chrono::seconds; + const auto now = chrono::system_clock::now() + + chrono::milliseconds(std::lround(tist_offset * 1000.0)); + + const auto offset = now - chrono::time_point_cast<Sec>(now); + if (offset >= chrono::seconds(1)) { + throw std::logic_error("Invalid startup offset calculation for TIST! " + + to_string(chrono::duration_cast<chrono::milliseconds>(offset).count()) + + " ms"); + } + const time_t t_now = chrono::system_clock::to_time_t(chrono::time_point_cast<Sec>(now)); + const auto offset_ms = chrono::duration_cast<chrono::milliseconds>(offset).count(); + + m_edi_time = t_now; + m_pps_offset_ms = std::lround(offset_ms / 24.0) * 24; + + const auto counter_offset = tist_at_fct0_ms / 24; + const auto offset_as_count = m_pps_offset_ms / 24; + + etiLog.level(debug) << "Init " << counter_offset << " " << offset_as_count; + + return (250 - counter_offset + offset_as_count) % 250; +} + +constexpr int TIMESTAMP_LEVEL_2_SHIFT = 14; + +void MuxTime::increment_timestamp() +{ + m_pps_offset_ms += 24; + if (m_pps_offset_ms >= 1000) { + m_pps_offset_ms -= 1000; + m_edi_time += 1; + + // Also update MNSC time for next time FP==0 + mnsc_increment_time = true; + } +} + +void MuxTime::set_tist_offset(double new_tist_offset) +{ + const int32_t new_tist_offset_ms = std::lround(new_tist_offset * 1000.0); + int32_t delta = m_tist_offset_ms - new_tist_offset_ms; + if (delta > 0) { + while (delta > 0) { + increment_timestamp(); + delta -= 24; + } + } + else if (delta < 0) { + while (delta < 0) { + m_edi_time -= 1; + delta += 1000; + } + // compensate the we subtracted too much + while (delta > 0) { + increment_timestamp(); + delta -= 24; + } + } + m_tist_offset_ms = new_tist_offset_ms; +} + +std::pair<uint32_t, std::time_t> MuxTime::get_tist_seconds() +{ + auto timestamp = m_pps_offset_ms * 16384; + return {timestamp % 0xfa0000, m_edi_time}; +} + +std::pair<uint32_t, std::time_t> MuxTime::get_milliseconds_seconds() +{ + auto tist_seconds = get_tist_seconds(); + return {tist_seconds.first >> TIMESTAMP_LEVEL_2_SHIFT, tist_seconds.second}; +} + + +DabMultiplexer::DabMultiplexer(boost::property_tree::ptree pt) : RemoteControllable("mux"), m_pt(pt), + m_time(), ensemble(std::make_shared<dabEnsemble>()), m_clock_tai(split_pipe_separated_string(pt.get("general.tai_clock_bulletins", ""))), - fig_carousel(ensemble) + fig_carousel(ensemble, [&]() { return m_time.get_milliseconds_seconds(); }) { RC_ADD_PARAMETER(frames, "Show number of frames generated [read-only]"); - RC_ADD_PARAMETER(tist_offset, "Timestamp offset in integral number of seconds"); rcs.enrol(&m_clock_tai); } @@ -99,59 +187,20 @@ void DabMultiplexer::prepare(bool require_tai_clock) throw MuxInitException(); } - /* At startup, derive edi_time, TIST and CIF count such that there is - * a consistency across mux restarts. Ensure edi_time and TIST represent - * current time. - * - * FCT and DLFC are directly derived from m_currentFrame. - * Every 6s, FCT overflows. DLFC overflows at 5000 every 120s. - * - * Keep a granularity of 24ms, which corresponds to the duration of an ETI - * frame, to get nicer timestamps. - */ - using Sec = chrono::seconds; - const auto now = chrono::system_clock::now(); - const time_t t_now = chrono::system_clock::to_time_t(chrono::time_point_cast<Sec>(now)); - - m_edi_time = t_now - (t_now % 6); - m_currentFrame = 0; - time_t edi_time_at_cif0 = t_now - (t_now % 120); - while (edi_time_at_cif0 < m_edi_time) { - edi_time_at_cif0 += 6; - m_currentFrame += 250; - } - - if (edi_time_at_cif0 != m_edi_time) { - throw std::logic_error("Invalid startup offset calculation for CIF!"); - } - - const auto offset = now - chrono::time_point_cast<Sec>(now); - if (offset >= chrono::seconds(1)) { - throw std::logic_error("Invalid startup offset calculation for TIST! " + - to_string(chrono::duration_cast<chrono::milliseconds>(offset).count()) + - " ms"); - } - - int64_t offset_ms = chrono::duration_cast<chrono::milliseconds>(offset).count(); - offset_ms += 1000 * (t_now - m_edi_time); - - m_timestamp = 0; - while (offset_ms >= 24) { - increment_timestamp(); - m_currentFrame++; - offset_ms -= 24; - } - - mnsc_increment_time = false; + const uint32_t tist_at_fct0_ms = m_pt.get<double>("general.tist_at_fct0", 0); + currentFrame = m_time.init(tist_at_fct0_ms, m_pt.get<double>("general.tist_offset", 0.0)); + m_time.mnsc_increment_time = false; bool tist_enabled = m_pt.get("general.tist", false); - m_tist_offset = m_pt.get<int>("general.tist_offset", 0); - mnsc_time = m_edi_time + m_tist_offset; + auto tist_edi_time = m_time.get_tist_seconds(); + const auto timestamp = tist_edi_time.first; + const auto edi_time = tist_edi_time.second; + m_time.mnsc_time = edi_time; etiLog.log(info, "Startup CIF Count %i with timestamp: %d + %f", - m_currentFrame, m_edi_time, - (m_timestamp & 0xFFFFFF) / 16384000.0); + currentFrame, edi_time, + (timestamp & 0xFFFFFF) / 16384000.0); // Try to load offset once @@ -391,17 +440,6 @@ void DabMultiplexer::prepare_data_inputs() } } -void DabMultiplexer::increment_timestamp() -{ - m_timestamp += 24 << 14; // Shift 24ms by 14 to Timestamp level 2 - if (m_timestamp > 0xf9FFff) { - m_timestamp -= 0xfa0000; // Substract 16384000, corresponding to one second - m_edi_time += 1; - - // Also update MNSC time for next time FP==0 - mnsc_increment_time = true; - } -} /* Each call creates one ETI frame */ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs) @@ -432,9 +470,14 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs etiLog.level(error) << "Could not get UTC-TAI offset for EDI timestamp"; } } - update_dab_time(); - const auto edi_time = m_edi_time + m_tist_offset; + auto tist_edi_time = m_time.get_tist_seconds(); + const auto timestamp = tist_edi_time.first; + const auto edi_time = tist_edi_time.second; + /* + etiLog.level(debug) << "Frame " << currentFrame << " " << edi_time << + " + " << (timestamp >> TIMESTAMP_LEVEL_2_SHIFT); + */ // Initialise the ETI frame memset(etiFrame, 0, 6144); @@ -450,7 +493,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs //****** Field FSYNC *****// // See ETS 300 799, 6.2.1.2 - if ((m_currentFrame & 1) == 0) { + if ((currentFrame & 1) == 0) { etiSync->FSYNC = ETI_FSYNC1; } else { @@ -468,9 +511,8 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs eti_FC *fc = (eti_FC *) &etiFrame[4]; //****** FCT ******// - // Incremente for each frame, overflows at 249 - fc->FCT = m_currentFrame % 250; - edi_tagDETI.dlfc = m_currentFrame % 5000; + fc->FCT = currentFrame % 250; + edi_tagDETI.dlfc = currentFrame % 5000; //****** FICF ******// // Fast Information Channel Flag, 1 bit, =1 if FIC present @@ -487,7 +529,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs /* Frame Phase, 3 bit counter, tells the COFDM generator * when to insert the TII. Is also used by the MNSC. */ - fc->FP = edi_tagDETI.fp = m_currentFrame & 0x7; + fc->FP = edi_tagDETI.fp = currentFrame & 0x7; //****** MID ******// //Mode Identity, 2 bits, 01 ModeI, 10 modeII, 11 ModeIII, 00 ModeIV @@ -562,7 +604,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs eoh->MNSC = 0; struct tm time_tm; - gmtime_r(&mnsc_time, &time_tm); + gmtime_r(&m_time.mnsc_time, &time_tm); switch (fc->FP & 0x3) { case 0: @@ -574,10 +616,10 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs mnsc->rfa = 0; } - if (mnsc_increment_time) + if (m_time.mnsc_increment_time) { - mnsc_increment_time = false; - mnsc_time += 1; + m_time.mnsc_increment_time = false; + m_time.mnsc_time += 1; } break; case 1: @@ -621,7 +663,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs // Insert all FIBs const bool fib3_present = (ensemble->transmission_mode == TransmissionMode_e::TM_III); - index += fig_carousel.write_fibs(&etiFrame[index], m_currentFrame, fib3_present); + index += fig_carousel.write_fibs(&etiFrame[index], currentFrame, fib3_present); /********************************************************************** ****** Input Data Reading ******************************************* @@ -633,12 +675,13 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs int sizeSubchannel = subchannel->getSizeByte(); // no need to check enableTist because we always increment the timestamp int result = subchannel->readFrame(&etiFrame[index], - sizeSubchannel, edi_time, tai_utc_offset, m_timestamp); + sizeSubchannel, + edi_time, tai_utc_offset, timestamp); if (result < 0) { etiLog.log(info, "Subchannel %d read failed at ETI frame number: %d", - subchannel->id, m_currentFrame); + subchannel->id, currentFrame); } // save pointer to Audio or Data Stream into correct TagESTn for EDI @@ -678,8 +721,8 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs bool enableTist = m_pt.get("general.tist", false); if (enableTist) { - tist->TIST = htonl(m_timestamp) | 0xff; - edi_tagDETI.tsta = m_timestamp & 0xffffff; + tist->TIST = htonl(timestamp) | 0xff; + edi_tagDETI.tsta = timestamp & 0xffffff; } else { tist->TIST = htonl(0xffffff) | 0xff; @@ -700,7 +743,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs output->setMetadata(md_edi_time); shared_ptr<OutputMetadata> md_dlfc = - make_shared<OutputMetadataDLFC>(m_currentFrame % 5000); + make_shared<OutputMetadataDLFC>(currentFrame % 5000); output->setMetadata(md_dlfc); } } @@ -716,8 +759,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs Approximate 8 ms 1 ms 3,91 us 488 ns 61 ns time resolution */ - - increment_timestamp(); + m_time.increment_timestamp(); /********************************************************************** *********** Section FRPD ***************************************** @@ -759,6 +801,12 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs } edi_sender->write(edi_tagpacket); + + for (const auto& stat : edi_sender->get_tcp_server_stats()) { + get_mgmt_server().update_edi_tcp_output_stat( + stat.listen_port, + stat.stats.size()); + } } #if _DEBUG @@ -769,7 +817,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs if (enableTist) { etiLog.log(info, "ETI frame number %i Timestamp: %d + %f", m_currentFrame, edi_time, - (m_timestamp & 0xFFFFFF) / 16384000.0); + (timestamp & 0xFFFFFF) / 16384000.0); } else { etiLog.log(info, "ETI frame number %i Time: %d, no TIST", @@ -778,7 +826,7 @@ void DabMultiplexer::mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs } #endif - m_currentFrame++; + currentFrame++; } void DabMultiplexer::print_info() @@ -799,7 +847,7 @@ void DabMultiplexer::set_parameter(const std::string& parameter, throw ParameterError(ss.str()); } else if (parameter == "tist_offset") { - m_tist_offset = std::stoi(value); + m_time.set_tist_offset(std::stod(value)); } else { stringstream ss; @@ -814,10 +862,10 @@ const std::string DabMultiplexer::get_parameter(const std::string& parameter) co { stringstream ss; if (parameter == "frames") { - ss << m_currentFrame; + ss << currentFrame; } else if (parameter == "tist_offset") { - ss << m_tist_offset; + ss << m_time.tist_offset(); } else { ss << "Parameter '" << parameter << @@ -831,8 +879,8 @@ const std::string DabMultiplexer::get_parameter(const std::string& parameter) co const json::map_t DabMultiplexer::get_all_values() const { json::map_t map; - map["frames"].v = m_currentFrame; - map["tist_offset"].v = m_tist_offset; + map["frames"].v = currentFrame; + map["tist_offset"].v = m_time.tist_offset(); return map; } diff --git a/src/DabMultiplexer.h b/src/DabMultiplexer.h index 89d547e..9306eed 100644 --- a/src/DabMultiplexer.h +++ b/src/DabMultiplexer.h @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2024 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li */ /* @@ -30,21 +30,12 @@ #endif #include "dabOutput/dabOutput.h" -#include "edioutput/TagItems.h" -#include "edioutput/TagPacket.h" -#include "edioutput/AFPacket.h" #include "edioutput/Transport.h" #include "fig/FIGCarousel.h" -#include "crc.h" -#include "utils.h" -#include "Socket.h" -#include "PcDebug.h" #include "MuxElements.h" #include "RemoteControl.h" -#include "Eti.h" #include "ClockTAI.h" #include <vector> -#include <chrono> #include <memory> #include <string> #include <memory> @@ -52,6 +43,37 @@ constexpr uint32_t ETI_FSYNC1 = 0x49C5F8; +class MuxTime { + private: + std::time_t m_edi_time = 0; + uint32_t m_pps_offset_ms = 0; + int64_t m_tist_offset_ms = 0; + + public: + std::pair<uint32_t, std::time_t> get_tist_seconds(); + std::pair<uint32_t, std::time_t> get_milliseconds_seconds(); + + + /* Pre v3 odr-dabmux did the MNSC calculation differently, + * which works with the easydabv2. The rework in odr-dabmux, + * deriving MNSC time from EDI time broke this. + * + * That's why we're now tracking MNSC time in separate variables, + * to get the same behaviour back. + * + * I'm not aware of any devices using MNSC time besides the + * easydab. ODR-DabMod now considers EDI seconds or ZMQ metadata. + */ + bool mnsc_increment_time = false; + std::time_t mnsc_time = 0; + + /* Setup the time and return the initial currentFrame counter value */ + uint64_t init(uint32_t tist_at_fct0_ms, double tist_offset); + void increment_timestamp(); + double tist_offset() const { return m_tist_offset_ms / 1000.0; } + void set_tist_offset(double new_tist_offset); +}; + class DabMultiplexer : public RemoteControllable { public: DabMultiplexer(boost::property_tree::ptree pt); @@ -61,8 +83,6 @@ class DabMultiplexer : public RemoteControllable { void prepare(bool require_tai_clock); - uint64_t getCurrentFrame() const { return m_currentFrame; } - void mux_frame(std::vector<std::shared_ptr<DabOutput> >& outputs); void print_info(void); @@ -82,60 +102,19 @@ class DabMultiplexer : public RemoteControllable { void prepare_subchannels(void); void prepare_services_components(void); void prepare_data_inputs(void); - void increment_timestamp(void); boost::property_tree::ptree m_pt; - uint32_t m_timestamp = 0; - std::time_t m_edi_time = 0; - - /* Pre v3 odr-dabmux did the MNSC calculation differently, - * which works with the easydabv2. The rework in odr-dabmux, - * deriving MNSC time from EDI time broke this. - * - * That's why we're now tracking MNSC time in separate variables, - * to get the same behaviour back. - * - * I'm not aware of any devices using MNSC time besides the - * easydab. ODR-DabMod now considers EDI seconds or ZMQ metadata. - */ - bool mnsc_increment_time = false; - std::time_t mnsc_time = 0; + MuxTime m_time; + uint64_t currentFrame = 0; edi::configuration_t edi_conf; std::shared_ptr<edi::Sender> edi_sender; - uint64_t m_currentFrame = 0; - std::shared_ptr<dabEnsemble> ensemble; - int m_tist_offset = 0; bool m_tai_clock_required = false; ClockTAI m_clock_tai; - /* New FIG Carousel */ FIC::FIGCarousel fig_carousel; }; - -// DAB Mode -#define DEFAULT_DAB_MODE 1 - -// Taille de la trame de donnee, sous-canal 3, nb de paquets de 64bits, -// STL3 * 8 = x kbytes par trame ETI - -// Data bitrate in kbits/s. Must be 64 kb/s multiple. -#define DEFAULT_DATA_BITRATE 384 -#define DEFAULT_PACKET_BITRATE 32 - -/* default ensemble parameters. Label must be max 16 chars, short label - * a subset of the label, max 8 chars - */ -#define DEFAULT_ENSEMBLE_LABEL "ODR Dab Mux" -#define DEFAULT_ENSEMBLE_SHORT_LABEL "ODRMux" -#define DEFAULT_ENSEMBLE_ID 0xc000 -#define DEFAULT_ENSEMBLE_ECC 0xa1 - -// start value for default service IDs (if not overridden by configuration) -#define DEFAULT_SERVICE_ID 50 -#define DEFAULT_PACKET_ADDRESS 0 - diff --git a/src/DabMux.cpp b/src/DabMux.cpp index 1a367da..bf525c1 100644 --- a/src/DabMux.cpp +++ b/src/DabMux.cpp @@ -327,6 +327,38 @@ int main(int argc, char *argv[]) if (outputuid == "edi") { ptree pt_edi = pt_outputs.get_child("edi"); + bool default_enable_pft = pt_edi.get<bool>("enable_pft", false); + edi_conf.verbose = pt_edi.get<bool>("verbose", false); + + unsigned int default_fec = pt_edi.get<unsigned int>("fec", 3); + unsigned int default_chunk_len = pt_edi.get<unsigned int>("chunk_len", 207); + + auto check_spreading_factor = [](int percent) { + if (percent < 0) { + throw std::runtime_error("EDI output: negative packet_spread value is invalid."); + } + double factor = (double)percent / 100.0; + if (factor > 30000) { + throw std::runtime_error("EDI output: interleaving set for more than 30 seconds!"); + } + return factor; + }; + + double default_spreading_factor = check_spreading_factor(pt_edi.get<int>("packet_spread", 95)); + + using pt_t = boost::property_tree::basic_ptree<std::basic_string<char>, std::basic_string<char>>; + auto handle_overrides = [&](edi::pft_settings_t& pft_settings, pt_t pt) { + pft_settings.chunk_len = pt.get<unsigned int>("chunk_len", default_chunk_len); + pft_settings.enable_pft = pt.get<bool>("enable_pft", default_enable_pft); + pft_settings.fec = pt.get<unsigned int>("fec", default_fec); + pft_settings.fragment_spreading_factor = default_spreading_factor; + auto override_spread_percent = pt.get_optional<int>("packet_spread"); + if (override_spread_percent) { + pft_settings.fragment_spreading_factor = check_spreading_factor(*override_spread_percent); + } + pft_settings.verbose = pt.get<bool>("verbose", edi_conf.verbose); + }; + for (auto pt_edi_dest : pt_edi.get_child("destinations")) { const auto proto = pt_edi_dest.second.get<string>("protocol", "udp"); if (proto == "udp") { @@ -346,6 +378,8 @@ int main(int argc, char *argv[]) dest->dest_port = pt_edi.get<unsigned int>("port"); } + handle_overrides(dest->pft_settings, pt_edi_dest.second); + edi_conf.destinations.push_back(dest); } else if (proto == "tcp") { @@ -355,6 +389,8 @@ int main(int argc, char *argv[]) double preroll = pt_edi_dest.second.get<double>("preroll-burst", 0.0); dest->tcp_server_preroll_buffers = ceil(preroll / 24e-3); + handle_overrides(dest->pft_settings, pt_edi_dest.second); + edi_conf.destinations.push_back(dest); } else { @@ -362,22 +398,6 @@ int main(int argc, char *argv[]) } } - edi_conf.dump = pt_edi.get<bool>("dump", false); - edi_conf.enable_pft = pt_edi.get<bool>("enable_pft", false); - edi_conf.verbose = pt_edi.get<bool>("verbose", false); - - edi_conf.fec = pt_edi.get<unsigned int>("fec", 3); - edi_conf.chunk_len = pt_edi.get<unsigned int>("chunk_len", 207); - - int spread_percent = pt_edi.get<int>("packet_spread", 95); - if (spread_percent < 0) { - throw std::runtime_error("EDI output: negative packet_spread value is invalid."); - } - edi_conf.fragment_spreading_factor = (double)spread_percent / 100.0; - if (edi_conf.fragment_spreading_factor > 30000) { - throw std::runtime_error("EDI output: interleaving set for more than 30 seconds!"); - } - edi_conf.tagpacket_alignment = pt_edi.get<unsigned int>("tagpacket_alignment", 8); mux.set_edi_config(edi_conf); diff --git a/src/ManagementServer.cpp b/src/ManagementServer.cpp index 568e80e..dff093a 100644 --- a/src/ManagementServer.cpp +++ b/src/ManagementServer.cpp @@ -2,7 +2,7 @@ Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2018 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org @@ -28,13 +28,12 @@ along with ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. */ -#include <errno.h> -#include <string.h> -#include <math.h> -#include <stdint.h> -#include <limits> #include <sstream> #include <algorithm> +#include <cstring> +#include <cmath> +#include <cstdint> +#include <limits> #include <boost/version.hpp> #include "ManagementServer.h" #include "Log.h" @@ -127,37 +126,42 @@ ManagementServer& get_mgmt_server() */ } -void ManagementServer::registerInput(InputStat* is) +void ManagementServer::register_input(InputStat* is) { unique_lock<mutex> lock(m_statsmutex); std::string id(is->get_name()); - if (m_inputStats.count(id) == 1) { + if (m_input_stats.count(id) == 1) { etiLog.level(error) << "Double registration in MGMT Server with id '" << id << "'"; return; } - m_inputStats[id] = is; + m_input_stats[id] = is; } -void ManagementServer::unregisterInput(std::string id) +void ManagementServer::unregister_input(std::string id) { unique_lock<mutex> lock(m_statsmutex); - if (m_inputStats.count(id) == 1) { - m_inputStats.erase(id); + if (m_input_stats.count(id) == 1) { + m_input_stats.erase(id); } } +// outputs will never disappear, no need to have a "remove" logic +void ManagementServer::update_edi_tcp_output_stat(uint16_t listen_port, size_t num_connections) +{ + m_output_stats[listen_port] = num_connections; +} bool ManagementServer::isInputRegistered(std::string& id) { unique_lock<mutex> lock(m_statsmutex); - if (m_inputStats.count(id) == 0) { + if (m_input_stats.count(id) == 0) { etiLog.level(error) << "Management Server: id '" << id << "' does was not registered"; @@ -166,7 +170,7 @@ bool ManagementServer::isInputRegistered(std::string& id) return true; } -std::string ManagementServer::getStatConfigJSON() +std::string ManagementServer::get_input_config_json() { unique_lock<mutex> lock(m_statsmutex); @@ -175,7 +179,7 @@ std::string ManagementServer::getStatConfigJSON() std::map<std::string,InputStat*>::iterator iter; int i = 0; - for(iter = m_inputStats.begin(); iter != m_inputStats.end(); + for (iter = m_input_stats.begin(); iter != m_input_stats.end(); ++iter, i++) { std::string id = iter->first; @@ -192,16 +196,15 @@ std::string ManagementServer::getStatConfigJSON() return ss.str(); } -std::string ManagementServer::getValuesJSON() +std::string ManagementServer::get_input_values_json() { unique_lock<mutex> lock(m_statsmutex); std::ostringstream ss; ss << "{ \"values\" : {\n"; - std::map<std::string,InputStat*>::iterator iter; int i = 0; - for(iter = m_inputStats.begin(); iter != m_inputStats.end(); + for (auto iter = m_input_stats.begin(); iter != m_input_stats.end(); ++iter, i++) { const std::string& id = iter->first; @@ -220,6 +223,31 @@ std::string ManagementServer::getValuesJSON() return ss.str(); } +std::string ManagementServer::get_output_values_json() +{ + unique_lock<mutex> lock(m_statsmutex); + + std::ostringstream ss; + ss << "{ \"output_values\" : {\n"; + + int i = 0; + for (auto iter = m_output_stats.begin(); iter != m_output_stats.end(); + ++iter, i++) + { + auto listen_port = iter->first; + auto num_connections = iter->second; + if (i > 0) { + ss << " ,\n"; + } + ss << " \"edi_tcp_" << listen_port << "\" : { \"num_connections\": " << + num_connections << "} "; + } + + ss << "}\n}\n"; + + return ss.str(); +} + ManagementServer::ManagementServer() : m_zmq_context(), m_zmq_sock(m_zmq_context, ZMQ_REP), @@ -323,10 +351,13 @@ void ManagementServer::handle_message(zmq::message_t& zmq_message) << "}\n"; } else if (data == "config") { - answer << getStatConfigJSON(); + answer << get_input_config_json(); } else if (data == "values") { - answer << getValuesJSON(); + answer << get_input_values_json(); + } + else if (data == "output_values") { + answer << get_output_values_json(); } else if (data == "getptree") { unique_lock<mutex> lock(m_configmutex); @@ -366,12 +397,12 @@ InputStat::InputStat(const std::string& name) : InputStat::~InputStat() { - get_mgmt_server().unregisterInput(m_name); + get_mgmt_server().unregister_input(m_name); } void InputStat::registerAtServer() { - get_mgmt_server().registerInput(this); + get_mgmt_server().register_input(this); } void InputStat::notifyBuffer(long bufsize) diff --git a/src/ManagementServer.h b/src/ManagementServer.h index 6e39922..c7a4222 100644 --- a/src/ManagementServer.h +++ b/src/ManagementServer.h @@ -2,7 +2,7 @@ Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2018 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org @@ -50,6 +50,7 @@ # include "config.h" #endif +#include "Socket.h" #include "zmq.hpp" #include <string> #include <map> @@ -167,8 +168,10 @@ class ManagementServer void open(int listenport); /* Un-/Register a statistics data source */ - void registerInput(InputStat* is); - void unregisterInput(std::string id); + void register_input(InputStat* is); + void unregister_input(std::string id); + + void update_edi_tcp_output_stat(uint16_t listen_port, size_t num_connections); /* Load a ptree given by the management server. * @@ -205,20 +208,25 @@ class ManagementServer std::thread m_restarter_thread; /******* Statistics Data ********/ - std::map<std::string, InputStat*> m_inputStats; + std::map<std::string, InputStat*> m_input_stats; + + // Holds information about EDI/TCP outputs + std::map<uint16_t /* port */, size_t /* num_connections */> m_output_stats; /* Return a description of the configuration that will * allow to define what graphs to be created * * returns: a JSON encoded configuration */ - std::string getStatConfigJSON(); + std::string get_input_config_json(); /* Return the values for the statistics as defined in the configuration * * returns: JSON encoded statistics */ - std::string getValuesJSON(); + std::string get_input_values_json(); + + std::string get_output_values_json(); // mutex for accessing the map std::mutex m_statsmutex; diff --git a/src/fig/FIG.h b/src/fig/FIG.h index 9752245..eda4671 100644 --- a/src/fig/FIG.h +++ b/src/fig/FIG.h @@ -35,11 +35,19 @@ namespace FIC { class FIGRuntimeInformation { public: - FIGRuntimeInformation(std::shared_ptr<dabEnsemble>& e) : + + using dab_time_t = std::pair<uint32_t /* milliseconds */, time_t>; + using get_time_func_t = std::function<dab_time_t()>; + + FIGRuntimeInformation( + std::shared_ptr<dabEnsemble>& e, + get_time_func_t getTimeFunc) : + getTimeFunc(getTimeFunc), currentFrame(0), ensemble(e), factumAnalyzer(false) {} + get_time_func_t getTimeFunc; unsigned long currentFrame; std::shared_ptr<dabEnsemble> ensemble; bool factumAnalyzer; diff --git a/src/fig/FIG0_10.cpp b/src/fig/FIG0_10.cpp index 56ce9fb..240aa19 100644 --- a/src/fig/FIG0_10.cpp +++ b/src/fig/FIG0_10.cpp @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2016 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li */ /* @@ -23,7 +23,6 @@ along with ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. */ -#include "fig/FIG0structs.h" #include "fig/FIG0_10.h" #include "utils.h" @@ -89,7 +88,7 @@ FillStatus FIG0_10::fill(uint8_t *buf, size_t max_size) return fs; } - //Time and country identifier + // Time and country identifier auto fig0_10 = (FIGtype0_10_LongForm*)buf; fig0_10->FIGtypeNumber = 0; @@ -102,9 +101,9 @@ FillStatus FIG0_10::fill(uint8_t *buf, size_t max_size) remaining -= 2; struct tm timeData; - time_t dab_time_seconds = 0; - uint32_t dab_time_millis = 0; - get_dab_time(&dab_time_seconds, &dab_time_millis); + const auto dab_time = m_rti->getTimeFunc(); + time_t dab_time_seconds = dab_time.second; + uint32_t dab_time_millis = dab_time.first; gmtime_r(&dab_time_seconds, &timeData); fig0_10->RFU = 0; diff --git a/src/fig/FIG0structs.h b/src/fig/FIG0structs.h index 5f514b3..2e107e8 100644 --- a/src/fig/FIG0structs.h +++ b/src/fig/FIG0structs.h @@ -24,19 +24,17 @@ */ #pragma once - #include <cstdint> - #include "fig/FIG.h" -#define FIG0_13_APPTYPE_SLIDESHOW 0x2 -#define FIG0_13_APPTYPE_WEBSITE 0x3 -#define FIG0_13_APPTYPE_TPEG 0x4 -#define FIG0_13_APPTYPE_DGPS 0x5 -#define FIG0_13_APPTYPE_TMC 0x6 -#define FIG0_13_APPTYPE_SPI 0x7 -#define FIG0_13_APPTYPE_DABJAVA 0x8 -#define FIG0_13_APPTYPE_JOURNALINE 0x44a +constexpr uint16_t FIG0_13_APPTYPE_SLIDESHOW = 0x2; +constexpr uint16_t FIG0_13_APPTYPE_WEBSITE = 0x3; +constexpr uint16_t FIG0_13_APPTYPE_TPEG = 0x4; +constexpr uint16_t FIG0_13_APPTYPE_DGPS = 0x5; +constexpr uint16_t FIG0_13_APPTYPE_TMC = 0x6; +constexpr uint16_t FIG0_13_APPTYPE_SPI = 0x7; +constexpr uint16_t FIG0_13_APPTYPE_DABJAVA = 0x8; +constexpr uint16_t FIG0_13_APPTYPE_JOURNALINE = 0x44a; struct FIGtype0 { diff --git a/src/fig/FIGCarousel.cpp b/src/fig/FIGCarousel.cpp index 9748dbf..ceda275 100644 --- a/src/fig/FIGCarousel.cpp +++ b/src/fig/FIGCarousel.cpp @@ -68,8 +68,11 @@ bool FIGCarouselElement::check_deadline() /**************** FIGCarousel *****************/ -FIGCarousel::FIGCarousel(std::shared_ptr<dabEnsemble> ensemble) : - m_rti(ensemble), +FIGCarousel::FIGCarousel( + std::shared_ptr<dabEnsemble> ensemble, + FIGRuntimeInformation::get_time_func_t getTimeFunc + ) : + m_rti(ensemble, getTimeFunc), m_fig0_0(&m_rti), m_fig0_1(&m_rti), m_fig0_2(&m_rti), diff --git a/src/fig/FIGCarousel.h b/src/fig/FIGCarousel.h index 1e33577..a2a8022 100644 --- a/src/fig/FIGCarousel.h +++ b/src/fig/FIGCarousel.h @@ -67,7 +67,9 @@ enum class FIBAllocation { class FIGCarousel { public: - FIGCarousel(std::shared_ptr<dabEnsemble> ensemble); + FIGCarousel( + std::shared_ptr<dabEnsemble> ensemble, + FIGRuntimeInformation::get_time_func_t getTimeFunc); /* Write all FIBs to the buffer, including correct padding and crc. * Returns number of bytes written. diff --git a/src/utils.cpp b/src/utils.cpp index e7ef224..7ea6293 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2021 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org @@ -29,36 +29,13 @@ #include <iostream> #include <memory> #include <boost/algorithm/string/join.hpp> -#include "DabMux.h" #include "utils.h" #include "fig/FIG0structs.h" using namespace std; -static time_t dab_time_seconds = 0; -static int dab_time_millis = 0; - static void printServices(const vector<shared_ptr<DabService> >& services); -void update_dab_time() -{ - if (dab_time_seconds == 0) { - dab_time_seconds = time(nullptr); - } else { - dab_time_millis+= 24; - if (dab_time_millis >= 1000) { - dab_time_millis -= 1000; - ++dab_time_seconds; - } - } -} - -void get_dab_time(time_t *time, uint32_t *millis) -{ - *time = dab_time_seconds; - *millis = dab_time_millis; -} - uint32_t gregorian2mjd(int year, int month, int day) { diff --git a/src/utils.h b/src/utils.h index 331a0b2..d037bb3 100644 --- a/src/utils.h +++ b/src/utils.h @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2020 + Copyright (C) 2025 Matthias P. Braendli, matthias.braendli@mpb.li This file contains a set of utility functions that are used to show @@ -34,10 +34,6 @@ #include <memory> #include "MuxElements.h" -/* Must be called once per ETI frame to update the time */ -void update_dab_time(void); -void get_dab_time(time_t *time, uint32_t *millis); - /* Convert a date and time into the modified Julian date * used in FIG 0/10 * diff --git a/src/zmq2edi/EDISender.cpp b/src/zmq2edi/EDISender.cpp new file mode 100644 index 0000000..06b7420 --- /dev/null +++ b/src/zmq2edi/EDISender.cpp @@ -0,0 +1,390 @@ +/* + Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, + 2011, 2012 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2018 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux 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. + + ODR-DabMux 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 ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "EDISender.h" +#include "Log.h" +#include <cmath> +#include <numeric> +#include <map> +#include <algorithm> + +using namespace std; + +EDISender::~EDISender() +{ + if (running.load()) { + running.store(false); + + // Unblock thread + frame_t emptyframe; + frames.push(emptyframe); + + process_thread.join(); + } +} + +void EDISender::start(const edi_configuration_t& conf, + int delay_ms, int max_delay_ms) +{ + edi_conf = conf; + tist_delay_ms = delay_ms; + tist_max_delay_ms = max_delay_ms; + + if (edi_conf.verbose) { + etiLog.log(info, "Setup EDI"); + } + + if (edi_conf.dump) { + edi_debug_file.open("./edi.debug"); + } + + if (edi_conf.enabled()) { + for (auto& edi_destination : edi_conf.destinations) { + auto edi_output = make_shared<UdpSocket>(edi_destination.source_port); + + if (not edi_destination.source_addr.empty()) { + int err = edi_output->setMulticastSource(edi_destination.source_addr.c_str()); + if (err) { + throw runtime_error("EDI socket set source failed!"); + } + err = edi_output->setMulticastTTL(edi_destination.ttl); + if (err) { + throw runtime_error("EDI socket set TTL failed!"); + } + } + + edi_destination.socket = edi_output; + } + } + + if (edi_conf.verbose) { + etiLog.log(info, "EDI set up"); + } + + // The AF Packet will be protected with reed-solomon and split in fragments + edi::PFT pft(edi_conf); + edi_pft = pft; + + if (edi_conf.interleaver_enabled()) { + edi_interleaver.SetLatency(edi_conf.latency_frames); + } + + startTime = std::chrono::steady_clock::now(); + running.store(true); + process_thread = thread(&EDISender::process, this); +} + +void EDISender::push_frame(const frame_t& frame) +{ + frames.push(frame); +} + +void EDISender::print_configuration() +{ + if (edi_conf.enabled()) { + etiLog.level(info) << "EDI"; + etiLog.level(info) << " verbose " << edi_conf.verbose; + for (auto& edi_dest : edi_conf.destinations) { + etiLog.level(info) << " to " << edi_dest.dest_addr << ":" << edi_conf.dest_port; + if (not edi_dest.source_addr.empty()) { + etiLog.level(info) << " source " << edi_dest.source_addr; + etiLog.level(info) << " ttl " << edi_dest.ttl; + } + etiLog.level(info) << " source port " << edi_dest.source_port; + } + if (edi_conf.interleaver_enabled()) { + etiLog.level(info) << " interleave " << edi_conf.latency_frames * 24 << " ms"; + } + } + else { + etiLog.level(info) << "EDI disabled"; + } +} + +void EDISender::send_eti_frame(uint8_t* p, metadata_t metadata) +{ + edi::TagDETI edi_tagDETI; + edi::TagStarPTR edi_tagStarPtr; + map<int, edi::TagESTn> edi_subchannelToTag; + // The above Tag Items will be assembled into a TAG Packet + edi::TagPacket edi_tagpacket(edi_conf.tagpacket_alignment); + + // SYNC + edi_tagDETI.stat = p[0]; + + // LIDATA FCT + edi_tagDETI.dlfc = metadata.dlfc; + + const int fct = p[4]; + if (metadata.dlfc % 250 != fct) { + etiLog.level(warn) << "Frame FCT=" << fct << + " does not correspond to DLFC=" << metadata.dlfc; + } + + bool ficf = (p[5] & 0x80) >> 7; + edi_tagDETI.ficf = ficf; + + const int nst = p[5] & 0x7F; + + edi_tagDETI.fp = (p[6] & 0xE0) >> 5; + const int mid = (p[6] & 0x18) >> 3; + edi_tagDETI.mid = mid; + //const int fl = (p[6] & 0x07) * 256 + p[7]; + + int ficl = 0; + if (ficf == 0) { + etiLog.level(warn) << "Not FIC in data stream!"; + return; + } + else if (mid == 3) { + ficl = 32; + } + else { + ficl = 24; + } + + vector<uint32_t> sad(nst); + vector<uint32_t> stl(nst); + // Loop over STC subchannels: + for (int i=0; i < nst; i++) { + // EDI stream index is 1-indexed + const int edi_stream_id = i + 1; + + uint32_t scid = (p[8 + 4*i] & 0xFC) >> 2; + sad[i] = (p[8+4*i] & 0x03) * 256 + p[9+4*i]; + uint32_t tpl = (p[10+4*i] & 0xFC) >> 2; + stl[i] = (p[10+4*i] & 0x03) * 256 + \ + p[11+4*i]; + + edi::TagESTn tag_ESTn; + tag_ESTn.id = edi_stream_id; + tag_ESTn.scid = scid; + tag_ESTn.sad = sad[i]; + tag_ESTn.tpl = tpl; + tag_ESTn.rfa = 0; // two bits + tag_ESTn.mst_length = stl[i]; + tag_ESTn.mst_data = nullptr; + + edi_subchannelToTag[i] = tag_ESTn; + } + + const uint16_t mnsc = p[8 + 4*nst] * 256 + \ + p[8 + 4*nst + 1]; + edi_tagDETI.mnsc = mnsc; + + /*const uint16_t crc1 = p[8 + 4*nst + 2]*256 + \ + p[8 + 4*nst + 3]; */ + + edi_tagDETI.fic_data = p + 12 + 4*nst; + edi_tagDETI.fic_length = ficl * 4; + + // loop over MSC subchannels + int offset = 0; + for (int i=0; i < nst; i++) { + edi::TagESTn& tag = edi_subchannelToTag[i]; + tag.mst_data = (p + 12 + 4*nst + ficf*ficl*4 + offset); + + offset += stl[i] * 8; + } + + /* + const uint16_t crc2 = p[12 + 4*nst + ficf*ficl*4 + offset] * 256 + \ + p[12 + 4*nst + ficf*ficl*4 + offset + 1]; */ + + // TIST + const size_t tist_ix = 12 + 4*nst + ficf*ficl*4 + offset + 4; + uint32_t tist = (uint32_t)(p[tist_ix]) << 24 | + (uint32_t)(p[tist_ix+1]) << 16 | + (uint32_t)(p[tist_ix+2]) << 8 | + (uint32_t)(p[tist_ix+3]); + + std::time_t posix_timestamp_1_jan_2000 = 946684800; + + // Wait until our time is tist_delay after the TIST before + // we release that frame + + using namespace std::chrono; + + const auto seconds = metadata.edi_time; + const auto pps_offset = milliseconds(std::lrint((tist & 0xFFFFFF) / 16384.0)); + const auto t_frame = system_clock::from_time_t( + seconds + posix_timestamp_1_jan_2000) + pps_offset; + + const auto t_release = t_frame + milliseconds(tist_delay_ms); + const auto t_now = system_clock::now(); + + /* + etiLog.level(debug) << "seconds " << seconds + posix_timestamp_1_jan_2000; + etiLog.level(debug) << "now " << system_clock::to_time_t(t_now); + etiLog.level(debug) << "wait " << wait_time.count(); + */ + + const auto wait_time = t_release - t_now; + wait_times.push_back(duration_cast<microseconds>(wait_time).count()); + + if (tist_max_delay_ms > 0) { + const auto t_latest_release = t_frame + milliseconds(tist_max_delay_ms); + + if (t_now > t_latest_release) { + // drop frame + num_dropped.fetch_add(1); + return; + } + } + + if (t_release > t_now) { + std::this_thread::sleep_for(wait_time); + } + + edi_tagDETI.tsta = tist; + edi_tagDETI.atstf = 1; + edi_tagDETI.utco = metadata.utc_offset; + edi_tagDETI.seconds = metadata.edi_time; + + if (edi_conf.enabled()) { + // put tags *ptr, DETI and all subchannels into one TagPacket + edi_tagpacket.tag_items.push_back(&edi_tagStarPtr); + edi_tagpacket.tag_items.push_back(&edi_tagDETI); + + for (auto& tag : edi_subchannelToTag) { + edi_tagpacket.tag_items.push_back(&tag.second); + } + + // Assemble into one AF Packet + edi::AFPacket edi_afpacket = edi_afPacketiser.Assemble(edi_tagpacket); + + if (edi_conf.enable_pft) { + // Apply PFT layer to AF Packet (Reed Solomon FEC and Fragmentation) + vector<edi::PFTFragment> edi_fragments = edi_pft.Assemble(edi_afpacket); + + if (edi_conf.verbose) { + fprintf(stderr, "EDI number of PFT fragment before interleaver %zu\n", + edi_fragments.size()); + } + + if (edi_conf.interleaver_enabled()) { + edi_fragments = edi_interleaver.Interleave(edi_fragments); + } + + // Send over ethernet + for (const auto& edi_frag : edi_fragments) { + for (auto& dest : edi_conf.destinations) { + InetAddress addr; + addr.setAddress(dest.dest_addr.c_str()); + addr.setPort(edi_conf.dest_port); + + dest.socket->send(edi_frag, addr); + } + + if (edi_conf.dump) { + std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file); + std::copy(edi_frag.begin(), edi_frag.end(), debug_iterator); + } + } + + if (edi_conf.verbose) { + fprintf(stderr, "EDI number of PFT fragments %zu\n", + edi_fragments.size()); + } + } + else { + // Send over ethernet + for (auto& dest : edi_conf.destinations) { + InetAddress addr; + addr.setAddress(dest.dest_addr.c_str()); + addr.setPort(edi_conf.dest_port); + + dest.socket->send(edi_afpacket, addr); + } + + if (edi_conf.dump) { + std::ostream_iterator<uint8_t> debug_iterator(edi_debug_file); + std::copy(edi_afpacket.begin(), edi_afpacket.end(), debug_iterator); + } + } + } +} + +void EDISender::process() +{ + while (running.load()) { + frame_t frame; + frames.wait_and_pop(frame); + + if (not running.load() or frame.first.empty()) { + break; + } + + if (frame.first.size() == 6144) { + send_eti_frame(frame.first.data(), frame.second); + } + else { + etiLog.level(warn) << "Ignoring short ETI frame, " + "DFLC=" << frame.second.dlfc << ", len=" << + frame.first.size(); + } + + if (wait_times.size() == 250) { // every six seconds + const double n = wait_times.size(); + + double sum = accumulate(wait_times.begin(), wait_times.end(), 0); + size_t num_late = std::count_if(wait_times.begin(), wait_times.end(), + [](double v){ return v < 0; }); + double mean = sum / n; + + double sq_sum = 0; + for (const auto t : wait_times) { + sq_sum += (t-mean) * (t-mean); + } + double stdev = sqrt(sq_sum / n); + auto min_max = minmax_element(wait_times.begin(), wait_times.end()); + + /* Debug code + stringstream ss; + ss << "times:"; + for (const auto t : wait_times) { + ss << " " << t; + } + etiLog.level(debug) << ss.str(); + */ + + const size_t dropped = num_dropped.exchange(0); + + etiLog.level(info) << "Wait time statistics [microseconds]:" + " min: " << *min_max.first << + " max: " << *min_max.second << + " mean: " << mean << + " stdev: " << stdev << + " late: " << + num_late << " of " << wait_times.size() << " (" << + num_late * 100.0 / n << "%)" << + " dropped: " << dropped; + + wait_times.clear(); + } + } +} diff --git a/src/zmq2edi/EDISender.h b/src/zmq2edi/EDISender.h new file mode 100644 index 0000000..44502c1 --- /dev/null +++ b/src/zmq2edi/EDISender.h @@ -0,0 +1,91 @@ +/* + Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, + 2011, 2012 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2018 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux 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. + + ODR-DabMux 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 ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. +*/ + +#pragma once +#include <iostream> +#include <iterator> +#include <thread> +#include <vector> +#include <chrono> +#include <atomic> +#include "ThreadsafeQueue.h" +#include "dabOutput/dabOutput.h" +#include "dabOutput/edi/TagItems.h" +#include "dabOutput/edi/TagPacket.h" +#include "dabOutput/edi/AFPacket.h" +#include "dabOutput/edi/PFT.h" +#include "dabOutput/edi/Interleaver.h" + +// This metadata gets transmitted in the zmq stream +struct metadata_t { + uint32_t edi_time = 0; + int16_t utc_offset = 0; + uint16_t dlfc = 0; +}; + +using frame_t = std::pair<std::vector<uint8_t>, metadata_t>; + +class EDISender { + public: + EDISender() = default; + EDISender(const EDISender& other) = delete; + EDISender& operator=(const EDISender& other) = delete; + ~EDISender(); + void start(const edi_configuration_t& conf, int delay_ms, int max_delay_ms); + void push_frame(const frame_t& frame); + void print_configuration(void); + + private: + void send_eti_frame(uint8_t* p, metadata_t metadata); + void process(void); + + int tist_delay_ms = 0; + int tist_max_delay_ms = 0; + std::atomic<bool> running = ATOMIC_VAR_INIT(false); + std::thread process_thread; + edi_configuration_t edi_conf; + std::chrono::steady_clock::time_point startTime; + ThreadsafeQueue<frame_t> frames; + std::ofstream edi_debug_file; + + // The TagPacket will then be placed into an AFPacket + edi::AFPacketiser edi_afPacketiser; + + // The AF Packet will be protected with reed-solomon and split in fragments + edi::PFT edi_pft; + + // To mitigate for burst packet loss, PFT fragments can be sent out-of-order + edi::Interleaver edi_interleaver; + + // For statistics about wait time before we transmit packets, + // in microseconds + std::vector<double> wait_times; + + // Number of frames dropped because their TIST was larger than max_delay + std::atomic<size_t> num_dropped = ATOMIC_VAR_INIT(0); + +}; diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp new file mode 100644 index 0000000..63c3228 --- /dev/null +++ b/src/zmq2edi/zmq2edi.cpp @@ -0,0 +1,419 @@ +/* + Copyright (C) 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, + 2011, 2012 Her Majesty the Queen in Right of Canada (Communications + Research Center Canada) + + Copyright (C) 2018 + Matthias P. Braendli, matthias.braendli@mpb.li + + http://www.opendigitalradio.org + */ +/* + This file is part of ODR-DabMux. + + ODR-DabMux 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. + + ODR-DabMux 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 ODR-DabMux. If not, see <http://www.gnu.org/licenses/>. +*/ + +#include "Log.h" +#include "zmq.hpp" +#include <math.h> +#include <getopt.h> +#include <string.h> +#include <iostream> +#include <iterator> +#include <vector> + +#include "EDISender.h" +#include "dabOutput/dabOutput.h" + +constexpr size_t MAX_ERROR_COUNT = 10; +constexpr long ZMQ_TIMEOUT_MS = 1000; + +static edi_configuration_t edi_conf; + +static EDISender edisender; + +void usage(void) +{ + using namespace std; + + cerr << "Usage:" << endl; + cerr << "odr-zmq2edi [options] <source>" << endl << endl; + + cerr << "Options:" << endl; + cerr << "The following options can be given only once:" << endl; + cerr << " <source> is a ZMQ URL that points to a ODR-DabMux ZMQ output." << endl; + cerr << " -w <delay> Keep every ETI frame until TIST is <delay> milliseconds after current system time." << endl; + cerr << " -W <max_delay> Drop ETI frames if TIST is <max_delay> later than current system time." << endl; + cerr << " -p <destination port> sets the destination port." << endl; + cerr << " -P Disable PFT and send AFPackets." << endl; + cerr << " -f <fec> sets the FEC." << endl; + cerr << " -i <interleave> enables the interleaved with this latency." << endl; + cerr << " -D dumps the EDI to edi.debug file." << endl; + cerr << " -v Enables verbose mode." << endl; + cerr << " -a <tagpacket alignement> sets the alignment of the TAG Packet (default 8)." << endl << endl; + + cerr << "The following options can be given several times, when more than once destination is addressed:" << endl; + cerr << " -d <destination ip> sets the destination ip." << endl; + cerr << " -s <source port> sets the source port." << endl; + cerr << " -S <source ip> select the source IP in case we want to use multicast." << endl; + cerr << " -t <ttl> set the packet's TTL." << endl << endl; + + cerr << "odr-zmq2edi will quit if it does not receive data for " << + (int)(MAX_ERROR_COUNT * ZMQ_TIMEOUT_MS / 1000.0) << " seconds." << endl; + cerr << "It is best practice to run this tool under a process supervisor that will restart it automatically." << endl; +} + +static metadata_t get_md_one_frame(uint8_t *buf, size_t size, size_t *consumed_bytes) +{ + size_t remaining = size; + if (remaining < 3) { + etiLog.level(warn) << "Insufficient data to parse metadata"; + throw std::runtime_error("Insufficient data"); + } + + metadata_t md; + bool utc_offset_received = false; + bool edi_time_received = false; + bool dlfc_received = false; + + while (remaining) { + uint8_t id = buf[0]; + uint16_t len = (((uint16_t)buf[1]) << 8) + buf[2]; + + if (id == static_cast<uint8_t>(output_metadata_id_e::separation_marker)) { + if (len != 0) { + etiLog.level(warn) << "Invalid length " << len << " for metadata: separation_marker"; + } + + if (not utc_offset_received or not edi_time_received or not dlfc_received) { + throw std::runtime_error("Incomplete metadata received"); + } + + remaining -= 3; + *consumed_bytes = size - remaining; + return md; + } + else if (id == static_cast<uint8_t>(output_metadata_id_e::utc_offset)) { + if (len != 2) { + etiLog.level(warn) << "Invalid length " << len << " for metadata: utc_offset"; + } + if (remaining < 2) { + throw std::runtime_error("Insufficient data for utc_offset"); + } + uint16_t utco; + std::memcpy(&utco, buf + 3, sizeof(utco)); + md.utc_offset = ntohs(utco); + utc_offset_received = true; + remaining -= 5; + buf += 5; + } + else if (id == static_cast<uint8_t>(output_metadata_id_e::edi_time)) { + if (len != 4) { + etiLog.level(warn) << "Invalid length " << len << " for metadata: edi_time"; + } + if (remaining < 4) { + throw std::runtime_error("Insufficient data for edi_time"); + } + uint32_t edi_time; + std::memcpy(&edi_time, buf + 3, sizeof(edi_time)); + md.edi_time = ntohl(edi_time); + edi_time_received = true; + remaining -= 7; + buf += 7; + } + else if (id == static_cast<uint8_t>(output_metadata_id_e::dlfc)) { + if (len != 2) { + etiLog.level(warn) << "Invalid length " << len << " for metadata: dlfc"; + } + if (remaining < 2) { + throw std::runtime_error("Insufficient data for dlfc"); + } + uint16_t dlfc; + std::memcpy(&dlfc, buf + 3, sizeof(dlfc)); + md.dlfc = ntohs(dlfc); + dlfc_received = true; + remaining -= 5; + buf += 5; + } + } + + throw std::runtime_error("Insufficient data"); +} + +/* There is some state inside the parsing of destination arguments, + * because several destinations can be given. */ + +static edi_destination_t edi_destination; +static bool source_port_set = false; +static bool source_addr_set = false; +static bool ttl_set = false; +static bool dest_addr_set = false; + +static void add_edi_destination(void) +{ + if (not dest_addr_set) { + throw std::runtime_error("Destination address not specified for destination number " + + std::to_string(edi_conf.destinations.size() + 1)); + } + + edi_conf.destinations.push_back(edi_destination); + edi_destination_t newdest; + edi_destination = newdest; + + source_port_set = false; + source_addr_set = false; + ttl_set = false; + dest_addr_set = false; +} + +static void parse_destination_args(char option) +{ + switch (option) { + case 's': + if (source_port_set) { + add_edi_destination(); + } + edi_destination.source_port = std::stoi(optarg); + source_port_set = true; + break; + case 'S': + if (source_addr_set) { + add_edi_destination(); + } + edi_destination.source_addr = optarg; + source_addr_set = true; + break; + case 't': + if (ttl_set) { + add_edi_destination(); + } + edi_destination.ttl = std::stoi(optarg); + ttl_set = true; + break; + case 'd': + if (dest_addr_set) { + add_edi_destination(); + } + edi_destination.dest_addr = optarg; + dest_addr_set = true; + break; + default: + throw std::logic_error("parse_destination_args invalid"); + } +} + +int start(int argc, char **argv) +{ + edi_conf.enable_pft = true; + + if (argc == 0) { + usage(); + return 1; + } + + int delay_ms = 500; + int max_delay_ms = 0; // no max delay + + int ch = 0; + while (ch != -1) { + ch = getopt(argc, argv, "d:p:s:S:t:Pf:i:Dva:w:W:"); + switch (ch) { + case -1: + break; + case 'd': + case 's': + case 'S': + case 't': + parse_destination_args(ch); + break; + case 'p': + edi_conf.dest_port = std::stoi(optarg); + break; + case 'P': + edi_conf.enable_pft = false; + break; + case 'f': + edi_conf.fec = std::stoi(optarg); + break; + case 'i': + { + double interleave_ms = std::stod(optarg); + if (interleave_ms != 0.0) { + if (interleave_ms < 0) { + throw std::runtime_error("EDI output: negative interleave value is invalid."); + } + + auto latency_rounded = lround(interleave_ms / 24.0); + if (latency_rounded * 24 > 30000) { + throw std::runtime_error("EDI output: interleaving set for more than 30 seconds!"); + } + + edi_conf.latency_frames = latency_rounded; + } + } + break; + case 'D': + edi_conf.dump = true; + break; + case 'v': + edi_conf.verbose = true; + break; + case 'a': + edi_conf.tagpacket_alignment = std::stoi(optarg); + break; + case 'w': + delay_ms = std::stoi(optarg); + break; + case 'W': + max_delay_ms = std::stoi(optarg); + break; + case 'h': + default: + usage(); + return 1; + } + } + + add_edi_destination(); + + if (optind >= argc) { + etiLog.level(error) << "source option is missing"; + return 1; + } + + if (edi_conf.dest_port == 0) { + etiLog.level(error) << "No EDI destination port defined"; + return 1; + } + + if (edi_conf.destinations.empty()) { + etiLog.level(error) << "No EDI destinations set"; + return 1; + } + + if (max_delay_ms > 0) { + etiLog.level(info) << "Setting up EDI Sender with delay " << delay_ms << " ms and max delay " << max_delay_ms << " ms"; + } + else { + etiLog.level(info) << "Setting up EDI Sender with delay " << delay_ms << " ms"; + } + edisender.start(edi_conf, delay_ms, max_delay_ms); + edisender.print_configuration(); + + const char* source_url = argv[optind]; + + + size_t frame_count = 0; + size_t error_count = 0; + + etiLog.level(info) << "Opening ZMQ input: " << source_url; + + zmq::context_t zmq_ctx(1); + zmq::socket_t zmq_sock(zmq_ctx, ZMQ_SUB); + zmq_sock.connect(source_url); + zmq_sock.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); // subscribe to all messages + + while (error_count < MAX_ERROR_COUNT) { + zmq::message_t incoming; + zmq::pollitem_t items[1]; + items[0].socket = zmq_sock; + items[0].events = ZMQ_POLLIN; + const int num_events = zmq::poll(items, 1, ZMQ_TIMEOUT_MS); + if (num_events == 0) { // timeout + error_count++; + } + else { + // Event received: recv will not block + zmq_sock.recv(&incoming); + + zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data(); + + if (dab_msg->version != 1) { + etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version; + error_count++; + } + + int offset = sizeof(dab_msg->version) + + NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen); + + std::list<std::pair<std::vector<uint8_t>, metadata_t> > all_frames; + + for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) { + if (dab_msg->buflen[i] <= 0 or dab_msg->buflen[i] > 6144) { + etiLog.level(error) << "ZeroMQ buffer " << i << + " has invalid length " << dab_msg->buflen[i]; + error_count++; + } + else { + std::vector<uint8_t> buf(6144, 0x55); + + const int framesize = dab_msg->buflen[i]; + + memcpy(&buf.front(), + ((uint8_t*)incoming.data()) + offset, + framesize); + + all_frames.emplace_back( + std::piecewise_construct, + std::make_tuple(std::move(buf)), + std::make_tuple()); + + offset += framesize; + } + } + + for (auto &f : all_frames) { + size_t consumed_bytes = 0; + + f.second = get_md_one_frame( + static_cast<uint8_t*>(incoming.data()) + offset, + incoming.size() - offset, + &consumed_bytes); + + offset += consumed_bytes; + } + + for (auto &f : all_frames) { + edisender.push_frame(f); + frame_count++; + } + } + } + + etiLog.level(info) << "Quitting after " << frame_count << " frames transferred"; + + return 0; +} + +int main(int argc, char **argv) +{ + etiLog.level(info) << "ZMQ2EDI converter from " << + PACKAGE_NAME << " " << +#if defined(GITVERSION) + GITVERSION << +#else + PACKAGE_VERSION << +#endif + " starting up"; + + try { + return start(argc, argv); + } + catch (std::runtime_error &e) { + etiLog.level(error) << "Error: " << e.what(); + } + + return 1; +} |