Move to CLOCK_REALTIME introduce clock.h and lf_atomic.h

CMakeLists.txt

@@ -22,6 +22,9 @@ if(DEFINED LF_SINGLE_THREADED)
     add_compile_definitions(LF_SINGLE_THREADED=1)
 endif()
 
+# Warnings as errors
+add_compile_options(-Werror)
+
 set(Test test)
 set(Lib lib)
 set(CoreLibPath core)

core/federated/RTI/CMakeLists.txt

@@ -66,6 +66,7 @@ add_executable(
     rti_remote.c
     ${CoreLib}/trace.c
     ${LF_PLATFORM_FILE}
+    ${CoreLib}/platform/lf_atomic_gcc_clang.c
     ${CoreLib}/platform/lf_unix_clock_support.c
     ${CoreLib}/utils/util.c
     ${CoreLib}/tag.c
@@ -92,6 +93,8 @@ target_compile_definitions(RTI PUBLIC PLATFORM_${CMAKE_SYSTEM_NAME})
 # Set RTI Tracing
 target_compile_definitions(RTI PUBLIC RTI_TRACE)
 
+# Warnings as errors
+target_compile_options(RTI PUBLIC -Werror)
 # Find threads and link to it
 find_package(Threads REQUIRED)
 target_link_libraries(RTI Threads::Threads)

core/federated/clock-sync.c

@@ -44,9 +44,8 @@ THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #include "net_util.h"
 #include "util.h"
 
-// Global variables defined in tag.c:
-extern interval_t _lf_time_physical_clock_offset;
-extern interval_t _lf_time_test_physical_clock_offset;
+interval_t _lf_clock_sync_offset = NSEC(0);
+interval_t _lf_clock_sync_constant_bias = NSEC(0);
 
 /**
  * Keep a record of connection statistics
@@ -77,6 +76,12 @@ instant_t _lf_last_clock_sync_instant = 0LL;
  */
 int _lf_rti_socket_UDP = -1;
 
+static void adjust_lf_clock_sync_offset(interval_t adjustment) {
+    // Do an atomic adjustment of the clock sync offset. This is needed
+    // to ensure thread-safety on 32bit platforms.
+    lf_atomic_fetch_add64(&_lf_clock_sync_offset, adjustment);
+}
+
 #ifdef _LF_CLOCK_SYNC_COLLECT_STATS
 /**
  * Update statistic on the socket based on the newly calculated network delay
@@ -380,7 +385,7 @@ void handle_T4_clock_sync_message(unsigned char* buffer, int socket, instant_t r
         // Apply a jitter attenuator to the estimated clock error to prevent
         // large jumps in the underlying clock.
         // Note that estimated_clock_error is calculated using lf_time_physical() which includes
-        // the _lf_time_physical_clock_offset adjustment.
+        // the clock sync adjustment.
         adjustment = estimated_clock_error / _LF_CLOCK_SYNC_ATTENUATION;
     } else {
         // Use of TCP socket means we are in the startup phase, so
@@ -420,21 +425,19 @@ void handle_T4_clock_sync_message(unsigned char* buffer, int socket, instant_t r
         // which means we can now adjust the clock offset.
         // For the AVG algorithm, history is a running average and can be directly
         // applied
-        _lf_time_physical_clock_offset += _lf_rti_socket_stat.history;
+        adjust_lf_clock_sync_offset(_lf_rti_socket_stat.history);
         // @note AVG and SD will be zero if collect-stats is set to false
         LF_PRINT_LOG("Clock sync:"
                     " New offset: " PRINTF_TIME "."
                     " Round trip delay to RTI (now): " PRINTF_TIME "."
                     " (AVG): " PRINTF_TIME "."
                     " (SD): " PRINTF_TIME "."
-                    " Local round trip delay: " PRINTF_TIME "."
-                    " Test offset: " PRINTF_TIME ".",
-                    _lf_time_physical_clock_offset,
+                    " Local round trip delay: " PRINTF_TIME ".",
+                    _lf_clock_sync_offset,
                     network_round_trip_delay,
                     stats.average,
                     stats.standard_deviation,
-                    _lf_rti_socket_stat.local_delay,
-                    _lf_time_test_physical_clock_offset);
+                    _lf_rti_socket_stat.local_delay);
         // Reset the stats
         reset_socket_stat(&_lf_rti_socket_stat);
         // Set the last instant at which the clocks were synchronized
@@ -554,4 +557,19 @@ int create_clock_sync_thread(lf_thread_t* thread_id) {
 #endif // _LF_CLOCK_SYNC_ON
     return 0;
 }
+
+#if defined (_LF_CLOCK_SYNC_ON)
+void clock_sync_apply_offset(instant_t *t) {
+    *t += (_lf_clock_sync_offset + _lf_clock_sync_constant_bias);
+}
+
+void clock_sync_remove_offset(instant_t *t) {
+    *t -= (_lf_clock_sync_offset + _lf_clock_sync_constant_bias);
+}
+
+void clock_sync_set_constant_bias(interval_t offset) {
+    _lf_clock_sync_constant_bias = offset;
+}
+#endif
+
 #endif

core/federated/federate.c

@@ -42,7 +42,6 @@
 #endif
 
 // Global variables defined in tag.c:
-extern instant_t _lf_last_reported_unadjusted_physical_time_ns;
 extern instant_t start_time;
 
 // Global variable defined in reactor_common.c:
@@ -1841,7 +1840,7 @@ void lf_connect_to_federate(uint16_t remote_federate_id) {
                             remote_federate_id, CONNECT_MAX_RETRIES);
                 return;
             }
-            lf_print_warning("Could not connect to federate %d. Will try again every %lld nanoseconds.\n",
+            lf_print_warning("Could not connect to federate %d. Will try again every" PRINTF_TIME "nanoseconds.\n",
                    remote_federate_id, ADDRESS_QUERY_RETRY_INTERVAL);
 
             // Check whether the RTI is still there.
@@ -2144,7 +2143,7 @@ void lf_enqueue_port_absent_reactions(environment_t* env){
         return;
     }
 #endif
-    LF_PRINT_DEBUG("Enqueueing port absent reactions at time %lld.", (long long) (env->current_tag.time - start_time));
+    LF_PRINT_DEBUG("Enqueueing port absent reactions at time " PRINTF_TIME ".", (env->current_tag.time - start_time));
     if (num_port_absent_reactions == 0) {
         LF_PRINT_DEBUG("No port absent reactions.");
         return;
@@ -2509,13 +2508,7 @@ tag_t lf_send_next_event_tag(environment_t* env, tag_t tag, bool wait_for_reply)
         // RTI. That amount of time will be no greater than ADVANCE_MESSAGE_INTERVAL in the future.
         LF_PRINT_DEBUG("Waiting for physical time to elapse or an event on the event queue.");
 
-        // The above call to bounded_NET called lf_time_physical()
-        // set _lf_last_reported_unadjusted_physical_time_ns, the
-        // time obtained using CLOCK_REALTIME before adjustment for
-        // clock synchronization. Since that is the clock used by
-        // lf_cond_timedwait, this is the clock we want to use.
-        instant_t wait_until_time_ns =
-                _lf_last_reported_unadjusted_physical_time_ns + ADVANCE_MESSAGE_INTERVAL;
+        instant_t wait_until_time_ns = lf_time_physical() + ADVANCE_MESSAGE_INTERVAL;
 
         // Regardless of the ADVANCE_MESSAGE_INTERVAL, do not let this
         // wait exceed the time of the next tag.

core/platform/CMakeLists.txt

@@ -12,12 +12,12 @@ set(LF_PLATFORM_FILES
     lf_zephyr_clock_counter.c
     lf_zephyr_clock_kernel.c
     lf_rp2040_support.c
+    lf_atomic_windows.c
+    lf_atomic_gcc_clang.c
+    lf_atomic_irq.c
 )
 
-if(${CMAKE_SYSTEM_NAME} STREQUAL "Windows")
-    set(CMAKE_SYSTEM_VERSION 10.0)
-    message("Using Windows SDK version ${CMAKE_VS_WINDOWS_TARGET_PLATFORM_VERSION}")
-elseif(${CMAKE_SYSTEM_NAME} STREQUAL "Nrf52")
+if(${CMAKE_SYSTEM_NAME} STREQUAL "Nrf52")
     list(APPEND REACTORC_COMPILE_DEFS PLATFORM_NRF52)
 elseif(${CMAKE_SYSTEM_NAME} STREQUAL "Zephyr")
     list(APPEND REACTORC_COMPILE_DEFS PLATFORM_ZEPHYR)

core/platform/arduino_mbed/ConditionWrapper.cpp

@@ -46,9 +46,9 @@ void condition_delete(void* condition){
     delete cv;
 }
 
-bool condition_wait_for(void* condition, int64_t absolute_time_ns){
+bool condition_wait_for(void* condition, int64_t wakeup_time){
     ConditionVariable* cv = (ConditionVariable*) condition;
-    return cv->wait_for(absolute_time_ns / 1000000LL);
+    return cv->wait_for(wakeup_time / 1000000LL);
 }
 
 int condition_wait(void* condition){

core/platform/lf_C11_threads_support.c

@@ -1,7 +1,6 @@
 #if !defined(LF_SINGLE_THREADED) && !defined(PLATFORM_ARDUINO)
 #include "platform.h"
 #include "lf_C11_threads_support.h"
-
 #include <threads.h>
 #include <stdlib.h>
 #include <stdint.h> // For fixed-width integral types
@@ -45,17 +44,19 @@ int lf_cond_wait(lf_cond_t* cond) {
     return cnd_wait((cnd_t*)&cond->condition, (mtx_t*)cond->mutex);
 }
 
-int lf_cond_timedwait(lf_cond_t* cond, int64_t absolute_time_ns) {
-    // Convert the absolute time to a timespec.
-    // timespec is seconds and nanoseconds.
-    struct timespec timespec_absolute_time
-            = {(time_t)absolute_time_ns / 1000000000LL, (long)absolute_time_ns % 1000000000LL};
-    int return_value = 0;
-    return_value = cnd_timedwait(
+int lf_cond_timedwait(lf_cond_t* cond, instant_t wakeup_time) {
+    clock_sync_remove_offset(&wakeup_time);
+    struct timespec timespec_absolute_time = {
+        .tv_sec = wakeup_time / BILLION,
+        .tv_nsec = wakeup_time % BILLION
+    };
+
+    int return_value = cnd_timedwait(
         (cnd_t*)&cond->condition,
         (mtx_t*)cond->mutex,
         &timespec_absolute_time
     );
+
     switch (return_value) {
         case thrd_timedout:
             return_value = LF_TIMEOUT;

core/platform/lf_POSIX_threads_support.c

@@ -1,6 +1,7 @@
 #if !defined(LF_SINGLE_THREADED) && !defined(PLATFORM_ARDUINO)
 #include "platform.h"
 #include "lf_POSIX_threads_support.h"
+#include "lf_unix_clock_support.h"
 
 #include <pthread.h>
 #include <errno.h>
@@ -62,13 +63,11 @@ int lf_cond_wait(lf_cond_t* cond) {
     return pthread_cond_wait((pthread_cond_t*)&cond->condition, (pthread_mutex_t*)cond->mutex);
 }
 
-int lf_cond_timedwait(lf_cond_t* cond, int64_t absolute_time_ns) {
-    // Convert the absolute time to a timespec.
-    // timespec is seconds and nanoseconds.
-    struct timespec timespec_absolute_time
-            = {(time_t)absolute_time_ns / 1000000000LL, (long)absolute_time_ns % 1000000000LL};
-    int return_value = 0;
-    return_value = pthread_cond_timedwait(
+int lf_cond_timedwait(lf_cond_t* cond, instant_t wakeup_time) {
+    // Remove the clock synchronization offset.
+    clock_sync_remove_offset(&wakeup_time);
+    struct timespec timespec_absolute_time = convert_ns_to_timespec(wakeup_time);
+    int return_value = pthread_cond_timedwait(
         (pthread_cond_t*)&cond->condition,
         (pthread_mutex_t*)cond->mutex,
         &timespec_absolute_time

core/platform/lf_arduino_support.c

@@ -68,6 +68,7 @@ static volatile uint32_t _lf_time_us_low_last = 0;
  */
 int _lf_interruptable_sleep_until_locked(environment_t* env, instant_t wakeup) {
     instant_t now;
+
     _lf_async_event = false;
     lf_disable_interrupts_nested();
 
@@ -122,11 +123,10 @@ int _lf_clock_now(instant_t* t) {
     }
 
     *t = COMBINE_HI_LO(_lf_time_us_high, now_us_low) * 1000ULL;
+    clock_sync_apply_offset(t);
     return 0;
 }
 
-#if defined(LF_SINGLE_THREADED)
-
 int lf_enable_interrupts_nested() {
     if (_lf_num_nested_critical_sections++ == 0) {
         // First nested entry into a critical section.
@@ -148,6 +148,7 @@ int lf_disable_interrupts_nested() {
     return 0;
 }
 
+#if defined(LF_SINGLE_THREADED)
 /**
  * Handle notifications from the runtime of changes to the event queue.
  * If a sleep is in progress, it should be interrupted.
@@ -219,10 +220,10 @@ int lf_cond_wait(lf_cond_t* cond) {
     return 0;
 }
 
-int lf_cond_timedwait(lf_cond_t* cond, instant_t absolute_time_ns) {
+int lf_cond_timedwait(lf_cond_t* cond, instant_t wakeup_time) {
     instant_t now;
     _lf_clock_now(&now);
-    interval_t sleep_duration_ns = absolute_time_ns - now;
+    interval_t sleep_duration_ns = wakeup_time - now;
     bool res = condition_wait_for(*cond, sleep_duration_ns);
     if (!res) {
         return 0;

core/platform/lf_atomic_gcc_clang.c

@@ -0,0 +1,40 @@
+#if defined(PLATFORM_Linux) || defined(PLATFORM_Darwin)
+#if defined(__GNUC__) || defined(__clang__)
+/**
+ * @author Soroush Bateni
+ * @author Erling Rennemo Jellum
+ * @copyright (c) 2023
+ * License: <a href="https://github.com/lf-lang/reactor-c/blob/main/LICENSE.md">BSD 2-clause</a>
+ * @brief Implements the atomics API using GCC/Clang APIs.
+ */
+
+#include "lf_atomic.h"
+#include "platform.h"
+
+int32_t lf_atomic_fetch_add32(int32_t *ptr, int32_t value) { 
+    return __sync_fetch_and_add(ptr, value);
+}
+int64_t lf_atomic_fetch_add64(int64_t *ptr, int64_t value) {
+    return __sync_fetch_and_add(ptr, value);
+}
+int32_t lf_atomic_add_fetch32(int32_t *ptr, int32_t value) {
+    return __sync_add_and_fetch(ptr, value);
+}
+int64_t lf_atomic_add_fetch64(int64_t *ptr, int64_t value) {
+    return __sync_add_and_fetch(ptr, value);
+}
+bool lf_atomic_bool_compare_and_swap32(int32_t *ptr, int32_t oldval, int32_t newval) {
+    return __sync_bool_compare_and_swap(ptr, oldval, newval);
+}
+bool lf_atomic_bool_compare_and_swap64(int64_t *ptr, int64_t oldval, int64_t newval) {
+    return __sync_bool_compare_and_swap(ptr, oldval, newval);
+}
+int32_t  lf_atomic_val_compare_and_swap32(int32_t *ptr, int32_t oldval, int32_t newval) {
+    return __sync_val_compare_and_swap(ptr, oldval, newval);
+}
+int64_t  lf_atomic_val_compare_and_swap64(int64_t *ptr, int64_t oldval, int64_t newval) {
+    return __sync_val_compare_and_swap(ptr, oldval, newval);
+}
+
+#endif
+#endif