fix(hook-frame): Rework framing hook to synchronize frame boundaries

Signed-off-by: Steffen Vogel <[email protected]>

Author: stv0g

doc/openapi/components/schemas/config/duration.yaml

@@ -0,0 +1,21 @@
+# yaml-language-server: $schema=http://json-schema.org/draft-07/schema
+# SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University
+# SPDX-License-Identifier: Apache-2.0
+---
+oneOf:
+- type: string
+  description: |
+    Duration as a string, e.g., "1h30m", "45s", "200ms".
+  pattern: (\d+d)?(\d+h)?(\d+m)?(\d+s)?(\d+ms)?(\d+us)?(\d+ns)?
+  examples:
+  - "6d23h30m50s40ms"
+  - "45s"
+  - "2d200ms"
+- type: integer
+  description: |
+    Duration as integer.
+  minimum: 0
+  examples:
+  - 5400000
+  - 45000
+  - 200

doc/openapi/components/schemas/config/hooks/frame.yaml

@@ -8,28 +8,14 @@ allOf:
     trigger:
       description: The trigger for new frames.
       type: string
-      default: sequence
+      default: timestamp
       enum:
       - sequence
       - timestamp
 
-    unit:
-      description: The type of a timestamp trigger.
-      type: string
-      enum:
-      - milliseconds
-      - seconds
-      - minutes
-      - hours
-
     interval:
       description: The interval in which frames are annotated.
-      type: number
-      default: 1
-
-    offset:
-      description: An offset in the interval for the annotation of new frames.
-      type: number
-      default: 0
+      default: "1s"
+      $ref: ../../duration.yaml
 
 - $ref: ../hook.yaml

etc/examples/hooks/frame.conf

@@ -13,8 +13,7 @@ paths = (
                 type = "frame"
 
                 trigger = "timestamp"
-                unit = "seconds"
-                interval = 10
+                interval = "3s"
             }
         )
     }

lib/hooks/frame.cpp

@@ -5,225 +5,92 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <chrono>
 #include <cstdint>
 #include <cstring>
-#include <limits>
-#include <memory>
-#include <optional>
+#include <variant>
 
 #include <jansson.h>
 
+#include <villas/config_helper.hpp>
 #include <villas/exceptions.hpp>
 #include <villas/hook.hpp>
 #include <villas/sample.hpp>
 
