自学教程：C++ Condition_Lock函数代码示例

/*** @brief   pop the read point  from cyc buffer** @author hankejia* @date 2012-11-05* @param[in] pthis			the pointer point to the CCycBuffer.* @return T_S32* @retval return the read point of cycbuffer,if return NULL failed*/static T_CHR * PopSingle( T_pVOID pthis, T_S32 iSize ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    //the cyc buffer is empty, wait for push    Condition_Lock( handle->mDataCon );    handle->mPopComplete = AK_FALSE;    while ( ( handle->mUseSize < iSize ) && ( !(handle->mForceQuit) ) ) {        //Condition_Unlock( handle->mDataCon );        Condition_Unlock( handle->mWriteDataCon );        //logi( "PopSingle buffer empty!/n" );        Condition_Wait( &(handle->mDataCon) );        //logi( "wake up from push signed!/n" );        Condition_Unlock( handle->mDataCon );        Condition_Lock( handle->mWriteDataCon );        Condition_Lock( handle->mDataCon );    }    if ( handle->mForceQuit ) {        handle->mPopComplete = AK_TRUE;        Condition_Unlock( handle->mDataCon );        return NULL;    }    assert( ( ( handle->mRead >= handle->mCycBuffer ) && ( handle->mRead < (handle->mCycBuffer + handle->mBufferSize) ) ) );    assert( ( ( handle->mWrite >= handle->mCycBuffer ) && ( handle->mWrite < (handle->mCycBuffer + handle->mBufferSize) ) ) );    Condition_Unlock( handle->mDataCon );    return handle->mRead;}

/*** @brief   write cyc buffer is data to file** @author hankejia* @date 2012-07-05* @param[in] pthis			the pointer point to the CCycBuffer.* @param[in] fd			file is descriptor.the cyc buffer is data will write to this file* @param[in] iSize			how many data in bytes from cyc buffer you want to write to the file* @param[in] pWriteBuffer	the space alloc by caller,fuction will use this space to load the data,and write to file system,*						if this param is NULL, function will alloc space by himself* @return T_S32* @retval return  >=0 the size of bytes write to file, < 0  failed*/static T_S32 WriteToFs( T_pVOID pthis, T_S32 fd, T_S32 iSize ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    T_CHR * pBuffer = NULL;    T_S32 iLineSize = 0, iLeavings = 0;    Condition_Lock( handle->mWriteDataCon );    //need to flash all data    if ( -1 == iSize ) {        iSize = handle->mUseSize;    }    pBuffer = PopSingle( pthis, iSize );    if ( pBuffer == NULL ) {        ResumeForceQuitState( pthis );        Condition_Unlock( handle->mWriteDataCon );        delay_loop( 0, 10000 ); // 10ms        return 0;    }    iLeavings = iSize;    while( iLeavings > 0 ) {        iLineSize = ( handle->mCycBuffer + handle->mBufferSize ) - pBuffer;        if ( iLineSize == 0 ) {            pBuffer = handle->mCycBuffer;            continue;        } else if ( iLineSize > iLeavings ) {            iLineSize = iLeavings;        }        if ( WriteComplete( fd, pBuffer, iLineSize ) < 0 ) {            Condition_Unlock( handle->mWriteDataCon );            loge( "WriteToFs::WriteComplete error!/n" );            return -1;        }        pBuffer += iLineSize;        iLeavings -= iLineSize;    }    Condition_Lock( handle->mDataCon );    handle->mRead = pBuffer;    handle->mUseSize -= iSize;    handle->mPopComplete = AK_TRUE;    Condition_Unlock( handle->mDataCon );    Condition_Unlock( handle->mWriteDataCon );    //logi( "pop signal send!/n" );    Condition_Signal( &(handle->mDataCon) );    return iSize;}

DEFINE_CONSTRUCTOR_END/*** @brief   simulate class CCycbuffer destructor** free all the simulate class member variable,* @author hankejia* @date 2012-07-05* @param[in] pthis   the pointer point to the CCycBuffer.* @return NONE*/DEFINE_DESTRUCTOR_BEGIN( CCycBuffer ){    CCycBuffer *this = (CCycBuffer *)pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    //虚构时，如果当前还有线程在Push流程中等待    //那么首先Pop出Buffer中的所有数据，唤醒在Push    //中等待的线程，等待其完成push操作。    Condition_Lock( handle->mDataCon );    if ( !( handle->mPushComplete ) ) {        Condition_Unlock( handle->mDataCon );        Clean( this );        //唤醒在Push中等待的线程        Condition_Signal( &(handle->mDataCon) );        Condition_Wait( &(handle->mDataCon) );        Condition_Unlock( handle->mDataCon );    } else        Condition_Unlock( handle->mDataCon );    //如果当前还有线程在pop流程中等待，那么首先    //向Buffer中Push数据，唤醒在Pop中等待的线程，等    //待其完成pop操作。    Condition_Lock( handle->mDataCon );    if ( !( handle->mPopComplete ) ) {        Condition_Unlock( handle->mDataCon );        FakePushFull( this );        //唤醒在Pop中等待的线程        Condition_Signal( &(handle->mDataCon) );        Condition_Wait( &handle->mDataCon );        Condition_Unlock( handle->mDataCon );    } else        Condition_Unlock( handle->mDataCon );    DestroyCycBuffer( this );    Condition_Destroy( &( handle->mDataCon ) );    Condition_Destroy( &( handle->mWriteDataCon ) );    //Condition_Destroy( &( handle->mWaitForPopCon ) );    //Condition_Destroy( &( handle->mWaitForPushCon ) );    free( handle );}

/* * This function is called only when the reporter thread * This function is the loop that the reporter thread processes */void reporter_spawn( thread_Settings *thread ) {    do {        // This section allows for safe exiting with Ctrl-C        Condition_Lock ( ReportCond );        if ( ReportRoot == NULL ) {            // Allow main thread to exit if Ctrl-C is received            thread_setignore();            Condition_Wait ( &ReportCond );            // Stop main thread from exiting until done with all reports            thread_unsetignore();        }        Condition_Unlock ( ReportCond );again:        if ( ReportRoot != NULL ) {            ReportHeader *temp = ReportRoot;            //Condition_Unlock ( ReportCond );            if(temp->report.mThreadMode == kMode_Client){            	synchronize();            }            if ( reporter_process_report ( temp ) ) {                // This section allows for more reports to be added while                // the reporter is processing reports without needing to                // stop the reporter or immediately notify it                Condition_Lock ( ReportCond );                if ( temp == ReportRoot ) {                    // no new reports                    ReportRoot = temp->next;                } else {                    // new reports added                    ReportHeader *itr = ReportRoot;                    while ( itr->next != temp ) {                        itr = itr->next;                    }                    itr->next = temp->next;                }                // finished with report so free it                free( temp );                Condition_Unlock ( ReportCond );                Condition_Signal( &ReportDoneCond );                if (ReportRoot)                    goto again;            }            Condition_Signal( &ReportDoneCond );            usleep(10000);        } else {            //Condition_Unlock ( ReportCond );        }    } while ( 1 );}

/* ------------------------------------------------------------------- * Start the specified object's thread execution. Increments thread * count, spawns new thread, and stores thread ID. * ------------------------------------------------------------------- */void thread_start( struct thread_Settings* thread ) {    // Make sure this object has not been started already    if ( thread_equalid( thread->mTID, thread_zeroid() ) ) {        // Check if we need to start another thread before this one        if ( thread->runNow != NULL ) {            thread_start( thread->runNow );        }        // increment thread count        Condition_Lock( thread_sNum_cond );        thread_sNum++;        Condition_Unlock( thread_sNum_cond );#if   defined( HAVE_POSIX_THREAD )        // pthreads -- spawn new thread        if ( pthread_create( &thread->mTID, NULL, thread_run_wrapper, thread ) != 0 ) {            WARN( 1, "pthread_create" );            // decrement thread count            Condition_Lock( thread_sNum_cond );            thread_sNum--;            Condition_Unlock( thread_sNum_cond );        }#elif defined( HAVE_WIN32_THREAD )        // Win32 threads -- spawn new thread        // Win32 has a thread handle in addition to the thread ID        thread->mHandle = CreateThread( NULL, 0, thread_run_wrapper, thread, 0, &thread->mTID );        if ( thread->mHandle == NULL ) {            WARN( 1, "CreateThread" );            // decrement thread count            Condition_Lock( thread_sNum_cond );            thread_sNum--;            Condition_Unlock( thread_sNum_cond );        }#else        // single-threaded -- call Run_Wrapper in this thread        thread_run_wrapper( thread );#endif    }} // end thread_start

/*** @brief   create the Cyc Buffer, malloc memory** call SetBufferSize fuction to set cyc buffer size before call this function* @author hankejia* @date 2012-07-05* @param[in] pthis   the pointer point to the CCycBuffer.* @return T_S32* @retval if return 0, create cyc buffer success, otherwise failed*/static T_S32 CreateCycBuffer( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    T_S32 ret = 0;    Condition_Lock( handle->mDataCon );    if ( handle->mCycBuffer != NULL )        DestroyCycBuffer( this );    handle->mCycBuffer = (T_CHR *)malloc( handle->mBufferSize );    if ( handle->mCycBuffer == NULL )    {        loge( "memory alloc error![%s::%s]/n", __FILE__, __LINE__ );        ret = -1;        goto End;    }    memset( handle->mCycBuffer, 0, handle->mBufferSize );    handle->mRead = handle->mWrite = handle->mCycBuffer;End:    Condition_Unlock( handle->mDataCon );    return ret;}

/* * ReportServerUDP will generate a report of the UDP * statistics as reported by the server on the client * side. */void ReportServerUDP( thread_Settings *agent, server_hdr *server ) {    if ( (ntohl(server->flags) & HEADER_VERSION1) != 0 &&         isServerReport( agent ) ) {        /*         * Create in one big chunk         */        ReportHeader *reporthdr = (ReportHeader *) malloc( sizeof(ReportHeader) );        Transfer_Info *stats = &reporthdr->report.info;        if ( reporthdr != NULL ) {            stats->transferID = agent->mSock;            stats->groupID = (agent->multihdr != NULL ? agent->multihdr->groupID                                                      : -1);            reporthdr->agentindex = -1;            reporthdr->reporterindex = -1;            reporthdr->report.type = SERVER_RELAY_REPORT;            reporthdr->report.mode = agent->mReportMode;            stats->mFormat = agent->mFormat;            stats->jitter = ntohl( server->jitter1 );            stats->jitter += ntohl( server->jitter2 ) / (double)rMillion;#ifdef HAVE_INT64_T            stats->TotalLen = (((max_size_t) ntohl( server->total_len1 )) << 32) +                                  ntohl( server->total_len2 );#else            stats->TotalLen = ntohl( server->total_len2 );#endif            stats->startTime = 0;            stats->endTime = ntohl( server->stop_sec );            stats->endTime += ntohl( server->stop_usec ) / (double)rMillion;            stats->cntError = ntohl( server->error_cnt );            stats->cntOutofOrder = ntohl( server->outorder_cnt );            stats->cntDatagrams = ntohl( server->datagrams );            stats->mUDP = (char)kMode_Server;            reporthdr->report.connection.peer = agent->local;            reporthdr->report.connection.size_peer = agent->size_local;            reporthdr->report.connection.local = agent->peer;            reporthdr->report.connection.size_local = agent->size_peer;#ifdef HAVE_THREAD            /*             * Update the ReportRoot to include this report.             */            Condition_Lock( ReportCond );            reporthdr->next = ReportRoot;            ReportRoot = reporthdr;            Condition_Signal( &ReportCond );            Condition_Unlock( ReportCond );#else            /*             * Process the report in this thread             */            reporthdr->next = NULL;            process_report ( reporthdr );#endif        } else {            FAIL(1, "Out of Memory!!/n", agent);        }    }}

/* ------------------------------------------------------------------- * Wait for all thread object's execution to complete. Depends on the * thread count being accurate and the threads sending a condition * signal when they terminate. * ------------------------------------------------------------------- */void thread_joinall( void ) {    Condition_Lock( thread_sNum_cond );    while ( thread_sNum > 0 ) {        Condition_Wait( &thread_sNum_cond );    }    Condition_Unlock( thread_sNum_cond );} // end Joinall

int photograph( void *pbuf, int size){	unsigned long outsize;	int ret;	VideoStream_Enc_Reset();		//printf("%d/n", h_eid);	void * buf;	if( 2 == index_file )		ret = frame_encode(h_eid , pbuf+(parse.width*parse.height*3/2), &buf, &outsize);	else		ret = frame_encode(h_eid, pbuf, &buf, &outsize);	if(ret == 0)	{		Condition_Lock( g_conMangerPh );		g_phsize = outsize;		Condition_Signal( &g_conMangerPh );		Condition_Unlock( g_conMangerPh );	}	VideoStream_Enc_Reset();		return 0;}

/* ------------------------------------------------------------------- * unset a thread from being ignorable, so joinall will wait on it * this simply increments the thread count that joinall uses. * This is utilized by the reporter thread which knows when it * is ok to quit (aka no pending reports). * ------------------------------------------------------------------- */void thread_unsetignore( void ) {    Condition_Lock( thread_sNum_cond );    IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Unsetting ignorable thread./r/n" ) );    thread_sNum++;    IPERF_DEBUGF( CONDITION_DEBUG | IPERF_DBG_TRACE, ( "Signaling thread_sNum_cond condition./r/n" ) );    Condition_Signal( &thread_sNum_cond );    Condition_Unlock( thread_sNum_cond );} // end thread_unsetignore

/* ------------------------------------------------------------------- * Start the specified object's thread execution. Increments thread * count, spawns new thread, and stores thread ID. * ------------------------------------------------------------------- */void thread_start( struct thread_Settings* thread ) {    // Make sure this object has not been started already    if ( thread_equalid( thread->mTID, thread_zeroid() ) ) {        // Check if we need to start another thread before this one        if ( thread->runNow != NULL ) {            thread_start( thread->runNow );        }        // increment thread count        Condition_Lock( thread_sNum_cond );        IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Incrementing thread count from %d to %d./r/n", thread_sNum, (thread_sNum + 1) ) );        thread_sNum++;        Condition_Unlock( thread_sNum_cond );        IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE | IPERF_DBG_STATE, ( "Spawning %s thread./r/n", thread_names[thread->mThreadMode] ) );#if   defined( HAVE_POSIX_THREAD )        // pthreads -- spawn new thread        if ( pthread_create( &thread->mTID, NULL, thread_run_wrapper, thread ) != 0 ) {            WARN( 1, ( "pthread_create failed!/r/n" ) );            // decrement thread count            Condition_Lock( thread_sNum_cond );            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Decrementing thread count from %d to %d./r/n", thread_sNum, (thread_sNum - 1) ) );            thread_sNum--;            Condition_Unlock( thread_sNum_cond );        }#elif defined( HAVE_WIN32_THREAD )        // Win32 threads -- spawn new thread        // Win32 has a thread handle in addition to the thread ID        thread->mHandle = CreateThread( NULL, 0, thread_run_wrapper, thread, 0, &thread->mTID );        if ( thread->mHandle == NULL ) {            WARN( 1, ( "CreateThread failed!/r/n" ) );            // decrement thread count            Condition_Lock( thread_sNum_cond );            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Decrementing thread count from %d to %d./r/n", thread_sNum, (thread_sNum - 1) ) );            thread_sNum--;            Condition_Unlock( thread_sNum_cond );        }#else        // single-threaded -- call Run_Wrapper in this thread        thread_run_wrapper( thread );#endif /* HAVE_POSIX_THREAD */    }} // end thread_start

/* ------------------------------------------------------------------- * this function releases all non-terminating threads from the list * of active threads, so that when all terminating threads quit * the joinall will complete. This is called on a Ctrl-C input. It is * also used by the -P usage on the server side * ------------------------------------------------------------------- */int thread_release_nonterm( int interrupt ) {    Condition_Lock( thread_sNum_cond );    thread_sNum -= nonterminating_num;    if ( thread_sNum > 1 && nonterminating_num > 0 && interrupt != 0 ) {        fprintf( stderr,"%s", wait_server_threads );    }    nonterminating_num = 0;    Condition_Signal( &thread_sNum_cond );    Condition_Unlock( thread_sNum_cond );    return thread_sNum;}

/*** @brief   call this function to resume the force quit state. If you call* ForceQuit before, and then you want to push or pop again, please* call this function first.** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return 0 for success, otherwise for failed*/static T_S32 ResumeForceQuitState( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    Condition_Lock( handle->mDataCon );    handle->mForceQuit = AK_FALSE;    Condition_Unlock( handle->mDataCon );    return 0;}

/* ------------------------------------------------------------------- * unset a thread from being non-terminating, so if you cancel through * Ctrl-C they can be ignored by the joinall. * ------------------------------------------------------------------- */void thread_unregister_nonterm( void ) {    Condition_Lock( thread_sNum_cond );    if ( nonterminating_num == 0 ) {        // nonterminating has been released with release_nonterm        // Add back to the threads to wait on        thread_sNum++;    } else {        nonterminating_num--;     }    Condition_Unlock( thread_sNum_cond );}

/* ------------------------------------------------------------------- * unset a thread from being non-terminating, so if you cancel through * Ctrl-C they can be ignored by the joinall. * ------------------------------------------------------------------- */void thread_unregister_nonterm( void ) {    Condition_Lock( thread_sNum_cond );    IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Unregistering non-terminating thread./r/n" ) );    if ( nonterminating_num == 0 ) {        // nonterminating has been released with release_nonterm        // Add back to the threads to wait on        thread_sNum++;    } else {        nonterminating_num--;     }    Condition_Unlock( thread_sNum_cond );} // end thread_unregister_nonterm

/* TODO: This doesn't work on Windows... causes early exit. */int thread_reporterdone( void ) {    int val;    Condition_Lock( thread_sNum_cond );    if ( thread_sNum <= 1 )        val = 1;    else        val = 0;    Condition_Unlock( thread_sNum_cond );    return val;} // end thread_reporterdone

/*** @brief   get cyc buffer is used size** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return cyc buffer is uesd size*/static T_S32 GetUsedSize( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    T_S32 ret = 0;    Condition_Lock( handle->mDataCon );    ret = handle->mUseSize;    Condition_Unlock( handle->mDataCon );    return ret;}

/* * ReportSettings will generate a summary report for * settings being used with Listeners or Clients */void ReportSettings( thread_Settings *agent ) {    if ( isSettingsReport( agent ) ) {        /*         * Create in one big chunk         */        ReportHeader *reporthdr = malloc( sizeof(ReportHeader) );        if ( reporthdr != NULL ) {            ReporterData *data = &reporthdr->report;            data->info.transferID = agent->mSock;            data->info.groupID = -1;            reporthdr->agentindex = -1;            reporthdr->reporterindex = -1;                    data->mHost = agent->mHost;            data->mLocalhost = agent->mLocalhost;            data->mode = agent->mReportMode;            data->type = SETTINGS_REPORT;            data->mBufLen = agent->mBufLen;            data->mMSS = agent->mMSS;            data->mTCPWin = agent->mTCPWin;            data->flags = agent->flags;            data->mThreadMode = agent->mThreadMode;            data->mPort = agent->mPort;            data->info.mFormat = agent->mFormat;            data->info.mTTL = agent->mTTL;            data->connection.peer = agent->peer;            data->connection.size_peer = agent->size_peer;            data->connection.local = agent->local;            data->connection.size_local = agent->size_local;            data->mUDPRate = agent->mUDPRate;            data->mUDPRateUnits = agent->mUDPRateUnits;    #ifdef HAVE_THREAD            /*             * Update the ReportRoot to include this report.             */            Condition_Lock( ReportCond );            reporthdr->next = ReportRoot;            ReportRoot = reporthdr;            Condition_Signal( &ReportCond );            Condition_Unlock( ReportCond );    #else            /*             * Process the report in this thread             */            reporthdr->next = NULL;            process_report ( reporthdr );    #endif         } else {            FAIL(1, "Out of Memory!!/n", agent);        }    }}

/* * ReportPacket is called by a transfer agent to record * the arrival or departure of a "packet" (for TCP it  * will actually represent many packets). This needs to * be as simple and fast as possible as it gets called for * every "packet". */void ReportPacket( ReportHeader* agent, ReportStruct *packet ) {    if ( agent != NULL ) {        int index = agent->reporterindex;        /*         * First find the appropriate place to put the information         */        if ( agent->agentindex == NUM_REPORT_STRUCTS ) {            // Just need to make sure that reporter is not on the first            // item            while ( index == 0 ) {                Condition_Signal( &ReportCond );                Condition_Lock( ReportCond );                Condition_Wait( &ReportDoneCond );                Condition_Unlock( ReportCond );                index = agent->reporterindex;            }            agent->agentindex = 0;        }        // Need to make sure that reporter is not about to be "lapped"        while ( index - 1 == agent->agentindex ) {            Condition_Signal( &ReportCond );            Condition_Lock( ReportCond );            Condition_Wait( &ReportDoneCond );            Condition_Unlock( ReportCond );            index = agent->reporterindex;        }        // Put the information there        memcpy( agent->data + agent->agentindex, packet, sizeof(ReportStruct) );                // Updating agentindex MUST be the last thing done        agent->agentindex++;#ifndef HAVE_THREAD        /*         * Process the report in this thread         */        process_report ( agent );#endif     }}

/* ------------------------------------------------------------------- * this function releases all non-terminating threads from the list * of active threads, so that when all terminating threads quit * the joinall will complete. This is called on a Ctrl-C input. It is * also used by the -P usage on the server side * ------------------------------------------------------------------- */int thread_release_nonterm( int interrupt ) {    Condition_Lock( thread_sNum_cond );    IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Releasing all non-terminating threads./r/n" ) );    thread_sNum -= nonterminating_num;    if ( thread_sNum > 1 && nonterminating_num > 0 && interrupt != 0 ) {        fprintf( stderr, "%s", wait_server_threads );    }    nonterminating_num = 0;    IPERF_DEBUGF( CONDITION_DEBUG | IPERF_DBG_TRACE, ( "Signaling thread_sNum_cond condition./r/n" ) );    Condition_Signal( &thread_sNum_cond );    Condition_Unlock( thread_sNum_cond );    return thread_sNum;} // end thread_release_nonterm

/* ------------------------------------------------------------------- * Stop the specified object's thread execution (if any) immediately. * Decrements thread count and resets the thread ID. * ------------------------------------------------------------------- */void thread_stop( struct thread_Settings* thread ) {#ifdef HAVE_THREAD    // Make sure we have been started    if ( ! thread_equalid( thread->mTID, thread_zeroid() ) ) {        // decrement thread count        Condition_Lock( thread_sNum_cond );        IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Decrementing thread count from %d to %d./r/n", thread_sNum, (thread_sNum - 1) ) );        thread_sNum--;        IPERF_DEBUGF( CONDITION_DEBUG | IPERF_DBG_TRACE, ( "Signaling thread_sNum_cond condition./r/n" ) );        Condition_Signal( &thread_sNum_cond );        Condition_Unlock( thread_sNum_cond );        // use exit()   if called from within this thread        // use cancel() if called from a different thread        if ( thread_equalid( thread_getid(), thread->mTID ) ) {            // Destroy the object            Settings_Destroy( thread );            // Exit#if   defined( HAVE_POSIX_THREAD )            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Exiting %s thread./r/n", thread_names[thread->mThreadMode] ) );            pthread_exit( NULL );#elif defined( HAVE_WIN32_THREAD )            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Exiting %s thread./r/n", thread_names[thread->mThreadMode] ) );            CloseHandle( thread->mHandle );            ExitThread( 0 );#endif /* HAVE_POSIX_THREAD */        } else {            // Cancel#if   defined( HAVE_POSIX_THREAD )            // Cray J90 doesn't have pthread_cancel; Iperf works okay without#ifdef HAVE_PTHREAD_CANCEL            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Canceling %s thread./r/n", thread_names[thread->mThreadMode] ) );            pthread_cancel( thread->mTID );#endif /* HAVE_PTHREAD_CANCEL */#elif defined(HAVE_WIN32_THREAD)            // this is a somewhat dangerous function; it's not            // suggested to Stop() threads a lot.            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Terminating %s thread./r/n", thread_names[thread->mThreadMode] ) );            TerminateThread( thread->mHandle, 0 );#endif /* HAVE_POSIX_THREAD */            // Destroy the object only after killing the thread            Settings_Destroy( thread );        }    }#endif /* HAVE_THREAD */} // end thread_stop

/*** @brief   if there are some threads block in push or pop process, call this function can* force all these threads to exit.** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return 0 for success, otherwise for failed*/static T_S32 ForceQuit( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    Condition_Lock( handle->mDataCon );    handle->mForceQuit = AK_TRUE;    //Condition_Broadcast( &(handle->mWaitForPushCon) );    //Condition_Broadcast( &(handle->mWaitForPopCon) );    Condition_Broadcast( &(handle->mDataCon) );    Condition_Unlock( handle->mDataCon );    return 0;}

/* * BarrierClient allows for multiple stream clients to be syncronized */void BarrierClient( ReportHeader *agent ) {    Condition_Lock(agent->multireport->barrier);    agent->multireport->threads--;    if ( agent->multireport->threads == 0 ) {        // last one set time and wake up everyone        gettimeofday( &(agent->multireport->startTime), NULL );        Condition_Broadcast( &agent->multireport->barrier );    } else {        Condition_Wait( &agent->multireport->barrier );    }    agent->multireport->threads++;    Condition_Unlock( agent->multireport->barrier );    agent->report.startTime = agent->multireport->startTime;    agent->report.nextTime = agent->report.startTime;    TimeAdd( agent->report.nextTime, agent->report.intervalTime );}

/*** @brief   cyc buffer is empty?** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return AK_TRUE if cyc buffer is empty, otherwise return AK_FALSE*/static T_BOOL IsEmpty( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    T_BOOL bRet = AK_FALSE;    Condition_Lock( handle->mDataCon );    if ( !(handle->mPushComplete) ) {        return AK_FALSE;    }    bRet = (handle->mUseSize == 0) ? AK_TRUE : AK_FALSE;    Condition_Unlock( handle->mDataCon );    return bRet;}

/* ------------------------------------------------------------------- * Stop the specified object's thread execution (if any) immediately. * Decrements thread count and resets the thread ID. * ------------------------------------------------------------------- */void thread_stop( struct thread_Settings* thread ) {#ifdef HAVE_THREAD    // Make sure we have been started    if ( ! thread_equalid( thread->mTID, thread_zeroid() ) ) {        // decrement thread count        Condition_Lock( thread_sNum_cond );        thread_sNum--;        Condition_Signal( &thread_sNum_cond );        Condition_Unlock( thread_sNum_cond );        // use exit()   if called from within this thread        // use cancel() if called from a different thread        if ( thread_equalid( thread_getid(), thread->mTID ) ) {            // Destroy the object            Settings_Destroy( thread );            // Exit#if   defined( HAVE_POSIX_THREAD )            pthread_exit( NULL );#else // Win32            CloseHandle( thread->mHandle );            ExitThread( 0 );#endif        } else {            // Cancel#if   defined( HAVE_POSIX_THREAD )            // Cray J90 doesn't have pthread_cancel; Iperf works okay without#ifdef HAVE_PTHREAD_CANCEL            pthread_cancel( thread->mTID );#endif#else // Win32            // this is a somewhat dangerous function; it's not            // suggested to Stop() threads a lot.            TerminateThread( thread->mHandle, 0 );#endif            // Destroy the object only after killing the thread            Settings_Destroy( thread );        }    }#endif} // end Stop

/* ------------------------------------------------------------------- * Wait for all thread object's execution to complete. Depends on the * thread count being accurate and the threads sending a condition * signal when they terminate. * ------------------------------------------------------------------- */void thread_joinall( void ) {    Condition_Lock( thread_sNum_cond );#if IPERF_DEBUG    int waiting_on = -1;#endif /* IPERF_DEBUG */    IPERF_DEBUGF( CONDITION_DEBUG | IPERF_DBG_TRACE, ( "Waiting on thread_sNum_cond condition./r/n" ) );    while ( thread_sNum > 0 ) {#if IPERF_DEBUG        if ( thread_sNum != waiting_on )        {            waiting_on = thread_sNum;            IPERF_DEBUGF( THREAD_DEBUG | IPERF_DBG_TRACE, ( "Waiting on %d threads./r/n", waiting_on ) );        }#endif /* IPERF_DEBUG */        Condition_Wait( &thread_sNum_cond );    }    Condition_Unlock( thread_sNum_cond );} // end Joinall

/*** @brief   let cyc buffer be full state, but no need to push any useful data** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return 0 for success, otherwise for failed*/static T_S32 FakePushFull( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    Condition_Lock( handle->mDataCon );    if ( handle->mUseSize == handle->mBufferSize ) {        Condition_Unlock( handle->mDataCon );        return 0;    }    //复位read write指针    handle->mRead = handle->mWrite = handle->mCycBuffer;    handle->mUseSize = handle->mBufferSize;    Condition_Unlock( handle->mDataCon );    return 0;}

/*** @brief   clean the cyc buffer** @author hankejia* @date 2012-07-05* @param[in] pthis		the pointer point to the CCycBuffer.* @return T_S32* @retval return 0 for success, otherwise for failed*/static T_S32 Clean( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    Condition_Lock( handle->mDataCon );    if ( handle->mUseSize == 0 ) {        Condition_Unlock( handle->mDataCon );        return 0;    }    //复位read write指针    handle->mRead = handle->mWrite = handle->mCycBuffer;    handle->mUseSize = 0;    Condition_Unlock( handle->mDataCon );    ResumeForceQuitState( this );    return 0;}

/*** @brief   destory the Cyc Buffer, free memory** @author hankejia* @date 2012-07-05* @param[in] pthis   the pointer point to the CCycBuffer.* @param[in] iSize   the cyc buffer size.* @return T_S32* @retval if return 0 destroy cyc buffer success, otherwise failed*/static T_S32 DestroyCycBuffer( T_pVOID pthis ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    Condition_Lock( handle->mDataCon );    if ( handle->mCycBuffer != NULL )        free( handle->mCycBuffer );    handle->mCycBuffer = NULL;    handle->mRead = NULL;    handle->mWrite = NULL;    handle->mBufferSize = 0;    handle->mUseSize = 0;    Condition_Unlock( handle->mDataCon );    return 0;}

DEFINE_DESTRUCTOR_END/*** @brief   set the Cyc Buffer is size in bytes* @author hankejia* @date 2012-07-05* @param[in] pthis   the pointer point to the CCycBuffer.* @param[in] iSize   the cyc buffer size.* @return T_S32* @retval if return 0, set buffer size success, otherwise failed*/static T_S32 SetBufferSize( T_pVOID pthis, T_S32 iSize ){    CCycBuffer *this = ( CCycBuffer * )pthis;    CycBuffer_handle * handle = (CycBuffer_handle *)this->handle;    assert( iSize > 0 );    Condition_Lock( handle->mDataCon );    handle->mBufferSize = iSize;    Condition_Unlock( handle->mDataCon );    return 0;}