threadpsx.cpp

来自「A*算法 A*算法 A*算法 A*算法A*算法A*算法」· C++ 代码 · 共 1,725 行 · 第 1/4 页

    int err = pthread_cond_broadcast(&m_cond);
    if ( err != 0 )
    {
        wxLogApiError(_T("pthread_cond_broadcast()"), err);

        return wxCOND_MISC_ERROR;
    }

    return wxCOND_NO_ERROR;
}

// ===========================================================================
// wxSemaphore implementation
// ===========================================================================

// ---------------------------------------------------------------------------
// wxSemaphoreInternal
// ---------------------------------------------------------------------------

// we implement the semaphores using mutexes and conditions instead of using
// the sem_xxx() POSIX functions because they're not widely available and also
// because it's impossible to implement WaitTimeout() using them
class wxSemaphoreInternal
{
public:
    wxSemaphoreInternal(int initialcount, int maxcount);

    bool IsOk() const { return m_isOk; }

    wxSemaError Wait();
    wxSemaError TryWait();
    wxSemaError WaitTimeout(unsigned long milliseconds);

    wxSemaError Post();

private:
    wxMutex m_mutex;
    wxCondition m_cond;

    size_t m_count,
           m_maxcount;

    bool m_isOk;
};

wxSemaphoreInternal::wxSemaphoreInternal(int initialcount, int maxcount)
                   : m_cond(m_mutex)
{

    if ( (initialcount < 0 || maxcount < 0) ||
            ((maxcount > 0) && (initialcount > maxcount)) )
    {
        wxFAIL_MSG( _T("wxSemaphore: invalid initial or maximal count") );

        m_isOk = FALSE;
    }
    else
    {
        m_maxcount = (size_t)maxcount;
        m_count = (size_t)initialcount;
    }

    m_isOk = m_mutex.IsOk() && m_cond.IsOk();
}

wxSemaError wxSemaphoreInternal::Wait()
{
    wxMutexLocker locker(m_mutex);

    while ( m_count == 0 )
    {
        wxLogTrace(TRACE_SEMA,
                   _T("Thread %ld waiting for semaphore to become signalled"),
                   wxThread::GetCurrentId());

        if ( m_cond.Wait() != wxCOND_NO_ERROR )
            return wxSEMA_MISC_ERROR;

        wxLogTrace(TRACE_SEMA,
                   _T("Thread %ld finished waiting for semaphore, count = %lu"),
                   wxThread::GetCurrentId(), (unsigned long)m_count);
    }

    m_count--;

    return wxSEMA_NO_ERROR;
}

wxSemaError wxSemaphoreInternal::TryWait()
{
    wxMutexLocker locker(m_mutex);

    if ( m_count == 0 )
        return wxSEMA_BUSY;

    m_count--;

    return wxSEMA_NO_ERROR;
}

wxSemaError wxSemaphoreInternal::WaitTimeout(unsigned long milliseconds)
{
    wxMutexLocker locker(m_mutex);

    wxLongLong startTime = wxGetLocalTimeMillis();

    while ( m_count == 0 )
    {
        wxLongLong elapsed = wxGetLocalTimeMillis() - startTime;
        long remainingTime = (long)milliseconds - (long)elapsed.GetLo();
        if ( remainingTime <= 0 )
        {
            // timeout
            return wxSEMA_TIMEOUT;
        }

        switch ( m_cond.WaitTimeout(remainingTime) )
        {
            case wxCOND_TIMEOUT:
                return wxSEMA_TIMEOUT;

            default:
                return wxSEMA_MISC_ERROR;

            case wxCOND_NO_ERROR:
                ;
        }
    }

    m_count--;

    return wxSEMA_NO_ERROR;
}

wxSemaError wxSemaphoreInternal::Post()
{
    wxMutexLocker locker(m_mutex);

    if ( m_maxcount > 0 && m_count == m_maxcount )
    {
        return wxSEMA_OVERFLOW;
    }

    m_count++;

    wxLogTrace(TRACE_SEMA,
               _T("Thread %ld about to signal semaphore, count = %lu"),
               wxThread::GetCurrentId(), (unsigned long)m_count);

    return m_cond.Signal() == wxCOND_NO_ERROR ? wxSEMA_NO_ERROR
                                              : wxSEMA_MISC_ERROR;
}

