Runtime support for watchdogs

core/reactor.c

@@ -363,6 +363,9 @@ int lf_reactor_c_main(int argc, const char* argv[]) {
         _lf_execution_started = true;
         _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

@@ -1316,18 +1316,70 @@ bool _lf_check_deadline(self_base_t* self, bool invoke_deadline_handler) {
     return false;
 }
 
+#ifdef LF_THREADED
+
+void* run_watchdog(watchdog_t* watchdog) {
+    watchdog->thread_active = true;
+    self_base_t* base = watchdog->base;
+
+    while (lf_time_physical() < watchdog->expiration) {
+        interval_t T = watchdog->expiration - lf_time_physical();
+        lf_mutex_unlock(&(base->watchdog_mutex));
+        lf_nanosleep(T);
+        lf_mutex_lock(&(base->watchdog_mutex));
+    }
+
+
+    watchdog_function_t watchdog_func = watchdog->watchdog_function;
+    (*watchdog_func)(watchdog);
+    lf_mutex_unlock(&(base->watchdog_mutex));
+}
+
+void _lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout) {
+    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 = false;
+}
+
+//FIXME: modif4watchdogs
+void _lf_watchdog_stop(watchdog_t* watchdog) {
+
+}
+#endif
+
 /**
  * Invoke the given reaction
  *
  * @param reaction The reaction that has just executed.
  * @param worker The thread number of the worker thread or 0 for unthreaded execution (for tracing).
  */
 void _lf_invoke_reaction(reaction_t* reaction, int worker) {
+
+#ifdef LF_THREADED
+    if (&(((self_base_t*) reaction->self)->watchdog_mutex) != NULL) {
+        lf_mutex_lock(&(((self_base_t*) reaction->self)->watchdog_mutex));
+    }
+#endif
+
     tracepoint_reaction_starts(reaction, worker);
     ((self_base_t*) reaction->self)->executing_reaction = reaction;
     reaction->function(reaction->self);
     ((self_base_t*) reaction->self)->executing_reaction = NULL;
     tracepoint_reaction_ends(reaction, worker);
+
+
+#ifdef LF_THREADED
+    if (&(((self_base_t*) reaction->self)->watchdog_mutex) != NULL) {
+        lf_mutex_unlock(&(((self_base_t*) reaction->self)->watchdog_mutex));
+    }
+#endif
 }
 
 /**

core/threaded/reactor_threaded.c

@@ -1140,6 +1140,8 @@ int lf_reactor_c_main(int argc, const char* argv[]) {
         // it can be probably called in that manner as well).
         _lf_initialize_start_tag();
 
+        _lf_initialize_watchdog_mutexes();
+
         start_threads();
 
         lf_mutex_unlock(&mutex);

include/api/api.h

@@ -169,6 +169,17 @@ 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);
+#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

@@ -141,6 +141,13 @@ 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;
 
@@ -210,6 +217,22 @@ 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.
+    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.
  */
@@ -279,6 +302,10 @@ typedef struct allocation_record_t {
 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. 
+#endif
 #ifdef MODAL_REACTORS
     reactor_mode_state_t _lf__mode_state;    // The current mode (for modal models).
 #endif

include/core/reactor.h

@@ -425,6 +425,11 @@ void _lf_initialize_timers(void);
  */
 void _lf_trigger_startup_reactions(void);
 
+/**
+ * Function to initialize mutexes for watchdogs
+ */
+void _lf_initialize_watchdog_mutexes(void);
+
 
 /**
  * Function (to be code generated) to terminate execution.
@@ -553,5 +558,39 @@ void _lf_fd_send_stop_request_to_rti(void);
  */
 bool _lf_check_deadline(self_base_t* self, bool invoke_deadline_handler);
 
+#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);
+#endif
+
+/**
+ * These functions must be implemented by both threaded and unthreaded
+ * runtime. Should be routed to appropriate API calls in platform.h
+*/
+
+/**
+ * @brief Notify other threads of new events on the event queue.
+ * 
+ */
+void _lf_notify_of_event();
+
+/**
+ * @brief Enter critical section. Must be paired with a
+ * `_lf_critical_section_exit()`
+ * 
+ */
+void _lf_critical_section_enter();
+
+/**
+ * @brief Leave critical section
+ */
+void _lf_critical_section_exit();
+
 #endif /* REACTOR_H */
 /** @} */

include/core/reactor_common.h

@@ -78,6 +78,11 @@ 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);
 bool _lf_check_deadline(self_base_t* self, bool invoke_deadline_handler);
+#ifdef LF_THREADED
+void* run_watchdog(watchdog_t* watchdog);
+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[]);

lib/schedule.c

@@ -192,3 +192,9 @@ trigger_handle_t lf_schedule_value(void* action, interval_t extra_delay, void* v
 bool lf_check_deadline(void* self, bool invoke_deadline_handler) {
 	return _lf_check_deadline((self_base_t*)self, invoke_deadline_handler);
 }
+
+#ifdef LF_THREADED
+void lf_watchdog_start(watchdog_t* watchdog, interval_t additional_timeout) {
+    return _lf_watchdog_start(watchdog, additional_timeout);
+}
+#endif

lingua-franca-ref.txt

@@ -1 +1 @@
-master
+watchdogs

test/src_gen_stub.c

@@ -7,4 +7,5 @@ bool _lf_trigger_shutdown_reactions() { return true; }
 void _lf_set_default_command_line_options() {}
 void _lf_trigger_startup_reactions() {}
 void _lf_initialize_timers() {}
+void _lf_initialize_watchdog_mutexes() {}
 void logical_tag_complete(tag_t tag_to_send) {}