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

Removed non ENABLE_EVENT_FD option in ITTI.

Browse Source 
Modified message processing for RT task when RTAI is enabled.
Clean-up logs format in RRC.
Fixed some warnings.

pre-ci Ok.

git-svn-id: http://svn.eurecom.fr/openair4G/trunk@4550 818b1a75-f10b-46b9-bf7c-635c3b92a50f
This commit is contained in:
winckel
2013-11-29 16:06:30 +00:00
parent 9a30eabe52
commit cd22e69f5e
7 changed files with 225 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/itti/intertask_interface.c
+168 -233
View File
@@ -41,14 +41,11 @@
# include <rtai_fifos.h>
#endif
#include "queue.h"
#include "assertions.h"
#if defined(ENABLE_EVENT_FD)
# include <sys/epoll.h>
# include <sys/eventfd.h>
# include "liblfds611.h"
#endif
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include "liblfds611.h"
#include "intertask_interface.h"
#include "intertask_interface_dump.h"
@@ -92,10 +89,6 @@ typedef enum task_state_s {
/* This list acts as a FIFO of messages received by tasks (RRC, NAS, ...) */
typedef struct message_list_s {
#if !defined(ENABLE_EVENT_FD)
STAILQ_ENTRY(message_list_s) next_element;
#endif
MessageDef *msg; ///< Pointer to the message
message_number_t message_number; ///< Unique message number
@@ -108,26 +101,11 @@ typedef struct thread_desc_s {
/* State of the thread */
volatile task_state_t task_state;
} thread_desc_t;
typedef struct task_desc_s {
/* Queue of messages belonging to the task */
#if !defined(ENABLE_EVENT_FD)
STAILQ_HEAD(message_queue_head, message_list_s) message_queue;
/* Number of messages in the queue */
volatile uint32_t message_in_queue;
/* Mutex for the message queue */
pthread_mutex_t message_queue_mutex;
/* Conditional var for message queue and task synchro */
pthread_cond_t message_queue_cond_var;
#else
struct lfds611_queue_state *message_queue;
/* This fd is used internally by ITTI. */
int epoll_fd;
/* The task fd */
/* The thread fd */
int task_event_fd;
/* Number of events to monitor */
@@ -141,7 +119,19 @@ typedef struct task_desc_s {
struct epoll_event *events;
int epoll_nb_events;
#ifdef RTAI
/* Flag to mark real time thread */
unsigned real_time;
/* Counter to indicate from RTAI threads that messages are pending for the thread */
unsigned messages_pending;
#endif
} thread_desc_t;
typedef struct task_desc_s {
/* Queue of messages belonging to the task */
struct lfds611_queue_state *message_queue;
} task_desc_t;
typedef struct itti_desc_s {
@@ -161,6 +151,11 @@ typedef struct itti_desc_s {
const message_info_t *messages_info;
itti_lte_time_t lte_time;
int running;
#ifdef RTAI
pthread_t rt_relay_thread;
#endif
} itti_desc_t;
static itti_desc_t itti_desc;
@@ -192,7 +187,7 @@ const char *itti_get_task_name(task_id_t task_id)
return (itti_desc.tasks_info[task_id].name);
}
static task_id_t itti_get_current_task_id()
static task_id_t itti_get_current_task_id(void)
{
task_id_t task_id;
thread_id_t thread_id;
@@ -280,9 +275,9 @@ inline MessageDef *itti_alloc_new_message(task_id_t origin_task_id, MessagesIds
return itti_alloc_new_message_sized(origin_task_id, message_id, itti_desc.messages_info[message_id].size);
}
int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message)
int itti_send_msg_to_task(task_id_t destination_task_id, instance_t instance, MessageDef *message)
{
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
thread_id_t destination_thread_id;
thread_id_t origin_task_id;
message_list_t *new;
uint32_t priority;
@@ -290,9 +285,10 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
uint32_t message_id;
DevAssert(message != NULL);
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(destination_task_id < itti_desc.task_max, destination_task_id, itti_desc.task_max, 0);
message->ittiMsgHeader.destinationTaskId = task_id;
destination_thread_id = TASK_GET_THREAD_ID(destination_task_id);
message->ittiMsgHeader.destinationTaskId = destination_task_id;
message->ittiMsgHeader.instance = instance;
message->ittiMsgHeader.lte_time.frame = itti_desc.lte_time.frame;
message->ittiMsgHeader.lte_time.slot = itti_desc.lte_time.slot;
@@ -303,7 +299,7 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
#if defined(OAI_EMU) || defined(RTAI)
vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_SEND_MSG,
task_id);
destination_task_id);
#endif
priority = itti_get_message_priority (message_id);
@@ -311,37 +307,23 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
/* Increment the global message number */
message_number = itti_increment_message_number ();
#ifdef RTAI
/*
*
#ifdef RTAI
if ((pthread_self() == itti_desc.threads[TASK_GET_THREAD_ID(origin_task_id)].task_thread) ||
(task_id == TASK_UNKNOWN) ||
((TASK_GET_PARENT_TASK_ID(origin_task_id) != TASK_UNKNOWN) &&
(pthread_self() == itti_desc.threads[TASK_GET_PARENT_TASK_ID(origin_task_id)].task_thread)))
#endif
itti_dump_queue_message (message_number, message, itti_desc.messages_info[message_id].name,
sizeof(MessageHeader) + message->ittiMsgHeader.ittiMsgSize);
*/
itti_dump_queue_message (message_number, message, itti_desc.messages_info[message_id].name,
sizeof(MessageHeader) + message->ittiMsgHeader.ittiMsgSize);
ITTI_DEBUG("Message %s, number %lu with priority %d successfully sent from %s to queue (%u:%s)\n",
itti_desc.messages_info[message_id].name,
message_number,
priority,
itti_get_task_name(origin_task_id),
task_id,
itti_get_task_name(task_id));
if (task_id != TASK_UNKNOWN)
if (destination_task_id != TASK_UNKNOWN)
{
/* We cannot send a message if the task is not running */
DevCheck(itti_desc.threads[thread_id].task_state == TASK_STATE_READY, itti_desc.threads[thread_id].task_state,
TASK_STATE_READY, thread_id);
#if !defined(ENABLE_EVENT_FD)
/* Lock the mutex to get exclusive access to the list */
pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
/* Check the number of messages in the queue */
DevCheck(itti_desc.tasks[task_id].message_in_queue < itti_desc.tasks_info[task_id].queue_size,
task_id, itti_desc.tasks[task_id].message_in_queue, itti_desc.tasks_info[task_id].queue_size);
#endif
DevCheck(itti_desc.threads[destination_thread_id].task_state == TASK_STATE_READY, itti_desc.threads[destination_thread_id].task_state,
TASK_STATE_READY, destination_thread_id);
/* Allocate new list element */
new = (message_list_t *) malloc (sizeof(struct message_list_s));
@@ -352,101 +334,68 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
new->message_number = message_number;
new->message_priority = priority;
#if defined(ENABLE_EVENT_FD)
# ifdef RTAI
if ((pthread_self() != itti_desc.threads[TASK_GET_THREAD_ID(origin_task_id)].task_thread) &&
(TASK_GET_PARENT_TASK_ID(origin_task_id) != TASK_UNKNOWN))
/* Enqueue message in destination task queue */
lfds611_queue_enqueue(itti_desc.tasks[destination_task_id].message_queue, new);
#ifdef RTAI
if (itti_desc.threads[TASK_GET_THREAD_ID(origin_task_id)].real_time)
{
/* This is the RT task, -> enqueue in the parent thread */
lfds611_queue_enqueue(itti_desc.tasks[TASK_GET_PARENT_TASK_ID(origin_task_id)].message_queue, new);
/* Signal from RT thread */
// rtf_sem_post(56);
} else
# endif
/* No need to use a event fd for subtasks */
if (TASK_GET_PARENT_TASK_ID(task_id) == TASK_UNKNOWN)
{
ssize_t write_ret;
uint64_t sem_counter = 1;
lfds611_queue_enqueue(itti_desc.tasks[task_id].message_queue, new);
/* Call to write for an event fd must be of 8 bytes */
write_ret = write(itti_desc.tasks[task_id].task_event_fd, &sem_counter, sizeof(sem_counter));
DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, task_id);
} else {
lfds611_queue_enqueue(itti_desc.tasks[task_id].message_queue, new);
/* This is a RT task, increase destination task messages pending counter */
__sync_fetch_and_add (&itti_desc.threads[destination_thread_id].messages_pending, 1);
}
#else
if (STAILQ_EMPTY (&itti_desc.tasks[task_id].message_queue)) {
STAILQ_INSERT_HEAD (&itti_desc.tasks[task_id].message_queue, new, next_element);
}
else {
// struct message_list_s *insert_after = NULL;
// struct message_list_s *temp;
//
// /* This method is inefficient... */
// STAILQ_FOREACH(temp, &itti_desc.tasks[task_id].message_queue, next_element) {
// struct message_list_s *next;
// next = STAILQ_NEXT(temp, next_element);
// /* Increment message priority to create a sort of
// * priority based scheduler */
// // if (temp->message_priority < TASK_PRIORITY_MAX) {
// // temp->message_priority++;
// // }
// if (next && next->message_priority < priority) {
// insert_after = temp;
// break;
// }
// }
// if (insert_after == NULL) {
STAILQ_INSERT_TAIL (&itti_desc.tasks[task_id].message_queue, new, next_element);
// } else {
// STAILQ_INSERT_AFTER(&itti_desc.tasks[task_id].message_queue, insert_after, new,
// next_element);
// }
}
/* Update the number of messages in the queue */
itti_desc.tasks[task_id].message_in_queue++;
if (itti_desc.tasks[task_id].message_in_queue == 1) {
/* Emit a signal to wake up target task thread */
pthread_cond_signal (&itti_desc.tasks[task_id].message_queue_cond_var);
}
/* Release the mutex */
pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
else
#endif
{
/* Only use event fd for tasks, subtasks will pool the queue */
if (TASK_GET_PARENT_TASK_ID(destination_task_id) == TASK_UNKNOWN)
{
ssize_t write_ret;
uint64_t sem_counter = 1;
/* Call to write for an event fd must be of 8 bytes */
write_ret = write (itti_desc.threads[destination_thread_id].task_event_fd, &sem_counter, sizeof(sem_counter));
DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, destination_thread_id);
}
}
ITTI_DEBUG("Message %s, number %lu with priority %d successfully sent from %s to queue (%u:%s)\n",
itti_desc.messages_info[message_id].name,
message_number,
priority,
itti_get_task_name(origin_task_id),
destination_task_id,
itti_get_task_name(destination_task_id));
}
#if defined(OAI_EMU) || defined(RTAI)
vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_SEND_MSG_END,
task_id);
destination_task_id);
#endif
return 0;
}
#if defined(ENABLE_EVENT_FD)
void itti_subscribe_event_fd(task_id_t task_id, int fd)
{
thread_id_t thread_id;
struct epoll_event event;
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(fd >= 0, fd, 0, 0);
itti_desc.tasks[task_id].nb_events++;
thread_id = TASK_GET_THREAD_ID(task_id);
itti_desc.threads[thread_id].nb_events++;
/* Reallocate the events */
itti_desc.tasks[task_id].events = realloc(
itti_desc.tasks[task_id].events,
itti_desc.tasks[task_id].nb_events * sizeof(struct epoll_event));
itti_desc.threads[thread_id].events = realloc(
itti_desc.threads[thread_id].events,
itti_desc.threads[thread_id].nb_events * sizeof(struct epoll_event));
event.events = EPOLLIN | EPOLLERR;
event.data.fd = fd;
/* Add the event fd to the list of monitored events */
if (epoll_ctl(itti_desc.tasks[task_id].epoll_fd, EPOLL_CTL_ADD, fd,
if (epoll_ctl(itti_desc.threads[thread_id].epoll_fd, EPOLL_CTL_ADD, fd,
&event) != 0)
{
ITTI_ERROR("epoll_ctl (EPOLL_CTL_ADD) failed for task %s, fd %d: %s\n",
@@ -460,11 +409,14 @@ void itti_subscribe_event_fd(task_id_t task_id, int fd)
void itti_unsubscribe_event_fd(task_id_t task_id, int fd)
{
thread_id_t thread_id;
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(fd >= 0, fd, 0, 0);
thread_id = TASK_GET_THREAD_ID(task_id);
/* Add the event fd to the list of monitored events */
if (epoll_ctl(itti_desc.tasks[task_id].epoll_fd, EPOLL_CTL_DEL, fd, NULL) != 0)
if (epoll_ctl(itti_desc.threads[thread_id].epoll_fd, EPOLL_CTL_DEL, fd, NULL) != 0)
{
ITTI_ERROR("epoll_ctl (EPOLL_CTL_DEL) failed for task %s and fd %d: %s\n",
itti_get_task_name(task_id), fd, strerror(errno));
@@ -472,23 +424,27 @@ void itti_unsubscribe_event_fd(task_id_t task_id, int fd)
DevAssert(0 == 1);
}
itti_desc.tasks[task_id].nb_events--;
itti_desc.tasks[task_id].events = realloc(
itti_desc.tasks[task_id].events,
itti_desc.tasks[task_id].nb_events * sizeof(struct epoll_event));
itti_desc.threads[thread_id].nb_events--;
itti_desc.threads[thread_id].events = realloc(
itti_desc.threads[thread_id].events,
itti_desc.threads[thread_id].nb_events * sizeof(struct epoll_event));
}
int itti_get_events(task_id_t task_id, struct epoll_event **events)
{
thread_id_t thread_id;
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
*events = itti_desc.tasks[task_id].events;
thread_id = TASK_GET_THREAD_ID(task_id);
*events = itti_desc.threads[thread_id].events;
return itti_desc.tasks[task_id].epoll_nb_events;
return itti_desc.threads[thread_id].epoll_nb_events;
}
static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t polling, MessageDef **received_msg)
{
thread_id_t thread_id;
int epoll_ret = 0;
int epoll_timeout = 0;
int i;
@@ -496,6 +452,7 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevAssert(received_msg != NULL);
thread_id = TASK_GET_THREAD_ID(task_id);
*received_msg = NULL;
if (polling) {
@@ -510,9 +467,9 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
}
do {
epoll_ret = epoll_wait(itti_desc.tasks[task_id].epoll_fd,
itti_desc.tasks[task_id].events,
itti_desc.tasks[task_id].nb_events,
epoll_ret = epoll_wait(itti_desc.threads[thread_id].epoll_fd,
itti_desc.threads[thread_id].events,
itti_desc.threads[thread_id].nb_events,
epoll_timeout);
} while (epoll_ret < 0 && errno == EINTR);
@@ -526,19 +483,19 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
return;
}
itti_desc.tasks[task_id].epoll_nb_events = epoll_ret;
itti_desc.threads[thread_id].epoll_nb_events = epoll_ret;
for (i = 0; i < epoll_ret; i++) {
/* Check if there is an event for ITTI for the event fd */
if ((itti_desc.tasks[task_id].events[i].events & EPOLLIN) &&
(itti_desc.tasks[task_id].events[i].data.fd == itti_desc.tasks[task_id].task_event_fd))
if ((itti_desc.threads[thread_id].events[i].events & EPOLLIN) &&
(itti_desc.threads[thread_id].events[i].data.fd == itti_desc.threads[thread_id].task_event_fd))
{
struct message_list_s *message;
uint64_t sem_counter;
ssize_t read_ret;
/* Read will always return 1 */
read_ret = read (itti_desc.tasks[task_id].task_event_fd, &sem_counter, sizeof(sem_counter));
read_ret = read (itti_desc.threads[thread_id].task_event_fd, &sem_counter, sizeof(sem_counter));
DevCheck(read_ret == sizeof(sem_counter), read_ret, sizeof(sem_counter), 0);
if (lfds611_queue_dequeue (itti_desc.tasks[task_id].message_queue, (void **) &message) == 0) {
@@ -551,7 +508,6 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
}
}
}
#endif
void itti_receive_msg(task_id_t task_id, MessageDef **received_msg)
{
@@ -559,39 +515,9 @@ void itti_receive_msg(task_id_t task_id, MessageDef **received_msg)
vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_RECV_MSG,
task_id);
#endif
#if defined(ENABLE_EVENT_FD)
itti_receive_msg_internal_event_fd(task_id, 0, received_msg);
#else
DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevAssert(received_msg != NULL);
// Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
if (itti_desc.tasks[task_id].message_in_queue == 0) {
ITTI_DEBUG("Message in queue[(%u:%s)] == 0, waiting\n", task_id, itti_get_task_name(task_id));
// Wait while list == 0
pthread_cond_wait (&itti_desc.tasks[task_id].message_queue_cond_var,
&itti_desc.tasks[task_id].message_queue_mutex);
ITTI_DEBUG("Receiver queue[(%u:%s)] got new message notification\n",
task_id, itti_get_task_name(task_id));
}
if (!STAILQ_EMPTY (&itti_desc.tasks[task_id].message_queue)) {
message_list_t *temp = STAILQ_FIRST (&itti_desc.tasks[task_id].message_queue);
/* Update received_msg reference */
*received_msg = temp->msg;
/* Remove message from queue */
STAILQ_REMOVE_HEAD (&itti_desc.tasks[task_id].message_queue, next_element);
free (temp);
itti_desc.tasks[task_id].message_in_queue--;
}
// Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif
#if defined(OAI_EMU) || defined(RTAI)
#if defined(OAI_EMU) || defined(RTAI)
vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_RECV_MSG_END,
task_id);
#endif
@@ -608,7 +534,6 @@ void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
task_id);
#endif
#if defined(ENABLE_EVENT_FD)
{
struct message_list_s *message;
@@ -618,33 +543,6 @@ void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
free (message);
}
}
#else
if (itti_desc.tasks[task_id].message_in_queue != 0) {
message_list_t *temp;
// Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
STAILQ_FOREACH (temp, &itti_desc.tasks[task_id].message_queue, next_element)
{
/* Update received_msg reference */
*received_msg = temp->msg;
/* Remove message from queue */
STAILQ_REMOVE (&itti_desc.tasks[task_id].message_queue, temp, message_list_s, next_element);
free (temp);
itti_desc.tasks[task_id].message_in_queue--;
ITTI_DEBUG(
"Receiver queue[(%u:%s)] got new message %s, number %lu\n",
task_id, itti_get_task_name(task_id), itti_desc.messages_info[temp->msg->ittiMsgHeader.messageId].name, temp->message_number);
break;
}
// Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
}
#endif
if ((itti_debug_poll) && (*received_msg == NULL)) {
ITTI_DEBUG("No message in queue[(%u:%s)]\n", task_id, itti_get_task_name(task_id));
@@ -678,6 +576,17 @@ int itti_create_task(task_id_t task_id, void *(*start_routine)(void *), void *ar
return 0;
}
#ifdef RTAI
void itti_set_task_real_time(task_id_t task_id)
{
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
itti_desc.threads[thread_id].real_time = TRUE;
}
#endif
void itti_mark_task_ready(task_id_t task_id)
{
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
@@ -696,20 +605,10 @@ void itti_mark_task_ready(task_id_t task_id)
/* Register the thread in itti dump */
itti_dump_thread_use_ring_buffer();
#if defined(ENABLE_EVENT_FD)
/* Mark the thread as using LFDS queue */
lfds611_queue_use(itti_desc.tasks[task_id].message_queue);
#else
// Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif
itti_desc.threads[thread_id].task_state = TASK_STATE_READY;
#if !defined(ENABLE_EVENT_FD)
// Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif
}
void itti_exit_task(void) {
@@ -727,6 +626,40 @@ void itti_terminate_tasks(task_id_t task_id) {
pthread_exit (NULL);
}
#ifdef RTAI
static void *itti_rt_relay_thread(void *arg)
{
thread_id_t thread_id;
unsigned pending_messages;
while (itti_desc.running)
{
usleep (100);
/* Checks for all non real time tasks if they have pending messages */
for (thread_id = THREAD_FIRST; thread_id < itti_desc.thread_max; thread_id++)
{
if ((itti_desc.threads[thread_id].task_state == TASK_STATE_READY)
&& (itti_desc.threads[thread_id].real_time == FALSE))
{
pending_messages = __sync_fetch_and_and (&itti_desc.threads[thread_id].messages_pending, 0);
if (pending_messages > 0)
{
ssize_t write_ret;
uint64_t sem_counter = pending_messages;
/* Call to write for an event fd must be of 8 bytes */
write_ret = write (itti_desc.threads[thread_id].task_event_fd, &sem_counter, sizeof(sem_counter));
DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, thread_id);
}
}
}
}
return NULL;
}
#endif
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, const char * const messages_definition_xml, const char * const dump_file_name) {
task_id_t task_id;
@@ -768,7 +701,6 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i
itti_desc.tasks_info[task_id].parent_task != TASK_UNKNOWN ?
itti_get_task_name(itti_desc.tasks_info[task_id].parent_task) : "");
#if defined(ENABLE_EVENT_FD)
ITTI_DEBUG("Creating queue of message of size %u\n", itti_desc.tasks_info[task_id].queue_size);
ret = lfds611_queue_new(&itti_desc.tasks[task_id].message_queue, itti_desc.tasks_info[task_id].queue_size);
@@ -784,32 +716,38 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i
ret = rtf_sem_init(56, 0);
}
# endif
}
itti_desc.tasks[task_id].epoll_fd = epoll_create1(0);
if (itti_desc.tasks[task_id].epoll_fd == -1) {
/* Initializing each thread */
for (thread_id = THREAD_FIRST; thread_id < itti_desc.thread_max; thread_id++)
{
itti_desc.threads[thread_id].task_state = TASK_STATE_NOT_CONFIGURED;
itti_desc.threads[thread_id].epoll_fd = epoll_create1(0);
if (itti_desc.threads[thread_id].epoll_fd == -1) {
ITTI_ERROR("Failed to create new epoll fd: %s\n", strerror(errno));
/* Always assert on this condition */
DevAssert(0 == 1);
}
itti_desc.tasks[task_id].task_event_fd = eventfd(0, EFD_SEMAPHORE);
if (itti_desc.tasks[task_id].task_event_fd == -1)
itti_desc.threads[thread_id].task_event_fd = eventfd(0, EFD_SEMAPHORE);
if (itti_desc.threads[thread_id].task_event_fd == -1)
{
ITTI_ERROR("eventfd failed: %s\n", strerror(errno));
/* Always assert on this condition */
DevAssert(0 == 1);
}
itti_desc.tasks[task_id].nb_events = 1;
itti_desc.threads[thread_id].nb_events = 1;
itti_desc.tasks[task_id].events = calloc(1, sizeof(struct epoll_event));
itti_desc.threads[thread_id].events = calloc(1, sizeof(struct epoll_event));
itti_desc.tasks[task_id].events->events = EPOLLIN | EPOLLERR;
itti_desc.tasks[task_id].events->data.fd = itti_desc.tasks[task_id].task_event_fd;
itti_desc.threads[thread_id].events->events = EPOLLIN | EPOLLERR;
itti_desc.threads[thread_id].events->data.fd = itti_desc.threads[thread_id].task_event_fd;
/* Add the event fd to the list of monitored events */
if (epoll_ctl(itti_desc.tasks[task_id].epoll_fd, EPOLL_CTL_ADD,
itti_desc.tasks[task_id].task_event_fd, itti_desc.tasks[task_id].events) != 0)
if (epoll_ctl(itti_desc.threads[thread_id].epoll_fd, EPOLL_CTL_ADD,
itti_desc.threads[thread_id].task_event_fd, itti_desc.threads[thread_id].events) != 0)
{
ITTI_ERROR("epoll_ctl (EPOLL_CTL_ADD) failed: %s\n", strerror(errno));
/* Always assert on this condition */
@@ -817,24 +755,19 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i
}
ITTI_DEBUG("Successfully subscribed fd %d for task %s\n",
itti_desc.tasks[task_id].task_event_fd, itti_get_task_name(task_id));
#else
STAILQ_INIT (&itti_desc.tasks[task_id].message_queue);
itti_desc.tasks[task_id].message_in_queue = 0;
itti_desc.threads[thread_id].task_event_fd, itti_get_task_name(task_id));
// Initialize mutexes
pthread_mutex_init (&itti_desc.tasks[task_id].message_queue_mutex, NULL);
// Initialize Cond vars
pthread_cond_init (&itti_desc.tasks[task_id].message_queue_cond_var, NULL);
#ifdef RTAI
itti_desc.threads[thread_id].real_time = FALSE;
itti_desc.threads[thread_id].messages_pending = 0;
#endif
}
/* Initializing each thread */
for (thread_id = THREAD_FIRST; thread_id < itti_desc.thread_max; thread_id++)
{
itti_desc.threads[thread_id].task_state = TASK_STATE_NOT_CONFIGURED;
}
itti_desc.running = TRUE;
#ifdef RTAI
/* Start RT relay thread */
DevAssert(pthread_create (&itti_desc.rt_relay_thread, NULL, itti_rt_relay_thread, NULL) >= 0);
#endif
itti_dump_init (messages_definition_xml, dump_file_name);
@@ -891,6 +824,8 @@ void itti_wait_tasks_end(void) {
}
} while ((ready_tasks > 0) && (retries--));
itti_desc.running = FALSE;
if (ready_tasks > 0) {
ITTI_DEBUG("Some threads are still running, force exit\n");
exit (0);