hs_os.c

OS接口代码,实现OS中间件功能,抽取OS接口,........

*
* Output:
*		phMsgQueueHandle:		消息队列句柄
*
* Return:
*		其他 						- 失败
*		HS_NO_ERROR				- 成功
*
* Others:
*
****************************************************************************/
HS_ErrorCode_e HS_OS_MsgQueueCreate (IN U8* 	pucMsgQueueName, 
									IN U32	maxMsgNums, 
									IN U32	maxMsgLength,
									OUT HANDLE *phMsgQueueHandle)
{
    S32 errsv;
    U32 ulMQId;
    U8 ucMQName[33] = "/";
    struct mq_attr attr;
    mqd_t mqd;

    if(FALSE == s_bMQInit)
    {
        if(MQInit() != HS_NO_ERROR)
        {
            hs_debug_print(ERROR_LEVEL, "Message queue Init Fail\n");
            return HS_ERROR_FAILURE;
        }
    }

    HS_OS_MutexLock(s_hMQMutex, HS_TIMEOUT_INFINITY, 0);
    
    ulMQId= FindUnusedMQ();
    if(ulMQId < 0)
    {
        hs_debug_print(ERROR_LEVEL, "The amount of message queues reached the system's limit!\n");
        return HS_ERROR_FAILURE;
    }

    strcat(ucMQName, pucMsgQueueName);
    strncpy(s_amMQ[ulMQId].mqname, ucMQName, 32);
    
    attr.mq_flags = O_NONBLOCK; //0 or O_NONBLOCK;
    attr.mq_maxmsg = maxMsgNums;
    attr.mq_msgsize = maxMsgLength;
    
    mqd = mq_open(ucMQName, O_RDWR | O_CREAT | O_EXCL, 777, &attr);
    errsv = errno;
    if(mqd == (mqd_t)(-1))
    {
        if(EEXIST == errsv)
        {
            struct mq_attr attrtest;
            hs_debug_print(INFO_LEVEL, "WARNING: Message queue '%s' already exists! Opened anyway(May cause problem!) \n", pucMsgQueueName); 
            mqd = mq_open(ucMQName, O_RDWR);
            mq_getattr(mqd, &attrtest);
            hs_debug_print(INFO_LEVEL, "Opened MsgQ: mq_msgsize = %d\nOpenedMsgQ: mq_maxmsg = %d\n", attrtest.mq_msgsize, attrtest.mq_maxmsg);
        }
        else
        {
            HS_OS_MutexUnlock(s_hMQMutex);
            hs_debug_print(ERROR_LEVEL, "mq_open failed, errno = %d %s, LINE:%d FILE:%s\n", errsv, strerror(errsv), __LINE__, __FILE__);
            hs_debug_print(ERROR_LEVEL, "Message queue create failed!\n");
            return HS_ERROR_FAILURE;
        }
    }

    // Message queue created. Now assign proper properties to mq array
    s_amMQ[ulMQId].bUsed = TRUE;
    s_amMQ[ulMQId].mqd = mqd;
    HS_OS_SemaphoreCreate(maxMsgNums, 0, &s_amMQ[ulMQId].hRcvSem);
    HS_OS_SemaphoreCreate(maxMsgNums, maxMsgNums, &s_amMQ[ulMQId].hSndSem);
    
    HS_OS_MutexUnlock(s_hMQMutex);
    *phMsgQueueHandle = ulMQId;
    return HS_NO_ERROR;
}

/***************************************************************************
* Function:
*		HS_OS_MsgQueueDelete
*
* Description:
*		删除一个消息队列
*
* Input:
*		hMsgQueueHandle:		消息队列句柄
*
* Output:
*		无
*
* Return:
*		其他 						- 失败
*		HS_NO_ERROR				- 成功
*
* Others:
*
****************************************************************************/
HS_ErrorCode_e HS_OS_MsgQueueDelete (HANDLE hMsgQueueHandle)
{
    S32 lRet;
    S32 errsv;

    lRet = mq_close(s_amMQ[hMsgQueueHandle].mqd);
    errsv = errno;
    if(lRet != 0)
    {
        hs_debug_print(ERROR_LEVEL, "Close message queue failed while deleting! (Bad handle?)\n");
        return HS_ERROR_FAILURE;
    }

    lRet = mq_unlink(s_amMQ[hMsgQueueHandle].mqname);
    if(lRet != 0)
    {
        hs_debug_print(ERROR_LEVEL, "Unlink message queue failed while deleting! (Bad handle?)\n");
        return HS_ERROR_FAILURE;
    }

    hs_semaphore_delete(s_amMQ[hMsgQueueHandle].hRcvSem);
    hs_semaphore_delete(s_amMQ[hMsgQueueHandle].hSndSem);
    
    s_amMQ[hMsgQueueHandle].bUsed = FALSE;

    return HS_NO_ERROR;
}

