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

itti dynamic queue creation, and 64 bits instance id

Browse Source
This commit is contained in:
Laurent
2020-12-23 13:56:01 +01:00
parent d23a7e83f6
commit 42a110338c
183 changed files with 718 additions and 12086 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/ocp_itti/intertask_interface.cpp
+75 -63
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);
task_list_t *tasks;
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) {
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,18 +145,20 @@ 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) {
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;
}
@@ -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,
@@ -295,8 +302,7 @@ 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];
pthread_mutex_lock(&t->queue_cond_lock);
@@ -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)
{
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);
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,38 @@ 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
}
void itti_create_queue(const task_info_t *task_info) {
pthread_mutex_lock (&lock_nb_queues);
int newQueue=nb_queues;
nb_queues++;
AssertFatal(reallocarray(tasks, nb_queues, sizeof(*tasks)),"no memory");
pthread_mutex_unlock (&lock_nb_queues);
LOG_I(TMR,"Starting itti queue: %s as task %d\n", tasks_info->name, newQueue);
pthread_mutex_init(&tasks[newQueue].queue_cond_lock, NULL);
memcpy(&tasks[newQueue].admin, tasks_info, 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);
}
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");
MessagesIds messages_id_max,
const task_info_t *tasks_info,
const message_info_t *messages_info) {
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_info[i]);
}
return 0;
@@ -359,11 +371,10 @@ 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)
{
long *timer_id) {
task_list_t *t=&tasks[task_id];
do {
@@ -397,8 +408,7 @@ 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);
@@ -439,5 +449,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;
}
}