thread.cpp

Wxpython Implemented on Windows CE, Source code

{
    if ( m_critRegion != kInvalidID )
        MPDeleteCriticalRegion( m_critRegion );

    MPYield();
}

wxMutexError wxMutexInternal::Lock()
{
    wxCHECK_MSG( m_isOk , wxMUTEX_MISC_ERROR , wxT("Invalid Mutex") );

    OSStatus err = MPEnterCriticalRegion( m_critRegion, kDurationForever);
    if (err != noErr)
    {
        wxLogSysError(wxT("Could not lock mutex"));
        return wxMUTEX_MISC_ERROR;
    }

    return wxMUTEX_NO_ERROR;
}

wxMutexError wxMutexInternal::TryLock()
{
    wxCHECK_MSG( m_isOk , wxMUTEX_MISC_ERROR , wxT("Invalid Mutex") ) ;

    OSStatus err = MPEnterCriticalRegion( m_critRegion, kDurationImmediate);
    if (err != noErr)
    {
        if ( err == kMPTimeoutErr)
            return wxMUTEX_BUSY;

        wxLogSysError( wxT("Could not try lock mutex") );
        return wxMUTEX_MISC_ERROR;
    }

    return wxMUTEX_NO_ERROR;
}

wxMutexError wxMutexInternal::Unlock()
{
    wxCHECK_MSG( m_isOk , wxMUTEX_MISC_ERROR , wxT("Invalid Mutex") ) ;

    OSStatus err = MPExitCriticalRegion( m_critRegion );
    MPYield() ;

    if (err != noErr)
    {
        wxLogSysError( wxT("Could not unlock mutex") );

        return wxMUTEX_MISC_ERROR;
    }

    return wxMUTEX_NO_ERROR;
}

#endif

// --------------------------------------------------------------------------
// wxSemaphore
// --------------------------------------------------------------------------

class wxSemaphoreInternal
{
public:
    wxSemaphoreInternal( int initialcount, int maxcount );
    virtual ~wxSemaphoreInternal();

    bool IsOk() const
    { return m_isOk; }

    wxSemaError Post();
    wxSemaError WaitTimeout( unsigned long milliseconds );

    wxSemaError Wait()
    { return WaitTimeout( kDurationForever); }

    wxSemaError TryWait()
    {
        wxSemaError err = WaitTimeout( kDurationImmediate );
        if (err == wxSEMA_TIMEOUT)
            err = wxSEMA_BUSY;

        return err;
    }

private:
    MPSemaphoreID m_semaphore;
    bool m_isOk;
};

wxSemaphoreInternal::wxSemaphoreInternal( int initialcount, int maxcount)
{
    m_isOk = false;
    m_semaphore = kInvalidID;
    if ( maxcount == 0 )
        // make it practically infinite
        maxcount = INT_MAX;

    verify_noerr( MPCreateSemaphore( maxcount, initialcount, &m_semaphore ) );
    m_isOk = ( m_semaphore != kInvalidID );

    if ( !IsOk() )
    {
        wxFAIL_MSG( wxT("Error when creating semaphore") );
    }
}

wxSemaphoreInternal::~wxSemaphoreInternal()
{
    if (m_semaphore != kInvalidID)
        MPDeleteSemaphore( m_semaphore );

    MPYield();
}

wxSemaError wxSemaphoreInternal::WaitTimeout( unsigned long milliseconds )
{
    OSStatus err = MPWaitOnSemaphore( m_semaphore, milliseconds );
    if (err != noErr)
    {
        if (err == kMPTimeoutErr)
            return wxSEMA_TIMEOUT;

        return wxSEMA_MISC_ERROR;
    }

    return wxSEMA_NO_ERROR;
}

wxSemaError wxSemaphoreInternal::Post()
{
    OSStatus err = MPSignalSemaphore( m_semaphore );
    MPYield();
    if (err != noErr)
        return wxSEMA_MISC_ERROR;

    return wxSEMA_NO_ERROR;
}

// ----------------------------------------------------------------------------
// wxCondition implementation
// ----------------------------------------------------------------------------

