msgpipe.c

一种操作系统源码核

#if !defined(__EBOS_H)
#include "ebos.h"
#endif

/*
*********************************************************************************************************
*                                          PIPE module implement FILE
*********************************************************************************************************
*/

#define DEFAULT_PIPE_SIZE 10

static PIPE_PCB *PipeTable[MAX_PIPE_NUM];//管道向量表
static void CleanupPipe(void *information);////管道清除函数
static PIPE_PCB *    PipeFreeList; //可用管道列表
/*
*********************************************************************************************************
*              获得一个可用的ID号                       

*********************************************************************************************************
*/
static SIGNED _NewPipeId()
{
	static UNSIGNED Next_ID;
	UNSIGNED  id;			/* process id to return		*/
	EnLock();
	if ( (id=Next_ID++) <MAX_PIPE_NUM){
	   	if (PipeTable[id] == NULL){
	        	UnLock();
			return(id);
	   }
	}
	else if(PipeFreeList){
	    	id=PipeFreeList->id;
	    	#if ENABLE_MEM
			FreeBuff(PipeFreeList);
	    	#endif
	    	PipeFreeList=PipeFreeList->next;
	    	UnLock();
	    	return(id);
	}
	UnLock();
	return(ERR_PIPE_FULL);

}
/*
*********************************************************************************************************
*                                   创建一个管道

*********************************************************************************************************
*/
#if ENABLE_MEM
SIGNED CreatePipe (void *pipe_start,USHORT pipe_size,UNSIGNED message_size)
#else
SIGNED CreatePipe (PIPE_PCB *pipe_ptr,void *pipe_start,INT16U pipe_size,UNSIGNED message_size)
#endif
{
 	SIGNED pipe_id;
 	STATUS err;
    	#if ENABLE_MEM
    		PIPE_PCB *pipe_ptr;
    	#endif
    	if (pipe_start==NULL ) return (ERR_INVALID_MEMORY);
    	if(message_size>pipe_size ||message_size==0 ||pipe_size==0)   return (ERR_INVALID_SIZE);
    	pipe_id=_NewPipeId();
    	if(pipe_id<0)
        	return(pipe_id);
    	#if ENABLE_MEM          
        	err=AllocBuff(KERNAL.event_pool,&(void *)pipe_ptr,-1);
		if(err!=OK)
	    		return (ERR_NO_MEMORY);
    	#else
        	if (pipe_ptr==NULL)  return(ERR_INVALID_PIPE);
        	else  if(PipeTable[pipe_ptr->id]==pipe_ptr) return (pipe_ptr->id);
    	#endif
    	#if DEBUG
        	DebugLog(CREATE_PIPE,pipe_ptr->id);//在日志中记录操作，时间等信息
    	#endif
    	PipeTable[pipe_id]=pipe_ptr;
    	pipe_ptr->id=pipe_id;
    	pipe_ptr->current_protect=NULL;
    	pipe_ptr->suspend_type=FIFO;
    	pipe_ptr->pipe_type=FIX_TYPE;
    	pipe_ptr->message_size=message_size;
    	pipe_ptr->pipe_size=pipe_size;
    	pipe_ptr->available=0;
    	pipe_ptr->pipe_start=pipe_start;
    	pipe_ptr->pipe_end=(INT8U *)pipe_start+pipe_size;
    	pipe_ptr->read_pointer=pipe_start;
    	pipe_ptr->write_pointer=pipe_start;
    	pipe_ptr->task_wait_header=NULL;
   
    	return(pipe_id);

}
/*
*********************************************************************************************************
*                                      Del A PIPE

*********************************************************************************************************
*/
STATUS DelPipe(SIGNED pipe_id)
{

   	STATUS state,oldstate;
   	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(DELETE_PIPE,pipe_id);//在日志中记录操作，时间等信息
   	#endif
   	DoProtect(pipe_ptr);
   	if (pipe_ptr == NULL)                                 /* Not allowed to Del idle PIPE     */
      		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));
   	}
   	#if ENABLE_MEM
      		FreeBuff(pipe_ptr);
   	#endif
   	UnProtect(pipe_ptr);
   	EnLock();
   	pipe_ptr->next=PipeFreeList;
   	PipeFreeList=pipe_ptr;
   	PipeTable[pipe_id]=NULL;
   	UnLock();
   
   	if(oldstate==TRUE)
       		_ControlToSystem();
   	return (OK);

}
/*
*********************************************************************************************************
*                                      管道参数设置函数

*********************************************************************************************************
*/
STATUS SysPipeCtl(SIGNED pipe_id,INT8U cmd,INT8U param)
{
	STATUS ret;
	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(SYS_PIPE_CTL,pipe_id);//在日志中记录操作，时间等信息
        #endif
        DoProtect(pipe_ptr);
	if (pipe_ptr == NULL) {
	   	return (ERR_INVALID_PIPE); /* Not allowed is idle task     */
	}
	switch(cmd){
	  	case SUSPEND_TYPE:
  	      		pipe_ptr->suspend_type=param;
	      		ret=OK;
	      		break;
	   	case PIPE_SIZE:
	      		pipe_ptr->pipe_size=param;
	      		ret=OK;
	      		break;
	   	case PIPE_TYPE:
	      		pipe_ptr->pipe_type=param;
	      		ret=OK;
	      		break;
	   	default:
	      		ret=ERR_INVALID_CMD;
	      		break;
	}
	UnProtect(pipe_ptr);
	return(ret);
}
/*
*********************************************************************************************************
*                                     SEND A MESSAGE TO A PIPE

*********************************************************************************************************
*/

