Qemu線程池介紹

有時咱們但願把一部分工做經過建立線程的方式異步執行，這樣咱們能夠在執行任務的同時，繼續執行其餘任務。可是若是這種需求比較多的話，頻繁的建立和銷燬線程帶來很大的性能損耗。若是咱們能建立一個或一些線程，而後重複使用它們，就能夠避免這個問題。

Qemu的實現

qemu模仿glib實現了線程池的功能，目前qemu中線程池主要應用在raw文件的支持上，當linux-aio不可用時，就像glibc，經過線程實現aio機制。咱們也可看到，表明線程池中的線程成員的數據結構 ThreadPoolElement 就包含了用來描述aio的BlockAIOCB結構。相關數據結構以下：

首先看一下線程池相關的數據結構：

typedef struct AIOCBInfo {

    void (*cancel_async)(BlockAIOCB *acb);

    AioContext *(*get_aio_context)(BlockAIOCB *acb);

    size_t aiocb_size;

} AIOCBInfo;

struct BlockAIOCB {

    const AIOCBInfo *aiocb_info;

    BlockDriverState *bs;

    BlockCompletionFunc *cb;

    void *opaque;

    int refcnt;

};

struct ThreadPoolElement {

    BlockAIOCB common;

    ThreadPool *pool;

    ThreadPoolFunc *func;

    void *arg;

    /* Moving state out of THREAD_QUEUED is protected by lock.  After

     * that, only the worker thread can write to it.  Reads and writes

     * of state and ret are ordered with memory barriers.

     */

    enum ThreadState state;

    int ret;

    /* Access to this list is protected by lock.  */

    QTAILQ_ENTRY(ThreadPoolElement) reqs;

    /* Access to this list is protected by the global mutex.  */

    QLIST_ENTRY(ThreadPoolElement) all;

};

struct ThreadPool {

    AioContext *ctx;

    QEMUBH *completion_bh;

    QemuMutex lock;

    QemuCond worker_stopped;

    QemuSemaphore sem;

    int max_threads;

    QEMUBH *new_thread_bh;

    /* The following variables are only accessed from one AioContext. */

    QLIST_HEAD(, ThreadPoolElement) head;

    /* The following variables are protected by lock.  */

    QTAILQ_HEAD(, ThreadPoolElement) request_list;

    int cur_threads;

    int idle_threads;

    int new_threads;     /* backlog of threads we need to create */

    int pending_threads; /* threads created but not running yet */

    bool stopping;

};

ThreadPool 數據結構負責維護線程池裏面的線程成員，線程的建立是經過下半部來實現的；ThreadPool 中有5個負責維護不一樣狀態下的線程成員的計數器，max_threads負責統計線程池中容許建立的線程的最大值； new_threads負責統計須要建立的線程數；pending_threads負責統計已建立但尚未運行的線程數；idle_threads負責統計空閒的線程數；cur_threads負責統計當前線程池中線程的個數；注意cur_threads包含new_threads中還沒有建立的線程。

線程池建立

首先經過thread_pool_new函數爲特定的AioContext實例建立一個新的線程池。在這個函數中初始化ThreadPool數據結構的各個成員，包括負責建立新線程的new_thread_bh和線程執行完畢後用來調度執行任務完成回調函數的completion_bh。

任務提交

咱們經過調用thread_pool_submit_aio函數來提交任務，這個函數的原型是：

BlockAIOCB *thread_pool_submit_aio(ThreadPool *pool,

ThreadPoolFunc *func, void *arg,

BlockCompletionFunc *cb, void *opaque)

這個函數爲提交的任務建立一個ThreadPoolElement實例添加到ThreadPool中，同時調用spawn_thread函數來建立一個qemu線程。spawn_thread函數是經過調度pool->new_thread_bh來建立qemu線程的。

線程執行

建立的線程執行worker_thread函數，這個函數從pool->request_list鏈表中取下第一個ThreadPoolElement節點，執行其任務函數，而後調度執行pool->completion_bh；這個bh遍歷pool->head鏈表，根據其ThreadPoolElement成員的狀態來決定是否執行實例上註冊的完成回調函數。

線程執行完一個任務後，也就是一個ThreadPoolElement實例被執行後，線程就出在idle狀態，等待下一個任務提交動做。任務提交與線程執行之間的同步是經過pool->sem來實現的。thread_pool_submit_aio中任務提交後會調用qemu_sem_post(&pool->sem)來增長pool->sem的計數，worker_thread在pool->sem上醒來後從pool->request_list鏈表上獲取下一個要執行的ThreadPoolElement節點。

總結

線程池計數提供了線程重複使用的功能，這在qemu有大量io操做的時候提升了性能；同時，也提供了除了linux-aio以外的aio實現。

 

參考：

https://developer.gnome.org/glib/2.46/glib-Thread-Pools.html