Improve and reduce critical sections (#276)

* Add get_lag() and improve docs

* Remove platform.h from doxy

* Reduce and improve the critical sections

* Smaller fixes

* First iteration of moving to a mutex type

* Add mutex to queues

* Format

* Add locks to the payload pools

* Remove critical sections

* Port over Zephyr, RIOT and Pico

* Format

* Fix aducm355

* aducm355 fixes

* Fix aducm355, flexpret and pico

* Formatting

* Remove unused field of ClockSynchronization

* Fix platform implementations

* Small fixes

* Formatting

* Trigger CI

Author: erlingrj

examples/flexpret/buildAll.sh

@@ -1,5 +1,5 @@
-# set -e
-# ${REACTOR_UC_PATH}/lfc/bin/lfc-dev src/Smoke.lf
-# cmake -Bbuild
-# make -C build
-# bin/fp-smoke
+set -e
+${REACTOR_UC_PATH}/lfc/bin/lfc-dev src/Smoke.lf
+cmake -Bbuild
+make -C build
+bin/fp-smoke

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

@@ -22,8 +22,8 @@ typedef struct {
   char msg[32];
 } msg_t;
 
-LF_DEFINE_ACTION_STRUCT(Sender, act, PHYSICAL_ACTION, 1, 0, 0, 10, bool);
-LF_DEFINE_ACTION_CTOR(Sender, act, PHYSICAL_ACTION, 1, 0, 0, 10, bool);
+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_REACTION_STRUCT(Sender, r, 1);
 LF_DEFINE_REACTION_CTOR(Sender, r, 0, NULL, NULL);
 

include/reactor-uc/action.h

@@ -4,10 +4,11 @@
 #include "reactor-uc/error.h"
 #include "reactor-uc/reaction.h"
 #include "reactor-uc/trigger.h"
+#include "reactor-uc/platform.h"
 
 typedef struct Action Action;
-typedef struct LogicalAction LogicalAction;
 typedef struct PhysicalAction PhysicalAction;
+typedef struct LogicalAction LogicalAction;
 
 typedef enum { LOGICAL_ACTION, PHYSICAL_ACTION } ActionType;
 
@@ -40,4 +41,22 @@ void Action_ctor(Action *self, ActionType type, interval_t min_offset, interval_
                  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);
+
+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);
 #endif

include/reactor-uc/clock_synchronization.h

@@ -55,8 +55,6 @@ struct ClockSynchronization {
   interval_t period;              // The period between sync request messages are sent to the neighbor master.
   ClockSyncTimestamps timestamps; // The timestamps used to compute clock offset.
   ClockServo servo;               // The PID controller
-  // TODO: Is this message here needed? Why cant we use the one on the bundle?
-  FederateMessage msg; // A FederateMessage used for transmitting sync request and follow-up messages.
   void (*handle_message_callback)(ClockSynchronization *self, const ClockSyncMessage *msg, size_t bundle_idx);
 };
 

include/reactor-uc/environment.h

@@ -10,18 +10,18 @@
 
 #include "reactor-uc/builtin_triggers.h"
 #include "reactor-uc/error.h"
-#include "reactor-uc/platform.h"
 #include "reactor-uc/reactor.h"
 #include "reactor-uc/scheduler.h"
 #include "reactor-uc/queues.h"
 