// ===========================================================================
// wxThread implementation
// ===========================================================================

// the thread callback functions must have the C linkage
extern "C"
{

#ifdef wxHAVE_PTHREAD_CLEANUP
    // thread exit function
    void wxPthreadCleanup(void *ptr);
#endif // wxHAVE_PTHREAD_CLEANUP

void *wxPthreadStart(void *ptr);

} // extern "C"

// ----------------------------------------------------------------------------
// wxThreadInternal
// ----------------------------------------------------------------------------

class wxThreadInternal
{
public:
    wxThreadInternal();
    ~wxThreadInternal();

    // thread entry function
    static void *PthreadStart(wxThread *thread);

    // thread actions
        // start the thread
    wxThreadError Run();
        // unblock the thread allowing it to run
    void SignalRun() { m_semRun.Post(); }
        // ask the thread to terminate
    void Wait();
        // go to sleep until Resume() is called
    void Pause();
        // resume the thread
    void Resume();

    // accessors
        // priority
    int GetPriority() const { return m_prio; }
    void SetPriority(int prio) { m_prio = prio; }
        // state
    wxThreadState GetState() const { return m_state; }
    void SetState(wxThreadState state)
    {
#ifdef __WXDEBUG__
        static const wxChar *stateNames[] =
        {
            _T("NEW"),
            _T("RUNNING"),
            _T("PAUSED"),
            _T("EXITED"),
        };

        wxLogTrace(TRACE_THREADS, _T("Thread %ld: %s => %s."),
                   (long)GetId(), stateNames[m_state], stateNames[state]);
#endif // __WXDEBUG__

        m_state = state;
    }
        // id
    pthread_t GetId() const { return m_threadId; }
    pthread_t *GetIdPtr() { return &m_threadId; }
        // "cancelled" flag
    void SetCancelFlag() { m_cancelled = TRUE; }
    bool WasCancelled() const { return m_cancelled; }
        // exit code
    void SetExitCode(wxThread::ExitCode exitcode) { m_exitcode = exitcode; }
    wxThread::ExitCode GetExitCode() const { return m_exitcode; }

        // the pause flag
    void SetReallyPaused(bool paused) { m_isPaused = paused; }
    bool IsReallyPaused() const { return m_isPaused; }

        // tell the thread that it is a detached one
    void Detach()
    {
        wxCriticalSectionLocker lock(m_csJoinFlag);

        m_shouldBeJoined = FALSE;
        m_isDetached = TRUE;
    }

#ifdef wxHAVE_PTHREAD_CLEANUP
    // this is used by wxPthreadCleanup() only
    static void Cleanup(wxThread *thread);
#endif // wxHAVE_PTHREAD_CLEANUP

private:
    pthread_t     m_threadId;   // id of the thread
    wxThreadState m_state;      // see wxThreadState enum
    int           m_prio;       // in wxWidgets units: from 0 to 100

    // this flag is set when the thread should terminate
    bool m_cancelled;

    // this flag is set when the thread is blocking on m_semSuspend
    bool m_isPaused;

    // the thread exit code - only used for joinable (!detached) threads and
    // is only valid after the thread termination
    wxThread::ExitCode m_exitcode;

    // many threads may call Wait(), but only one of them should call
    // pthread_join(), so we have to keep track of this
    wxCriticalSection m_csJoinFlag;
    bool m_shouldBeJoined;
    bool m_isDetached;

    // this semaphore is posted by Run() and the threads Entry() is not
    // called before it is done
    wxSemaphore m_semRun;

    // this one is signaled when the thread should resume after having been
    // Pause()d
    wxSemaphore m_semSuspend;
};

// ----------------------------------------------------------------------------
// thread startup and exit functions
// ----------------------------------------------------------------------------

void *wxPthreadStart(void *ptr)
{
    return wxThreadInternal::PthreadStart((wxThread *)ptr);
}

