Add action scheduling policies (#334)

* Add "update" to SPACING_VIOLATION_POLICIES in LFValidator

* Add DelayedActionDuplicate.lf test checking all action policies

* Add ActionPolicy enumeration

* Add ActionPolicy parameter to macros_internal.h

* Add ActionPolicy to UcActionGenerator

* Add inline functions for event comparison

* Update ActionLib constructors to use "defer" parameter

* Add get_event_queue method to Scheduler interface

* Add action scheduling policies and remove recursive heapify from event queue

* Apply code formatting

* Add more event queue tests

* Free previous payload with "replace" policy

* Update Sender action constructor to use 'defer' policy

* Update src/queues.c

Co-authored-by: Edward A. Lee <eal@berkeley.edu>

* Add javadoc to event.h

* Add update policy tests

* Fix update policy

* Add docs

* Apply code formatting

* Refactor comments

* Refactor action update policies: move event cancel/replace from Action into Scheduler

* Add unit tests for defer, drop, replace, and update action policies

* Apply code formatting

* Add minSpacing validation for "defer" policy in LFValidator

* Refactor event comparison functions to explicitly return boolean values

* Refactor type assignment in FederatedInputConnection_ctor for implicit conversion 'bool' -> 'int'

* Add caching for Xtext ANTLR generator in build action

* Revert changes: implicit conversion 'bool' -> 'int'

* Refactor action policies to use ACTION_POLICY constants and update error return values for consistency

* Refactor FederatedInputConnection_ctor to use explicit if-else for connection type assignment

* Fix clang-tidy warnings

* Remove clang-tidy installation from memory workflow dependencies

* Fix memory report crash when base and PR branches have different tests

* Add "update" to SPACING_VIOLATION_POLICIES in LFValidator

* Add DelayedActionDuplicate.lf test checking all action policies

* Add ActionPolicy enumeration

* Add ActionPolicy parameter to macros_internal.h

* Add ActionPolicy to UcActionGenerator

* Add inline functions for event comparison

* Update ActionLib constructors to use "defer" parameter

* Add get_event_queue method to Scheduler interface

* Add action scheduling policies and remove recursive heapify from event queue

* Apply code formatting

* Add more event queue tests

* Free previous payload with "replace" policy

* Update Sender action constructor to use 'defer' policy

* Update src/queues.c

Co-authored-by: Edward A. Lee <eal@berkeley.edu>

* Add javadoc to event.h

* Add update policy tests

* Fix update policy

* Add docs

* Apply code formatting

* Refactor comments

* Refactor action update policies: move event cancel/replace from Action into Scheduler

* Add unit tests for defer, drop, replace, and update action policies

* Apply code formatting

* Add minSpacing validation for "defer" policy in LFValidator

* Refactor event comparison functions to explicitly return boolean values

* Add caching for Xtext ANTLR generator in build action

* Refactor action policies to use ACTION_POLICY constants and update error return values for consistency

* Fix clang-tidy warnings

* Remove clang-tidy installation from memory workflow dependencies

* Fix memory report crash when base and PR branches have different tests

* Fix event matching logic in find_matching_event_idx function

* Set maxNumberOfPendingEvents default value to SIZE_MAX for void actions

* Update maxNumPendingEvents default value to 20 and add VoidAction test case

* Enforce action policy when the payload_buffer is full

* Restore max number of pending events to 1 when not specified

* Fix clang-tidy errors

* Add unit tests for action buffer policies: DEFER, DROP, REPLACE, and UPDATE

* Refactor action minSpacing validation and generation with `defer` policy

* Refactor target properties with annotations

* Apply code formatting

* Fix max_pending_events annotation

* Revert changes when policy is defer and minSpacing is 0

* Fix action scheduling when minSpacing is equals to 0 and defer policy

* Fix minSpacing reference in reactor constructor code generation

* Add imports for Action and ActionOrigin in LFValidator

* Apply code formatting

---------

Co-authored-by: Edward A. Lee <eal@berkeley.edu>

Author: sbgaia

.github/actions/lingua-franca/action.yml

@@ -8,5 +8,10 @@ runs:
       with:
         distribution: temurin
         java-version: 17
+    - name: Cache Xtext ANTLR generator
+      uses: actions/cache@v4
+      with:
+        path: lfc/core/.antlr-generator-3.2.0-patch.jar
+        key: antlr-generator-3.2.0-patch
     - name: Gradle Build Action
       uses: gradle/gradle-build-action@v2.8.0

.github/workflows/memory.yml

@@ -20,7 +20,7 @@ jobs:
       - name: Install deps
         run: |
           sudo apt-get install lcov pipx -y
-          sudo pipx install clang-format clang-tidy
+          sudo pipx install clang-format
       - name: Generate memory usage report (base branch)
         id: base
         uses: ./.github/actions/memory-report

examples/zephyr/basic_federated/federated_sender/src/sender.c

@@ -23,7 +23,7 @@ typedef struct {
 } msg_t;
 
 LF_DEFINE_ACTION_STRUCT(Sender, act, PhysicalAction, 1, 0, 0, 10, bool);
-LF_DEFINE_ACTION_CTOR(Sender, act, PhysicalAction, 1, 0, 0, 10, bool);
+LF_DEFINE_ACTION_CTOR(Sender, act, PhysicalAction, ACTION_POLICY_DEFER, 1, 0, 0, 10, bool);
 LF_DEFINE_REACTION_STRUCT(Sender, r, 1);
 LF_DEFINE_REACTION_CTOR(Sender, r, 0, NULL, NULL);
 

include/reactor-uc/action.h

@@ -12,9 +12,30 @@ typedef struct LogicalAction LogicalAction;
 
 typedef enum { LOGICAL_ACTION, PHYSICAL_ACTION } ActionType;
 
+/** @brief Policy for handling scheduled events that violate the specified minimum
+ *  interarrival time, or that exceed the action's pending-event capacity.
+ *
+ * The same policy is applied in both situations:
+ *
+ * - `ACTION_POLICY_DEFER` (default): adjust the tag so the minimum interarrival time is
+ *   satisfied. If the buffer is full, the new event is dropped (no slot is available to
+ *   defer into) and `LF_VALUE_BUFFER_FULL` is returned.
+ * - `ACTION_POLICY_DROP`: drop the event. Returns `LF_OK` for min_spacing drops, or
+ *   `LF_VALUE_BUFFER_FULL` when the buffer was full.
+ * - `ACTION_POLICY_REPLACE`: attempt to replace the payload of the preceding event. If
+ *   the preceding event has already been popped off the event queue, falls back to
+ *   `ACTION_POLICY_DEFER` (or to dropping with `LF_VALUE_BUFFER_FULL` when the buffer is
+ *   full). Naturally handles buffer-full because it does not require a new slot.
+ * - `ACTION_POLICY_UPDATE`: cancel the preceding event (if still queued) and schedule
+ *   the new one in its place. Naturally handles buffer-full because cancelling frees a
+ *   slot.
+ */
+typedef enum { ACTION_POLICY_DEFER, ACTION_POLICY_DROP, ACTION_POLICY_REPLACE, ACTION_POLICY_UPDATE } ActionPolicy;
+
 struct Action {
   Trigger super;
   ActionType type;
+  ActionPolicy policy;        // Policy for handling events that violate min_spacing
   interval_t min_offset;      // The minimum offset from the current time that an event can be scheduled on this action.
   interval_t min_spacing;     // The minimum spacing between two events scheduled on this action.
   instant_t last_event_time;  // Logical time of most recent event scheduled on this action.
@@ -36,27 +57,27 @@ struct Action {
   lf_ret_t (*schedule)(Action* self, interval_t offset, const void* value);
 };
 
-void Action_ctor(Action* self, ActionType type, interval_t min_offset, interval_t min_spacing, Reactor* parent,
-                 Reaction** sources, size_t sources_size, Reaction** effects, size_t effects_size, Reaction** observers,
-                 size_t observers_size, void* value_ptr, size_t value_size, void* payload_buf, bool* payload_used_buf,
-                 size_t event_bound);
+void Action_ctor(Action* self, ActionType type, ActionPolicy policy, interval_t min_offset, interval_t min_spacing,
+                 Reactor* parent, Reaction** sources, size_t sources_size, Reaction** effects, size_t effects_size,
+                 Reaction** observers, size_t observers_size, void* value_ptr, size_t value_size, void* payload_buf,
+                 bool* payload_used_buf, size_t event_bound);
 
 struct LogicalAction {
   Action super;
 };
 
-void LogicalAction_ctor(LogicalAction* self, interval_t min_offset, interval_t min_spacing, Reactor* parent,
-                        Reaction** sources, size_t sources_size, Reaction** effects, size_t effects_size,
-                        Reaction** observers, size_t observers_size, void* value_ptr, size_t value_size,
-                        void* payload_buf, bool* payload_used_buf, size_t event_bound);
+void LogicalAction_ctor(LogicalAction* self, ActionPolicy policy, interval_t min_offset, interval_t min_spacing,
+                        Reactor* parent, Reaction** sources, size_t sources_size, Reaction** effects,
+                        size_t effects_size, Reaction** observers, size_t observers_size, void* value_ptr,
+                        size_t value_size, void* payload_buf, bool* payload_used_buf, size_t event_bound);
 
 struct PhysicalAction {
   Action super;
   MUTEX_T mutex;
 };
 
-void PhysicalAction_ctor(PhysicalAction* self, interval_t min_offset, interval_t min_spacing, Reactor* parent,
-                         Reaction** sources, size_t sources_size, Reaction** effects, size_t effects_size,
-                         Reaction** observers, size_t observers_size, void* value_ptr, size_t value_size,
-                         void* payload_buf, bool* payload_used_buf, size_t event_bound);
+void PhysicalAction_ctor(PhysicalAction* self, ActionPolicy policy, interval_t min_offset, interval_t min_spacing,
+                         Reactor* parent, Reaction** sources, size_t sources_size, Reaction** effects,
+                         size_t effects_size, Reaction** observers, size_t observers_size, void* value_ptr,
+                         size_t value_size, void* payload_buf, bool* payload_used_buf, size_t event_bound);
 #endif

include/reactor-uc/error.h

@@ -20,7 +20,8 @@ typedef enum {
   LF_EVENT_QUEUE_FULL,
   LF_VALUE_BUFFER_FULL,
   LF_NETWORK_CHANNEL_RETRY,
-  LF_NETWORK_CHANNEL_EMPTY
+  LF_NETWORK_CHANNEL_EMPTY,
+  LF_EVENT_NOT_FOUND
 } lf_ret_t;
 
 /**

include/reactor-uc/event.h

@@ -74,4 +74,71 @@ struct SystemEventHandler {
 void EventPayloadPool_ctor(EventPayloadPool* self, char* buffer, bool* used, size_t element_size, size_t capacity,
                            size_t reserved);
 
+/**
+ * @brief Get the tag of an arbitrary event.
+ * @param arbitrary_event Pointer to the arbitrary event.
+ * @return The tag of the event.
+ */
+static inline tag_t get_tag(ArbitraryEvent* arbitrary_event) { return arbitrary_event->event.super.tag; }
+
+/**
+ * @brief Compare the tags of two arbitrary events.
+ * @param evt1 Pointer to the first arbitrary event.
+ * @param evt2 Pointer to the second arbitrary event.
+ * @return true if the tags are the same, false otherwise.
+ */
+static inline bool events_same_tag(ArbitraryEvent* evt1, ArbitraryEvent* evt2) {
+  return lf_tag_compare(get_tag(evt1), get_tag(evt2)) == 0;
+}
+
+/**
+ * @brief Compare the triggers of two arbitrary events.
+ * @param evt1 Pointer to the first arbitrary event.
+ * @param evt2 Pointer to the second arbitrary event.
+ * @return true if the triggers of both events are the same, false otherwise.
+ */
+static inline bool events_same_trigger(ArbitraryEvent* evt1, ArbitraryEvent* evt2) {
+  if (evt1->event.super.type != EVENT || evt2->event.super.type != EVENT) {
+    return false;
+  }
+  Event* event1 = &evt1->event;
+  Event* event2 = &evt2->event;
+  return event1->trigger == event2->trigger;
+}
+
+/**
+ * @brief Compare the handlers of two arbitrary events.
+ * @param evt1 Pointer to the first arbitrary event.
+ * @param evt2 Pointer to the second arbitrary event.
+ * @return true if the handlers of both events are the same, false otherwise.
+ */
+static inline bool events_same_handler(ArbitraryEvent* evt1, ArbitraryEvent* evt2) {
+  if (evt1->event.super.type != SYSTEM_EVENT || evt2->event.super.type != SYSTEM_EVENT) {
+    return false;
+  }
+  SystemEvent* event1 = &evt1->system_event;
+  SystemEvent* event2 = &evt2->system_event;
+  return event1->handler == event2->handler;
+}
+
+/**
+ * @brief Compare the tags and triggers of two arbitrary events.
+ * @param evt1 Pointer to the first arbitrary event.
+ * @param evt2 Pointer to the second arbitrary event.
+ * @return true if the tags and triggers of both events are the same, false otherwise.
+ */
+static inline bool events_same_tag_and_trigger(ArbitraryEvent* evt1, ArbitraryEvent* evt2) {
+  return (events_same_tag(evt1, evt2) && events_same_trigger(evt1, evt2)) != 0;
+}
+
+/**
+ * @brief Compare the tags and handlers of two arbitrary events.
+ * @param evt1 Pointer to the first arbitrary event.
+ * @param evt2 Pointer to the second arbitrary event.
+ * @return true if the tags and handlers of both events are the same, false otherwise.
+ */
+static inline bool events_same_tag_and_handler(ArbitraryEvent* evt1, ArbitraryEvent* evt2) {
+  return (events_same_tag(evt1, evt2) && events_same_handler(evt1, evt2)) != 0;
+}
+
 #endif

include/reactor-uc/macros_internal.h

@@ -312,21 +312,22 @@
     Reaction* observers[(ObserverSize)];                                                                               \
   } ReactorName##_##ActionName;
 
-#define LF_DEFINE_ACTION_CTOR(ReactorName, ActionName, ActionType, EffectSize, SourceSize, ObserverSize,               \
+#define LF_DEFINE_ACTION_CTOR(ReactorName, ActionName, ActionType, ActionPolicy, EffectSize, SourceSize, ObserverSize, \
                               MaxPendingEvents, BufferType)                                                            \
   void ReactorName##_##ActionName##_ctor(ReactorName##_##ActionName* self, Reactor* parent, interval_t min_delay,      \
                                          interval_t min_spacing) {                                                     \
-    ActionType##_ctor(&self->super, min_delay, min_spacing, parent, self->sources, (SourceSize), self->effects,        \
-                      (EffectSize), self->observers, ObserverSize, &self->value, sizeof(self->value),                  \
+    ActionType##_ctor(&self->super, ActionPolicy, min_delay, min_spacing, parent, self->sources, (SourceSize),         \
+                      self->effects, (EffectSize), self->observers, ObserverSize, &self->value, sizeof(self->value),   \
                       (void*)&self->payload_buf, self->payload_used_buf, (MaxPendingEvents));                          \
   }
 
