queue.c

来自「用c/c++实现的一个CMPP API」· C语言 代码 · 共 776 行 · 第 1/2 页

    recQCell* pQLink;   /*队列的指针*/
    int nSize;      /*申请的总共内容包内存大小*/
    void*   pSAddr;     /*内容包的开始地址*/
    void*   pEAddr;     /*内容包的结束地址*/
    int     nCount;     /*内容包的个数*/
    
        
    if ( pQAddr == NULL )
    return QUEUE_FAIL;
    
    pQLink = (recQCell*)pQAddr;
    
    nSize  = pQLink->nShmSize;      /*申请的总共内容包内存大小*/
    pSAddr = pQLink->pSAddr;        /*内容包的开始地址*/
    pEAddr = pQLink->pEAddr;        /*内容包的结束地址*/
    nCount = pQLink->nMsgMaxNum;    /*内容包的个数*/
    
    pQLink->nInitFlag = 0; /*重新复位内存信息*/
    
    nRet = nInitQLink( pQAddr, nSize, pSAddr, pEAddr, nCount );
    if ( nRet != 0 )
    {
        ErrMsg( "Error When nInitQLink, pvShmAddr[x] = %x\n", (int)pQAddr );
        return QUEUE_FAIL;
    }
    
    return QUEUE_OK;
}


static  int nInitQLink( void* pAddr, int nSize, void* pSAddr, void* pEAddr, int nCount )
{
    recQCell* pQAddr;
    recMsgCell*     pCellStart;
    int i;
    
    if ( pAddr == NULL )
    return QUEUE_FAIL;
    
    pQAddr = (recQCell* )pAddr;
    
    if( pQAddr->nInitFlag == 1) /*如果初始化过了*/
    return QUEUE_OK;
    
    /*初始化数据*/
    pQAddr->nInitFlag   = 1;        /*是否初始化标志*/
    pQAddr->nMsgMaxNum  = nCount;   /*队列的包最多个数*/
    pQAddr->nHead       = 0;        /*队列的头*/
    pQAddr->nTail       = 0;        /*队列的尾*/
    pQAddr->nShmSize    = nSize;    /*申请的总共内容包内存大小*/
    pQAddr->pSAddr      = pSAddr;   /*内容包的开始地址*/
    pQAddr->pEAddr      = pEAddr;   /*内容包的结束地址*/
    pQAddr->pCurSAddr   = pSAddr;   /*当前内容包的开始地址*/
    pQAddr->pCurEAddr   = pSAddr;   /*当前内容包的结束地址*/
    
    pQAddr->prCell = (recMsgCell**)( (char*)pAddr + sizeof( recQCell ) );
    
    pCellStart = (recMsgCell*) ( (char*)pAddr + sizeof( recQCell ) 
                                + sizeof( void* )*(pQAddr->nMsgMaxNum)
                                ); 
    //for( i=0; i<MacQueueCellNum; i++ )
    for( i=0; i<pQAddr->nMsgMaxNum; i++ )  
    {
        /* 初始化每个包首内容 */
        pQAddr->prCell[i] = (recMsgCell*)( pCellStart );
        pCellStart++;                            
        //memset( &( pQAddr->r1Cell[i] ) ,0, sizeof( recMsgCell ) ); 
        memset(  pQAddr->prCell[i], 0, sizeof( recMsgCell ) ); 
    }
    
    /* 初始化每个包体内容 */
    memset( pSAddr, 0, (size_t)((int)pEAddr -(int)pSAddr) );    
    
    return QUEUE_OK;
}


int nWriteQueue_Api(  void* pQAddr, void* pAddr, int nSize )
{
    int nRet;       /*函数调用结果值*/
    recQCell* pQLink;   /*队列的指针*/

        
    if ( pQAddr == NULL || pAddr == NULL || nSize <=0  )
    return QUEUE_FAIL;
    
    pQLink = (recQCell*)pQAddr;
    
    nRet = nCheckQFull( pQLink, nSize );
    if ( nRet < 0 ) // QUEUE FULL
    {
        return QUEUE_FULL;
    }
    
    /*
    *如果内存分段，需要两次memcpy 
    */
    if( nWriteCellPck( pQLink, pAddr, nSize ) < 0 )     
    {
        ErrMsg( "Error When nWriteFinish\n" ); 
        return QUEUE_FAIL;
    }
        
    if( nWriteFinish( pQLink, nSize, 0  ) < 0 )
    {
        ErrMsg( "Error When nWriteFinish\n" ); 
        return QUEUE_FAIL;
    }
    return QUEUE_OK;
    
}

