- Fix some performances issues for softmodem

- Remove eventfd if task has only polling mode - Created dummy thread to convert eventfd from RT thread to non RT threads - Added eNB app to softmodem - Set poll mode for itti dumper thread in softmodem - Disable timers for softmodem git-svn-id: http://svn.eurecom.fr/openair4G/trunk@4545 818b1a75-f10b-46b9-bf7c-635c3b92a50f
2013-11-28 15:51:49 +00:00
parent 18f113c486
commit 6102fa4b60
12 changed files with 421 additions and 178 deletions
@@ -37,6 +37,10 @@
 #include <errno.h>
 #include <signal.h>

+#ifdef RTAI
+# include <rtai_fifos.h>
+#endif
+
 #include "queue.h"
 #include "assertions.h"

@@ -49,6 +53,10 @@
 #include "intertask_interface.h"
 #include "intertask_interface_dump.h"

+#if defined(OAI_EMU) || defined(RTAI)
+# include "vcd_signal_dumper.h"
+#endif
+
 /* Includes "intertask_interface_init.h" to check prototype coherence, but
 * disable threads and messages information generation.
 */
@@ -62,10 +70,18 @@
 const int itti_debug = 0;
 const int itti_debug_poll = 0;

-#define ITTI_DEBUG(x, args...) do { if (itti_debug) fprintf(stdout, "[ITTI][D]"x, ##args); fflush (stdout); } \
+/* Don't flush if using RTAI */
+#ifdef RTAI
+# define ITTI_DEBUG(x, args...) do { if (itti_debug) rt_printk("[ITTI][D]"x, ##args); } \
    while(0)
-#define ITTI_ERROR(x, args...) do { fprintf(stdout, "[ITTI][E]"x, ##args); fflush (stdout); } \
+# define ITTI_ERROR(x, args...) do { rt_printk("[ITTI][E]"x, ##args); } \
    while(0)
+#else
+# define ITTI_DEBUG(x, args...) do { if (itti_debug) fprintf(stdout, "[ITTI][D]"x, ##args); fflush (stdout); } \
+    while(0)
+# define ITTI_ERROR(x, args...) do { fprintf(stdout, "[ITTI][E]"x, ##args); fflush (stdout); } \
+    while(0)
+#endif

 /* Global message size */
 #define MESSAGE_SIZE(mESSAGEiD) (sizeof(MessageHeader) + itti_desc.messages_info[mESSAGEiD].size)