+typedef struct Platform Platform;
 typedef struct Environment Environment;
 extern Environment *_lf_environment; // NOLINT
 
 struct Environment {
-  Reactor *main;         // The top-level reactor of the program.
-  Scheduler *scheduler;  // The scheduler in charge of executing the reactions.
-  Platform *platform;    // The platform that provides the physical time and sleep functions.
+  Reactor *main;        // The top-level reactor of the program.
+  Scheduler *scheduler; // The scheduler in charge of executing the reactions.
+  Platform *platform;
   bool has_async_events; // Whether the program has multiple execution contexts and can receive async events and thus
                          // need critical sections.
   bool fast_mode; // Whether the program is executing in fast mode where we do not wait for physical time to elapse
@@ -50,7 +50,7 @@ struct Environment {
    * This function must be called from a critical section.
    *
    */
-  lf_ret_t (*wait_until_locked)(Environment *self, instant_t wakeup_time);
+  lf_ret_t (*wait_until)(Environment *self, instant_t wakeup_time);
 
   /**
    * @brief Sleep for a duration.
@@ -113,18 +113,6 @@ struct Environment {
    */
   interval_t (*get_lag)(Environment *self);
 
-  /**
-   * @private
-   * @brief Enter a critical section. Either by disabling interrupts, using a mutex or both.
-   */
-  void (*enter_critical_section)(Environment *self);
-
-  /**
-   * @private
-   * @brief Leave a critical section. Either by enabling interrupts, releasing a mutex or both.
-   */
-  void (*leave_critical_section)(Environment *self);
-
   /**
    * @brief Request the termination of the program.
    * @param self The environment.

include/reactor-uc/event.h

@@ -3,6 +3,7 @@
 
 #include "reactor-uc/error.h"
 #include "reactor-uc/tag.h"
+#include "reactor-uc/platform.h"
 #include <stdbool.h>
 
 #define EVENT_INIT(Tag, Trigger, Payload)                                                                              \
@@ -55,6 +56,8 @@ struct EventPayloadPool {
   /** Number of payloads reserved to be allocated through `allocate_reserved` */
   size_t reserved;
 
+  MUTEX_T mutex;
+
   /** Allocate a payload from the pool. */
   lf_ret_t (*allocate)(EventPayloadPool *self, void **payload);
   lf_ret_t (*allocate_reserved)(EventPayloadPool *self, void **payload);

include/reactor-uc/federated.h

@@ -94,6 +94,7 @@ void FederatedOutputConnection_ctor(FederatedOutputConnection *self, Reactor *pa
  */
 struct FederatedInputConnection {
   Connection super;
+  MUTEX_T mutex;
   interval_t delay;     // The amount of delay on this connection
   ConnectionType type;  // Whether this is a logical or physical connection
   tag_t last_known_tag; // The latest tag this input is known at.

include/reactor-uc/macros_internal.h

@@ -282,7 +282,7 @@
 #define LF_DEFINE_ACTION_STRUCT(ReactorName, ActionName, ActionType, EffectSize, SourceSize, ObserverSize,             \
                                 MaxPendingEvents, BufferType)                                                          \
   typedef struct {                                                                                                     \
-    Action super;                                                                                                      \
+    ActionType super;                                                                                                  \
     BufferType value;                                                                                                  \
     BufferType payload_buf[(MaxPendingEvents)];                                                                        \
     bool payload_used_buf[(MaxPendingEvents)];                                                                         \
@@ -294,7 +294,7 @@
 #define LF_DEFINE_ACTION_STRUCT_ARRAY(ReactorName, ActionName, ActionType, EffectSize, SourceSize, ObserverSize,       \
                                       MaxPendingEvents, BufferType, ArrayLength)                                       \
   typedef struct {                                                                                                     \
-    Action super;                                                                                                      \
+    ActionType super;                                                                                                  \
     BufferType value[(ArrayLength)];                                                                                   \
     BufferType payload_buf[(ArrayLength)][(MaxPendingEvents)];                                                         \
     bool payload_used_buf[(MaxPendingEvents)];                                                                         \
@@ -306,7 +306,7 @@
 #define LF_DEFINE_ACTION_STRUCT_VOID(ReactorName, ActionName, ActionType, EffectSize, SourceSize, ObserverSize,        \
                                      MaxPendingEvents)                                                                 \
   typedef struct {                                                                                                     \
-    Action super;                                                                                                      \
+    ActionType super;                                                                                                  \
     Reaction *sources[(SourceSize)];                                                                                   \
     Reaction *effects[(EffectSize)];                                                                                   \
     Reaction *observers[(ObserverSize)];                                                                               \
@@ -316,17 +316,17 @@
                               MaxPendingEvents, BufferType)                                                            \
   void ReactorName##_##ActionName##_ctor(ReactorName##_##ActionName *self, Reactor *parent, interval_t min_delay,      \
                                          interval_t min_spacing) {                                                     \
-    Action_ctor(&self->super, ActionType, 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));                               \
+    ActionType##_ctor(&self->super, 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)                                                                   \
   void ReactorName##_##ActionName##_ctor(ReactorName##_##ActionName *self, Reactor *parent, interval_t min_delay,      \
                                          interval_t min_spacing) {                                                     \
-    Action_ctor(&self->super, ActionType, min_delay, min_spacing, parent, self->sources, (SourceSize), self->effects,  \
-                (EffectSize), self->observers, ObserverSize, NULL, 0, NULL, NULL, (MaxPendingEvents));                 \
+    ActionType##_ctor(&self->super, 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/physical_clock.h

@@ -1,6 +1,7 @@
 #ifndef REACTOR_UC_PHYSICAL_CLOCK_H
 #define REACTOR_UC_PHYSICAL_CLOCK_H
 
+#include "reactor-uc/platform.h"
 #include "reactor-uc/tag.h"
 #include "reactor-uc/error.h"
 #include <stdbool.h>
@@ -10,6 +11,7 @@ typedef struct Environment Environment;
 
 struct PhysicalClock {
   Environment *env;
+  MUTEX_T mutex;
   interval_t offset;             // Constant offset applied to each reading of the HW clock
   instant_t adjustment_epoch_hw; // The time at which the frequency adjustment should by applied from.
   double adjustment;             // The frequency adjustment factor.

include/reactor-uc/platform.h

@@ -6,14 +6,24 @@
 #include <stdarg.h>
 
 typedef struct Platform Platform;
+typedef struct Mutex Mutex;
+
+/**
+ * @brief Each supported platform must provide a mutex, this is used by the runtime
+ * to protect shared data structures.
+ *
+ * The mutex can be implemented using the underlying OS, or just disabling/enabling interrupts.
+ *
+ */
+struct Mutex {
+  void (*lock)(Mutex *super);
+  void (*unlock)(Mutex *super);
+};
+
+/** Construct a Mutex*/
+void Mutex_ctor(Mutex *super);
 
 struct Platform {
-  /**
-   * @brief Perform any platform initialization such as initializing semaphores,
-   * configuring clocks, etc. Called once after construction.
-   * @param super The platform
-   */
-  lf_ret_t (*initialize)(Platform *super);
   /**
    * @brief Return the current physical time in nanoseconds.
    */
@@ -32,30 +42,43 @@ struct Platform {
    * @brief Put the system to sleep until the wakeup time or until an
    * asynchronous event occurs. Must be called from a critical section.
    */
-  lf_ret_t (*wait_until_interruptible_locked)(Platform *super, instant_t wakeup_time);
+  lf_ret_t (*wait_until_interruptible)(Platform *super, instant_t wakeup_time);
 
   /**
    * @brief Signal the occurrence of an asynchronous event. This should wake
    * up the platform if it is sleeping on `wait_until_interruptible_locked`.
    */
-  void (*new_async_event)(Platform *super);
-
-  /**
-   * @brief Enter and leave a critical section. This must support nested critical sections. Critical sections
-   * can be implemented by disabling interrupts or acquiring a mutex.
-   */
-  void (*enter_critical_section)(Platform *super);
-  void (*leave_critical_section)(Platform *super);
+  void (*notify)(Platform *super);
 };
 
-// Return a pointer to a Platform object. Must be implemented for each
-// target platform.
-Platform *Platform_new(void);
-
 // Construct a Platform object. Must be implemented for each target platform.
 void Platform_ctor(Platform *super);
 
+// Returns a pointer to the platform.P
+Platform *Platform_new();
+
 // Allow each platform to provide its own implementation for printing.
 void Platform_vprintf(const char *fmt, va_list args);
 
+#if defined(PLATFORM_POSIX)
+#include "reactor-uc/platform/posix/posix.h"
+#elif defined(PLATFORM_RIOT)
+#include "reactor-uc/platform/riot/riot.h"
+#elif defined(PLATFORM_ZEPHYR)
+#include "reactor-uc/platform/zephyr/zephyr.h"
+#elif defined(PLATFORM_FLEXPRET)
+#include "reactor-uc/platform/flexpret/flexpret.h"
+#elif defined(PLATFORM_PICO)
+#include "reactor-uc/platform/pico/pico.h"
+#elif defined(PLATFORM_PATMOS)
+#include "reactor-uc/platform/patmos/patmos.h"
+#elif defined(PLATFORM_ADUCM355)
+#include "reactor-uc/platform/aducm355/aducm355.h"
+#else
+#error "NO PLATFORM SPECIFIED"
+#endif
+
+#define MUTEX_LOCK(Mutex) (Mutex).super.lock(&(Mutex).super)
+#define MUTEX_UNLOCK(Mutex) (Mutex).super.unlock(&(Mutex).super)
+
 #endif