Watchdog support

core/CMakeLists.txt

@@ -105,5 +105,6 @@ define(NUMBER_OF_FEDERATES)
 define(NUMBER_OF_WORKERS)
 define(SCHEDULER)
 define(LF_SOURCE_DIRECTORY)
+define(LF_PACKAGE_DIRECTORY)
 define(LF_FILE_SEPARATOR)
 define(WORKERS_NEEDED_FOR_FEDERATE)

core/reactor.c

@@ -368,8 +368,6 @@ int lf_reactor_c_main(int argc, const char* argv[]) {
         _lf_trigger_startup_reactions();
         _lf_initialize_timers();
 
-        _lf_initialize_watchdog_mutexes();
-
         // If the stop_tag is (0,0), also insert the shutdown
         // reactions. This can only happen if the timeout time
         // was set to 0.

core/reactor_common.c

@@ -56,6 +56,14 @@ THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #include "hashset/hashset.h"
 #include "hashset/hashset_itr.h"
 
+#ifdef LF_THREADED
+#include "watchdog.h"
+
+// Code generated global variables.
+extern int _lf_watchdog_number;
+extern watchdog_t* _lf_watchdogs;
+#endif
+
 // Global variable defined in tag.c:
 extern tag_t current_tag;
 extern instant_t start_time;
@@ -1291,50 +1299,6 @@ trigger_handle_t _lf_schedule_int(lf_action_base_t* action, interval_t extra_del
     return _lf_schedule_value(action, extra_delay, container, 1);
 }
 
-#ifdef LF_THREADED
-
-void* run_watchdog(void* arg) {
-    watchdog_t* watchdog = (watchdog_t*)arg;
-
-    self_base_t* base = watchdog->base;
-    lf_mutex_lock(&(base->watchdog_mutex));
-
-    while (lf_time_physical() < watchdog->expiration) {
-        interval_t T = watchdog->expiration - lf_time_physical();
-        lf_mutex_unlock(&(base->watchdog_mutex));
-        lf_sleep(T);
-        lf_mutex_lock(&(base->watchdog_mutex));
-    }
-
-    if (watchdog->expiration != NEVER) {
-        watchdog_function_t watchdog_func = watchdog->watchdog_function;
-        (*watchdog_func)(base);
-    }
-    watchdog->thread_active = false;
-
-    lf_mutex_unlock(&(base->watchdog_mutex));
-    watchdog->thread_active = false;
-    return NULL;
-}
-
-void _lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout) {
-    // Assumes reaction mutex is already held.
-
-    self_base_t* base = watchdog->base;
-
-    watchdog->expiration = lf_time_logical() + watchdog->min_expiration + additional_timeout;
-
-    if (!watchdog->thread_active) {
-        lf_thread_create(&(watchdog->thread_id), run_watchdog, watchdog);
-        watchdog->thread_active = true;
-    } 
-}
-
-void _lf_watchdog_stop(watchdog_t* watchdog) {
-    watchdog->expiration = NEVER;
-}
-#endif
-
 /**
 
  * Invoke the given reaction
@@ -1345,8 +1309,8 @@ void _lf_watchdog_stop(watchdog_t* watchdog) {
 void _lf_invoke_reaction(reaction_t* reaction, int worker) {
 
 #ifdef LF_THREADED
-    if (((self_base_t*) reaction->self)->has_watchdog == true) {
-        lf_mutex_lock(&(((self_base_t*) reaction->self)->watchdog_mutex));
+    if (((self_base_t*) reaction->self)->reactor_mutex != NULL) {
+        lf_mutex_lock((lf_mutex_t*)((self_base_t*)reaction->self)->reactor_mutex);
     }
 #endif
 
@@ -1358,8 +1322,8 @@ void _lf_invoke_reaction(reaction_t* reaction, int worker) {
 
 
 #ifdef LF_THREADED
-    if (((self_base_t*) reaction->self)->has_watchdog == true) {
-        lf_mutex_unlock(&(((self_base_t*) reaction->self)->watchdog_mutex));
+    if (((self_base_t*) reaction->self)->reactor_mutex != NULL) {
+        lf_mutex_unlock((lf_mutex_t*)((self_base_t*)reaction->self)->reactor_mutex);
     }
 #endif
 }
@@ -1802,6 +1766,13 @@ void termination(void) {
         lf_print_warning("Memory allocated for tokens has not been freed!");
         lf_print_warning("Number of unfreed tokens: %d.", _lf_count_token_allocations);
     }
+#ifdef LF_THREADED
+    for (int i = 0; i < _lf_watchdog_number; i++) {
+        if (_lf_watchdogs[i].base->reactor_mutex != NULL) {
+            free(_lf_watchdogs[i].base->reactor_mutex);
+        }
+    }
+#endif
     _lf_free_all_reactors();
     free(_lf_is_present_fields);
     free(_lf_is_present_fields_abbreviated);

core/threaded/CMakeLists.txt

@@ -7,6 +7,7 @@ set(
     scheduler_NP.c
     scheduler_PEDF_NP.c
     scheduler_sync_tag_advance.c
+    watchdog.c
 )
 list(APPEND INFO_SOURCES ${THREADED_SOURCES})
 

core/threaded/watchdog.c

@@ -0,0 +1,83 @@
+/**
+ * @file
+ * @author Benjamin Asch
+ * @author Edward A. Lee
+ * @copyright (c) 2023, The University of California at Berkeley.
+ * License: <a href="https://github.com/lf-lang/reactor-c/blob/main/LICENSE.md">BSD 2-clause</a>
+ * @brief Definitions for watchdogs.
+ */
+
+#include <assert.h>
+#include "watchdog.h"
+
+extern int _lf_watchdog_number;
+extern watchdog_t* _lf_watchdogs;
+
+/**
+ * @brief Initialize watchdog mutexes.
+ * For any reactor with one or more watchdogs, the self struct should have a non-NULL
+ * `reactor_mutex` field which points to an instance of `lf_mutex_t`.
+ * This function initializes those mutexes.
+ */
+void _lf_initialize_watchdog_mutexes() {
+    for (int i = 0; i < _lf_watchdog_number; i++) {
+        self_base_t* current_base = _lf_watchdogs[i].base;
+        if (current_base->reactor_mutex != NULL) {
+            lf_mutex_init((lf_mutex_t*)(current_base->reactor_mutex));
+        }
+    }
+}
+
+/**
+ * @brief Thread function for watchdog.
+ * This function sleeps until physical time exceeds the expiration time of
+ * the watchdog and then invokes the watchdog expiration handler function.
+ * In normal usage, the expiration time is incremented while the thread is
+ * sleeping, so the watchdog never expires and the handler function is never
+ * invoked.
+ * This function acquires the reaction mutex and releases it while sleeping.
+ * 
+ * @param arg A pointer to the watchdog struct
+ * @return NULL
+ */
+void* _lf_run_watchdog(void* arg) {
+    watchdog_t* watchdog = (watchdog_t*)arg;
+
+    self_base_t* base = watchdog->base;
+    assert(base->reactor_mutex != NULL);
+    lf_mutex_lock((lf_mutex_t*)(base->reactor_mutex));
+    instant_t physical_time = lf_time_physical();
+    while (physical_time < watchdog->expiration) {
+        interval_t T = watchdog->expiration - physical_time;
+        lf_mutex_unlock((lf_mutex_t*)base->reactor_mutex);
+        lf_sleep(T);
+        lf_mutex_lock((lf_mutex_t*)(base->reactor_mutex));
+        physical_time = lf_time_physical();
+    }
+
+    if (watchdog->expiration != NEVER) {
+        watchdog_function_t watchdog_func = watchdog->watchdog_function;
+        (*watchdog_func)(base);
+    }
+    watchdog->thread_active = false;
+
+    lf_mutex_unlock((lf_mutex_t*)(base->reactor_mutex));
+    return NULL;
+}
+
+void lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout) {
+    // Assumes reaction mutex is already held.
+
+    self_base_t* base = watchdog->base;
+
+    watchdog->expiration = lf_time_logical() + watchdog->min_expiration + additional_timeout;
+
+    if (!watchdog->thread_active) {
+        lf_thread_create(&(watchdog->thread_id), _lf_run_watchdog, watchdog);
+        watchdog->thread_active = true;
+    } 
+}
+
+void lf_watchdog_stop(watchdog_t* watchdog) {
+    watchdog->expiration = NEVER;
+}