/*
*写入队列包地址空间，前提是内存空间足够。 如果内存分段，需要两次memcpy 
*输入参数： pQLink（ 内存指针 ）， pSrcAddr( 源地址 ）nSize（包大小）
*返回值： =0 成功， 〈0 失败 
*/
static  int nWriteCellPck( const recQCell* pQLink, void* pSrcAddr, int nSize )
{
    char    *pCurSAddr, *pCurEAddr;
    char    *pSAddr, *pEAddr;
    
    int nSize1; /* 分段的两部分 no1 */
    int nSize2; /* 分段的两部分 no2 */
    
    pCurSAddr = (char*)pQLink->pCurSAddr;
    pCurEAddr = (char*)pQLink->pCurEAddr;
    pSAddr  = (char*)pQLink->pSAddr;
    pEAddr  = (char*)pQLink->pEAddr;
    
    if( pCurEAddr + nSize >= pEAddr ) /* 已经分段 */
    {
        nSize1 = ((int)pEAddr - (int)pCurEAddr);
        memcpy( (void*)pCurEAddr, (void*)pSrcAddr, nSize1 );
                
        nSize2 = nSize - nSize1;
        memcpy( (void*)pSAddr, (void*)((char*)pSrcAddr+nSize1), nSize2 );
    }
    else    /* 不用分段 */
    {
        memcpy( (void*)pCurEAddr, (void*)pSrcAddr, nSize );
    }
    return QUEUE_OK;
}

/*
*检查队列是否还有可用空间（ 可能是内存不够，也可能是达到队列的最大值 ）
*输入参数： pQLink（ 内存指针 ）， nSize（包大小）
*返回值： =0 成功， 〈0 失败 
*/
static  int nCheckQFull( const recQCell* pQLink, int nSize )
{
    
    int nHead, nTail;
    void    *pCurSAddr, *pCurEAddr;
    void    *pSAddr, *pEAddr;
    int nMemFree;
    int nMemUsed;
    
    /*
    *to check Q Full 
    */
    nHead = pQLink->nHead;
    nTail = pQLink->nTail;
    
    nHead = (nHead+1)%(pQLink->nMsgMaxNum);
    if( nHead == nTail ) /*queue is full */
    {
        return QUEUE_FULL;
    }
    
    /*
    *to check Mem Space Enough 
    */
    pCurSAddr = pQLink->pCurSAddr;
    pCurEAddr = pQLink->pCurEAddr;
    pSAddr  = pQLink->pSAddr;
    pEAddr  = pQLink->pEAddr;
    

    if( pCurSAddr <= pCurEAddr )
    {
        nMemUsed = (int)pCurEAddr - (int)pCurSAddr ;
        nMemFree = ( (int)pEAddr - (int)pSAddr ) - nMemUsed;
    }
    else
    {
        nMemFree = (int)pCurSAddr - (int)pCurEAddr ;
    }
        
    if ( nMemFree <= nSize ) /* note the ptr size to mem size convert */
    {
        return QUEUE_FULL;
    }
    
    return QUEUE_OK;
}

static  int nWriteFinish( recQCell* pQLink, int nSize, long lKey)
{
    int nHead;
    char    *pCurEAddr;
    char    *pSAddr, *pEAddr;
    
        
    pCurEAddr = (char*)pQLink->pCurEAddr;
    pSAddr  = (char*)pQLink->pSAddr;
    pEAddr  = (char*)pQLink->pEAddr;
    
    if( pCurEAddr + nSize > pEAddr ) /* 内存分段问题 */
    {
        pQLink->pCurEAddr = (char*)( 
                                    (int)pSAddr +
                                    ( (int) pCurEAddr +(int)nSize - (int)pEAddr)
                                    );
    }
    else
    {
        pQLink->pCurEAddr = (char*)pQLink->pCurEAddr + nSize ;
    }
    
    nHead = pQLink->nHead;
    
    
    pQLink->prCell[nHead]->pPckAddr  = pCurEAddr;    /*包首地址*/
    pQLink->prCell[nHead]->nPckSize  = nSize;        /*包内容大小*/
    //pQLink->r1Cell[nHead].pPckAddr  = pCurEAddr;    /*包首地址*/
    //pQLink->r1Cell[nHead].nPckSize  = nSize;        /*包内容大小*/
    
    pQLink->nHead = (pQLink->nHead+1) % pQLink->nMsgMaxNum ;
                
    return QUEUE_OK;
}

