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

Modified ITTI to create one message queue for each task and sub task.

Browse Source 
Still only one thread is created by task.

pre-ci ok.

git-svn-id: http://svn.eurecom.fr/openair4G/trunk@4350 818b1a75-f10b-46b9-bf7c-635c3b92a50f
This commit is contained in:
winckel
2013-11-08 09:43:03 +00:00
parent cc7f377552
commit cefa14650f
13 changed files with 168 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/itti/intertask_interface.c
+134 -123
View File
@@ -85,6 +85,13 @@ struct message_list_s {
uint32_t message_priority; ///< Message priority uint32_t message_priority; ///< Message priority
}; };
typedef struct thread_desc_s {
/* pthread associated with the thread */
pthread_t task_thread;
/* State of the thread */
volatile task_state_t task_state;
} thread_desc_t;
typedef struct task_desc_s { typedef struct task_desc_s {
/* Queue of messages belonging to the task */ /* Queue of messages belonging to the task */
#if !defined(ENABLE_EVENT_FD) #if !defined(ENABLE_EVENT_FD)
@@ -117,23 +124,22 @@ typedef struct task_desc_s {
int epoll_nb_events; int epoll_nb_events;
#endif #endif
/* pthread associated with the task */
pthread_t task_thread;
/* State of the task */
volatile task_state_t task_state;
} task_desc_t; } task_desc_t;
struct itti_desc_s { struct itti_desc_s {
thread_desc_t *threads;
task_desc_t *tasks; task_desc_t *tasks;
/* Current message number. Incremented every call to send_msg_to_task */ /* Current message number. Incremented every call to send_msg_to_task */
message_number_t message_number __attribute__((aligned(8))); message_number_t message_number __attribute__((aligned(8)));
thread_id_t thread_max; thread_id_t thread_max;
task_id_t task_max;
MessagesIds messages_id_max; MessagesIds messages_id_max;
pthread_t thread_handling_signals; pthread_t thread_handling_signals;
const char * const *threads_name; const task_info_t *tasks_info;
const message_info_t *messages_info; const message_info_t *messages_info;
}; };
@@ -161,16 +167,15 @@ const char *itti_get_message_name(MessagesIds message_id) {
const char *itti_get_task_name(task_id_t task_id) const char *itti_get_task_name(task_id_t task_id)
{ {
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0); return (itti_desc.tasks_info[task_id].name);
return (itti_desc.threads_name[thread_id]);
} }
int itti_send_broadcast_message(MessageDef *message_p) { int itti_send_broadcast_message(MessageDef *message_p) {
task_id_t destination_task_id;
thread_id_t origin_thread_id; thread_id_t origin_thread_id;
uint32_t i; uint32_t thread_id;
int ret = 0; int ret = 0;
int result; int result;
@@ -178,19 +183,24 @@ int itti_send_broadcast_message(MessageDef *message_p) {
origin_thread_id = TASK_GET_THREAD_ID(message_p->header.originTaskId); origin_thread_id = TASK_GET_THREAD_ID(message_p->header.originTaskId);
for (i = THREAD_FIRST; i < itti_desc.thread_max; i++) { destination_task_id = TASK_FIRST;
for (thread_id = THREAD_FIRST; thread_id < itti_desc.thread_max; thread_id++) {
MessageDef *new_message_p; MessageDef *new_message_p;
while (thread_id != TASK_GET_THREAD_ID(destination_task_id))
{
destination_task_id++;
}
/* Skip task that broadcast the message */ /* Skip task that broadcast the message */
if (i != origin_thread_id) { if (thread_id != origin_thread_id) {
/* Skip tasks which are not running */ /* Skip tasks which are not running */
if (itti_desc.tasks[i].task_state == TASK_STATE_READY) { if (itti_desc.threads[thread_id].task_state == TASK_STATE_READY) {
new_message_p = malloc (sizeof(MessageDef)); new_message_p = malloc (sizeof(MessageDef));
DevAssert(message_p != NULL); DevAssert(message_p != NULL);
memcpy (new_message_p, message_p, sizeof(MessageDef)); memcpy (new_message_p, message_p, sizeof(MessageDef));
result = itti_send_msg_to_task (TASK_SHIFT_THREAD_ID(i), INSTANCE_DEFAULT, new_message_p); result = itti_send_msg_to_task (destination_task_id, INSTANCE_DEFAULT, new_message_p);
DevCheck(result >= 0, message_p->header.messageId, i, 0); DevCheck(result >= 0, message_p->header.messageId, thread_id, destination_task_id);
} }
} }
} }
@@ -222,7 +232,7 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
uint32_t message_id; uint32_t message_id;
DevAssert(message != NULL); DevAssert(message != NULL);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
message->header.destinationTaskId = task_id; message->header.destinationTaskId = task_id;
message->header.instance = instance; message->header.instance = instance;
@@ -232,17 +242,17 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
priority = itti_get_message_priority (message_id); priority = itti_get_message_priority (message_id);
/* We cannot send a message if the task is not running */ /* We cannot send a message if the task is not running */
DevCheck(itti_desc.tasks[thread_id].task_state == TASK_STATE_READY, itti_desc.tasks[thread_id].task_state, DevCheck(itti_desc.threads[thread_id].task_state == TASK_STATE_READY, itti_desc.threads[thread_id].task_state,
TASK_STATE_READY, thread_id); TASK_STATE_READY, thread_id);
#if !defined(ENABLE_EVENT_FD) #if !defined(ENABLE_EVENT_FD)
/* Lock the mutex to get exclusive access to the list */ /* Lock the mutex to get exclusive access to the list */
pthread_mutex_lock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
/* Check the number of messages in the queue */ /* Check the number of messages in the queue */
DevCheck((itti_desc.tasks[thread_id].message_in_queue * sizeof(MessageDef)) < ITTI_QUEUE_SIZE_PER_TASK, DevCheck((itti_desc.tasks[task_id].message_in_queue * sizeof(MessageDef)) < ITTI_QUEUE_SIZE_PER_TASK,
(itti_desc.tasks[thread_id].message_in_queue * sizeof(MessageDef)), ITTI_QUEUE_SIZE_PER_TASK, (itti_desc.tasks[task_id].message_in_queue * sizeof(MessageDef)), ITTI_QUEUE_SIZE_PER_TASK,
itti_desc.tasks[thread_id].message_in_queue); itti_desc.tasks[task_id].message_in_queue);
#endif #endif
/* Allocate new list element */ /* Allocate new list element */
@@ -264,21 +274,21 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
{ {
uint64_t sem_counter = 1; uint64_t sem_counter = 1;
lfds611_queue_enqueue(itti_desc.tasks[thread_id].message_queue, new); lfds611_queue_enqueue(itti_desc.tasks[task_id].message_queue, new);
/* Call to write for an event fd must be of 8 bytes */ /* Call to write for an event fd must be of 8 bytes */
write(itti_desc.tasks[thread_id].task_event_fd, &sem_counter, sizeof(sem_counter)); write(itti_desc.tasks[task_id].task_event_fd, &sem_counter, sizeof(sem_counter));
} }
#else #else
if (STAILQ_EMPTY (&itti_desc.tasks[thread_id].message_queue)) { if (STAILQ_EMPTY (&itti_desc.tasks[task_id].message_queue)) {
STAILQ_INSERT_HEAD (&itti_desc.tasks[thread_id].message_queue, new, next_element); STAILQ_INSERT_HEAD (&itti_desc.tasks[task_id].message_queue, new, next_element);
} }
else { else {
// struct message_list_s *insert_after = NULL; // struct message_list_s *insert_after = NULL;
// struct message_list_s *temp; // struct message_list_s *temp;
// //
// /* This method is inefficient... */ // /* This method is inefficient... */
// STAILQ_FOREACH(temp, &itti_desc.tasks[thread_id].message_queue, next_element) { // STAILQ_FOREACH(temp, &itti_desc.tasks[task_id].message_queue, next_element) {
// struct message_list_s *next; // struct message_list_s *next;
// next = STAILQ_NEXT(temp, next_element); // next = STAILQ_NEXT(temp, next_element);
// /* Increment message priority to create a sort of // /* Increment message priority to create a sort of
@@ -292,26 +302,26 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
// } // }
// } // }
// if (insert_after == NULL) { // if (insert_after == NULL) {
STAILQ_INSERT_TAIL (&itti_desc.tasks[thread_id].message_queue, new, next_element); STAILQ_INSERT_TAIL (&itti_desc.tasks[task_id].message_queue, new, next_element);
// } else { // } else {
// STAILQ_INSERT_AFTER(&itti_desc.tasks[thread_id].message_queue, insert_after, new, // STAILQ_INSERT_AFTER(&itti_desc.tasks[task_id].message_queue, insert_after, new,
// next_element); // next_element);
// } // }
} }
/* Update the number of messages in the queue */ /* Update the number of messages in the queue */
itti_desc.tasks[thread_id].message_in_queue++; itti_desc.tasks[task_id].message_in_queue++;
if (itti_desc.tasks[thread_id].message_in_queue == 1) { if (itti_desc.tasks[task_id].message_in_queue == 1) {
/* Emit a signal to wake up target task thread */ /* Emit a signal to wake up target task thread */
pthread_cond_signal (&itti_desc.tasks[thread_id].message_queue_cond_var); pthread_cond_signal (&itti_desc.tasks[task_id].message_queue_cond_var);
} }
/* Release the mutex */ /* Release the mutex */
pthread_mutex_unlock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif #endif
ITTI_DEBUG( ITTI_DEBUG(
"Message %s, number %lu with priority %d successfully sent to queue (%u:%s)\n", "Message %s, number %lu with priority %d successfully sent to queue (%u:%s)\n",
itti_desc.messages_info[message_id].name, message_number, priority, thread_id, itti_desc.threads_name[thread_id]); itti_desc.messages_info[message_id].name, message_number, priority, task_id, itti_get_task_name(task_id));
return 0; return 0;
} }
@@ -319,23 +329,22 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
void itti_subscribe_event_fd(task_id_t task_id, int fd) void itti_subscribe_event_fd(task_id_t task_id, int fd)
{ {
struct epoll_event event; struct epoll_event event;
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); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(fd >= 0, fd, 0, 0); DevCheck(fd >= 0, fd, 0, 0);
itti_desc.tasks[thread_id].nb_events++; itti_desc.tasks[task_id].nb_events++;
/* Reallocate the events */ /* Reallocate the events */
itti_desc.tasks[thread_id].events = realloc( itti_desc.tasks[task_id].events = realloc(
itti_desc.tasks[thread_id].events, itti_desc.tasks[task_id].events,
itti_desc.tasks[thread_id].nb_events * sizeof(struct epoll_event)); itti_desc.tasks[task_id].nb_events * sizeof(struct epoll_event));
event.events = EPOLLIN; event.events = EPOLLIN;
event.data.fd = fd; event.data.fd = fd;
/* Add the event fd to the list of monitored events */ /* Add the event fd to the list of monitored events */
if (epoll_ctl(itti_desc.tasks[thread_id].epoll_fd, EPOLL_CTL_ADD, fd, if (epoll_ctl(itti_desc.tasks[task_id].epoll_fd, EPOLL_CTL_ADD, fd,
&event) != 0) &event) != 0)
{ {
ITTI_ERROR("epoll_ctl (EPOLL_CTL_ADD) failed for task %s, fd %d: %s\n", ITTI_ERROR("epoll_ctl (EPOLL_CTL_ADD) failed for task %s, fd %d: %s\n",
@@ -347,13 +356,11 @@ void itti_subscribe_event_fd(task_id_t task_id, int fd)
void itti_unsubscribe_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 = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
DevCheck(fd >= 0, fd, 0, 0); DevCheck(fd >= 0, fd, 0, 0);
/* Add the event fd to the list of monitored events */ /* Add the event fd to the list of monitored events */
if (epoll_ctl(itti_desc.tasks[thread_id].epoll_fd, EPOLL_CTL_DEL, fd, NULL) != 0) if (epoll_ctl(itti_desc.tasks[task_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_ERROR("epoll_ctl (EPOLL_CTL_DEL) failed for task %s and fd %d: %s\n",
itti_get_task_name(task_id), fd, strerror(errno)); itti_get_task_name(task_id), fd, strerror(errno));
@@ -361,21 +368,19 @@ void itti_unsubscribe_event_fd(task_id_t task_id, int fd)
DevAssert(0 == 1); DevAssert(0 == 1);
} }
itti_desc.tasks[thread_id].nb_events--; itti_desc.tasks[task_id].nb_events--;
itti_desc.tasks[thread_id].events = realloc( itti_desc.tasks[task_id].events = realloc(
itti_desc.tasks[thread_id].events, itti_desc.tasks[task_id].events,
itti_desc.tasks[thread_id].nb_events * sizeof(struct epoll_event)); itti_desc.tasks[task_id].nb_events * sizeof(struct epoll_event));
} }
int itti_get_events(task_id_t task_id, struct epoll_event **events) int itti_get_events(task_id_t task_id, struct epoll_event **events)
{ {
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0); *events = itti_desc.tasks[task_id].events;
*events = itti_desc.tasks[thread_id].events; return itti_desc.tasks[task_id].epoll_nb_events;
return itti_desc.tasks[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) static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t polling, MessageDef **received_msg)
@@ -384,9 +389,7 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
int epoll_timeout = 0; int epoll_timeout = 0;
int i; int i;
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
DevAssert(received_msg != NULL); DevAssert(received_msg != NULL);
*received_msg = NULL; *received_msg = NULL;
@@ -403,9 +406,9 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
} }
do { do {
epoll_ret = epoll_wait(itti_desc.tasks[thread_id].epoll_fd, epoll_ret = epoll_wait(itti_desc.tasks[task_id].epoll_fd,
itti_desc.tasks[thread_id].events, itti_desc.tasks[task_id].events,
itti_desc.tasks[thread_id].nb_events, itti_desc.tasks[task_id].nb_events,
epoll_timeout); epoll_timeout);
} while (epoll_ret < 0 && errno == EINTR); } while (epoll_ret < 0 && errno == EINTR);
@@ -419,22 +422,22 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t
return; return;
} }
itti_desc.tasks[thread_id].epoll_nb_events = epoll_ret; itti_desc.tasks[task_id].epoll_nb_events = epoll_ret;
for (i = 0; i < epoll_ret; i++) { for (i = 0; i < epoll_ret; i++) {
/* Check if there is an event for ITTI for the event fd */ /* Check if there is an event for ITTI for the event fd */
if ((itti_desc.tasks[thread_id].events[i].events & EPOLLIN) && if ((itti_desc.tasks[task_id].events[i].events & EPOLLIN) &&
(itti_desc.tasks[thread_id].events[i].data.fd == itti_desc.tasks[thread_id].task_event_fd)) (itti_desc.tasks[task_id].events[i].data.fd == itti_desc.tasks[task_id].task_event_fd))
{ {
struct message_list_s *message; struct message_list_s *message;
uint64_t sem_counter; uint64_t sem_counter;
/* Read will always return 1 */ /* Read will always return 1 */
read (itti_desc.tasks[thread_id].task_event_fd, &sem_counter, sizeof(sem_counter)); read (itti_desc.tasks[task_id].task_event_fd, &sem_counter, sizeof(sem_counter));
if (lfds611_queue_dequeue (itti_desc.tasks[thread_id].message_queue, (void **) &message) == 0) { if (lfds611_queue_dequeue (itti_desc.tasks[task_id].message_queue, (void **) &message) == 0) {
/* No element in list -> this should not happen */ /* No element in list -> this should not happen */
DevParam(thread_id, task_id, epoll_ret); DevParam(task_id, epoll_ret, 0);
} }
*received_msg = message->msg; *received_msg = message->msg;
free (message); free (message);
@@ -449,43 +452,39 @@ void itti_receive_msg(task_id_t task_id, MessageDef **received_msg)
#if defined(ENABLE_EVENT_FD) #if defined(ENABLE_EVENT_FD)
itti_receive_msg_internal_event_fd(task_id, 0, received_msg); itti_receive_msg_internal_event_fd(task_id, 0, received_msg);
#else #else
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
DevAssert(received_msg != NULL); DevAssert(received_msg != NULL);
// Lock the mutex to get exclusive access to the list // Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
if (itti_desc.tasks[thread_id].message_in_queue == 0) { if (itti_desc.tasks[task_id].message_in_queue == 0) {
ITTI_DEBUG("Message in queue[(%u:%s)] == 0, waiting\n", thread_id, itti_desc.threads_name[thread_id]); ITTI_DEBUG("Message in queue[(%u:%s)] == 0, waiting\n", task_id, itti_get_task_name(task_id));
// Wait while list == 0 // Wait while list == 0
pthread_cond_wait (&itti_desc.tasks[thread_id].message_queue_cond_var, pthread_cond_wait (&itti_desc.tasks[task_id].message_queue_cond_var,
&itti_desc.tasks[thread_id].message_queue_mutex); &itti_desc.tasks[task_id].message_queue_mutex);
ITTI_DEBUG("Receiver queue[(%u:%s)] got new message notification for task %x\n", ITTI_DEBUG("Receiver queue[(%u:%s)] got new message notification\n",
thread_id, itti_desc.threads_name[thread_id], task_id); task_id, itti_get_task_name(task_id));
} }
if (!STAILQ_EMPTY (&itti_desc.tasks[thread_id].message_queue)) { if (!STAILQ_EMPTY (&itti_desc.tasks[task_id].message_queue)) {
struct message_list_s *temp = STAILQ_FIRST (&itti_desc.tasks[thread_id].message_queue); struct message_list_s *temp = STAILQ_FIRST (&itti_desc.tasks[task_id].message_queue);
/* Update received_msg reference */ /* Update received_msg reference */
*received_msg = temp->msg; *received_msg = temp->msg;
/* Remove message from queue */ /* Remove message from queue */
STAILQ_REMOVE_HEAD (&itti_desc.tasks[thread_id].message_queue, next_element); STAILQ_REMOVE_HEAD (&itti_desc.tasks[task_id].message_queue, next_element);
free (temp); free (temp);
itti_desc.tasks[thread_id].message_in_queue--; itti_desc.tasks[task_id].message_in_queue--;
} }
// Release the mutex // Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif #endif
} }
void itti_poll_msg(task_id_t task_id, instance_t instance, MessageDef **received_msg) { void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); DevCheck(task_id < itti_desc.task_max, task_id, itti_desc.task_max, 0);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
DevAssert(received_msg != NULL); DevAssert(received_msg != NULL);
*received_msg = NULL; *received_msg = NULL;
@@ -493,38 +492,35 @@ void itti_poll_msg(task_id_t task_id, instance_t instance, MessageDef **received
#if defined(ENABLE_EVENT_FD) #if defined(ENABLE_EVENT_FD)
itti_receive_msg_internal_event_fd(task_id, 1, received_msg); itti_receive_msg_internal_event_fd(task_id, 1, received_msg);
#else #else
if (itti_desc.tasks[thread_id].message_in_queue != 0) { if (itti_desc.tasks[task_id].message_in_queue != 0) {
struct message_list_s *temp; struct message_list_s *temp;
// Lock the mutex to get exclusive access to the list // Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
STAILQ_FOREACH (temp, &itti_desc.tasks[thread_id].message_queue, next_element) STAILQ_FOREACH (temp, &itti_desc.tasks[task_id].message_queue, next_element)
{ {
if ((temp->msg->header.destinationTaskId == task_id) /* Update received_msg reference */
&& ((instance == INSTANCE_ALL) || (temp->msg->header.instance == instance))) { *received_msg = temp->msg;
/* Update received_msg reference */
*received_msg = temp->msg;
/* Remove message from queue */ /* Remove message from queue */
STAILQ_REMOVE (&itti_desc.tasks[thread_id].message_queue, temp, message_list_s, next_element); STAILQ_REMOVE (&itti_desc.tasks[task_id].message_queue, temp, message_list_s, next_element);
free (temp); free (temp);
itti_desc.tasks[thread_id].message_in_queue--; itti_desc.tasks[task_id].message_in_queue--;
ITTI_DEBUG( ITTI_DEBUG(
"Receiver queue[(%u:%s)] got new message %s, number %lu for task %x\n", "Receiver queue[(%u:%s)] got new message %s, number %lu\n",
thread_id, itti_desc.threads_name[thread_id], itti_desc.messages_info[temp->msg->header.messageId].name, temp->message_number, task_id); task_id, itti_get_task_name(task_id), itti_desc.messages_info[temp->msg->header.messageId].name, temp->message_number);
break; break;
}
} }
// Release the mutex // Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
} }
#endif #endif
if (*received_msg == NULL) { if (*received_msg == NULL) {
ITTI_DEBUG("No message in queue[(%u:%s)] for task %x\n", thread_id, itti_desc.threads_name[thread_id], task_id); ITTI_DEBUG("No message in queue[(%u:%s)]\n", task_id, itti_get_task_name(task_id));
} }
} }
@@ -534,16 +530,16 @@ int itti_create_task(task_id_t task_id, void *(*start_routine)(void *), void *ar
DevAssert(start_routine != NULL); DevAssert(start_routine != NULL);
DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0); DevCheck(thread_id < itti_desc.thread_max, thread_id, itti_desc.thread_max, 0);
DevCheck(itti_desc.tasks[thread_id].task_state == TASK_STATE_NOT_CONFIGURED, task_id, thread_id, DevCheck(itti_desc.threads[thread_id].task_state == TASK_STATE_NOT_CONFIGURED, task_id, thread_id,
itti_desc.tasks[thread_id].task_state); itti_desc.threads[thread_id].task_state);
itti_desc.tasks[thread_id].task_state = TASK_STATE_STARTING; itti_desc.threads[thread_id].task_state = TASK_STATE_STARTING;
result = pthread_create (&itti_desc.tasks[thread_id].task_thread, NULL, start_routine, args_p); result = pthread_create (&itti_desc.threads[thread_id].task_thread, NULL, start_routine, args_p);
DevCheck(result>= 0, task_id, thread_id, result); DevCheck(result>= 0, task_id, thread_id, result);
/* Wait till the thread is completely ready */ /* Wait till the thread is completely ready */
while (itti_desc.tasks[thread_id].task_state != TASK_STATE_READY) while (itti_desc.threads[thread_id].task_state != TASK_STATE_READY)
; ;
return 0; return 0;
} }
@@ -555,14 +551,14 @@ void itti_mark_task_ready(task_id_t task_id) {
#if !defined(ENABLE_EVENT_FD) #if !defined(ENABLE_EVENT_FD)
// Lock the mutex to get exclusive access to the list // Lock the mutex to get exclusive access to the list
pthread_mutex_lock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_lock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif #endif
itti_desc.tasks[thread_id].task_state = TASK_STATE_READY; itti_desc.threads[thread_id].task_state = TASK_STATE_READY;
#if !defined(ENABLE_EVENT_FD) #if !defined(ENABLE_EVENT_FD)
// Release the mutex // Release the mutex
pthread_mutex_unlock (&itti_desc.tasks[thread_id].message_queue_mutex); pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
#endif #endif
} }
@@ -581,27 +577,31 @@ void itti_terminate_tasks(task_id_t task_id) {
pthread_exit (NULL); pthread_exit (NULL);
} }
int itti_init(thread_id_t thread_max, MessagesIds messages_id_max, const char * const *threads_name, 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) { const message_info_t *messages_info, const char * const messages_definition_xml, const char * const dump_file_name) {
int i; int i;
itti_desc.message_number = 1; itti_desc.message_number = 1;
ITTI_DEBUG("Init: %d threads, %d messages\n", thread_max, messages_id_max); ITTI_DEBUG("Init: %d tasks, %d threads, %d messages\n", task_max, thread_max, messages_id_max);
CHECK_INIT_RETURN(signal_init()); CHECK_INIT_RETURN(signal_init());
/* Saves threads and messages max values */ /* Saves threads and messages max values */
itti_desc.task_max = task_max;
itti_desc.thread_max = thread_max; itti_desc.thread_max = thread_max;
itti_desc.messages_id_max = messages_id_max; itti_desc.messages_id_max = messages_id_max;
itti_desc.thread_handling_signals = -1; itti_desc.thread_handling_signals = -1;
itti_desc.threads_name = threads_name; itti_desc.tasks_info = tasks_info;
itti_desc.messages_info = messages_info; itti_desc.messages_info = messages_info;
/* Allocates memory for tasks info */ /* Allocates memory for tasks info */
itti_desc.tasks = calloc (itti_desc.thread_max, sizeof(task_desc_t)); itti_desc.tasks = calloc (itti_desc.task_max, sizeof(task_desc_t));
/* Allocates memory for threads info */
itti_desc.threads = calloc (itti_desc.thread_max, sizeof(thread_desc_t));
/* Initializing each queue and related stuff */ /* Initializing each queue and related stuff */
for (i = THREAD_FIRST; i < itti_desc.thread_max; i++) for (i = TASK_FIRST; i < itti_desc.task_max; i++)
{ {
#if defined(ENABLE_EVENT_FD) #if defined(ENABLE_EVENT_FD)
ITTI_DEBUG("Creating queue of message of size %u\n", ITTI_DEBUG("Creating queue of message of size %u\n",
@@ -652,9 +652,14 @@ int itti_init(thread_id_t thread_max, MessagesIds messages_id_max, const char *
// Initialize Cond vars // Initialize Cond vars
pthread_cond_init (&itti_desc.tasks[i].message_queue_cond_var, NULL); pthread_cond_init (&itti_desc.tasks[i].message_queue_cond_var, NULL);
#endif #endif
itti_desc.tasks[i].task_state = TASK_STATE_NOT_CONFIGURED;
} }
/* Initializing each thread */
for (i = THREAD_FIRST; i < itti_desc.thread_max; i++)
{
itti_desc.threads[i].task_state = TASK_STATE_NOT_CONFIGURED;
}
itti_dump_init (messages_definition_xml, dump_file_name); itti_dump_init (messages_definition_xml, dump_file_name);
CHECK_INIT_RETURN(timer_init ()); CHECK_INIT_RETURN(timer_init ());
@@ -664,7 +669,8 @@ int itti_init(thread_id_t thread_max, MessagesIds messages_id_max, const char *
void itti_wait_tasks_end(void) { void itti_wait_tasks_end(void) {
int end = 0; int end = 0;
int i; int thread_id;
task_id_t task_id;
int ready_tasks; int ready_tasks;
int result; int result;
int retries = 10; int retries = 10;
@@ -679,17 +685,22 @@ void itti_wait_tasks_end(void) {
do { do {
ready_tasks = 0; ready_tasks = 0;
for (i = THREAD_FIRST; i < itti_desc.thread_max; i++) { task_id = TASK_FIRST;
for (thread_id = THREAD_FIRST; thread_id < itti_desc.task_max; thread_id++) {
/* Skip tasks which are not running */ /* Skip tasks which are not running */
if (itti_desc.tasks[i].task_state == TASK_STATE_READY) { if (itti_desc.threads[thread_id].task_state == TASK_STATE_READY) {
while (thread_id != TASK_GET_THREAD_ID(task_id))
{
task_id++;
}
result = pthread_tryjoin_np (itti_desc.tasks[i].task_thread, NULL); result = pthread_tryjoin_np (itti_desc.threads[thread_id].task_thread, NULL);
ITTI_DEBUG("Thread %s join status %d\n", itti_desc.threads_name[i], result); ITTI_DEBUG("Thread %s join status %d\n", itti_get_task_name(task_id), result);
if (result == 0) { if (result == 0) {
/* Thread has terminated */ /* Thread has terminated */
itti_desc.tasks[i].task_state = TASK_STATE_ENDED; itti_desc.threads[thread_id].task_state = TASK_STATE_ENDED;
} }
else { else {
/* Thread is still running, count it */ /* Thread is still running, count it */
common/utils/itti/intertask_interface.h
+8 -3
View File
@@ -86,6 +86,12 @@ enum task_priorities {
TASK_PRIORITY_MIN = 10, TASK_PRIORITY_MIN = 10,
}; };
typedef struct task_info_s {
thread_id_t thread;
/* Printable name */
const char * const name;
} task_info_t;
/** \brief Send a broadcast message to every task /** \brief Send a broadcast message to every task
\param message_p Pointer to the message to send \param message_p Pointer to the message to send
@returns < 0 on failure, 0 otherwise @returns < 0 on failure, 0 otherwise
@@ -129,12 +135,11 @@ int itti_get_events(task_id_t task_id, struct epoll_event **events);
**/ **/
void itti_receive_msg(task_id_t task_id, MessageDef **received_msg); void itti_receive_msg(task_id_t task_id, MessageDef **received_msg);
/** \brief Try to retrieves a message in the queue associated to task_id and matching requested instance. /** \brief Try to retrieves a message in the queue associated to task_id.
\param task_id Task ID of the receiving task \param task_id Task ID of the receiving task
\param instance Instance of the task used for virtualization
\param received_msg Pointer to the allocated message \param received_msg Pointer to the allocated message
**/ **/
void itti_poll_msg(task_id_t task_id, instance_t instance, MessageDef **received_msg); void itti_poll_msg(task_id_t task_id, MessageDef **received_msg);
/** \brief Start thread associated to the task /** \brief Start thread associated to the task
* \param task_id task to start * \param task_id task to start
common/utils/itti/intertask_interface_dump.c
+1 -1
View File
@@ -29,7 +29,7 @@
*******************************************************************************/ *******************************************************************************/
/** @brief Intertask Interface Signal Dumper /** @brief Intertask Interface Signal Dumper
* Allows users to connect their itti_debugger to this process and dump * Allows users to connect their itti_analyzer to this process and dump
* signals exchanged between tasks. * signals exchanged between tasks.
* @author Sebastien Roux <sebastien.roux@eurecom.fr> * @author Sebastien Roux <sebastien.roux@eurecom.fr>
*/ */
common/utils/itti/intertask_interface_init.h
+6 -6
View File
@@ -54,11 +54,11 @@ const char * const messages_definition_xml = {
#include "messages_xml.h" #include "messages_xml.h"
}; };
/* Map thread id to printable name. */ /* Map task id to printable name. */
const char * const threads_name[] = { const task_info_t tasks_info[] = {
"unused", {0, "TASK_UNKNOWN"},
#define TASK_DEF(tHREADiD, pRIO) #tHREADiD, #define TASK_DEF(tHREADiD, pRIO) {tHREADiD##_THREAD, #tHREADiD},
#define SUB_TASK_DEF(tHREADiD, sUBtASKiD) #define SUB_TASK_DEF(tHREADiD, sUBtASKiD) {sUBtASKiD##_THREAD, #sUBtASKiD},
#include <tasks_def.h> #include <tasks_def.h>
#undef SUB_TASK_DEF #undef SUB_TASK_DEF
#undef TASK_DEF #undef TASK_DEF
@@ -79,7 +79,7 @@ const message_info_t messages_info[] = {
* \param threads_name Pointer on the threads name information as created by this include file * \param threads_name Pointer on the threads name information as created by this include file
* \param messages_info Pointer on messages information as created by this include file * \param messages_info Pointer on messages information as created by this include file
**/ **/
int itti_init(thread_id_t thread_max, MessagesIds messages_id_max, const char * const *threads_name, 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 message_info_t *messages_info, const char * const messages_definition_xml,
const char * const dump_file_name); const char * const dump_file_name);
common/utils/itti/intertask_interface_types.h
+10 -13
View File
@@ -56,12 +56,7 @@
#define TASK_SUB_TASK_ID_LENGTH 8 #define TASK_SUB_TASK_ID_LENGTH 8
/* Defines to extract task ID fields */ /* Defines to extract task ID fields */
#define TASK_GET_THREAD_ID(tASKiD) UL_FIELD_EXTRACT(tASKiD, TASK_THREAD_ID_OFFSET, TASK_THREAD_ID_LENGTH) #define TASK_GET_THREAD_ID(tASKiD) (itti_desc.tasks_info[tASKiD].thread)
#define TASK_GET_SUB_TASK_ID(tASKiD) UL_FIELD_EXTRACT(tASKiD, TASK_SUB_TASK_ID_OFFSET, TASK_SUB_TASK_ID_LENGTH)
/* Defines to shift task ID fields */
#define TASK_SHIFT_THREAD_ID(tHREADiD) UL_BIT_SHIFT(tHREADiD, TASK_THREAD_ID_OFFSET)
#define TASK_SHIFT_SUB_TASK_ID(sUBtASKiD) UL_BIT_SHIFT(sUBtASKiD, TASK_SUB_TASK_ID_OFFSET)
#include <messages_types.h> #include <messages_types.h>
@@ -86,29 +81,31 @@ typedef enum
#undef SUB_TASK_DEF #undef SUB_TASK_DEF
#undef TASK_DEF #undef TASK_DEF
THREAD_MAX, THREAD_END = THREAD_MAX, THREAD_MAX,
} thread_id_t; } thread_id_t;
//! Sub-tasks id, to defined offset form thread id //! Sub-tasks id, to defined offset form thread id
typedef enum typedef enum
{ {
#define TASK_DEF(tHREADiD, pRIO) SUB_TASK_INIT_##tHREADiD = 0, #define TASK_DEF(tHREADiD, pRIO) tHREADiD##_THREAD = THREAD_##tHREADiD,
#define SUB_TASK_DEF(tHREADiD, sUBtASKiD) SUB_TASK_OFFSET_##sUBtASKiD, #define SUB_TASK_DEF(tHREADiD, sUBtASKiD) sUBtASKiD##_THREAD = THREAD_##tHREADiD,
#include <tasks_def.h> #include <tasks_def.h>
#undef SUB_TASK_DEF #undef SUB_TASK_DEF
#undef TASK_DEF #undef TASK_DEF
} sub_task_id_t; } task_thread_id_t;
//! Tasks id of each task //! Tasks id of each task
typedef enum typedef enum
{ {
#define TASK_DEF(tHREADiD, pRIO) tHREADiD = TASK_SHIFT_THREAD_ID(THREAD_##tHREADiD), TASK_FIRST = 1, TASK_UNKNOWN = 0,
#define SUB_TASK_DEF(tHREADiD, sUBtASKiD) sUBtASKiD = (TASK_SHIFT_THREAD_ID(THREAD_##tHREADiD) | TASK_SHIFT_SUB_TASK_ID(SUB_TASK_OFFSET_##sUBtASKiD)),
#define TASK_DEF(tHREADiD, pRIO) tHREADiD,
#define SUB_TASK_DEF(tHREADiD, sUBtASKiD) sUBtASKiD,
#include <tasks_def.h> #include <tasks_def.h>
#undef SUB_TASK_DEF #undef SUB_TASK_DEF
#undef TASK_DEF #undef TASK_DEF
TASK_UNKNOWN = 0xFFFF, TASK_MAX,
} task_id_t; } task_id_t;
typedef union msg_s typedef union msg_s
openair-cn/OAISIM_MME/oaisim_mme.c
+1 -1
View File
@@ -68,7 +68,7 @@ int main(int argc, char *argv[])
/* Calling each layer init function */ /* Calling each layer init function */
CHECK_INIT_RETURN(log_init(&mme_config, oai_mme_log_specific)); CHECK_INIT_RETURN(log_init(&mme_config, oai_mme_log_specific));
CHECK_INIT_RETURN(itti_init(THREAD_MAX, MESSAGES_ID_MAX, threads_name, messages_info, messages_definition_xml, NULL)); CHECK_INIT_RETURN(itti_init(TASK_MAX, THREAD_MAX, MESSAGES_ID_MAX, tasks_info, messages_info, messages_definition_xml, NULL));
CHECK_INIT_RETURN(nas_init(&mme_config)); CHECK_INIT_RETURN(nas_init(&mme_config));
CHECK_INIT_RETURN(sctp_init(&mme_config)); CHECK_INIT_RETURN(sctp_init(&mme_config));
openair-cn/OAI_EPC/oai_epc.c
+1 -1
View File
@@ -74,7 +74,7 @@ int main(int argc, char *argv[])
/* Calling each layer init function */ /* Calling each layer init function */
CHECK_INIT_RETURN(log_init(&mme_config, oai_epc_log_specific)); CHECK_INIT_RETURN(log_init(&mme_config, oai_epc_log_specific));
CHECK_INIT_RETURN(itti_init(THREAD_MAX, MESSAGES_ID_MAX, threads_name, messages_info, messages_definition_xml, NULL)); CHECK_INIT_RETURN(itti_init(TASK_MAX, THREAD_MAX, MESSAGES_ID_MAX, tasks_info, messages_info, messages_definition_xml, NULL));
CHECK_INIT_RETURN(nas_init(&mme_config)); CHECK_INIT_RETURN(nas_init(&mme_config));
CHECK_INIT_RETURN(sctp_init(&mme_config)); CHECK_INIT_RETURN(sctp_init(&mme_config));
openair-cn/OAI_SGW/oai_sgw.c
+1 -1
View File
@@ -60,7 +60,7 @@ int main(int argc, char *argv[])
/* Calling each layer init function */ /* Calling each layer init function */
CHECK_INIT_RETURN(log_init(&mme_config, oai_sgw_log_specific)); CHECK_INIT_RETURN(log_init(&mme_config, oai_sgw_log_specific));
CHECK_INIT_RETURN(itti_init(THREAD_MAX, MESSAGES_ID_MAX, threads_name, messages_info, messages_definition_xml, NULL)); CHECK_INIT_RETURN(itti_init(TASK_MAX, THREAD_MAX, MESSAGES_ID_MAX, tasks_info, messages_info, messages_definition_xml, NULL));
CHECK_INIT_RETURN(udp_init(&mme_config)); CHECK_INIT_RETURN(udp_init(&mme_config));
CHECK_INIT_RETURN(s11_sgw_init(&mme_config)); CHECK_INIT_RETURN(s11_sgw_init(&mme_config));
openair-cn/TEST/oaisim_mme_itti_test.c
+1 -1
View File
@@ -78,7 +78,7 @@ int main(int argc, char *argv[])
/* Calling each layer init function */ /* Calling each layer init function */
log_init(&mme_config); log_init(&mme_config);
itti_init(THREAD_MAX, MESSAGES_ID_MAX, threads_name, messages_info, messages_definition_xml, NULL); itti_init(TASK_MAX, THREAD_MAX, MESSAGES_ID_MAX, tasks_info, messages_info, messages_definition_xml, NULL);
sctp_init(&mme_config); sctp_init(&mme_config);
udp_init(&mme_config); udp_init(&mme_config);
s1ap_mme_init(&mme_config); s1ap_mme_init(&mme_config);
openair2/LAYER2/MAC/eNB_scheduler.c
+1 -1
View File
@@ -4212,7 +4212,7 @@ void eNB_dlsch_ulsch_scheduler(u8 Mod_id,u8 cooperation_flag, u32 frame, u8 subf
#if defined(ENABLE_ITTI) #if defined(ENABLE_ITTI)
do { do {
// Checks if a message has been sent to MAC sub-task // Checks if a message has been sent to MAC sub-task
itti_poll_msg (TASK_MAC_ENB, INSTANCE_ALL, &msg_p); itti_poll_msg (TASK_MAC_ENB, &msg_p);
if (msg_p != NULL) { if (msg_p != NULL) {
msg_name = ITTI_MSG_NAME (msg_p); msg_name = ITTI_MSG_NAME (msg_p);
openair2/LAYER2/MAC/ue_procedures.c
+1 -1
View File
@@ -1289,7 +1289,7 @@ UE_L2_STATE_t ue_scheduler(u8 Mod_id,u32 frame, u8 subframe, lte_subframe_t dire
#if defined(ENABLE_ITTI) #if defined(ENABLE_ITTI)
do { do {
// Checks if a message has been sent to MAC sub-task // Checks if a message has been sent to MAC sub-task
itti_poll_msg (TASK_MAC_UE, INSTANCE_ALL, &msg_p); itti_poll_msg (TASK_MAC_UE, &msg_p);
if (msg_p != NULL) { if (msg_p != NULL) {
msg_name = ITTI_MSG_NAME (msg_p); msg_name = ITTI_MSG_NAME (msg_p);
openair2/LAYER2/PDCP_v10.1.0/pdcp.c
+1 -1
View File
@@ -500,7 +500,7 @@ void pdcp_run (u32_t frame, u8 eNB_flag, u8 UE_index, u8 eNB_index) {
#if defined(ENABLE_ITTI) #if defined(ENABLE_ITTI)
do { do {
// Checks if a message has been sent to PDCP sub-task // Checks if a message has been sent to PDCP sub-task
itti_poll_msg (eNB_flag ? TASK_PDCP_ENB : TASK_PDCP_UE, INSTANCE_ALL, &msg_p); itti_poll_msg (eNB_flag ? TASK_PDCP_ENB : TASK_PDCP_UE, &msg_p);
if (msg_p != NULL) { if (msg_p != NULL) {
msg_name = ITTI_MSG_NAME (msg_p); msg_name = ITTI_MSG_NAME (msg_p);
targets/SIMU/USER/oaisim.c
+2 -2
View File
@@ -430,7 +430,7 @@ void *l2l1_task(void *args_p) {
#if defined(ENABLE_ITTI) #if defined(ENABLE_ITTI)
do { do {
// Checks if a message has been sent to L2L1 task // Checks if a message has been sent to L2L1 task
itti_poll_msg (TASK_L2L1, INSTANCE_ALL, &message_p); itti_poll_msg (TASK_L2L1, &message_p);
if (message_p != NULL) { if (message_p != NULL) {
switch (message_p->header.messageId) { switch (message_p->header.messageId) {
@@ -936,7 +936,7 @@ int main(int argc, char **argv) {
} }
#if defined(ENABLE_ITTI) #if defined(ENABLE_ITTI)
itti_init(THREAD_MAX, MESSAGES_ID_MAX, threads_name, messages_info, messages_definition_xml, oai_emulation.info.itti_dump_file); itti_init(TASK_MAX, THREAD_MAX, MESSAGES_ID_MAX, tasks_info, messages_info, messages_definition_xml, oai_emulation.info.itti_dump_file);
#endif #endif
#ifdef OPENAIR2 #ifdef OPENAIR2