summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/input/Zmq.cpp4
-rw-r--r--src/input/Zmq.h4
-rw-r--r--src/zmq2edi/EDISender.cpp98
-rw-r--r--src/zmq2edi/EDISender.h22
-rw-r--r--src/zmq2edi/zmq2edi.cpp167
5 files changed, 186 insertions, 109 deletions
diff --git a/src/input/Zmq.cpp b/src/input/Zmq.cpp
index 603f514..305653b 100644
--- a/src/input/Zmq.cpp
+++ b/src/input/Zmq.cpp
@@ -369,7 +369,7 @@ int ZmqMPEG::readFromSocket(size_t framesize)
if ( msg.size() >= sizeof(zmq_frame_header_t) and
msg.size() == ZMQ_FRAME_SIZE(frame) and
frame->version == 1 and
- frame->encoder == ZMQ_ENCODER_TOOLAME) {
+ frame->encoder == ZMQ_ENCODER_MPEG_L2) {
datalen = frame->datasize;
data = ZMQ_FRAME_DATA(frame);
@@ -439,7 +439,7 @@ int ZmqAAC::readFromSocket(size_t framesize)
if ( msg.size() >= sizeof(zmq_frame_header_t) and
msg.size() == ZMQ_FRAME_SIZE(frame) and
frame->version == 1 and
- frame->encoder == ZMQ_ENCODER_FDK) {
+ frame->encoder == ZMQ_ENCODER_AACPLUS) {
datalen = frame->datasize;
data = ZMQ_FRAME_DATA(frame);
diff --git a/src/input/Zmq.h b/src/input/Zmq.h
index f4992f1..c101da0 100644
--- a/src/input/Zmq.h
+++ b/src/input/Zmq.h
@@ -119,8 +119,8 @@ struct dab_input_zmq_config_t
std::string curve_encoder_keyfile;
};
-#define ZMQ_ENCODER_FDK 1
-#define ZMQ_ENCODER_TOOLAME 2
+#define ZMQ_ENCODER_AACPLUS 1
+#define ZMQ_ENCODER_MPEG_L2 2
/* This defines the on-wire representation of a ZMQ message header.
*
diff --git a/src/zmq2edi/EDISender.cpp b/src/zmq2edi/EDISender.cpp
index 2188f8a..4a70105 100644
--- a/src/zmq2edi/EDISender.cpp
+++ b/src/zmq2edi/EDISender.cpp
@@ -3,7 +3,7 @@
2011, 2012 Her Majesty the Queen in Right of Canada (Communications
Research Center Canada)
- Copyright (C) 2019
+ Copyright (C) 2020
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -31,6 +31,7 @@
#include <numeric>
#include <map>
#include <algorithm>
+#include <limits>
using namespace std;
@@ -47,8 +48,7 @@ EDISender::~EDISender()
}
}
-void EDISender::start(const edi::configuration_t& conf,
- int delay_ms, bool drop_late_packets)
+void EDISender::start(const edi::configuration_t& conf, int delay_ms, bool drop_late_packets)
{
edi_conf = conf;
tist_delay_ms = delay_ms;
@@ -56,14 +56,13 @@ void EDISender::start(const edi::configuration_t& conf,
edi_sender = make_shared<edi::Sender>(edi_conf);
- startTime = std::chrono::steady_clock::now();
running.store(true);
process_thread = thread(&EDISender::process, this);
}
-void EDISender::push_frame(const frame_t& frame)
+void EDISender::push_frame(frame_t&& frame)
{
- frames.push(frame);
+ frames.push(move(frame));
}
void EDISender::print_configuration()
@@ -76,8 +75,10 @@ void EDISender::print_configuration()
}
}
-void EDISender::send_eti_frame(uint8_t* p, metadata_t metadata)
+void EDISender::send_eti_frame(frame_t& frame)
{
+ uint8_t *p = frame.data.data();
+
edi::TagDETI edi_tagDETI;
edi::TagStarPTR edi_tagStarPtr("DETI");
map<int, edi::TagESTn> edi_subchannelToTag;
@@ -88,12 +89,12 @@ void EDISender::send_eti_frame(uint8_t* p, metadata_t metadata)
edi_tagDETI.stat = p[0];
// LIDATA FCT
- edi_tagDETI.dlfc = metadata.dlfc;
+ edi_tagDETI.dlfc = frame.metadata.dlfc;
const int fct = p[4];
- if (metadata.dlfc % 250 != fct) {
+ if (frame.metadata.dlfc % 250 != fct) {
etiLog.level(warn) << "Frame FCT=" << fct <<
- " does not correspond to DLFC=" << metadata.dlfc;
+ " does not correspond to DLFC=" << frame.metadata.dlfc;
}
bool ficf = (p[5] & 0x80) >> 7;
@@ -182,25 +183,32 @@ void EDISender::send_eti_frame(uint8_t* p, metadata_t metadata)
const auto pps_offset = milliseconds(std::lrint((tist & 0xFFFFFF) / 16384.0));
const auto t_frame = system_clock::from_time_t(
- metadata.edi_time + posix_timestamp_1_jan_2000 - metadata.utc_offset) + pps_offset;
+ 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 auto wait_time = t_release - t_now;
- wait_times.push_back(duration_cast<microseconds>(wait_time).count());
+ const bool late = t_release < t_now;
+
+ buffering_stat_t stat;
+ stat.late = late;
- if (t_release > t_now) {
+ if (not late) {
+ const auto wait_time = t_release - t_now;
std::this_thread::sleep_for(wait_time);
}
- else if (drop_late) {
+
+ stat.buffering_time_us = duration_cast<microseconds>(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 = metadata.utc_offset;
- edi_tagDETI.seconds = metadata.edi_time;
+ 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
@@ -221,53 +229,69 @@ void EDISender::process()
frame_t frame;
frames.wait_and_pop(frame);
- if (not running.load() or frame.first.empty()) {
+ if (not running.load() or frame.data.empty()) {
break;
}
- if (frame.first.size() == 6144) {
- send_eti_frame(frame.first.data(), frame.second);
+ if (frame.data.size() == 6144) {
+ send_eti_frame(frame);
}
else {
etiLog.level(warn) << "Ignoring short ETI frame, "
- "DFLC=" << frame.second.dlfc << ", len=" <<
- frame.first.size();
+ "DFLC=" << frame.metadata.dlfc << ", len=" <<
+ frame.data.size();
}
- if (wait_times.size() == 250) { // every six seconds
- const double n = wait_times.size();
+ if (buffering_stats.size() == 250) { // every six seconds
+ const double n = buffering_stats.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; });
+ 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<double>::max();
+ double max = -std::numeric_limits<double>::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 t : wait_times) {
+ 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);
- 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;
+ for (const auto t : buffering_stats) {
+ ss << " " << lrint(t.buffering_time_us / 1000.0);
}
etiLog.level(debug) << ss.str();
- */
+ // */
- etiLog.level(info) << "Wait time statistics [microseconds]:"
- " min: " << *min_max.first <<
- " max: " << *min_max.second <<
+ etiLog.level(info) << "Buffering time statistics [milliseconds]:"
+ " min: " << min <<
+ " max: " << max <<
" mean: " << mean <<
" stdev: " << stdev <<
" late: " <<
- num_late << " of " << wait_times.size() << " (" <<
+ num_late << " of " << buffering_stats.size() << " (" <<
num_late * 100.0 / n << "%)";
- wait_times.clear();
+ buffering_stats.clear();
}
}
}
diff --git a/src/zmq2edi/EDISender.h b/src/zmq2edi/EDISender.h
index 3525b4b..c953563 100644
--- a/src/zmq2edi/EDISender.h
+++ b/src/zmq2edi/EDISender.h
@@ -3,7 +3,7 @@
2011, 2012 Her Majesty the Queen in Right of Canada (Communications
Research Center Canada)
- Copyright (C) 2019
+ Copyright (C) 2020
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -45,7 +45,11 @@ struct metadata_t {
uint16_t dlfc;
};
-using frame_t = std::pair<std::vector<uint8_t>, metadata_t>;
+struct frame_t {
+ std::vector<uint8_t> data;
+ metadata_t metadata;
+ std::chrono::steady_clock::time_point received_at;
+};
class EDISender {
public:
@@ -55,11 +59,11 @@ class EDISender {
~EDISender();
void start(const edi::configuration_t& conf,
int delay_ms, bool drop_late_packets);
- void push_frame(const frame_t& frame);
+ void push_frame(frame_t&& frame);
void print_configuration(void);
private:
- void send_eti_frame(uint8_t* p, metadata_t metadata);
+ void send_eti_frame(frame_t& frame);
void process(void);
int tist_delay_ms;
@@ -67,13 +71,15 @@ class EDISender {
std::atomic<bool> running;
std::thread process_thread;
edi::configuration_t edi_conf;
- std::chrono::steady_clock::time_point startTime;
ThreadsafeQueue<frame_t> frames;
std::shared_ptr<edi::Sender> edi_sender;
- // For statistics about wait time before we transmit packets,
- // in microseconds
- std::vector<double> wait_times;
+ 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_stat_t> buffering_stats;
};
diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp
index f7d733c..20ddcbe 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) 2019
+ Copyright (C) 2020
Matthias P. Braendli, matthias.braendli@mpb.li
http://www.opendigitalradio.org
@@ -59,6 +59,8 @@ static void usage()
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 << " Negative delay values are also allowed." << endl;
+ cerr << " -C <path to script> 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 << " -p <destination port> Set the destination port." << endl;
cerr << " -P Disable PFT and send AFPackets." << endl;
@@ -222,6 +224,8 @@ static void parse_destination_args(char option)
}
}
+class FCTDiscontinuity { };
+
int start(int argc, char **argv)
{
edi_conf.enable_pft = true;
@@ -234,13 +238,17 @@ int start(int argc, char **argv)
int delay_ms = 500;
bool drop_late_packets = false;
uint32_t backoff_after_reset_ms = DEFAULT_BACKOFF;
+ std::string startupcheck;
int ch = 0;
while (ch != -1) {
- ch = getopt(argc, argv, "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:xh");
switch (ch) {
case -1:
break;
+ case 'C':
+ startupcheck = optarg;
+ break;
case 'd':
case 's':
case 'S':
@@ -298,6 +306,25 @@ int start(int argc, char **argv)
}
}
+ if (not startupcheck.empty()) {
+ etiLog.level(info) << "Running startup check '" << startupcheck << "'";
+ int wstatus = system(startupcheck.c_str());
+
+ if (WIFEXITED(wstatus)) {
+ if (WEXITSTATUS(wstatus) == 0) {
+ etiLog.level(info) << "Startup check ok";
+ }
+ else {
+ etiLog.level(error) << "Startup check failed, returned " << WEXITSTATUS(wstatus);
+ return 1;
+ }
+ }
+ else {
+ etiLog.level(error) << "Startup check failed, child didn't terminate normally";
+ return 1;
+ }
+ }
+
add_edi_destination();
if (optind >= argc) {
@@ -325,88 +352,100 @@ int start(int argc, char **argv)
zmq::context_t zmq_ctx(1);
etiLog.level(info) << "Opening ZMQ input: " << source_url;
- size_t num_consecutive_resets = 0;
while (true) {
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
size_t error_count = 0;
+ int previous_fct = -1;
+
+ try {
+ 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);
+ const auto received_at = std::chrono::steady_clock::now();
- 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 {
- num_consecutive_resets = 0;
+ zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
- // Event received: recv will not block
- zmq_sock.recv(&incoming);
+ if (dab_msg->version != 1) {
+ etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
+ error_count++;
+ }
- zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
+ int offset = sizeof(dab_msg->version) +
+ NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
- if (dab_msg->version != 1) {
- etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
- error_count++;
- }
+ std::vector<frame_t> all_frames;
+ all_frames.reserve(NUM_FRAMES_PER_ZMQ_MESSAGE);
- int offset = sizeof(dab_msg->version) +
- NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
+ 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 {
+ frame_t frame;
+ frame.data.resize(6144, 0x55);
+ frame.received_at = received_at;
- std::list<std::pair<std::vector<uint8_t>, metadata_t> > all_frames;
+ const int framesize = dab_msg->buflen[i];
- 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);
+ memcpy(frame.data.data(),
+ ((uint8_t*)incoming.data()) + offset,
+ framesize);
- const int framesize = dab_msg->buflen[i];
+ const int fct = frame.data[4];
- memcpy(&buf.front(),
- ((uint8_t*)incoming.data()) + offset,
- framesize);
+ const int expected_fct = (previous_fct + 1) % 250;
+ if (previous_fct != -1 and expected_fct != fct) {
+ etiLog.level(error) << "ETI wrong frame counter FCT=" << fct << " expected " << expected_fct;
+ throw FCTDiscontinuity();
+ }
+ previous_fct = fct;
- all_frames.emplace_back(
- std::piecewise_construct,
- std::make_tuple(std::move(buf)),
- std::make_tuple());
+ all_frames.push_back(std::move(frame));
- offset += framesize;
+ offset += framesize;
+ }
}
- }
- for (auto &f : all_frames) {
- size_t consumed_bytes = 0;
+ 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);
+ f.metadata = get_md_one_frame(
+ static_cast<uint8_t*>(incoming.data()) + offset,
+ incoming.size() - offset,
+ &consumed_bytes);
- offset += consumed_bytes;
- }
+ offset += consumed_bytes;
+ }
- for (auto &f : all_frames) {
- edisender.push_frame(f);
+ for (auto &f : all_frames) {
+ edisender.push_frame(std::move(f));
+ }
}
}
- }
- num_consecutive_resets++;
+ etiLog.level(info) << "Backoff " << backoff_after_reset_ms <<
+ "ms due to ZMQ input (" << source_url << ") timeout";
+ }
+ catch (const FCTDiscontinuity&) {
+ etiLog.level(info) << "Backoff " << backoff_after_reset_ms << "ms FCT discontinuity";
+ }
zmq_sock.close();
std::this_thread::sleep_for(std::chrono::milliseconds(backoff_after_reset_ms));
- etiLog.level(info) << "ZMQ input (" << source_url << ") timeout after " <<
- num_consecutive_resets << " consecutive resets.";
}
return 0;
@@ -423,12 +462,20 @@ int main(int argc, char **argv)
#endif
" starting up";
+ int ret = 1;
+
try {
- return start(argc, argv);
+ 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();
}
- catch (std::runtime_error &e) {
- etiLog.level(error) << "Error: " << e.what();
+ catch (const std::logic_error &e) {
+ etiLog.level(error) << "Logic error! " << e.what();
}
- return 1;
+ return ret;
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-27 08:52:00 +0000