#if 0

class wxConditionInternal
{
public:
    wxConditionInternal( wxMutex& mutex )
    :
    m_mutex( mutex ),
    m_semaphore( 0, 1 ),
    m_gate( 1, 1 )
    {
        m_waiters = 0;
        m_signals = 0;
        m_canceled = 0;
    }

    virtual ~wxConditionInternal() {}

    bool IsOk() const
    { return m_mutex.IsOk(); }

    wxCondError Wait()
    { return WaitTimeout( kDurationForever ); }

    wxCondError WaitTimeout( unsigned long msectimeout );

    wxCondError Signal()
    { return DoSignal( false); }

    wxCondError Broadcast()
    { return DoSignal( true ); }

private:
    wxCondError DoSignal( bool signalAll );

    wxMutex& m_mutex;
    wxSemaphoreInternal m_semaphore;  // Signals the waiting threads.
    wxSemaphoreInternal m_gate;
    wxCriticalSection m_varSection;
    size_t m_waiters; // Number of threads waiting for a signal.
    size_t m_signals; // Number of signals to send.
    size_t m_canceled; // Number of canceled waiters in m_waiters.
};

wxCondError wxConditionInternal::WaitTimeout( unsigned long msectimeout )
{
    m_gate.Wait();

    if ( ++ m_waiters == INT_MAX )
    {
        m_varSection.Enter();

        m_waiters -= m_canceled;
        m_signals -= m_canceled;
        m_canceled = 0;

        m_varSection.Leave();
    }

    m_gate.Post();
    m_mutex.Unlock();

    wxSemaError err = m_semaphore.WaitTimeout( msectimeout);
    wxASSERT( err == wxSEMA_NO_ERROR || err == wxSEMA_TIMEOUT);

    m_varSection.Enter();

    if ( err != wxSEMA_NO_ERROR )
    {
        if ( m_signals > m_canceled )
        {
            // A signal is being sent after we timed out.
            if ( m_waiters == m_signals )
            {
                // There are no excess waiters to catch the signal, so
                // we must throw it away.
                wxSemaError err2 = m_semaphore.Wait();
                if ( err2 != wxSEMA_NO_ERROR )
                {
                    wxLogSysError( wx("Error while waiting on semaphore") );
                }

                wxASSERT( err2 == wxSEMA_NO_ERROR);

                --m_waiters;
                if ( --m_signals == m_canceled )
                {
                    // This was the last signal. open the gate.
                    wxASSERT( m_waiters == m_canceled );
                    m_gate.Post();
                }
            }
            else
            {
                // There are excess waiters to catch the signal, leave it be.
                --m_waiters;
            }
        }
        else
        {
            // No signals is being sent:
            // the gate may be open or closed, so we can't touch m_waiters.
            ++m_canceled;
            ++m_signals;
        }
    }
    else
    {
        // We caught a signal.
        wxASSERT( m_signals > m_canceled );

        --m_waiters;

        if ( --m_signals == m_canceled)
        {
            // This was the last signal. open the gate.
            wxASSERT( m_waiters == m_canceled );

            m_gate.Post();
        }
    }

    m_varSection.Leave();
    m_mutex.Lock();

    if (err != noErr)
        return err == wxSEMA_TIMEOUT ? wxCOND_TIMEOUT : wxCOND_MISC_ERROR;

    return wxCOND_NO_ERROR;
}


wxCondError wxConditionInternal::DoSignal( bool signalAll)
{
    m_gate.Wait();
    m_varSection.Enter();

    wxASSERT( m_signals == m_canceled );

    if ( m_waiters == m_canceled)
    {
        m_varSection.Leave();
        m_gate.Post();
        return wxCOND_NO_ERROR;
    }

    if ( m_canceled > 0)
    {
        m_waiters -= m_canceled;
        m_signals = 0;
        m_canceled = 0;
    }

    m_signals = signalAll ? m_waiters : 1;
    size_t n = m_signals;

    m_varSection.Leave();

    // Let the waiters inherit the gate lock.

    do
    {
        wxSemaError err = m_semaphore.Post();
        wxASSERT( err == wxSEMA_NO_ERROR );
    }
    while ( --n );

    return wxCOND_NO_ERROR;
}

