qemu\u91cc\u9762\u6709\u4e2a\u670d\u52a1\u4e8eaio\u7684\u7ebf\u7a0b\u6c60\uff1a<\/p>\n
struct ThreadPool {\n AioContext *ctx;\n QEMUBH *completion_bh;\n QemuMutex lock;\n QemuCond check_cancel;\n QemuCond worker_stopped;\n QemuSemaphore sem;\n int max_threads;\n QEMUBH *new_thread_bh;\n\n \/* The following variables are only accessed from one AioContext. *\/\n QLIST_HEAD(, ThreadPoolElement) head;\n\n \/* The following variables are protected by lock. *\/\n QTAILQ_HEAD(, ThreadPoolElement) request_list;\n int cur_threads;\n int idle_threads;\n int new_threads; \/* backlog of threads we need to create *\/\n int pending_threads; \/* threads created but not running yet *\/\n int pending_cancellations; \/* whether we need a cond_broadcast *\/\n bool stopping;\n};<\/pre>\nqemu\u5728thread_pool_init_one\u5efa\u6c60\u5b50\u7684\u65f6\u5019\u6ce8\u518c\u4e86\u4e00\u4e2a\u5ef6\u8fdf\u6267\u884c\u7684work\uff1a
\npool->new_thread_bh = aio_bh_new(ctx, spawn_thread_bh_fn, pool);<\/p>\n
\u8fd9\u4e2awork\u5c06\u5728spawn_thread\u91cc\u9762\u88ab\u5b89\u6392\u8c03\u5ea6<\/p>\n
static void spawn_thread(ThreadPool *pool)\n{\n pool->cur_threads++;\n pool->new_threads++; \/\/\u8fd9\u91cc\u662f\u4e3a\u4e86\u8bb0\u5f55\u4e0bspawn thread\u88ab\u8c03\u7528\u7684\u6b21\u6570\uff0c\u4e3a\u540e\u9762\u5b89\u6392\u7684\u521b\u5efa\u7ebf\u7a0b\u7684\u540e\u534a\u90e8\u63d0\u4f9b\u6570\u636e\uff0c\u56e0\u4e3a\u4e0d\u662f\u6bcf\u6b21\u8c03\u7528\u8fdb\u6765\u90fd\u4f1a\u5b89\u6392\u540e\u534a\u90e8\u5de5\u4f5c\u7684\n \/* If there are threads being created, they will spawn new workers, so\n * we don't spend time creating many threads in a loop holding a mutex or\n * starving the current vcpu.\n *\n * If there are no idle threads, ask the main thread to create one, so we\n * inherit the correct affinity instead of the vcpu affinity.\n *\/\n if (!pool->pending_threads) { \/\/\u5982\u679c\u524d\u9762\u5df2\u7ecf\u6709\u5b89\u6392\uff0c\u5c31\u4e0d\u518d\u91cd\u590d\u5b89\u6392\n qemu_bh_schedule(pool->new_thread_bh);\n }<\/pre>\n <\/p>\n
\u5f53work\u88ab\u8c03\u5ea6\u5230\u7684\u65f6\u5019\uff0c\u6267\u884cspawn_thread_bh_fn<\/p>\n
static void spawn_thread_bh_fn(void *opaque)\n{\n ThreadPool *pool = opaque;\n\n qemu_mutex_lock(&pool->lock);\n do_spawn_thread(pool);\n qemu_mutex_unlock(&pool->lock);\n}<\/pre>\nstatic void do_spawn_thread(ThreadPool *pool)\n{\n QemuThread t;\n\n \/* Runs with lock taken. *\/\n if (!pool->new_threads) {\n return;\n }\n\n pool->new_threads--;\n pool->pending_threads++; \/\/\u907f\u514d\u91cd\u590d\u5b89\u6392\n\n qemu_thread_create(&t, \"worker\", worker_thread, pool, QEMU_THREAD_DETACHED);\n}<\/pre>\n <\/p>\n
\u8fd9\u91cc\u91c7\u7528\u4e86\u9012\u5f52\u7684\u65b9\u5f0f\u6765\u4e00\u6b21\u6027\u521b\u5efa\u591a\u4e2aworker\uff0c\u65b0\u521b\u5efa\u7684worker\u7ebf\u7a0b\u4f1a\u9012\u5f52\u8c03\u52a8do_spawn_thread\u6765\u521b\u5efa\u4e0b\u4e00\u4e2aworker\uff0c\u76f4\u5230\u53d1\u73b0new_threads\u4e3a0<\/p>\n
\u8fd9\u91cc\u6709\u4e2a\u95ee\u9898\uff0c\u5982\u679cqemu_thread_create\u521b\u5efa\u65b0\u7ebf\u7a0b\u5931\u8d25\uff0c\u90a3\u4e48\u5c31\u4f1a\u5bfc\u81f4\u540e\u9762\u65b0\u7684\u7ebf\u7a0b\u6c38\u8fdc\u65e0\u6cd5\u521b\u5efa\uff0c\u56e0\u4e3apending_threads\u4e0d\u4f1a\u88ab\u51cf\u6263\u4e3a0\uff0c\u524d\u9762spawn_thread\u5c31\u4e0d\u4f1a\u518d\u5b89\u6392\u65b0\u7684\u4e0b\u534a\u90e8\u5de5\u4f5c\u6765\u521b\u5efa\u7ebf\u7a0b\u4e86\u3002<\/p>\n
\u4e0b\u9762\u662faio\u6d3e\u53d1\u4efb\u52a1\u7684\u51fd\u6570thread_pool_submit_aio\uff0c\u53ef\u4ee5\u770b\u5230\uff0c\u6c60\u5b50\u4e2d\u7684\u7ebf\u7a0b\u6570\u662f\u52a8\u6001\u589e\u52a0\u7684\uff0c\u5982\u679c\u6709\u7a7a\u95f2\u7684\u7ebf\u7a0b\u6216\u8005\u7ebf\u7a0b\u6570\u5df2\u8fbe\u4e0a\u9650\u662f\u4e0d\u4f1a\u521b\u5efa\u65b0\u7684\u7ebf\u7a0b\u7684\uff0c\u5e76\u4e14\u91c7\u7528\u4e86\u4fe1\u53f7\u91cf\u901a\u77e5\u7684\u65b9\u6cd5\u6765\u51cf\u5c11\u7ebf\u7a0b\u8f6e\u8be2\u5f00\u9500\uff0c\u6709\u4efb\u52a1\u7684\u65f6\u5019\u624d\u653e\u5f00\u4e00\u4e2a\u989d\u5ea6\uff0c\u800c\u4e0d\u662f\u8ba9\u7ebf\u7a0b\u4e00\u76f4\u5c1d\u8bd5\u62ff\u540e\u9762\u7684mutex\u9501\u518d\u53bb\u770b\u4e0brequest list\u662f\u4e0d\u662f\u7a7a\u3002<\/p>\n
qemu_mutex_lock(&pool->lock);\n if (pool->idle_threads == 0 && pool->cur_threads < pool->max_threads) {\n spawn_thread(pool);\n }\n QTAILQ_INSERT_TAIL(&pool->request_list, req, reqs);\n qemu_mutex_unlock(&pool->lock);\n qemu_sem_post(&pool->sem); \/\/\u8fd9\u91cc\u4f1a\u589e\u52a0sem\u7684\u8ba1\u6570\uff0c\u8ba9\u4e00\u4e2aidle\u7684worker\u83b7\u5f97\u6267\u884c\u673a\u4f1a\n return &req->common;<\/pre>\n <\/p>\n
worker\u521b\u5efa\u7684\u65f6\u5019\u4ee5\u53caworker\u5728func\u6267\u884c\u5b8c\u540e\u4f1a\u518d\u6b21\u5c1d\u8bd5\u83b7\u53d6pool->sem\uff08\u7b49\u5f85sem>0\uff09\uff0c\u7b49\u5f85\u65b0\u7684\u4efb\u52a1\uff1a<\/p>\n
pool->idle_threads++;\nqemu_mutex_unlock(&pool->lock);\nret = qemu_sem_timedwait(&pool->sem, 10000);\/\/\u51cf\u6263sem\u8ba1\u6570\nqemu_mutex_lock(&pool->lock);\npool->idle_threads--;<\/pre>\n\u6c60\u5b50\u8d44\u6e90\u91ca\u653e\u7684\u65f6\u5019\uff0c\u4f1a\u6807\u8bb0pool->stopping\u5e76\u7ed9\u6240\u6709worker\u4e00\u4e2a\u6700\u540e\u7684\u4efb\u52a1\uff08\u901a\u8fc7sem_post\uff09\uff1a\u91ca\u653e\u81ea\u5df1\uff1a<\/p>\n
thread_pool_free:\n\n\/* Stop new threads from spawning *\/\nqemu_bh_delete(pool->new_thread_bh);\npool->cur_threads -= pool->new_threads;\npool->new_threads = 0;\n\n\/* Wait for worker threads to terminate *\/\npool->stopping = true;\nwhile (pool->cur_threads > 0) {\nqemu_sem_post(&pool->sem);\nqemu_cond_wait(&pool->worker_stopped, &pool->lock);\n}<\/pre>\n