multithread thread don't move

[lzpmail]

hello, i write a program that create 3 threads, and use QWaitCondition and QMutex let's them sync, but when the program run awhile, it stop in enqQue.wait(&mutex), and the program not die, it does not process data, please help me what wrong. thank you

[mcosta]

if you can post your code

[lzpmail]

this is my htree run functin code, this problem is the program run awhile, but after stoped in the second run() function's deqWait.wait(&mutQue), and now the dataBuff size is 8192(most big), the dataQue size is 0, please help what wrong. thanks

void UrineThread::run()
{
    int readlen, retval, i;
    char str[TEMPDATABUFSIZE];
    fd_set rfds;
    struct timeval tv ;
 
    tv.tv_sec = 0;
    tv.tv_usec = 100000;
 
    printf("thread....\n");
    while (!stopThread)
    {
        FD_ZERO(&rfds);                           // 清空串口接收端口集
        FD_SET(fd,&rfds);                         // 设置串口接收端口集
 
        while(FD_ISSET(fd, &rfds))                 // 检测串口是否有读写动作
        {
            FD_ZERO(&rfds);                                          // 清空串口接收端口集  每次循环都要清空，否则不会检测到有变化
            FD_SET(fd,&rfds);                                        // 设置串口接收端口集
            retval = select(fd+1,&rfds,NULL,NULL,&tv);
 
            if(retval == -1)
            {
                printf("an error accured.\n");
                break;
            }
            else if(retval)                                           //retval > 0
            {
                readlen = ::read(fd, str, TEMPDATABUFSIZE);         //读取数据
 
                printf("readlen value: %d\n", readlen);
                printf("data Buff size : %d\n", dataBuff.size());
 
                for (i=0; i<readlen; ++i)
                {
                    mutex.lock();
                    while (dataBuff.size() >= DATABUFFSIZE)
                        buffFull.wait(&mutex);                            //ç*‰å¾…
                    dataBuff.enqueue(str);
                    buffEmpty.wakeAll();
                    enqWait.wakeAll();                                    //唤醒所有线程
                    mutex.unlock();
                }
            /*    str[readlen] = '\0';                                    //ç»™è¯»å–çš„æ•°æ®åŠ ç»“å°¾ç¬¦
                dataBuff.append(str); */                                   //å°†æ•°æ®åŠ å…¥åˆ°ç¼“å†²åŒºä¸*
 
                FD_ZERO(&rfds);
                FD_SET(fd,&rfds);
 
                retval = select(fd+1,&rfds,NULL,NULL,&tv);              //判æ–*是否还有数据
 
                if(!retval)                                             //如果没有数据则退出第二层循环
                {
                    break;
                }
            }
        }
        msleep(60);                                                     //æ— æ•°æ®æ—¶ï¼Œç¡çœ 100毫秒， 新增的
    }
}
 
void UrineShowThread::run()
{
    bool result;
    PacketType currPack;
 
    while (!stopThread)
    {
        mutQue.lock();
        while (packQue.size() < 1)
            deqWait.wait(&mutQue);
        printf("data Buff size: %d, data Que size: %d\n", dataBuff.size(), packQue.size());
        currPack = packQue.dequeue();
        enqWait.wakeAll();
        buffEmpty.wakeAll();
        mutQue.unlock();
 
        result = sHostPackHandler[gHostPackInfo[currPack.ID].type]( currPack );
 
        if (FALSE == result)
        {
            printf("pack handler fault.\n");
            // the command has not been processed, add code here to process it
        }
    }
}
 
void UrineProcThread::run()
{
    printf("data process.\n");                          //在æ*¤å¤„åŠ æ•°æ®å¤„ç†ä»£ç 
 
    gHostPackMan.MakePack();                            //解包，并å*˜äºŽåŒ…队列ä¸*
    printf("out the make pack recursive\n");
}
 