-#define LF_DEFINE_ACTION_CTOR_VOID(ReactorName, ActionName, ActionType, EffectSize, SourceSize, ObserverSize,          \
-                                   MaxPendingEvents)                                                                   \
+#define LF_DEFINE_ACTION_CTOR_VOID(ReactorName, ActionName, ActionType, ActionPolicy, EffectSize, SourceSize,          \
+                                   ObserverSize, MaxPendingEvents)                                                     \
   void ReactorName##_##ActionName##_ctor(ReactorName##_##ActionName* self, Reactor* parent, interval_t min_delay,      \
                                          interval_t min_spacing) {                                                     \
-    ActionType##_ctor(&self->super, min_delay, min_spacing, parent, self->sources, (SourceSize), self->effects,        \
-                      (EffectSize), self->observers, ObserverSize, NULL, 0, NULL, NULL, (MaxPendingEvents));           \
+    ActionType##_ctor(&self->super, ActionPolicy, min_delay, min_spacing, parent, self->sources, (SourceSize),         \
+                      self->effects, (EffectSize), self->observers, ObserverSize, NULL, 0, NULL, NULL,                 \
+                      (MaxPendingEvents));                                                                             \
   }
 
 #define LF_ACTION_INSTANCE(ReactorName, ActionName) ReactorName##_##ActionName ActionName

