Added functionality for UDP/TCP/SCTP transferring of data over the ASYNC channel interface.

Added selection of interfaces through the configuration file with a new block of statements as follows:

    FLEXSPLIT_INTERFACES :
    {
        DU_INTERFACE_NAME_FOR_F1U           = "lo";
        DU_IPV4_ADDRESS_FOR_F1U             = "127.0.0.1/24";
        DU_PORT_FOR_F1U			    = 2210;

        CU_INTERFACE_NAME_FOR_F1U           = "lo";
        CU_IPV4_ADDRESS_FOR_F1U             = "127.0.0.1";	//Address to search the DU
        CU_PORT_FOR_F1U			    = 2210;

        // One of TCP/UDP/SCTP
        F1_U_TRANSPORT_TYPE 		    = "UDP";
    };

openair2/UTIL/ASYNC_IF/link_manager.c

@@ -57,7 +57,7 @@ static void *link_manager_sender_thread(void *_manager)
 
   while (manager->run) {
     while (message_get(manager->send_queue, &data, &size, &priority) == 0) {
-      link_send_packet(manager->socket_link, data, size);
+      link_send_packet(manager->socket_link, data, size, manager->type, manager->peer_addr, manager->port);
       free(data);
     }
     //    if (message_get(manager->send_queue, &data, &size, &priority))
@@ -86,7 +86,7 @@ static void *link_manager_receiver_thread(void *_manager)
   LOG_D(MAC, "starting link manager receiver thread\n");
 
   while (manager->run) {
-    if (link_receive_packet(manager->socket_link, &data, &size))
+    if (link_receive_packet(manager->socket_link, &data, &size, manager->type, manager->peer_addr, manager->port))
       goto error;
     /* todo: priority */
     if (message_put(manager->receive_queue, data, size, 0))
@@ -105,7 +105,10 @@ error:
 link_manager_t *create_link_manager(
         message_queue_t *send_queue,
         message_queue_t *receive_queue,
-        socket_link_t   *link)
+        socket_link_t   *link,
+	uint16_t 	type,
+	char 		*peer_addr,
+ 	int		port	  )
 {
   link_manager_t *ret = NULL;
   pthread_attr_t attr;
@@ -120,6 +123,9 @@ link_manager_t *create_link_manager(
   ret->send_queue = send_queue;
   ret->receive_queue = receive_queue;
   ret->socket_link = link;
+  ret->type = type;
+  ret->peer_addr = peer_addr;
+  ret->port = port;
   ret->run = 1;
 
   if (pthread_attr_init(&attr))

openair2/UTIL/ASYNC_IF/link_manager.h

@@ -41,6 +41,7 @@
 #include "message_queue.h"
 //#include "ringbuffer_queue.h"
 #include "socket_link.h"
+#include <stdint.h>
 
 #include <pthread.h>
 
@@ -52,6 +53,9 @@ typedef struct {
   message_queue_t *send_queue;
   message_queue_t *receive_queue;
   socket_link_t   *socket_link;
+  uint16_t	  type;
+  char 		  *peer_addr;
+  int		  port;
   pthread_t       sender;
   pthread_t       receiver;
   volatile int    run;
@@ -60,7 +64,11 @@ typedef struct {
 link_manager_t *create_link_manager(
         message_queue_t *send_queue,
         message_queue_t *receive_queue,
-        socket_link_t   *link);
+        socket_link_t   *link,
+	uint16_t 	type,
+	char 		*peer_addr,
+	int 		port
+ 				  );
 void destroy_link_manager(link_manager_t *);
 
 #ifdef __cplusplus

openair2/UTIL/ASYNC_IF/message_queue.h

@@ -38,6 +38,7 @@
 #ifndef MESSAGE_QUEUE_H
 #define MESSAGE_QUEUE_H
 
+#include <stdint.h>
 #include <pthread.h>
 
 #ifdef __cplusplus
@@ -48,6 +49,7 @@ typedef struct message_t {
   void *data;
   int  size;
   int  priority;
+  uint16_t type;
   struct message_t *next;
 } message_t;
 

openair2/UTIL/ASYNC_IF/socket_link.c

@@ -48,6 +48,7 @@
 #include <netinet/ip.h>
 #include <netinet/tcp.h>
 #include <netinet/udp.h>
+#include <netinet/sctp.h>
 #include <arpa/inet.h>
 #include <stdint.h>
 
@@ -106,7 +107,6 @@ socket_link_t *new_link_server(int port)
     LOG_E(PROTO_AGENT, "%s:%d: accept: %s\n", __FILE__, __LINE__, strerror(errno));
     goto error;
   }
-  printf("Accepted new connection from client\n");
   close(socket_server);
 
   LOG_D(PROTO_AGENT, "connection from %s:%d\n", inet_ntoa(addr.sin_addr), ntohs(addr.sin_port));
@@ -169,6 +169,202 @@ error:
   return NULL;
 }
 
+
+socket_link_t *new_link_udp_server(int port){
+
+  socket_link_t  *ret = NULL;
+
+  struct sockaddr_in si_me, si_other;
+  int socket_server, i, slen = sizeof(si_other) , recv_bytes;
+  char buf[1500];
+
+  ret = calloc(1, sizeof(socket_link_t));
+  if (ret == NULL) {
+    LOG_D(PROTO_AGENT, "%s:%d: out of memory\n", __FILE__, __LINE__);
+    goto error;
+  }
+  ret->socket_fd = -1;
+
+  LOG_D(PROTO_AGENT, "create a new udp link server socket at port %d\n", port);
+
+  //create a UDP socket
+  if ((socket_server=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) {
+        goto error;
+  }
+
+  // zero out the structure
+  memset((char *) &si_me, 0, sizeof(si_me));
+
+  si_me.sin_family = AF_INET;
+  si_me.sin_port = htons(port);
+  si_me.sin_addr.s_addr = htonl(INADDR_ANY);
+
+  //bind socket to port
+  if( bind(socket_server , (struct sockaddr*)&si_me, sizeof(si_me) ) == -1) {
+        goto error;
+  }
+  ret->socket_fd = socket_server;
+  ret->peer_port = 0;
+  return ret;
+  
+error:
+  close(socket_server);
+  if (ret != NULL) close(ret->socket_fd);
+  free(ret);
+  LOG_E(PROTO_AGENT, "ERROR in new_link_udp_server (see above), returning NULL\n");
+  return NULL;
+}
+
+
+socket_link_t *new_link_udp_client(char *server, int port){
+
+  socket_link_t      *ret = NULL;
+  ret = calloc(1, sizeof(socket_link_t));
+  if (ret == NULL) {
+    LOG_E(MAC, "%s:%d: out of memory\n", __FILE__, __LINE__);
+    //goto error;
+  }
+  ret->socket_fd = -1;
+
+  struct sockaddr_in si_other, server_info;
+  int s, i, slen=sizeof(si_other);
+  char buf[1500];
+  char message[1500];
+ 
+  if ( (s=socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1){
+        goto error;
+  }
+ 
+  memset((char *) &si_other, 0, sizeof(si_other));
+  si_other.sin_family = AF_INET;
+  si_other.sin_port = 0; //htons(port);
+  
+
+  
+  if (inet_aton(server, &si_other.sin_addr) == 0){
+        fprintf(stderr, "inet_aton() failed\n");
+        goto error;
+  }
+  connect(s, (struct sockaddr *)&si_other, sizeof(si_other)); 
+  
+  getsockname(s, (struct sockaddr *)&si_other, &slen);
+
+  ret->socket_fd = s;
+  ret->peer_port = port; //ntohs(si_other.sin_port);
+ 
+  return ret;
+error:
+  if (ret != NULL) close(ret->socket_fd);
+  free(ret);
+  LOG_E(MAC, "ERROR in new_link_udp_client (see above), returning NULL\n");
+  return NULL;
+}
+
+
+socket_link_t *new_link_sctp_server(int port)
+{
+
+  socket_link_t  *ret = NULL;
+  ret = calloc(1, sizeof(socket_link_t));
+  if (ret == NULL) {
+    LOG_D(PROTO_AGENT, "%s:%d: out of memory\n", __FILE__, __LINE__);
+    goto error;
+  }
+  ret->socket_fd = -1;
+
+  int listenSock, temp;
+  struct sockaddr_in servaddr;
+ 
+  listenSock = socket (AF_INET, SOCK_STREAM, IPPROTO_SCTP);
+  if(listenSock == -1)
+  {
+      perror("socket()");
+      exit(1);
+  }
+ 
+  bzero ((void *) &servaddr, sizeof (servaddr));
+  servaddr.sin_family = AF_INET;
+  servaddr.sin_addr.s_addr = htonl (INADDR_ANY);
+  servaddr.sin_port = htons(port);
+ 
+  temp = bind (listenSock, (struct sockaddr *) &servaddr, sizeof (servaddr));
+ 
+  if(temp == -1 )
+  {
+      perror("bind()");
+      close(listenSock);
+      exit(1);
+  }
+ 
+  temp = listen (listenSock, 5);
+    if(temp == -1 )
+  {
+      perror("listen()");
+      close(listenSock);
+      exit(1);
+  }
+
+  ret->socket_fd = accept (listenSock, (struct sockaddr *) NULL, (int *) NULL);
+  
+  return ret;
+
+error:
+  close(listenSock);
+  if (ret != NULL) close(ret->socket_fd);
+  free(ret);
+  LOG_E(PROTO_AGENT, "ERROR in new_link_sctp_server (see above), returning NULL\n");
+  return NULL;
+}
+
+socket_link_t *new_link_sctp_client(char *server, int port)
+{
+
+  socket_link_t      *ret = NULL;
+  ret = calloc(1, sizeof(socket_link_t));
+  if (ret == NULL) {
+    LOG_D(PROTO_AGENT, "%s:%d: out of memory\n", __FILE__, __LINE__);
+    goto error;
+  }
+  ret->socket_fd = -1;
+
+  int temp;
+  struct sockaddr_in servaddr;
+    
+  ret->socket_fd = socket (AF_INET, SOCK_STREAM, IPPROTO_SCTP);
+ 
+  if (ret->socket_fd == -1)
+  {
+      perror("socket()");
+      exit(1);
+  }
+ 
+  bzero ((void *) &servaddr, sizeof (servaddr));
+  servaddr.sin_family = AF_INET;
+  servaddr.sin_port = htons (port);
+  if (inet_aton(server, &servaddr.sin_addr) == 0) {
+    LOG_E(PROTO_AGENT, "invalid IP address '%s', use a.b.c.d notation\n", server);
+    goto error;
+  }
+
+  temp = connect (ret->socket_fd, (struct sockaddr *) &servaddr, sizeof (servaddr));
+ 
+  if (temp == -1)
+  {
+      perror("connect()");
+      close(ret->socket_fd);
+      exit(1);
+  }
+
+  return ret;
+
+error:
+  if (ret != NULL) close(ret->socket_fd);
+  free(ret);
+  LOG_E(MAC, "ERROR in new_link_sctp_client (see above), returning NULL\n");
+  return NULL;
+}
+
+
 /*
  * return -1 on error and 0 if the sending was fine
  */
@@ -192,6 +388,43 @@ error:
   return -1;
 }
 
+static int socket_udp_send(int socket_fd, void *buf, int size, char *peer_addr, int port)
+{
+  
+  struct sockaddr_in si_other;
+  int slen = sizeof(si_other);
+  char *s = buf;
+  int   l;
+  int my_socket;
+  
+  LOG_D(PROTO_AGENT,"UDP send\n");
+
+  my_socket = socket_fd;
+  memset((char *) &si_other, 0, sizeof(si_other));
+  si_other.sin_family = AF_INET;
+  si_other.sin_port = htons(port);
+     
+  if (inet_aton(peer_addr , &si_other.sin_addr) == 0) 
+  {
+      fprintf(stderr, "inet_aton() failed\n");
+      exit(1);
+  }
+
+  while (size) {
+    l = sendto(my_socket, s, size, 0, (struct sockaddr *) &si_other, slen);
+    if (l == -1) goto error;
+    if (l == 0) { LOG_E(PROTO_AGENT, "%s:%d: this cannot happen, normally...\n", __FILE__, __LINE__); abort(); }
+    size -= l;
+    s += l;
+  }
+  
+  return 0;
+error:
+  LOG_E(PROTO_AGENT, "socket_udp_send: ERROR: %s\n", strerror(errno));
+  return -1;
+}
+
+
 /*
  * return -1 on error and 0 if the receiving was fine
  */
@@ -219,30 +452,85 @@ socket_closed:
   return -1;
 }
 
+/*
+ * return -1 on error and 0 if the receiving was fine
+ */
+static int socket_udp_receive(int socket_fd, void *buf, int size)
+{
+  LOG_D(PROTO_AGENT,"UDP RECEIVE\n");
+
+  struct sockaddr_in client;
+  int slen = sizeof(client);
+  char *s = buf;
+  int   l;
+
+  while (size) {
+    l = recvfrom(socket_fd, s, size, 0, (struct sockaddr *) &client, &slen);
+    getsockname(s, (struct sockaddr *)&client, &slen);
+    LOG_D(PROTO_AGENT, "Got message from src port: %u\n", ntohs(client.sin_port));
+    if (l == -1) goto error;
+    if (l == 0) goto socket_closed;
+    size -= l;
+    s += l;
+  }
+
+  return ntohs(client.sin_port);
+
+error:
+  LOG_E(MAC, "socket_udp_receive: ERROR: %s\n", strerror(errno));
+  return -1;
+
+socket_closed:
+  LOG_E(MAC, "socket_udp_receive: socket closed\n");
+  return -1;
+}
+
+
 /*
  * return -1 on error and 0 if the sending was fine
  */
-int link_send_packet(socket_link_t *link, void *data, int size)
+int link_send_packet(socket_link_t *link, void *data, int size, uint16_t proto_type, char *peer_addr, int port)
 {
   char sizebuf[4];
   int32_t s = size;
-
+  
   /* send the size first, maximum is 2^31 bytes */
   sizebuf[0] = (s >> 24) & 255;
   sizebuf[1] = (s >> 16) & 255;
   sizebuf[2] = (s >> 8) & 255;
   sizebuf[3] = s & 255;
 
-  if (socket_send(link->socket_fd, sizebuf, 4) == -1)
-    goto error;
+  if ((proto_type == 0) || (proto_type == 2))
+  {
+    if (socket_send(link->socket_fd, sizebuf, 4) == -1)
+      goto error;
+          
+    link->bytes_sent += 4;
 
-  link->bytes_sent += 4;
+    if (socket_send(link->socket_fd, data, size) == -1)
+      goto error;
 
-  if (socket_send(link->socket_fd, data, size) == -1)
-    goto error;
+    link->bytes_sent += size;
+    link->packets_sent++;
+  }
+  else if (proto_type == 1 )
+  {
+    while (link->peer_port == 0)
+    {
+      sleep(0.1);
+    }
+    LOG_D(PROTO_AGENT, "peer port is %d", link->peer_port);
+    if (socket_udp_send(link->socket_fd, sizebuf, 4, peer_addr, link->peer_port) == -1)
+      goto error;
+      
+    link->bytes_sent += 4;
 
-  link->bytes_sent += size;
-  link->packets_sent++;
+    if (socket_udp_send(link->socket_fd, data, size, peer_addr, link->peer_port) == -1)
+      goto error;
+
+    link->bytes_sent += size;
+    link->packets_sent++;
+  }
 
   return 0;
 
@@ -253,16 +541,28 @@ error:
 /*
  * return -1 on error and 0 if the sending was fine
  */
-int link_receive_packet(socket_link_t *link, void **ret_data, int *ret_size)
+int link_receive_packet(socket_link_t *link, void **ret_data, int *ret_size, uint16_t proto_type, char *peer_addr, int port)
 {
   unsigned char sizebuf[4];
   int32_t       size;
   void          *data = NULL;
+  int 		peer_port = 0;
 
+  if ((proto_type == 0) || (proto_type == 2))
+  {
   /* received the size first, maximum is 2^31 bytes */
-  if (socket_receive(link->socket_fd, sizebuf, 4) == -1)
-    goto error;
-
+    if (socket_receive(link->socket_fd, sizebuf, 4) == -1)
+      goto error;
+  }
+  else if (proto_type == 1)
+  {
+      /* received the size first, maximum is 2^31 bytes */
+    peer_port = socket_udp_receive(link->socket_fd, sizebuf, 4);
+      if ( peer_port == -1)
+	goto error;
+    if (link->peer_port == 0) link->peer_port = peer_port;
+  }
+  
   size = (sizebuf[0] << 24) |
          (sizebuf[1] << 16) |
          (sizebuf[2] << 8)  |
@@ -277,10 +577,17 @@ int link_receive_packet(socket_link_t *link, void **ret_data, int *ret_size)
     LOG_E(MAC, "%s:%d: out of memory\n", __FILE__, __LINE__);
     goto error;
   }
-
-  if (socket_receive(link->socket_fd, data, size) == -1)
-    goto error;
-
+  
+  if ((proto_type == 0) || (proto_type == 2))
+  {
+    if (socket_receive(link->socket_fd, data, size) == -1)
+      goto error;
+  }
+  else if (proto_type == 1)
+  {
+    if (socket_udp_receive(link->socket_fd, data, size) == -1)
+      goto error;
+  }
   link->bytes_received += size;
   link->packets_received++;
   *ret_data = data;

openair2/UTIL/ASYNC_IF/socket_link.h

@@ -44,8 +44,10 @@
 extern "C" {
 #endif
 
+  
 typedef struct {
   int      socket_fd;
+  int	   peer_port;
   uint64_t bytes_sent;
   uint64_t packets_sent;
   uint64_t bytes_received;
@@ -54,8 +56,12 @@ typedef struct {
 
 socket_link_t *new_link_server(int port);
 socket_link_t *new_link_client(char *server, int port);
-int link_send_packet(socket_link_t *link, void *data, int size);
-int link_receive_packet(socket_link_t *link, void **data, int *size);
+socket_link_t *new_link_udp_server(int port);
+socket_link_t *new_link_udp_client(char *server, int port);
+socket_link_t *new_link_sctp_server(int port);
+socket_link_t *new_link_sctp_client(char *server, int port);
+int link_send_packet(socket_link_t *link, void *data, int size, uint16_t proto_type, char *peer_addr, int port);
+int link_receive_packet(socket_link_t *link, void **data, int *size, uint16_t proto_type, char *peer_addr, int port);
 int close_link(socket_link_t *link);
 
 #ifdef __cplusplus