Appearance
线程间通信
一般可以使用全局变量以及线程间同步进行实现
RT-Thread也提供了一部分的通信机制
邮箱
一个线程发送, 另外的线程接受信息, 进行处理
使用邮箱的时候每一次只能发送一个四字节的数据(32位处理器),特点是开销比较低，效率较高
可以发送一个地址从而达到发送多个数据的目的
非阻塞方式的邮件发送过程能够安全的应用于中断服务中, 发送以及接受信息的时候可以使用阻塞的模式
邮箱有一个缓存区, 使用rt_mailbox_t进行控制
实际使用
创建(初始化)
c
/** 动态的方式创建
* This function will create a mailbox object from system resource
*
* @param name the name of mailbox 记录一个名字
* @param size the size of mailbox 记录一下缓存区的大小
* @param flag the flag of mailbox 一个标志位
*
* @return the created mailbox, RT_NULL on error happen
*/
rt_mailbox_t rt_mb_create(const char *name, rt_size_t size, rt_uint8_t flag)
/** 静态的方式创建
* This function will initialize a mailbox and put it under control of resource
* management.
*
* @param mb the mailbox object 邮箱的句柄
* @param name the name of mailbox 名字
* @param msgpool the begin address of buffer to save received mail 缓存区的地址
* @param size the size of mailbox 缓冲区大小
* @param flag the flag of mailbox 标志
*
* @return the operation status, RT_EOK on successful
*/
rt_err_t rt_mb_init(rt_mailbox_t mb,
const char *name,
void *msgpool,
rt_size_t size,
rt_uint8_t flag)这一个标志位可以为RT_IPC_FLAG_FIFO或RT_IPC_FLAG_PRIO, 设置的是挂起任务被释放的时候是按照进入的顺序先进入的先出去还是优先级比较高的先出去
删除
c
/**动态
* This function will delete a mailbox object and release the memory
*
* @param mb the mailbox object
*
* @return the error code
*/
rt_err_t rt_mb_delete(rt_mailbox_t mb)
/**静态
* This function will detach a mailbox from resource management
*
* @param mb the mailbox object
*
* @return the operation status, RT_EOK on successful
*/
rt_err_t rt_mb_detach(rt_mailbox_t mb)发送邮件
c
/**
* This function will send a mail to mailbox object, if there are threads
* suspended on mailbox object, it will be waked up. This function will return
* immediately, if you want blocking send, use rt_mb_send_wait instead.
*
* @param mb the mailbox object
* @param value the mail 要发送的数据
*
* @return the error code
*/
rt_err_t rt_mb_send(rt_mailbox_t mb, rt_ubase_t value)这是一个不等待的时钟发送函数
c
/**
* This function will send a mail to mailbox object. If the mailbox is full,
* current thread will be suspended until timeout.
*
* @param mb the mailbox object
* @param value the mail
* @param timeout the waiting time 多了一个等待时间
*
* @return the error code
*/
rt_err_t rt_mb_send_wait(rt_mailbox_t mb,
rt_ubase_t value,
rt_int32_t timeout)接收
c
/**
* This function will receive a mail from mailbox object, if there is no mail
* in mailbox object, the thread shall wait for a specified time.
*
* @param mb the mailbox object
* @param value the received mail will be saved in 给出一个存放收到的数据的位置
* @param timeout the waiting time
*
* @return the error code
*/
rt_err_t rt_mb_recv(rt_mailbox_t mb, rt_ubase_t *value, rt_int32_t timeout)使用技巧
可以使用一个这样的结构体, 每次发送这一个结构体的地址, 进行大于32字节的数据发送
c
struct msg
{
uint32_t *data;
uint32_t size;
}消息队列
是邮箱的扩展, 没有4字节的限制
消息队列能够接收来自线程或中断服务例程中不固定长度的消息，并把消息缓存在自己的内存空间中。这些消息使用的是链表进行连接, 消息先进先出
这一个使用的拷贝的模式进行传输, 不建议直接发送大量数据(可以发送一个地址)
可以用于发送不定长的数据, 实际使用的时候可以使用消息队列发送消息, 使用邮箱表示接收到数据了
实际使用
创建
c
/**动态
* This function will create a message queue object from system resource
*
* @param name the name of message queue 名字
* @param msg_size the size of message 每一个消息的大小(字节)
* @param max_msgs the maximum number of message in queue 记录一下消息的最大的个数
* @param flag the flag of message queue 一个标志
*
* @return the created message queue, RT_NULL on error happen
*/
rt_mq_t rt_mq_create(const char *name,
rt_size_t msg_size,
rt_size_t max_msgs,
rt_uint8_t flag)
/**静态
* This function will initialize a message queue and put it under control of
* resource management.
*
* @param mq the message object 对象的句柄
* @param name the name of message queue
* @param msgpool the beginning address of buffer to save messages 缓冲区的地址, 动态申请
* 的时候这个的大小是 (一个数据的大小+sizeof(struct rt_mq_message)) * mq->max_msgs
* @param msg_size the maximum size of message一个消息的大小
* @param pool_size the size of buffer to save messages 缓冲区的大小
* @param flag the flag of message queue
*
* @return the operation status, RT_EOK on successful
*/
rt_err_t rt_mq_init(rt_mq_t mq,
const char *name,
void *msgpool,
rt_size_t msg_size,
rt_size_t pool_size,
rt_uint8_t flag)这一个标志位可以为RT_IPC_FLAG_FIFO或RT_IPC_FLAG_PRIO, 设置的是挂起任务被释放的时候是按照进入的顺序先进入的先出去还是优先级比较高的先出去
删除
c
/**动态
* This function will delete a message queue object and release the memory
*
* @param mq the message queue object
*
* @return the error code
*/
rt_err_t rt_mq_delete(rt_mq_t mq)
/**静态
* This function will detach a message queue object from resource management
*
* @param mq the message queue object
*
* @return the operation status, RT_EOK on successful
*/
rt_err_t rt_mq_detach(rt_mq_t mq)发送消息
c
/**
* This function will send a message to message queue object, if there are
* threads suspended on message queue object, it will be waked up.
*
* @param mq the message queue object
* @param buffer the message 发送的消息的地址
* @param size the size of buffer 发送的数据的大小
*
* @return the error code
*/
rt_err_t rt_mq_send(rt_mq_t mq, const void *buffer, rt_size_t size)不等待
c
/**
* This function will send a message to message queue object. If the message queue is full,
* current thread will be suspended until timeout.
*
* @param mq the message queue object
* @param buffer the message
* @param size the size of buffer
* @param timeout the waiting time
*
* @return the error code
*/
rt_err_t rt_mq_send_wait(rt_mq_t mq,
const void *buffer,
rt_size_t size,
rt_int32_t timeout)等待
c
/**
* This function will send an urgent message to message queue object, which
* means the message will be inserted to the head of message queue. If there
* are threads suspended on message queue object, it will be waked up.
*
* @param mq the message queue object
* @param buffer the message
* @param size the size of buffer
*
* @return the error code
*/
rt_err_t rt_mq_urgent(rt_mq_t mq, const void *buffer, rt_size_t size)发送一个紧急的消息, 这一个消息会直接放到队首
接收
c
/**
* This function will receive a message from message queue object, if there is
* no message in message queue object, the thread shall wait for a specified
* time.
*
* @param mq the message queue object
* @param buffer the received message will be saved in 接收到的数据
* @param size the size of buffer
* @param timeout the waiting time
*
* @return the error code
*/
rt_err_t rt_mq_recv(rt_mq_t mq,
void *buffer,
rt_size_t size,
rt_int32_t timeout)信号(软件中断信号)
**注: **信号这块应该是要在微内核里使用，如果你是用宏内核版本，不推荐使用信号功能。
POSIX标准定义了sigset_t类型来定义一个信号集, 实际是一个unsigned long类型的数据, 应用程序能够使用的信号为SIGUSR1（10）和SIGUSR2（12）
他的本质是一个软件中断
收到信号的线程实际的处理方法有三种
- 类似中断的处理程序，对于需要处理的信号，线程可以指定处理函数，由该函数来处理。
- 忽略某个信号，对该信号不做任何处理，就像未发生过一样。
- 对该信号的处理保留系统的默认值。
需要定义RT_USING_SIGNALS这一个宏
实际使用
安装信号
如果线程要处理某一信号，那么就要在线程中安装该信号。
主要用来确定信号值及线程针对该信号值的动作之间的映射关系，即线程将要处理哪个信号，该信号被传递给线程时，将执行何种操作。
c
rt_sighandler_t rt_signal_install(int signo, rt_sighandler_t handler)(这一个函数是给现在的线程安装)
signo信号值（只有SIGUSR1和SIGUSR2是开放给用户使用的）
handler设置对信号值的处理方式, 这一个的实际的函数是void (*rt_sighandler_t)(int signo);
也可以使用SIG_IGN，忽略某个信号, SIG_DFL，系统会调用默认的处理函数_signal_default_handler()
返回安装信号前的handler值表示成功
阻塞(屏蔽)信号
该信号将不会递达给安装此信号的线程，也不会引发软中断处理。
c
void rt_signal_mask(int signo)解除信号阻塞
c
void rt_signal_unmask(int signo)发送信号
c
int rt_thread_kill(rt_thread_t tid, int sig)tid: 接收信号的线程
sig: 信号值
等待信号
c
int rt_signal_wait(const rt_sigset_t *set, rt_siginfo_t *si, rt_int32_t timeout)set: 指定等待的信号
si: 指向存储等到信号信息的指针
RT-Thread代码分析
邮箱
实际是实现是一个对环形缓存区的使用
c
struct rt_mailbox
{
struct rt_ipc_object parent; /**< inherit from ipc_object */
rt_ubase_t *msg_pool; /**< start address of
message buffer 缓存区地址*/
rt_uint16_t size; /**< size of message pool 大小*/
rt_uint16_t entry; /**< index of messages in msg_pool
记录一下数量*/
rt_uint16_t in_offset; /**< input offset of the message buffer记录一下写的位置 */
rt_uint16_t out_offset; /**< output offset of the message 记录一下读的位置buffer */
rt_list_t suspend_sender_thread; /**< sender thread suspended on
挂起的任务链表this mailbox */
};
typedef struct rt_mailbox *rt_mailbox_t;c
struct rt_ipc_object
{
struct rt_object parent; /**< inherit from rt_object */
rt_list_t suspend_thread; /**< threads pended on this resource 记录接收的任务的队列 */
};创建
c
rt_mailbox_t rt_mb_create(const char *name, rt_size_t size, rt_uint8_t flag)
{
rt_mailbox_t mb;
RT_DEBUG_NOT_IN_INTERRUPT;
/* allocate object 这一个在对象创建的那一篇里面分析过了*/
mb = (rt_mailbox_t)rt_object_allocate(RT_Object_Class_MailBox, name);
if (mb == RT_NULL)
return mb;
/* set parent */
mb->parent.parent.flag = flag;
/* initialize ipc object 实际上就是初始化一个用于记录接收任务挂起的链表 */
rt_ipc_object_init(&(mb->parent));
/* initialize mailbox 记录一些信息 */
mb->size = size;
//获取一个缓存区
mb->msg_pool = (rt_ubase_t *)RT_KERNEL_MALLOC(mb->size * sizeof(rt_ubase_t));
if (mb->msg_pool == RT_NULL)
{
//获取失败
/* delete mailbox object */
rt_object_delete(&(mb->parent.parent));
return RT_NULL;
}
mb->entry = 0;
mb->in_offset = 0;
mb->out_offset = 0;
/* initialize an additional list of sender suspend thread 发送者的队列 */
rt_list_init(&(mb->suspend_sender_thread));
return mb;
}c
//实际就是记录一下信息
rt_err_t rt_mb_init(rt_mailbox_t mb,
const char *name,
void *msgpool,
rt_size_t size,
rt_uint8_t flag)
{
RT_ASSERT(mb != RT_NULL);
/* initialize object 具体看对象的分析那一章*/
rt_object_init(&(mb->parent.parent), RT_Object_Class_MailBox, name);
/* set parent flag */
mb->parent.parent.flag = flag;
/* initialize ipc object */
rt_ipc_object_init(&(mb->parent));
/* initialize mailbox */
mb->msg_pool = (rt_ubase_t *)msgpool;
mb->size = size;
mb->entry = 0;
mb->in_offset = 0;
mb->out_offset = 0;
/* initialize an additional list of sender suspend thread */
rt_list_init(&(mb->suspend_sender_thread));
return RT_EOK;
}删除
c
rt_err_t rt_mb_delete(rt_mailbox_t mb)
{
RT_DEBUG_NOT_IN_INTERRUPT;
/* resume all suspended thread 把所有的挂起的接收线程释放, 这个同步那一章里面有 */
rt_ipc_list_resume_all(&(mb->parent.suspend_thread));
/* also resume all mailbox private suspended thread 发送线程释放 */
rt_ipc_list_resume_all(&(mb->suspend_sender_thread));
/* free mailbox pool 释放缓存内存 */
RT_KERNEL_FREE(mb->msg_pool);
/* delete mailbox object 控制块会在空闲任务释放 */
rt_object_delete(&(mb->parent.parent));
return RT_EOK;
}发送
c
rt_err_t rt_mb_send(rt_mailbox_t mb, rt_ubase_t value)
{
return rt_mb_send_wait(mb, value, 0);
}c
rt_err_t rt_mb_send_wait(rt_mailbox_t mb,
rt_ubase_t value,
rt_int32_t timeout)
{
struct rt_thread *thread;
register rt_ubase_t temp;
rt_uint32_t tick_delta;
/* initialize delta tick */
tick_delta = 0;
/* get current thread 获取当前线程 */
thread = rt_thread_self();
//回调函数
RT_OBJECT_HOOK_CALL(rt_object_put_hook, (&(mb->parent.parent)));
/* disable interrupt 临界区 */
temp = rt_hw_interrupt_disable();
/* for non-blocking call 看一看有没有位置 */
if (mb->entry == mb->size && timeout == 0)
{
//没有并且不等待
rt_hw_interrupt_enable(temp);
return -RT_EFULL;
}
/* mailbox is full */
while (mb->entry == mb->size)
{
//邮箱满了
/* reset error number in thread */
thread->error = RT_EOK;
/* no waiting, return timeout */
if (timeout == 0)
{
/* enable interrupt 这一个线程等待时间到了 */
rt_hw_interrupt_enable(temp);
return -RT_EFULL;
}
//这个使用的时候需要已经开始调度了
RT_DEBUG_IN_THREAD_CONTEXT;
/* suspend current thread */
//把这一个任务记录在挂起队列里面(具体分析看任务同步的那一篇)
rt_ipc_list_suspend(&(mb->suspend_sender_thread),
thread,
mb->parent.parent.flag);
/* has waiting time, start thread timer */
if (timeout > 0)
{
//需要等待
/* get the start tick of timer 获取现在的时间 */
tick_delta = rt_tick_get();
RT_DEBUG_LOG(RT_DEBUG_IPC, ("mb_send_wait: start timer of thread:%s\n",
thread->name));
/* reset the timeout of thread timer and start it 开启一个时钟用于超时时候的唤醒*/
rt_timer_control(&(thread->thread_timer),
RT_TIMER_CTRL_SET_TIME,
&timeout);
rt_timer_start(&(thread->thread_timer));
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* re-schedule 启动一次调度 */
rt_schedule();
//这里线程被唤醒, 看一看这这时候是不是超时了
/* resume from suspend state */
if (thread->error != RT_EOK)
{
/* return error */
return thread->error;
}
/* disable interrupt 在这之前可能有一个高优先级把这一个位置又使用了, 需要再看看可不可以发送*/
temp = rt_hw_interrupt_disable();
/* if it's not waiting forever and then re-calculate timeout tick 跟新一下发送的时间, 再试着发送一次 */
if (timeout > 0)
{
tick_delta = rt_tick_get() - tick_delta;
timeout -= tick_delta;
//时间已经到了
if (timeout < 0)
timeout = 0;
}
}
//可以发送信息
/* set ptr 记录一下信息*/
mb->msg_pool[mb->in_offset] = value;
/* increase input offset 更新一下记录的指针 */
++ mb->in_offset;
if (mb->in_offset >= mb->size)
mb->in_offset = 0;//大小大于这一个环形缓冲区, 回头部
/* increase message entry */
mb->entry ++;//大小加一
/* resume suspended thread 看一看接收的有没有在等的 */
if (!rt_list_isempty(&mb->parent.suspend_thread))
{
//唤醒一下第一个等待的线程以及把时钟关了(看线程同步)
rt_ipc_list_resume(&(mb->parent.suspend_thread));
/* enable interrupt */
rt_hw_interrupt_enable(temp);
//切换一下任务
rt_schedule();
return RT_EOK;
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
return RT_EOK;
}接收
c
//基本和发送一样, 只是循环判断是为空, 以及标识符加减反过来了
rt_err_t rt_mb_recv(rt_mailbox_t mb, rt_ubase_t *value, rt_int32_t timeout)
{
struct rt_thread *thread;
register rt_ubase_t temp;
rt_uint32_t tick_delta;
/* parameter check */
RT_ASSERT(mb != RT_NULL);
RT_ASSERT(rt_object_get_type(&mb->parent.parent) == RT_Object_Class_MailBox);
/* initialize delta tick */
tick_delta = 0;
/* get current thread */
thread = rt_thread_self();
RT_OBJECT_HOOK_CALL(rt_object_trytake_hook, (&(mb->parent.parent)));
/* disable interrupt */
temp = rt_hw_interrupt_disable();
/* for non-blocking call */
if (mb->entry == 0 && timeout == 0)
{
rt_hw_interrupt_enable(temp);
return -RT_ETIMEOUT;
}
/* mailbox is empty 看看是不是空的 */
while (mb->entry == 0)
{
//需要等待
/* reset error number in thread */
thread->error = RT_EOK;
/* no waiting, return timeout */
if (timeout == 0)
{
//等待的时间到了, 失败
/* enable interrupt */
rt_hw_interrupt_enable(temp);
thread->error = -RT_ETIMEOUT;
return -RT_ETIMEOUT;
}
//必须打开调度器了
RT_DEBUG_IN_THREAD_CONTEXT;
//一个典型的icp挂起处理
/* suspend current thread */
rt_ipc_list_suspend(&(mb->parent.suspend_thread),
thread,
mb->parent.parent.flag);
/* has waiting time, start thread timer 还有时间, 挂起*/
if (timeout > 0)
{
/* get the start tick of timer */
tick_delta = rt_tick_get();
RT_DEBUG_LOG(RT_DEBUG_IPC, ("mb_recv: start timer of thread:%s\n",
thread->name));
/* reset the timeout of thread timer and start it */
rt_timer_control(&(thread->thread_timer),
RT_TIMER_CTRL_SET_TIME,
&timeout);
rt_timer_start(&(thread->thread_timer));
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* re-schedule */
rt_schedule();
//切换, 看一看是不是出错了(超时等)
/* resume from suspend state */
if (thread->error != RT_EOK)
{
/* return error */
return thread->error;
}
/* disable interrupt 最后检测一下有没有位置 */
temp = rt_hw_interrupt_disable();
/* if it's not waiting forever and then re-calculate timeout tick 更新时间 */
if (timeout > 0)
{
tick_delta = rt_tick_get() - tick_delta;
timeout -= tick_delta;
if (timeout < 0)
timeout = 0;
}
}
/* fill ptr */
*value = mb->msg_pool[mb->out_offset];
/* increase output offset */
++ mb->out_offset;
if (mb->out_offset >= mb->size)
mb->out_offset = 0;
/* decrease message entry */
mb->entry --;
/* resume suspended thread */
if (!rt_list_isempty(&(mb->suspend_sender_thread)))
{
//释放第一个线程
rt_ipc_list_resume(&(mb->suspend_sender_thread));
/* enable interrupt */
rt_hw_interrupt_enable(temp);
RT_OBJECT_HOOK_CALL(rt_object_take_hook, (&(mb->parent.parent)));
rt_schedule();
return RT_EOK;
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
RT_OBJECT_HOOK_CALL(rt_object_take_hook, (&(mb->parent.parent)));
return RT_EOK;
}消息队列
实际上这是一个对链表的使用
消息链表: 使用两个指针记录消息链表的头以及尾
空闲链表: 类似栈, 使用一个指针记录空闲任务尾部
c
struct rt_messagequeue
{
struct rt_ipc_object parent; /**< inherit from ipc_object */
void *msg_pool; /**< start address of message
queue 存放消息的缓冲区 */
rt_uint16_t msg_size; /**< message size of each message
记录可以容纳的每一个消息的大小*/
rt_uint16_t max_msgs; /**< max number of messages
记录消息的个数*/
rt_uint16_t entry; /**< index of messages in the queue 记录现在消息的个数 */
void *msg_queue_head; /**< list head 链表头 */
void *msg_queue_tail; /**< list tail 链表尾*/
void *msg_queue_free; /**< pointer indicated the free node of queue 记录缓冲区里下一个空闲消息的链表 */
rt_list_t suspend_sender_thread; /**< sender thread suspended on this message queue 发送线程的挂起的等待队列*/
};c
struct rt_mq_message
{
struct rt_mq_message *next;
};一个管理信息的链表
创建
初始化以后的缓冲区
c
rt_mq_t rt_mq_create(const char *name,
rt_size_t msg_size,
rt_size_t max_msgs,
rt_uint8_t flag)
{
struct rt_messagequeue *mq;
struct rt_mq_message *head;
register rt_base_t temp;
//这一个函数不应该在中断里面使用
RT_DEBUG_NOT_IN_INTERRUPT;
/* allocate object 获取一个邮箱的对象 */
mq = (rt_mq_t)rt_object_allocate(RT_Object_Class_MessageQueue, name);
if (mq == RT_NULL)
return mq;
/* set parent 记录一下标志 */
mq->parent.parent.flag = flag;
/* initialize ipc object 主要是初始化一个ipc的链表(这一部分在信号同步的的那一篇里面有) */
rt_ipc_object_init(&(mq->parent));
/* initialize message queue */
/* get correct message size 把这一个按照四字节对齐 */
mq->msg_size = RT_ALIGN(msg_size, RT_ALIGN_SIZE);
mq->max_msgs = max_msgs;
/* allocate message pool 获取一个存信息的内存, 实际的大小是(信息大小 + 管理结构体(一个链表)) * 数量 */
mq->msg_pool = RT_KERNEL_MALLOC((mq->msg_size + sizeof(struct rt_mq_message)) * mq->max_msgs);
if (mq->msg_pool == RT_NULL)
{
//获取失败的时候
rt_object_delete(&(mq->parent.parent));
return RT_NULL;
}
/* initialize message list */
mq->msg_queue_head = RT_NULL;
mq->msg_queue_tail = RT_NULL;
/* initialize message empty list 初始化这一个缓存里面的信息 */
mq->msg_queue_free = RT_NULL;
for (temp = 0; temp < mq->max_msgs; temp ++)
{
head = (struct rt_mq_message *)((rt_uint8_t *)mq->msg_pool +
temp * (mq->msg_size + sizeof(struct rt_mq_message)));//计算一下第n个信息的位置
head->next = (struct rt_mq_message *)mq->msg_queue_free;//初始化链表指向数组前一个信息
mq->msg_queue_free = head;//更新一下可以使用的下一个的位置
}
/* the initial entry is zero */
mq->entry = 0;
/* initialize an additional list of sender suspend thread */
rt_list_init(&(mq->suspend_sender_thread));
return mq;
}c
//静态, 实际就是记录一下用到的值
rt_err_t rt_mq_init(rt_mq_t mq,
const char *name,
void *msgpool,
rt_size_t msg_size,
rt_size_t pool_size,
rt_uint8_t flag)
{
struct rt_mq_message *head;
register rt_base_t temp;
/* parameter check */
RT_ASSERT(mq != RT_NULL);
/* initialize object */
rt_object_init(&(mq->parent.parent), RT_Object_Class_MessageQueue, name);
/* set parent flag */
mq->parent.parent.flag = flag;
/* initialize ipc object */
rt_ipc_object_init(&(mq->parent));
/* set message pool */
mq->msg_pool = msgpool;
/* get correct message size 计算一下实际可以存储的信息的个数 */
mq->msg_size = RT_ALIGN(msg_size, RT_ALIGN_SIZE);
mq->max_msgs = pool_size / (mq->msg_size + sizeof(struct rt_mq_message));
/* initialize message list */
mq->msg_queue_head = RT_NULL;
mq->msg_queue_tail = RT_NULL;
/* initialize message empty list */
mq->msg_queue_free = RT_NULL;
for (temp = 0; temp < mq->max_msgs; temp ++)
{
head = (struct rt_mq_message *)((rt_uint8_t *)mq->msg_pool +
temp * (mq->msg_size + sizeof(struct rt_mq_message)));
head->next = (struct rt_mq_message *)mq->msg_queue_free;
mq->msg_queue_free = head;
}
/* the initial entry is zero */
mq->entry = 0;
/* initialize an additional list of sender suspend thread */
rt_list_init(&(mq->suspend_sender_thread));
return RT_EOK;
}发送消息
c
rt_err_t rt_mq_send_wait(rt_mq_t mq,
const void *buffer,
rt_size_t size,
rt_int32_t timeout)
{
register rt_ubase_t temp;
struct rt_mq_message *msg;
rt_uint32_t tick_delta;
struct rt_thread *thread;
/* greater than one message size */
if (size > mq->msg_size)
//发送的消息太大了, 不能发送
return -RT_ERROR;
/* initialize delta tick */
tick_delta = 0;
/* get current thread 获取当前线程 */
thread = rt_thread_self();
RT_OBJECT_HOOK_CALL(rt_object_put_hook, (&(mq->parent.parent)));
/* disable interrupt 临界区 */
temp = rt_hw_interrupt_disable();
/* get a free list, there must be an empty item */
msg = (struct rt_mq_message *)mq->msg_queue_free;
/* for non-blocking call */
if (msg == RT_NULL && timeout == 0)
{
/* enable interrupt 没有位置, 并且不等待 */
rt_hw_interrupt_enable(temp);
return -RT_EFULL;
}
/* message queue is full 没有可以用于发送的空闲缓冲区 */
while ((msg = mq->msg_queue_free) == RT_NULL)
{
/* reset error number in thread */
thread->error = RT_EOK;
/* no waiting, return timeout */
if (timeout == 0)
{
/* enable interrupt 时间到了 */
rt_hw_interrupt_enable(temp);
return -RT_EFULL;
}
//一个典型的ipc挂起
RT_DEBUG_IN_THREAD_CONTEXT;
/* suspend current thread */
rt_ipc_list_suspend(&(mq->suspend_sender_thread),
thread,
mq->parent.parent.flag);
/* has waiting time, start thread timer */
if (timeout > 0)
{
//还需要等待
/* get the start tick of timer 记录现在的时间, 用于计算是不是超时 */
tick_delta = rt_tick_get();
RT_DEBUG_LOG(RT_DEBUG_IPC, ("mq_send_wait: start timer of thread:%s\n",
thread->name));
/* reset the timeout of thread timer and start it */
rt_timer_control(&(thread->thread_timer),
RT_TIMER_CTRL_SET_TIME,
&timeout);
rt_timer_start(&(thread->thread_timer));
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* re-schedule 任务切换*/
rt_schedule();
//回来了, 可能有位置或者出错(超时)
/* resume from suspend state */
if (thread->error != RT_EOK)
{
//是出错回来的
/* return error */
return thread->error;
}
/* disable interrupt */
temp = rt_hw_interrupt_disable();
//最后更新时间以及检测一下是不是真的有位置
/* if it's not waiting forever and then re-calculate timeout tick */
if (timeout > 0)
{
tick_delta = rt_tick_get() - tick_delta;
timeout -= tick_delta;
if (timeout < 0)
timeout = 0;
}
}
//有位置
/* move free list pointer 获取一个位置, 这个msg是记录空闲位置的那一个指针 */
mq->msg_queue_free = msg->next;
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* the msg is the new tailer of list, the next shall be NULL 这是一个新加入的信息, 没有下一个*/
msg->next = RT_NULL;
/* copy buffer 把信息拷贝到缓冲区 */
rt_memcpy(msg + 1, buffer, size);
/* disable interrupt */
temp = rt_hw_interrupt_disable();
/* link msg to message queue 看一看链表里面有没有信息*/
if (mq->msg_queue_tail != RT_NULL)
{
//这是不第一个信息, 更新上一条信息的下一条为这个新的消息
/* if the tail exists, */
((struct rt_mq_message *)mq->msg_queue_tail)->next = msg;
}
/* set new tail 尾部记录为这一个信息 */
mq->msg_queue_tail = msg;
/* if the head is empty, set head 这时候没有信息的话记录一下这一条信息为第一条 */
if (mq->msg_queue_head == RT_NULL)
mq->msg_queue_head = msg;
/* increase message entry 数量加一 */
mq->entry ++;
/* resume suspended thread 看一看有没有可以释放的任务*/
if (!rt_list_isempty(&mq->parent.suspend_thread))
{
//释放第一个任务
rt_ipc_list_resume(&(mq->parent.suspend_thread));
/* enable interrupt */
rt_hw_interrupt_enable(temp);
rt_schedule();
return RT_EOK;
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
return RT_EOK;
}接收信息
c
//基本一样, 主要看链表操作
rt_err_t rt_mq_recv(rt_mq_t mq,
void *buffer,
rt_size_t size,
rt_int32_t timeout)
{
struct rt_thread *thread;
register rt_ubase_t temp;
struct rt_mq_message *msg;
rt_uint32_t tick_delta;
/* parameter check */
RT_ASSERT(mq != RT_NULL);
RT_ASSERT(rt_object_get_type(&mq->parent.parent) == RT_Object_Class_MessageQueue);
RT_ASSERT(buffer != RT_NULL);
RT_ASSERT(size != 0);
/* initialize delta tick */
tick_delta = 0;
/* get current thread */
thread = rt_thread_self();
RT_OBJECT_HOOK_CALL(rt_object_trytake_hook, (&(mq->parent.parent)));
/* disable interrupt */
temp = rt_hw_interrupt_disable();
/* for non-blocking call */
if (mq->entry == 0 && timeout == 0)
{
//没有可以获取的信息, 并且不等待
rt_hw_interrupt_enable(temp);
return -RT_ETIMEOUT;
}
/* message queue is empty 没有信息 */
while (mq->entry == 0)
{
RT_DEBUG_IN_THREAD_CONTEXT;
/* reset error number in thread */
thread->error = RT_EOK;
/* no waiting, return timeout 超时 */
if (timeout == 0)
{
/* enable interrupt */
rt_hw_interrupt_enable(temp);
thread->error = -RT_ETIMEOUT;
return -RT_ETIMEOUT;
}
//ipc挂起
/* suspend current thread */
rt_ipc_list_suspend(&(mq->parent.suspend_thread),
thread,
mq->parent.parent.flag);
/* has waiting time, start thread timer */
if (timeout > 0)
{
/* get the start tick of timer */
tick_delta = rt_tick_get();
RT_DEBUG_LOG(RT_DEBUG_IPC, ("set thread:%s to timer list\n",
thread->name));
/* reset the timeout of thread timer and start it */
rt_timer_control(&(thread->thread_timer),
RT_TIMER_CTRL_SET_TIME,
&timeout);
rt_timer_start(&(thread->thread_timer));
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* re-schedule */
rt_schedule();
/* recv message */
if (thread->error != RT_EOK)
{
/* return error */
return thread->error;
}
/* disable interrupt */
temp = rt_hw_interrupt_disable();
/* if it's not waiting forever and then re-calculate timeout tick */
if (timeout > 0)
{
tick_delta = rt_tick_get() - tick_delta;
timeout -= tick_delta;
if (timeout < 0)
timeout = 0;
}
}
/* get message from queue 获取当前的第一个信息 */
msg = (struct rt_mq_message *)mq->msg_queue_head;
/* move message queue head 更新一下下一条消息的位置 */
mq->msg_queue_head = msg->next;
/* reach queue tail, set to NULL 这个里面没有消息了(头尾一样, 只有一条信息) */
if (mq->msg_queue_tail == msg)
mq->msg_queue_tail = RT_NULL;
/* decrease message entry 数量更新 */
mq->entry --;
/* enable interrupt */
rt_hw_interrupt_enable(temp);
/* copy message 拷贝一下信息 */
rt_memcpy(buffer, msg + 1, size > mq->msg_size ? mq->msg_size : size);
/* disable interrupt */
temp = rt_hw_interrupt_disable();
/* put message to free list 把这个回归空闲队列 */
msg->next = (struct rt_mq_message *)mq->msg_queue_free;
mq->msg_queue_free = msg;
/* resume suspended thread 释放等待线程*/
if (!rt_list_isempty(&(mq->suspend_sender_thread)))
{
rt_ipc_list_resume(&(mq->suspend_sender_thread));
/* enable interrupt */
rt_hw_interrupt_enable(temp);
RT_OBJECT_HOOK_CALL(rt_object_take_hook, (&(mq->parent.parent)));
rt_schedule();
return RT_EOK;
}
/* enable interrupt */
rt_hw_interrupt_enable(temp);
RT_OBJECT_HOOK_CALL(rt_object_take_hook, (&(mq->parent.parent)));
return RT_EOK;
}信号
c
#if defined(RT_USING_SIGNALS)
rt_sigset_t sig_pending; /**< the pending signals 记录来了的信号 */
rt_sigset_t sig_mask; /**< the mask bits of signal 记录屏蔽的信号 */
rt_sighandler_t *sig_vectors; /**< vectors of signal handler 记录处理函数 */
void *si_list; /**< the signal infor list 挂起的信号的信息链表 */
#endif线程管理结构体
c
typedef void (*rt_sighandler_t)(int signo);处理函数
c
struct siginfo_node
{
siginfo_t si;
struct rt_slist_node list;
};这一个是用来记录挂起的信号的的信息
信号的处理除了会在这里面显示的位置进行, 还会在切换任务的时候处理
安装
c
rt_sighandler_t rt_signal_install(int signo, rt_sighandler_t handler)
{
rt_base_t level;
rt_sighandler_t old = RT_NULL;
rt_thread_t tid = rt_thread_self();
//看一看是不是有效的值
if (!sig_valid(signo)) return SIG_ERR;
level = rt_hw_interrupt_disable();
if (tid->sig_vectors == RT_NULL)
{
//这一个线程之前没有安转过
rt_thread_alloc_sig(tid);
}
if (tid->sig_vectors)
{
old = tid->sig_vectors[signo];
if (handler == SIG_IGN) tid->sig_vectors[signo] = RT_NULL;
else if (handler == SIG_DFL) tid->sig_vectors[signo] = _signal_default_handler;
else tid->sig_vectors[signo] = handler;
}
rt_hw_interrupt_enable(level);
return old;
}c
void rt_thread_alloc_sig(rt_thread_t tid)
{
int index;
rt_base_t level;
rt_sighandler_t *vectors;
//获取一个足以记录处理所有信号的函数的数组
vectors = (rt_sighandler_t *)RT_KERNEL_MALLOC(sizeof(rt_sighandler_t) * RT_SIG_MAX);
RT_ASSERT(vectors != RT_NULL);
for (index = 0; index < RT_SIG_MAX; index ++)
{
//初始化为默认的函数
vectors[index] = _signal_default_handler;
}
//把这一个数组记录在线程里面
level = rt_hw_interrupt_disable();
tid->sig_vectors = vectors;
rt_hw_interrupt_enable(level);
}c
//默认的函数
static void _signal_default_handler(int signo)
{
LOG_I("handled signo[%d] with default action.", signo);
return ;
}删除(屏蔽)
c
//实际上是更新一下线程里面的屏蔽值
void rt_signal_mask(int signo)
{
rt_base_t level;
rt_thread_t tid = rt_thread_self();
level = rt_hw_interrupt_disable();
tid->sig_mask &= ~sig_mask(signo);
rt_hw_interrupt_enable(level);
}解除
c
void rt_signal_unmask(int signo)
{
rt_base_t level;
rt_thread_t tid = rt_thread_self();
level = rt_hw_interrupt_disable();
//改一下标志
tid->sig_mask |= sig_mask(signo);
/* let thread handle pended signals */
if (tid->sig_mask & tid->sig_pending)
{
//有需要处理的标志
rt_hw_interrupt_enable(level);
_signal_deliver(tid);
}
else
{
rt_hw_interrupt_enable(level);
}
}c
//根据要处理的有信号的线程的状态进行分支处理
static void _signal_deliver(rt_thread_t tid)
{
rt_ubase_t level;
level = rt_hw_interrupt_disable();
/* thread is not interested in pended signals */
if (!(tid->sig_pending & tid->sig_mask))
{
//没有待处理的信号
rt_hw_interrupt_enable(level);
return;
}
if ((tid->stat & RT_THREAD_STAT_MASK) == RT_THREAD_SUSPEND)
{
//这一个任务挂起了(他在等待这一个信号)
/* resume thread to handle signal */
rt_thread_resume(tid);
/* add signal state */
tid->stat |= (RT_THREAD_STAT_SIGNAL | RT_THREAD_STAT_SIGNAL_PENDING);
rt_hw_interrupt_enable(level);
//恢复这一个任务
/* re-schedule */
rt_schedule();
//这时候已经看完是不是这一个信号待处理了
}
else
{
//这一个任务运行或ready中
if (tid == rt_thread_self())
{
//是当前的在运行的任务
/* add signal state */
tid->stat |= RT_THREAD_STAT_SIGNAL;
rt_hw_interrupt_enable(level);
/* do signal action in self thread context */
if (rt_interrupt_get_nest() == 0)
{
//直接开启这一个软件线程
rt_thread_handle_sig(RT_TRUE);
}
}
else if (!((tid->stat & RT_THREAD_STAT_SIGNAL_MASK) & RT_THREAD_STAT_SIGNAL))
{
//不是在运行的任务, 这个时候会为这一个任务开启一个新的栈
/* add signal state 更新一下标志 */
tid->stat |= (RT_THREAD_STAT_SIGNAL | RT_THREAD_STAT_SIGNAL_PENDING);
/* point to the signal handle entry */
tid->stat &= ~RT_THREAD_STAT_SIGNAL_PENDING;
tid->sig_ret = tid->sp;//记录一下之前的栈
//这一个看线程处理篇
//实际处理使用的之前栈下面的一部分
tid->sp = rt_hw_stack_init((void *)_signal_entry, RT_NULL,
(void *)((char *)tid->sig_ret - 32), RT_NULL);//设置一下这一个线程返回以后的处理
rt_hw_interrupt_enable(level);
LOG_D("signal stack pointer @ 0x%08x", tid->sp);
/* re-schedule */
rt_schedule();
}
else
{
rt_hw_interrupt_enable(level);
}
}
}c
//这是一个软件中断的线程, 如果需要执行这一个的线程不在runing状态, 会使用一个新的栈空间执行这一个线程
void rt_thread_handle_sig(rt_bool_t clean_state)
{
rt_base_t level;
rt_thread_t tid = rt_thread_self();
struct siginfo_node *si_node;
level = rt_hw_interrupt_disable();
if (tid->sig_pending & tid->sig_mask)
{
/* if thread is not waiting for signal 等一个信号的话直接返回 */
if (!(tid->stat & RT_THREAD_STAT_SIGNAL_WAIT))
{
//这个时候不是在等一个信号
while (tid->sig_pending & tid->sig_mask)
{
//依次处理信号
int signo, error;
rt_sighandler_t handler;
//获取一个待处理的信号
si_node = (struct siginfo_node *)tid->si_list;
if (!si_node) break;
/* remove this sig info node from list */
if (si_node->list.next == RT_NULL)
tid->si_list = RT_NULL;//这是最后一个信号
else
tid->si_list = (void *)rt_slist_entry(si_node->list.next, struct siginfo_node, list);//记录下一个信号
signo = si_node->si.si_signo;//获取标号
handler = tid->sig_vectors[signo];//获取处理函数
tid->sig_pending &= ~sig_mask(signo);//更新挂起的标志
rt_hw_interrupt_enable(level);
LOG_D("handle signal: %d, handler 0x%08x", signo, handler);
if (handler) handler(signo);//执行处理函数
level = rt_hw_interrupt_disable();
error = -RT_EINTR;
rt_mp_free(si_node); /* release this siginfo node 从内存池里面释放 */
/* set errno in thread tcb */
tid->error = error;
}
/* whether clean signal status */
//这一个标志需要清除
if (clean_state == RT_TRUE)
{
tid->stat &= ~RT_THREAD_STAT_SIGNAL;
}
else
{
return;
}
}
}
rt_hw_interrupt_enable(level);
}c
//非当前线程的时候线程返回时候的处理函数(软件中断)
static void _signal_entry(void *parameter)
{
rt_thread_t tid = rt_thread_self();
/* handle signal 处理一下信号 */
rt_thread_handle_sig(RT_FALSE);
/* return to thread 返回之前的在处理的任务状态 */
tid->sp = tid->sig_ret;//改变栈
tid->sig_ret = RT_NULL;
LOG_D("switch back to: 0x%08x\n", tid->sp);
tid->stat &= ~RT_THREAD_STAT_SIGNAL;
//线程处理的里面分析了, 主要是切换一下运行位置以及栈
rt_hw_context_switch_to((rt_ubase_t)&(tid->sp));
}发送信号
c
#define sig_mask(sig_no) (1u << sig_no)
int rt_thread_kill(rt_thread_t tid, int sig)
{
siginfo_t si;
rt_base_t level;
struct siginfo_node *si_node;
if (!sig_valid(sig)) return -RT_EINVAL;
LOG_I("send signal: %d", sig);
si.si_signo = sig;
si.si_code = SI_USER;
si.si_value.sival_ptr = RT_NULL;
level = rt_hw_interrupt_disable();
if (tid->sig_pending & sig_mask(sig))
{
//这一个信号标志已经挂起了, 用这一个新的信息队列覆盖之前的信息
/* whether already emits this signal? */
struct rt_slist_node *node;
struct siginfo_node *entry;
//获取挂起的信号的信息链表
si_node = (struct siginfo_node *)tid->si_list;
if (si_node)
node = (struct rt_slist_node *)&si_node->list;
else
node = RT_NULL;
/* update sig info */
for (; (node) != RT_NULL; node = node->next)
{
//遍历当前任务待处理的所有信号信息
entry = rt_slist_entry(node, struct siginfo_node, list);
if (entry->si.si_signo == sig)
{
//用新的信息覆盖
memcpy(&(entry->si), &si, sizeof(siginfo_t));
rt_hw_interrupt_enable(level);
return 0;
}
}
}
rt_hw_interrupt_enable(level);
//这时候是标志没有挂起, 或者挂起了但是没有信息处理的链表项
//获取一个内存块
si_node = (struct siginfo_node *) rt_mp_alloc(_rt_siginfo_pool, 0);
if (si_node)
{
rt_slist_init(&(si_node->list));
//更新一下信息
memcpy(&(si_node->si), &si, sizeof(siginfo_t));
level = rt_hw_interrupt_disable();
//把这个挂入链表里面
if (tid->si_list)
{
struct siginfo_node *si_list;
//这个里面前面有节点
si_list = (struct siginfo_node *)tid->si_list;
rt_slist_append(&(si_list->list), &(si_node->list));
}
else
{
//前面没有, 这就是第一个
tid->si_list = si_node;
}
/* a new signal 记录一下标志 */
tid->sig_pending |= sig_mask(sig);
rt_hw_interrupt_enable(level);
}
else
{
LOG_E("The allocation of signal info node failed.");
}
/* deliver signal to this thread */
_signal_deliver(tid);
return RT_EOK;
}c
int rt_system_signal_init(void)
{
//这一个会设置rt_mp_alloc返回的大小
_rt_siginfo_pool = rt_mp_create("signal", RT_SIG_INFO_MAX, sizeof(struct siginfo_node));
if (_rt_siginfo_pool == RT_NULL)
{
LOG_E("create memory pool for signal info failed.");
RT_ASSERT(0);
}
return 0;
}等待信号
这一个实际是一直在等待那一个信号, 那一个信号来之前一直挂起, 不会处理其他信号
c
int rt_signal_wait(const rt_sigset_t *set, rt_siginfo_t *si, rt_int32_t timeout)
{
int ret = RT_EOK;
rt_base_t level;
rt_thread_t tid = rt_thread_self();
struct siginfo_node *si_node = RT_NULL, *si_prev = RT_NULL;
/* current context checking */
RT_DEBUG_IN_THREAD_CONTEXT;
/* parameters check */
if (set == NULL || *set == 0 || si == NULL )
{
ret = -RT_EINVAL;
goto __done_return;
}
/* clear siginfo to avoid unknown value 清空一下, 用于记录 */
memset(si, 0x0, sizeof(rt_siginfo_t));
level = rt_hw_interrupt_disable();
/* already pending */
if (tid->sig_pending & *set) goto __done;
if (timeout == 0)
{
ret = -RT_ETIMEOUT;
goto __done_int;
}
/* suspend self thread 把自己挂起 */
rt_thread_suspend(tid);
/* set thread stat as waiting for signal */
tid->stat |= RT_THREAD_STAT_SIGNAL_WAIT;
/* start timeout timer */
if (timeout != RT_WAITING_FOREVER)
{
/* reset the timeout of thread timer and start it */
rt_timer_control(&(tid->thread_timer),
RT_TIMER_CTRL_SET_TIME,
&timeout);
rt_timer_start(&(tid->thread_timer));
}
rt_hw_interrupt_enable(level);
/* do thread scheduling */
rt_schedule();
//返回, 可能超时或者有信号来了
level = rt_hw_interrupt_disable();
/* remove signal waiting flag */
tid->stat &= ~RT_THREAD_STAT_SIGNAL_WAIT;
/* check errno of thread */
if (tid->error == -RT_ETIMEOUT)
{
//是超时
tid->error = RT_EOK;
rt_hw_interrupt_enable(level);
/* timer timeout */
ret = -RT_ETIMEOUT;
goto __done_return;
}
__done:
//是信号来了
/* to get the first matched pending signals */
si_node = (struct siginfo_node *)tid->si_list;
while (si_node)
{
//遍历一下所有的节点
int signo;
signo = si_node->si.si_signo;
if (sig_mask(signo) & *set)
{
//是在等的这一个
*si = si_node->si;
LOG_D("sigwait: %d sig raised!", signo);
if (si_prev) si_prev->list.next = si_node->list.next;//这一个信号的链表不在第一个
else
{
//是第一个
struct siginfo_node *node_next;
if (si_node->list.next)
{
//不是最后一个
node_next = (void *)rt_slist_entry(si_node->list.next, struct siginfo_node, list);
tid->si_list = node_next;
}
else
{
//唯一的信号
tid->si_list = RT_NULL;
}
}
/* clear pending */
tid->sig_pending &= ~sig_mask(signo);//记录为这一个链表处理完了
rt_mp_free(si_node);//释放一下
break;
}
si_prev = si_node;
if (si_node->list.next)
{
//后面还有, 获取下一个
si_node = (void *)rt_slist_entry(si_node->list.next, struct siginfo_node, list);
}
else
{
si_node = RT_NULL;
}
}//while
__done_int:
rt_hw_interrupt_enable(level);
__done_return:
return ret;
}其他
c
#ifdef RT_USING_SIGNALS
/* check stat of thread for signal */
level = rt_hw_interrupt_disable();
if (rt_current_thread->stat & RT_THREAD_STAT_SIGNAL_PENDING)
{
extern void rt_thread_handle_sig(rt_bool_t clean_state);
rt_current_thread->stat &= ~RT_THREAD_STAT_SIGNAL_PENDING;
rt_hw_interrupt_enable(level);
/* check signal status 处理信号 */
rt_thread_handle_sig(RT_TRUE);
}
else
{
rt_hw_interrupt_enable(level);
}
#endif在任务切换的时候rt_schedule里面