Fix: rounding error in st10_tai_to_media_clk (#1342)

For 59.94 fps RTP timestamp increment should alternate between 1501,
1502 for two consecutive frames.
Before, the same increment value was sometimes repeated, this fixes it.

Additionally:
- noctx integration tests for st40 user pacing was added
- added tests for st10_tai_to_media_clk and st10_media_clk_to_ns

Fixes:
#1325

Signed-off-by: Kasiewicz, Marek <[email protected]>

Author: Sakoram

lib/src/st2110/pipeline/st40_pipeline_rx.c

@@ -88,11 +88,7 @@ static int rx_st40p_rtp_ready(void* priv) {
 
   struct rte_mbuf* pkt = mbuf;
   enum mtl_port port = mt_port_by_id(ctx->impl, pkt->port);
-  uint64_t receive_timestamp = 0;
-  if (port < MTL_PORT_MAX)
-    receive_timestamp = mt_mbuf_time_stamp(ctx->impl, pkt, port);
-  else
-    receive_timestamp = mtl_ptp_read_time(ctx->impl);
+  uint64_t receive_timestamp = mt_mbuf_time_stamp(ctx->impl, pkt, port);
 
   uint32_t hdr_bytes = sizeof(struct st40_rfc8331_rtp_hdr);
   if (len < hdr_bytes) {

lib/src/st2110/st_fmt.c

@@ -906,16 +906,35 @@ bool st_frame_fmt_equal_transport(enum st_frame_fmt fmt, enum st20_fmt tfmt) {
   return (fmt == to_fmt) ? true : false;
 }
 
+static uint64_t st_muldiv_u64_round_closest(uint64_t value, uint64_t multiplier,
+                                            uint64_t divisor) {
+  /* keep conversions reproducible without relying on floating point */
+  __uint128_t product = (__uint128_t)value * multiplier;
+  __uint128_t quotient = product / divisor;
+  __uint128_t remainder = product - quotient * divisor;
+  __uint128_t half = divisor / 2;
+
+  if (remainder > half) quotient++; /* ties round down to keep jitter bounded */
+
+  return (uint64_t)quotient;
+}
+
 uint32_t st10_tai_to_media_clk(uint64_t tai_ns, uint32_t sampling_rate) {
-  long double ts = (long double)tai_ns * sampling_rate / NS_PER_S;
-  uint64_t tmstamp64 = ts;
-  uint32_t tmstamp32 = tmstamp64;
-  return tmstamp32;
+  if (!sampling_rate) {
+    err("%s, invalid sampling rate\n", __func__);
+    return 0;
+  }
+
+  return (uint32_t)st_muldiv_u64_round_closest(tai_ns, sampling_rate, NS_PER_S);
 }
 
 uint64_t st10_media_clk_to_ns(uint32_t media_ts, uint32_t sampling_rate) {
-  long double ts = (long double)media_ts * NS_PER_S / sampling_rate;
-  return ts;
+  if (!sampling_rate) {
+    err("%s, invalid sampling rate\n", __func__);
+    return 0;
+  }
+
+  return st_muldiv_u64_round_closest(media_ts, NS_PER_S, sampling_rate);
 }
 
 int st_draw_logo(struct st_frame* frame, struct st_frame* logo, uint32_t x, uint32_t y) {

tests/integration_tests/noctx/core/test_fixture.cpp

@@ -15,9 +15,37 @@
 
 #include "handlers/st20p_handler.hpp"
 #include "handlers/st30p_handler.hpp"
+#include "handlers/st40p_handler.hpp"
 #include "session.hpp"
 #include "strategy.hpp"
 
+namespace {
+
+struct TestPtpClockState {
+  std::atomic<uint64_t> start_ns{0};
+  std::atomic<bool> running{false};
+};
+
+TestPtpClockState g_test_ptp_clock;
+
+uint64_t monotonicNowNs() {
+  struct timespec spec;
+  clock_gettime(CLOCK_MONOTONIC, &spec);
+  return (uint64_t)spec.tv_sec * NS_PER_S + spec.tv_nsec;
+}
+
+void startClock(uint64_t now) {
+  g_test_ptp_clock.start_ns.store(now, std::memory_order_release);
+  g_test_ptp_clock.running.store(true, std::memory_order_release);
+}
+
+void resetClock() {
+  g_test_ptp_clock.running.store(false, std::memory_order_release);
+  g_test_ptp_clock.start_ns.store(0, std::memory_order_release);
+}
+
+} /* namespace */
+
 void NoCtxTest::SetUp() {
   ctx = new st_tests_context;
   if (!ctx) {
@@ -42,10 +70,12 @@ void NoCtxTest::SetUp() {
   ctx->para.tx_queues_cnt[MTL_PORT_R] = 16;
   ctx->para.rx_queues_cnt[MTL_PORT_P] = 16;
   ctx->para.rx_queues_cnt[MTL_PORT_R] = 16;
+  ResetFakePtpClock();
   defaultTestDuration = 20;
 }
 
 void NoCtxTest::TearDown() {
+  st40pHandlers.clear();
   st30pHandlers.clear();
   st20pHandlers.clear();
   frameTestStrategies.clear();
@@ -61,34 +91,37 @@ void NoCtxTest::TearDown() {
   }
 }
 
-uint64_t NoCtxTest::TestPtpSourceSinceEpoch(void* priv) {
-  struct timespec spec;
-  static std::atomic<uint64_t> adjustment_ns{0};
-
-  if (adjustment_ns.load() == 0 || priv == nullptr) {
-    struct timespec spec_adjustment_to_epoch;
-    clock_gettime(CLOCK_MONOTONIC, &spec_adjustment_to_epoch);
-    uint64_t temp_adjustment = (uint64_t)spec_adjustment_to_epoch.tv_sec * NS_PER_S +
-                               spec_adjustment_to_epoch.tv_nsec;
+uint64_t NoCtxTest::FakePtpClockNow(void* priv) {
+  (void)priv;
+  if (!g_test_ptp_clock.running.load(std::memory_order_acquire)) {
+    StartFakePtpClock();
+  }
 
-    adjustment_ns.store(temp_adjustment);
+  const uint64_t start = g_test_ptp_clock.start_ns.load(std::memory_order_acquire);
+  const uint64_t now = monotonicNowNs();
+  if (now <= start) {
+    return 0;
   }
 
-  clock_gettime(CLOCK_MONOTONIC, &spec);
-  uint64_t result =
-      ((uint64_t)spec.tv_sec * NS_PER_S + spec.tv_nsec) - adjustment_ns.load();
+  return now - start;
+}
+
+void NoCtxTest::StartFakePtpClock() {
+  startClock(monotonicNowNs());
+}
 
-  return result;
+void NoCtxTest::ResetFakePtpClock() {
+  resetClock();
 }
 
 void NoCtxTest::sleepUntilFailure(int sleep_duration) {
   if (!sleep_duration) {
     sleep_duration = defaultTestDuration;
   }
 
-  for (int i = 0; i < sleep_duration; ++i) {
+  for (int i = 0; i < sleep_duration * 10; ++i) {
     if (HasFailure()) break;
-    sleep(1);
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }
 }
 
@@ -106,6 +139,8 @@ NoCtxTest::St20pHandlerBundle NoCtxTest::createSt20pHandlerBundle(
     configure(handler);
   }
 
+  handler->normalizeSessionOps();
+
   std::unique_ptr<FrameTestStrategy> strategyOwned;
   FrameTestStrategy* strategy = nullptr;
   if (strategyFactory) {
@@ -153,6 +188,8 @@ NoCtxTest::St30pHandlerBundle NoCtxTest::createSt30pHandlerBundle(
     configure(handler);
   }
 
+  handler->normalizeSessionOps();
+
   std::unique_ptr<FrameTestStrategy> strategyOwned;
   FrameTestStrategy* strategy = nullptr;
   if (strategyFactory) {
@@ -185,12 +222,58 @@ NoCtxTest::St30pHandlerBundle NoCtxTest::registerSt30pResources(
   return bundle;
 }
 
-void NoCtxTest::initSt20pDefaultContext() {
+NoCtxTest::St40pHandlerBundle NoCtxTest::createSt40pHandlerBundle(
+    bool createTx, bool createRx,
+    std::function<FrameTestStrategy*(St40pHandler*)> strategyFactory,
+    std::function<void(St40pHandler*)> configure) {
+  if (!ctx) {
+    throw std::runtime_error("createSt40pHandlerBundle expects initialized ctx");
+  }
+
+  auto handlerOwned = std::make_unique<St40pHandler>(ctx);
+  auto* handler = handlerOwned.get();
+  if (configure) {
+    configure(handler);
+  }
+
+  std::unique_ptr<FrameTestStrategy> strategyOwned;
+  FrameTestStrategy* strategy = nullptr;
+  if (strategyFactory) {
+    strategyOwned.reset(strategyFactory(handler));
+    strategy = strategyOwned.get();
+    handler->setFrameTestStrategy(strategy);
+  }
+
+  if (createRx) {
+    handler->createSessionRx();
+  }
+  if (createTx) {
+    handler->createSessionTx();
+  }
+
+  return registerSt40pResources(std::move(handlerOwned), std::move(strategyOwned));
+}
+
+NoCtxTest::St40pHandlerBundle NoCtxTest::registerSt40pResources(
+    std::unique_ptr<St40pHandler> handler, std::unique_ptr<FrameTestStrategy> strategy) {
+  St40pHandlerBundle bundle;
+  if (handler) {
+    bundle.handler = handler.get();
+    st40pHandlers.emplace_back(std::move(handler));
+  }
+  if (strategy) {
+    bundle.strategy = strategy.get();
+    frameTestStrategies.emplace_back(std::move(strategy));
+  }
+  return bundle;
+}
+
+void NoCtxTest::initDefaultContext() {
   if (!ctx) {
-    throw std::runtime_error("initSt20pDefaultContext expects initialized ctx");
+    throw std::runtime_error("initDefaultContext expects initialized ctx");
   }
 
-  ctx->para.ptp_get_time_fn = NoCtxTest::TestPtpSourceSinceEpoch;
+  ctx->para.ptp_get_time_fn = NoCtxTest::FakePtpClockNow;
   ctx->para.log_level = MTL_LOG_LEVEL_INFO;
   ctx->para.flags &= ~MTL_FLAG_DEV_AUTO_START_STOP;
   ctx->handle = mtl_init(&ctx->para);

tests/integration_tests/noctx/core/test_fixture.hpp

@@ -13,6 +13,7 @@
 
 #include "handlers/st20p_handler.hpp"
 #include "handlers/st30p_handler.hpp"
+#include "handlers/st40p_handler.hpp"
 #include "tests.hpp"
 
 class Session;
@@ -35,10 +36,13 @@ class NoCtxTest : public ::testing::Test {
 
   using St20pHandlerBundle = HandlerBundle<St20pHandler>;
   using St30pHandlerBundle = HandlerBundle<St30pHandler>;
+  using St40pHandlerBundle = HandlerBundle<St40pHandler>;
 
   uint defaultTestDuration = 0;
 
-  static uint64_t TestPtpSourceSinceEpoch(void* priv);
+  static uint64_t FakePtpClockNow(void* priv);
+  static void StartFakePtpClock();
+  static void ResetFakePtpClock();
 
   void sleepUntilFailure(int sleepDuration = 0);
   St20pHandlerBundle createSt20pHandlerBundle(
@@ -53,11 +57,18 @@ class NoCtxTest : public ::testing::Test {
       std::function<void(St30pHandler*)> configure = nullptr);
   St30pHandlerBundle registerSt30pResources(std::unique_ptr<St30pHandler> handler,
                                             std::unique_ptr<FrameTestStrategy> strategy);
-  void initSt20pDefaultContext();
+  St40pHandlerBundle createSt40pHandlerBundle(
+      bool createTx, bool createRx,
+      std::function<FrameTestStrategy*(St40pHandler*)> strategyFactory,
+      std::function<void(St40pHandler*)> configure = nullptr);
+  St40pHandlerBundle registerSt40pResources(std::unique_ptr<St40pHandler> handler,
+                                            std::unique_ptr<FrameTestStrategy> strategy);
+  void initDefaultContext();
   bool waitForSession(Session& session,
                       std::chrono::milliseconds timeout =
                           std::chrono::milliseconds(SessionStartTimeoutMs));
 
+  std::vector<std::unique_ptr<St40pHandler>> st40pHandlers;
   std::vector<std::unique_ptr<St30pHandler>> st30pHandlers;
   std::vector<std::unique_ptr<St20pHandler>> st20pHandlers;
   std::vector<std::unique_ptr<FrameTestStrategy>> frameTestStrategies;