Draft: Create Network driver instead of directly using sockets.

core/federated/RTI/CMakeLists.txt

@@ -70,6 +70,7 @@ add_executable(
     ${CoreLib}/utils/util.c
     ${CoreLib}/tag.c
     ${CoreLib}/federated/network/net_util.c
+    ${CoreLib}/federated/network/lf_socket_support.c
     ${CoreLib}/utils/pqueue_base.c
     ${CoreLib}/utils/pqueue_tag.c
     ${CoreLib}/utils/pqueue.c

core/federated/RTI/main.c

@@ -281,7 +281,8 @@ int main(int argc, const char* argv[]) {
         initialize_federate(fed_info, i);
         rti.base.scheduling_nodes[i] = (scheduling_node_t *) fed_info;
     }
-
+    _lf_initialize_clock();
+
     int socket_descriptor = start_rti_server(rti.user_specified_port);
     wait_for_federates(socket_descriptor);
     free_scheduling_nodes(rti.base.scheduling_nodes, rti.base.number_of_scheduling_nodes);

core/federated/RTI/rti_remote.c

@@ -27,7 +27,7 @@
  */
 
 #include "rti_remote.h"
-#include "net_util.h"
+// #include "net_util.h"
 #include <string.h>
 
 
@@ -1561,7 +1561,6 @@ int32_t start_rti_server(uint16_t port) {
         // Use the default starting port.
         port = STARTING_PORT;
     }
-    _lf_initialize_clock();
     // Create the TCP socket server
     rti_remote->socket_descriptor_TCP = create_server(specified_port, port, TCP);
     lf_print("RTI: Listening for federates.");
@@ -1632,6 +1631,10 @@ void initialize_RTI(rti_remote_t *rti){
     initialize_rti_common(&rti_remote->base);
     rti_remote->base.mutex = &rti_mutex;
 
+    // TODO: How to make this compile dependent? When should the options be determined?
+    rti_remote->rti_netdrv = socket_init(TCP);
+    rti_remote->clock_netdrv = socket_init(UDP);
+
     // federation_rti related initializations
     rti_remote->max_start_time = 0LL;
     rti_remote->num_feds_proposed_start = 0;

core/federated/RTI/rti_remote.h

@@ -22,6 +22,7 @@
 #include <unistd.h>     // Defines read(), write(), and close()
 #include <strings.h>    // Defines bzero().
 
+#include "net_util.h"
 #include "rti_common.h"
 
 #ifdef __RTI_AUTH__
@@ -32,6 +33,9 @@
 #include "lf_types.h"
 #include "message_record/message_record.h"
 
+//TODO: Need to be fixed later.
+#include "lf_socket_support.h"
+
 /////////////////////////////////////////////
 //// Data structures
 
@@ -119,6 +123,11 @@ typedef struct rti_remote_t {
      */
     const char* federation_id;
 
+    //TODO: Does it have to be a pointer?
+    netdrv_t *rti_netdrv;
+    netdrv_t *clock_netdrv;
+
+
     /************* TCP server information *************/
     /** The desired port specified by the user on the command line. */
     uint16_t user_specified_port;

core/federated/network/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(LF_NETWORK_FILES net_util.c)
+set(LF_NETWORK_FILES net_util.c lf_socket_support.c)
 list(APPEND INFO_SOURCES ${LF_NETWORK_FILES})
 
 list(TRANSFORM LF_NETWORK_FILES PREPEND federated/network/)

core/federated/network/lf_socket_support.c

@@ -0,0 +1,99 @@
+
+#include <netinet/ether.h>
+#include <linux/if.h>		/* IFNAMSIZ */
+#include <arpa/inet.h>		/* htons */
+#include <unistd.h>
+#include <string.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <errno.h>
+
+#include "util.h"
+#include "lf_socket_support.h"
+// #include "net_util.h"
+
+static socket_priv_t *get_priv(netdrv_t *drv)
+{
+	if (!drv)
+		return NULL;
+	return (socket_priv_t *)(drv+1);
+}
+
+// static int socket_open(netdrv_t *drv)
+// {
+// 	socket_priv_t *priv = get_priv(drv);
+// 	priv->sock = socket(AF_PACKET, SOCK_RAW, htons(priv->proto));
+// 	if (priv->sock < 0)
+// 		return -1;
+
+// 	/* If Rx, timeout is a good thing */
+// 	if (priv->timeout_us > 0) {
+// 		struct timeval tv = {
+// 			.tv_sec = priv->timeout_us / 1e6,
+// 			.tv_usec = priv->timeout_us % 1000000,
+// 		};
+// 		if (setsockopt(priv->sock, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv) == -1) {
+// 			printf("%s(): Could not set timeout on socket (%d): %s\n",
+// 				__func__, priv->sock, strerror(errno));
+// 			close(priv->sock);
+// 			return -1;
+// 		}
+// 	}
+// 	/* bind to device, ... */
+
+// 	printf("Socket created\n");
+// 	return 0;
+// }
+
+// static void socket_close(netdrv_t *drv)
+// {
+// 	if (!drv)
+// 		return;
+// 	socket_priv_t *priv = get_priv(drv);
+// 	if (priv->sock > 0) {
+// 		close(priv->sock);
+// 		priv->sock = -1;
+// 	}
+// }
+
+// static int socket_recv(netdrv_t *drv, void * buffer, int size)
+// {
+// 	if (!drv)
+// 		return -1;
+// 	socket_priv_t *priv = get_priv(drv);
+// 	if (priv->timeout_us > 0) {
+// 		int res = -1;
+// 		do {
+// 			res = recv(priv->sock, buffer, size, MSG_TRUNC);
+// 		} while (res > 0);
+// 		return res;
+// 	}
+// 	return recv(priv->sock, buffer, size, MSG_TRUNC);
+// }
+
+// static int socket_send(netdrv_t *drv, void * buffer, int size)
+// {
+// 	if (!drv)
+// 		return -1;
+// 	socket_priv_t *priv = get_priv(drv);
+// 	return send(priv->sock, buffer, size, MSG_DONTWAIT);
+// }
+
+netdrv_t * socket_init(int protocol) {
+    //TODO: Should it be malloc? To support different network stacks operate simulatneously?
+	netdrv_t *drv = malloc(sizeof(*drv) + sizeof(socket_priv_t)); //Don't need to call malloc() twice.
+	// if (!drv) {// check if malloc worked.
+	// 	lf_print_error_and_exit("Falied to malloc netdrv_t.");
+    // }
+	// memset(drv, 0, sizeof(*drv));
+
+	socket_priv_t *priv = get_priv(drv); //drv+1 return.
+	priv->proto = protocol;
+
+	// drv->init = socket_open;
+	// drv->close = socket_close;
+	// drv->read = socket_recv;
+	// drv->write = socket_send;
+	return drv;
+}

core/federated/network/net_util.c

@@ -56,6 +56,33 @@ THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 /** Number of nanoseconds to sleep before retrying a socket read. */
 #define SOCKET_READ_RETRY_INTERVAL 1000000
 
+int  netdrv_open(netdrv_t *drv)
+{
+	if (!drv)
+		return -1;
+	return drv->open(drv);
+}
+void netdrv_close(netdrv_t *drv)
+{
+	if (drv)
+		drv->close(drv);
+}
+
+int netdrv_read(netdrv_t *drv, void * buffer, int size)
+{
+	if (!drv)
+		return -1;
+	return drv->read(drv, buffer, size);
+}
+
+int  netdrv_write(netdrv_t *drv, void * buffer, int size)
+{
+	if (!drv)
+		return -1;
+	return drv->write(drv, buffer, size);
+}
+
+
 int create_real_time_tcp_socket_errexit() {
     int sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
     if (sock < 0) {

include/core/federated/network/lf_socket_support.h

@@ -0,0 +1,15 @@
+#ifndef LF_SOCKET_SUPPORT_H
+#define LF_SOCKET_SUPPORT_H
+
+#include "net_util.h"
+
+typedef struct socket_priv_t {
+	int sock;
+	int proto;
+    int port;
+    int user_specified_port;
+} socket_priv_t;
+
+netdrv_t * socket_init(int protocol);
+
+#endif // LF_SOCKET_SUPPORT_H

include/core/federated/network/net_util.h

@@ -62,6 +62,17 @@ int host_is_big_endian(void);
 
 #ifdef FEDERATED
 
+typedef struct netdrv_t {
+    int  ( *open)(struct netdrv_t *drv);
+    void (*close)(struct netdrv_t *drv);
+    int  ( *read)(struct netdrv_t *drv, void * buffer, int size);
+    int  (*write)(struct netdrv_t *drv, void * buffer, int size);
+} netdrv_t;
+
+int  netdrv_open(netdrv_t *drv);
+void netdrv_close(netdrv_t *drv);
+int  netdrv_read(netdrv_t *drv, void * buffer, int size);
+int  netdrv_write(netdrv_t *drv, void * buffer, int size);
 
 /**
  * @brief Create an IPv4 TCP socket with Nagle's algorithm disabled