void  PackMan::MakePack( void )
{
    UCHAR currChar;
    BOOL  result;
                                           // repeat untill no data in receive buffer
    while (!stopThread)                    //只有当所有数据都处理后才退出
    {
        mutex.lock();
                                                  // if (dataBuff.size() < 1)
        while (dataBuff.size() < 1)
            buffEmpty.wait(&mutex);
        currChar = dataBuff.dequeue();                //从缓冲区ä¸*获得一个数据
        buffFull.wakeAll();
        mutex.unlock();
 
        // packet ID has been received
        if (mPackIdGot)
        {
            // current byte is a valid packet data
            if ( 0x80 <= currChar )
            {
                // be careful: data stored begin from the second byte
                mCurrPack.buffer[mCurrPackLen] = currChar;
                ++mCurrPackLen;
                --mRestByte;
 
                // whole packet has been received,
                if ( 0 >= mRestByte )
                {
                    result = UnpackWithCheckSum( &mCurrPack.buffer[0], mCurrPackLen );
 
                    if (result)            //解包成功
                    {
                        mutQue.lock();
                        while (packQue.size() >= PACKQUEUESIZE)
                            enqWait.wait(&mutQue);
                        packQue.enqueue(mCurrPack);       //把解的包放到队列ä¸*去
                        deqWait.wakeAll();
                        mutQue.unlock();
                    }
                    else
                    {
                        printf("check sum fault.\n");
//                        mErrorPack ++;
//
//                        if( NULL != gPtrView )
//                        {
//                            ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
//                        }
                    }
                    mPackIdGot = 0;
                }
            }
            // current byte is not a valid packet data, maybe is a packet ID,
            // unget it for further analysis
            else
            {
                // there must be a error, because current packet is not integral
                //mPacksReceived.Put(Pack_ErrPack);
//                mErrorPack ++;
//
//                if( NULL != gPtrView )
//                {
//                    ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
//                }
                printf("the data is fault.\n");
                mPackIdGot = 0;
 
                mutex.lock();
                while (dataBuff.size() >= DATABUFFSIZE)
                    buffFull.wait(&mutex);                            //ç*‰å¾…
                 dataBuff.prepend(currChar);           //currChar 可能是一个包， 重新插入队列
                 buffEmpty.wakeAll();                                  //唤醒所有线程
                mutex.unlock();
//                mUART.mRxCharQue.Unget( );
            }
        }
        // packet ID has not been received
        else
        {
            // check whether currChar is a valid packet ID
            if ( ( mMaxPackID > currChar ) && ( 0 < mPackInfo[currChar].len ) )
            {
                // ****** >>> ****** //
                mRestByte    = mPackInfo[currChar].len - 1 ;
                mCurrPackLen = 1;
                mCurrPack.ID = currChar;
                mPackIdGot   = 1;
                printf("rest byte len: %d\n", mRestByte);
 
                // if this kind of packet only has an ID, a whole packet received
                if ( 0 == mRestByte )
                {
                    mutQue.lock();
                      while (packQue.size() >= PACKQUEUESIZE)
                        enqWait.wait(&mutQue);
                    packQue.enqueue(mCurrPack);       //把解的包放到队列ä¸*去
                    deqWait.wakeAll();
                    mutQue.unlock();
 
                    mPackIdGot = 0;
                }
            }
            // currChar is not a valid packet ID
            else
            {
                printf("fault id: %d\n", currChar);
/*                mErrorPack ++;
 
                if( NULL != gPtrView )
                {
                    ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
                }
 
                //  there must be a error, because current packet is not integral
                mPacksReceived.Put(Pack_UnknownPack);
                */
            }
        }
    } // while
 
    return;
}

void UrineThread::run()
{
    int readlen, retval, i;
    char str[TEMPDATABUFSIZE];
    fd_set rfds;
    struct timeval tv ;

    tv.tv_sec = 0;
    tv.tv_usec = 100000;

    printf("thread....\n");
    while (!stopThread)
    {
        FD_ZERO(&rfds);                           // 清空串口接收端口集
        FD_SET(fd,&rfds);                         // 设置串口接收端口集

        while(FD_ISSET(fd, &rfds))                 // 检测串口是否有读写动作
        {
            FD_ZERO(&rfds);                                          // 清空串口接收端口集  每次循环都要清空，否则不会检测到有变化
            FD_SET(fd,&rfds);                                        // 设置串口接收端口集
            retval = select(fd+1,&rfds,NULL,NULL,&tv);

            if(retval == -1)
            {
                printf("an error accured.\n");
                break;
            }
            else if(retval)                                           //retval > 0
            {
                readlen = ::read(fd, str, TEMPDATABUFSIZE);         //读取数据

                printf("readlen value: %d\n", readlen);
                printf("data Buff size : %d\n", dataBuff.size());

                for (i=0; i<readlen; ++i)
                {
                    mutex.lock();
                    while (dataBuff.size() >= DATABUFFSIZE)
                        buffFull.wait(&mutex);                            //ç*‰å¾…
                    dataBuff.enqueue(str);
                    buffEmpty.wakeAll();
                    enqWait.wakeAll();                                    //唤醒所有线程
                    mutex.unlock();
                }
            /*    str[readlen] = '\0';                                    //ç»™è¯»å–çš„æ•°æ®åŠ ç»“å°¾ç¬¦
                dataBuff.append(str); */                                   //å°†æ•°æ®åŠ å…¥åˆ°ç¼“å†²åŒºä¸*

                FD_ZERO(&rfds);
                FD_SET(fd,&rfds);

                retval = select(fd+1,&rfds,NULL,NULL,&tv);              //判æ–*是否还有数据

                if(!retval)                                             //如果没有数据则退出第二层循环
                {
                    break;
                }
            }
        }
        msleep(60);                                                     //æ— æ•°æ®æ—¶ï¼Œç¡çœ 100毫秒， 新增的
    }
}

