aboutsummaryrefslogtreecommitdiffstats
path: root/src/zmq2edi/EDISender.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/zmq2edi/EDISender.cpp')
-rw-r--r--src/zmq2edi/EDISender.cpp98
1 files changed, 61 insertions, 37 deletions
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();
}
}
}