Synnax 52.9 (#2064)

Author: emilbon99

arc/cpp/runtime/BUILD.bazel

@@ -27,6 +27,7 @@ cc_test(
     srcs = ["runtime_test.cpp"],
     deps = [
         ":runtime",
+        "//arc/cpp/ir/testutil",
         "//arc/cpp/runtime/testutil",
         "//x/cpp/telem",
         "//x/cpp/test",

arc/cpp/runtime/loop/loop.h

@@ -249,8 +249,14 @@ struct Loop {
     /// @brief Block until timer/external event or breaker stops.
     /// Must be called from the runtime thread only.
     /// @param breaker Controls loop termination; wait() returns when breaker stops.
+    /// @param max_timeout Upper bound on how long to sleep. When positive, the loop
+    /// will wake after at most this duration even if no timer or input fires.
+    /// A value of 0 means no deadline constraint (use the loop's configured timing).
     /// @return WakeReason indicating why wait() returned.
-    virtual WakeReason wait(x::breaker::Breaker &breaker) = 0;
+    virtual WakeReason wait(
+        x::breaker::Breaker &breaker,
+        x::telem::TimeSpan max_timeout = x::telem::TimeSpan(0)
+    ) = 0;
 
     /// @brief Initialize loop resources. Must be called before wait().
     /// Applies RT configuration (priority, affinity, memory lock) if configured.

arc/cpp/runtime/loop/loop_darwin.cpp

@@ -39,19 +39,22 @@ class DarwinLoop final : public Loop {
 
     ~DarwinLoop() override { this->close_fds(); }
 
-    WakeReason wait(x::breaker::Breaker &breaker) override {
+    WakeReason wait(
+        x::breaker::Breaker &breaker,
+        x::telem::TimeSpan max_timeout = x::telem::TimeSpan(0)
+    ) override {
         if (this->kqueue_fd_ == -1) return WakeReason::Shutdown;
 
         switch (this->config_.mode) {
             case ExecutionMode::AUTO:
             case ExecutionMode::EVENT_DRIVEN:
-                return this->event_driven_wait();
+                return this->event_driven_wait(max_timeout);
             case ExecutionMode::BUSY_WAIT:
                 return this->busy_wait(breaker);
             case ExecutionMode::HIGH_RATE:
                 return this->high_rate_wait(breaker);
             case ExecutionMode::HYBRID:
-                return this->hybrid_wait(breaker);
+                return this->hybrid_wait(breaker, max_timeout);
             case ExecutionMode::RT_EVENT:
                 return this->high_rate_wait(breaker);
         }
@@ -195,7 +198,10 @@ class DarwinLoop final : public Loop {
     }
 
     /// @brief HYBRID: Spin for configured duration, then block with timeout.
-    WakeReason hybrid_wait(const x::breaker::Breaker &breaker) const {
+    WakeReason hybrid_wait(
+        const x::breaker::Breaker &breaker,
+        const x::telem::TimeSpan max_timeout
+    ) const {
         const auto spin_start = std::chrono::steady_clock::now();
         const auto spin_duration = this->config_.spin_duration.chrono();
         struct timespec timeout = {0, 0};
@@ -205,18 +211,22 @@ class DarwinLoop final : public Loop {
             const int n = kevent(this->kqueue_fd_, nullptr, 0, events, 8, &timeout);
             if (n > 0) return this->classify_events(events, n);
         }
-        const auto block_timeout_ns = timing::HYBRID_BLOCK_TIMEOUT.nanoseconds();
-        timeout.tv_sec = 0;
-        timeout.tv_nsec = block_timeout_ns;
+        const auto block_ns = max_timeout.nanoseconds() > 0
+                                ? max_timeout.nanoseconds()
+                                : timing::HYBRID_BLOCK_TIMEOUT.nanoseconds();
+        timeout = ns_to_timespec(block_ns);
         const int n = kevent(this->kqueue_fd_, nullptr, 0, events, 8, &timeout);
         if (n > 0) return this->classify_events(events, n);
         return WakeReason::Timeout;
     }
 
     /// @brief EVENT_DRIVEN: Block on kqueue events with timeout.
-    WakeReason event_driven_wait() const {
+    WakeReason event_driven_wait(const x::telem::TimeSpan max_timeout) const {
         struct kevent events[8];
-        const struct timespec timeout = {0, timing::EVENT_DRIVEN_TIMEOUT.nanoseconds()};
+        const auto timeout_ns = max_timeout.nanoseconds() > 0
+                                  ? max_timeout.nanoseconds()
+                                  : timing::EVENT_DRIVEN_TIMEOUT.nanoseconds();
+        const auto timeout = ns_to_timespec(timeout_ns);
         const int n = kevent(this->kqueue_fd_, nullptr, 0, events, 8, &timeout);
 
         if (n > 0) return this->classify_events(events, n);
@@ -237,6 +247,10 @@ class DarwinLoop final : public Loop {
         return WakeReason::Shutdown;
     }
 
+    static constexpr timespec ns_to_timespec(const int64_t ns) {
+        return {ns / 1'000'000'000, ns % 1'000'000'000};
+    }
+
     Config config_;
     int kqueue_fd_ = -1;
     bool kqueue_timer_enabled_ = false;

arc/cpp/runtime/loop/loop_linux.cpp

@@ -30,7 +30,10 @@ class LinuxLoop final : public Loop {
 
     ~LinuxLoop() override { this->close_fds(); }
 
-    WakeReason wait(x::breaker::Breaker &breaker) override {
+    WakeReason wait(
+        x::breaker::Breaker &breaker,
+        x::telem::TimeSpan max_timeout = x::telem::TimeSpan(0)
+    ) override {
         if (this->epoll_fd_ == -1) return WakeReason::Shutdown;
 
         switch (this->config_.mode) {
@@ -39,12 +42,12 @@ class LinuxLoop final : public Loop {
             case ExecutionMode::HIGH_RATE:
                 return this->high_rate_wait(breaker);
             case ExecutionMode::RT_EVENT:
-                return this->event_driven_wait(true);
+                return this->event_driven_wait(true, max_timeout);
             case ExecutionMode::HYBRID:
-                return this->hybrid_wait(breaker);
+                return this->hybrid_wait(breaker, max_timeout);
             case ExecutionMode::AUTO:
             case ExecutionMode::EVENT_DRIVEN:
-                return this->event_driven_wait(true);
+                return this->event_driven_wait(true, max_timeout);
         }
         return WakeReason::Shutdown;
     }
@@ -210,10 +213,13 @@ class LinuxLoop final : public Loop {
         return WakeReason::Timer;
     }
 
-    WakeReason event_driven_wait(bool blocking) {
+    WakeReason event_driven_wait(bool blocking, const x::telem::TimeSpan max_timeout) {
         struct epoll_event events[2];
-        const int timeout_ms = blocking ? timing::EVENT_DRIVEN_TIMEOUT.milliseconds()
+        const int default_ms = blocking ? timing::EVENT_DRIVEN_TIMEOUT.milliseconds()
                                         : timing::POLL_TIMEOUT.milliseconds();
+        const int timeout_ms = max_timeout.nanoseconds() > 0
+                                 ? static_cast<int>(max_timeout.milliseconds())
+                                 : default_ms;
         const int n = epoll_wait(this->epoll_fd_, events, 2, timeout_ms);
 
         if (n > 0) return this->consume_events(events, n);
@@ -223,7 +229,10 @@ class LinuxLoop final : public Loop {
         return WakeReason::Shutdown;
     }
 
-    WakeReason hybrid_wait(const x::breaker::Breaker &breaker) {
+    WakeReason hybrid_wait(
+        const x::breaker::Breaker &breaker,
+        const x::telem::TimeSpan max_timeout
+    ) {
         const auto spin_start = std::chrono::steady_clock::now();
         const auto spin_duration = std::chrono::nanoseconds(
             this->config_.spin_duration.nanoseconds()
@@ -238,7 +247,9 @@ class LinuxLoop final : public Loop {
             if (n > 0) return this->consume_events(events, n);
         }
 
-        const int timeout_ms = timing::HYBRID_BLOCK_TIMEOUT.milliseconds();
+        const int timeout_ms = max_timeout.nanoseconds() > 0
+                                 ? static_cast<int>(max_timeout.milliseconds())
+                                 : timing::HYBRID_BLOCK_TIMEOUT.milliseconds();
         const int n = epoll_wait(this->epoll_fd_, events, 2, timeout_ms);
         if (n > 0) return this->consume_events(events, n);
         return WakeReason::Timeout;

arc/cpp/runtime/loop/loop_polling.cpp

@@ -46,7 +46,10 @@ class PollingLoop final : public Loop {
 
     ~PollingLoop() override { this->timer_.reset(); }
 
-    WakeReason wait(x::breaker::Breaker &breaker) override {
+    WakeReason wait(
+        x::breaker::Breaker &breaker,
+        x::telem::TimeSpan max_timeout = x::telem::TimeSpan(0)
+    ) override {
         if (!this->started_) return WakeReason::Shutdown;
 
         if (this->config_.interval.nanoseconds() > 0 && this->timer_) {
@@ -56,9 +59,12 @@ class PollingLoop final : public Loop {
             )
                                      .count();
 
-            if (elapsed < this->config_.interval.nanoseconds()) {
-                const int64_t remaining_ns = this->config_.interval.nanoseconds() -
-                                             elapsed;
+            auto interval_ns = this->config_.interval.nanoseconds();
+            if (max_timeout.nanoseconds() > 0)
+                interval_ns = std::min(interval_ns, max_timeout.nanoseconds());
+
+            if (elapsed < interval_ns) {
+                const int64_t remaining_ns = interval_ns - elapsed;
 
                 switch (this->config_.mode) {
                     case ExecutionMode::BUSY_WAIT:
@@ -78,7 +84,9 @@ class PollingLoop final : public Loop {
                 this->last_tick_ = now;
             }
         } else {
-            if (this->config_.mode == ExecutionMode::BUSY_WAIT) {
+            if (max_timeout.nanoseconds() > 0) {
+                std::this_thread::sleep_for(max_timeout.chrono());
+            } else if (this->config_.mode == ExecutionMode::BUSY_WAIT) {
                 std::this_thread::sleep_for(x::telem::MICROSECOND.chrono());
             } else {
                 std::this_thread::sleep_for(timing::HIGH_RATE_POLL_INTERVAL.chrono());

arc/cpp/runtime/loop/loop_test.cpp

@@ -809,3 +809,88 @@ TEST(WakeReasonTest, DistinguishesTimerFromInputWhenBothConfigured) {
 
     breaker.stop();
 }
+
+/// @brief EVENT_DRIVEN with max_timeout should wake after max_timeout.
+TEST(MaxTimeoutTest, EventDriven_WakesAfterMaxTimeout) {
+    Config config;
+    config.mode = ExecutionMode::EVENT_DRIVEN;
+    config.interval = x::telem::TimeSpan(0);
+
+    const auto loop = ASSERT_NIL_P(create(config));
+
+    x::breaker::Breaker breaker;
+
+    const auto sw = x::telem::Stopwatch();
+    const auto reason = loop->wait(breaker, 20 * x::telem::MILLISECOND);
+
+    const auto elapsed = sw.elapsed();
+    EXPECT_GE(elapsed, 15 * x::telem::MILLISECOND);
+    EXPECT_LE(elapsed, test_timing::TIMER_UPPER_BOUND);
+    EXPECT_EQ(reason, WakeReason::Timeout);
+}
+
+/// @brief max_timeout should override a longer configured interval.
+TEST(MaxTimeoutTest, EventDriven_MaxTimeoutOverridesLongerInterval) {
+    Config config;
+    config.mode = ExecutionMode::EVENT_DRIVEN;
+    config.interval = 500 * x::telem::MILLISECOND;
+
+    const auto loop = ASSERT_NIL_P(create(config));
+
+    x::breaker::Breaker breaker;
+
+    const auto sw = x::telem::Stopwatch();
+    loop->wait(breaker, 20 * x::telem::MILLISECOND);
+
+    const auto elapsed = sw.elapsed();
+    EXPECT_GE(elapsed, 15 * x::telem::MILLISECOND);
+    EXPECT_LE(elapsed, test_timing::TIMER_UPPER_BOUND);
+}
+
+/// @brief Input arriving before max_timeout should wake immediately.
+TEST(MaxTimeoutTest, EventDriven_InputWakesBeforeMaxTimeout) {
+    Config config;
+    config.mode = ExecutionMode::EVENT_DRIVEN;
+    config.interval = x::telem::TimeSpan(0);
+
+    const auto loop = ASSERT_NIL_P(create(config));
+
+    auto notifier = x::notify::create();
+    ASSERT_TRUE(loop->watch(*notifier));
+
+    x::breaker::Breaker breaker;
+
+    std::atomic<WakeReason> reason{WakeReason::Shutdown};
+    std::thread waiter([&]() {
+        reason.store(loop->wait(breaker, 500 * x::telem::MILLISECOND));
+    });
+
+    std::this_thread::sleep_for(test_timing::THREAD_STARTUP.chrono());
+    notifier->signal();
+    waiter.join();
+
+    EXPECT_EQ(reason.load(), WakeReason::Input);
+}
+
+/// @brief HYBRID mode with max_timeout should use it for the blocking phase.
+TEST(MaxTimeoutTest, Hybrid_MaxTimeoutConstrainsBlockPhase) {
+    Config config;
+    config.mode = ExecutionMode::HYBRID;
+    config.interval = x::telem::TimeSpan(0);
+    config.spin_duration = 50 * x::telem::MICROSECOND;
+
+    const auto loop = ASSERT_NIL_P(create(config));
+
+    x::breaker::Breaker breaker;
+    breaker.start();
+
+    const auto sw = x::telem::Stopwatch();
+    const auto reason = loop->wait(breaker, 20 * x::telem::MILLISECOND);
+
+    const auto elapsed = sw.elapsed();
+    EXPECT_GE(elapsed, 15 * x::telem::MILLISECOND);
+    EXPECT_LE(elapsed, test_timing::TIMER_UPPER_BOUND);
+    EXPECT_EQ(reason, WakeReason::Timeout);
+
+    breaker.stop();
+}

arc/cpp/runtime/loop/loop_windows.cpp

@@ -34,7 +34,10 @@ class WindowsLoop final : public Loop {
 
     ~WindowsLoop() override { this->close_handles(); }
 
-    WakeReason wait(x::breaker::Breaker &breaker) override {
+    WakeReason wait(
+        x::breaker::Breaker &breaker,
+        x::telem::TimeSpan max_timeout = x::telem::TimeSpan(0)
+    ) override {
         if (this->wake_event_ == NULL) return WakeReason::Shutdown;
 
         switch (this->config_.mode) {
@@ -43,12 +46,12 @@ class WindowsLoop final : public Loop {
             case ExecutionMode::HIGH_RATE:
                 return this->high_rate_wait(breaker);
             case ExecutionMode::RT_EVENT:
-                return this->event_driven_wait(false);
+                return this->event_driven_wait(false, max_timeout);
             case ExecutionMode::HYBRID:
-                return this->hybrid_wait(breaker);
+                return this->hybrid_wait(breaker, max_timeout);
             case ExecutionMode::AUTO:
             case ExecutionMode::EVENT_DRIVEN:
-                return this->event_driven_wait(true);
+                return this->event_driven_wait(true, max_timeout);
         }
         return WakeReason::Shutdown;
     }
@@ -178,18 +181,21 @@ class WindowsLoop final : public Loop {
         return WakeReason::Timer;
     }
 
-    WakeReason event_driven_wait(bool blocking) {
+    WakeReason event_driven_wait(bool blocking, const x::telem::TimeSpan max_timeout) {
         HANDLE handles[3];
         const DWORD count = this->build_handles(handles);
         if (count == 0) return WakeReason::Shutdown;
 
-        const DWORD timeout_ms = blocking
+        const DWORD default_ms = blocking
                                    ? static_cast<DWORD>(
                                          timing::EVENT_DRIVEN_TIMEOUT.milliseconds()
                                      )
                                    : static_cast<DWORD>(
                                          timing::HYBRID_BLOCK_TIMEOUT.milliseconds()
                                      );
+        const DWORD timeout_ms = max_timeout.nanoseconds() > 0
+                                   ? static_cast<DWORD>(max_timeout.milliseconds())
+                                   : default_ms;
 
         const DWORD result = WaitForMultipleObjects(count, handles, FALSE, timeout_ms);
         if (result == WAIT_TIMEOUT) return WakeReason::Timeout;
@@ -200,7 +206,8 @@ class WindowsLoop final : public Loop {
         return this->classify_result(result, handles);
     }
 
-    WakeReason hybrid_wait(x::breaker::Breaker &breaker) {
+    WakeReason
+    hybrid_wait(x::breaker::Breaker &breaker, const x::telem::TimeSpan max_timeout) {
         HANDLE handles[3];
         const DWORD count = this->build_handles(handles);
         if (count == 0) return WakeReason::Shutdown;
@@ -218,9 +225,11 @@ class WindowsLoop final : public Loop {
                 return this->classify_result(result, handles);
         }
 
-        const DWORD timeout_ms = static_cast<DWORD>(
-            timing::HYBRID_BLOCK_TIMEOUT.milliseconds()
-        );
+        const DWORD timeout_ms = max_timeout.nanoseconds() > 0
+                                   ? static_cast<DWORD>(max_timeout.milliseconds())
+                                   : static_cast<DWORD>(
+                                         timing::HYBRID_BLOCK_TIMEOUT.milliseconds()
+                                     );
         const DWORD result = WaitForMultipleObjects(count, handles, FALSE, timeout_ms);
         if (result == WAIT_TIMEOUT) return WakeReason::Timeout;
         if (result < WAIT_OBJECT_0 + count)

arc/cpp/runtime/node/node.h

@@ -32,6 +32,8 @@ struct Context {
     /// Time-based nodes should only fire when reason is TimerTick.
     RunReason reason;
     std::function<void(const std::string &output_param)> mark_changed;
+    std::function<void()> mark_self_changed;
+    std::function<void(x::telem::TimeSpan)> set_deadline;
     std::function<void(const x::errors::Error &)> report_error;
     std::function<void()> activate_stage;
 };