msgpipe.c

一种操作系统源码核

            		}
            		pipe_ptr->tasks_waiting--;  
	    		DelList(&(void *)(pipe_ptr->task_wait_header),task_suspend);
	    		Unlock(); 
	    		return (OK);

        	}
        	else{
            		UnProtect(pipe_ptr);
            		return (ERR_PIPE_FULL);
        	}
     	}
     	else{ //队列空间够,消息放入队列,
        	for(i=message_size-1;i>=0;i--){
            		pipe_ptr->available++;
            		if(pipe_ptr->read_pointer==pipe_ptr->pipe_start)
               			pipe_ptr->read_pointer=pipe_ptr->pipe_end;
            		else  
               			(pipe_ptr->read_pointer)--;
            		if (i==0 && pipe_ptr->pipe_type==VAR_TYPE)
               			*(pipe_ptr->read_pointer)=size;
            		else
               			*(pipe_ptr->read_pointer)=*((INT8U *)message+i);
         	}
         	UnProtect(pipe_ptr);
         	return (OK);
     	}



}
/*
*********************************************************************************************************
        广播一条消息到管道,当前在管道上等待接收消息的任务都接收到该消息并恢复
*********************************************************************************************************
*/
STATUS BroadToPipe(SIGNED pipe_id, void *message,UNSIGNED size,SIGNED suspend)
{
   	WAIT_QUEUE *linknode;
   	TASK_TCB *task_ptr;
   	WAIT_QUEUE *task_suspend;
   	STATUS status,oldstate;
   	UNSIGNED message_size,i;
   	PIPE_PCB *pipe_ptr;
   	#if WAIT_QUEUE_ALLOC_MODE	
      		task_suspend=WaitQueueFreeList;
      		WaitQueueFreeList=WaitQueueFreeList->queue_ptr;
      		
   	#else
     		WAIT_QUEUE  pipe_wait_queue;
      		task_suspend=&pipe_wait_queue;
   	#endif   
   	if (message == NULL)                                 /* Not allowed to Del idle PIPE     */
      		return (ERR_INVALID_POINTER);
   	if (size>pipe_ptr->message_size) return (ERR_INVALID_SIZE);   
   	if (pipe_id<0 || pipe_id >=MAX_PIPE_NUM) {                                 /* Not allowed to Del idle task     */
      		return (ERR_INVALID_PIPE);
   	}
   	pipe_ptr=PipeTable[pipe_id];
   	#if DEBUG
      		DebugLog(BROADTO_PIPE,pipe_id);//在日志中记录操作，时间等信息
   	#endif
   	DoProtect(pipe_ptr);
   	if (pipe_ptr == NULL)                                /* Not allowed to Del idle PIPE     */
      		return (ERR_INVALID_PIPE);
   	if(pipe_ptr->pipe_type==VAR_TYPE)
      		message_size=size+1;
   	else
      		message_size=pipe_ptr->message_size;
   	if (pipe_ptr->available==NULL && pipe_ptr->task_wait_header){//有任务在等待接收,消息并且放入队列中的消息域,恢复任务,
      		linknode=pipe_ptr->task_wait_header;
      		do{
         		task_ptr=linknode->task_ptr;
	 		linknode->queue_message=message;
	 		linknode->queue_message_size=size;
	 		status=_ResumeTask(task_ptr,EVENT_SUSPEND);
         		if (status==TRUE)
             			oldstate =status;
         		linknode=(WAIT_QUEUE *)(linknode->next);
      		}while(linknode!=pipe_ptr->task_wait_header);
      		UnProtect(pipe_ptr);
      		if(oldstate==TRUE)
         		_ControlToSystem();
      		return (OK);
   	}
   	else if(pipe_ptr->available+message_size>pipe_ptr->pipe_size){   //队列满
      		if(suspend!=NO_SUSPEND){//任务挂起
         		if(((TASK_TCB *)KERNAL.current_thread)->type!=TASK){
             			UnProtect(pipe_ptr);
             			return(ERR_INVALID_SUSPEND);
         		}
         		pipe_ptr->tasks_waiting++;
         		task_suspend->task_ptr=((TASK_TCB *)KERNAL.current_thread);
	 		task_suspend->queue_message_size=size;
	 		task_suspend->priority=GetPriority(((TASK_TCB *)KERNAL.current_thread)->id);
         		AddList(&(void *)(pipe_ptr->task_wait_header),task_suspend,pipe_ptr->suspend_type);
         		UnProtect(pipe_ptr);
	 		_SuspendTask(KERNAL.current_thread,EVENT_SUSPEND,CleanupPipe,pipe_ptr,suspend);
         		_ControlToSystem();
         		Enlock(); 
           		#if WAIT_QUEUE_ALLOC_MODE
 				task_suspend->queue_ptr=WaitQueueFreeList;
 				WaitQueueFreeList=task_suspend;
 			#endif
         		if(pipe_ptr==NULL){
         			Unlock();
             			return(ERR_PIPE_DELETED);
             		}
         		if (pipe_ptr->available+message_size>pipe_ptr->pipe_size){
             			Unlock();
             			return(ERR_TIMEOUT);
          		}
          		if(pipe_ptr->task_wait_header==NULL){
             			Unlock();
             			return(ERR_PIPE_RESET);
          		}
          		//队列消息被另一个任务接收导致队列为空,该任务放消息到队列
          		for(i=0;i<message_size;i++){
             			if (i==0 && pipe_ptr->pipe_type==VAR_TYPE)
                 			*(pipe_ptr->write_pointer)=size;
             			else
	         			*(pipe_ptr->write_pointer)=*((INT8U *)message+i);
             			if(pipe_ptr->write_pointer==pipe_ptr->pipe_end)
                 			pipe_ptr->write_pointer=pipe_ptr->pipe_start;
             			else  
                 			(pipe_ptr->write_pointer)++;
             			pipe_ptr->available++;
          		}
          		pipe_ptr->tasks_waiting--;  
	  		DelList(&(void *)(pipe_ptr->task_wait_header),task_suspend);
          		Unlock();
         	 	return (OK);
      		}
      		else{
          		UnProtect(pipe_ptr);
          		return (ERR_PIPE_FULL);
      		}
   	}
   	else{//队列不满,消息放入队列
      		for(i=0;i<message_size;i++){
          		if (i==0 && pipe_ptr->pipe_type==VAR_TYPE)
             			*(pipe_ptr->write_pointer)=size;
          		else
             			*(pipe_ptr->write_pointer)=*((INT8U *)message+i);
	  		if(pipe_ptr->write_pointer==pipe_ptr->pipe_end)
	     			pipe_ptr->write_pointer=pipe_ptr->pipe_start;
          		else 
            			(pipe_ptr->write_pointer)++;
          		pipe_ptr->available++;
       		}
       		UnProtect(pipe_ptr);
       		return (OK);
    	}
}


