pa_unix_util.c

mediastreamer2是开源的网络传输媒体流的库

    goto end;
}

PaError PaUnixThread_Terminate( PaUnixThread* self, int wait, PaError* exitResult )
{
    PaError result = paNoError;
    void* pret;

    if( exitResult )
    {
        *exitResult = paNoError;
    }
#if 0
    if( watchdogExitResult )
        *watchdogExitResult = paNoError;

    if( th->watchdogRunning )
    {
        pthread_cancel( th->watchdogThread );
        PA_ENSURE_SYSTEM( pthread_join( th->watchdogThread, &pret ), 0 );

        if( pret && pret != PTHREAD_CANCELED )
        {
            if( watchdogExitResult )
                *watchdogExitResult = *(PaError *) pret;
            free( pret );
        }
    }
#endif

    /* Only kill the thread if it isn't in the process of stopping (flushing adaptation buffers) */
    /* TODO: Make join time out */
    self->stopRequested = wait;
    if( !wait )
    {
        PA_DEBUG(( "%s: Canceling thread %d\n", __FUNCTION__, self->thread ));
        /* XXX: Safe to call this if the thread has exited on its own? */
        pthread_cancel( self->thread );
    }
    PA_DEBUG(( "%s: Joining thread %d\n", __FUNCTION__, self->thread ));
    PA_ENSURE_SYSTEM( pthread_join( self->thread, &pret ), 0 );

    if( pret && PTHREAD_CANCELED != pret )
    {
        if( exitResult )
        {
            *exitResult = *(PaError*)pret;
        }
        free( pret );
    }

error:
    PA_ASSERT_CALL( PaUnixMutex_Terminate( &self->mtx ), paNoError );
    PA_ASSERT_CALL( pthread_cond_destroy( &self->cond ), 0 );

    return result;
}

PaError PaUnixThread_PrepareNotify( PaUnixThread* self )
{
    PaError result = paNoError;
    PA_UNLESS( self->parentWaiting, paInternalError );

    PA_ENSURE( PaUnixMutex_Lock( &self->mtx ) );
    self->locked = 1;

error:
    return result;
}

PaError PaUnixThread_NotifyParent( PaUnixThread* self )
{
    PaError result = paNoError;
    PA_UNLESS( self->parentWaiting, paInternalError );

    if( !self->locked )
    {
        PA_ENSURE( PaUnixMutex_Lock( &self->mtx ) );
        self->locked = 1;
    }
    self->parentWaiting = 0;
    pthread_cond_signal( &self->cond );
    PA_ENSURE( PaUnixMutex_Unlock( &self->mtx ) );
    self->locked = 0;

error:
    return result;
}

int PaUnixThread_StopRequested( PaUnixThread* self )
{
    return self->stopRequested;
}

PaError PaUnixMutex_Initialize( PaUnixMutex* self )
{
    PaError result = paNoError;
    PA_ASSERT_CALL( pthread_mutex_init( &self->mtx, NULL ), 0 );
    return result;
}

PaError PaUnixMutex_Terminate( PaUnixMutex* self )
{
    PaError result = paNoError;
    PA_ASSERT_CALL( pthread_mutex_destroy( &self->mtx ), 0 );
    return result;
}

/** Lock mutex.
 *
 * We're disabling thread cancellation while the thread is holding a lock, so mutexes are 
 * properly unlocked at termination time.
 */
PaError PaUnixMutex_Lock( PaUnixMutex* self )
{
    PaError result = paNoError;
    int oldState;
    
    PA_ENSURE_SYSTEM( pthread_setcancelstate( PTHREAD_CANCEL_DISABLE, &oldState ), 0 );
    PA_ENSURE_SYSTEM( pthread_mutex_lock( &self->mtx ), 0 );

error:
    return result;
}

/** Unlock mutex.
 *
 * Thread cancellation is enabled again after the mutex is properly unlocked.
 */
PaError PaUnixMutex_Unlock( PaUnixMutex* self )
{
    PaError result = paNoError;
    int oldState;

    PA_ENSURE_SYSTEM( pthread_mutex_unlock( &self->mtx ), 0 );
    PA_ENSURE_SYSTEM( pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, &oldState ), 0 );

error:
    return result;
}

#if 0
/* Threading utility struct */
typedef struct PaAlsaThreading
{
    pthread_t watchdogThread;
    pthread_t callbackThread;
    int watchdogRunning;
    int rtSched;
    int rtPrio;
    int useWatchdog;
    unsigned long throttledSleepTime;
    volatile PaTime callbackTime;
    volatile PaTime callbackCpuTime;
    PaUtilCpuLoadMeasurer *cpuLoadMeasurer;
} PaAlsaThreading;

static void OnWatchdogExit( void *userData )
{
    PaAlsaThreading *th = (PaAlsaThreading *) userData;
    struct sched_param spm = { 0 };
    assert( th );

    PA_ASSERT_CALL( pthread_setschedparam( th->callbackThread, SCHED_OTHER, &spm ), 0 );    /* Lower before exiting */
    PA_DEBUG(( "Watchdog exiting\n" ));
}

static PaError BoostPriority( PaAlsaThreading *th )
{
    PaError result = paNoError;
    struct sched_param spm = { 0 };
    spm.sched_priority = th->rtPrio;

    assert( th );

    if( pthread_setschedparam( th->callbackThread, SCHED_FIFO, &spm ) != 0 )
    {
        PA_UNLESS( errno == EPERM, paInternalError );  /* Lack permission to raise priority */
        PA_DEBUG(( "Failed bumping priority\n" ));
        result = 0;
    }
    else
        result = 1; /* Success */
error:
    return result;
}

