Sync service for RFsimulator

Replace the FD <-> buffer_t hashtable with epoll_event_t mapping.
This simplifies the code, reduces the probability of memory leaks and removes a
possible data race on the hashtable during rfsimulator shutdown.

radio/rfsimulator/CMakeLists.txt

@@ -1,9 +1,10 @@
 add_library(rfsimulator MODULE
             simulator.c
             apply_channelmod.c
+            sync_service.c
             ../../openair1/PHY/TOOLS/signal_energy.c
 )
-target_link_libraries(rfsimulator PRIVATE SIMU HASHTABLE)
+target_link_libraries(rfsimulator PRIVATE SIMU LOG)
 set_target_properties(rfsimulator PROPERTIES LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR})
 
 add_executable(replay_node stored_node.c)

radio/rfsimulator/simulator.c

@@ -50,9 +50,10 @@
 #define CHANNELMOD_DYNAMICLOAD
 #include <openair1/SIMULATION/TOOLS/sim.h>
 #include "rfsimulator.h"
-#include "hashtable.h"
+#include "sync_service.h"
 
 #define PORT 4043 //default TCP port for this simulator
+#define SYNC_PORT 4044 //default TCP port for sync service
 //
 // CirSize defines the number of samples inquired for a read cycle
 // It is bounded by a slot read capability (which depends on bandwidth and numerology)
@@ -70,22 +71,6 @@
 #define byteToSample(a,b) ((a)/(sizeof(sample_t)*(b)))
 
 #define GENERATE_CHANNEL 10 // each frame (or slot?) in DL
-
-// This needs to be re-architected in the future
-//
-// File Descriptors management in rfsimulator is not optimized
-// Relying on FD_SETSIZE (actually 1024) is not appropriated
-// Also the use of fd value as returned by Linux as an index for buf[] structure is not appropriated
-// especially for client (UE) side since only 1 fd per connection to a gNB is needed. On the server
-// side the value should be tuned to the maximum number of connections with UE's which corresponds
-// to the maximum number of UEs hosted by a gNB which is unlikely to be in the order of thousands
-// since all I/Q's would flow through the same TCP transport.
-// Until a convenient management is implemented, the MAX_FD_RFSIMU is used everywhere (instead of
-// FD_SETSIE) and reduced to 125. This should allow for around 20 simultaeous UEs.
-//
-// An indirection level via hashtable was added to allow the software to use FDs above MAX_FD_RFSIMU.
-//
-// #define MAX_FD_RFSIMU FD_SETSIZE
 #define MAX_FD_RFSIMU 250
 #define SEND_BUFF_SIZE 100000000 // Socket buffer size
 
@@ -102,7 +87,7 @@ typedef enum { SIMU_ROLE_SERVER = 1, SIMU_ROLE_CLIENT } simuRole;
   "        chanmod:   enable channel modelisation\n"\
   "        saviq:     enable saving written iqs to a file\n"
 #define CONFIG_HELP_HANG_WORKAROUND "Enable workaroud for server mode hanging on new client connection.\n"
-
+#define CONFIG_HELP_SYNC_SERVICE "Enable sync service: 0 (default) disable, 1 enable server, 2 enable client.\n"
 /*-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------*/
 /*                                            configuration parameters for the rfsimulator device                                                                              */
 /*   optname                     helpstr                     paramflags           XXXptr                               defXXXval                          type         numelt  */
@@ -120,6 +105,7 @@ typedef enum { SIMU_ROLE_SERVER = 1, SIMU_ROLE_CLIENT } simuRole;
     {"prop_delay",             "<propagation delay in ms>\n",         simOpt,  .dblptr=&(rfsimulator->prop_delay_ms),  .defdblval=0.0,                   TYPE_DOUBLE,    0 },\
     {"wait_timeout",           "<wait timeout if no UE connected>\n", simOpt,  .iptr=&(rfsimulator->wait_timeout),     .defintval=1,                     TYPE_INT,       0 },\
     {"hanging-workaround",     CONFIG_HELP_HANG_WORKAROUND,           simOpt,  .iptr=&rfsimulator->hanging_workaround, .defintval=0,                     TYPE_INT,       0 },\