#else
class wxConditionInternal
{
public:
    wxConditionInternal( wxMutex& mutex );

    bool IsOk() const
    { return m_mutex.IsOk() && m_semaphore.IsOk(); }

    wxCondError Wait();
    wxCondError WaitTimeout( unsigned long milliseconds );

    wxCondError Signal();
    wxCondError Broadcast();

private:
    // the number of threads currently waiting for this condition
    SInt32 m_numWaiters;

    // the critical section protecting m_numWaiters
    wxCriticalSection m_csWaiters;

    wxMutex& m_mutex;
    wxSemaphore m_semaphore;

    DECLARE_NO_COPY_CLASS(wxConditionInternal)
};

wxConditionInternal::wxConditionInternal( wxMutex& mutex )
    : m_mutex(mutex)
{
    // another thread can't access it until we return from ctor, so no need to
    // protect access to m_numWaiters here
    m_numWaiters = 0;
}

wxCondError wxConditionInternal::Wait()
{
    // increment the number of waiters
    IncrementAtomic( &m_numWaiters );

    m_mutex.Unlock();

    // a potential race condition can occur here
    //
    // after a thread increments nwaiters, and unlocks the mutex and before the
    // semaphore.Wait() is called, if another thread can cause a signal to be
    // generated
    //
    // this race condition is handled by using a semaphore and incrementing the
    // semaphore only if 'nwaiters' is greater that zero since the semaphore,
    // can 'remember' signals the race condition will not occur

    // wait ( if necessary ) and decrement semaphore
    wxSemaError err = m_semaphore.Wait();
    m_mutex.Lock();

    return err == wxSEMA_NO_ERROR ? wxCOND_NO_ERROR : wxCOND_MISC_ERROR;
}

wxCondError wxConditionInternal::WaitTimeout( unsigned long milliseconds )
{
    IncrementAtomic( &m_numWaiters );

    m_mutex.Unlock();

    // a race condition can occur at this point in the code
    //
    // please see the comments in Wait(), for details

    wxSemaError err = m_semaphore.WaitTimeout(milliseconds);

    if ( err == wxSEMA_BUSY )
    {
        // another potential race condition exists here it is caused when a
        // 'waiting' thread timesout, and returns from WaitForSingleObject, but
        // has not yet decremented 'nwaiters'.
        //
        // at this point if another thread calls signal() then the semaphore
        // will be incremented, but the waiting thread will miss it.
        //
        // to handle this particular case, the waiting thread calls
        // WaitForSingleObject again with a timeout of 0, after locking
        // 'nwaiters_mutex'. this call does not block because of the zero
        // timeout, but will allow the waiting thread to catch the missed
        // signals.
        wxCriticalSectionLocker lock(m_csWaiters);

        err = m_semaphore.WaitTimeout(0);

        if ( err != wxSEMA_NO_ERROR )
        {
            m_numWaiters--;
        }
    }

    m_mutex.Lock();

    return err == wxSEMA_NO_ERROR ? wxCOND_NO_ERROR : wxCOND_MISC_ERROR;
}

wxCondError wxConditionInternal::Signal()
{
    wxCriticalSectionLocker lock(m_csWaiters);

    if ( m_numWaiters > 0 )
    {
        // increment the semaphore by 1
        if ( m_semaphore.Post() != wxSEMA_NO_ERROR )
            return wxCOND_MISC_ERROR;

        m_numWaiters--;
    }

    return wxCOND_NO_ERROR;
}

wxCondError wxConditionInternal::Broadcast()
{
    wxCriticalSectionLocker lock(m_csWaiters);

    while ( m_numWaiters > 0 )
    {
        if ( m_semaphore.Post() != wxSEMA_NO_ERROR )
            return wxCOND_MISC_ERROR;

        m_numWaiters--;
    }

    return wxCOND_NO_ERROR;
}
#endif