qemu里面有个服务于aio的线程池：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
struct ThreadPool { AioContext *ctx; QEMUBH *completion_bh; QemuMutex lock; QemuCond check_cancel; 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 */ int pending_cancellations; /* whether we need a cond_broadcast */ bool stopping; }; |
qemu在thread_pool_init_one建池子的时候注册了一个延迟执行的work：
pool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);
这个work将在spawn_thread里面被安排调度
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
static void spawn_thread(ThreadPool *pool) { pool->cur_threads++; pool->new_threads++; //这里是为了记录下spawn thread被调用的次数，为后面安排的创建线程的后半部提供数据，因为不是每次调用进来都会安排后半部工作的 /* If there are threads being created, they will spawn new workers, so * we don't spend time creating many threads in a loop holding a mutex or * starving the current vcpu. * * If there are no idle threads, ask the main thread to create one, so we * inherit the correct affinity instead of the vcpu affinity. */ if (!pool->pending_threads) { //如果前面已经有安排，就不再重复安排 qemu_bh_schedule(pool->new_thread_bh); } |
当work被调度到的时候，执行spawn_thread_bh_fn
|
1 2 3 4 5 6 7 8 |
static void spawn_thread_bh_fn(void *opaque) { ThreadPool *pool = opaque; qemu_mutex_lock(&pool->lock); do_spawn_thread(pool); qemu_mutex_unlock(&pool->lock); } |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
static void do_spawn_thread(ThreadPool *pool) { QemuThread t; /* Runs with lock taken. */ if (!pool->new_threads) { return; } pool->new_threads--; pool->pending_threads++; //避免重复安排 qemu_thread_create(&t, "worker", worker_thread, pool, QEMU_THREAD_DETACHED); } |
这里采用了递归的方式来一次性创建多个worker，新创建的worker线程会递归调动do_spawn_thread来创建下一个worker，直到发现new_threads为0
这里有个问题，如果qemu_thread_create创建新线程失败，那么就会导致后面新的线程永远无法创建，因为pending_threads不会被减扣为0，前面spawn_thread就不会再安排新的下半部工作来创建线程了。
下面是aio派发任务的函数thread_pool_submit_aio，可以看到，池子中的线程数是动态增加的，如果有空闲的线程或者线程数已达上限是不会创建新的线程的，并且采用了信号量通知的方法来减少线程轮询开销，有任务的时候才放开一个额度，而不是让线程一直尝试拿后面的mutex锁再去看下request list是不是空。
|
1 2 3 4 5 6 7 8 |
qemu_mutex_lock(&pool->lock); if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) { spawn_thread(pool); } QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs); qemu_mutex_unlock(&pool->lock); qemu_sem_post(&pool->sem); //这里会增加sem的计数，让一个idle的worker获得执行机会 return &req->common; |
worker创建的时候以及worker在func执行完后会再次尝试获取pool->sem（等待sem>0），等待新的任务：
|
1 2 3 4 5 |
pool->idle_threads++; qemu_mutex_unlock(&pool->lock); ret = qemu_sem_timedwait(&pool->sem, 10000);//减扣sem计数 qemu_mutex_lock(&pool->lock); pool->idle_threads--; |
池子资源释放的时候，会标记pool->stopping并给所有worker一个最后的任务（通过sem_post）：释放自己：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
thread_pool_free: /* Stop new threads from spawning */ qemu_bh_delete(pool->new_thread_bh); pool->cur_threads -= pool->new_threads; pool->new_threads = 0; /* Wait for worker threads to terminate */ pool->stopping = true; while (pool->cur_threads > 0) { qemu_sem_post(&pool->sem); qemu_cond_wait(&pool->worker_stopped, &pool->lock); } |
