Draft: Apply pluggable security for federated execution.

core/federated/RTI/main.c

@@ -137,6 +137,7 @@ void usage(int argc, const char* argv[]) {
   lf_print("  -a, --auth Turn on HMAC authentication options.\n");
   lf_print("  -t, --tracing Turn on tracing.\n");
   lf_print("  -d, --disable_dnet Turn off the use of DNET signals.\n");
+  lf_print("  -sst, --sst SST config path for RTI.\n");
 
   lf_print("Command given:");
   for (int i = 0; i < argc; i++) {
@@ -234,7 +235,7 @@ int process_args(int argc, const char* argv[]) {
       rti.base.number_of_scheduling_nodes = (int32_t)num_federates; // FIXME: Loses numbers on 64-bit machines
       lf_print("RTI: Number of federates: %d", rti.base.number_of_scheduling_nodes);
     } else if (strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) {
-#ifdef COMM_TYPE_TCP
+#if defined(COMM_TYPE_TCP) || defined(COMM_TYPE_SST)
       if (argc < i + 2) {
         lf_print_error("--port needs a short unsigned integer argument ( > 0 and < %d).", UINT16_MAX);
         usage(argc, argv);
@@ -266,6 +267,15 @@ int process_args(int argc, const char* argv[]) {
       return 0;
 #endif
       rti.authentication_enabled = true;
+    } else if (strcmp(argv[i], "-sst") == 0 || strcmp(argv[i], "--sst") == 0) {
+#ifndef COMM_TYPE_SST
+      lf_print_error("--sst requires the RTI to be built with the --DCOMM_TYPE=SST option.");
+      usage(argc, argv);
+      return 0;
+#else
+      i++;
+      lf_set_sst_config_path(argv[i]);
+#endif
     } else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--tracing") == 0) {
       rti.base.tracing_enabled = true;
     } else if (strcmp(argv[i], "-d") == 0 || strcmp(argv[i], "--dnet_disabled") == 0) {

core/federated/RTI/rti_remote.c

@@ -1572,6 +1572,7 @@ void initialize_RTI(rti_remote_t* rti) {
 
   // Initialize thread synchronization primitives
   LF_MUTEX_INIT(&rti_mutex);
+  LF_MUTEX_INIT(&shutdown_mutex);
   LF_COND_INIT(&received_start_times, &rti_mutex);
   LF_COND_INIT(&sent_start_time, &rti_mutex);
 

core/federated/federate.c

@@ -46,6 +46,9 @@ extern bool _lf_termination_executed;
 
 // Global variables references in federate.h
 lf_mutex_t lf_outbound_netchan_mutex;
+
+lf_mutex_t lf_inbound_netchan_mutex;
+
 lf_cond_t lf_port_status_changed;
 
 /**
@@ -400,12 +403,11 @@ static trigger_handle_t schedule_message_received_from_network_locked(environmen
  *  federate.
  */
 static void close_inbound_netchan(int fed_id) {
-  LF_MUTEX_LOCK(&netchan_mutex);
-  if (_fed.netchans_for_inbound_p2p_connections[fed_id] != NULL) {
-    shutdown_netchan(_fed.netchans_for_inbound_p2p_connections[fed_id], false);
-    _fed.netchans_for_inbound_p2p_connections[fed_id] = NULL;
+  LF_MUTEX_LOCK(&lf_inbound_netchan_mutex);
+  if (_fed.netchans_for_inbound_p2p_connections[fed_id] >= 0) {
+    shutdown_netchan(&_fed.netchans_for_inbound_p2p_connections[fed_id], false);
   }
-  LF_MUTEX_UNLOCK(&netchan_mutex);
+  LF_MUTEX_UNLOCK(&lf_inbound_netchan_mutex);
 }
 
 /**
@@ -722,7 +724,6 @@ static int handle_port_absent_message(netchan_t netchan, int fed_id) {
  * network channel in _fed.netchans_for_inbound_p2p_connections
  * to -1 and returns, terminating the thread.
  * @param _args The remote federate ID (cast to void*).
- * @param fed_id_ptr A pointer to a uint16_t containing federate ID being listened to.
  *  This procedure frees the memory pointed to before returning.
  */
 static void* listen_to_federates(void* _args) {
@@ -1948,18 +1949,10 @@ void lf_connect_to_rti(const char* hostname, int port) {
 
 void lf_create_server(int specified_port) {
   assert(specified_port <= UINT16_MAX && specified_port >= 0);
-
-  netchan_t server_netchan = initialize_netchan();
-  set_my_port(server_netchan, specified_port);
-
-  if (create_server(server_netchan, false)) {
-    lf_print_error_system_failure("RTI failed to create server: %s.", strerror(errno));
+  if (create_server(specified_port, &_fed.server_socket, (uint16_t*)&_fed.server_port, TCP, false)) {
+    lf_print_error_system_failure("RTI failed to create TCP server: %s.", strerror(errno));
   };
-  _fed.server_netchan = server_netchan;
-  // Get the final server port to send to the RTI on an MSG_TYPE_ADDRESS_ADVERTISEMENT message.
-  int32_t server_port = get_my_port(server_netchan);
-
-  LF_PRINT_LOG("Server for communicating with other federates started using port %d.", server_port);
+  LF_PRINT_LOG("Server for communicating with other federates started using port %d.", _fed.server_port);
 
   // Send the server port number to the RTI
   // on an MSG_TYPE_ADDRESS_ADVERTISEMENT message (@see net_common.h).
@@ -2086,12 +2079,12 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
     int result = lf_thread_create(&_fed.inbound_netchan_listeners[received_federates], listen_to_federates, fed_id_arg);
     if (result != 0) {
       // Failed to create a listening thread.
-      LF_MUTEX_LOCK(&netchan_mutex);
+      LF_MUTEX_LOCK(&lf_inbound_netchan_mutex);
       if (_fed.netchans_for_inbound_p2p_connections[remote_fed_id] != NULL) {
         shutdown_netchan(_fed.netchans_for_inbound_p2p_connections[remote_fed_id], false);
         _fed.netchans_for_inbound_p2p_connections[remote_fed_id] = NULL;
       }
-      LF_MUTEX_UNLOCK(&netchan_mutex);
+      LF_MUTEX_UNLOCK(&lf_inbound_netchan_mutex);
       lf_print_error_and_exit("Failed to create a thread to listen for incoming physical connection. Error code: %d.",
                               result);
     }

core/reactor_common.c

@@ -919,6 +919,9 @@ void usage(int argc, const char* argv[]) {
   printf("   The address of the RTI, which can be in the form of user@host:port or ip:port.\n\n");
   printf("  -l\n");
   printf("   Send stdout to individual log files for each federate.\n\n");
+#ifdef COMM_TYPE_SST
+  printf("  -sst, --sst <n>\n");
+#endif
 #endif
 
   printf("Command given:\n");
@@ -1068,6 +1071,17 @@ int process_args(int argc, const char* argv[]) {
         return 0;
       }
     }
+#endif
+#ifdef COMM_TYPE_SST
+    else if (strcmp(arg, "-sst") == 0 || strcmp(arg, "--sst") == 0) {
+      if (argc < i + 1) {
+        lf_print_error("--sst needs a string argument.");
+        usage(argc, argv);
+        return 0;
+      }
+      const char* fid = argv[i++];
+      lf_set_sst_config_path(fid);
+    }
 #endif
     else if (strcmp(arg, "--ros-args") == 0) {
       // FIXME: Ignore ROS arguments for now

core/threaded/reactor_threaded.c

@@ -199,9 +199,8 @@ bool wait_until(instant_t wait_until_time, lf_cond_t* condition) {
     LF_PRINT_DEBUG("-------- Waiting until physical time " PRINTF_TIME, wait_until_time - start_time);
     // Check whether we actually need to wait, or if we have already passed the timepoint.
     interval_t wait_duration = wait_until_time - lf_time_physical();
-    if (wait_duration < MIN_SLEEP_DURATION) {
-      LF_PRINT_DEBUG("Wait time " PRINTF_TIME " is less than MIN_SLEEP_DURATION " PRINTF_TIME ". Skipping wait.",
-                     wait_duration, MIN_SLEEP_DURATION);
+    if (wait_duration < 0) {
+      LF_PRINT_DEBUG("We have already passed " PRINTF_TIME ". Skipping wait.", wait_until_time);
       return true;
     }
 

include/core/federated/federate.h

@@ -215,10 +215,15 @@ typedef enum parse_rti_code_t { SUCCESS, INVALID_PORT, INVALID_HOST, INVALID_USE
 // Global variables
 
 /**
- * Mutex lock held while performing network channel write and close operations.
+ * Mutex lock held while performing outbound network channel write and close operations.
  */
 extern lf_mutex_t lf_outbound_netchan_mutex;
 
+/**
+ * Mutex lock held while performing inbound network channel write and close operations.
+ */
+extern lf_mutex_t lf_inbound_netchan_mutex;
+
 /**
  * Condition variable for blocking on unkonwn federate input ports.
  */

include/core/threaded/reactor_threaded.h

@@ -93,8 +93,8 @@ void lf_synchronize_with_other_federates(void);
  * if that event time matches or exceeds the specified time.
  *
  * The mutex lock associated with the condition argument is assumed to be held by
- * the calling thread. This mutex is released while waiting. If the wait time is
- * too small to actually wait (less than MIN_SLEEP_DURATION), then this function
+ * the calling thread. This mutex is released while waiting. If the current physical
+ * time has already passed the specified time, then this function
  * immediately returns true and the mutex is not released.
  *
  * @param env Environment within which we are executing.

logging/api/logging_macros.h

@@ -107,4 +107,4 @@ static const bool _lf_log_level_is_debug = LOG_LEVEL >= LOG_LEVEL_DEBUG;
     }                                                                                                                  \
   } while (0)
 #endif // NDEBUG
-#endif // LOGGING_MACROS_H
+#endif // LOGGING_MACROS_H

low_level_platform/api/low_level_platform.h

@@ -113,12 +113,12 @@ int lf_mutex_lock(lf_mutex_t* mutex);
 /**
  * @brief Get the number of cores on the host machine.
  */
-int lf_available_cores();
+int lf_available_cores(void);
 
 /**
  * @brief Return the lf_thread_t of the calling thread.
  */
-lf_thread_t lf_thread_self();
+lf_thread_t lf_thread_self(void);
 
 /**
  * Create a new thread, starting with execution of lf_thread
@@ -273,12 +273,12 @@ int _lf_cond_timedwait(lf_cond_t* cond, instant_t wakeup_time);
  * @brief The ID of the current thread. The only guarantee is that these IDs will be a contiguous range of numbers
  * starting at 0.
  */
-int lf_thread_id();
+int lf_thread_id(void);
 
 /**
  * @brief Initialize the thread ID for the current thread.
  */
-void initialize_lf_thread_id();
+void initialize_lf_thread_id(void);
 #endif // !defined(LF_SINGLE_THREADED)
 
 /**

network/api/lf_sst_support.h

@@ -0,0 +1,15 @@
+#ifndef LF_SST_SUPPORT_H
+#define LF_SST_SUPPORT_H
+
+#include "socket_common.h"
+#include <sst-c-api/c_api.h>
+
+typedef struct sst_priv_t {
+  socket_priv_t* socket_priv;
+  SST_ctx_t* sst_ctx;
+  SST_session_ctx_t* session_ctx;
+} sst_priv_t;
+
+void lf_set_sst_config_path(const char* config_path);
+
+#endif /* LF_SST_SUPPORT_H */

network/api/net_driver.h

@@ -3,13 +3,17 @@
 
 #include "socket_common.h"
 
+#if defined(COMM_TYPE_SST)
+#include "lf_sst_support.h"
+#endif
+
 typedef void* netchan_t;
 
 /**
  * Allocate memory for the network channel.
  * @return netchan_t Initialized network channel.
  */
-netchan_t initialize_netchan();
+netchan_t initialize_netchan(void);
 
 /**
  * Create a netchannel server. This is such as a server socket which accepts connections.

network/api/socket_common.h

@@ -77,9 +77,9 @@
 typedef enum socket_type_t { TCP, UDP } socket_type_t;
 
 /**
- * Mutex protecting socket close operations.
+ * Mutex protecting socket shutdown operations.
  */
-extern lf_mutex_t netchan_mutex;
+extern lf_mutex_t shutdown_mutex;
 
 typedef struct socket_priv_t {
   int socket_descriptor;

network/impl/CMakeLists.txt

@@ -10,6 +10,10 @@ target_sources(lf-network-impl PUBLIC
 
 if(COMM_TYPE MATCHES TCP)
     target_sources(lf-network-impl PUBLIC ${CMAKE_CURRENT_LIST_DIR}/src/lf_socket_support.c)
+elseif(COMM_TYPE MATCHES SST)
+    find_package(sst-lib REQUIRED)
+    target_sources(lf-network-impl PUBLIC ${CMAKE_CURRENT_LIST_DIR}/src/lf_sst_support.c)
+    target_link_libraries(lf-network-impl PUBLIC sst-lib::sst-c-api)
 else()
     message(FATAL_ERROR "Your communication type is not supported! The C target supports TCP.")
 endif()