Commit 9ee91703 authored by laurent's avatar laurent

fix non reentrant issue in itti

parent abf68010
...@@ -45,9 +45,7 @@ extern "C" { ...@@ -45,9 +45,7 @@ extern "C" {
std::vector<MessageDef *> message_queue; std::vector<MessageDef *> message_queue;
std::map<long,timer_elm_t> timer_map; std::map<long,timer_elm_t> timer_map;
uint64_t next_timer=UINT64_MAX; uint64_t next_timer=UINT64_MAX;
struct epoll_event *events =NULL;
int nb_fd_epoll=0; int nb_fd_epoll=0;
int nb_events=0;
int epoll_fd=-1; int epoll_fd=-1;
int sem_fd=-1; int sem_fd=-1;
} task_list_t; } task_list_t;
...@@ -167,8 +165,6 @@ extern "C" { ...@@ -167,8 +165,6 @@ extern "C" {
struct epoll_event event; struct epoll_event event;
task_list_t *t=tasks[task_id]; task_list_t *t=tasks[task_id];
t->nb_fd_epoll++; t->nb_fd_epoll++;
t->events = (struct epoll_event *)realloc((void *)t->events,
t->nb_fd_epoll * sizeof(struct epoll_event));
event.events = EPOLLIN | EPOLLERR; event.events = EPOLLIN | EPOLLERR;
event.data.u64 = 0; event.data.u64 = 0;
event.data.fd = fd; event.data.fd = fd;
...@@ -185,7 +181,7 @@ extern "C" { ...@@ -185,7 +181,7 @@ extern "C" {
t->nb_fd_epoll--; t->nb_fd_epoll--;
} }
static inline int itti_get_events_locked(task_id_t task_id, struct epoll_event **events) { static inline int itti_get_events_locked(task_id_t task_id, struct epoll_event *events, int nb_events) {
task_list_t *t=tasks[task_id]; task_list_t *t=tasks[task_id];
uint64_t current_time=0; uint64_t current_time=0;
...@@ -234,40 +230,39 @@ extern "C" { ...@@ -234,40 +230,39 @@ extern "C" {
pthread_mutex_unlock(&t->queue_cond_lock); pthread_mutex_unlock(&t->queue_cond_lock);
LOG_D(ITTI,"enter blocking wait for %s, timeout: %d ms\n", itti_get_task_name(task_id), epoll_timeout); LOG_D(ITTI,"enter blocking wait for %s, timeout: %d ms\n", itti_get_task_name(task_id), epoll_timeout);
t->nb_events = epoll_wait(t->epoll_fd,t->events,t->nb_fd_epoll, epoll_timeout); nb_events = epoll_wait(t->epoll_fd, events,t->nb_fd_epoll, epoll_timeout);
if ( t->nb_events < 0 && (errno == EINTR || errno == EAGAIN ) ) if ( nb_events < 0 && (errno == EINTR || errno == EAGAIN ) )
pthread_mutex_lock(&t->queue_cond_lock); pthread_mutex_lock(&t->queue_cond_lock);
} while (t->nb_events < 0 && (errno == EINTR || errno == EAGAIN ) ); } while (nb_events < 0 && (errno == EINTR || errno == EAGAIN ) );
AssertFatal (t->nb_events >=0, AssertFatal (nb_events >=0,
"epoll_wait failed for task %s, nb fds %d, timeout %lu: %s!\n", "epoll_wait failed for task %s, nb fds %d, timeout %lu: %s!\n",
itti_get_task_name(task_id), t->nb_fd_epoll, t->next_timer != UINT64_MAX ? t->next_timer-current_time : -1, strerror(errno)); itti_get_task_name(task_id), t->nb_fd_epoll, t->next_timer != UINT64_MAX ? t->next_timer-current_time : -1, strerror(errno));
LOG_D(ITTI,"receive on %d descriptors for %s\n", t->nb_events, itti_get_task_name(task_id)); LOG_D(ITTI,"receive on %d descriptors for %s\n", nb_events, itti_get_task_name(task_id));
if (t->nb_events == 0) if (nb_events == 0)
/* No data to read -> return */ /* No data to read -> return */
return 0; return 0;
for (int i = 0; i < t->nb_events; i++) { for (int i = 0; i < nb_events; 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 ((t->events[i].events & EPOLLIN) && if ((events[i].events & EPOLLIN) &&
(t->events[i].data.fd == t->sem_fd)) { (events[i].data.fd == t->sem_fd)) {
eventfd_t sem_counter; eventfd_t sem_counter;
/* Read will always return 1 */ /* Read will always return 1 */
AssertFatal( sizeof(sem_counter) == read (t->sem_fd, &sem_counter, sizeof(sem_counter)), ""); AssertFatal( sizeof(sem_counter) == read (t->sem_fd, &sem_counter, sizeof(sem_counter)), "");
/* Mark that the event has been processed */ /* Mark that the event has been processed */
t->events[i].events &= ~EPOLLIN; events[i].events &= ~EPOLLIN;
} }
} }
*events = t->events; return nb_events;
return t->nb_events;
} }
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, int nb_evts) {
pthread_mutex_lock(&tasks[task_id]->queue_cond_lock); pthread_mutex_lock(&tasks[task_id]->queue_cond_lock);
return itti_get_events_locked(task_id, events); return itti_get_events_locked(task_id, events, nb_evts);
} }
void itti_receive_msg(task_id_t task_id, MessageDef **received_msg) { void itti_receive_msg(task_id_t task_id, MessageDef **received_msg) {
...@@ -275,15 +270,16 @@ extern "C" { ...@@ -275,15 +270,16 @@ extern "C" {
task_list_t *t=tasks[task_id]; task_list_t *t=tasks[task_id];
pthread_mutex_lock(&t->queue_cond_lock); pthread_mutex_lock(&t->queue_cond_lock);
struct epoll_event events[t->nb_fd_epoll];
// Weird condition to deal with crap legacy itti interface // Weird condition to deal with crap legacy itti interface
if ( t->nb_fd_epoll == 1 ) { if ( t->nb_fd_epoll == 1 ) {
while (t->message_queue.empty()) { while (t->message_queue.empty()) {
itti_get_events_locked(task_id, &t->events); itti_get_events_locked(task_id, events, t->nb_fd_epoll);
pthread_mutex_lock(&t->queue_cond_lock); pthread_mutex_lock(&t->queue_cond_lock);
} }
} else { } else {
if (t->message_queue.empty()) { if (t->message_queue.empty()) {
itti_get_events_locked(task_id, &t->events); itti_get_events_locked(task_id, events, t->nb_fd_epoll);
pthread_mutex_lock(&t->queue_cond_lock); pthread_mutex_lock(&t->queue_cond_lock);
} }
} }
......
...@@ -479,7 +479,7 @@ void itti_unsubscribe_event_fd(task_id_t task_id, int fd); ...@@ -479,7 +479,7 @@ void itti_unsubscribe_event_fd(task_id_t task_id, int fd);
\param events events list \param events events list
@returns number of events to handle @returns number of events to handle
**/ **/
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, int nb_max_evts);
/** \brief Retrieves a message in the queue associated to task_id. /** \brief Retrieves a message in the queue associated to task_id.
If the queue is empty, the thread is blocked till a new message arrives. If the queue is empty, the thread is blocked till a new message arrives.
......
...@@ -1009,13 +1009,11 @@ int rrc_mac_config_req_eNB(module_id_t Mod_idP, ...@@ -1009,13 +1009,11 @@ int rrc_mac_config_req_eNB(module_id_t Mod_idP,
LOG_I(MAC, "[eNB %d][CONFIG] LTE_MBSFNAreaConfiguration_r9_t(%p) commonSF_AllocPeriod_r9(%d)\n",Mod_idP,mbms_AreaConfig, RC.mac[Mod_idP]->common_channels[0].commonSF_AllocPeriod_r9); LOG_I(MAC, "[eNB %d][CONFIG] LTE_MBSFNAreaConfiguration_r9_t(%p) commonSF_AllocPeriod_r9(%d)\n",Mod_idP,mbms_AreaConfig, RC.mac[Mod_idP]->common_channels[0].commonSF_AllocPeriod_r9);
for(i=0; i < mbms_AreaConfig->commonSF_Alloc_r9.list.count; i++){ for(i=0; i < mbms_AreaConfig->commonSF_Alloc_r9.list.count; i++){
RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i] = mbms_AreaConfig->commonSF_Alloc_r9.list.array[i]; RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i] = mbms_AreaConfig->commonSF_Alloc_r9.list.array[i];
LOG_I(RRC,"[eNB %d][CONFIG] MBSFNArea[%d] commonSF_Alloc_r9: radioframeAllocationPeriod(%ldn),radioframeAllocationOffset(%ld), subframeAllocation(%x,%x,%x)\n" LOG_I(RRC,"[eNB %d][CONFIG] MBSFNArea[%d] commonSF_Alloc_r9: radioframeAllocationPeriod(%ldn),radioframeAllocationOffset(%ld), subframeAllocation(%x)\n"
,Mod_idP,i ,Mod_idP,i
,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->radioframeAllocationPeriod ,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->radioframeAllocationPeriod
,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->radioframeAllocationOffset ,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->radioframeAllocationOffset
,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->subframeAllocation.choice.oneFrame.buf[0] ,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->subframeAllocation.choice.oneFrame.buf[0]);
,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->subframeAllocation.choice.oneFrame.buf[1]
,RC.mac[Mod_idP]->common_channels[0].commonSF_Alloc_r9_mbsfn_SubframeConfig[i]->subframeAllocation.choice.oneFrame.buf[2]);
} }
} }
......
...@@ -86,7 +86,6 @@ void nas_user_api_id_initialize(nas_user_t *user) { ...@@ -86,7 +86,6 @@ void nas_user_api_id_initialize(nas_user_t *user) {
void *nas_ue_task(void *args_p) void *nas_ue_task(void *args_p)
{ {
int nb_events; int nb_events;
struct epoll_event *events;
MessageDef *msg_p; MessageDef *msg_p;
instance_t instance; instance_t instance;
unsigned int Mod_id; unsigned int Mod_id;
...@@ -306,8 +305,8 @@ void *nas_ue_task(void *args_p) ...@@ -306,8 +305,8 @@ void *nas_ue_task(void *args_p)
AssertFatal (result == EXIT_SUCCESS, "Failed to free memory (%d)!\n", result); AssertFatal (result == EXIT_SUCCESS, "Failed to free memory (%d)!\n", result);
msg_p = NULL; msg_p = NULL;
} }
struct epoll_event events[20];
nb_events = itti_get_events(TASK_NAS_UE, &events); nb_events = itti_get_events(TASK_NAS_UE, events, 20);
if ((nb_events > 0) && (events != NULL)) { if ((nb_events > 0) && (events != NULL)) {
if (nas_ue_process_events(users, events, nb_events) == true) { if (nas_ue_process_events(users, events, nb_events) == true) {
......
...@@ -216,7 +216,7 @@ sctp_eNB_accept_associations_multi( ...@@ -216,7 +216,7 @@ sctp_eNB_accept_associations_multi(
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void static void
sctp_handle_new_association_req_multi( sctp_handle_new_association_req_multi(
const instance_t instance, const instance_t instance,
const task_id_t requestor, const task_id_t requestor,
...@@ -355,7 +355,7 @@ sctp_handle_new_association_req_multi( ...@@ -355,7 +355,7 @@ sctp_handle_new_association_req_multi(
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void static void
sctp_handle_new_association_req( sctp_handle_new_association_req(
const instance_t instance, const instance_t instance,
const task_id_t requestor, const task_id_t requestor,
...@@ -617,7 +617,7 @@ sctp_handle_new_association_req( ...@@ -617,7 +617,7 @@ sctp_handle_new_association_req(
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void sctp_send_data( static void sctp_send_data(
instance_t instance, instance_t instance,
task_id_t task_id, task_id_t task_id,
sctp_data_req_t *sctp_data_req_p) sctp_data_req_t *sctp_data_req_p)
...@@ -1058,16 +1058,13 @@ sctp_eNB_read_from_socket( ...@@ -1058,16 +1058,13 @@ sctp_eNB_read_from_socket(
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void void
sctp_eNB_flush_sockets( static sctp_eNB_flush_sockets(
struct epoll_event *events, int nb_events) struct epoll_event *events, int nb_events)
{ {
printf("entre sctp_eNB_flush_sockets %p\n",events);
int i; int i;
struct sctp_cnx_list_elm_s *sctp_cnx = NULL; struct sctp_cnx_list_elm_s *sctp_cnx = NULL;
if (events == NULL) {
return;
}
for (i = 0; i < nb_events; i++) { for (i = 0; i < nb_events; i++) {
sctp_cnx = sctp_get_cnx(-1, events[i].data.fd); sctp_cnx = sctp_get_cnx(-1, events[i].data.fd);
...@@ -1090,7 +1087,7 @@ sctp_eNB_flush_sockets( ...@@ -1090,7 +1087,7 @@ sctp_eNB_flush_sockets(
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void sctp_eNB_init(void) static void sctp_eNB_init(void)
{ {
SCTP_DEBUG("Starting SCTP layer\n"); SCTP_DEBUG("Starting SCTP layer\n");
...@@ -1100,10 +1097,9 @@ void sctp_eNB_init(void) ...@@ -1100,10 +1097,9 @@ void sctp_eNB_init(void)
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
void *sctp_eNB_process_itti_msg(void *notUsed) static void sctp_eNB_process_itti_msg()
{ {
int nb_events; int nb_events;
struct epoll_event *events;
MessageDef *received_msg = NULL; MessageDef *received_msg = NULL;
int result; int result;
...@@ -1184,12 +1180,12 @@ void *sctp_eNB_process_itti_msg(void *notUsed) ...@@ -1184,12 +1180,12 @@ void *sctp_eNB_process_itti_msg(void *notUsed)
AssertFatal (result == EXIT_SUCCESS, "Failed to free memory (%d)!\n", result); AssertFatal (result == EXIT_SUCCESS, "Failed to free memory (%d)!\n", result);
received_msg = NULL; received_msg = NULL;
} }
struct epoll_event events[20];
nb_events = itti_get_events(TASK_SCTP, &events); nb_events = itti_get_events(TASK_SCTP, events, 20);
/* Now handle notifications for other sockets */ /* Now handle notifications for other sockets */
sctp_eNB_flush_sockets(events, nb_events); sctp_eNB_flush_sockets(events, nb_events);
return NULL; return;
} }
//------------------------------------------------------------------------------ //------------------------------------------------------------------------------
...@@ -1198,7 +1194,7 @@ void *sctp_eNB_task(void *arg) ...@@ -1198,7 +1194,7 @@ void *sctp_eNB_task(void *arg)
sctp_eNB_init(); sctp_eNB_init();
while (1) { while (1) {
(void) sctp_eNB_process_itti_msg(NULL); sctp_eNB_process_itti_msg();
} }
return NULL; return NULL;
......
...@@ -22,8 +22,6 @@ ...@@ -22,8 +22,6 @@
#ifndef SCTP_ENB_TASK_H_ #ifndef SCTP_ENB_TASK_H_
#define SCTP_ENB_TASK_H_ #define SCTP_ENB_TASK_H_
void sctp_eNB_init(void);
void *sctp_eNB_process_itti_msg(void *);
void *sctp_eNB_task(void *arg); void *sctp_eNB_task(void *arg);
#endif /* SCTP_ENB_TASK_H_ */ #endif /* SCTP_ENB_TASK_H_ */
...@@ -1278,9 +1278,8 @@ void *gtpv1uTask(void *args) { ...@@ -1278,9 +1278,8 @@ void *gtpv1uTask(void *args) {
AssertFatal(EXIT_SUCCESS==itti_free(TASK_GTPV1_U, message_p), "Failed to free memory!\n"); AssertFatal(EXIT_SUCCESS==itti_free(TASK_GTPV1_U, message_p), "Failed to free memory!\n");
} }
struct epoll_event *events; struct epoll_event events[20];
int nb_events = itti_get_events(TASK_GTPV1_U, events, 20);
int nb_events = itti_get_events(TASK_GTPV1_U, &events);
for (int i = 0; i < nb_events; i++) for (int i = 0; i < nb_events; i++)
if ((events[i].events&EPOLLIN)) if ((events[i].events&EPOLLIN))
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment