itten
268 lines
12 KiB
C++
268 lines
12 KiB
C++
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <cstring>
|
|
#include <exception>
|
|
#include <limits>
|
|
|
|
#include <mxl/flow.h>
|
|
#include <mxl/time.h>
|
|
|
|
#include "NodeBase.hpp"
|
|
#include "ST2110Receiver.hpp"
|
|
#include "V210.hpp"
|
|
|
|
namespace dmf {
|
|
|
|
namespace {
|
|
|
|
class RtpIndexMapper {
|
|
public:
|
|
RtpIndexMapper(const mxlRational& rate, int latency_frames)
|
|
: rate_(rate), latency_frames_(latency_frames) {}
|
|
|
|
uint64_t index_for(uint32_t rtp_timestamp, uint64_t current_mxl_index) {
|
|
const uint64_t rtp_ext = extend_rtp(rtp_timestamp);
|
|
if (!anchored_) {
|
|
anchored_ = true;
|
|
base_rtp_ext_ = rtp_ext;
|
|
base_mxl_index_ = current_mxl_index + static_cast<uint64_t>(latency_frames_);
|
|
last_offset_ = 0;
|
|
return base_mxl_index_;
|
|
}
|
|
|
|
const uint64_t rtp_delta = rtp_ext - base_rtp_ext_;
|
|
const uint64_t denominator = 90'000ULL * static_cast<uint64_t>(rate_.denominator);
|
|
const uint64_t numerator = rtp_delta * static_cast<uint64_t>(rate_.numerator);
|
|
const uint64_t offset = (numerator + denominator / 2) / denominator;
|
|
if (offset <= last_offset_) {
|
|
duplicate_or_backwards_++;
|
|
last_gap_frames_ = 0;
|
|
} else if (offset > last_offset_ + 1) {
|
|
last_gap_frames_ = offset - (last_offset_ + 1);
|
|
rtp_gap_frames_ += last_gap_frames_;
|
|
} else {
|
|
last_gap_frames_ = 0;
|
|
}
|
|
last_offset_ = offset;
|
|
return base_mxl_index_ + offset;
|
|
}
|
|
|
|
uint32_t last_rtp_delta() const { return last_rtp_delta_; }
|
|
uint32_t expected_rtp_delta() const {
|
|
return static_cast<uint32_t>(
|
|
(90'000ULL * static_cast<uint64_t>(rate_.denominator)) /
|
|
static_cast<uint64_t>(rate_.numerator));
|
|
}
|
|
uint64_t last_gap_frames() const { return last_gap_frames_; }
|
|
uint64_t rtp_gap_frames() const { return rtp_gap_frames_; }
|
|
uint64_t duplicate_or_backwards() const { return duplicate_or_backwards_; }
|
|
uint64_t base_mxl_index() const { return base_mxl_index_; }
|
|
|
|
private:
|
|
uint64_t extend_rtp(uint32_t rtp_timestamp) {
|
|
if (!have_last_rtp_) {
|
|
have_last_rtp_ = true;
|
|
last_rtp_ = rtp_timestamp;
|
|
return rtp_timestamp;
|
|
}
|
|
|
|
if (rtp_timestamp < last_rtp_ &&
|
|
static_cast<uint32_t>(last_rtp_ - rtp_timestamp) > 0x80000000u) {
|
|
rtp_cycles_ += 0x1'0000'0000ULL;
|
|
}
|
|
last_rtp_delta_ = rtp_timestamp - last_rtp_;
|
|
last_rtp_ = rtp_timestamp;
|
|
return rtp_cycles_ + rtp_timestamp;
|
|
}
|
|
|
|
mxlRational rate_{};
|
|
int latency_frames_ = 2;
|
|
bool anchored_ = false;
|
|
bool have_last_rtp_ = false;
|
|
uint32_t last_rtp_ = 0;
|
|
uint64_t rtp_cycles_ = 0;
|
|
uint64_t base_rtp_ext_ = 0;
|
|
uint64_t base_mxl_index_ = 0;
|
|
uint64_t last_offset_ = 0;
|
|
uint64_t rtp_gap_frames_ = 0;
|
|
uint64_t duplicate_or_backwards_ = 0;
|
|
uint64_t last_gap_frames_ = 0;
|
|
uint32_t last_rtp_delta_ = 0;
|
|
};
|
|
|
|
} // namespace
|
|
|
|
class ST2110In : public NodeBase {
|
|
void run() override {
|
|
ST2110ReceiverConfig cfg;
|
|
try {
|
|
cfg = parse_st2110_receiver_config(config());
|
|
} catch (const std::exception& e) {
|
|
log("config error: %s", e.what());
|
|
return;
|
|
}
|
|
|
|
log("SMPTE 2110-20 RX %s:%u from %s on %s backend=%s local=%s %dx%d depth=%d @ %d/%d latency=%d index_mode=%s",
|
|
cfg.mcast_ip.c_str(), cfg.udp_port, cfg.source_ip.c_str(), cfg.ifname.c_str(),
|
|
cfg.backend.c_str(), cfg.local_ip.c_str(), cfg.width, cfg.height, cfg.depth,
|
|
cfg.fps_num, cfg.fps_den, cfg.mxl_latency_frames, cfg.mxl_index_mode.c_str());
|
|
|
|
try {
|
|
MTLContext mtl(cfg);
|
|
ST20RxSession rx(mtl.get(), mtl.port_name(), cfg);
|
|
MXLVideoWriter writer(instance(), cfg, node_id());
|
|
|
|
const mxlRational video_rate = {cfg.fps_num, cfg.fps_den};
|
|
const uint32_t video_stride = writer.config().discrete.sliceSizes[0];
|
|
const uint32_t uyvy_stride = static_cast<uint32_t>(cfg.width * 2);
|
|
const auto expected_fmt = cfg.output_fmt();
|
|
RtpIndexMapper rtp_mapper(video_rate, cfg.mxl_latency_frames);
|
|
uint64_t last_published_index = std::numeric_limits<uint64_t>::max();
|
|
uint64_t frames_written = 0;
|
|
uint64_t incomplete_frames = 0;
|
|
uint64_t unexpected_format_frames = 0;
|
|
uint64_t mxl_open_failures = 0;
|
|
uint64_t index_resyncs = 0;
|
|
uint64_t skipped_indices = 0;
|
|
uint64_t backwards_indices = 0;
|
|
uint64_t last_report_frames = 0;
|
|
auto last_report = std::chrono::steady_clock::now();
|
|
|
|
log("MTL version=%s port=%s output_stride=%u input_stride=%u direct_v210=%d",
|
|
mtl_version(), mtl.port_name().c_str(), video_stride,
|
|
cfg.direct_v210() ? video_stride : uyvy_stride, cfg.direct_v210() ? 1 : 0);
|
|
|
|
while (g_running.load(std::memory_order_relaxed)) {
|
|
st_frame* frame = st20p_rx_get_frame(rx.get());
|
|
if (!frame) {
|
|
continue;
|
|
}
|
|
|
|
if (!st_is_frame_complete(frame->status)) {
|
|
incomplete_frames++;
|
|
st20p_rx_put_frame(rx.get(), frame);
|
|
continue;
|
|
}
|
|
|
|
if (frame->fmt != expected_fmt) {
|
|
unexpected_format_frames++;
|
|
log("unexpected MTL frame fmt=%d; expected %d", frame->fmt, expected_fmt);
|
|
st20p_rx_put_frame(rx.get(), frame);
|
|
continue;
|
|
}
|
|
|
|
mxlGrainInfo grain{};
|
|
uint8_t* video_buf = nullptr;
|
|
const uint64_t current_index = mxlGetCurrentIndex(&video_rate);
|
|
uint64_t video_index = 0;
|
|
if (cfg.mxl_index_mode == "rtp") {
|
|
video_index = rtp_mapper.index_for(frame->rtp_timestamp, current_index);
|
|
if (rtp_mapper.last_gap_frames() > 0) {
|
|
log("RTP gap frame=%llu rtp=%u delta=%u expected=%u gap_frames=%llu",
|
|
static_cast<unsigned long long>(frames_written),
|
|
frame->rtp_timestamp,
|
|
rtp_mapper.last_rtp_delta(),
|
|
rtp_mapper.expected_rtp_delta(),
|
|
static_cast<unsigned long long>(rtp_mapper.last_gap_frames()));
|
|
}
|
|
} else {
|
|
video_index = current_index + static_cast<uint64_t>(cfg.mxl_latency_frames);
|
|
}
|
|
|
|
if (last_published_index != std::numeric_limits<uint64_t>::max()) {
|
|
if (video_index <= last_published_index) {
|
|
backwards_indices++;
|
|
video_index = last_published_index + 1;
|
|
} else if (video_index > last_published_index + 1) {
|
|
skipped_indices += video_index - (last_published_index + 1);
|
|
}
|
|
}
|
|
|
|
mxlStatus st = mxlFlowWriterOpenGrain(writer.get(), video_index, &grain, &video_buf);
|
|
if (st != MXL_STATUS_OK) {
|
|
mxl_open_failures++;
|
|
log("mxlFlowWriterOpenGrain failed (%s) index=%llu current=%llu", mxl_status_str(st),
|
|
static_cast<unsigned long long>(video_index),
|
|
static_cast<unsigned long long>(current_index));
|
|
st20p_rx_put_frame(rx.get(), frame);
|
|
continue;
|
|
}
|
|
|
|
if (last_published_index != std::numeric_limits<uint64_t>::max() &&
|
|
video_index > last_published_index + 1) {
|
|
index_resyncs++;
|
|
log("MXL index resync last=%llu next=%llu current=%llu",
|
|
static_cast<unsigned long long>(last_published_index),
|
|
static_cast<unsigned long long>(video_index),
|
|
static_cast<unsigned long long>(current_index));
|
|
}
|
|
|
|
if (cfg.direct_v210()) {
|
|
const auto* src = static_cast<const uint8_t*>(frame->addr[0]);
|
|
const size_t src_stride = frame->linesize[0] ? frame->linesize[0]
|
|
: static_cast<size_t>(cfg.width / 6) * 16;
|
|
const size_t row_bytes = static_cast<size_t>(cfg.width / 6) * 16;
|
|
for (int y = 0; y < cfg.height; ++y) {
|
|
std::memcpy(video_buf + static_cast<size_t>(y) * video_stride,
|
|
src + static_cast<size_t>(y) * src_stride, row_bytes);
|
|
}
|
|
} else {
|
|
const uint32_t src_stride = frame->linesize[0]
|
|
? static_cast<uint32_t>(frame->linesize[0])
|
|
: uyvy_stride;
|
|
v210::UYVYtoV210(static_cast<const uint8_t*>(frame->addr[0]), video_buf,
|
|
cfg.width, cfg.height, src_stride, video_stride);
|
|
}
|
|
grain.flags = 0;
|
|
grain.validSlices = grain.totalSlices;
|
|
mxlFlowWriterCommitGrain(writer.get(), &grain);
|
|
st20p_rx_put_frame(rx.get(), frame);
|
|
|
|
frames_written++;
|
|
last_published_index = video_index;
|
|
const auto now = std::chrono::steady_clock::now();
|
|
const auto elapsed = std::chrono::duration<double>(now - last_report).count();
|
|
if (elapsed >= 5.0) {
|
|
const uint64_t delta = frames_written - last_report_frames;
|
|
const double measured_fps = static_cast<double>(delta) / elapsed;
|
|
log("stats frames=%llu fps=%.2f incomplete=%llu bad_fmt=%llu mxl_open_fail=%llu resync=%llu skipped=%llu backwards=%llu rtp_gap=%llu rtp_dup=%llu last_index=%llu current=%llu",
|
|
static_cast<unsigned long long>(frames_written), measured_fps,
|
|
static_cast<unsigned long long>(incomplete_frames),
|
|
static_cast<unsigned long long>(unexpected_format_frames),
|
|
static_cast<unsigned long long>(mxl_open_failures),
|
|
static_cast<unsigned long long>(index_resyncs),
|
|
static_cast<unsigned long long>(skipped_indices),
|
|
static_cast<unsigned long long>(backwards_indices),
|
|
static_cast<unsigned long long>(rtp_mapper.rtp_gap_frames()),
|
|
static_cast<unsigned long long>(rtp_mapper.duplicate_or_backwards()),
|
|
static_cast<unsigned long long>(last_published_index),
|
|
static_cast<unsigned long long>(mxlGetCurrentIndex(&video_rate)));
|
|
last_report = now;
|
|
last_report_frames = frames_written;
|
|
}
|
|
}
|
|
|
|
log("stopped frames=%llu incomplete=%llu bad_fmt=%llu mxl_open_fail=%llu resync=%llu skipped=%llu backwards=%llu rtp_gap=%llu rtp_dup=%llu",
|
|
static_cast<unsigned long long>(frames_written),
|
|
static_cast<unsigned long long>(incomplete_frames),
|
|
static_cast<unsigned long long>(unexpected_format_frames),
|
|
static_cast<unsigned long long>(mxl_open_failures),
|
|
static_cast<unsigned long long>(index_resyncs),
|
|
static_cast<unsigned long long>(skipped_indices),
|
|
static_cast<unsigned long long>(backwards_indices),
|
|
static_cast<unsigned long long>(rtp_mapper.rtp_gap_frames()),
|
|
static_cast<unsigned long long>(rtp_mapper.duplicate_or_backwards()));
|
|
} catch (const std::exception& e) {
|
|
log("error: %s", e.what());
|
|
}
|
|
}
|
|
};
|
|
|
|
} // namespace dmf
|
|
|
|
int main() {
|
|
dmf::ST2110In node;
|
|
return node.execute();
|
|
}
|