STATUS SendToPipe(SIGNED pipe_id, void *message, UNSIGNED size,SIGNED suspend)
{
   
   	TASK_TCB *task_ptr;
   	STATUS status;
   	UNSIGNED i;
   	UNSIGNED message_size;
   	PIPE_PCB *pipe_ptr;
   	WAIT_QUEUE *task_suspend;
   	#if WAIT_QUEUE_ALLOC_MODE	
      		WAIT_QUEUE  pipe_wait_queue;
      		task_suspend=&pipe_wait_queue;
   	#else
      		task_suspend=WaitQueueFreeList;
      		WaitQueueFreeList=WaitQueueFreeList->queue_ptr;
   	#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(SENDTO_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){//有任务在等待接收消息,恢复任务,消
      		pipe_ptr->task_wait_header->queue_message=message;
      		pipe_ptr->task_wait_header->queue_message_size=size;
      		task_ptr=(pipe_ptr->task_wait_header)->task_ptr;
      		UnProtect(pipe_ptr);
      		status=_ResumeTask(task_ptr,EVENT_SUSPEND);
      		if (status==TRUE)
         		_ControlToSystem();
     		return (OK);
   	}
   	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);
          		}
           		//队列消息被另一个任务接收导致队列有空间,则该任务放消息到队列
         
          		pipe_ptr->tasks_waiting--;  
	  		DelList(&(void *)(pipe_ptr->task_wait_header),task_suspend);
	  		
	  		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++;
          		}
          		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);
    	}
  
  

}
/*
*********************************************************************************************************
*                                     SEND A MESSAGE TO Front Of A PIPE

*********************************************************************************************************
*/
STATUS SendToFrontOfPipe(SIGNED pipe_id, void *message, UNSIGNED size, SIGNED suspend)
{
  	WAIT_QUEUE *task_suspend;
  	TASK_TCB *task_ptr;
  	STATUS status;
  	UNSIGNED message_size;
  	SIGNED i;
  	PIPE_PCB *pipe_ptr;
  	#if WAIT_QUEUE_ALLOC_MODE	
      		WAIT_QUEUE  pipe_wait_queue;
      		task_suspend=&pipe_wait_queue;
  	#else
     		 task_suspend=WaitQueueFreeList;
      		WaitQueueFreeList=WaitQueueFreeList->queue_ptr;
  	#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(SENDTO_FRONTOFPIPE,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){//有任务等待接收消息,恢复任务,消息放入队列队列中的任务的挂起结构的消息域
      		pipe_ptr->task_wait_header->queue_message=message;
      		pipe_ptr->task_wait_header->queue_message_size=size;
      		task_ptr=(pipe_ptr->task_wait_header)->task_ptr;
      		UnProtect(pipe_ptr);
      		status=_ResumeTask(task_ptr,EVENT_SUSPEND);
      		if (status==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=message_size;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==1 && pipe_ptr->pipe_type==VAR_TYPE)
                   			*(pipe_ptr->read_pointer)=size;
                		else
           	   			*(pipe_ptr->read_pointer)=*((INT8U *)message+i-1);