+#include "villas/timing.hpp"
+
 namespace villas {
 namespace node {
 
-enum class Trigger {
-  TIMESTAMP,
-  SEQUENCE,
-};
-
-enum class Unit {
-  MILLISECONDS,
-  SECONDS,
-  MINUTES,
-  HOURS,
-};
-
-uint64_t unit_wrap(std::optional<Unit> unit) {
-  if (unit) {
-    switch (*unit) {
-    case Unit::HOURS:
-      return 24;
-    case Unit::MINUTES:
-      return 60;
-    case Unit::SECONDS:
-      return 60;
-    case Unit::MILLISECONDS:
-      return 1000;
-    }
-  }
-
-  return std::numeric_limits<uint64_t>::max();
-}
-
 class FrameHook : public Hook {
-  using sample_ptr = std::unique_ptr<Sample, decltype(&sample_decref)>;
+  using TimeInterval = std::chrono::milliseconds;
+  using SequenceInterval = uint64_t;
 
 private:
-  Trigger trigger;
-  uint64_t interval;
-  uint64_t offset;
-  std::optional<Unit> unit;
-  sample_ptr last_smp;
+  std::variant<TimeInterval, SequenceInterval> interval;
+  Sample::PtrUnique last_smp;
 
   bool updateInterval(Sample const *next_smp) {
     bool changed = false;
 
     if (!last_smp.get() ||
         (next_smp->flags & (int)SampleFlags::NEW_SIMULATION)) {
       changed = true;
-    } else if (trigger == Trigger::SEQUENCE) {
+    } else if (auto *i = std::get_if<SequenceInterval>(&interval)) {
       if (!(next_smp->flags & (int)SampleFlags::HAS_SEQUENCE))
-        throw RuntimeError{"Missing sequence number."};
+        throw RuntimeError("Missing sequence number");
+
+      auto last_interval = (last_smp->sequence + *i) / *i;
+      auto next_interval = (next_smp->sequence + *i) / *i;
 
-      auto last_interval = (last_smp->sequence + interval - offset) / interval;
-      auto next_interval = (next_smp->sequence + interval - offset) / interval;
       changed = last_interval != next_interval;
-    } else {
+    } else if (auto *i = std::get_if<TimeInterval>(&interval)) {
       if (!(next_smp->flags & (int)SampleFlags::HAS_TS_ORIGIN))
-        throw RuntimeError{"Missing origin timestamp."};
-
-      switch (unit.value()) {
-      case Unit::HOURS: {
-        auto last_hour = last_smp->ts.origin.tv_sec / 3'600;
-        auto next_hour = next_smp->ts.origin.tv_sec / 3'600;
-
-        auto last_day = last_hour / 24;
-        auto next_day = next_hour / 24;
-        if (last_day != next_day) {
-          changed = true;
-          break;
-        }
-
-        auto last_hour_of_day = last_hour - 24 * last_day;
-        auto next_hour_of_day = next_hour - 24 * next_day;
-        auto last_interval_of_day =
-            (last_hour_of_day + interval - offset) / interval;
-        auto next_interval_of_day =
-            (next_hour_of_day + interval - offset) / interval;
-        changed = last_interval_of_day != next_interval_of_day;
-        break;
-      }
+        throw RuntimeError("Missing origin timestamp");
 
-      case Unit::MINUTES: {
-        auto last_minute = last_smp->ts.origin.tv_sec / 60;
-        auto next_minute = next_smp->ts.origin.tv_sec / 60;
+      auto last_ts = time_to_timepoint<TimeInterval>(&last_smp->ts.origin);
+      auto next_ts = time_to_timepoint<TimeInterval>(&next_smp->ts.origin);
 
-        auto last_hour = last_minute / 60;
-        auto next_hour = next_minute / 60;
-        if (last_hour != next_hour) {
-          changed = true;
-          break;
-        }
+      auto last_interval = (last_ts.time_since_epoch() + *i) / *i;
+      auto next_interval = (next_ts.time_since_epoch() + *i) / *i;
 
-        auto last_minute_of_hour = last_minute - 60 * last_hour;
-        auto next_minute_of_hour = next_minute - 60 * next_hour;
-        auto last_interval_of_hour =
-            (last_minute_of_hour + interval - offset) / interval;
-        auto next_interval_of_hour =
-            (next_minute_of_hour + interval - offset) / interval;
-        changed = last_interval_of_hour != next_interval_of_hour;
-        break;
-      }
-
-      case Unit::SECONDS: {
-        auto last_second = last_smp->ts.origin.tv_sec;
-        auto next_second = next_smp->ts.origin.tv_sec;
-
-        auto last_minute = last_second / 60;
-        auto next_minute = next_second / 60;
-        if (last_minute != next_minute) {
-          changed = true;
-          break;
-        }
-
-        auto last_second_of_minute = last_second - 60 * last_minute;
-        auto next_second_of_minute = next_second - 60 * next_minute;
-        auto last_interval_of_minute =
-            (last_second_of_minute + interval - offset) / interval;
-        auto next_interval_of_minute =
-            (next_second_of_minute + interval - offset) / interval;
-        changed = last_interval_of_minute != next_interval_of_minute;
-        break;
-      }
-
-      case Unit::MILLISECONDS: {
-        auto last_second = last_smp->ts.origin.tv_sec;
-        auto next_second = next_smp->ts.origin.tv_sec;
-        if (last_second != next_second) {
-          changed = true;
-          break;
-        }
-
-        auto last_millisecond_of_second =
-            last_smp->ts.origin.tv_nsec / 1'000'000;
-        auto next_millisecond_of_second =
-            next_smp->ts.origin.tv_nsec / 1'000'000;
-        auto last_interval_of_second =
-            (last_millisecond_of_second + interval - offset) / interval;
-        auto next_interval_of_second =
-            (next_millisecond_of_second + interval - offset) / interval;
-        changed = last_interval_of_second != next_interval_of_second;
-        break;
-      }
-      }
+      changed = last_interval != next_interval;
     }
 
-    if (changed)
-      logger->debug("new frame");
-
     return changed;
   }
 
 public:
   FrameHook(Path *p, Node *n, int fl, int prio, bool en = true)
-      : Hook(p, n, fl, prio, en), trigger(Trigger::SEQUENCE), interval(1),
-        offset(0), unit{std::nullopt}, last_smp{nullptr, &sample_decref} {}
+      : Hook(p, n, fl, prio, en), interval(TimeInterval(0)),
+        last_smp{nullptr, &sample_decref} {}
 
   virtual ~FrameHook() { (void)last_smp.release(); }
 
   void parse(json_t *json) override {
     Hook::parse(json);
 
     char *trigger_str = nullptr;
-    char *unit_str = nullptr;
-    int interval_int = -1;
-    int offset_int = -1;
+    json_t *json_interval = nullptr;
 
     json_error_t err;
-    auto ret = json_unpack_ex(json, &err, 0, "{ s?: s, s?: s, s?: i, s?: i }",
-                              "trigger", &trigger_str, "unit", &unit_str,
-                              "interval", &interval_int, "offset", &offset_int);
+    auto ret = json_unpack_ex(json, &err, 0, "{ s?: s, s: o }", "trigger",
+                              &trigger_str, "interval", &json_interval);
     if (ret)
       throw ConfigError(json, err, "node-config-hook-frame");
 
-    if (trigger_str) {
-      if (!strcmp(trigger_str, "sequence"))
-        trigger = Trigger::SEQUENCE;
-      else if (!strcmp(trigger_str, "timestamp"))
-        trigger = Trigger::TIMESTAMP;
-      else
-        throw ConfigError(json, "node-config-hook-frame-unit");
-    }
-
-    if (trigger == Trigger::TIMESTAMP) {
-      if (!strcmp(unit_str, "milliseconds"))
-        unit = Unit::MILLISECONDS;
-      else if (!strcmp(unit_str, "seconds"))
-        unit = Unit::SECONDS;
-      else if (!strcmp(unit_str, "minutes"))
-        unit = Unit::MINUTES;
-      else if (!strcmp(unit_str, "hours"))
-        unit = Unit::HOURS;
-      else
-        throw ConfigError(json, "node-config-hook-frame-unit");
-    }
-
-    if (interval_int != -1) {
-      if (interval_int <= 0 || (uint64_t)interval_int > unit_wrap(unit))
-        throw ConfigError(json, "node-config-hook-frame-interval");
-
-      interval = interval_int;
-    }
+    if (trigger_str == nullptr || !strcmp(trigger_str, "timestamp")) {
+      interval = parse_duration<TimeInterval>(json_interval);
 
-    if (offset_int != -1) {
-      if (offset_int < 0 || (uint64_t)offset_int >= unit_wrap(unit))
-        throw ConfigError(json, "node-config-hook-frame-offset");
+      if (std::get<TimeInterval>(interval) == TimeInterval::zero())
+        throw ConfigError(json, "node-config-hook-frame-interval",
+                          "Interval must be greater than zero");
+    } else if (!strcmp(trigger_str, "sequence")) {
+      if (!json_is_integer(json))
+        throw ConfigError(json, "node-config-hook-frame-interval",
+                          "Interval must be an integer");
 
-      offset = offset_int;
+      interval = SequenceInterval(json_integer_value(json_interval));
     }
   }
 
@@ -236,7 +103,7 @@ class FrameHook : public Hook {
       smp->flags &= ~(int)SampleFlags::NEW_FRAME;
 
     sample_incref(smp);
-    last_smp = sample_ptr{smp, &sample_decref};
+    last_smp = Sample::PtrUnique{smp, &sample_decref};
 
     return Reason::OK;
   }