/*
*********************************************************************************************************
*                                    RECEIVE A MESSAGE FROM A PIPE

*********************************************************************************************************
*/
STATUS RecvFromPipe(SIGNED pipe_id,INT8U *message,INT8U *actual_size,SIGNED suspend)
{
	WAIT_QUEUE *task_suspend;
	TASK_TCB *task_ptr;
  	STATUS status;
  	UNSIGNED message_size,i;
  	PIPE_PCB *pipe_ptr;
	#if WAIT_QUEUE_ALLOC_MODE
		task_suspend=WaitQueueFreeList;
		WaitQueueFreeList=WaitQueueFreeList->queue_ptr;

	#else
		WAIT_QUEUE  pipe_wait_queue;
		task_suspend=&pipe_wait_queue;
  	#endif      
  	if (pipe_id<0 || pipe_id >=MAX_PIPE_NUM) {                                 /* Not allowed to Del idle task     */
      		return (ERR_INVALID_PIPE);
  	}
  	pipe_ptr=PipeTable[pipe_id];
  	if (actual_size==NULL || message==NULL ) return ERR_INVALID_POINTER;
  	#if DEBUG
      		DebugLog(RECV_FROM_PIPE,pipe_id);//在日志中记录操作，时间等信息
  	#endif
  	DoProtect(pipe_ptr);
  	if (pipe_ptr == NULL)                                  /* Not allowed to Del idle PIPE     */
      		return (ERR_INVALID_PIPE);
  	if(pipe_ptr->available==0){   //队列空
      		if(suspend!=NO_SUSPEND){//任务挂起
	 		if (((TASK_TCB *)KERNAL.current_thread)->type!=TASK){
	    			UnProtect(pipe_ptr);
            			return (ERR_INVALID_SUSPEND);
	 		}
         		pipe_ptr->tasks_waiting++;  
			task_suspend->task_ptr=((TASK_TCB *)KERNAL.current_thread);
         		task_suspend->priority=GetPriority(((TASK_TCB *)KERNAL.current_thread)->id);
	 		AddList(&(void *)(pipe_ptr->task_wait_header),task_suspend,pipe_ptr->suspend_type);
	 		UnProtect(pipe_ptr);
	 		_SuspendTask(KERNAL.current_thread,EVENT_SUSPEND,CleanupPipe,pipe_ptr,suspend);
	 		_ControlToSystem();
	 		Enlock(); 
           		#if WAIT_QUEUE_ALLOC_MODE
 				task_suspend->queue_ptr=WaitQueueFreeList;
 				WaitQueueFreeList=task_suspend;
 			#endif
	 		if(pipe_ptr==NULL){
	 			Unlock();
	    			return (ERR_PIPE_DELETED);
	    		}
         		if((pipe_ptr->task_wait_header)==NULL) {
	    			Unlock();
	    			return (ERR_PIPE_RESET);
	 		}
	 		if (task_suspend->queue_message==NULL) {
	    			Unlock();
	    			return (ERR_TIMEOUT);
	 		}
	   	 	//接收到消息,移去挂起结构
         		message=task_suspend->queue_message;
         		*actual_size=task_suspend->queue_message_size;
         		pipe_ptr->tasks_waiting--;  
         		DelList(&(void *)(pipe_ptr->task_wait_header),task_suspend);
         		Unlock();
         		return (OK);
      		}
      		else{
	 		UnProtect(pipe_ptr);
		 	return (ERR_PIPE_EMPTY);
      		}
   	}
   	else{ ////队列中有消息,取得消息
      		if(pipe_ptr->pipe_type==VAR_TYPE){
       	 		message_size=*(pipe_ptr->read_pointer);
         		i=0;
      		}
      		else{
			message[0]=*(pipe_ptr->read_pointer);
         		message_size=pipe_ptr->message_size;
			pipe_ptr->available--;
         		i=1;
      		}
      		*actual_size=message_size;
      		while (i<message_size){
         		pipe_ptr->available--;
         		if(pipe_ptr->read_pointer==pipe_ptr->pipe_end)
	   		 	pipe_ptr->read_pointer=pipe_ptr->pipe_start;
	 		else  
	    			(pipe_ptr->read_pointer)++;
        		 message[i]=*(pipe_ptr->read_pointer);
         		i++;
      		}
      		if(pipe_ptr->task_wait_header){  //这时的等待的任务肯定是要发送的任务,判断空间是否足够若够则恢复该任务
	 		if (pipe_ptr->available+pipe_ptr->task_wait_header->queue_message_size+1<=pipe_ptr->pipe_size){
	    			task_ptr=(pipe_ptr->task_wait_header)->task_ptr;
	    			UnProtect(pipe_ptr);
	    			status=_ResumeTask(task_ptr,EVENT_SUSPEND);
            			if (status==TRUE) 
                			_ControlToSystem();
          		}
       		}
       		UnProtect(pipe_ptr);
       		return (OK);
   	}
}
/*
*********************************************************************************************************
*                                    在一个管道上的等待列表上的某个任务恢复或删除时，清除相应等待列表上的项

*********************************************************************************************************
*/
static void CleanupPipe(void *information)
{
    	PIPE_PCB *pipe_ptr;
    	TASK_TCB *task_ptr;
    	WAIT_QUEUE *linknode;
    	pipe_ptr=(PIPE_PCB *)information;
    	UnProtect(pipe_ptr);
    	EnLock();
    	linknode=pipe_ptr->task_wait_header;
    	do{
        	task_ptr=linknode->task_ptr;
		if (task_ptr==((TASK_TCB *)KERNAL.current_thread)){
	    		pipe_ptr->tasks_waiting--;  
	    		DelList(&(void *)(pipe_ptr->task_wait_header),linknode);
	    		#if WAIT_QUEUE_ALLOC_MODE
   				linknode->queue_ptr=WaitQueueFreeList;
 				WaitQueueFreeList=linknode;
      			#endif
		}
		linknode=(WAIT_QUEUE *)(linknode->next);
		if (linknode==NULL) break;
   	}while(linknode!=pipe_ptr->task_wait_header);
  	UnLock();

}
/*
*********************************************************************************************************
*                              列出系统中存在的管道信息

*********************************************************************************************************
*/
STATUS ListPipe(SIGNED pipe_id,PIPE_DATA **pipe_lst)
{
    	PIPE_PCB *pipe_ptr;
    	UNSIGNED i;
    	if ((pipe_id+1)<0 || pipe_id >=MAX_PIPE_NUM) {                                 /* Not allowed to Del idle task     */
        	return (ERR_INVALID_PIPE);
     	}
     	if(pipe_id>=0){
    	  	pipe_ptr=PipeTable[pipe_id];
              	DoProtect(pipe_ptr);
    	       	if (pipe_ptr == NULL)                               /* Not allowed to Del idle task     */
	 		return (ERR_INVALID_PIPE);
		(*pipe_lst)->pipe_start=pipe_ptr->pipe_start;
		(*pipe_lst)->pipe_size=pipe_ptr->pipe_size;
		(*pipe_lst)->pipe_type=pipe_ptr->pipe_type;
		(*pipe_lst)->tasks_waiting=pipe_ptr->tasks_waiting;
		(*pipe_lst)->available=pipe_ptr->available;
		(*pipe_lst)->suspend_type=pipe_ptr->suspend_type;
		(*pipe_lst)->task_id=(pipe_ptr->task_wait_header)->task_ptr->id;
		UnProtect(pipe_ptr);
		return (OK);
       }
       else{
		for(i=0;i<MAX_PIPE_NUM;i++){
			pipe_ptr=PipeTable[i];
			DoProtect(pipe_ptr);
			if(pipe_ptr){
				(*pipe_lst)->pipe_start=pipe_ptr->pipe_start;
				(*pipe_lst)->pipe_size=pipe_ptr->pipe_size;
				(*pipe_lst)->pipe_type=pipe_ptr->pipe_type;
				(*pipe_lst)->tasks_waiting=pipe_ptr->tasks_waiting;
				(*pipe_lst)->available=pipe_ptr->available;
				(*pipe_lst)->suspend_type=pipe_ptr->suspend_type;
				(*pipe_lst)->task_id=(pipe_ptr->task_wait_header)->task_ptr->id;
				pipe_lst++;
    			}
    			UnProtect(pipe_ptr);
    		}
    		
    		return (OK);
    	}			
         
       
       		
}        	 