static void *WatchdogFunc( void *userData )
{
    PaError result = paNoError, *pres = NULL;
    int err;
    PaAlsaThreading *th = (PaAlsaThreading *) userData;
    unsigned intervalMsec = 500;
    const PaTime maxSeconds = 3.;   /* Max seconds between callbacks */
    PaTime timeThen = PaUtil_GetTime(), timeNow, timeElapsed, cpuTimeThen, cpuTimeNow, cpuTimeElapsed;
    double cpuLoad, avgCpuLoad = 0.;
    int throttled = 0;

    assert( th );

    /* Execute OnWatchdogExit when exiting */
    pthread_cleanup_push( &OnWatchdogExit, th );

    /* Boost priority of callback thread */
    PA_ENSURE( result = BoostPriority( th ) );
    if( !result )
    {
        /* Boost failed, might as well exit */
        pthread_exit( NULL );
    }

    cpuTimeThen = th->callbackCpuTime;
    {
        int policy;
        struct sched_param spm = { 0 };
        pthread_getschedparam( pthread_self(), &policy, &spm );
        PA_DEBUG(( "%s: Watchdog priority is %d\n", __FUNCTION__, spm.sched_priority ));
    }

    while( 1 )
    {
        double lowpassCoeff = 0.9, lowpassCoeff1 = 0.99999 - lowpassCoeff;
        
        /* Test before and after in case whatever underlying sleep call isn't interrupted by pthread_cancel */
        pthread_testcancel();
        Pa_Sleep( intervalMsec );
        pthread_testcancel();

        if( PaUtil_GetTime() - th->callbackTime > maxSeconds )
        {
            PA_DEBUG(( "Watchdog: Terminating callback thread\n" ));
            /* Tell thread to terminate */
            err = pthread_kill( th->callbackThread, SIGKILL );
            pthread_exit( NULL );
        }

        PA_DEBUG(( "%s: PortAudio reports CPU load: %g\n", __FUNCTION__, PaUtil_GetCpuLoad( th->cpuLoadMeasurer ) ));

        /* Check if we should throttle, or unthrottle :P */
        cpuTimeNow = th->callbackCpuTime;
        cpuTimeElapsed = cpuTimeNow - cpuTimeThen;
        cpuTimeThen = cpuTimeNow;

        timeNow = PaUtil_GetTime();
        timeElapsed = timeNow - timeThen;
        timeThen = timeNow;
        cpuLoad = cpuTimeElapsed / timeElapsed;
        avgCpuLoad = avgCpuLoad * lowpassCoeff + cpuLoad * lowpassCoeff1;
        /*
        if( throttled )
            PA_DEBUG(( "Watchdog: CPU load: %g, %g\n", avgCpuLoad, cpuTimeElapsed ));
            */
        if( PaUtil_GetCpuLoad( th->cpuLoadMeasurer ) > .925 )
        {
            static int policy;
            static struct sched_param spm = { 0 };
            static const struct sched_param defaultSpm = { 0 };
            PA_DEBUG(( "%s: Throttling audio thread, priority %d\n", __FUNCTION__, spm.sched_priority ));

            pthread_getschedparam( th->callbackThread, &policy, &spm );
            if( !pthread_setschedparam( th->callbackThread, SCHED_OTHER, &defaultSpm ) )
            {
                throttled = 1;
            }
            else
                PA_DEBUG(( "Watchdog: Couldn't lower priority of audio thread: %s\n", strerror( errno ) ));

            /* Give other processes a go, before raising priority again */
            PA_DEBUG(( "%s: Watchdog sleeping for %lu msecs before unthrottling\n", __FUNCTION__, th->throttledSleepTime ));
            Pa_Sleep( th->throttledSleepTime );

            /* Reset callback priority */
            if( pthread_setschedparam( th->callbackThread, SCHED_FIFO, &spm ) != 0 )
            {
                PA_DEBUG(( "%s: Couldn't raise priority of audio thread: %s\n", __FUNCTION__, strerror( errno ) ));
            }

            if( PaUtil_GetCpuLoad( th->cpuLoadMeasurer ) >= .99 )
                intervalMsec = 50;
            else
                intervalMsec = 100;

            /*
            lowpassCoeff = .97;
            lowpassCoeff1 = .99999 - lowpassCoeff;
            */
        }
        else if( throttled && avgCpuLoad < .8 )
        {
            intervalMsec = 500;
            throttled = 0;

            /*
            lowpassCoeff = .9;
            lowpassCoeff1 = .99999 - lowpassCoeff;
            */
        }
    }

    pthread_cleanup_pop( 1 );   /* Execute cleanup on exit */

error:
    /* Shouldn't get here in the normal case */

    /* Pass on error code */
    pres = malloc( sizeof (PaError) );
    *pres = result;
    
    pthread_exit( pres );
}

static void CallbackUpdate( PaAlsaThreading *th )
{
    th->callbackTime = PaUtil_GetTime();
    th->callbackCpuTime = PaUtil_GetCpuLoad( th->cpuLoadMeasurer );
}
#endif

/*
static void *CanaryFunc( void *userData )
{
    const unsigned intervalMsec = 1000;
    PaUtilThreading *th = (PaUtilThreading *) userData;

    while( 1 )
    {
        th->canaryTime = PaUtil_GetTime();

        pthread_testcancel();
        Pa_Sleep( intervalMsec );
    }

    pthread_exit( NULL );
}
*/