/***************************************************************************
* Function:
*		HS_OS_SendMessage
*
* Description:
*		往指定消息队列发送一个消息
*
* Input:
*		hMsgQueueHandle:		消息队列句柄
*		enmTimeoutFlagType: 	等待超时的类型
*		TimeoutMs:			指定等待的时间，当等待类型为
*							HS_TIMEOUT_SETVALUE  时有效
*		pMsg:				指向要发送的消息的指针
*		msgLength			消息长度，单位Byte
*
* Output:
*		无
*
* Return:
*		其他 						- 失败
*		HS_NO_ERROR				- 成功
*
* Others:
*
****************************************************************************/
HS_ErrorCode_e HS_OS_SendMessage (	
								IN HANDLE hMsgQueueHandle, 
								IN HS_OS_TimeoutFlag_e enmTimeoutFlagType, 
								IN U32 TimeoutMs,
								IN const void* pMsg,
								IN S32 msgLength
								)
{
    S32 lRet = 0;
    S32 errsv = 0;
#if 1
    hs_error_e eRet;
    eRet = HS_OS_SemaphoreWait(s_amMQ[hMsgQueueHandle].hSndSem, enmTimeoutFlagType, TimeoutMs);
    if(HS_ERROR_TIMEOUT == eRet)
    {
        hs_debug_print(INFO_LEVEL, "Msg queue full, send message timed out!\n");
        return HS_ERROR_TIMEOUT;
    }
    
    lRet = mq_send(s_amMQ[hMsgQueueHandle].mqd, pMsg, msgLength, NULL);
    errsv = errno;
    
    if(lRet < 0)
    {
        if(ETIMEDOUT == errsv || EAGAIN == errsv)
        {
            hs_debug_print(INFO_LEVEL, "Msg queue full, send message timed out!\n");
            return HS_ERROR_TIMEOUT;
        }
        else
        {
            hs_debug_print(ERROR_LEVEL, "Send message failed! errno = %u %s\n", errsv, strerror(errsv));
            return HS_ERROR_FAILURE;
        }
    }

    HS_OS_SemaphoreSignal(s_amMQ[hMsgQueueHandle].hRcvSem);

    return HS_NO_ERROR;
#else
    struct timespec tsTimeout;
    struct timeval tvCurTime;
    struct mq_attr attrtest;
    mq_getattr(s_amMQ[messagequeue].mqd, &attrtest);
    printf("SEND MSG: mq_msgsize = %d\nSEND MSG: mq_maxmsg = %d\n", attrtest.mq_msgsize, attrtest.mq_maxmsg);
    if(type == HS_TIMEOUT_IMMEDIATELY)
    {
        struct mq_attr attr, oldattr;
        attr.mq_flags = O_NONBLOCK;
        mq_setattr(s_amMQ[messagequeue].mqd, &attr, &oldattr);
        lRet = mq_send(s_amMQ[messagequeue].mqd, p_message, msgbytes, NULL);
        errsv = errno;
        mq_setattr(s_amMQ[messagequeue].mqd, &oldattr, NULL);
    }
    else if(type == HS_TIMEOUT_INFINITY)
    {
        lRet = mq_send(s_amMQ[messagequeue].mqd, p_message, msgbytes, NULL);
        errsv = errno;
    }
    else if(type == HS_TIMEOUT_SETVALUE)
    {

        /* Acquire abs_timespec */
        gettimeofday(&tvCurTime, NULL);
        printf("tv_sec = %u, tv_nsec = %d\n", tvCurTime.tv_sec, tvCurTime.tv_usec);
        tsTimeout.tv_nsec = tvCurTime.tv_usec * 1000 + (timeout % 1000) * 1000 * 1000;
        tsTimeout.tv_sec = tvCurTime.tv_sec + timeout / 1000 + tsTimeout.tv_nsec / 1000000000;
        tsTimeout.tv_nsec = tsTimeout.tv_nsec % 1000000000;
        printf("tv_sec = %u, tv_nsec = %d\n", tsTimeout.tv_sec, tsTimeout.tv_nsec);
        if(tsTimeout.tv_nsec > 1000000000)
        {
            printf("TV_NSEC > 1000000000!\n");
        }
        lRet = mq_timedsend(s_amMQ[messagequeue].mqd, p_message, msgbytes, NULL, &tsTimeout);
        errsv = errno;
        if(errsv == 22)
        {
            hs_error_e eRetAlt;
            struct mq_attr attr, oldattr;

            hs_debug_print(INFO_LEVEL, "mq_timedsend bug! using alternative method\n");

            eRetAlt = hs_semaphore_wait(s_amMQ[messagequeue].hSndSem, HS_TIMEOUT_SETVALUE, timeout);
            attr.mq_flags = O_NONBLOCK;
            mq_setattr(s_amMQ[messagequeue].mqd, &attr, &oldattr);
            lRet = mq_send(s_amMQ[messagequeue].mqd, p_message, msgbytes, NULL);
            errsv = errno;
            mq_setattr(s_amMQ[messagequeue].mqd, &oldattr, NULL);
        }
    }
    else
    {
        hs_debug_print(ERROR_LEVEL, "Invalid timeout type!\n");
        return HS_ERROR_FAILURE;
    }

    if(lRet < 0)
    {
        if(ETIMEDOUT == errsv || EAGAIN == errsv)
        {
            hs_debug_print(INFO_LEVEL, "Msg quere full, send message timed out!\n");
            return HS_ERROR_TIMEOUT;
        }
        else
        {
            hs_debug_print(ERROR_LEVEL, "Send message failed! errno = %u %s\n", errsv, strerror(errsv));
            return HS_ERROR_FAILURE;
        }
    }

    if(s_amMQ[messagequeue].hRcvSem != 0)
    {
        hs_semaphore_release(s_amMQ[messagequeue].hRcvSem);
    }
       
    return HS_NO_ERROR;
#endif
}

