maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

time management module

Browse Source 
This commit introduces a time management module.

See doc/time_management.md for details.
This commit is contained in:
Cedric Roux
2025-04-18 09:02:54 +02:00
parent 9bccad31b5
commit 7cbedb1575
30 changed files with 1983 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/time_manager/time_server.c
+251
View File
@@ -0,0 +1,251 @@
/*
* 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
*
* 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 "time_server.h"
#include <stdbool.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <poll.h>
#include <errno.h>
#include <sys/eventfd.h>
#include <unistd.h>
#include <string.h>
#include "common/utils/system.h"
#include "common/utils/LOG/log.h"
typedef struct {
pthread_t thread_id;
_Atomic bool exit;
int server_socket;
int tick_fd;
void (*callback)(void *);
void *callback_data;
} time_server_private_t;
static void *time_server_thread(void *ts)
{
time_server_private_t *time_server = ts;
struct pollfd *polls = NULL;
int *socket_fds = NULL;
int client_count = 0;
polls = calloc(2, sizeof(struct pollfd));
DevAssert(polls != NULL);
while (!time_server->exit) {
/* polls[0] is to receive ticks */
polls[0].fd = time_server->tick_fd;
polls[0].events = POLLIN;
/* polls[1] is to accept connections from clients */
polls[1].fd = time_server->server_socket;
polls[1].events = POLLIN;
/* add clients */
for (int i = 0; i < client_count; i++) {
polls[i + 2].fd = socket_fds[i];
polls[i + 2].events = POLLIN;
}
int ret = poll(polls, client_count + 2, -1);
DevAssert(ret >= 1 || (ret == -1 && errno == EINTR));
if (time_server->exit)
break;
/* check tick */
if (polls[0].revents) {
/* do not accept errors */
DevAssert(polls[0].revents == POLLIN);
uint64_t ticks;
int ret = read(time_server->tick_fd, &ticks, 8);
DevAssert(ret == 8);
/* call the callback ticks time */
for (int i = 0; i < ticks; i++)
time_server->callback(time_server->callback_data);
/* send ticks by blocks of 255 max (what fits in a uint8_t) */
while (ticks) {
/* get min(ticks, 255) */
uint8_t count = ticks < 255 ? ticks : 255;
/* send to all clients */
for (int i = 0; i < client_count; i++) {
if (write(socket_fds[i], &count, 1) == 1)
continue;
/* error writing - remove client */
LOG_W(UTIL, "time server: client disconnects\n");
shutdown(socket_fds[i], SHUT_RDWR);
close(socket_fds[i]);
client_count--;
i--;
if (client_count == 0) {
free(socket_fds);
socket_fds = NULL;
} else {
memmove(&socket_fds[i + 1], &socket_fds[i + 2], sizeof(int) * (client_count - i - 1));
socket_fds = realloc(socket_fds, sizeof(int) * client_count);
}
polls = realloc(polls, sizeof(struct pollfd) * (client_count + 2));
DevAssert(polls != NULL);
}
ticks -= count;
}
}
/* check new client */
if (polls[1].revents) {
/* do not accept errors */
DevAssert(polls[1].revents == POLLIN);
int new_socket = accept(time_server->server_socket, NULL, NULL);
DevAssert(new_socket != -1);
/* add client */
client_count++;
socket_fds = realloc(socket_fds, sizeof(int) * client_count);
DevAssert(socket_fds != NULL);
socket_fds[client_count - 1] = new_socket;
polls = realloc(polls, sizeof(struct pollfd) * (client_count + 2));
DevAssert(polls != NULL);
continue;
}
/* any event on a client socket is to be considered as an error */
for (int i = 0; i < client_count; i++) {
if (polls[i + 2].revents == 0)
continue;
LOG_W(UTIL,
"time server: error on client's socket, removing client (revents is %d, fd is %d i is %d)\n",
polls[i + 2].revents,
polls[i + 2].fd,
i);
shutdown(socket_fds[i], SHUT_RDWR);
close(socket_fds[i]);
client_count--;
i--;
if (client_count == 0) {
free(socket_fds);
socket_fds = NULL;
} else {
memmove(&socket_fds[i + 1], &socket_fds[i + 2], sizeof(int) * (client_count - i - 1));
socket_fds = realloc(socket_fds, sizeof(int) * client_count);
DevAssert(socket_fds != NULL);
}
if (client_count)
memmove(&polls[i + 2 + 1], &polls[i + 2 + 2], sizeof(struct pollfd) * (client_count - i - 1));
polls = realloc(polls, sizeof(struct pollfd) * (client_count + 2));
DevAssert(polls != NULL);
}
}
/* close all clients' sockets */
for (int i = 0; i < client_count; i++) {
shutdown(socket_fds[i], SHUT_RDWR);
close(socket_fds[i]);
}
free(polls);
free(socket_fds);
return NULL;
}
time_server_t *new_time_server(const char *ip,
int port,
void (*callback)(void *),
void *callback_data)
{
time_server_private_t *ts;
int ret;
ts = calloc(1, sizeof(*ts));
DevAssert(ts != NULL);
ts->server_socket = socket(AF_INET, SOCK_STREAM, 0);
DevAssert(ts->server_socket != -1);
int v = 1;
ret = setsockopt(ts->server_socket, SOL_SOCKET, SO_REUSEADDR, &v, sizeof(v));
DevAssert(ret == 0);
v = 1;
ret = setsockopt(ts->server_socket, IPPROTO_TCP, TCP_NODELAY, &v, sizeof(v));
DevAssert(ret == 0);
ts->tick_fd = eventfd(0, 0);
DevAssert(ts->tick_fd != -1);
struct sockaddr_in ip_addr;
ip_addr.sin_family = AF_INET;
ip_addr.sin_port = htons(port);
ret = inet_aton(ip, &ip_addr.sin_addr);
DevAssert(ret != 0);
ret = bind(ts->server_socket, (struct sockaddr *)&ip_addr, sizeof(ip_addr));
DevAssert(ret == 0);
ret = listen(ts->server_socket, 10); /* 10: arbitrary value */
DevAssert(ret == 0);
ts->callback = callback;
ts->callback_data = callback_data;
threadCreate(&ts->thread_id, time_server_thread, ts, "time server", -1, SCHED_OAI);
return ts;
}
void free_time_server(time_server_t *ts)
{
time_server_private_t *time_server = ts;
/* flag exit */
time_server->exit = true;
/* send a fake tick to wake up the server thread */
uint64_t v = 1;
int ret = write(time_server->tick_fd, &v, 8);
DevAssert(ret == 8);
/* wait for thread termination */
void *retval;
ret = pthread_join(time_server->thread_id, &retval);
/* free memory */
shutdown(time_server->server_socket, SHUT_RDWR);
close(time_server->server_socket);
close(time_server->tick_fd);
free(time_server);
}
static void time_server_tick_callback(void *ts)
{
time_server_private_t *time_server = ts;
uint64_t v = 1;
int ret = write(time_server->tick_fd, &v, 8);
DevAssert(ret == 8);
}
void time_server_attach_time_source(time_server_t *time_server,
time_source_t *time_source)
{
time_source_set_callback(time_source, time_server_tick_callback, time_server);
}