void *wxThreadInternal::PthreadStart(wxThread *thread)
{
    wxThreadInternal *pthread = thread->m_internal;

    wxLogTrace(TRACE_THREADS, _T("Thread %ld started."), THR_ID(pthread));

    // associate the thread pointer with the newly created thread so that
    // wxThread::This() will work
    int rc = pthread_setspecific(gs_keySelf, thread);
    if ( rc != 0 )
    {
        wxLogSysError(rc, _("Cannot start thread: error writing TLS"));

        return (void *)-1;
    }

    // have to declare this before pthread_cleanup_push() which defines a
    // block!
    bool dontRunAtAll;

#ifdef wxHAVE_PTHREAD_CLEANUP
    // install the cleanup handler which will be called if the thread is
    // cancelled
    pthread_cleanup_push(wxPthreadCleanup, thread);
#endif // wxHAVE_PTHREAD_CLEANUP

    // wait for the semaphore to be posted from Run()
    pthread->m_semRun.Wait();

    // test whether we should run the run at all - may be it was deleted
    // before it started to Run()?
    {
        wxCriticalSectionLocker lock(thread->m_critsect);

        dontRunAtAll = pthread->GetState() == STATE_NEW &&
                       pthread->WasCancelled();
    }

    if ( !dontRunAtAll )
    {
        // call the main entry
        wxLogTrace(TRACE_THREADS,
                   _T("Thread %ld about to enter its Entry()."),
                   THR_ID(pthread));

        pthread->m_exitcode = thread->Entry();

        wxLogTrace(TRACE_THREADS,
                   _T("Thread %ld Entry() returned %lu."),
                   THR_ID(pthread), (unsigned long)pthread->m_exitcode);

        {
            wxCriticalSectionLocker lock(thread->m_critsect);

            // change the state of the thread to "exited" so that
            // wxPthreadCleanup handler won't do anything from now (if it's
            // called before we do pthread_cleanup_pop below)
            pthread->SetState(STATE_EXITED);
        }
    }

    // NB: at least under Linux, pthread_cleanup_push/pop are macros and pop
    //     contains the matching '}' for the '{' in push, so they must be used
    //     in the same block!
#ifdef wxHAVE_PTHREAD_CLEANUP
    // remove the cleanup handler without executing it
    pthread_cleanup_pop(FALSE);
#endif // wxHAVE_PTHREAD_CLEANUP

    if ( dontRunAtAll )
    {
        // FIXME: deleting a possibly joinable thread here???
        delete thread;

        return EXITCODE_CANCELLED;
    }
    else
    {
        // terminate the thread
        thread->Exit(pthread->m_exitcode);

        wxFAIL_MSG(wxT("wxThread::Exit() can't return."));

        return NULL;
    }
}

#ifdef wxHAVE_PTHREAD_CLEANUP

// this handler is called when the thread is cancelled
extern "C" void wxPthreadCleanup(void *ptr)
{
    wxThreadInternal::Cleanup((wxThread *)ptr);
}

void wxThreadInternal::Cleanup(wxThread *thread)
{
    if (pthread_getspecific(gs_keySelf) == 0) return;
    {
        wxCriticalSectionLocker lock(thread->m_critsect);
        if ( thread->m_internal->GetState() == STATE_EXITED )
        {
            // thread is already considered as finished.
            return;
        }
    }

    // exit the thread gracefully
    thread->Exit(EXITCODE_CANCELLED);
}

#endif // wxHAVE_PTHREAD_CLEANUP

// ----------------------------------------------------------------------------
// wxThreadInternal
// ----------------------------------------------------------------------------

wxThreadInternal::wxThreadInternal()
{
    m_state = STATE_NEW;
    m_cancelled = FALSE;
    m_prio = WXTHREAD_DEFAULT_PRIORITY;
    m_threadId = 0;
    m_exitcode = 0;

    // set to TRUE only when the thread starts waiting on m_semSuspend
    m_isPaused = FALSE;

    // defaults for joinable threads
    m_shouldBeJoined = TRUE;
    m_isDetached = FALSE;
}

wxThreadInternal::~wxThreadInternal()
{
}

wxThreadError wxThreadInternal::Run()
{
    wxCHECK_MSG( GetState() == STATE_NEW, wxTHREAD_RUNNING,
                 wxT("thread may only be started once after Create()") );

    SetState(STATE_RUNNING);

    // wake up threads waiting for our start
    SignalRun();