/***************************************************************************
* Function:
*		HS_OS_GetMessage
*
* Description:
*		从指定消息队列获得消息
*
* Input:
*		hMsgQueueHandle:		消息队列句柄
*		enmTimeoutFlagType: 	等待超时的类型
*		TimeoutMs:			指定等待的时间，当等待类型为
*							HS_TIMEOUT_SETVALUE  时有效
*
* Output:
*		pMsg:				指向要接收消息的缓冲区的地址
*		msgLength			缓冲区大小，单位Byte
*
* Return:
*		其他 						- 失败
*		HS_NO_ERROR				- 成功
*
* Others:
*
****************************************************************************/
HS_ErrorCode_e HS_OS_GetMessage (	
								IN HANDLE hMsgQueueHandle,
								IN HS_OS_TimeoutFlag_e enmTimeoutFlagType, 
								IN U32 TimeoutMs,
								OUT const void* pMsg,
								IN S32 msgLength
								)
{
    hs_error_e eRet;
    S32 lRet;
    S32 errsv;
#if 1
    eRet = HS_OS_SemaphoreWait(s_amMQ[hMsgQueueHandle].hRcvSem, enmTimeoutFlagType, TimeoutMs);
    if(HS_ERROR_TIMEOUT == eRet)
    {
        //hs_debug_print(INFO_LEVEL, "Msg queue empty, get message timed out!\n");
        return HS_ERROR_TIMEOUT;
    }
    
    lRet = mq_receive(s_amMQ[hMsgQueueHandle].mqd, pMsg, msgLength, NULL);
    errsv = errno;
    
    if(lRet < 0)
    {
        if(ETIMEDOUT == errsv || EAGAIN == errsv)
        {
            //hs_debug_print(INFO_LEVEL, "Msg queue empty, get message timed out!\n");
            return HS_ERROR_TIMEOUT;
        }
        else
        {
            hs_debug_print(ERROR_LEVEL, "Get message failed! errno = %u %s\n", errsv, strerror(errsv));
            return HS_ERROR_FAILURE;
        }
    }

    HS_OS_SemaphoreSignal(s_amMQ[hMsgQueueHandle].hSndSem);

    return HS_NO_ERROR;
#else
    if(type == HS_TIMEOUT_IMMEDIATELY)
    {
        struct mq_attr attr, oldattr;
        attr.mq_flags = O_NONBLOCK;
        mq_setattr(s_amMQ[messagequeue].mqd, &attr, &oldattr);
        lRet = mq_receive(s_amMQ[messagequeue].mqd, p_buffer, bufferlength, NULL);
        errsv = errno;
        mq_setattr(s_amMQ[messagequeue].mqd, &oldattr, NULL);
    }
    else if(type == HS_TIMEOUT_INFINITY)
    {
        lRet = mq_receive(s_amMQ[messagequeue].mqd, p_buffer, bufferlength, NULL);
        errsv = errno;
    }
    else if(type == HS_TIMEOUT_SETVALUE)
    {
        struct timespec tsTimeout;
        struct timeval tvCurTime;
        
        /* Acquire abs_timespec */
        gettimeofday(&tvCurTime, NULL);
        
        tsTimeout.tv_nsec = tvCurTime.tv_usec * 1000 + (timeout % 1000) * 1000 * 1000;
        tsTimeout.tv_sec = tvCurTime.tv_sec + timeout / 1000 + tsTimeout.tv_nsec / 1000000000;
        tsTimeout.tv_nsec = tsTimeout.tv_nsec % 1000000000;
        printf("tv_sec = %d, tv_nsec = %d\n", tsTimeout.tv_sec, tsTimeout.tv_nsec);

        lRet = mq_timedreceive(s_amMQ[messagequeue].mqd, p_buffer, bufferlength, NULL, &tsTimeout);
        errsv = errno;
        if(errsv == 22)
        {
            hs_error_e eRetAlt;
            struct mq_attr attr, oldattr;

            hs_debug_print(INFO_LEVEL, "mq_timedreceive bug! using alternative method\n");
            printf("sem_mq wait\n");
            eRetAlt = hs_semaphore_wait(s_amMQ[messagequeue].hRcvSem, HS_TIMEOUT_SETVALUE, timeout);
            printf("sem_mq done waiting\n");
            attr.mq_flags = O_NONBLOCK;
            mq_setattr(s_amMQ[messagequeue].mqd, &attr, &oldattr);
            lRet = mq_receive(s_amMQ[messagequeue].mqd, p_buffer, bufferlength, NULL);
            errsv = errno;
            mq_setattr(s_amMQ[messagequeue].mqd, &oldattr, NULL);
        }
    }
    else
    {
        hs_debug_print(ERROR_LEVEL, "Invalid timeout type!\n");
        return HS_ERROR_FAILURE;
    }

    if(lRet < 0)
    {
        if(ETIMEDOUT == errsv || EAGAIN == errsv)
        {
            hs_debug_print(INFO_LEVEL, "Msg queue empty, get message timed out!\n");
            return HS_ERROR_TIMEOUT;
        }
        else
        {
            hs_debug_print(ERROR_LEVEL, "Get message failed! errno = %u %s\n", errsv, strerror(errsv));
            return HS_ERROR_FAILURE;
        }
    }

    if(s_amMQ[messagequeue].hSndSem != 0)
    {
        hs_semaphore_release(s_amMQ[messagequeue].hSndSem);
    }
       
    return HS_NO_ERROR;
#endif
}

