C++：多線程類庫的設(shè)計與實現(xiàn)（六）

just_person 2012-03-15

展開全文

線程類的具體實現(xiàn)

一、將接口和實現(xiàn)相分離

分離接口和實現(xiàn)是OO設(shè)計中的重要思想。它其實反映了現(xiàn)實世界中抽象和具象、一般和特殊的基本現(xiàn)象和規(guī)律。而我們，正是基于這種思想，來完成我們類庫的跨平臺特性。

基本上所有的OO開發(fā)環(huán)境都提供了支持接口和實現(xiàn)分離的機制。很多機制是基于編程語言本身的，如可繼承性和多態(tài)性，而有些機制還基于體系結(jié)構(gòu)特性，比如CORBA和COM。

我們把有關(guān)線程的具體實現(xiàn)封裝到“實現(xiàn)類”里面，命名為TheadImpl。實際上ThreadImpl只是一個薄層封裝，為上層Thread類提供一個統(tǒng)一線程API調(diào)用界面。

二、包裝系統(tǒng)的API

雖然只是一個薄層封裝，我們不得不做常常是讓類庫設(shè)計者最頭疼的一件事情---統(tǒng)一操作系統(tǒng)API的調(diào)用界面。每種操作系統(tǒng)都提供API，常常都是相互風(fēng)格迥異，這給統(tǒng)一包裝API界面帶來很多困難---雖然同時也增加了這項工作的藝術(shù)性。設(shè)計者不得不充分調(diào)研各種對象操作系統(tǒng)，了解每個API的調(diào)用規(guī)格，詳細歸納，慎重取舍，以設(shè)計出良好的統(tǒng)一包裝界面。在這里系統(tǒng)編程經(jīng)驗至關(guān)重要。

雖然本系列的文章順序是先介紹線程類Thread的接口，再來討論ThreadImpl的具體實現(xiàn)，這可能會給讀者帶來自上而下（Top down）的設(shè)計路線的印象。而實際上，和許多工作的實際過程一樣，C++類庫設(shè)計也不是簡單的自上而下或者自下而上（Bottom up），而是一個PDCA循環(huán)（規(guī)劃 Plan-執(zhí)行 DO-檢查 Check-措施 Action）。據(jù)我的個人經(jīng)驗，每個類庫在第一版設(shè)計完成后都要經(jīng)過重復(fù)數(shù)次的修改，以達到一個相對穩(wěn)定的版本。

以上的經(jīng)驗之談是希望讀者能理解和大多數(shù)工作一樣C++類庫設(shè)計也不是一件一蹴而就的直線性工作。下面介紹我們的對象系統(tǒng)的線程API---我們的包裝對象。

三、盤點POSIX與Windows的線程API
和其他POSIX系統(tǒng)API一樣，POSIX線程的API在POSIX規(guī)范 IEEE 1003系列中被定義，目前的版本為2004年版。POSIX線程提供豐富的系統(tǒng)調(diào)用，不過我們的線程類只使用其中的一部分。如果讀者手頭沒有IEEE的文檔也沒有關(guān)系，因為POSIX線程被廣泛實現(xiàn)于各種Linux平臺，絕大多數(shù)Linux系統(tǒng)的手冊頁（Manual page）都有相關(guān)的敘述。

創(chuàng)建線程：

POSIX：
int pthread_create (
     pthread_t *restrict thread,
     const pthread_attr_t *restrict attr,
     void *(*start_routine)(void*),
     void *restrict arg
);

pthread_create()創(chuàng)建線程并立刻執(zhí)行它。

Windows：
HANDLE CreateThread(
      LPSECURITY_ATTRIBUTES lpThreadAttributes,
    SIZE_T dwStackSize,
    LPTHREAD_START_ROUTINE lpStartAddress,
    LPVOID lpParameter,
    DWORD dwCreationFlags,
    LPDWORD lpThreadId
);

CreateThread()允許在dwCreationFlags參數(shù)中指定線程是否以掛起狀態(tài)被創(chuàng)建（CREATE_SUSPENDED標(biāo)志）。另外CreateThread允許通過dwStackSize指定線程初始棧大小。

這兩個函數(shù)都接受一個用戶例程函數(shù)指針，作為線程運行的用戶代碼入口。

用戶例程的形式定義：
POSIX：
void * start_routine(void * param);

Windows：
DWORD WINAPI ThreadProc(LPVOID lpParameter);

設(shè)置線程初始棧：

Posix：
通過線程屬性設(shè)定。

int pthread_attr_setstacksize(pthread_attr_t *attr, size_t stacksize);

Windows：
在CreateThread()中指定。

判斷線程死活/有效性

POSIX：
IEEE 1003中并沒有明確指定判斷線程死活的API。不過很多人使用pthread_kill（）：
int pthread_kill(pthread_t thread, int sig);