+    {"sync-service",           CONFIG_HELP_SYNC_SERVICE,              simOpt,  .iptr=&rfsimulator->sync_service,      .defintval=SYNC_SERVICE_DISABLED,  TYPE_INT,       0 },\
   };
 
 static void getset_currentchannels_type(char *buf, int debug, webdatadef_t *tdata, telnet_printfunc_t prnt);
@@ -169,8 +155,6 @@ typedef struct {
   int saveIQfile;
   buffer_t buf[MAX_FD_RFSIMU];
   int next_buf;
-  // Hashtable used as an indirection level between file descriptor and the buf array
-  hash_table_t *fd_to_buf_map;
   int rx_num_channels;
   int tx_num_channels;
   double sample_rate;
@@ -185,44 +169,18 @@ typedef struct {
   int wait_timeout;
   double prop_delay_ms;
   int hanging_workaround;
+  int sync_service;
+  sync_service_t *sync_service_ptr;
 } rfsimulator_state_t;
 
-
-static buffer_t *get_buff_from_socket(rfsimulator_state_t *simulator_state, int socket)
+static buffer_t* allocCirBuf(rfsimulator_state_t *bridge, int sock)
 {
-  uint64_t buffer_index;
-  if (hashtable_get(simulator_state->fd_to_buf_map, socket, (void **)&buffer_index) == HASH_TABLE_OK) {
-    return &simulator_state->buf[buffer_index];
-  } else {
-    return NULL;
-  }
-}
-
-static void add_buff_to_socket_mapping(rfsimulator_state_t *simulator_state, int socket, uint64_t buff_index)
-{
-  hashtable_rc_t rc = hashtable_insert(simulator_state->fd_to_buf_map, socket, (void *)buff_index);
-  AssertFatal(rc == HASH_TABLE_OK,
-              "%s sock = %d\n",
-              rc == HASH_TABLE_INSERT_OVERWRITTEN_DATA ? "Duplicate entry in hashtable" : "Hashtable is not allocated",
-              socket);
-}
-
-static void remove_buff_to_socket_mapping(rfsimulator_state_t *simulator_state, int socket)
-{
-  // Failure is fine here
-  hashtable_remove(simulator_state->fd_to_buf_map, socket);
-}
-
-static int allocCirBuf(rfsimulator_state_t *bridge, int sock)
-{
-  /* TODO: cleanup code so that this AssertFatal becomes useless */
-  AssertFatal(sock >= 0 && sock < sizeofArray(bridge->buf), "socket %d is not in range\n", sock);
   uint64_t buff_index = bridge->next_buf++ % MAX_FD_RFSIMU;
   buffer_t *ptr=&bridge->buf[buff_index];
   ptr->circularBuf = calloc(1, sampleToByte(CirSize, 1));
   if (ptr->circularBuf == NULL) {
     LOG_E(HW, "malloc(%lu) failed\n", sampleToByte(CirSize, 1));
-    return -1;
+    return NULL;
   }
   ptr->circularBufEnd=((char *)ptr->circularBuf)+sampleToByte(CirSize,1);
   ptr->conn_sock=sock;
@@ -234,14 +192,14 @@ static int allocCirBuf(rfsimulator_state_t *bridge, int sock)
   int sendbuff = SEND_BUFF_SIZE;
   if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff)) != 0) {
     LOG_E(HW, "setsockopt(SO_SNDBUF) failed\n");
-    return -1;
+    return NULL;
   }
   struct epoll_event ev= {0};
   ev.events = EPOLLIN | EPOLLRDHUP;
-  ev.data.fd = sock;
+  ev.data.ptr = ptr;
   if (epoll_ctl(bridge->epollfd, EPOLL_CTL_ADD, sock, &ev) != 0) {
     LOG_E(HW, "epoll_ctl(EPOLL_CTL_ADD) failed\n");
-    return -1;
+    return NULL;
   }
 
   if ( bridge->channelmod > 0) {
@@ -269,7 +227,7 @@ static int allocCirBuf(rfsimulator_state_t *bridge, int sock)
       ptr->channel_model = find_channel_desc_fromname(legacy_model_name);
       if (!ptr->channel_model) {
         LOG_E(HW, "Channel model %s/%s not found, check config file\n", modelname, legacy_model_name);
-        return -1;
+        return NULL;
       }
     }
 