void UrineShowThread::run()
{
    bool result;
    PacketType currPack;

    while (!stopThread)
    {
        mutQue.lock();
        while (packQue.size() < 1)
            deqWait.wait(&mutQue);
        printf("data Buff size: %d, data Que size: %d\n", dataBuff.size(), packQue.size());
        currPack = packQue.dequeue();
        enqWait.wakeAll();
        buffEmpty.wakeAll();
        mutQue.unlock();

        result = sHostPackHandler[gHostPackInfo[currPack.ID].type]( currPack );

        if (FALSE == result)
        {
            printf("pack handler fault.\n");
            // the command has not been processed, add code here to process it
        }
    }
}

void UrineProcThread::run()
{
    printf("data process.\n");                          //在æ*¤å¤„åŠ æ•°æ®å¤„ç†ä»£ç 

    gHostPackMan.MakePack();                            //解包，并å*˜äºŽåŒ…队列ä¸*
    printf("out the make pack recursive\n");
}

void  PackMan::MakePack( void )
{
    UCHAR currChar;
    BOOL  result;
                                           // repeat untill no data in receive buffer
    while (!stopThread)                    //只有当所有数据都处理后才退出
    {
        mutex.lock();
                                                  // if (dataBuff.size() < 1)
        while (dataBuff.size() < 1)
            buffEmpty.wait(&mutex);
        currChar = dataBuff.dequeue();                //从缓冲区ä¸*获得一个数据
        buffFull.wakeAll();
        mutex.unlock();

        // packet ID has been received
        if (mPackIdGot)
        {
            // current byte is a valid packet data
            if ( 0x80 <= currChar )
            {
                // be careful: data stored begin from the second byte
                mCurrPack.buffer[mCurrPackLen] = currChar;
                ++mCurrPackLen;
                --mRestByte;

                // whole packet has been received,
                if ( 0 >= mRestByte )
                {
                    result = UnpackWithCheckSum( &mCurrPack.buffer[0], mCurrPackLen );

                    if (result)            //解包成功
                    {
                        mutQue.lock();
                        while (packQue.size() >= PACKQUEUESIZE)
                            enqWait.wait(&mutQue);
                        packQue.enqueue(mCurrPack);       //把解的包放到队列ä¸*去
                        deqWait.wakeAll();
                        mutQue.unlock();
                    }
                    else
                    {
                        printf("check sum fault.\n");
//                        mErrorPack ++;
//
//                        if( NULL != gPtrView )
//                        {
//                            ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
//                        }
                    }
                    mPackIdGot = 0;
                }
            }
            // current byte is not a valid packet data, maybe is a packet ID,
            // unget it for further analysis
            else
            {
                // there must be a error, because current packet is not integral
                //mPacksReceived.Put(Pack_ErrPack);
//                mErrorPack ++;
//
//                if( NULL != gPtrView )
//                {
//                    ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
//                }
                printf("the data is fault.\n");
                mPackIdGot = 0;

                mutex.lock();
                while (dataBuff.size() >= DATABUFFSIZE)
                    buffFull.wait(&mutex);                            //ç*‰å¾…
                 dataBuff.prepend(currChar);           //currChar 可能是一个包， 重新插入队列
                 buffEmpty.wakeAll();                                  //唤醒所有线程
                mutex.unlock();
//                mUART.mRxCharQue.Unget( );
            }
        }
        // packet ID has not been received
        else
        {
            // check whether currChar is a valid packet ID
            if ( ( mMaxPackID > currChar ) && ( 0 < mPackInfo[currChar].len ) )
            {
                // ****** >>> ****** //
                mRestByte    = mPackInfo[currChar].len - 1 ;
                mCurrPackLen = 1;
                mCurrPack.ID = currChar;
                mPackIdGot   = 1;
                printf("rest byte len: %d\n", mRestByte);

                // if this kind of packet only has an ID, a whole packet received
                if ( 0 == mRestByte )
                {
                    mutQue.lock();
                      while (packQue.size() >= PACKQUEUESIZE)
                        enqWait.wait(&mutQue);
                    packQue.enqueue(mCurrPack);       //把解的包放到队列ä¸*去
                    deqWait.wakeAll();
                    mutQue.unlock();

                    mPackIdGot = 0;
                }
            }
            // currChar is not a valid packet ID
            else
            {
                printf("fault id: %d\n", currChar);
/*                mErrorPack ++;

                if( NULL != gPtrView )
                {
                    ::PostMessage( gPtrView->m_hWnd, WM_COMM_ERROR, mErrorPack, 0 );
                }

                //  there must be a error, because current packet is not integral
                mPacksReceived.Put(Pack_UnknownPack);
                */
            }
        }
    } // while

    return;
}

To copy to clipboard, switch view to plain text mode 

[wysota]

I hate to spoil your fun using threads but why don't you just use QSocketNotifier instead of all that C code?

[lzpmail]

yea, when i use thread before, i use the QSocketNotifier to listen the serial file, but i don't know why, the serial data lost sometime, and the interface will be die, so i change to use thread, i have no way.

[wysota]

Well, you must have been doing something wrong then. QSocketNotifier does more or less the same you're doing here only that it... works. There is no need for any threads here.