/***************************************************************************
* Function:
*		HS_OS_Malloc
*
* Description:
*		申请内存
*
* Input:
*		size:		内存大小
*
* Output:
*		无
*
* Return:
*		其他				- 成功，返回内存地址
*		NULL				- 失败
*
* Others:
*
****************************************************************************/
void *HS_OS_Malloc (IN U32 size)
{
    void *pMem;

    pMem = malloc(size);

    if (NULL == pMem)
    {
        hs_debug_print(ERROR_LEVEL, "Memroy malloc fail!\n");
    }

    return (pMem);
}

/***************************************************************************
* Function:
*		HS_OS_MallocEx
*
* Description:
*		申请内存，需制定内存分区
*
* Input:
*		partition:	指定将要分配内存的分区
*		size:		内存大小
*
* Output:
*		无
*
* Return:
*		其他				- 成功，返回内存地址
*		NULL				- 失败
*
* Others:
*
****************************************************************************/
void *HS_OS_MallocEx(IN U32 partition, IN U32 size)
{
    return HS_OS_Malloc(size);
}

/***************************************************************************
* Function:
*		HS_OS_Calloc
*
* Description:
*		为一系列连续的元素申请内存
*
* Input:
*		num:	元素数量
*		size:		每个元素的内存大小
*
* Output:
*		无
*
* Return:
*		其他				- 成功，返回第一个元素的内存地址
*		NULL				- 失败
*
* Others:
*
****************************************************************************/
void *HS_OS_Calloc (IN U32 num, IN U32 size)
{
    void *pMem;

    pMem = calloc(num, size);

    if (NULL == pMem)
    {
        hs_debug_print(ERROR_LEVEL, "Memory calloc fail!\n");
    }

    return (pMem);
}

/***************************************************************************
* Function:
*		HS_OS_CallocEx
*
* Description:
*		为一系列连续的元素申请内存，需制定内存分区
*
* Input: