Commit b45af4a3 authored by winckel's avatar winckel

Moved files related to assertions, backtrace and signals handling into ITTI directory.

Renamed all ITTI functions, to make them start with module name ("itti_").
Added generic signals handling support in ITTI.
Changed ITTI log port for oaisim to 10006.
Introduce an option to use ITTI in oaisim and reoragized code.

git-svn-id: http://svn.eurecom.fr/openair4G/trunk@4267 818b1a75-f10b-46b9-bf7c-635c3b92a50f
parent 228aa778
...@@ -2,6 +2,8 @@ ITTI_DIR = $(COMMON_UTILS_DIR)/itti ...@@ -2,6 +2,8 @@ ITTI_DIR = $(COMMON_UTILS_DIR)/itti
ITTI_OBJS = $(ITTI_DIR)/intertask_interface.o ITTI_OBJS = $(ITTI_DIR)/intertask_interface.o
ITTI_OBJS += $(ITTI_DIR)/intertask_interface_dump.o ITTI_OBJS += $(ITTI_DIR)/intertask_interface_dump.o
ITTI_OBJS += $(ITTI_DIR)/backtrace.o
ITTI_OBJS += $(ITTI_DIR)/signals.o
ITTI_OBJS += $(ITTI_DIR)/timer.o ITTI_OBJS += $(ITTI_DIR)/timer.o
UTILS_OBJS = $(ITTI_OBJS) UTILS_OBJS = $(ITTI_OBJS)
......
...@@ -17,4 +17,7 @@ libitti_la_SOURCES = \ ...@@ -17,4 +17,7 @@ libitti_la_SOURCES = \
udp_message_def.h udp_messages_types.h \ udp_message_def.h udp_messages_types.h \
intertask_interface.c intertask_interface.h \ intertask_interface.c intertask_interface.h \
intertask_interface_dump.c intertask_interface_dump.h \ intertask_interface_dump.c intertask_interface_dump.h \
assertions.h \
backtrace.c backtrace.h \
signals.c signals.h \
timer.c timer.h timer.c timer.h
\ No newline at end of file
/******************************************************************************* /*******************************************************************************
Eurecom OpenAirInterface Eurecom OpenAirInterface
Copyright(c) 1999 - 2012 Eurecom Copyright(c) 1999 - 2012 Eurecom
This program is free software; you can redistribute it and/or modify it This program is free software; you can redistribute it and/or modify it
under the terms and conditions of the GNU General Public License, under the terms and conditions of the GNU General Public License,
version 2, as published by the Free Software Foundation. version 2, as published by the Free Software Foundation.
This program is distributed in the hope it will be useful, but WITHOUT This program is distributed in the hope it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
more details. more details.
You should have received a copy of the GNU General Public License along with You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc., this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA. 51 Franklin St - Fifth Floor, Boston, MA 02110-1301 USA.
The full GNU General Public License is included in this distribution in The full GNU General Public License is included in this distribution in
the file called "COPYING". the file called "COPYING".
Contact Information Contact Information
Openair Admin: openair_admin@eurecom.fr Openair Admin: openair_admin@eurecom.fr
Openair Tech : openair_tech@eurecom.fr Openair Tech : openair_tech@eurecom.fr
Forums : http://forums.eurecom.fr/openairinterface Forums : http://forums.eurecom.fr/openairinterface
Address : EURECOM, Campus SophiaTech, 450 Route des Chappes Address : EURECOM, Campus SophiaTech, 450 Route des Chappes
06410 Biot FRANCE 06410 Biot FRANCE
*******************************************************************************/ *******************************************************************************/
#include <pthread.h> #include <pthread.h>
#include <stdio.h> #include <stdio.h>
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include <unistd.h> #include <unistd.h>
#include <string.h> #include <string.h>
#include <errno.h> #include <errno.h>
#include <signal.h>
#include "queue.h" #include "queue.h"
#include "assertions.h" #include "assertions.h"
...@@ -45,6 +46,7 @@ ...@@ -45,6 +46,7 @@
#include "intertask_interface_init.h" #include "intertask_interface_init.h"
#undef CHECK_PROTOTYPE_ONLY #undef CHECK_PROTOTYPE_ONLY
#include "signals.h"
#include "timer.h" #include "timer.h"
int itti_debug = 1; int itti_debug = 1;
...@@ -58,125 +60,124 @@ int itti_debug = 1; ...@@ -58,125 +60,124 @@ int itti_debug = 1;
#define MESSAGE_SIZE(mESSAGEiD) (sizeof(MessageHeader) + itti_desc.messages_info[mESSAGEiD].size) #define MESSAGE_SIZE(mESSAGEiD) (sizeof(MessageHeader) + itti_desc.messages_info[mESSAGEiD].size)
typedef enum task_state_s { typedef enum task_state_s {
TASK_STATE_NOT_CONFIGURED, TASK_STATE_NOT_CONFIGURED, TASK_STATE_STARTING, TASK_STATE_READY, TASK_STATE_ENDED, TASK_STATE_MAX,
TASK_STATE_STARTING,
TASK_STATE_READY,
TASK_STATE_MAX,
} task_state_t; } task_state_t;
/* This list acts as a FIFO of messages received by tasks (RRC, NAS, ...) */ /* This list acts as a FIFO of messages received by tasks (RRC, NAS, ...) */
struct message_list_s { struct message_list_s {
STAILQ_ENTRY(message_list_s) next_element; STAILQ_ENTRY(message_list_s) next_element;
MessageDef *msg; ///< Pointer to the message MessageDef *msg; ///< Pointer to the message
message_number_t message_number; ///< Unique message number message_number_t message_number; ///< Unique message number
uint32_t message_priority; ///< Message priority uint32_t message_priority; ///< Message priority
}; };
typedef struct task_desc_s { typedef struct task_desc_s {
/* Queue of messages belonging to the task */ /* Queue of messages belonging to the task */
STAILQ_HEAD(message_queue_head, message_list_s) message_queue; STAILQ_HEAD(message_queue_head, message_list_s)
message_queue;
/* Number of messages in the queue */ /* Number of messages in the queue */
volatile uint32_t message_in_queue; volatile uint32_t message_in_queue;
/* Mutex for the message queue */ /* Mutex for the message queue */
pthread_mutex_t message_queue_mutex; pthread_mutex_t message_queue_mutex;
/* Conditional var for message queue and task synchro */ /* Conditional var for message queue and task synchro */
pthread_cond_t message_queue_cond_var; pthread_cond_t message_queue_cond_var;
pthread_t task_thread; pthread_t task_thread;
volatile task_state_t task_state; volatile task_state_t task_state;
} task_desc_t; } task_desc_t;
struct itti_desc_s { struct itti_desc_s {
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;
MessagesIds messages_id_max; MessagesIds messages_id_max;
pthread_t thread_handling_signals;
const char * const *threads_name; const char * const *threads_name;
const message_info_t *messages_info; const message_info_t *messages_info;
}; };
static struct itti_desc_s itti_desc; static struct itti_desc_s itti_desc;
static inline static inline message_number_t itti_increment_message_number(void) {
message_number_t increment_message_number(void)
{
/* Atomic operation supported by GCC: returns the current message number /* Atomic operation supported by GCC: returns the current message number
* and then increment it by 1. * and then increment it by 1.
* This can be done without mutex. * This can be done without mutex.
*/ */
return __sync_fetch_and_add(&itti_desc.message_number, 1); return __sync_fetch_and_add (&itti_desc.message_number, 1);
} }
static inline uint32_t get_message_priority(MessagesIds message_id) static inline uint32_t itti_get_message_priority(MessagesIds message_id) {
{
DevCheck(message_id < itti_desc.messages_id_max, message_id, itti_desc.messages_id_max, 0); DevCheck(message_id < itti_desc.messages_id_max, message_id, itti_desc.messages_id_max, 0);
return (itti_desc.messages_info[message_id].priority); return (itti_desc.messages_info[message_id].priority);
} }
char *get_message_name(MessagesIds message_id) char *itti_get_message_name(MessagesIds message_id) {
{
DevCheck(message_id < itti_desc.messages_id_max, message_id, itti_desc.messages_id_max, 0); DevCheck(message_id < itti_desc.messages_id_max, message_id, itti_desc.messages_id_max, 0);
return (itti_desc.messages_info[message_id].name); return (itti_desc.messages_info[message_id].name);
} }
int send_broadcast_message(MessageDef *message_p) int itti_send_broadcast_message(MessageDef *message_p) {
{ thread_id_t origin_thread_id;
uint32_t i; uint32_t i;
int ret = 0; int ret = 0;
int temp; int result;
for (i = THREAD_FIRST; i < itti_desc.thread_max; i++) if (message_p == NULL) {
{ ITTI_ERROR("Message to broadcast is NULL (%s:%d)\n", __FILE__, __LINE__);
MessageDef *new_message_p; return -1;
}
/* Skip tasks which are not running */ origin_thread_id = TASK_GET_THREAD_ID(message_p->header.originTaskId);
if (itti_desc.tasks[i].task_state == TASK_STATE_READY)
{
new_message_p = malloc(sizeof(MessageDef));
if (new_message_p == NULL) { for (i = THREAD_FIRST; i < itti_desc.thread_max; i++) {
ITTI_ERROR("Failed to allocate memory (%s:%d)\n", MessageDef *new_message_p;
__FILE__, __LINE__);
return -1; /* Skip task that broadcast the message */
} if (i != origin_thread_id) {
memcpy(new_message_p, message_p, sizeof(MessageDef)); /* Skip tasks which are not running */
temp = send_msg_to_task(TASK_SHIFT_THREAD_ID(i), INSTANCE_DEFAULT, new_message_p); if (itti_desc.tasks[i].task_state == TASK_STATE_READY) {
if (temp < 0) { new_message_p = malloc (sizeof(MessageDef));
ITTI_ERROR("Failed to send broadcast message (%s) to queue (%u:%s)\n",
itti_desc.messages_info[message_p->header.messageId].name, i, itti_desc.threads_name[i]); if (new_message_p == NULL) {
ret = temp; ITTI_ERROR("Failed to allocate memory (%s:%d)\n", __FILE__, __LINE__);
free(new_message_p); return -1;
}
memcpy (new_message_p, message_p, sizeof(MessageDef));
result = itti_send_msg_to_task (TASK_SHIFT_THREAD_ID(i), INSTANCE_DEFAULT, new_message_p);
if (result < 0) {
ITTI_ERROR("Failed to send broadcast message (%s) to queue (%u:%s)\n",
itti_desc.messages_info[message_p->header.messageId].name, i, itti_desc.threads_name[i]);
ret = result;
free (new_message_p);
}
} }
} }
} }
free (message_p);
return ret; return ret;
} }
inline MessageDef *alloc_new_message( inline MessageDef *itti_alloc_new_message(task_id_t origin_task_id, MessagesIds message_id) {
task_id_t origin_task_id,
MessagesIds message_id)
{
MessageDef *temp = NULL; MessageDef *temp = NULL;
if ((TASK_GET_THREAD_ID(origin_task_id) >= itti_desc.thread_max) || if (message_id >= itti_desc.messages_id_max) {
(message_id >= itti_desc.messages_id_max)) ITTI_ERROR("Invalid message id %d (%s:%d)\n", message_id, __FILE__, __LINE__);
{
return NULL; return NULL;
} }
temp = calloc(1, MESSAGE_SIZE(message_id)); temp = calloc (1, MESSAGE_SIZE(message_id));
if (temp == NULL) { if (temp == NULL) {
ITTI_ERROR("Cannot allocate memory for new message (%s:%d)\n", ITTI_ERROR("Cannot allocate memory for new message (%s:%d)\n", __FILE__, __LINE__);
__FILE__, __LINE__);
return NULL; return NULL;
} }
...@@ -187,16 +188,14 @@ inline MessageDef *alloc_new_message( ...@@ -187,16 +188,14 @@ inline MessageDef *alloc_new_message(
return temp; return temp;
} }
int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message) int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message) {
{ thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); struct message_list_s *new;
struct message_list_s *new; uint32_t priority;
uint32_t priority; message_number_t message_number;
message_number_t message_number; uint32_t message_id;
uint32_t message_id;
if (thread_id >= itti_desc.thread_max) if (thread_id >= itti_desc.thread_max) {
{
return -1; return -1;
} }
...@@ -208,43 +207,41 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message ...@@ -208,43 +207,41 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message
DevCheck(thread_id < itti_desc.thread_max, thread_id, 0, 0); DevCheck(thread_id < itti_desc.thread_max, thread_id, 0, 0);
DevCheck(message_id < itti_desc.messages_id_max, itti_desc.messages_id_max, message_id, 0); DevCheck(message_id < itti_desc.messages_id_max, itti_desc.messages_id_max, message_id, 0);
priority = get_message_priority(message_id); priority = itti_get_message_priority (message_id);
/* 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[thread_id].message_queue_mutex);
/* 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, DevCheck(itti_desc.tasks[thread_id].task_state == TASK_STATE_READY, itti_desc.tasks[thread_id].task_state,
itti_desc.tasks[thread_id].task_state, TASK_STATE_READY, thread_id); TASK_STATE_READY, thread_id);
/* 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)) DevCheck((itti_desc.tasks[thread_id].message_in_queue * sizeof(MessageDef)) < ITTI_QUEUE_SIZE_PER_TASK,
< ITTI_QUEUE_SIZE_PER_TASK, (itti_desc.tasks[thread_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[thread_id].message_in_queue); itti_desc.tasks[thread_id].message_in_queue);
/* Allocate new list element */ /* Allocate new list element */
if ((new = (struct message_list_s *)malloc(sizeof(struct message_list_s))) == NULL) if ((new = (struct message_list_s *) malloc (sizeof(struct message_list_s))) == NULL) {
{ ITTI_ERROR("Cannot allocate memory for new message (%s:%d)\n", __FILE__, __LINE__);
ITTI_ERROR("Cannot allocate memory for new message (%s:%d)\n",
__FILE__, __LINE__);
return -1; return -1;
} }
/* Increment the global message number */ /* Increment the global message number */
message_number = increment_message_number(); message_number = itti_increment_message_number ();
/* Fill in members */ /* Fill in members */
new->msg = message; new->msg = message;
new->message_number = message_number; new->message_number = message_number;
new->message_priority = priority; new->message_priority = priority;
itti_dump_queue_message(message_number, message, itti_desc.messages_info[message_id].name, MESSAGE_SIZE(message_id)); itti_dump_queue_message (message_number, message, itti_desc.messages_info[message_id].name,
MESSAGE_SIZE(message_id));
if (STAILQ_EMPTY(&itti_desc.tasks[thread_id].message_queue)) { if (STAILQ_EMPTY (&itti_desc.tasks[thread_id].message_queue)) {
STAILQ_INSERT_HEAD(&itti_desc.tasks[thread_id].message_queue, new, next_element); STAILQ_INSERT_HEAD (&itti_desc.tasks[thread_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;
// //
...@@ -263,7 +260,7 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message ...@@ -263,7 +260,7 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message
// } // }
// } // }
// 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[thread_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[thread_id].message_queue, insert_after, new,
// next_element); // next_element);
...@@ -274,84 +271,78 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message ...@@ -274,84 +271,78 @@ int send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *message
itti_desc.tasks[thread_id].message_in_queue++; itti_desc.tasks[thread_id].message_in_queue++;
if (itti_desc.tasks[thread_id].message_in_queue == 1) { if (itti_desc.tasks[thread_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[thread_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[thread_id].message_queue_mutex);
ITTI_DEBUG("Message %s, number %lu with priority %d successfully sent to queue (%u:%s)\n", ITTI_DEBUG(
itti_desc.messages_info[message_id].name, message_number, "Message %s, number %lu with priority %d successfully sent to queue (%u:%s)\n",
priority, thread_id, itti_desc.threads_name[thread_id]); itti_desc.messages_info[message_id].name, message_number, priority, thread_id, itti_desc.threads_name[thread_id]);
return 0; return 0;
} }
void receive_msg(task_id_t task_id, MessageDef **received_msg) void itti_receive_msg(task_id_t task_id, MessageDef **received_msg) {
{
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
DevCheck(thread_id < itti_desc.thread_max, thread_id, 0, 0); DevCheck(thread_id < itti_desc.thread_max, thread_id, 0, 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[thread_id].message_queue_mutex);
if (itti_desc.tasks[thread_id].message_in_queue == 0) { if (itti_desc.tasks[thread_id].message_in_queue == 0) {
ITTI_DEBUG("Message in queue[(%u:%s)] == 0, waiting\n", ITTI_DEBUG("Message in queue[(%u:%s)] == 0, waiting\n", thread_id, itti_desc.threads_name[thread_id]);
thread_id, itti_desc.threads_name[thread_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[thread_id].message_queue_cond_var,
&itti_desc.tasks[thread_id].message_queue_mutex); &itti_desc.tasks[thread_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 for task %x\n",
thread_id, itti_desc.threads_name[thread_id], task_id); thread_id, itti_desc.threads_name[thread_id], task_id);
} }
if (!STAILQ_EMPTY(&itti_desc.tasks[thread_id].message_queue)) { if (!STAILQ_EMPTY (&itti_desc.tasks[thread_id].message_queue)) {
struct message_list_s *temp = STAILQ_FIRST( struct message_list_s *temp = STAILQ_FIRST (&itti_desc.tasks[thread_id].message_queue);
&itti_desc.tasks[thread_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[thread_id].message_queue, next_element);
free(temp); free (temp);
itti_desc.tasks[thread_id].message_in_queue--; itti_desc.tasks[thread_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[thread_id].message_queue_mutex);
} }
void poll_msg(task_id_t task_id, instance_t instance, MessageDef **received_msg) void itti_poll_msg(task_id_t task_id, instance_t instance, MessageDef **received_msg) {
{
thread_id_t thread_id = TASK_GET_THREAD_ID(task_id); thread_id_t thread_id = TASK_GET_THREAD_ID(task_id);
DevCheck (thread_id < itti_desc.thread_max, thread_id, 0, 0); DevCheck(thread_id < itti_desc.thread_max, thread_id, 0, 0);
DevAssert (received_msg != NULL); DevAssert(received_msg != NULL);
*received_msg = NULL; *received_msg = NULL;
if (itti_desc.tasks[thread_id].message_in_queue != 0) if (itti_desc.tasks[thread_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[thread_id].message_queue_mutex);
STAILQ_FOREACH(temp, &itti_desc.tasks[thread_id].message_queue, next_element) STAILQ_FOREACH (temp, &itti_desc.tasks[thread_id].message_queue, next_element)
{ {
if ((temp->msg->header.destinationTaskId == task_id) if ((temp->msg->header.destinationTaskId == task_id)
&& ((instance == INSTANCE_ALL) || (temp->msg->header.instance == instance))) && ((instance == INSTANCE_ALL) || (temp->msg->header.instance == instance))) {
{
/* Update received_msg reference */ /* Update received_msg reference */
*received_msg = temp->msg; *received_msg = temp->msg;
/* Remove message from queue */ /* Remove message from queue */