@@ -278,34 +236,27 @@ static int allocCirBuf(rfsimulator_state_t *bridge, int sock)
     random_channel(ptr->channel_model,false);
     LOG_I(HW, "Random channel %s in rfsimulator activated\n", modelname);
   }
-  add_buff_to_socket_mapping(bridge, sock, buff_index);
-  return 0;
+  return ptr;
 }
 
-static void removeCirBuf(rfsimulator_state_t *bridge, int sock) {
-  if (epoll_ctl(bridge->epollfd, EPOLL_CTL_DEL, sock, NULL) != 0) {
+static void removeCirBuf(rfsimulator_state_t *bridge, buffer_t *buf) {
+  if (epoll_ctl(bridge->epollfd, EPOLL_CTL_DEL, buf->conn_sock, NULL) != 0) {
     LOG_E(HW, "epoll_ctl(EPOLL_CTL_DEL) failed\n");
   }
-  close(sock);
-  buffer_t* buf = get_buff_from_socket(bridge, sock);
-  if (buf) {
-    free(buf->circularBuf);
-    // Fixme: no free_channel_desc_scm(bridge->buf[sock].channel_model) implemented
-    // a lot of mem leaks
-    //free(bridge->buf[sock].channel_model);
-    memset(buf, 0, sizeof(buffer_t));
-    buf->conn_sock=-1;
-    remove_buff_to_socket_mapping(bridge, sock);
-    nb_ue--;
-  }
+  close(buf->conn_sock);
+  free(buf->circularBuf);
+  // Fixme: no free_channel_desc_scm(bridge->buf[sock].channel_model) implemented
+  // a lot of mem leaks
+  //free(bridge->buf[sock].channel_model);
+  memset(buf, 0, sizeof(buffer_t));
+  buf->conn_sock = -1;
+  nb_ue--;
 }
 
-static void socketError(rfsimulator_state_t *bridge, int sock) {
-  buffer_t* buf = get_buff_from_socket(bridge, sock);
-  if (!buf) return;
+static void socketError(rfsimulator_state_t *bridge, buffer_t *buf) {
   if (buf->conn_sock != -1) {
     LOG_W(HW, "Lost socket\n");
-    removeCirBuf(bridge, sock);
+    removeCirBuf(bridge, buf);
 
     if (bridge->role == SIMU_ROLE_CLIENT)
       exit(1);
@@ -661,7 +612,7 @@ static int startServer(openair0_device *device)
   }
   struct epoll_event ev = {0};
   ev.events = EPOLLIN;
-  ev.data.fd = t->listen_sock;
+  ev.data.ptr = NULL;
   if (epoll_ctl(t->epollfd, EPOLL_CTL_ADD, t->listen_sock, &ev) != 0) {
     LOG_E(HW, "epoll_ctl(EPOLL_CTL_ADD) failed, errno(%d)\n", errno);
     return -1;
@@ -711,6 +662,9 @@ static int client_try_connect(const char *host, uint16_t port)
 static int startClient(openair0_device *device)
 {
   rfsimulator_state_t *t = device->priv;
+  if (t->sync_service != SYNC_SERVICE_DISABLED) {
+    t->sync_service_ptr = sync_service_init(t->sync_service, t->ip, SYNC_PORT);
+  }
   t->role = SIMU_ROLE_CLIENT;
   int sock;
 
@@ -730,7 +684,7 @@ static int startClient(openair0_device *device)
   if (setblocking(sock, notBlocking) == -1) {
     return -1;
   }
-  return allocCirBuf(t, sock);
+  return allocCirBuf(t, sock) == NULL ? -1 : 0;
 }
 
 static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags, bool alreadyLocked) {
@@ -815,9 +769,9 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
   }
 
   for (int nbEv = 0; nbEv < nfds; ++nbEv) {
-    int fd=events[nbEv].data.fd;
+    buffer_t *b = events[nbEv].data.ptr;
 
-    if (events[nbEv].events & EPOLLIN && fd == t->listen_sock) {
+    if (events[nbEv].events & EPOLLIN && b == NULL) {
       int conn_sock;
       conn_sock = accept(t->listen_sock, NULL, NULL);
       if (conn_sock == -1) {
@@ -827,7 +781,8 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
       if (setblocking(conn_sock, notBlocking)) {
         return false;
       }
-      if (allocCirBuf(t, conn_sock) == -1) {
+      buffer_t *new_buf = allocCirBuf(t, conn_sock);
+      if (new_buf == NULL) {
         return false;
       }
       LOG_I(HW, "A client connects, sending the current time\n");
@@ -840,17 +795,13 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
 
       rfsimulator_write_internal(t, t->lastWroteTS > 1 ? t->lastWroteTS - 1 : 0, samplesVoid, 1, t->tx_num_channels, 1, false);
 
-      buffer_t *b = get_buff_from_socket(t, conn_sock);
-      if (b->channel_model)
-        b->channel_model->start_TS = t->lastWroteTS;
+      if (new_buf->channel_model)
+        new_buf->channel_model->start_TS = t->lastWroteTS;
     } else {
       if ( events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP) ) {
-        socketError(t,fd);
+        socketError(t, b);
         continue;
       }
-
-      buffer_t *b = get_buff_from_socket(t, fd);
-      if (!b) continue;
       if ( b->circularBuf == NULL ) {
         LOG_E(HW, "Received data on not connected socket %d\n", events[nbEv].data.fd);
         continue;
@@ -865,7 +816,7 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
                  b->remainToTransfer :
                  b->circularBufEnd - b->transferPtr ;
 
-      ssize_t sz=recv(fd, b->transferPtr, blockSz, MSG_DONTWAIT);
+      ssize_t sz=recv(b->conn_sock, b->transferPtr, blockSz, MSG_DONTWAIT);
 
       if ( sz < 0 ) {
         if ( errno != EAGAIN ) {
@@ -925,7 +876,7 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
         pthread_mutex_lock(&Sockmutex);
 
         if (t->lastWroteTS != 0 && (fabs((double)t->lastWroteTS-b->lastReceivedTS) > (double)CirSize))
-          LOG_W(HW, "UEsock(%d) Tx/Rx shift too large Tx:%lu, Rx:%lu\n", fd, t->lastWroteTS, b->lastReceivedTS);
+          LOG_W(HW, "UEsock(%d) Tx/Rx shift too large Tx:%lu, Rx:%lu\n", b->conn_sock, t->lastWroteTS, b->lastReceivedTS);
 
         pthread_mutex_unlock(&Sockmutex);
         b->transferPtr=(char *)&b->circularBuf[(b->lastReceivedTS*b->th.nbAnt)%CirSize];
@@ -935,11 +886,11 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
       if ( b->headerMode==false ) {
         if ( ! b->trashingPacket ) {
           b->lastReceivedTS=b->th.timestamp+b->th.size-byteToSample(b->remainToTransfer,b->th.nbAnt);
-          LOG_D(HW, "UEsock: %d Set b->lastReceivedTS %ld\n", fd, b->lastReceivedTS);
+          LOG_D(HW, "UEsock: %d Set b->lastReceivedTS %ld\n", b->conn_sock, b->lastReceivedTS);
         }
 
         if ( b->remainToTransfer==0) {
-          LOG_D(HW, "UEsock: %d Completed block reception: %ld\n", fd, b->lastReceivedTS);
+          LOG_D(HW, "UEsock: %d Completed block reception: %ld\n", b->conn_sock, b->lastReceivedTS);
           b->headerMode=true;
           b->transferPtr=(char *)&b->th;
           b->remainToTransfer = sizeof(samplesBlockHeader_t);
@@ -999,6 +950,11 @@ static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimest
           }
       }
 
+      if (t->sync_service_ptr != NULL) {
+        int ret = sync_service_sync(t->sync_service_ptr, t->nextRxTstamp + nsamps);
+        AssertFatal(ret == 0, "sync_service_sync() failed: %d\n", ret);
+      }
+
       if (have_to_wait) {
         LOG_D(HW,
               "Waiting on socket, current last ts: %ld, expected at least : %ld\n",
@@ -1142,10 +1098,9 @@ static void rfsimulator_end(openair0_device *device) {
   for (int i = 0; i < MAX_FD_RFSIMU; i++) {
     buffer_t *b = &s->buf[i];
     if (b->conn_sock >= 0 )
-      removeCirBuf(s, b->conn_sock);
+      removeCirBuf(s, b);
   }
   close(s->epollfd);
-  hashtable_destroy(&s->fd_to_buf_map);
   free(s);
 }
 static void stopServer(openair0_device *device)
@@ -1171,11 +1126,6 @@ static int rfsimulator_write_init(openair0_device *device) {
   return 0;
 }
 
-void do_not_free_integer(void *integer)
-{
-  (void)integer;
-}
-
 __attribute__((__visibility__("default")))
 int device_init(openair0_device *device, openair0_config_t *openair0_cfg) {
   // to change the log level, use this on command line
@@ -1222,7 +1172,6 @@ int device_init(openair0_device *device, openair0_config_t *openair0_cfg) {
   for (int i = 0; i < MAX_FD_RFSIMU; i++)
     rfsimulator->buf[i].conn_sock=-1;
   rfsimulator->next_buf = 0;
-  rfsimulator->fd_to_buf_map = hashtable_create(MAX_FD_RFSIMU, NULL, do_not_free_integer);
 
   AssertFatal((rfsimulator->epollfd = epoll_create1(0)) != -1, "epoll_create1() failed, errno(%d)", errno);
   // we need to call randominit() for telnet server (use gaussdouble=>uniformrand)

radio/rfsimulator/sync_service.c

@@ -0,0 +1,171 @@
+/*
+ * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The OpenAirInterface Software Alliance licenses this file to You under
+ * the OAI Public License, Version 1.1  (the "License"); you may not use this file
+ * except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.openairinterface.org/?page_id=698
+ *
+ * Author and copyright: Laurent Thomas, open-cells.com
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *-------------------------------------------------------------------------------
+ * For more information about the OpenAirInterface (OAI) Software Alliance:
+ *      contact@openairinterface.org
+ */
+
+#include "sync_service.h"
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <errno.h>
+#include "common/utils/utils.h"
+#include <fcntl.h>
+#include "log.h"
+
+#define MAX_CLIENTS 10
+
+struct sync_service_t {
+  sync_service_mode_t mode;
+  int sockfd;
+  struct sockaddr_in addr;
+  uint64_t last_received_time;
+  int client_sockfd[MAX_CLIENTS];
+};
+
+// Initialize sync service as server or client.
+// For server, bind_addr and port specify where to listen.
+// For client, server_addr and port specify where to connect.
+sync_service_t* sync_service_init(sync_service_mode_t mode, const char* addr, uint16_t port)
+{
+  sync_service_t* svc = calloc_or_fail(1, sizeof(sync_service_t));
+  svc->mode = mode;
+  svc->sockfd = socket(AF_INET, SOCK_STREAM, 0);
+  if (svc->sockfd < 0) {
+    free(svc);
+    return NULL;
+  }
+  memset(&svc->addr, 0, sizeof(svc->addr));
+  svc->addr.sin_family = AF_INET;
+  svc->addr.sin_port = htons(port);
+  svc->addr.sin_addr.s_addr = inet_addr(addr);
+  if (mode == SYNC_SERVICE_SERVER) {
+    LOG_I(HW, "Initializing sync_service as SERVER on %s:%d\n", addr, port);
+  } else {
+    LOG_I(HW, "Initializing sync_service as CLIENT connecting to %s:%d\n", addr, port);
+  }
+
+  if (mode == SYNC_SERVICE_SERVER) {
+    if (bind(svc->sockfd, (struct sockaddr*)&svc->addr, sizeof(svc->addr)) < 0) {
+      close(svc->sockfd);
+      free(svc);
+      return NULL;
+    }
+    if (listen(svc->sockfd, 1) < 0) {
+      close(svc->sockfd);
+      free(svc);
+      return NULL;
+    }
+    int flags = fcntl(svc->sockfd, F_GETFL, 0);
+    if (flags >= 0) {
+        fcntl(svc->sockfd, F_SETFL, flags | O_NONBLOCK);
+    }
+    memset(svc->client_sockfd, 0, sizeof(svc->client_sockfd));
+  } else {
+    int retries = 5;
+    while (retries-- > 0) {
+      if (connect(svc->sockfd, (struct sockaddr*)&svc->addr, sizeof(svc->addr)) == 0) {
+        break;
+      }
+      if (retries == 0) {
+        close(svc->sockfd);
+        free(svc);
+        return NULL;
+      }
+      sleep(1);
+    }
+    // As client, receive the first sync message to initialize last_received_time
+    uint64_t initial_time = 0;
+    ssize_t recvd = recv(svc->sockfd, &initial_time, sizeof(initial_time), 0);
+    if (recvd != sizeof(initial_time)) {
+      close(svc->sockfd);
+      free(svc);
+      return NULL;
+    }
+    svc->last_received_time = initial_time;
+  }
+  return svc;
+}
+
+// Destroy sync service and free resources.
+void sync_service_destroy(sync_service_t* svc)
+{
+  if (!svc)
+    return;
+  close(svc->sockfd);
+  free(svc);
+}
+
+// Server: sends sync message with 'time' to client(s).
+// Client: blocks until a sync message with at least 'time' is received.
+int sync_service_sync(sync_service_t* svc, uint64_t time)
+{
+  if (!svc) {
+    AssertFatal(0, "recv() failed, errno(%d)\n", errno);
+  }
+  if (svc->mode == SYNC_SERVICE_SERVER) {
+    // Accept 1 new client per call
+    socklen_t addr_len = sizeof(svc->addr);
+    int client_fd = accept(svc->sockfd, (struct sockaddr*)&svc->addr, &addr_len);
+    if (client_fd >= 0) {
+      for (int i = 0; i < MAX_CLIENTS; i++) {
+        if (svc->client_sockfd[i] == 0) {
+          svc->client_sockfd[i] = client_fd;
+          break;
+        }
+      }
+    }
+
+    for (int i = 0; i < MAX_CLIENTS; i++) {
+      if (svc->client_sockfd[i] > 0) {
+        // Send sync message to each client
+        ssize_t sent = send(svc->client_sockfd[i], &time, sizeof(time), 0);
+        if (sent != sizeof(time)) {
+          close(svc->client_sockfd[i]);
+          svc->client_sockfd[i] = 0;
+        }
+      }
+    }
+    return 0;
+  } else {
+    // Wait for a time value >= requested time
+    if (svc->last_received_time >= time)
+      return 0;
+    uint64_t recv_time = 0;
+    while (1) {
+      ssize_t recvd = recv(svc->sockfd, &recv_time, sizeof(recv_time), 0);
+      if (recvd == sizeof(recv_time) && recv_time >= time) {
+        svc->last_received_time = recv_time;
+        return 0;
+
+        if (recvd < 0 && errno != EINTR) {
+          AssertFatal(0, "recv() failed, errno(%d)\n", errno);
+          LOG_E(HW, "recv() failed, errno(%d)\n", errno);
+          close(svc->sockfd);
+          free(svc);
+          return -1;
+        }
+      }
+    }
+  }
+}

radio/rfsimulator/sync_service.h

@@ -0,0 +1,53 @@
+/*
+* Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
+* contributor license agreements.  See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The OpenAirInterface Software Alliance licenses this file to You under
+* the OAI Public License, Version 1.1  (the "License"); you may not use this file
+* except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*      http://www.openairinterface.org/?page_id=698
+*
+* Author and copyright: Laurent Thomas, open-cells.com
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*-------------------------------------------------------------------------------
+* For more information about the OpenAirInterface (OAI) Software Alliance:
+*      contact@openairinterface.org
+*/
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef enum {
+    SYNC_SERVICE_DISABLED,
+    SYNC_SERVICE_SERVER,
+    SYNC_SERVICE_CLIENT
+} sync_service_mode_t;
+
+typedef struct sync_service_t sync_service_t;
+
+// Initialize sync service as server or client.
+// For server, bind_addr and port specify where to listen.
+// For client, server_addr and port specify where to connect.
+sync_service_t* sync_service_init(sync_service_mode_t mode, const char* addr, uint16_t port);
+
+// Destroy sync service and free resources.
+void sync_service_destroy(sync_service_t* svc);
+
+// Server: sends sync message with 'time' to client(s).
+// Client: blocks until a sync message with at least 'time' is received.
+int sync_service_sync(sync_service_t* svc, uint64_t time);
+
+#ifdef __cplusplus
+}
+#endif