1. 简述

OAI各个模块拥有自己的消息队列，当其他模块需要向该模块发送消息时，只需将封装好的message压入对端模块队列，本模块进行消息接收时，从本模块队列依次取出message，进行解析。

OAI模块比较多，消息收发统一调用itti模块接口函数来进行，通过函数解析来了解OAI的消息收发机制。

2. 模块消息接收循环

while (1) { // Wait for a message itti_receive_msg(TASK_RRC_GNB, &msg_p); msg_name_p = ITTI_MSG_NAME(msg_p); instance = ITTI_MSG_DESTINATION_INSTANCE(msg_p); switch (ITTI_MSG_ID(msg_p)) { case NR_RRC_MAC_CCCH_DATA_IND: ...... nr_rrc_gNB_decode_ccch(&ctxt, (uint8_t *)NR_RRC_MAC_CCCH_DATA_IND(msg_p).sdu, NR_RRC_MAC_CCCH_DATA_IND(msg_p).sdu_size, NR_RRC_MAC_CCCH_DATA_IND(msg_p).du_to_cu_rrc_container, NR_RRC_MAC_CCCH_DATA_IND(msg_p).CC_id); break; ...... } }

上面是RRC模块主线程函数的消息接收代码，由于代码比较长，下面switch处理就不全部列出来了，OAI其他模块都一样，均使用while(1)来循环接收消息

itti_receive_msg(TASK_RRC_GNB, &msg_p)：负责接收消息，下面会进行介绍

ITTI_MSG_NAME(msg_p)：解析获取消息名称

#defineITTI_MSG_NAME(mSGpTR)itti_get_message_name(ITTI_MSG_ID(mSGpTR))constchar*itti_get_message_name(MessagesIds message_id){returnmessages_info[message_id].name;}

ITTI_MSG_DESTINATION_INSTANCE(msg_p)：目前没发现用途

ITTI_MSG_ID(msg_p)：解析获取消息ID，通过消息ID来判断该消息应进入哪一种处理流程。

#defineITTI_MSG_ID(mSGpTR)((mSGpTR)->ittiMsgHeader.messageId)

3. itti消息接收

voiditti_receive_msg(task_id_ttask_id,MessageDef**received_msg){// Reception of one message, blocking callertask_list_t*t=tasks[task_id];pthread_mutex_lock(&t->queue_cond_lock);// Weird condition to deal with crap legacy itti interfaceif(t->nb_fd_epoll==1){while(t->message_queue.empty()){itti_get_events_locked(task_id,&t->events);pthread_mutex_lock(&t->queue_cond_lock);}}else{if(t->message_queue.empty()){itti_get_events_locked(task_id,&t->events);pthread_mutex_lock(&t->queue_cond_lock);}}// Legacy design: we return even if we have no message// in this case, *received_msg is NULLif(t->message_queue.empty()){*received_msg=NULL;LOG_D(TMR,"task %s received even from other fd (total fds: %d), returning msg NULL\n",t->admin.name,t->nb_fd_epoll);}else{*received_msg=t->message_queue.back();t->message_queue.pop_back();LOG_D(TMR,"task %s received a message\n",t->admin.name);}pthread_mutex_unlock(&t->queue_cond_lock);}

入参

task_id：实体id，也就是模块id，OAI将MAC,RLC,PDCP,RRC等模块进行了编号，在使用时根据索引号可快速确认当前操作属于哪个模块；

received_msg：消息指针，这里传入的是地址，从队列取出的消息放入该段内存

函数内部

tasks[task_id]：这里上篇已经提及，每一个模块实体都有一个task list，实体的各种属性存放于tasks中；

pthread_mutex_lock(&t->queue_cond_lock)：线程锁，在同一时间可能有多个模块向该模块队列发送消息，同时本模块也在取消息，为保证数据安全，读写操作需要加锁

itti_get_events_locked(task_id, &t->events)：如果模块队列是空的，去处理残留的定时器

*received_msg=NULL：队列为空，将received_msg置空

*received_msg=t->message_queue.back()：取出实体队列中最后一条消息，置给received_msg，这就是这一次消息接收所获得的数据。

t->message_queue.pop_back()：删除队尾元素，队列长度-1

pthread_mutex_unlock (&t->queue_cond_lock)：接收完毕，进行解锁

4. itti消息发送

模块在发送消息时调用itti_send_msg_to_task()进行发送，该函数比较简单，就不作说明，看一下它所调用的itti_send_msg_to_task()。

staticinlineintitti_send_msg_to_task(task_id_tdestination_task_id,instance_tdestinationInstance,MessageDef*message){task_list_t*t=tasks[destination_task_id];message->ittiMsgHeader.destinationTaskId=destination_task_id;message->ittiMsgHeader.destinationInstance=destinationInstance;message->ittiMsgHeader.lte_time.frame=0;message->ittiMsgHeader.lte_time.slot=0;intmessage_id=message->ittiMsgHeader.messageId;size_ts=t->message_queue.size();if(s>t->admin.queue_size)LOG_E(TMR,"Queue for %s task contains %ld messages\n",itti_get_task_name(destination_task_id),s);if(s>50)LOG_I(TMR,"Queue for %s task size: %ld\n",itti_get_task_name(destination_task_id),s+1);t->message_queue.insert(t->message_queue.begin(),message);eventfd_tsem_counter=1;AssertFatal(sizeof(sem_counter)==write(t->sem_fd,&sem_counter,sizeof(sem_counter)),"");LOG_D(TMR,"sent messages id=%d to %s\n",message_id,t->admin.name);return0;}

入参

destination_task_id：目标模块ID，指示message发往哪个模块实体

destinationInstance：目前没发现用途

message：需要发送的消息体

函数内部

当s > t->admin.queue_size时，给出报错，s为模块消息队列实际的元素个数，t->admin.queue_size为模块对队列元素的使用计数，前者比后者大，证明消息收发不对称。本意是好的，但实际OAI虽然定义了admin.queue_size，但并没有在程序中用到。

t->message_queue.insert(t->message_queue.begin(), message)：将message插入到该模块队列的message_queue.begin()前，begin为头部元素，即将message插入队首。