include/reactor-uc/queues.h

@@ -11,19 +11,49 @@
 typedef struct EventQueue EventQueue;
 typedef struct ReactionQueue ReactionQueue;
 
+/**
+ * @brief Min-heap priority event queue ordered by tag.
+ */
 struct EventQueue {
+  /** @brief Return the tag of the earliest event in the queue, or FOREVER_TAG
+   * if empty.
+   */
   tag_t (*next_tag)(EventQueue* self);
+  /**
+   * @brief Insert an event into the queue. Returns LF_EVENT_QUEUE_FULL if the
+   * queue is full.
+   */
   lf_ret_t (*insert)(EventQueue* self, AbstractEvent* event);
+  /** @brief Remove and return the earliest event in the queue. Returns
+   * LF_EVENT_QUEUE_EMPTY if the queue is empty.
+   */
   lf_ret_t (*pop)(EventQueue* self, AbstractEvent* event);
+  /** @brief Return true if the queue contains no events. */
   bool (*empty)(EventQueue* self);
-  void (*heapify_locked)(EventQueue* self, size_t idx);
+  /** @brief Restore the heap invariant for the entire queue. Should be called after bulk
+   * insertion of events. */
+  void (*build_heap)(EventQueue* self);
+  /** @brief Restore the heap invariant downward from @p idx. */
+  void (*heapify)(EventQueue* self, size_t idx);
+  /** @brief Find an event with the same tag and trigger as @p event, or NULL if not found. */
+  ArbitraryEvent* (*find_equal_same_tag)(EventQueue* self, AbstractEvent* event);
+  /** @brief Remove the event equal to @p event from the queue. Returns
+   * LF_EVENT_NOT_FOUND if no such event exists, LF_OK otherwise.
+   */
+  lf_ret_t (*remove)(EventQueue* self, AbstractEvent* event);
 
-  size_t size;
-  size_t capacity;
-  ArbitraryEvent* array;
-  MUTEX_T mutex;
+  size_t size;           /**< @brief Current number of events in the queue. */
+  size_t capacity;       /**< @brief Maximum number of events the queue can hold. */
+  ArbitraryEvent* array; /**< @brief Backing array of the event queue. */
+  MUTEX_T mutex;         /**< @brief Mutex protecting concurrent access. */
 };
 