通過發(fā)送信號0來測試線程的有效性。這樣的做的依據(jù)可能來源于標(biāo)準(zhǔn)IEEE 1003關(guān)于pthread_kill（）的一段注釋：
“Upon successful completion, the function shall return a value of zero. The pthread_kill ( ) function shall request that a signal be delivered to the specified thread. As in kill( ), if sig is zero, error checking shall be performed but no signal shall actually be sent.”

另外規(guī)范中對于pthread_kill()的返回值定義如下：
[ESRCH] No thread could be found corresponding to that specified by the given thread ID.
[EINVAL] The value of the sig argument is an invalid or unsupported signal number.

調(diào)用pthread_kill，發(fā)送信號0，檢測pthread_kill的返回值。如果pthread_t型的ID指向一個有效并且執(zhí)行中的線程，則pthread_kill應(yīng)該返回0;如果是一個無效ID或者線程已經(jīng)結(jié)束，則pthread_kill應(yīng)該返回ESRCH。

這看起來是個不錯的辦法，可惜筆者在實踐中發(fā)現(xiàn)，在某些POSIX實現(xiàn)（具體的說是Cygwin的某些版本）中，調(diào)用pthread_kill并傳遞一個無效的線程ID將直接導(dǎo)致指針操作異常而使程序異常結(jié)束。

筆者發(fā)現(xiàn)使用另一個API好像更為安全：

int pthread_getschedparam(
    pthread_t thread,
    int *restrict policy,
    struct sched_param *restrict param
);

pthread_getschedparam()有類似的調(diào)用語義規(guī)范：

“RETURN VALUE
If successful, the pthread_getschedparam( ) and pthread_setschedparam( ) functions shall return zero; otherwise, an error number shall be returned to indicate the error.
The pthread_getschedparam() function may fail if:
[ESRCH] The value specified by thread does not refer to an existing thread.”

所以，在我們的版本中，用pthread_getschedparam取代pthread_kill來測試線程死活。pthread_getschedparam的這一用途在筆者所測試的Cygwin和Linux版本中發(fā)揮正常。

Windows：
DWORD WaitForSingleObject(
    HANDLE hHandle,
    DWORD dwMilliseconds
);

Windows編程的規(guī)范做法，是調(diào)用WaitForSingleObject，指定等待時間為0。如果返回值為WAIT_TIMEOUT則說明線程為有效，否則為無效線程HANDLE或者線程已經(jīng)結(jié)束。

線程的睡眠函數(shù)：
POSIX:
unsigned sleep(unsigned seconds);
int usleep(useconds_t useconds);

這兩個API提供不同時間粒度的睡眠。我們可以把它們組合起來以實現(xiàn)毫秒級粒度的睡眠功能。另外某些老版本的Linux實現(xiàn)中把sleep()和usleep()解釋為進程級別的睡眠API---這已經(jīng)過時了。根據(jù)IEEE 1003 2004的說明：
“The sleep( ) function shall cause the calling thread to be suspended from execution until either the number of realtime seconds specified by the argument seconds has elapsed or a signal is delivered to the calling thread and its action is to invoke a signal-catching function or to terminate the process.”

usleep也有內(nèi)似的說明。目前的流行Linux版本的sleep和usleep都是線程級別的。

Windows：

VOID Sleep(DWORD dwMilliseconds);

強行結(jié)束線程：
POSIX：
int pthread_cancel(pthread_t thread);

Windows：
BOOL TerminateThread(HANDLE hThread, DWORD dwExitCode);

程序員應(yīng)該盡量避免調(diào)用強行結(jié)束線程的API，尤其是在C++程序設(shè)計中。我們將在后續(xù)的文章中解釋其原因。

四、完成線程實現(xiàn)類的代碼
下面給出POSIX線程實現(xiàn)類代碼。代碼中使用了Java風(fēng)格的注釋。

//---- ThreadImpl_Linux.h----

#ifndef THREAD_IMPL_LINUX_H
#define THREAD_IMPL_LINUX_H

#include <pthread.h>
#include <unistd.h>

class Thread;

/**
* Linux(POSIX)版本的線程實現(xiàn)類。
*/
class ThreadImpl
{

public:
    static const int        WAIT_INFINITE            = -1;
    static const int        INVALID_THREAD_ID        = 0;
    static const size_t     DEFAULT_STACK_SIZE       = 0;
    static const int        WAIT_TIME_SLICE        = 10;


public:

    /**
    * 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。包含線程的平臺相關(guān)信息。
    */
    struct ThreadStruct {
        pthread_t     tThread;

        ThreadStruct()
            :
            tThread(INVALID_THREAD_ID)

        {
        }

