From 78b4f04de2975da7b6240983fe1c6a496289a067 Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Mon, 29 Jan 2024 14:31:55 +0100 Subject: Add ZMQ output to odr-zmq2edi --- src/zmq2edi/EDISender.cpp | 297 ------------------------------------------ src/zmq2edi/EDISender.h | 85 ------------ src/zmq2edi/Sender.cpp | 320 ++++++++++++++++++++++++++++++++++++++++++++++ src/zmq2edi/Sender.h | 98 ++++++++++++++ src/zmq2edi/zmq2edi.cpp | 106 ++++++++++----- 5 files changed, 489 insertions(+), 417 deletions(-) delete mode 100644 src/zmq2edi/EDISender.cpp delete mode 100644 src/zmq2edi/EDISender.h create mode 100644 src/zmq2edi/Sender.cpp create mode 100644 src/zmq2edi/Sender.h (limited to 'src/zmq2edi') diff --git a/src/zmq2edi/EDISender.cpp b/src/zmq2edi/EDISender.cpp deleted file mode 100644 index 4a70105..0000000 --- a/src/zmq2edi/EDISender.cpp +++ /dev/null @@ -1,297 +0,0 @@ -/* - 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) 2020 - 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 . -*/ - -#include "EDISender.h" -#include "Log.h" -#include -#include -#include -#include -#include - -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, bool drop_late_packets) -{ - edi_conf = conf; - tist_delay_ms = delay_ms; - drop_late = drop_late_packets; - - edi_sender = make_shared(edi_conf); - - running.store(true); - process_thread = thread(&EDISender::process, this); -} - -void EDISender::push_frame(frame_t&& frame) -{ - frames.push(move(frame)); -} - -void EDISender::print_configuration() -{ - if (edi_conf.enabled()) { - edi_conf.print(); - } - else { - etiLog.level(info) << "EDI disabled"; - } -} - -void EDISender::send_eti_frame(frame_t& frame) -{ - uint8_t *p = frame.data.data(); - - edi::TagDETI edi_tagDETI; - edi::TagStarPTR edi_tagStarPtr("DETI"); - map 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 = frame.metadata.dlfc; - - const int fct = p[4]; - if (frame.metadata.dlfc % 250 != fct) { - etiLog.level(warn) << "Frame FCT=" << fct << - " does not correspond to DLFC=" << frame.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 sad(nst); - vector 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; - } - - uint16_t mnsc = 0; - std::memcpy(&mnsc, p + 8 + 4*nst, sizeof(uint16_t)); - 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 pps_offset = milliseconds(std::lrint((tist & 0xFFFFFF) / 16384.0)); - const auto t_frame = system_clock::from_time_t( - frame.metadata.edi_time + posix_timestamp_1_jan_2000 - frame.metadata.utc_offset) + pps_offset; - - const auto t_release = t_frame + milliseconds(tist_delay_ms); - const auto t_now = system_clock::now(); - - const bool late = t_release < t_now; - - buffering_stat_t stat; - stat.late = late; - - if (not late) { - const auto wait_time = t_release - t_now; - std::this_thread::sleep_for(wait_time); - } - - stat.buffering_time_us = duration_cast(steady_clock::now() - frame.received_at).count(); - buffering_stats.push_back(std::move(stat)); - - if (late and drop_late) { - return; - } - - edi_tagDETI.tsta = tist; - edi_tagDETI.atstf = 1; - edi_tagDETI.utco = frame.metadata.utc_offset; - edi_tagDETI.seconds = frame.metadata.edi_time; - - if (edi_sender and 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); - } - - edi_sender->write(edi_tagpacket); - } -} - -void EDISender::process() -{ - while (running.load()) { - frame_t frame; - frames.wait_and_pop(frame); - - if (not running.load() or frame.data.empty()) { - break; - } - - if (frame.data.size() == 6144) { - send_eti_frame(frame); - } - else { - etiLog.level(warn) << "Ignoring short ETI frame, " - "DFLC=" << frame.metadata.dlfc << ", len=" << - frame.data.size(); - } - - if (buffering_stats.size() == 250) { // every six seconds - const double n = buffering_stats.size(); - - size_t num_late = std::count_if(buffering_stats.begin(), buffering_stats.end(), - [](const buffering_stat_t& s){ return s.late; }); - - double sum = 0.0; - double min = std::numeric_limits::max(); - double max = -std::numeric_limits::max(); - for (const auto& s : buffering_stats) { - // convert to milliseconds - const double t = s.buffering_time_us / 1000.0; - sum += t; - - if (t < min) { - min = t; - } - - if (t > max) { - max = t; - } - } - double mean = sum / n; - - double sq_sum = 0; - for (const auto& s : buffering_stats) { - const double t = s.buffering_time_us / 1000.0; - sq_sum += (t-mean) * (t-mean); - } - double stdev = sqrt(sq_sum / n); - - /* Debug code - stringstream ss; - ss << "times:"; - for (const auto t : buffering_stats) { - ss << " " << lrint(t.buffering_time_us / 1000.0); - } - etiLog.level(debug) << ss.str(); - // */ - - etiLog.level(info) << "Buffering time statistics [milliseconds]:" - " min: " << min << - " max: " << max << - " mean: " << mean << - " stdev: " << stdev << - " late: " << - num_late << " of " << buffering_stats.size() << " (" << - num_late * 100.0 / n << "%)"; - - buffering_stats.clear(); - } - } -} diff --git a/src/zmq2edi/EDISender.h b/src/zmq2edi/EDISender.h deleted file mode 100644 index 9c37e75..0000000 --- a/src/zmq2edi/EDISender.h +++ /dev/null @@ -1,85 +0,0 @@ -/* - 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) 2020 - 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 . -*/ - -#pragma once -#include -#include -#include -#include -#include -#include -#include "ThreadsafeQueue.h" -#include "dabOutput/dabOutput.h" -#include "edioutput/TagItems.h" -#include "edioutput/TagPacket.h" -#include "edioutput/Transport.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; -}; - -struct frame_t { - std::vector data; - metadata_t metadata; - std::chrono::steady_clock::time_point received_at; -}; - -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, bool drop_late_packets); - void push_frame(frame_t&& frame); - void print_configuration(void); - - private: - void send_eti_frame(frame_t& frame); - void process(void); - - int tist_delay_ms; - bool drop_late; - std::atomic running; - std::thread process_thread; - edi::configuration_t edi_conf; - ThreadsafeQueue frames; - - std::shared_ptr edi_sender; - - struct buffering_stat_t { - // Time between when we received the packets and when we transmit packets, in microseconds - double buffering_time_us = 0.0; - bool late = false; - }; - std::vector buffering_stats; - -}; diff --git a/src/zmq2edi/Sender.cpp b/src/zmq2edi/Sender.cpp new file mode 100644 index 0000000..fe46846 --- /dev/null +++ b/src/zmq2edi/Sender.cpp @@ -0,0 +1,320 @@ +/* + 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) 2024 + 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 . +*/ + +#include "Sender.h" +#include "Log.h" +#include +#include +#include +#include +#include + +using namespace std; + +Sender::Sender() : + zmq_ctx(2) +{ +} + +Sender::~Sender() +{ + if (running.load()) { + running.store(false); + + // Unblock thread + frame_t emptyframe; + frames.push(std::move(emptyframe)); + + process_thread.join(); + } +} + +void Sender::start(const edi::configuration_t& conf, + const zmq_send_config_t& zmq_conf, + int delay_ms, bool drop_late_packets) +{ + edi_conf = conf; + tist_delay_ms = delay_ms; + drop_late = drop_late_packets; + + edi_sender = make_shared(edi_conf); + + for (const auto& url : zmq_conf.urls) { + zmq::socket_t zmq_sock(zmq_ctx, ZMQ_PUB); + zmq_sock.bind(url.c_str()); + zmq_sockets.emplace_back(std::move(zmq_sock)); + } + + running.store(true); + process_thread = thread(&Sender::process, this); +} + +void Sender::push_frame(frame_t&& frame) +{ + frames.push(std::move(frame)); +} + +void Sender::print_configuration() +{ + if (edi_conf.enabled()) { + edi_conf.print(); + } + else { + etiLog.level(info) << "EDI disabled"; + } +} + +void Sender::send_eti_frame(frame_t& frame) +{ + uint8_t *p = frame.data.data(); + + edi::TagDETI edi_tagDETI; + edi::TagStarPTR edi_tagStarPtr("DETI"); + map 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 = frame.metadata.dlfc; + + const int fct = p[4]; + if (frame.metadata.dlfc % 250 != fct) { + etiLog.level(warn) << "Frame FCT=" << fct << + " does not correspond to DLFC=" << frame.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 sad(nst); + vector 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; + } + + uint16_t mnsc = 0; + std::memcpy(&mnsc, p + 8 + 4*nst, sizeof(uint16_t)); + 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 pps_offset = milliseconds(std::lrint((tist & 0xFFFFFF) / 16384.0)); + const auto t_frame = system_clock::from_time_t( + frame.metadata.edi_time + posix_timestamp_1_jan_2000 - frame.metadata.utc_offset) + pps_offset; + + const auto t_release = t_frame + milliseconds(tist_delay_ms); + const auto t_now = system_clock::now(); + + const bool late = t_release < t_now; + + buffering_stat_t stat; + stat.late = late; + + if (not late) { + const auto wait_time = t_release - t_now; + std::this_thread::sleep_for(wait_time); + } + + stat.buffering_time_us = duration_cast(steady_clock::now() - frame.received_at).count(); + buffering_stats.push_back(std::move(stat)); + + if (late and drop_late) { + return; + } + + edi_tagDETI.tsta = tist; + edi_tagDETI.atstf = 1; + edi_tagDETI.utco = frame.metadata.utc_offset; + edi_tagDETI.seconds = frame.metadata.edi_time; + + if (edi_sender and 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); + } + + edi_sender->write(edi_tagpacket); + } + + if (not frame.original_zmq_message.empty()) { + for (auto& sock : zmq_sockets) { + const auto send_result = sock.send(frame.original_zmq_message, zmq::send_flags::dontwait); + if (not send_result.has_value()) { + num_zmq_send_errors++; + } + } + } +} + +void Sender::process() +{ + while (running.load()) { + frame_t frame; + frames.wait_and_pop(frame); + + if (not running.load() or frame.data.empty()) { + break; + } + + if (frame.data.size() == 6144) { + send_eti_frame(frame); + } + else { + etiLog.level(warn) << "Ignoring short ETI frame, " + "DFLC=" << frame.metadata.dlfc << ", len=" << + frame.data.size(); + } + + if (buffering_stats.size() == 250) { // every six seconds + const double n = buffering_stats.size(); + + size_t num_late = std::count_if(buffering_stats.begin(), buffering_stats.end(), + [](const buffering_stat_t& s){ return s.late; }); + + double sum = 0.0; + double min = std::numeric_limits::max(); + double max = -std::numeric_limits::max(); + for (const auto& s : buffering_stats) { + // convert to milliseconds + const double t = s.buffering_time_us / 1000.0; + sum += t; + + if (t < min) { + min = t; + } + + if (t > max) { + max = t; + } + } + double mean = sum / n; + + double sq_sum = 0; + for (const auto& s : buffering_stats) { + const double t = s.buffering_time_us / 1000.0; + sq_sum += (t-mean) * (t-mean); + } + double stdev = sqrt(sq_sum / n); + + /* Debug code + stringstream ss; + ss << "times:"; + for (const auto t : buffering_stats) { + ss << " " << lrint(t.buffering_time_us / 1000.0); + } + etiLog.level(debug) << ss.str(); + // */ + + etiLog.level(info) << "Buffering time statistics [milliseconds]:" + " min: " << min << + " max: " << max << + " mean: " << mean << + " stdev: " << stdev << + " late: " << + num_late << " of " << buffering_stats.size() << " (" << + num_late * 100.0 / n << "%) " << + "Num ZMQ send errors: " << num_zmq_send_errors; + + buffering_stats.clear(); + } + } +} diff --git a/src/zmq2edi/Sender.h b/src/zmq2edi/Sender.h new file mode 100644 index 0000000..6dfd615 --- /dev/null +++ b/src/zmq2edi/Sender.h @@ -0,0 +1,98 @@ +/* + 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) 2024 + 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 . +*/ + +#pragma once +#include +#include +#include +#include +#include +#include +#include "ThreadsafeQueue.h" +#include "dabOutput/dabOutput.h" +#include "edioutput/TagItems.h" +#include "edioutput/TagPacket.h" +#include "edioutput/Transport.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; +}; + +struct frame_t { + // Since a zmq message actually contains 4 frames, the + // original_zmq_msg is only non-empty for the first of the + // four calls to Sender::send_edi_frame(). + zmq::message_t original_zmq_message; + std::vector data; + metadata_t metadata; + std::chrono::steady_clock::time_point received_at; +}; + +struct zmq_send_config_t { + std::vector urls; +}; + +class Sender { + public: + Sender(); + Sender(const Sender& other) = delete; + Sender& operator=(const Sender& other) = delete; + ~Sender(); + void start(const edi::configuration_t& conf, + const zmq_send_config_t& zmq_conf, + int delay_ms, bool drop_late_packets); + void push_frame(frame_t&& frame); + void print_configuration(void); + + private: + void send_eti_frame(frame_t& frame); + void process(void); + + int tist_delay_ms; + bool drop_late; + std::atomic running; + std::thread process_thread; + edi::configuration_t edi_conf; + ThreadsafeQueue frames; + + std::shared_ptr edi_sender; + + zmq::context_t zmq_ctx; + std::vector zmq_sockets; + + struct buffering_stat_t { + // Time between when we received the packets and when we transmit packets, in microseconds + double buffering_time_us = 0.0; + bool late = false; + }; + std::vector buffering_stats; + size_t num_zmq_send_errors = 0; + +}; diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp index cff16c7..41d92b5 100644 --- a/src/zmq2edi/zmq2edi.cpp +++ b/src/zmq2edi/zmq2edi.cpp @@ -3,7 +3,7 @@ 2011, 2012 Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2020 + Copyright (C) 2024 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org @@ -36,7 +36,7 @@ #include #include -#include "EDISender.h" +#include "Sender.h" #include "dabOutput/dabOutput.h" constexpr size_t MAX_ERROR_COUNT = 10; @@ -45,7 +45,7 @@ constexpr long DEFAULT_BACKOFF = 5000; static edi::configuration_t edi_conf; -static EDISender edisender; +static Sender edisender; static void usage() { @@ -54,6 +54,8 @@ static void usage() cerr << "Usage:" << endl; cerr << "odr-zmq2edi [options] " << endl << endl; + cerr << "ODR-ZMQ2EDI can output to both EDI and ZMQ. It buffers and releases frames according to their timestamp." << endl; + cerr << "Options:" << endl; cerr << "The following options can be given only once:" << endl; cerr << " is a ZMQ URL that points to a ODR-DabMux ZMQ output." << endl; @@ -62,15 +64,20 @@ static void usage() cerr << " -C Before starting, run the given script, and only start if it returns 0." << endl; cerr << " This is useful for checking that NTP is properly synchronised" << endl; cerr << " -x Drop frames where for which the wait time would be negative, i.e. frames that arrived too late." << endl; + cerr << " -b Number of milliseconds to backoff after an input reset (default " << DEFAULT_BACKOFF << ")." << endl << endl; + + cerr << " ZMQ Output options:" << endl; + cerr << " -Z Add a zmq output to URL, e.g. tcp:*:9876 to listen for connections on port 9876 " << endl << endl; + + cerr << " EDI Output options:" << endl; + cerr << " -v Enables verbose mode." << endl; cerr << " -P Disable PFT and send AFPackets." << endl; cerr << " -f Set the FEC." << endl; cerr << " -i Configure the UDP packet spread/interleaver with given percentage: 0% send all fragments at once, 100% spread over 24ms, >100% spread and interleave. Default 95%\n"; cerr << " -D Dump the EDI to edi.debug file." << endl; - cerr << " -v Enables verbose mode." << endl; cerr << " -a Set the alignment of the TAG Packet (default 8)." << endl; - cerr << " -b Number of milliseconds to backoff after an input reset (default " << DEFAULT_BACKOFF << ")." << endl << endl; - cerr << "The following options can be given several times, when more than UDP destination is desired:" << endl; + cerr << "The following options can be given several times, when more than EDI/UDP destination is desired:" << endl; cerr << " -d Set the destination ip." << endl; cerr << " -p Set the destination port." << endl; cerr << " -s Set the source port." << endl; @@ -176,7 +183,7 @@ static void add_edi_destination(void) std::to_string(edi_conf.destinations.size() + 1)); } - edi_conf.destinations.push_back(move(edi_destination)); + edi_conf.destinations.push_back(std::move(edi_destination)); edi_destination = std::make_shared(); dest_port_set = false; @@ -249,9 +256,11 @@ int start(int argc, char **argv) uint32_t backoff_after_reset_ms = DEFAULT_BACKOFF; std::string startupcheck; + zmq_send_config_t zmq_conf; + int ch = 0; while (ch != -1) { - ch = getopt(argc, argv, "C:d:p:s:S:t:Pf:i:Dva:b:w:xh"); + ch = getopt(argc, argv, "C:d:p:s:S:t:Pf:i:Dva:b:w:xhZ:"); switch (ch) { case -1: break; @@ -302,6 +311,9 @@ int start(int argc, char **argv) case 'x': drop_late_packets = true; break; + case 'Z': + zmq_conf.urls.push_back(optarg); + break; case 'h': default: usage(); @@ -309,6 +321,32 @@ int start(int argc, char **argv) } } + if (dest_addr_set) { + add_edi_destination(); + } + + if (optind >= argc) { + etiLog.level(error) << "source option is missing"; + return 1; + } + + if (edi_conf.destinations.empty() and zmq_conf.urls.empty()) { + etiLog.level(error) << "No destinations set"; + return 1; + } + + if (not edi_conf.destinations.empty()) { + edisender.print_configuration(); + } + + if (not zmq_conf.urls.empty()) { + etiLog.level(info) << "Setting up ZMQ to:"; + for (const auto& url : zmq_conf.urls) { + etiLog.level(info) << " " << url; + } + } + + if (not startupcheck.empty()) { etiLog.level(info) << "Running startup check '" << startupcheck << "'"; int wstatus = system(startupcheck.c_str()); @@ -328,22 +366,9 @@ int start(int argc, char **argv) } } - add_edi_destination(); - - if (optind >= argc) { - etiLog.level(error) << "source option is missing"; - return 1; - } - - if (edi_conf.destinations.empty()) { - etiLog.level(error) << "No EDI destinations set"; - return 1; - } - - etiLog.level(info) << "Setting up EDI Sender with delay " << delay_ms << " ms. " << + etiLog.level(info) << "Setting up Sender with delay " << delay_ms << " ms. " << (drop_late_packets ? "Will" : "Will not") << " drop late packets"; - edisender.start(edi_conf, delay_ms, drop_late_packets); - edisender.print_configuration(); + edisender.start(edi_conf, zmq_conf, delay_ms, drop_late_packets); const char* source_url = argv[optind]; @@ -370,26 +395,33 @@ int start(int argc, char **argv) } else { // Event received: recv will not block - zmq_sock.recv(incoming, zmq::recv_flags::none); + const auto recv_result = zmq_sock.recv(incoming, zmq::recv_flags::none); + if (not recv_result.has_value()) { + continue; + } + const auto received_at = std::chrono::steady_clock::now(); - zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data(); + // Casting incoming.data() to zmq_dab_message_t* is not allowed, because + // it might be misaligned + zmq_dab_message_t dab_msg; + memcpy(&dab_msg, incoming.data(), ZMQ_DAB_MESSAGE_HEAD_LENGTH); - if (dab_msg->version != 1) { - etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version; + 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); + int offset = sizeof(dab_msg.version) + + NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg.buflen); std::vector all_frames; all_frames.reserve(NUM_FRAMES_PER_ZMQ_MESSAGE); for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) { - if (dab_msg->buflen[i] <= 0 or dab_msg->buflen[i] > 6144) { + 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]; + " has invalid length " << dab_msg.buflen[i]; error_count++; } else { @@ -397,7 +429,7 @@ int start(int argc, char **argv) frame.data.resize(6144, 0x55); frame.received_at = received_at; - const int framesize = dab_msg->buflen[i]; + const int framesize = dab_msg.buflen[i]; memcpy(frame.data.data(), ((uint8_t*)incoming.data()) + offset, @@ -429,6 +461,10 @@ int start(int argc, char **argv) offset += consumed_bytes; } + if (not all_frames.empty()) { + all_frames[0].original_zmq_message = std::move(incoming); + } + for (auto &f : all_frames) { edisender.push_frame(std::move(f)); } @@ -476,9 +512,6 @@ int main(int argc, char **argv) try { ret = start(argc, argv); - - // To make sure things get printed to stderr - std::this_thread::sleep_for(std::chrono::milliseconds(300)); } catch (const std::runtime_error &e) { etiLog.level(error) << "Runtime error: " << e.what(); @@ -487,5 +520,8 @@ int main(int argc, char **argv) etiLog.level(error) << "Logic error! " << e.what(); } + // To make sure things get printed to stderr + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + return ret; } -- cgit v1.2.3