+/**
+ * @brief Initialize an EventQueue.
+ * @param self     The EventQueue to initialize.
+ * @param array    Backing array of at least @p capacity ArbitraryEvent elements.
+ * @param capacity Maximum number of events the queue can hold.
+ */
 void EventQueue_ctor(EventQueue* self, ArbitraryEvent* array, size_t capacity);
 
 struct ReactionQueue {

include/reactor-uc/scheduler.h

@@ -57,6 +57,18 @@ struct Scheduler {
    * queue.
    */
   void (*prepare_timestep)(Scheduler* self, tag_t tag);
+
+  /**
+   * @brief Cancel a pending event for a `trigger` at `event_time`. Returns
+   * LF_EVENT_NOT_FOUND if no matching event exists.
+   */
+  lf_ret_t (*cancel_event)(Scheduler* self, Trigger* trigger, instant_t event_time);
+
+  /**
+   * @brief Replace the payload of a pending event for a `trigger` at
+   * `event_time`. Returns LF_EVENT_NOT_FOUND if no matching event exists.
+   */
+  lf_ret_t (*replace_event_payload)(Scheduler* self, Trigger* trigger, instant_t event_time, const void* new_value);
 };
 
 Scheduler* Scheduler_new(Environment* env, EventQueue* event_queue, EventQueue* system_event_queue,

lfc/core/src/main/java/org/lflang/validation/LFValidator.java

@@ -50,6 +50,8 @@
 import org.lflang.InferredType;
 import org.lflang.ModelInfo;
 import org.lflang.ast.ASTUtils;
+import org.lflang.lf.Action;
+import org.lflang.lf.ActionOrigin;
 import org.lflang.lf.Attribute;
 import org.lflang.lf.BracedListExpression;
 import org.lflang.lf.BuiltinTrigger;
@@ -98,6 +100,27 @@ public class LFValidator extends BaseLFValidator {
   // NOTE: please list methods in alphabetical order, and follow a naming convention checkClass,
   // where Class is the AST class.
 
+  @Check(CheckType.FAST)
+  public void checkAction(Action action) {
+    checkName(action.getName(), Literals.VARIABLE__NAME);
+    if (action.getOrigin() == ActionOrigin.NONE) {
+      error(
+          "Action must have modifier {@code logical} or {@code physical}.",
+          Literals.ACTION__ORIGIN);
+    }
+    if (action.getPolicy() != null && !SPACING_VIOLATION_POLICIES.contains(action.getPolicy())) {
+      error(
+          "Unrecognized spacing violation policy: "
+              + action.getPolicy()
+              + ". Available policies are: "
+              + String.join(", ", SPACING_VIOLATION_POLICIES)
+              + ".",
+          Literals.ACTION__POLICY);
+    }
+    checkExpressionIsTime(action.getMinDelay(), Literals.ACTION__MIN_DELAY);
+    checkExpressionIsTime(action.getMinSpacing(), Literals.ACTION__MIN_SPACING);
+  }
+
   public void checkReactorName(String name) throws IOException {
     // Check for illegal names.
     checkName(name, Literals.REACTOR_DECL__NAME);
@@ -874,7 +897,8 @@ private boolean sameType(Type type1, Type type2) {
       "Reserved words in the target language are not allowed for objects (inputs, outputs, actions,"
           + " timers, parameters, state, reactor definitions, and reactor instantiation): ";
 
-  private static List<String> SPACING_VIOLATION_POLICIES = List.of("defer", "drop", "replace");
+  private static List<String> SPACING_VIOLATION_POLICIES =
+      List.of("defer", "drop", "replace", "update");
 
   private static String UNDERSCORE_MESSAGE =
       "Names of objects (inputs, outputs, actions, timers, parameters, "