嵌入式操作系统内核原理和开发(互斥量)

时间:2022-06-13 20:17:03


【 声明:版权所有,欢迎转载,请勿用于商业用途。  联系信箱:feixiaoxing @163.com】 

 
     今天下午打开邮箱,打开rawos作者给我发的邮件,甚是惊喜。感谢他对我的支持,因为自己阅读过很多os的代码,包括ucos、rtthread、vxWorks、linux等等,所以阅读rawos对于我来说不算特别辛苦的事情。除了某些细节之外,我对整个系统的设计还算得上是比较了解的,所以也打算把这个代码介绍给大家。能在现实的硬件中使用当然最好,如果没有这样的机会,也可以提高个人的认识水平,或者介绍给内部的团队成员,大家一起分析和学习也不失为一个很好的方法。
 
     闲话不多说,话题还是转到我们今天的主题上面,即互斥量。学过操作系统课程的朋友对这个词汇肯定不会很陌生。和信号量相比,互斥保护的资源一般是唯一的。也就是说,资源就一份,你占有了,我就没有办法占有;当然如果你释放了,此时我就有机会占有了。
 
     一切的一切看上去没有什么问题。但是,我们都知道在实时嵌入式系统当中,线程之间的调度是严格按照优先级来进行调度。比方说,优先级为10的任务必须比优先级为11的任务优先得到调度。那么,有同学会问了,那优先级为11的任务什么时候才能得到调度呢,其实这个要求还是蛮苛刻的。要想优先级为11的任务得到调度,此时必须没有优先级10的任务、或者任务pend到资源上了、或者自身delay、或者被人suspend了。否则,优先级为10的任务会这么一直运行下去。那,这和我们的互斥量有什么关系呢?请听我一一讲来。
 
     我们假设现在有两个任务都准备运行,分别人任务A、B,优先级依次是10、11。某一段时间后,优先级为10和优先级为11的任务都在尝试获取某个资源。本来按照优先级的先后顺序,优先级为10的任务应该率先获取资源,这都没问题。但是,假设在尝试获取资源前,优先级为10的任务开了个小差,sleep一会,那么这个时候优先级为11的任务就可以开始运行了。等到优先级为10的任务苏醒过来,想重新获取资源的时候,惊讶地发现资源早就被别人给占了。因为资源目前只有一份,所以它只好把自己pend到等待队列里面,慢慢等待好心人能快点把资源释放出来。一切的一切看上去没有什么问题,但是这却和实时系统设计的初衷是相违背的。前面我们规定高优先级的任务必须优先得到运行的机会,而目前这种情况和我们的设计原则是背道而驰的。
 
     当然这个问题很早就被大家发现了,大家也在尝试不同的方法来解决。目前使用的比较多的就是两种方法,一种是给互斥量设定一个优先级,另外一种就是对优先级进行继承处理。看上去是两种方法,其实目的只有一个,就是让那些占有互斥量的thread提高优先级,赶快运行结束,把资源还给后面真正需要的人。看上去一切解决得都很完美,但是大家有没有考虑过这样一个问题,如果线程连续占有多个互斥量,优先级又该怎么处理?如果pend的任务被修改了优先级该怎么处理?如果这两种方法一起被使用,那又该怎么处理?我想,这就是作者在后期对互斥量代码进行重构的原因吧。当然了,上面讨论的内容已经是比较深的了,大家可以看看早期互斥量是怎么设计的,慢慢来,这样才会对作者的设计意图更加了解一些。
 
     老规矩,我们首先看看互斥量是怎么设计的,

