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

Merge branch 'master' into develop_SA_L3

Browse Source 
Conflicts:
	openair2/GNB_APP/gnb_config.c
	openair2/RRC/NR/rrc_gNB.c
	openair2/RRC/NR/rrc_gNB_NGAP.c
	openair2/RRC/NR_UE/rrc_UE.c
This commit is contained in:
Xue Song
2021-02-03 14:58:07 +09:00
parent c49220610d 3b2d50374a
commit 6a3b7b40a5
243 changed files with 2999 additions and 12777 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/ocp_itti/intertask_interface.cpp
+90 -77
View File
@@ -30,30 +30,32 @@ extern "C" {
#include <intertask_interface.h>
#include <common/utils/system.h>
typedef struct timer_elm_s {
timer_type_t type; ///< Timer type
long instance;
long duration;
uint64_t timeout;
void *timer_arg; ///< Optional argument that will be passed when timer expires
} timer_elm_t ;
typedef struct timer_elm_s {
timer_type_t type; ///< Timer type
long instance;
long duration;
uint64_t timeout;
void *timer_arg; ///< Optional argument that will be passed when timer expires
} timer_elm_t ;
typedef struct task_list_s {
task_info_t admin;
pthread_t thread;
pthread_mutex_t queue_cond_lock;
std::vector<MessageDef *> message_queue;
std::map<long,timer_elm_t> timer_map;
uint64_t next_timer=UINT64_MAX;
struct epoll_event *events =NULL;
int nb_fd_epoll=0;
int nb_events=0;
int epoll_fd=-1;
int sem_fd=-1;
} task_list_t;
typedef struct task_list_s {
task_info_t admin;
pthread_t thread;
pthread_mutex_t queue_cond_lock;
std::vector<MessageDef *> message_queue;
std::map<long,timer_elm_t> timer_map;
uint64_t next_timer=UINT64_MAX;
struct epoll_event *events =NULL;
int nb_fd_epoll=0;
int nb_events=0;
int epoll_fd=-1;
int sem_fd=-1;
} task_list_t;
int timer_expired(int fd);
task_list_t tasks[TASK_MAX];
int timer_expired(int fd);
static task_list_t **tasks=NULL;
static int nb_queues=0;
static pthread_mutex_t lock_nb_queues;
void *pool_buffer_init (void) {
return 0;
@@ -99,7 +101,7 @@ task_list_t tasks[TASK_MAX];
// #define S1AP_REGISTER_ENB_REQ(mSGpTR) (mSGpTR)->ittiMsg.s1ap_register_enb_req
// would become
// #define S1AP_REGISTER_ENB_REQ(mSGpTR) (s1ap_register_enb_req) mSGpTR)->ittiMsg
MessageDef *itti_alloc_new_message_sized(task_id_t origin_task_id, MessagesIds message_id, MessageHeaderSize size) {
MessageDef *itti_alloc_new_message_sized(task_id_t origin_task_id, instance_t originInstance, MessagesIds message_id, MessageHeaderSize size) {
MessageDef *temp = (MessageDef *)itti_malloc (origin_task_id, TASK_UNKNOWN, sizeof(MessageHeader) +32 + size);
temp->ittiMsgHeader.messageId = message_id;
temp->ittiMsgHeader.originTaskId = origin_task_id;
@@ -107,23 +109,24 @@ task_list_t tasks[TASK_MAX];
return temp;
}
MessageDef *itti_alloc_new_message(task_id_t origin_task_id, MessagesIds message_id) {
MessageDef *itti_alloc_new_message(task_id_t origin_task_id, instance_t originInstance, MessagesIds message_id) {
int size=sizeof(MessageHeader) + 32 + messages_info[message_id].size;
MessageDef *temp = (MessageDef *)itti_malloc (origin_task_id, TASK_UNKNOWN, size);
temp->ittiMsgHeader.messageId = message_id;
temp->ittiMsgHeader.originTaskId = origin_task_id;
temp->ittiMsgHeader.ittiMsgSize = size;
temp->ittiMsgHeader.destinationTaskId=TASK_UNKNOWN;
temp->ittiMsgHeader.instance=0;
temp->ittiMsgHeader.lte_time={0};
temp->ittiMsgHeader.originInstance=originInstance;
temp->ittiMsgHeader.destinationInstance=0;
temp->ittiMsgHeader.lte_time= {0};
return temp;
//return itti_alloc_new_message_sized(origin_task_id, message_id, messages_info[message_id].size);
}
static inline int itti_send_msg_to_task_locked(task_id_t destination_task_id, instance_t instance, MessageDef *message) {
task_list_t *t=tasks+destination_task_id;
static inline int itti_send_msg_to_task_locked(task_id_t destination_task_id, instance_t destinationInstance, MessageDef *message) {
task_list_t *t=tasks[destination_task_id];
message->ittiMsgHeader.destinationTaskId = destination_task_id;
message->ittiMsgHeader.instance = instance;
message->ittiMsgHeader.destinationInstance = destinationInstance;
message->ittiMsgHeader.lte_time.frame = 0;
message->ittiMsgHeader.lte_time.slot = 0;
int message_id = message->ittiMsgHeader.messageId;
@@ -142,25 +145,27 @@ task_list_t tasks[TASK_MAX];
return 0;
}
int itti_send_msg_to_task(task_id_t destination_task_id, instance_t instance, MessageDef *message) {
task_list_t *t=&tasks[destination_task_id];
int itti_send_msg_to_task(task_id_t destination_task_id, instance_t destinationInstance, MessageDef *message) {
task_list_t *t=tasks[destination_task_id];
pthread_mutex_lock (&t->queue_cond_lock);
int ret=itti_send_msg_to_task_locked(destination_task_id, instance, message);
int ret=itti_send_msg_to_task_locked(destination_task_id, destinationInstance, message);
while ( t->message_queue.size()>0 && t->admin.func != NULL ) {
if (t->message_queue.size()>1)
LOG_W(TMR,"queue in no thread mode is %ld\n", t->message_queue.size());
LOG_W(TMR,"queue in no thread mode is %ld\n", t->message_queue.size());
pthread_mutex_unlock (&t->queue_cond_lock);
t->admin.func(NULL);
pthread_mutex_lock (&t->queue_cond_lock);
}
pthread_mutex_unlock (&t->queue_cond_lock);
return ret;
}
void itti_subscribe_event_fd(task_id_t task_id, int fd) {
struct epoll_event event;
task_list_t *t=&tasks[task_id];
task_list_t *t=tasks[task_id];
t->nb_fd_epoll++;
t->events = (struct epoll_event *)realloc((void *)t->events,
t->nb_fd_epoll * sizeof(struct epoll_event));
@@ -173,7 +178,7 @@ task_list_t tasks[TASK_MAX];
}
void itti_unsubscribe_event_fd(task_id_t task_id, int fd) {
task_list_t *t=&tasks[task_id];
task_list_t *t=tasks[task_id];
AssertFatal (epoll_ctl(t->epoll_fd, EPOLL_CTL_DEL, fd, NULL) == 0,
"epoll_ctl (EPOLL_CTL_DEL) failed for task %s, fd %d: %s!\n",
itti_get_task_name(task_id), fd, strerror(errno));
@@ -181,7 +186,7 @@ task_list_t tasks[TASK_MAX];
}
static inline int itti_get_events_locked(task_id_t task_id, struct epoll_event **events) {
task_list_t *t=&tasks[task_id];
task_list_t *t=tasks[task_id];
uint64_t current_time=0;
do {
@@ -194,10 +199,11 @@ task_list_t tasks[TASK_MAX];
t->next_timer=UINT64_MAX;
// Proceed expired timer
for ( auto it=t->timer_map.begin() , next_it = it; it != t->timer_map.end() ; it = next_it ) {
for ( auto it=t->timer_map.begin(), next_it = it; it != t->timer_map.end() ; it = next_it ) {
++next_it;
if ( it->second.timeout < current_time ) {
MessageDef *message = itti_alloc_new_message(TASK_TIMER, TIMER_HAS_EXPIRED);
MessageDef *message = itti_alloc_new_message(TASK_TIMER, it->second.instance,TIMER_HAS_EXPIRED);
message->ittiMsg.timer_has_expired.timer_id=it->first;
message->ittiMsg.timer_has_expired.arg=it->second.timer_arg;
@@ -229,8 +235,9 @@ task_list_t tasks[TASK_MAX];
pthread_mutex_unlock(&t->queue_cond_lock);
LOG_D(TMR,"enter blocking wait for %s\n", itti_get_task_name(task_id));
t->nb_events = epoll_wait(t->epoll_fd,t->events,t->nb_fd_epoll, epoll_timeout);
if ( t->nb_events < 0 && (errno == EINTR || errno == EAGAIN ) )
pthread_mutex_lock(&t->queue_cond_lock);
pthread_mutex_lock(&t->queue_cond_lock);
} while (t->nb_events < 0 && (errno == EINTR || errno == EAGAIN ) );
AssertFatal (t->nb_events >=0,
@@ -259,13 +266,13 @@ task_list_t tasks[TASK_MAX];
}
int itti_get_events(task_id_t task_id, struct epoll_event **events) {
pthread_mutex_lock(&tasks[task_id].queue_cond_lock);
pthread_mutex_lock(&tasks[task_id]->queue_cond_lock);
return itti_get_events_locked(task_id, events);
}
void itti_receive_msg(task_id_t task_id, MessageDef **received_msg) {
// Reception of one message, blocking caller
task_list_t *t=&tasks[task_id];
task_list_t *t=tasks[task_id];
pthread_mutex_lock(&t->queue_cond_lock);
// Weird condition to deal with crap legacy itti interface
@@ -295,10 +302,9 @@ task_list_t tasks[TASK_MAX];
pthread_mutex_unlock (&t->queue_cond_lock);
}
void itti_poll_msg(task_id_t task_id, MessageDef **received_msg)
{
void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
//reception of one message, non-blocking
task_list_t *t=&tasks[task_id];
task_list_t *t=tasks[task_id];
pthread_mutex_lock(&t->queue_cond_lock);
if (!t->message_queue.empty()) {
@@ -312,11 +318,10 @@ task_list_t tasks[TASK_MAX];
}
int itti_create_task(task_id_t task_id,
void *(*start_routine)(void *),
void *args_p)
{
task_list_t *t=&tasks[task_id];
threadCreate (&t->thread, start_routine, args_p, (char*)itti_get_task_name(task_id),-1,OAI_PRIORITY_RT);
void *(*start_routine)(void *),
void *args_p) {
task_list_t *t=tasks[task_id];
threadCreate (&t->thread, start_routine, args_p, (char *)itti_get_task_name(task_id),-1,OAI_PRIORITY_RT);
LOG_I(TMR,"Created Posix thread %s\n", itti_get_task_name(task_id) );
return 0;
}
@@ -325,31 +330,39 @@ task_list_t tasks[TASK_MAX];
pthread_exit (NULL);
}
void itti_terminate_tasks(task_id_t task_id)
{
void itti_terminate_tasks(task_id_t task_id) {
// Sends Terminate signals to all tasks.
itti_send_terminate_message (task_id);
usleep(100*1000); // Allow the tasks to receive the message before going returning to main thread
}
int itti_create_queue(const task_info_t *taskInfo) {
pthread_mutex_lock (&lock_nb_queues);
int newQueue=nb_queues++;
AssertFatal(tasks=(task_list_t **) realloc(tasks, nb_queues * sizeof(*tasks)),"");
tasks[newQueue]= new task_list_t;
pthread_mutex_unlock (&lock_nb_queues);
LOG_I(TMR,"Starting itti queue: %s as task %d\n", taskInfo->name, newQueue);
pthread_mutex_init(&tasks[newQueue]->queue_cond_lock, NULL);
memcpy(&tasks[newQueue]->admin, taskInfo, sizeof(task_info_t));
AssertFatal( ( tasks[newQueue]->epoll_fd = epoll_create1(0) ) >=0, "");
AssertFatal( ( tasks[newQueue]->sem_fd = eventfd(0, EFD_SEMAPHORE) ) >=0, "");
itti_subscribe_event_fd((task_id_t)newQueue, tasks[newQueue]->sem_fd);
if (tasks[newQueue]->admin.threadFunc != NULL)
itti_create_task((task_id_t)newQueue, tasks[newQueue]->admin.threadFunc, NULL);
return newQueue;
}
int itti_init(task_id_t task_max,
thread_id_t thread_max,
MessagesIds messages_id_max,
const task_info_t *tasks_info,
const message_info_t *messages_info)
{
AssertFatal(TASK_MAX<UINT16_MAX, "Max itti tasks");
const task_info_t *tasks
) {
pthread_mutex_init(&lock_nb_queues, NULL);
nb_queues=0;
for(int i=0; i<task_max; ++i) {
LOG_I(TMR,"Starting itti queue: %s as task %d\n", tasks_info[i].name, i);
pthread_mutex_init(&tasks[i].queue_cond_lock, NULL);
memcpy(&tasks[i].admin, &tasks_info[i], sizeof(task_info_t));
AssertFatal( ( tasks[i].epoll_fd = epoll_create1(0) ) >=0, "");
AssertFatal( ( tasks[i].sem_fd = eventfd(0, EFD_SEMAPHORE) ) >=0, "");
itti_subscribe_event_fd((task_id_t)i, tasks[i].sem_fd);
if (tasks[i].admin.threadFunc != NULL)
itti_create_task((task_id_t)i, tasks[i].admin.threadFunc, NULL);
itti_create_queue(&tasks[i]);
}
return 0;
@@ -359,12 +372,11 @@ task_list_t tasks[TASK_MAX];
uint32_t interval_sec,
uint32_t interval_us,
task_id_t task_id,
int32_t instance,
instance_t instance,
timer_type_t type,
void *timer_arg,
long *timer_id)
{
task_list_t *t=&tasks[task_id];
long *timer_id) {
task_list_t *t=tasks[task_id];
do {
// set the taskid in the timer id to keep compatible with the legacy API
@@ -397,13 +409,12 @@ task_list_t tasks[TASK_MAX];
return 0;
}
int timer_remove(long timer_id)
{
int timer_remove(long timer_id) {
task_id_t task_id=(task_id_t)(timer_id&0xffff);
int ret;
pthread_mutex_lock (&tasks[task_id].queue_cond_lock);
ret=tasks[task_id].timer_map.erase(timer_id);
pthread_mutex_unlock (&tasks[task_id].queue_cond_lock);
pthread_mutex_lock (&tasks[task_id]->queue_cond_lock);
ret=tasks[task_id]->timer_map.erase(timer_id);
pthread_mutex_unlock (&tasks[task_id]->queue_cond_lock);
if (ret==1)
return 0;
@@ -418,7 +429,7 @@ task_list_t tasks[TASK_MAX];
}
const char *itti_get_task_name(task_id_t task_id) {
return tasks[task_id].admin.name;
return tasks[task_id]->admin.name;
}
// void for compatibility
@@ -439,5 +450,7 @@ task_list_t tasks[TASK_MAX];
void itti_wait_ready(int wait_tasks) {
// Stupid function, kept for compatibility (the parameter is meaningless!!!)
}
int signal_mask(void) { return 0;}
int signal_mask(void) {
return 0;
}
}
common/utils/ocp_itti/intertask_interface.h
+14 -10
View File
@@ -45,7 +45,7 @@ typedef struct {
uint32_t interval_sec;
uint32_t interval_us;
long task_id;
int32_t instance;
instance_t instance;
timer_type_t type;
void *timer_arg;
long timer_id;
@@ -263,7 +263,7 @@ void *rrc_enb_process_itti_msg(void *);
#include <openair3/NGAP/ngap_gNB.h>
#ifdef ITTI_SIM
#include <openair2/COMMON/itti_sim_messages_types.h>
#include <openair2/COMMON/itti_sim_messages_types.h>
#endif
/*
@@ -300,7 +300,7 @@ typedef struct {
//TASK_DEF(TASK_RRC_ENB, TASK_PRIORITY_MED, 200, NULL, NULL)
//TASK_DEF(TASK_GTPV1_U, TASK_PRIORITY_MED, 1000,NULL, NULL)
//TASK_DEF(TASK_UDP, TASK_PRIORITY_MED, 1000, NULL, NULL)
void * rrc_enb_process_msg(void*);
void *rrc_enb_process_msg(void *);
#define FOREACH_TASK(TASK_DEF) \
TASK_DEF(TASK_UNKNOWN, TASK_PRIORITY_MED, 50, NULL, NULL) \
TASK_DEF(TASK_TIMER, TASK_PRIORITY_MED, 10, NULL, NULL) \
@@ -402,7 +402,8 @@ typedef struct MessageHeader_s {
MessagesIds messageId; /**< Unique message id as referenced in enum MessagesIds */
task_id_t originTaskId; /**< ID of the sender task */
task_id_t destinationTaskId; /**< ID of the destination task */
instance_t instance; /**< Task instance for virtualization */
instance_t originInstance;
instance_t destinationInstance;
itti_lte_time_t lte_time;
MessageHeaderSize ittiMsgSize; /**< Message size (not including header size) */
} MessageHeader;
@@ -432,11 +433,11 @@ typedef struct MessageDef_s {
/* Extract the instance from a message */
#define ITTI_MESSAGE_GET_INSTANCE(mESSAGE) ((mESSAGE)->ittiMsgHeader.instance)
#define ITTI_MSG_ID(mSGpTR) ((mSGpTR)->ittiMsgHeader.messageId)
#define ITTI_MSG_ORIGIN_ID(mSGpTR) ((mSGpTR)->ittiMsgHeader.originTaskId)
#define ITTI_MSG_DESTINATION_ID(mSGpTR) ((mSGpTR)->ittiMsgHeader.destinationTaskId)
#define ITTI_MSG_INSTANCE(mSGpTR) ((mSGpTR)->ittiMsgHeader.instance)
#define ITTI_MSG_ORIGIN_INSTANCE(mSGpTR) ((mSGpTR)->ittiMsgHeader.originInstance)
#define ITTI_MSG_DESTINATION_INSTANCE(mSGpTR) ((mSGpTR)->ittiMsgHeader.destinationInstance)
#define ITTI_MSG_NAME(mSGpTR) itti_get_message_name(ITTI_MSG_ID(mSGpTR))
#define ITTI_MSG_ORIGIN_NAME(mSGpTR) itti_get_task_name(ITTI_MSG_ORIGIN_ID(mSGpTR))
#define ITTI_MSG_DESTINATION_NAME(mSGpTR) itti_get_task_name(ITTI_MSG_DESTINATION_ID(mSGpTR))
@@ -499,6 +500,8 @@ int itti_create_task(task_id_t task_id,
void *(*start_routine) (void *),
void *args_p);
int itti_create_queue(const task_info_t *task_info);
/** \brief Exit the current task.
**/
void itti_exit_task(void);
@@ -529,6 +532,7 @@ const char *itti_get_task_name(task_id_t task_id);
**/
MessageDef *itti_alloc_new_message(
task_id_t origin_task_id,
instance_t originInstance,
MessagesIds message_id);
/** \brief Alloc and memset(0) a new itti message.
@@ -539,6 +543,7 @@ MessageDef *itti_alloc_new_message(
**/
MessageDef *itti_alloc_new_message_sized(
task_id_t origin_task_id,
instance_t originInstance,
MessagesIds message_id,
MessageHeaderSize size);
@@ -546,7 +551,7 @@ MessageDef *itti_alloc_new_message_sized(
This function should be called from the main thread after having created all ITTI tasks.
**/
void itti_wait_tasks_end(void);
#define THREAD_MAX 0 //for compatibility
#define ADDED_QUEUES_MAX 10 // Fix me: MSC component too hard to understand, we keep room for 10 queues created dynamically
void itti_set_task_real_time(task_id_t task_id);
/** \brief Send a termination message to all tasks.
@@ -560,13 +565,12 @@ void *malloc_or_fail(size_t size);
int memory_read(const char *datafile, void *data, size_t size);
int itti_free(task_id_t task_id, void *ptr);
int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_id_max, const task_info_t *tasks_info,
const message_info_t *messages_info);
int itti_init(task_id_t task_max, const task_info_t *tasks_info);
int timer_setup(
uint32_t interval_sec,
uint32_t interval_us,
task_id_t task_id,
int32_t instance,
instance_t instance,
timer_type_t type,
void *timer_arg,
long *timer_id);