        ~ThreadStruct() {
        }
    };


public:
    typedef int (* THREAD_FUNCTION) (Thread *pParam);
    typedef void * (*POSIX_THREAD_ROUTINE) (void *pParam);

public:

    /**
    * 創(chuàng)建線程。
    * @param ts線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    * @param thread_func 用戶例程函數(shù)指針。
    * @param cbStackSize 線程初始棧大小，字節(jié)單位。
    * @param pThread 綁定的Thread對象指針。
    * @return 調(diào)用成功返回true，失敗返回false。
    */
    static bool CreateThread(ThreadStruct & ts,
                    THREAD_FUNCTION thread_func,
                    size_t cbStackSize,
                    Thread * pThread) {


        pthread_attr_t *pAttr = NULL;
        pthread_attr_t attr;

        if(cbStackSize != DEFAULT_STACK_SIZE) {

            if(0 != pthread_attr_init(&attr)) {
                return false;
            }

            if(0 != pthread_attr_setstacksize(&attr, cbStackSize)) {
                pthread_attr_destroy(&attr);
                return false;
            }

            pAttr = &attr;
        }

        int iRes = pthread_create(&(ts.tThread),
                            pAttr,
                            (POSIX_THREAD_ROUTINE)thread_func,
                            (void*)pThread);

        if(NULL != pAttr) {
            pthread_attr_destroy(&attr);
        }

        if(0 != iRes) {
            return false;
        }

        pthread_detach(ts.tThread);

        return true;
    }

    /**
    * 銷毀線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static void DestroyThread(ThreadStruct & ts) {
    }

    /**
    * 在指定時間內(nèi)等待線程結(jié)束。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。
    * @param iTimeout 指定的等待時間。
    * @return 線程在指定時間內(nèi)結(jié)束的情況下返回true，否則返回false。
    * 如果傳入的線程結(jié)構(gòu)體未關(guān)聯(lián)到有效線程，則返回true。
    */
    static bool WaitForThreadEnd(const ThreadStruct & ts, int iTimeout) {

        int iDelta = WAIT_TIME_SLICE;
        int iTotal = iTimeout;

        if(iTimeout == WAIT_INFINITE) { // Cause to do unlimited loop.
            iDelta = 0;
            iTotal = 1;
        }

        for(int i=0; i<iTotal; i+=iDelta) {
            if(!IsAlive(ts)) {
                return true;
            } else {
                Sleep(WAIT_TIME_SLICE);
            }
        }

        return false;
    }

    /**
    * 強行結(jié)束線程。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static void TerminateThread(ThreadStruct & ts) {
        ::pthread_cancel(ts.tThread);
    }

    /**
    * 判斷線程死活。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    * @return 線程為有效則返回true，否則返回false。
    */
    static bool IsAlive(const ThreadStruct & ts) {

        int iPolicy;
        struct sched_param sp;

        int iRes = pthread_getschedparam(ts.tThread, &iPolicy, &sp);

        if(0 == iRes) {
            return true;
        } else {
            return false;
        }
    }

    /**
    * 獲取線程ID。該ID在進程域內(nèi)唯一。
    * @param ts線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static int GetThreadId(const ThreadStruct & ts) {
        return (int) ts.tThread;
    }

    /**
    * 導(dǎo)致調(diào)用線程被掛起若干時間。
    * @param iMs 線程掛起時間，毫秒單位。
    */
    static void Sleep(int iMs) {

        int iS = iMs / 1000;
        int iUs = (iMs % 1000) * 1000;

        if(iS > 0) {
            sleep(iS);
        }

        if(iUs > 0) {
            usleep(iUs);
        }

        return;
    }

private:
    // 禁止構(gòu)造函數(shù)。
    ThreadImpl();

    // 禁止拷貝構(gòu)造函數(shù)。
    ThreadImpl(const ThreadImpl &) throw();

    // 禁止賦值操作符。
    void operator=(const ThreadImpl &);
}; // class ThreadImpl

#endif // #ifndef ThreadImpl_LINUX_H

//----EOF----

Windows的線程實現(xiàn)類的代碼如下：

//----ThreadImpl_WIN32.h----

#ifndef THREAD_IMPL_WIN32_H
#define THREAD_IMPL_WIN32_H

#include <Windows.h>

class Thread;

/**
* Windows 32位平臺版本的線程實現(xiàn)類。
*/
class ThreadImpl
{

public:
    static const int        WAIT_INFINITE            = -1;
    static const int        INVALID_THREAD_ID        = 0;
    static const size_t    DEFAULT_STACK_SIZE        = 0;

public:
    /**
    * 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。包含線程的平臺相關(guān)信息。
    */
    struct ThreadStruct {

        HANDLE     hThread;
        DWORD    dwThreadId;

        ThreadStruct()
            :
            hThread(NULL),
            dwThreadId(INVALID_THREAD_ID)
        {
        }

