Draft: Create network channel.

CMakeLists.txt

@@ -30,7 +30,6 @@ set(PlatformLib platform)
 include_directories(${CMAKE_SOURCE_DIR}/include)
 include_directories(${CMAKE_SOURCE_DIR}/include/core)
 include_directories(${CMAKE_SOURCE_DIR}/include/core/federated)
-include_directories(${CMAKE_SOURCE_DIR}/include/core/federated/network)
 include_directories(${CMAKE_SOURCE_DIR}/include/core/modal_models)
 include_directories(${CMAKE_SOURCE_DIR}/include/core/platform)
 include_directories(${CMAKE_SOURCE_DIR}/include/core/threaded)

core/CMakeLists.txt

@@ -18,7 +18,6 @@ list(APPEND REACTORC_SOURCES ${GENERAL_SOURCES})
 # Add sources for either threaded or single-threaded runtime
 if(DEFINED FEDERATED)
     include(federated/CMakeLists.txt)
-    include(federated/network/CMakeLists.txt)
 endif()
 
 # Add sources for either threaded or single-threaded runtime
@@ -96,10 +95,19 @@ include(${LF_ROOT}/platform/impl/CMakeLists.txt)
 target_link_libraries(reactor-c PUBLIC lf::platform-api)
 target_link_libraries(reactor-c PRIVATE lf::platform-impl)
 
+if(DEFINED FEDERATED)
+    if(NOT DEFINED COMM_TYPE)
+        set(COMM_TYPE TCP)
+    endif()
+    include(${LF_ROOT}/network/api/CMakeLists.txt)
+    include(${LF_ROOT}/network/impl/CMakeLists.txt)
+    target_link_libraries(reactor-c PUBLIC lf::network-api)
+    target_link_libraries(reactor-c PRIVATE lf::network-impl)
+endif()
+
 target_include_directories(reactor-c PUBLIC ../include)
 target_include_directories(reactor-c PUBLIC ../include/core)
 target_include_directories(reactor-c PUBLIC ../include/core/federated)
-target_include_directories(reactor-c PUBLIC ../include/core/federated/network)
 target_include_directories(reactor-c PUBLIC ../include/core/platform)
 target_include_directories(reactor-c PUBLIC ../include/core/modal_models)
 target_include_directories(reactor-c PUBLIC ../include/core/threaded)

core/federated/RTI/CMakeLists.txt

@@ -16,8 +16,6 @@ add_library(${RTI_LIB} STATIC
     ${CoreLib}/utils/util.c
     ${CoreLib}/tag.c
     ${CoreLib}/clock.c
-    ${CoreLib}/federated/network/net_util.c
-    ${CoreLib}/federated/network/socket_common.c
     ${CoreLib}/utils/pqueue_base.c
     ${CoreLib}/utils/pqueue_tag.c
     ${CoreLib}/utils/pqueue.c
@@ -29,14 +27,17 @@ add_executable(${RTI_MAIN} main.c)
 target_include_directories(${RTI_LIB} PUBLIC ../../../include)
 target_include_directories(${RTI_LIB} PUBLIC ${IncludeDir})
 target_include_directories(${RTI_LIB} PUBLIC ${IncludeDir}/federated)
-target_include_directories(${RTI_LIB} PUBLIC ${IncludeDir}/federated/network)
 target_include_directories(${RTI_LIB} PUBLIC ${IncludeDir}/modal_models)
 target_include_directories(${RTI_LIB} PUBLIC ${IncludeDir}/utils)
 
 if (NOT DEFINED LOG_LEVEL)
     set(LOG_LEVEL 0)
 ENDIF(NOT DEFINED LOG_LEVEL)
 
+if(NOT DEFINED COMM_TYPE)
+    set(COMM_TYPE TCP)
+endif()
+
 IF(CMAKE_BUILD_TYPE MATCHES DEBUG)
     # Set the LOG_LEVEL to 4 to get DEBUG messages
     message("-- Building RTI with DEBUG messages enabled")
@@ -71,13 +72,22 @@ target_link_libraries(${RTI_LIB} PUBLIC  lf::low-level-platform-impl)
 include(${LF_ROOT}/low_level_platform/api/CMakeLists.txt)
 target_link_libraries(${RTI_LIB} PUBLIC  lf::low-level-platform-api)
 
+include(${LF_ROOT}/network/impl/CMakeLists.txt)
+target_link_libraries(${RTI_LIB} PUBLIC  lf::network-impl)
+
+include(${LF_ROOT}/network/api/CMakeLists.txt)
+target_link_libraries(${RTI_LIB} PUBLIC  lf::network-api)
+
 # Set the STANDALONE_RTI flag to include the rti_remote and rti_common.
 target_compile_definitions(${RTI_LIB} PUBLIC STANDALONE_RTI=1)
 
 # Set FEDERATED to get federated compilation support
 target_compile_definitions(${RTI_LIB} PUBLIC FEDERATED=1)
 target_compile_definitions(${RTI_LIB} PUBLIC PLATFORM_${CMAKE_SYSTEM_NAME})
 
+# Set communication type.
+target_compile_definitions(${RTI_LIB} PUBLIC COMM_TYPE_${COMM_TYPE})
+
 # Set RTI Tracing
 target_compile_definitions(${RTI_LIB} PUBLIC RTI_TRACE)
 

core/federated/RTI/main.c

@@ -80,11 +80,11 @@ static void send_failed_signal(federate_info_t* fed) {
   if (rti.base.tracing_enabled) {
     tracepoint_rti_to_federate(send_FAILED, fed->enclave.id, NULL);
   }
-  int failed = write_to_socket(fed->socket, bytes_to_write, &(buffer[0]));
+  int failed = write_to_netchan(fed->fed_netchan, bytes_to_write, &(buffer[0]));
   if (failed == 0) {
     LF_PRINT_LOG("RTI has sent failed signal to federate %d due to abnormal termination.", fed->enclave.id);
   } else {
-    lf_print_error("RTI failed to send failed signal to federate %d on socket ID %d.", fed->enclave.id, fed->socket);
+    lf_print_error("RTI failed to send failed signal to federate %d.", fed->enclave.id);
   }
 }
 
@@ -234,6 +234,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 (argc < i + 2) {
         lf_print_error("--port needs a short unsigned integer argument ( > 0 and < %d).", UINT16_MAX);
         usage(argc, argv);
@@ -247,6 +248,9 @@ int process_args(int argc, const char* argv[]) {
         return 0;
       }
       rti.user_specified_port = (uint16_t)RTI_port;
+#else
+      lf_print_error("--port is only available for TCP.");
+#endif
     } else if (strcmp(argv[i], "-c") == 0 || strcmp(argv[i], "--clock_sync") == 0) {
       if (argc < i + 2) {
         lf_print_error("--clock-sync needs off|init|on.");
@@ -327,9 +331,8 @@ int main(int argc, const char* argv[]) {
     rti.base.scheduling_nodes[i] = (scheduling_node_t*)fed_info;
   }
 
-  int socket_descriptor = start_rti_server(rti.user_specified_port);
-  if (socket_descriptor >= 0) {
-    wait_for_federates(socket_descriptor);
+  if (!start_rti_server()) {
+    wait_for_federates();
     normal_termination = true;
     if (rti.base.tracing_enabled) {
       // No need for a mutex lock because all threads have exited.