typedef struct RAW_MUTEX  { 
RAW_COMMON_BLOCK_OBJECT common_block_obj;
RAW_U8 count;

/*ponit to occupy task*/
RAW_TASK_OBJ *occupy;
/*occupy task original priority*/
RAW_U8 occupy_original_priority;
} RAW_MUTEX;

     看上去互斥量的东西多一点,其实也还可以,只要大家明白了互斥量处理逻辑再回头来看看这些东西的时候,认识就会更加深刻。我们看看,数据结构里面都有什么,
     (1)通用互斥结构,这在前面信号量的时候已经介绍过一遍;
     (2)计数,判断资源是否还在;
     (3)当前所属的任务;
     (4)该任务原来的优先级。
 
     说好了基本结构,我们看看互斥量的构造、申请、释放、删除函数是怎么设计的,首先当然还是初始化函数,
RAW_U16 raw_mutex_create(RAW_MUTEX *mutex_ptr, RAW_U8 *name_ptr)  {  	#if (RAW_MUTEX_FUNCTION_CHECK > 0)    	if (mutex_ptr == 0)  		return RAW_NULL_OBJECT;  	  	#endif    	/*Init the list*/  	list_init(&mutex_ptr->common_block_obj.block_list);  	mutex_ptr->common_block_obj.block_way = 0;  	mutex_ptr->common_block_obj.name  =  name_ptr;    	/*No one occupy mutex yet*/  	mutex_ptr->occupy     = 0;    	/*resource is available at init state*/   	mutex_ptr->count   = 1;          	mutex_ptr->occupy_original_priority =  0;  	    	return RAW_SUCCESS;  } 
    初始化的函数还是比较简单的,主要做了下面的流程,
     (1)初始化互斥结构的公共属性,比如名字、阻塞方式等等;
     (2)初始化当前资源数量;
     (3)初始化占有资源的线程指针,还有就是线程的优先级。
 
     创建了互斥量之后,我们就要看看互斥量是怎么申请的?代码有点长,同学们可以心理调整一下了,
RAW_U16 raw_mutex_get(RAW_MUTEX *mutex_ptr, RAW_U32 wait_option)  {  	RAW_U16 error_status;  	RAW_SR_ALLOC();    	#if (RAW_MUTEX_FUNCTION_CHECK > 0)      	if (mutex_ptr == 0) {  		return RAW_NULL_OBJECT;  	}    	if (raw_int_nesting) {  		  		return RAW_NOT_CALLED_BY_ISR;  		  	}  	  	#endif  	  	RAW_CRITICAL_ENTER();  	  	 /* mutex is available */  	if (mutex_ptr->count) {     		mutex_ptr->occupy       =  raw_task_active;                                   		mutex_ptr->occupy_original_priority =  raw_task_active->priority;  		mutex_ptr->count   = 0;    		RAW_CRITICAL_EXIT();    		return RAW_SUCCESS;  	}      	/*if the same task get the same mutex again, it causes deadlock*/   	if (raw_task_active == mutex_ptr->occupy) {         		#if (CONFIG_RAW_ASSERT > 0)  		RAW_ASSERT(0);  		#endif  		    		RAW_CRITICAL_EXIT();     		return RAW_MUTEX_DEADLOCK;     }    	/*Cann't get mutex, and return immediately if wait_option is  RAW_NO_WAIT*/  	if (wait_option == RAW_NO_WAIT) {     		RAW_CRITICAL_EXIT();    		return RAW_NO_PEND_WAIT;    	}    	/*system is locked so task can not be blocked just return immediately*/  	if (raw_sched_lock) {     		RAW_CRITICAL_EXIT();	  		return RAW_SCHED_DISABLE;  	}        /*if current task is a higher priority task and block on  the mutex  	  *priority inverse condition happened, priority inherit method is used here*/  	  	if (raw_task_active->priority < mutex_ptr->occupy->priority) {    		switch (mutex_ptr->occupy->task_state) {  			  			case RAW_RDY:  				/*remove from the ready list*/  				remove_ready_list(&raw_ready_queue, mutex_ptr->occupy);  				/*raise the occupy task priority*/  				mutex_ptr->occupy->priority = raw_task_active->priority;	  				/*readd to the ready list head*/  				add_ready_list_head(&raw_ready_queue, mutex_ptr->occupy);  				break;  			    			case RAW_DLY:  			case RAW_DLY_SUSPENDED:  			case RAW_SUSPENDED:  				/*occupy task is not on any list, so just change the priority*/   				mutex_ptr->occupy->priority = raw_task_active->priority;	  				break;    			case RAW_PEND:                        /* Change the position of the task in the wait list       */  			case RAW_PEND_TIMEOUT:  			case RAW_PEND_SUSPENDED:  			case RAW_PEND_TIMEOUT_SUSPENDED:	  				/*occupy task is on the block list so change the priority on the block list*/  				mutex_ptr->occupy->priority = raw_task_active->priority;	  				change_pend_list_priority(mutex_ptr->occupy);   				break;    			default:  				RAW_CRITICAL_EXIT();	  				return RAW_INVALID_STATE;  		}  		  	}    	/*Any way block the current task*/  	raw_pend_object(&mutex_ptr->common_block_obj, raw_task_active, wait_option);    	RAW_CRITICAL_EXIT();	    	/*find the next highest priority task ready to run*/  	raw_sched();                                                 	/*So the task is waked up, need know which reason cause wake up.*/  	error_status = block_state_post_process(raw_task_active, 0);  	  	return error_status;  } 
    这段代码其实开头都还好,关键是末尾要结束的时候有一段代码比较费解。我想,这就是我前面说过的优先级反转问题。为了解决这一问题,在rawos版本中采取了优先级继承的方法。我们还是详细看一下逻辑本身是怎么样的,
     (1)判断参数合法性;
     (2)判断资源是否可取,如果可取,则在记录当前线程和优先级后返回;
     (3)如果资源被自己重复申请,返回;
     (4)如果线程不愿等待,返回;
     (5)如果此时禁止调度,返回;
     (6)如果此时优先级大于互斥量占有者的优先级,分情况处理
             a)占有者处于ready的状态,那么修改它的优先级,重新加入调度队列;
             b)占有者处于sleep的状态,直接修改优先级即可;
             c)占有者也被pend到别的资源上面了,那么修改那个资源的pend列表,可能设计到调度顺序问题。
     (7)线程把自己pend到互斥量等待队列上面;
     (8)线程调用系统调度函数,切换到其他线程运行;
     (9)线程再次得到运行的机会,从task获取结果后返回。
 
     基本上上面的介绍算得上是很详细了,那么互斥量的释放基本上是一个逆操作的过程,朋友也可以思考一下应该怎么解决才好,