        void Cleanup() {
            if(hThread != NULL) {
                CloseHandle(hThread);
                hThread = NULL;
            }

        }

        ~ThreadStruct() {
            Cleanup();
        }
    };

public:
    typedef int (* THREAD_FUNCTION) (Thread *pParam);

public:

    /**
    * 創(chuàng)建線程。
    * @param ts線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    * @param thread_func 用戶例程函數(shù)指針。
    * @param cbStackSize 線程初始棧大小，字節(jié)單位。
    * @param pThread 綁定的Thread對象指針。
    * @return 調(diào)用成功返回true，失敗返回false。
    */

    static bool CreateThread(ThreadStruct & ts,
                    THREAD_FUNCTION thread_func,
                    size_t cbStackSize,
                    Thread * pThread) {


        ts.hThread = ::CreateThread(NULL,
                    (DWORD) cbStackSize,
                    (LPTHREAD_START_ROUTINE) thread_func,
                    (LPVOID) pThread,
                    0,
                    &ts.dwThreadId);

        if(ts.hThread == NULL) {
            return false;
        }

        return true;
    }

    /**
    * 銷毀線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static void DestroyThread(ThreadStruct & ts) {
        ts.Cleanup();
    }

    /**
    * 在指定時間內(nèi)等待線程結(jié)束。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體。
    * @param iTimeout 指定的等待時間。
    * @return 線程在指定時間內(nèi)結(jié)束的情況下返回true，否則返回false。
    * 如果傳入的線程結(jié)構(gòu)體未關(guān)聯(lián)到有效線程，則返回true。
    */
    static bool WaitForThreadEnd(const ThreadStruct & ts, int iTimeout) {

        DWORD dwTO = (DWORD) iTimeout;

        if(iTimeout == WAIT_INFINITE) {
            dwTO = INFINITE;
        }

        DWORD dwRes = WaitForSingleObject(ts.hThread, dwTO);

        if(dwRes == WAIT_TIMEOUT) {
            return false;
        } else {
            return true;
        }
    }

    /**
    * 強行結(jié)束線程。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static void TerminateThread(const ThreadStruct & ts) {
        ::TerminateThread(ts.hThread, -1);
    }

    /**
    * 判斷線程死活。
    * @param ts 線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    * @return 線程為有效則返回true，否則返回false。
    */
    static bool IsAlive(const ThreadStruct & ts) {

        if(NULL == ts.hThread) {
            return false;
        }

        DWORD dwRes = WaitForSingleObject(ts.hThread, 0);

        if(WAIT_TIMEOUT == dwRes) {
            return true;
        } else if(WAIT_OBJECT_0 == dwRes || WAIT_FAILED == dwRes) {
            return false;
        } else {
            return false;
        }
    }

    /**
    * 獲取線程ID。該ID在進程域內(nèi)唯一。
    * @param ts線程實現(xiàn)類的內(nèi)部結(jié)構(gòu)體型變量。
    */
    static int GetThreadId(const ThreadStruct & ts) {
        return (int) ts.dwThreadId;
    }

    /**
    * 導(dǎo)致調(diào)用線程被掛起若干時間。
    * @param iMs 線程掛起時間，毫秒單位。
    */
    static void Sleep(int iMs) {
        return ::Sleep(iMs);
    }

private:
    // 禁止構(gòu)造函數(shù)。
    ThreadImpl();

    // 禁止拷貝構(gòu)造函數(shù)。
    ThreadImpl(const ThreadImpl &) throw();
// 禁止賦值操作符。
    void operator=(const ThreadImpl &);
}; // class ThreadImpl

#endif // #ifndef ThreadImpl_WIN32_H

//----EOF----

對上面的代碼稍微做一點說明。

Linux版本的CreateThread()中，創(chuàng)建完線程后立即調(diào)用pthread_detach()，以通知POSIX線程庫可以在該線程結(jié)束時立即釋放線程資源。注意到我們不使用pthread_join()來等待線程結(jié)束。POSIX的現(xiàn)程模型使用pthread_join()來實現(xiàn)簡單的線程調(diào)度，在筆者看來用起來很不方便，因為pthread_join()只能是無限等待，不提供超時機制，大大降低了程序設(shè)計的靈活性。

為了實現(xiàn)帶超時的線程等待函數(shù)，如我們的WaitForThreadEnd()，我們不得不把等待時間分解成更小的時間片輪詢。這樣做損失了一點CPU資源，并稍微影響了超時判斷的時間精度。

Windows版本的TerminateThread()中，我們將線程的結(jié)束代碼(Exit code)一律指定為-1。我們的上層線程類不使用這個結(jié)束代碼。

Linux版本和Windows版本的線程實現(xiàn)類都不提供線程優(yōu)先級的相關(guān)功能。