include/api/api.h

@@ -170,19 +170,6 @@ trigger_handle_t lf_schedule_value(void* action, interval_t extra_delay, void* v
  */
 bool lf_check_deadline(void* self, bool invoke_deadline_handler);
 
-#ifdef LF_THREADED
-/**
- * Begin the watchdog.
- * 
- * @param watchdog The watchdog to be started.
- * @param additional_timeout The timeout to be added to the minimum
- * experiation of the watchdog.
- **/
-void lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout);
-
-void lf_watchdog_stop(watchdog_t* watchdog);
-#endif
-
 /**
  * Compare two tags. Return -1 if the first is less than
  * the second, 0 if they are equal, and +1 if the first is

include/core/lf_types.h

@@ -140,13 +140,6 @@ typedef pqueue_pri_t index_t;
  */
 typedef void(*reaction_function_t)(void*);
 
-/** 
- * Watchdog function type. The argument passed to one of 
- * these watchdog functions is a pointer to the self struct
- * for the reactor.
- */
-typedef void(*watchdog_function_t)(void*);
-
 /** Trigger struct representing an output, timer, action, or input. See below. */
 typedef struct trigger_t trigger_t;
 
@@ -216,23 +209,6 @@ struct event_t {
     event_t* next;            // Pointer to the next event lined up in superdense time.
 };
 