/*
*********************************************************************************************************
*                           管道模块的初始化

*********************************************************************************************************
*/
void PipeInit(void)
{
     	UNSIGNED i;
     	for(i=0;i<MAX_PIPE_NUM;i++)
     		PipeTable[i]=NULL;

}
/*
*********************************************************************************************************
*                             This function resets a PIPE back to the initial state.

*********************************************************************************************************
*/
STATUS ResetPipe(SIGNED pipe_id)
{
   	WAIT_QUEUE *linknode;
   	STATUS oldstate,state;
   	PIPE_PCB *pipe_ptr;
   	if (pipe_id<0 || pipe_id >=MAX_PIPE_NUM) {                                 /* Not allowed to Del idle task     */
       		return (ERR_INVALID_PIPE);
   	}
   	pipe_ptr=PipeTable[pipe_id];
   	#if DEBUG
       		DebugLog(RESET_PIPE,pipe_id);//在日志中记录操作，时间等信息
   	#endif
   	oldstate=0;
   	DoProtect(pipe_ptr);
   	if (pipe_ptr == NULL)                                  /* Not allowed to Del idle task     */
       		return (ERR_INVALID_PIPE);
   	while(pipe_ptr->task_wait_header){
       		state=_ResumeTask(pipe_ptr->task_wait_header->task_ptr,EVENT_SUSPEND);
       		if(state==TRUE)
           		oldstate =state;
       		DelHead(&(void *)(pipe_ptr->task_wait_header));
   	}
   	UnProtect(pipe_ptr);
   	if(oldstate==TRUE)
       		_ControlToSystem();
   	return (OK);
}