@@ -89,6 +105,7 @@ typedef struct message_list_s {
 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;
@@ -266,6 +283,7 @@ inline MessageDef *itti_alloc_new_message(task_id_t origin_task_id, MessagesIds
 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 origin_task_id;
    message_list_t *new;
    uint32_t priority;
    message_number_t message_number;
@@ -281,13 +299,31 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
    message_id = message->ittiMsgHeader.messageId;
    DevCheck(message_id < itti_desc.messages_id_max, itti_desc.messages_id_max, message_id, 0);

+    origin_task_id = ITTI_MSG_ORIGIN_ID(message);
+
+#if defined(OAI_EMU) || defined(RTAI)
+    vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_SEND_MSG,
+                                            task_id);
+#endif
+
    priority = itti_get_message_priority (message_id);

    /* Increment the global message number */
    message_number = itti_increment_message_number ();

-    itti_dump_queue_message (message_number, message, itti_desc.messages_info[message_id].name,
-                             sizeof(MessageHeader) + message->ittiMsgHeader.ittiMsgSize);
+#ifdef RTAI
+    if (pthread_self() != itti_desc.threads[TASK_GET_THREAD_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_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)
    {
@@ -314,6 +350,19 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
        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))
+        {
+            /* 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;
@@ -322,7 +371,9 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me

            /* 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, 0);
+            DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, task_id);
+        } else {
+            lfds611_queue_enqueue(itti_desc.tasks[task_id].message_queue, new);
        }
 #else
        if (STAILQ_EMPTY (&itti_desc.tasks[task_id].message_queue)) {
@@ -365,9 +416,11 @@ int itti_send_msg_to_task(task_id_t task_id, instance_t instance, MessageDef *me
 #endif
    }

-    ITTI_DEBUG(
-            "Message %s, number %lu with priority %d successfully sent to queue (%u:%s)\n",
-            itti_desc.messages_info[message_id].name, message_number, priority, task_id, itti_get_task_name(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);
+#endif
+
    return 0;
 }

@@ -499,6 +552,10 @@ static inline void itti_receive_msg_internal_event_fd(task_id_t task_id, uint8_t

 void itti_receive_msg(task_id_t task_id, MessageDef **received_msg)
 {
+#if defined(OAI_EMU) || defined(RTAI)
+    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
@@ -531,6 +588,10 @@ void itti_receive_msg(task_id_t task_id, MessageDef **received_msg)
    // Release the mutex
    pthread_mutex_unlock (&itti_desc.tasks[task_id].message_queue_mutex);
 #endif
+#if defined(OAI_EMU) || defined(RTAI)
+    vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_RECV_MSG_END,
+                                            task_id);
+#endif
 }

 void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
@@ -539,8 +600,21 @@ void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {

    *received_msg = NULL;

+#if defined(OAI_EMU) || defined(RTAI)
+    vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_POLL_MSG,
+                                            task_id);
+#endif
+
 #if defined(ENABLE_EVENT_FD)
-    itti_receive_msg_internal_event_fd(task_id, 1, received_msg);
+    {
+        struct message_list_s *message;
+
+        if (lfds611_queue_dequeue (itti_desc.tasks[task_id].message_queue, (void **) &message) == 1)
+        {
+            *received_msg = message->msg;
+            free (message);
+        }
+    }
 #else
    if (itti_desc.tasks[task_id].message_in_queue != 0) {
        message_list_t *temp;
@@ -572,6 +646,11 @@ void itti_poll_msg(task_id_t task_id, MessageDef **received_msg) {
    if ((itti_debug_poll) && (*received_msg == NULL)) {
        ITTI_DEBUG("No message in queue[(%u:%s)]\n", task_id, itti_get_task_name(task_id));
    }
+
+#if defined(OAI_EMU) || defined(RTAI)
+    vcd_signal_dumper_dump_variable_by_name(VCD_SIGNAL_DUMPER_VARIABLE_ITTI_POLL_MSG_END,
+                                            task_id);
+#endif
 }

 int itti_create_task(task_id_t task_id, void *(*start_routine)(void *), void *args_p) {
@@ -596,15 +675,28 @@ int itti_create_task(task_id_t task_id, void *(*start_routine)(void *), void *ar
    return 0;
 }

-void itti_mark_task_ready(task_id_t task_id) {
+void itti_mark_task_ready(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);

+#ifdef RTAI
+    /* Assign low priority to created threads */
+    {
+        struct sched_param sched_param;
+        sched_param.sched_priority = sched_get_priority_min(SCHED_FIFO) + 1;
+        sched_setscheduler(0, SCHED_FIFO, &sched_param);
+    }
+#endif
+
    /* Register the thread in itti dump */
    itti_dump_thread_use_ring_buffer();

-#if !defined(ENABLE_EVENT_FD)
+#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
@@ -636,6 +728,8 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i
              const message_info_t *messages_info, const char * const messages_definition_xml, const char * const dump_file_name) {
    task_id_t task_id;
    thread_id_t thread_id;
+    int ret;
+
    itti_desc.message_number = 1;

    ITTI_DEBUG("Init: %d tasks, %d threads, %d messages\n", task_max, thread_max, messages_id_max);
@@ -664,14 +758,30 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i
    /* Initializing each queue and related stuff */
    for (task_id = TASK_FIRST; task_id < itti_desc.task_max; task_id++)
    {
+        ITTI_DEBUG("Initializing %stask %s%s%s\n",
+                   itti_desc.tasks_info[task_id].parent_task != TASK_UNKNOWN ? "sub-" : "",
+                   itti_desc.tasks_info[task_id].name,
+                   itti_desc.tasks_info[task_id].parent_task != TASK_UNKNOWN ? " with parent " : "",
+                   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);
-        if (lfds611_queue_new(&itti_desc.tasks[task_id].message_queue, itti_desc.tasks_info[task_id].queue_size) < 0)
+
+        ret = lfds611_queue_new(&itti_desc.tasks[task_id].message_queue, itti_desc.tasks_info[task_id].queue_size);
+        if (ret < 0)
        {
            ITTI_ERROR("lfds611_queue_new failed for task %u\n", task_id);
            DevAssert(0 == 1);
        }

+# ifdef RTAI
+        if (task_id == TASK_L2L1)
+        {
+            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) {
            ITTI_ERROR("Failed to create new epoll fd: %s\n", strerror(errno));
@@ -689,7 +799,7 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i

        itti_desc.tasks[task_id].nb_events = 1;

-        itti_desc.tasks[task_id].events = malloc(sizeof(struct epoll_event));
+        itti_desc.tasks[task_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;
@@ -725,7 +835,9 @@ int itti_init(task_id_t task_max, thread_id_t thread_max, MessagesIds messages_i

    itti_dump_init (messages_definition_xml, dump_file_name);

-    CHECK_INIT_RETURN(timer_init ());
+#ifndef RTAI
+     CHECK_INIT_RETURN(timer_init ());
+#endif

    return 0;
 }
@@ -38,6 +38,10 @@
 # include <sys/epoll.h>
 #endif

+#ifdef RTAI
+# include <rtai_sem.h>
+#endif
+
 #ifndef INTERTASK_INTERFACE_H_
 #define INTERTASK_INTERFACE_H_

@@ -91,6 +95,7 @@ typedef enum task_priorities_e {

 typedef struct task_info_s {
    thread_id_t thread;
+    task_id_t   parent_task;
    task_priorities_t priority;
    unsigned int queue_size;
    /* Printable name */
@@ -58,14 +58,25 @@
 #include "intertask_interface.h"
 #include "intertask_interface_dump.h"

+#if defined(RTAI)
+#include "vcd_signal_dumper.h"
+#endif
+
 #define SIGNAL_NAME_LENGTH  48

 static const int itti_dump_debug = 0;

-#define ITTI_DUMP_DEBUG(x, args...) do { if (itti_dump_debug) fprintf(stdout, "[ITTI][D]"x, ##args); } \
+#ifdef RTAI
+# define ITTI_DUMP_DEBUG(x, args...) do { if (itti_dump_debug) rt_printk("[ITTI][D]"x, ##args); } \
    while(0)
-#define ITTI_DUMP_ERROR(x, args...) do { fprintf(stdout, "[ITTI][E]"x, ##args); } \
+# define ITTI_DUMP_ERROR(x, args...) do { rt_printk("[ITTI][E]"x, ##args); } \
    while(0)
+#else
+# define ITTI_DUMP_DEBUG(x, args...) do { if (itti_dump_debug) fprintf(stdout, "[ITTI][D]"x, ##args); } \
+    while(0)
+# define ITTI_DUMP_ERROR(x, args...) do { fprintf(stdout, "[ITTI][E]"x, ##args); } \
+    while(0)
+#endif

 /* Message sent is an intertask dump type */
 #define ITTI_DUMP_MESSAGE_TYPE      0x1
@@ -100,12 +111,14 @@ typedef struct itti_desc_s {
     */
    struct lfds611_ringbuffer_state *itti_message_queue;

-    uint32_t queue_size;
-
    int nb_connected;

+#ifndef RTAI
    /* Event fd used to notify new messages (semaphore) */
    int event_fd;
+#else
+    unsigned long messages_in_queue __attribute__((aligned(8)));
+#endif

    int itti_listen_socket;

@@ -245,15 +258,16 @@ static int itti_dump_send_xml_definition(const int sd, const char *message_defin
 }

 static int itti_dump_enqueue_message(itti_dump_queue_item_t *new, uint32_t message_size,
-                                uint32_t message_type)
+                                     uint32_t message_type)
 {
-    ssize_t  write_ret;
-    uint64_t sem_counter = 1;
-
    struct lfds611_freelist_element *new_queue_element = NULL;

    DevAssert(new != NULL);

+#if defined(RTAI)
+    vcd_signal_dumper_dump_function_by_name(VCD_SIGNAL_DUMPER_FUNCTIONS_ITTI_DUMP_ENQUEUE_MESSAGE, VCD_FUNCTION_IN);
+#endif
+
    new->message_type = message_type;
    new->message_size = message_size;

@@ -265,9 +279,22 @@ static int itti_dump_enqueue_message(itti_dump_queue_item_t *new, uint32_t messa
    lfds611_ringbuffer_put_write_element(itti_dump_queue.itti_message_queue,
                                         new_queue_element);

-    /* Call to write for an event fd must be of 8 bytes */
-    write_ret = write(itti_dump_queue.event_fd, &sem_counter, sizeof(sem_counter));
-    DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, 0);
+#ifdef RTAI
+    __sync_fetch_and_add (&itti_dump_queue.messages_in_queue, 1);
+#else
+    {
+        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_dump_queue.event_fd, &sem_counter, sizeof(sem_counter));
+        DevCheck(write_ret == sizeof(sem_counter), write_ret, sem_counter, 0);
+    }
+#endif
+
+#if defined(RTAI)
+    vcd_signal_dumper_dump_function_by_name(VCD_SIGNAL_DUMPER_FUNCTIONS_ITTI_DUMP_ENQUEUE_MESSAGE, VCD_FUNCTION_OUT);
+#endif

    return 0;
 }
@@ -281,7 +308,6 @@ int itti_dump_queue_message(message_number_t message_number,
    {
        itti_dump_queue_item_t *new;
        size_t message_name_length;
-        int i;

        DevAssert(message_name != NULL);
        DevAssert(message_p != NULL);
@@ -311,17 +337,73 @@ int itti_dump_queue_message(message_number_t message_number,
        memcpy(new->message_name, message_name, message_name_length);

        itti_dump_enqueue_message(new, message_size, ITTI_DUMP_MESSAGE_TYPE);
-
-        for (i = 0; i < ITTI_DUMP_MAX_CON; i++) {
-            if (itti_dump_queue.itti_clients[i].sd == -1)
-                continue;
-            itti_dump_send_message(itti_dump_queue.itti_clients[i].sd, new);
-        }
    }

    return 0;
 }

+static void itti_dump_flush_ring_buffer(int flush_all)
+{
+    struct lfds611_freelist_element *element = NULL;
+    void *user_data;
+    int   j;
+
+#ifdef RTAI
+    unsigned long number_of_messages;
+
+    number_of_messages = itti_dump_queue.messages_in_queue;
+
+    ITTI_DUMP_DEBUG("%lu elements in queue\n", number_of_messages);
+
+    if (number_of_messages == 0) {
+        return;
+    }
+
+    __sync_sub_and_fetch(&itti_dump_queue.messages_in_queue, number_of_messages);
+#endif
+
+    do {
+        /* Acquire the ring element */
+        lfds611_ringbuffer_get_read_element(itti_dump_queue.itti_message_queue, &element);
+
+        DevAssert(element != NULL);
+
+        /* Retrieve user part of the message */
+        lfds611_freelist_get_user_data_from_element(element, &user_data);
+
+        if (((itti_dump_queue_item_t *)user_data)->message_type == ITTI_DUMP_EXIT_SIGNAL)
+        {
+#ifndef RTAI
+            close(itti_dump_queue.event_fd);
+#endif
+            close(itti_dump_queue.itti_listen_socket);
+
+            lfds611_ringbuffer_put_read_element(itti_dump_queue.itti_message_queue, element);
+
+            /* Leave the thread as we detected end signal */
+            pthread_exit(NULL);
+        }
+
+        /* Write message to file */
+        itti_dump_fwrite_message((itti_dump_queue_item_t *)user_data);
+
+        /* Send message to remote analyzer */
+        for (j = 0; j < ITTI_DUMP_MAX_CON; j++) {
+            if (itti_dump_queue.itti_clients[j].sd > 0) {
+                itti_dump_send_message(itti_dump_queue.itti_clients[j].sd,
+                                        (itti_dump_queue_item_t *)user_data);
+            }
+        }
+
+        /* We have finished with this element, reinsert it in the ring buffer */
+        lfds611_ringbuffer_put_read_element(itti_dump_queue.itti_message_queue, element);
+    } while(flush_all
+#ifdef RTAI
+            && --number_of_messages
+#endif
+            );
+}
+
 static void *itti_dump_socket(void *arg_p)
 {
    uint32_t message_definition_xml_length;
@@ -332,6 +414,11 @@ static void *itti_dump_socket(void *arg_p)
    fd_set read_set, working_set;
    struct sockaddr_in servaddr; /* socket address structure */

+    struct timeval *timeout_p = NULL;
+#ifdef RTAI
+    struct timeval  timeout;
+#endif
+
    ITTI_DUMP_DEBUG("Creating TCP dump socket on port %u\n", ITTI_PORT);

    message_definition_xml = (char *)arg_p;
@@ -383,11 +470,15 @@ static void *itti_dump_socket(void *arg_p)
    /* Add the listener */
    FD_SET(itti_listen_socket, &read_set);

+#ifndef RTAI
    /* Add the event fd */
    FD_SET(itti_dump_queue.event_fd, &read_set);

    /* Max of both sd */
    max_sd = itti_listen_socket > itti_dump_queue.event_fd ? itti_listen_socket : itti_dump_queue.event_fd;
+#else
+    max_sd = itti_listen_socket;
+#endif

    itti_dump_queue.itti_listen_socket = itti_listen_socket;

@@ -400,15 +491,26 @@ static void *itti_dump_socket(void *arg_p)
        int i;

        memcpy(&working_set, &read_set, sizeof(read_set));
+#ifdef RTAI
+        timeout.tv_sec  = 0;
+        timeout.tv_usec = 100000;
+
+        timeout_p = &timeout;
+#else
+        timeout_p = NULL;
+#endif

        /* No timeout: select blocks till a new event has to be handled
         * on sd's.
         */
-        rc = select(max_sd + 1, &working_set, NULL, NULL, NULL);
+        rc = select(max_sd + 1, &working_set, NULL, NULL, timeout_p);

        if (rc < 0) {
            ITTI_DUMP_ERROR("select failed (%d:%s)\n", errno, strerror(errno));
            pthread_exit(NULL);
+        } else if (rc == 0) {
+            /* Timeout */
+            itti_dump_flush_ring_buffer(1);
        }

        desc_ready = rc;
@@ -418,14 +520,11 @@ static void *itti_dump_socket(void *arg_p)
            {
                desc_ready -= 1;

+#ifndef RTAI
                if (i == itti_dump_queue.event_fd) {
                    /* Notification of new element to dump from other tasks */
                    uint64_t sem_counter;
                    ssize_t  read_ret;
-                    void    *user_data;
-                    int j;
-
-                    struct lfds611_freelist_element *element;

                    /* Read will always return 1 */
                    read_ret = read (itti_dump_queue.event_fd, &sem_counter, sizeof(sem_counter));
@@ -435,41 +534,12 @@ static void *itti_dump_socket(void *arg_p)
                    }
                    DevCheck(read_ret == sizeof(sem_counter), read_ret, sizeof(sem_counter), 0);

-                    /* Acquire the ring element */
-                    lfds611_ringbuffer_get_read_element(itti_dump_queue.itti_message_queue, &element);
-
-                    DevAssert(element != NULL);
-
-                    /* Retrieve user part of the message */
-                    lfds611_freelist_get_user_data_from_element(element, &user_data);
-
-                    if (((itti_dump_queue_item_t *)user_data)->message_type == ITTI_DUMP_EXIT_SIGNAL)
-                    {
-                        close(itti_dump_queue.event_fd);
-                        close(itti_dump_queue.itti_listen_socket);
-
-                        lfds611_ringbuffer_put_read_element(itti_dump_queue.itti_message_queue, element);
-
-                        /* Leave the thread as we detected end signal */
-                        pthread_exit(NULL);
-                    }
-
-                    /* Write message to file */
-                    itti_dump_fwrite_message((itti_dump_queue_item_t *)user_data);
-
-                    /* Send message to remote analyzer */
-                    for (j = 0; j < ITTI_DUMP_MAX_CON; j++) {
-                        if (itti_dump_queue.itti_clients[i].sd > 0) {
-                            itti_dump_send_message(itti_dump_queue.itti_clients[i].sd,
-                                                   (itti_dump_queue_item_t *)user_data);
-                        }
-                    }
-
-                    /* We have finished with this element, reinsert it in the ring buffer */
-                    lfds611_ringbuffer_put_read_element(itti_dump_queue.itti_message_queue, element);
+                    itti_dump_flush_ring_buffer(0);

                    ITTI_DUMP_DEBUG("Write element to file\n");
-                } else if (i == itti_listen_socket) {
+                } else
+#endif
+                if (i == itti_listen_socket) {
                    do {
                        client_socket = accept(itti_listen_socket, NULL, NULL);
                        if (client_socket < 0) {
@@ -581,11 +651,6 @@ int itti_dump_handle_new_connection(int sd, const char *xml_definition, uint32_t
    return 0;
 }

-int itti_dump_user_data_init_function(void **user_data, void *user_state)
-{
-    return 0;
-}
-
 /* This function should be called by each thread that will use the ring buffer */
 void itti_dump_thread_use_ring_buffer(void)
 {
@@ -597,7 +662,7 @@ int itti_dump_init(const char * const messages_definition_xml, const char * cons
    int i, ret;
    struct sched_param scheduler_param;

-    scheduler_param.sched_priority = 10;
+    scheduler_param.sched_priority = sched_get_priority_min(SCHED_FIFO) + 1;

    if (dump_file_name != NULL)
    {
@@ -637,6 +702,9 @@ int itti_dump_init(const char * const messages_definition_xml, const char * cons
        DevAssert(0 == 1);
    }

+#ifdef RTAI
+    itti_dump_queue.messages_in_queue = 0;
+#else
    itti_dump_queue.event_fd = eventfd(0, EFD_SEMAPHORE);
    if (itti_dump_queue.event_fd == -1)
    {
@@ -644,8 +712,8 @@ int itti_dump_init(const char * const messages_definition_xml, const char * cons
        /* Always assert on this condition */
        DevAssert(0 == 1);
    }
+#endif

-    itti_dump_queue.queue_size = 0;
    itti_dump_queue.nb_connected = 0;

    for(i = 0; i < ITTI_DUMP_MAX_CON; i++) {
@@ -660,7 +728,7 @@ int itti_dump_init(const char * const messages_definition_xml, const char * cons
        DevAssert(0 == 1);
    }

-    ret = pthread_attr_setschedpolicy(&itti_dump_queue.attr, SCHED_RR);
+    ret = pthread_attr_setschedpolicy(&itti_dump_queue.attr, SCHED_FIFO);
    if (ret < 0) {
        ITTI_DUMP_ERROR("pthread_attr_setschedpolicy (SCHED_IDLE) failed (%d:%s)\n", errno, strerror(errno));
        DevAssert(0 == 1);
@@ -681,7 +749,7 @@ int itti_dump_init(const char * const messages_definition_xml, const char * cons
    return 0;
 }

-void itti_dump_user_data_delete_function(void *user_data, void *user_state)
+static void itti_dump_user_data_delete_function(void *user_data, void *user_state)
 {
    if (user_data != NULL)
    {
@@ -56,9 +56,9 @@ const char * const messages_definition_xml = {

 /* Map task id to printable name. */
 const task_info_t tasks_info[] = {
-    {0, 0, 0, "TASK_UNKNOWN"},
-#define TASK_DEF(tHREADiD, pRIO, qUEUEsIZE)             { tHREADiD##_THREAD, pRIO, qUEUEsIZE, #tHREADiD },
-#define SUB_TASK_DEF(tHREADiD, sUBtASKiD, qUEUEsIZE)    { sUBtASKiD##_THREAD, 0, qUEUEsIZE, #sUBtASKiD },
+    {0, TASK_UNKNOWN, 0, 0, "TASK_UNKNOWN"},
+#define TASK_DEF(tHREADiD, pRIO, qUEUEsIZE)          { tHREADiD##_THREAD, TASK_UNKNOWN, pRIO, qUEUEsIZE, #tHREADiD },
+#define SUB_TASK_DEF(tHREADiD, sUBtASKiD, qUEUEsIZE) { sUBtASKiD##_THREAD, tHREADiD##_THREAD, 0, qUEUEsIZE, #sUBtASKiD },
 #include <tasks_def.h>
 #undef SUB_TASK_DEF
 #undef TASK_DEF
@@ -57,6 +57,7 @@

 /* Defines to extract task ID fields */
 #define TASK_GET_THREAD_ID(tASKiD)          (itti_desc.tasks_info[tASKiD].thread)
+#define TASK_GET_PARENT_TASK_ID(tASKiD)     (itti_desc.tasks_info[tASKiD].parent_task)
 /* Extract the instance from a message */
 #define ITTI_MESSAGE_GET_INSTANCE(mESSAGE)  ((mESSAGE)->ittiMsgHeader.instance)

@@ -80,6 +80,7 @@ do {                                                            \

 int timer_handle_signal(siginfo_t *info)
 {
+#if !defined(RTAI)
    struct timer_elm_s  *timer_p;
    MessageDef          *message_p;
    timer_has_expired_t *timer_expired_p;
@@ -121,6 +122,7 @@ int timer_handle_signal(siginfo_t *info)
        free(message_p);
        return -1;
    }
+#endif

    return 0;
 }
@@ -134,6 +136,7 @@ int timer_setup(
    void         *timer_arg,
    long         *timer_id)
 {
+#if !defined(RTAI)
    struct sigevent     se;
    struct itimerspec   its;
    struct timer_elm_s *timer_p;
@@ -201,14 +204,16 @@ int timer_setup(
    pthread_mutex_lock(&timer_desc.timer_list_mutex);
    STAILQ_INSERT_TAIL(&timer_desc.timer_queue, timer_p, entries);
    pthread_mutex_unlock(&timer_desc.timer_list_mutex);
+#endif

    return 0;
 }

 int timer_remove(long timer_id)
 {
-    struct timer_elm_s *timer_p;
    int rc = 0;
+#if !defined(RTAI)
+    struct timer_elm_s *timer_p;

    TMR_DEBUG("Removing timer 0x%lx\n", timer_id);

@@ -231,6 +236,7 @@ int timer_remove(long timer_id)
    }
    free(timer_p);
    timer_p = NULL;
+#endif
    return rc;
 }