前言
笔者这几天利用空闲时间为Sparrow添加了消息队列和互斥锁，已经有十余天没有更过文章了，今天下笔想写一写博客，遂继续为Sparrow更新一篇拓展篇。
其实也没什么好讲的，信号量、消息队列、互斥锁这些IPC机制都大差不差，只是小细节有些不一样，学会了信号量，其他的都差不多。
消息队列
消息队列的使用就是传递消息，当然，也可以用于同步。FreeRTOS中对消息队列的设计十分有趣，消息队列、信号量、互斥锁都是调用通用的函数的，这样的代码复用率很高。因为这三种IPC都是基于队列模型的。
简单讲讲FreeRTOS的三种IPC机制：消息队列、信号量、互斥锁的设计。
FreeRTOS的设计
常见的IPC机制中，通常P操作和V操作是核心
FreeRTOS的IPC机制
FreeRTOS的消息队列使用number记录队列中可用的消息数量，初始化时，numer为0，发信息是V操作，numer++，接收信息是P操作，number–。
从信号量的思想看，那么消息队列就是同步模型，因为它初始化时value为0。
设计消息队列
那么也很好推测这个过程了，发消息时把消息复制到队列中，可用的话就number++，不可用就阻塞，发消息成功后，如果有任务在等消息就唤醒它，
接收消息时同理，从队列中读取消息，有消息可以读的话就number–，不可读就阻塞，接收消息成功后，如果有任务在等待发送消息就唤醒它。
利用消息队列设计信号量
通过观察，我们发现消息队列里有value（就是numer），有P操作，有V操作，那么设计信号量不是很简单吗？不对消息进行操作不就行了，发信息时不复制，接受信息时也不从队列中读，队列也不用开辟内存空间。只要有value，有P操作和V操作就行了，稍微改改现成的消息队列就可以了。
利用消息队列设计互斥锁
前面笔者已经说过了，互斥锁就是信号量的一种，只不过添加了优先级继承机制。把信号量的value初始化为1，也可以用于资源的互斥访问，只不过会触发优先级反转问题就是了。
既然已经利用消息队列设计出了信号量，那么初始化消息队列的number为1，然后在P操作和V操作里面加上优先级反转机制就行了。值得一提的是，FreeRTOS初始化互斥锁的value为1的方法很有意思，就是先调用一次V操作，也就是发送消息，这样就能够使number的值为1了。
FreeRTOS的三种IPC机制设计如上，其实只要学懂了Dijkstra大神的信号量，IPC机制没什么复杂的。
信号量才是IPC机制的精髓所在！
Sparrow的IPC机制
Sparrow讲究的是抽象和模块化开发，老实说，笔者认为FreeRTOS的消息队列的代码写得很臃肿，不过性能稳定，代码复用率也高，只不过笔者并不在于那点内存，而是喜欢开发更加优雅的代码，所以笔者将这三种机制单独作为三种对象。
Sparrow的消息队列
Sparrow的调度层跟IPC层是隔离的，消息队列采用接口和模块化开发，让我们看看调度层的接口：
uint32_t xEnterCritical( void );
void xExitCritical( uint32_t xReturn );uint8_t FindHighestPriority( uint32_t Table );void TaskDelay( uint16_t ticks );typedef void (* TaskFunction_t)( void * );
typedef struct TCB_t *TaskHandle_t;void xTaskCreate( TaskFunction_t pxTaskCode,uint16_t usStackDepth,void *pvParameters,uint32_t uxPriority,TaskHandle_t *self );
TaskHandle_t TaskPrioritySet(TaskHandle_t taskHandle,uint8_t priority);void schedule( void );
void SchedulerInit( void );
void SchedulerStart( void );uint32_t StateAdd( TaskHandle_t taskHandle,uint8_t State);
uint32_t StateRemove( TaskHandle_t taskHandle, uint8_t State);
uint8_t CheckState( TaskHandle_t taskHandle,uint8_t State );TaskHandle_t GetTaskHandle( uint8_t i);
uint8_t GetTaskPriority( TaskHandle_t taskHandle);TaskHandle_t GetCurrentTCB(void);Sparrow的消息队列大体上的算法思路跟FreeRTOS的思路大差不差，因为消息队列的算法思想就是这样，就像信号量必须有PV操作一样，很难在算法上作更改，只能是具体的代码实现方式不一样。
程序 = 数据结构 + 算法，看看数据结构便知：
Class(Queue_struct)
{uint8_t *startPoint;uint8_t *endPoint;uint8_t *readPoint;uint8_t *writePoint;uint8_t MessageNumber;uint32_t SendTable;uint32_t ReceiveTable;uint32_t NodeSize;uint32_t NodeNumber;
};创建函数如下：
Queue_struct* queue_creat(uint32_t queue_length,uint32_t queue_size)
{size_t Qsize = (size_t)( queue_length * queue_size);Queue_struct *queue = heap_malloc(sizeof (Queue_struct) + Qsize);uint8_t *message_start = (uint8_t *)queue + sizeof(Queue_struct);*queue = (Queue_struct){.startPoint = message_start,.endPoint = (uint8_t *)(message_start + Qsize),.readPoint = (uint8_t *)( message_start + ( queue_length - 1) * queue_size ),.writePoint = message_start,.MessageNumber = 0UL,.SendTable = 0UL,.ReceiveTable = 0UL,.NodeNumber = queue_length,.NodeSize = queue_size,};return queue;
}
摘取一段代码：可以看出，StateRemove和StateAdd就是调度层的接口了，发送消息后，如果接收列表有阻塞的任务，那么就转移任务状态唤醒任务。接收消息同理。
void WriteToQueue( Queue_struct *queue , uint32_t *buf, uint8_t CurrentTcbPriority)
{memcpy((void *) queue->writePoint, buf, (size_t) queue->NodeSize);queue->writePoint += queue->NodeSize;if (queue->writePoint >= queue->endPoint) {queue->writePoint = queue->startPoint;}if (queue->ReceiveTable!= 0) {//Wake up the highest priority task in the receiving listuint8_t uxPriority = GetTopTCBIndex(queue->ReceiveTable);TaskHandle_t taskHandle = GetTaskHandle(uxPriority);queue->ReceiveTable &= ~(1 << uxPriority );//it belongs to the IPC layer,can't use State port!StateRemove(taskHandle,Block);// Also synchronize with the total blocking stateStateRemove(taskHandle,Delay);StateAdd(taskHandle, Ready);if(uxPriority > CurrentTcbPriority){schedule();}}(queue->MessageNumber)++;
}void ExtractFromQueue( Queue_struct *queue, uint32_t *buf, uint8_t CurrentTcbPriority)
{queue->readPoint += queue->NodeSize;if( queue->readPoint >= queue->endPoint ){queue->readPoint = queue->startPoint;}memcpy( ( void * ) buf, ( void * ) queue->readPoint, ( size_t ) queue->NodeSize );if (queue->SendTable != 0) {//Wake up the highest priority task in the sending listuint8_t uxPriority = GetTopTCBIndex(queue->SendTable);TaskHandle_t taskHandle = GetTaskHandle(uxPriority);queue->SendTable &= ~(1 << uxPriority );//it belongs to the IPC layer,can't use State port!StateRemove(taskHandle,Block);// Also synchronize with the total blocking stateStateRemove(taskHandle,Delay);StateAdd(taskHandle, Ready);if(uxPriority > CurrentTcbPriority ){schedule();}}(queue->MessageNumber)--;
}Sparrow的信号量
信号量在上一章已经讨论过了。读者可以看之前的文章，信号量就是三个部分：value，P操作和V操作。
Sparrow的互斥锁
互斥锁是特殊的信号量，数据结构如下：
Class(Mutex_struct)
{uint8_t value;uint32_t original_priority;uint32_t WaitTable;TaskHandle_t owner;
};Mutex_Handle mutex_creat(void)
{Mutex_struct *mutex = heap_malloc(sizeof (Mutex_struct) );*mutex = (Mutex_struct){.value = 1,.original_priority = 0UL,.WaitTable = 0UL,.owner = NULL};return mutex;
}
相比于信号量，它多了优先级和所有者的概念。
我们看看优先级继承机制即可，代码来自mutex_lock操作：
这是发现互斥锁被其他任务持有时，进行阻塞操作，如果阻塞的任务优先级比互斥锁使用者的优先级大，那么更改互斥锁使用者的优先级为阻塞任务的优先级+1，+1是因为Sparrow不支持相同优先级，当然，如果你在使用Sparrow时，也使用了互斥锁，那么请注意你可能因为互斥锁而阻塞的任务的优先级，每个使用互斥锁的任务的上一个优先级最好不要设计任务，因为发生优先级继承时可能会导致灾难。
if(Ticks > 0){StateAdd(CurrentTCB,Block);mutex->WaitTable |= (1 << CurrentTcbPriority);//it belongs to the IPC layer,can't use State port!TaskDelay(Ticks);uint8_t MutexOwnerPriority = GetTaskPriority(mutex->owner);if( MutexOwnerPriority < CurrentTcbPriority) {StateRemove(mutex->owner, Ready);TaskPrioritySet(mutex->owner, CurrentTcbPriority + 1);StateAdd(mutex->owner, Ready);}}
还原的代码，代码取自mutex_unlock操作：
if(owner_priority != mutex->original_priority) {StateRemove(mutex->owner, Ready);TaskPrioritySet(mutex->owner, mutex->original_priority);StateAdd(mutex->owner, Ready);}
这样，优先级继承机制就完成了。
总结
以上就是Sparrow的IPC机制的设计，有兴趣的读者可以阅读源码并使用Sparrow，然后观察现象以及调试。笔者已经测试过了这三种IPC机制。
三种IPC机制测试的工程地址：skaiui2/SKRTOS_sparrow: Lightweight rtos inspired by SKRTOS
单独的源码地址：skaiui2/SKRTOS_sparrow at source
最后，祝大家学得愉快！