/*
*描述：从内存区取包
*输入参数： pQAddr( 队列的内存地址 )，pAddr 包的首地址，nSize 包大小
*返回值： QUEUE_OK 成功 
*       QUEUE_FAIL 失败     
*       QUEUE_EMPTY没有包
*/
int nReadQueue_Api( void* pQAddr, void* pAddr, int nSize )
{
    int nRet;       /*函数调用结果值*/
    recQCell* pQLink;   /*队列的指针*/
    int nActSize;   /*当前包实际长度*/
    recMsgCell* pMsgCell;   /* 包首 */
    
    
    if ( pQAddr == NULL || pAddr == NULL || nSize <=0 )
    return QUEUE_FAIL;
    
    pQLink = (recQCell*)pQAddr;
    nRet = nCheckQCell( pQLink );
    if ( nRet == QUEUE_EMPTY ) /* need to add block state in here */ 
    {       
        return QUEUE_EMPTY;
    }
    //vDumpQueue_Api( stdout, pQAddr  );

    //nActSize = pQLink->r1Cell[ pQLink->nTail ].nPckSize ;
    pMsgCell = pQLink->prCell[ pQLink->nTail ];
    nActSize = pMsgCell->nPckSize ;
    
    if ( nSize < nActSize )/*如果pAddr输入的空间不够，会发生内存益处*/
    {
        ErrMsg( "Error Because input Buffer nSize < Actual Cell Size " );
        return QUEUE_FAIL;
        
    }
    else
    {
        if (nReadCellPck( pQLink, pAddr, nActSize ) < 0 )
        {
            ErrMsg( "Error When nReadCellPck\n" );
            return QUEUE_FAIL;  
        }
    }
        
    if( nReadFinish( pQLink, nActSize ) < 0 )
    {
        ErrMsg( "Error When nReadFinish\n" );
        return QUEUE_FAIL; 
    }

    return QUEUE_OK ;
}



/*
*读队列包， 如果内存分段，需要两次memcpy 
*输入参数： pQLink（ 内存指针 ）， pSrcAddr( 源地址 ）nSize（包大小）
*返回值： =0 成功， 〈0 失败 
*/
static  int nReadCellPck( const recQCell* pQLink, void* pDstAddr, int nSize )
{
    char    *pCurSAddr;
    char    *pSAddr, *pEAddr;
    int nSize1; /* 分段的两部分 no1 */
    int nSize2; /* 分段的两部分 no2 */
    
    pCurSAddr = (char*)pQLink->pCurSAddr;
    pSAddr  = (char*)pQLink->pSAddr;
    pEAddr  = (char*)pQLink->pEAddr;
    
    
    if( pCurSAddr + nSize >= pEAddr ) /* 已经分段 */
    {
        
        nSize1 = (int)pEAddr - (int)pCurSAddr;
        memcpy( pDstAddr, pCurSAddr, nSize1 );
        
        nSize2 = nSize - nSize1;
        memcpy( (char*)pDstAddr+nSize1, pSAddr,  nSize2 );
    }
    else    /* 不用分段 */
    {
        memcpy( pDstAddr, pCurSAddr, nSize );
        
    }
    return QUEUE_OK;
}


/*
*检查队列是否有包 
*输入参数： pQLink（ 内存指针 ）
*返回值： =0 成功， 〈0 失败 
*/
static  int nCheckQCell( const recQCell* pQLink )
{
        
    if( pQLink->nHead == pQLink->nTail ) /* no cell in Q */
    {
        return QUEUE_EMPTY;
    }
    return QUEUE_OK;
}

static  int nReadFinish( recQCell* pQLink, int nSize )
{

    int nTail;
    char    *pCurSAddr;
    char    *pSAddr, *pEAddr;
    
    pCurSAddr = (char*)pQLink->pCurSAddr;
    pSAddr  = (char*)pQLink->pSAddr;
    pEAddr  = (char*)pQLink->pEAddr;
    
    if( pCurSAddr + nSize > pEAddr ) /* 内存分段问题 */
    {
        pQLink->pCurSAddr = (char*)( 
                                    (int)pSAddr +
                                    ( (int) pCurSAddr +nSize - (int)pEAddr ) 
                                    );
    }
    else
    {
        pQLink->pCurSAddr = (char*)pQLink->pCurSAddr + nSize ;
    }
    
    nTail = pQLink->nTail;
    //memset( &(pQLink->r1Cell[nTail]), 0, sizeof( recMsgCell ) );
    memset( pQLink->prCell[nTail], 0, sizeof( recMsgCell ) );       
    pQLink->nTail = (pQLink->nTail+1) % pQLink->nMsgMaxNum;
    
    return QUEUE_OK;   
}

int nRemoveShm_Api( int nShmId )
{

/*
    if ( nShmId <= 0 )
    return QUEUE_FAIL;
*/

    #ifdef   NOUSESHAREMEMORY    
        free( (void*)nShmId  );
        return QUEUE_OK;
    #else
        return shmctl( nShmId, IPC_RMID, NULL );
    #endif      
    
}