-/** Typdef for watchdog_t struct, used to call watchdog handler. */
-typedef struct watchdog_t watchdog_t;
-
-#ifdef LF_THREADED
-/** Watchdog struct for handler. */
-struct watchdog_t {
-    struct self_base_t* base;                // The reactor that contains the watchdog.
-    trigger_t* trigger;                      // The trigger associated with this watchdog.
-    instant_t expiration;                   // The expiration instant for the watchdog. (Initialized to NEVER)
-    interval_t min_expiration;              // The minimum expiration interval for the watchdog.
-    lf_thread_t thread_id;                  // The thread that the watchdog is meant to run on.
-    bool thread_active;                     // Boolean indicating whether or not thread is active.  
-    watchdog_function_t watchdog_function;  // The function/handler for the watchdog.
-};
-#endif 
-
-
 /**
  * Trigger struct representing an output, timer, action, or input.
  */
@@ -303,9 +279,8 @@ typedef struct self_base_t {
 	struct allocation_record_t *allocations;
 	struct reaction_t *executing_reaction;   // The currently executing reaction of the reactor.
 #ifdef LF_THREADED
-    lf_mutex_t watchdog_mutex; // The mutex for this reactor to be acquired before reaction
-                               // invocation. 
-    bool has_watchdog;       // Boolean signifying initialization of watchdog_mutex
+    void* reactor_mutex; // If not null, this is expected to point to an lf_mutex_t.
+                          // It is not declared as such to avoid a dependence on platform.h.
 #endif
 #ifdef MODAL_REACTORS
     reactor_mode_state_t _lf__mode_state;    // The current mode (for modal models).

include/core/platform.h

@@ -36,6 +36,10 @@ THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #ifndef PLATFORM_H
 #define PLATFORM_H
 
+#ifdef __cplusplus
+extern "C" {
+#endif
+
 #include "lf_types.h"
 
 #if defined(LF_THREADED) && defined(LF_UNTHREADED)
@@ -313,4 +317,8 @@ extern int lf_sleep_until_locked(instant_t wakeup_time);
  */
 DEPRECATED(extern int lf_nanosleep(interval_t sleep_duration));
 
+#ifdef __cplusplus
+}
+#endif
+
 #endif // PLATFORM_H

include/core/reactor.h

@@ -535,20 +535,6 @@ trigger_handle_t _lf_schedule_copy(lf_action_base_t* action, interval_t offset,
  */
 void _lf_fd_send_stop_request_to_rti(void);
 
-
-#ifdef LF_THREADED
-/** 
- * Function to start the watchdog.
- * 
- * @param watchdog The watchdog to be started
- * @param additional_timeout Additional timeout to be added to the watchdog's
- * minimum expiration.
- **/
-void _lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout);
-
-void _lf_watchdog_stop(watchdog_t* watchdog);
-#endif
-
 /**
  * These functions must be implemented by both threaded and unthreaded
  * runtime. Should be routed to appropriate API calls in platform.h

include/core/reactor_common.h

@@ -82,11 +82,6 @@ trigger_handle_t _lf_insert_reactions_for_trigger(trigger_t* trigger, lf_token_t
 void _lf_advance_logical_time(instant_t next_time);
 
 trigger_handle_t _lf_schedule_int(lf_action_base_t* action, interval_t extra_delay, int value);
-#ifdef LF_THREADED
-void* run_watchdog(void* arg);
-void _lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout);
-void _lf_watchdog_stop(watchdog_t* watchdog);
-#endif
 void _lf_invoke_reaction(reaction_t* reaction, int worker);
 void schedule_output_reactions(reaction_t* reaction, int worker);
 int process_args(int argc, const char* argv[]);