RAW_U16 raw_mutex_put(RAW_MUTEX *mutex_ptr)  {    	LIST *block_list_head;  	  	RAW_SR_ALLOC();    	#if (RAW_MUTEX_FUNCTION_CHECK > 0)    	if (mutex_ptr == 0) {  		return RAW_NULL_OBJECT;  	}  	  	#endif    	block_list_head = &mutex_ptr->common_block_obj.block_list;  	  	RAW_CRITICAL_ENTER();    	/*Must release the mutex by self*/  	if (raw_task_active != mutex_ptr->occupy) {             		RAW_CRITICAL_EXIT();  		return RAW_MUTEX_NOT_RELEASE_BY_OCCYPY;  	}    	/*if no block task on this list just return*/  	if (is_list_empty(block_list_head)) {          		mutex_ptr->count   = 1;                                      		RAW_CRITICAL_EXIT();  		return RAW_SUCCESS;  	}       /*if priority was changed, just change it back to original priority*/  	   	if (raw_task_active->priority != mutex_ptr->occupy_original_priority) {  		  		remove_ready_list(&raw_ready_queue, raw_task_active);  		raw_task_active->priority = mutex_ptr->occupy_original_priority;     		add_ready_list_end(&raw_ready_queue, raw_task_active);    	}    	/* there must have task blocked on this mutex object*/ 																												  	mutex_ptr->occupy       =   list_entry(block_list_head->next, RAW_TASK_OBJ, task_list);   	/*the first blocked task became the occupy task*/  	mutex_ptr->occupy_original_priority = mutex_ptr->occupy->priority;  	/*mutex resource is occupied*/  	mutex_ptr->count   = 0;      	/*Wake up the occupy task, which is the highst priority task on the list*/																										   	raw_wake_object(mutex_ptr->occupy);  	  	RAW_CRITICAL_EXIT();      	raw_sched();                                          	return RAW_SUCCESS;  	  } 
    和之前的信号量释放相比,互斥量的释放要复杂一切,关键就在于修改优先级的问题。我们来梳理一下,
     (1)判断参数合法性;
     (2)判断线程是否为互斥量的占有线程,不是则返回;
     (3)判断等待队列是否为空,为空的话则返回;
     (4)判断占有任务的优先级有没有发生变化,如果有则需要重新修改优先级,重新加入调度队列中;
     (5)选择下一个可以调度的线程;
     (6)函数返回。
 
     说了这么些,就剩下最后一个删除互斥量了,大家再接再厉,一起去学习。
RAW_U16  raw_mutex_delete(RAW_MUTEX *mutex_ptr)  {  	LIST *block_list_head;  	  	RAW_TASK_OBJ  *mutex_occupy;  	  	RAW_SR_ALLOC();    	#if (RAW_MUTEX_FUNCTION_CHECK > 0)    	if (mutex_ptr == 0) {  		return RAW_NULL_OBJECT;  	}  	  	#endif    	block_list_head = &mutex_ptr->common_block_obj.block_list;  	  	RAW_CRITICAL_ENTER();    	mutex_occupy = mutex_ptr->occupy;       	/*if mutex is occupied and occupy priority is not the original priority*/  	if ((mutex_occupy) &&  (mutex_occupy->priority !=  mutex_ptr->occupy_original_priority)) {  		switch (mutex_occupy->task_state) {                        			case RAW_RDY:	  				/*remove from the ready list*/  				remove_ready_list(&raw_ready_queue, mutex_ptr->occupy);  				/*raise the occupy task priority*/  				mutex_occupy->priority =  mutex_ptr->occupy_original_priority;  				/*readd to the ready list head*/  				add_ready_list_end(&raw_ready_queue, mutex_ptr->occupy);  				break;    			case RAW_DLY:  			case RAW_SUSPENDED:  			case RAW_DLY_SUSPENDED:  				/*occupy task is not on any list, so just change the priority*/   				mutex_occupy->priority = mutex_ptr->occupy_original_priority;    				break;    			case RAW_PEND:  			case RAW_PEND_TIMEOUT:  			case RAW_PEND_SUSPENDED:  			case RAW_PEND_TIMEOUT_SUSPENDED:  				/*occupy task is on the block list so change the priority on the block list*/  				mutex_occupy->priority = mutex_ptr->occupy_original_priority;    				change_pend_list_priority(mutex_occupy);     				break;    			default:  				RAW_CRITICAL_EXIT();  				return RAW_STATE_UNKNOWN;  		}  	}    	  	/*All task blocked on this queue is waken up*/  	while (!is_list_empty(block_list_head)) {  		delete_pend_obj(list_entry(block_list_head->next, RAW_TASK_OBJ, task_list));	  	}                  	RAW_CRITICAL_EXIT();    	raw_sched();   	  	return RAW_SUCCESS;  } 

    互斥量的操作在实际情形下未必是存在的,所以作者在设计的时候添加了一个编译宏。不过删除所做的工作也不难理解,一个是处理好当前占有者的关系,一个是处理好等待队列的关系。我们来细看一下流程,
    (1)判断当前参数合法性;
    (2)判断占有者的情况,修改任务优先级,这里的情形和上面申请互斥量的处理方式是一样的,不再赘述;
    (3)唤醒所有的等待线程,如果线程已经suspend掉了,那么继续suspend;
    (4)调度到其他线程,防止有优先级高的任务已经被释放出来了;
    (5)函数返回,结束。