Thanks Changpeng for your reply. I will find it out. On 2021/11/12 12:59, Liu, Changpeng wrote: > Hi Jieyue, > > Thanks for submitting patch to SPDK, but we use gerrit for code development, you can refer https://spdk.io/development/ for submitting your patch. > > >> -----Original Message----- >> From: majieyue(a)linux.alibaba.com <majieyue(a)linux.alibaba.com&gt; >> Sent: Friday, November 12, 2021 12:56 PM >> To: spdk(a)lists.01.org >> Cc: Ma Jie Yue <majieyue(a)linux.alibaba.com&gt; >> Subject: [SPDK] [RFC PATCH v1] [RFC] Use multiple threads to handle vhost >> virtqueues >> >> From: Ma Jie Yue <majieyue(a)linux.alibaba.com&gt; >> >> Currently the vhost virtqueues of the same device are handled by only one spdk >> thread, even we have many reactors running, which means the performance of a >> vhost device can not be scaled up with multiple cores. >> >> This patch bind each virtqueue to an individual spdk thread, and leverage the >> spdk scheduler ability to run these threads on different reactors. Now only >> the vhost blk module is finished, and just leave the vhost scsi later. >> >> During the test, the spdk_top shows theses threads are indeed dispatched to >> different reactors, and the IO performance is also increased with the number >> of queues. >> >> Signed-off-by: Ma Jie Yue <majieyue(a)linux.alibaba.com&gt; >> --- >> lib/vhost/vhost.c | 158 +++++++++++++++---- >> lib/vhost/vhost_blk.c | 384 +++++++++++++++++++++------------------------ >> lib/vhost/vhost_internal.h | 28 +++- >> 3 files changed, 324 insertions(+), 246 deletions(-) >> >> diff --git a/lib/vhost/vhost.c b/lib/vhost/vhost.c >> index edae5938e..80343dd1c 100644 >> --- a/lib/vhost/vhost.c >> +++ b/lib/vhost/vhost.c >> @@ -1089,6 +1089,46 @@ vhost_session_stop_done(struct spdk_vhost_session >> *vsession, int response) >> vhost_session_cb_done(response); >> } >> >> +void >> +vhost_session_start_vq_done(struct spdk_vhost_virtqueue *vq, int response) >> +{ >> + struct spdk_vhost_session *vsession = vq->vsession; >> + >> + if (response == 0) { >> + vq->started = true; >> + vsession->active_queues++; >> + >> + if (vsession->active_queues == vsession->max_queues) { >> + vsession->started = true; >> + >> + assert(vsession->vdev->active_session_num < >> UINT32_MAX); >> + vsession->vdev->active_session_num++; >> + } >> + } >> + >> + vhost_session_cb_done(response); >> +} >> + >> +void >> +vhost_session_stop_vq_done(struct spdk_vhost_virtqueue *vq, int response) >> +{ >> + struct spdk_vhost_session *vsession = vq->vsession; >> + >> + if (response == 0) { >> + vq->started = false; >> + vsession->active_queues--; >> + >> + if (vsession->active_queues == 0) { >> + vsession->started = false; >> + >> + assert(vsession->vdev->active_session_num > 0); >> + vsession->vdev->active_session_num--; >> + } >> + } >> + >> + vhost_session_cb_done(response); >> +} >> + >> static void >> vhost_event_cb(void *arg1) >> { >> @@ -1101,7 +1141,7 @@ vhost_event_cb(void *arg1) >> } >> >> vsession = vhost_session_find_by_id(ctx->vdev, ctx->vsession_id); >> - ctx->cb_fn(ctx->vdev, vsession, NULL); >> + ctx->cb_fn(ctx->vdev, vsession, ctx->user_ctx); >> pthread_mutex_unlock(&g_vhost_mutex); >> } >> >> @@ -1126,6 +1166,34 @@ vhost_session_send_event(struct spdk_vhost_session >> *vsession, >> return g_dpdk_response; >> } >> >> +int >> +vhost_session_send_event_mt(struct spdk_vhost_session *vsession, >> + spdk_vhost_session_fn cb_fn, unsigned timeout_sec, >> + const char *errmsg) >> +{ >> + struct vhost_session_fn_ctx ev_ctx = {0}; >> + struct spdk_vhost_dev *vdev = vsession->vdev; >> + unsigned long i; >> + >> + ev_ctx.vdev = vdev; >> + ev_ctx.vsession_id = vsession->id; >> + ev_ctx.cb_fn = cb_fn; >> + >> + for (i = 0; i < vsession->max_queues; i++) { >> + ev_ctx.user_ctx = (void *)i; >> + spdk_thread_send_msg(vsession->thread[i], vhost_event_cb, >> &ev_ctx); >> + >> + pthread_mutex_unlock(&g_vhost_mutex); >> + wait_for_semaphore(timeout_sec, errmsg); >> + pthread_mutex_lock(&g_vhost_mutex); >> + >> + if (g_dpdk_response) >> + break; >> + } >> + >> + return g_dpdk_response; >> +} >> + >> static void >> foreach_session_finish_cb(void *arg1) >> { >> @@ -1250,7 +1318,7 @@ int >> vhost_stop_device_cb(int vid) >> { >> struct spdk_vhost_session *vsession; >> - int rc; >> + int i, rc; >> >> pthread_mutex_lock(&g_vhost_mutex); >> vsession = vhost_session_find_by_vid(vid); >> @@ -1267,6 +1335,14 @@ vhost_stop_device_cb(int vid) >> } >> >> rc = _stop_session(vsession); >> + >> + /* clean up the threads */ >> + if (!rc) { >> + for (i = 0; i < vsession->max_queues; i++) { >> + spdk_thread_send_msg(vsession->thread[i], >> vhost_dev_thread_exit, NULL); >> + } >> + } >> + >> pthread_mutex_unlock(&g_vhost_mutex); >> >> return rc; >> @@ -1280,6 +1356,7 @@ vhost_start_device_cb(int vid) >> int rc = -1; >> uint16_t i; >> bool packed_ring; >> + struct spdk_cpuset *cpumask; >> >> pthread_mutex_lock(&g_vhost_mutex); >> >> @@ -1304,9 +1381,11 @@ vhost_start_device_cb(int vid) >> packed_ring = ((vsession->negotiated_features & (1ULL << >> VIRTIO_F_RING_PACKED)) != 0); >> >> vsession->max_queues = 0; >> + cpumask = spdk_thread_get_cpumask(vdev->thread); >> memset(vsession->virtqueue, 0, sizeof(vsession->virtqueue)); >> for (i = 0; i < SPDK_VHOST_MAX_VQUEUES; i++) { >> struct spdk_vhost_virtqueue *q = &vsession->virtqueue[i]; >> + char *name; >> >> q->vsession = vsession; >> q->vring_idx = -1; >> @@ -1362,6 +1441,16 @@ vhost_start_device_cb(int vid) >> } >> >> q->packed.packed_ring = packed_ring; >> + >> + name = spdk_sprintf_alloc("%s.%u", vsession->name, i); >> + vsession->thread[i] = spdk_thread_create(name, cpumask); >> + free(name); >> + if (!vsession->thread[i]) { >> + SPDK_ERRLOG("Failed to create thread for >> virtqueue %s.%u", vsession->name, i); >> + rc = -EIO; >> + goto out; >> + } >> + >> vsession->max_queues = i + 1; >> } >> >> @@ -1401,55 +1490,56 @@ vhost_start_device_cb(int vid) >> } >> >> out: >> + if (rc) { >> + for (i = 0; i < vsession->max_queues; i++) { >> + spdk_thread_send_msg(vsession->thread[i], >> vhost_dev_thread_exit, NULL); >> + } >> + } >> pthread_mutex_unlock(&g_vhost_mutex); >> return rc; >> } >> >> void >> -vhost_session_set_interrupt_mode(struct spdk_vhost_session *vsession, bool >> interrupt_mode) >> +vhost_session_set_vq_interrupt_mode(struct spdk_vhost_virtqueue *q, bool >> interrupt_mode) >> { >> - uint16_t i; >> bool packed_ring; >> int rc = 0; >> + uint64_t num_events = 1; >> + struct spdk_vhost_session *vsession = q->vsession; >> >> packed_ring = ((vsession->negotiated_features & (1ULL << >> VIRTIO_F_RING_PACKED)) != 0); >> >> - for (i = 0; i < vsession->max_queues; i++) { >> - struct spdk_vhost_virtqueue *q = &vsession->virtqueue[i]; >> - uint64_t num_events = 1; >> + /* vring.desc and vring.desc_packed are in a union struct >> + * so q->vring.desc can replace q->vring.desc_packed. >> + */ >> + if (q->vring.desc == NULL || q->vring.size == 0) { >> + return; >> + } >> >> - /* vring.desc and vring.desc_packed are in a union struct >> - * so q->vring.desc can replace q->vring.desc_packed. >> - */ >> - if (q->vring.desc == NULL || q->vring.size == 0) { >> - continue; >> + if (interrupt_mode) { >> + /* Enable I/O submission notifications, we'll be interrupting. */ >> + if (packed_ring) { >> + * (volatile uint16_t *) &q->vring.device_event->flags = >> VRING_PACKED_EVENT_FLAG_ENABLE; >> + } else { >> + * (volatile uint16_t *) &q->vring.used->flags = 0; >> } >> >> - if (interrupt_mode) { >> - /* Enable I/O submission notifications, we'll be >> interrupting. */ >> - if (packed_ring) { >> - * (volatile uint16_t *) &q->vring.device_event- >>> flags = VRING_PACKED_EVENT_FLAG_ENABLE; >> - } else { >> - * (volatile uint16_t *) &q->vring.used->flags = 0; >> - } >> - >> - /* In case of race condition, always kick vring when >> switch to intr */ >> - rc = write(q->vring.kickfd, &num_events, >> sizeof(num_events)); >> - if (rc < 0) { >> - SPDK_ERRLOG("failed to kick vring: %s.\n", >> spdk_strerror(errno)); >> - } >> + /* In case of race condition, always kick vring when switch to intr >> */ >> + rc = write(q->vring.kickfd, &num_events, sizeof(num_events)); >> + if (rc < 0) { >> + SPDK_ERRLOG("failed to kick vring: %s.\n", >> spdk_strerror(errno)); >> + } >> >> - vsession->interrupt_mode = true; >> + vsession->interrupt_mode = true; >> + } else { >> + /* Disable I/O submission notifications, we'll be polling. */ >> + if (packed_ring) { >> + * (volatile uint16_t *) &q->vring.device_event->flags = >> VRING_PACKED_EVENT_FLAG_DISABLE; >> } else { >> - /* Disable I/O submission notifications, we'll be polling. >> */ >> - if (packed_ring) { >> - * (volatile uint16_t *) &q->vring.device_event- >>> flags = VRING_PACKED_EVENT_FLAG_DISABLE; >> - } else { >> - * (volatile uint16_t *) &q->vring.used->flags = >> VRING_USED_F_NO_NOTIFY; >> - } >> - >> - vsession->interrupt_mode = false; >> + * (volatile uint16_t *) &q->vring.used->flags = >> VRING_USED_F_NO_NOTIFY; >> } >> + >> + vsession->interrupt_mode = false; >> } >> } >> >> diff --git a/lib/vhost/vhost_blk.c b/lib/vhost/vhost_blk.c >> index 55fb82530..78821ce09 100644 >> --- a/lib/vhost/vhost_blk.c >> +++ b/lib/vhost/vhost_blk.c >> @@ -102,17 +102,20 @@ struct spdk_vhost_blk_session { >> /* The parent session must be the very first field in this struct */ >> struct spdk_vhost_session vsession; >> struct spdk_vhost_blk_dev *bvdev; >> - struct spdk_poller *requestq_poller; >> - struct spdk_io_channel *io_channel; >> - struct spdk_poller *stop_poller; >> + struct spdk_poller *requestq_poller[SPDK_VHOST_MAX_VQUEUES]; >> + struct spdk_io_channel *io_channel[SPDK_VHOST_MAX_VQUEUES]; >> + struct spdk_poller *stop_poller[SPDK_VHOST_MAX_VQUEUES]; >> }; >> >> /* forward declaration */ >> +static int vhost_blk_stop_vq_cb(struct spdk_vhost_dev *vdev, >> + struct spdk_vhost_session *vsession, void >> *unused); >> + >> static const struct spdk_vhost_dev_backend vhost_blk_device_backend; >> >> static int >> process_blk_request(struct spdk_vhost_blk_task *task, >> - struct spdk_vhost_blk_session *bvsession); >> + struct spdk_vhost_virtqueue *vq); >> >> static struct spdk_vhost_blk_session * >> to_blk_session(struct spdk_vhost_session *vsession) >> @@ -124,8 +127,8 @@ to_blk_session(struct spdk_vhost_session *vsession) >> static void >> blk_task_finish(struct spdk_vhost_blk_task *task) >> { >> - assert(task->bvsession->vsession.task_cnt > 0); >> - task->bvsession->vsession.task_cnt--; >> + assert(task->vq->task_cnt > 0); >> + task->vq->task_cnt--; >> task->used = false; >> } >> >> @@ -421,7 +424,7 @@ blk_request_resubmit(void *arg) >> struct spdk_vhost_blk_task *task = (struct spdk_vhost_blk_task *)arg; >> int rc = 0; >> >> - rc = process_blk_request(task, task->bvsession); >> + rc = process_blk_request(task, task->vq); >> if (rc == 0) { >> SPDK_DEBUGLOG(vhost_blk, "====== Task %p resubmitted >> ======\n", task); >> } else { >> @@ -435,12 +438,13 @@ blk_request_queue_io(struct spdk_vhost_blk_task >> *task) >> int rc; >> struct spdk_vhost_blk_session *bvsession = task->bvsession; >> struct spdk_bdev *bdev = bvsession->bvdev->bdev; >> + struct spdk_vhost_virtqueue *vq = task->vq; >> >> task->bdev_io_wait.bdev = bdev; >> task->bdev_io_wait.cb_fn = blk_request_resubmit; >> task->bdev_io_wait.cb_arg = task; >> >> - rc = spdk_bdev_queue_io_wait(bdev, bvsession->io_channel, &task- >>> bdev_io_wait); >> + rc = spdk_bdev_queue_io_wait(bdev, bvsession->io_channel[vq- >>> vring_idx], &task->bdev_io_wait); >> if (rc != 0) { >> SPDK_ERRLOG("%s: failed to queue I/O, rc=%d\n", bvsession- >>> vsession.name, rc); >> invalid_blk_request(task, VIRTIO_BLK_S_IOERR); >> @@ -449,8 +453,10 @@ blk_request_queue_io(struct spdk_vhost_blk_task *task) >> >> static int >> process_blk_request(struct spdk_vhost_blk_task *task, >> - struct spdk_vhost_blk_session *bvsession) >> + struct spdk_vhost_virtqueue *vq) >> { >> + struct spdk_vhost_session *vsession = vq->vsession; >> + struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> struct spdk_vhost_blk_dev *bvdev = bvsession->bvdev; >> const struct virtio_blk_outhdr *req; >> struct virtio_blk_discard_write_zeroes *desc; >> @@ -503,12 +509,12 @@ process_blk_request(struct spdk_vhost_blk_task *task, >> >> if (type == VIRTIO_BLK_T_IN) { >> task->used_len = payload_len + sizeof(*task->status); >> - rc = spdk_bdev_readv(bvdev->bdev_desc, bvsession- >>> io_channel, >> + rc = spdk_bdev_readv(bvdev->bdev_desc, bvsession- >>> io_channel[vq->vring_idx], >> &task->iovs[1], task->iovcnt, req- >>> sector * 512, >> payload_len, >> blk_request_complete_cb, task); >> } else if (!bvdev->readonly) { >> task->used_len = sizeof(*task->status); >> - rc = spdk_bdev_writev(bvdev->bdev_desc, bvsession- >>> io_channel, >> + rc = spdk_bdev_writev(bvdev->bdev_desc, bvsession- >>> io_channel[vq->vring_idx], >> &task->iovs[1], task->iovcnt, req- >>> sector * 512, >> payload_len, >> blk_request_complete_cb, task); >> } else { >> @@ -540,7 +546,7 @@ process_blk_request(struct spdk_vhost_blk_task *task, >> return -1; >> } >> >> - rc = spdk_bdev_unmap(bvdev->bdev_desc, bvsession- >>> io_channel, >> + rc = spdk_bdev_unmap(bvdev->bdev_desc, bvsession- >>> io_channel[vq->vring_idx], >> desc->sector * 512, desc->num_sectors * 512, >> blk_request_complete_cb, task); >> if (rc) { >> @@ -570,7 +576,7 @@ process_blk_request(struct spdk_vhost_blk_task *task, >> (uint64_t)desc->sector * 512, (uint64_t)desc- >>> num_sectors * 512); >> } >> >> - rc = spdk_bdev_write_zeroes(bvdev->bdev_desc, bvsession- >>> io_channel, >> + rc = spdk_bdev_write_zeroes(bvdev->bdev_desc, bvsession- >>> io_channel[vq->vring_idx], >> desc->sector * 512, desc->num_sectors >> * 512, >> blk_request_complete_cb, task); >> if (rc) { >> @@ -590,7 +596,7 @@ process_blk_request(struct spdk_vhost_blk_task *task, >> invalid_blk_request(task, VIRTIO_BLK_S_IOERR); >> return -1; >> } >> - rc = spdk_bdev_flush(bvdev->bdev_desc, bvsession->io_channel, >> + rc = spdk_bdev_flush(bvdev->bdev_desc, bvsession- >>> io_channel[vq->vring_idx], >> 0, flush_bytes, >> blk_request_complete_cb, task); >> if (rc) { >> @@ -639,7 +645,7 @@ process_blk_task(struct spdk_vhost_virtqueue *vq, >> uint16_t req_idx) >> return; >> } >> >> - task->bvsession->vsession.task_cnt++; >> + vq->task_cnt++; >> >> blk_task_init(task); >> >> @@ -653,7 +659,7 @@ process_blk_task(struct spdk_vhost_virtqueue *vq, >> uint16_t req_idx) >> return; >> } >> >> - if (process_blk_request(task, task->bvsession) == 0) { >> + if (process_blk_request(task, vq) == 0) { >> SPDK_DEBUGLOG(vhost_blk, "====== Task %p req_idx %d >> submitted ======\n", task, >> req_idx); >> } else { >> @@ -702,7 +708,7 @@ process_packed_blk_task(struct spdk_vhost_virtqueue >> *vq, uint16_t req_idx) >> req_idx, (req_idx + num_descs - 1) % >> vq->vring.size, >> &task->inflight_head); >> >> - task->bvsession->vsession.task_cnt++; >> + vq->task_cnt++; >> >> blk_task_init(task); >> >> @@ -715,7 +721,7 @@ process_packed_blk_task(struct spdk_vhost_virtqueue >> *vq, uint16_t req_idx) >> return; >> } >> >> - if (process_blk_request(task, task->bvsession) == 0) { >> + if (process_blk_request(task, vq) == 0) { >> SPDK_DEBUGLOG(vhost_blk, "====== Task %p req_idx %d >> submitted ======\n", task, >> task_idx); >> } else { >> @@ -760,7 +766,7 @@ process_packed_inflight_blk_task(struct >> spdk_vhost_virtqueue *vq, >> /* It's for cleaning inflight entries */ >> task->inflight_head = req_idx; >> >> - task->bvsession->vsession.task_cnt++; >> + vq->task_cnt++; >> >> blk_task_init(task); >> >> @@ -773,7 +779,7 @@ process_packed_inflight_blk_task(struct >> spdk_vhost_virtqueue *vq, >> return; >> } >> >> - if (process_blk_request(task, task->bvsession) == 0) { >> + if (process_blk_request(task, vq) == 0) { >> SPDK_DEBUGLOG(vhost_blk, "====== Task %p req_idx %d >> submitted ======\n", task, >> task_idx); >> } else { >> @@ -893,20 +899,6 @@ vdev_vq_worker(void *arg) >> return _vdev_vq_worker(vq); >> } >> >> -static int >> -vdev_worker(void *arg) >> -{ >> - struct spdk_vhost_blk_session *bvsession = arg; >> - struct spdk_vhost_session *vsession = &bvsession->vsession; >> - uint16_t q_idx; >> - >> - for (q_idx = 0; q_idx < vsession->max_queues; q_idx++) { >> - _vdev_vq_worker(&vsession->virtqueue[q_idx]); >> - } >> - >> - return SPDK_POLLER_BUSY; >> -} >> - >> static void >> no_bdev_process_vq(struct spdk_vhost_blk_session *bvsession, struct >> spdk_vhost_virtqueue *vq) >> { >> @@ -985,9 +977,9 @@ _no_bdev_vdev_vq_worker(struct spdk_vhost_virtqueue >> *vq) >> >> vhost_session_vq_used_signal(vq); >> >> - if (vsession->task_cnt == 0 && bvsession->io_channel) { >> - spdk_put_io_channel(bvsession->io_channel); >> - bvsession->io_channel = NULL; >> + if (vq->task_cnt == 0 && bvsession->io_channel[vq->vring_idx]) { >> + spdk_put_io_channel(bvsession->io_channel[vq->vring_idx]); >> + bvsession->io_channel[vq->vring_idx] = NULL; >> } >> >> return SPDK_POLLER_BUSY; >> @@ -1001,75 +993,55 @@ no_bdev_vdev_vq_worker(void *arg) >> return _no_bdev_vdev_vq_worker(vq); >> } >> >> -static int >> -no_bdev_vdev_worker(void *arg) >> -{ >> - struct spdk_vhost_blk_session *bvsession = arg; >> - struct spdk_vhost_session *vsession = &bvsession->vsession; >> - uint16_t q_idx; >> - >> - for (q_idx = 0; q_idx < vsession->max_queues; q_idx++) { >> - _no_bdev_vdev_vq_worker(&vsession->virtqueue[q_idx]); >> - } >> - >> - return SPDK_POLLER_BUSY; >> -} >> - >> static void >> -vhost_blk_session_unregister_interrupts(struct spdk_vhost_blk_session >> *bvsession) >> +vhost_blk_session_unregister_vq_interrupts(struct spdk_vhost_blk_session >> *bvsession, >> + int vq_idx) >> { >> struct spdk_vhost_session *vsession = &bvsession->vsession; >> struct spdk_vhost_virtqueue *vq; >> - int i; >> - >> - SPDK_DEBUGLOG(vhost_blk, "unregister virtqueues interrupt\n"); >> - for (i = 0; i < vsession->max_queues; i++) { >> - vq = &vsession->virtqueue[i]; >> - if (vq->intr == NULL) { >> - break; >> - } >> >> - SPDK_DEBUGLOG(vhost_blk, "unregister vq[%d]'s kickfd is %d\n", >> - i, vq->vring.kickfd); >> - spdk_interrupt_unregister(&vq->intr); >> + SPDK_DEBUGLOG(vhost_blk, "unregister virtqueues %d interrupt\n", >> vq_idx); >> + vq = &vsession->virtqueue[vq_idx]; >> + if (vq->intr == NULL) { >> + return; >> } >> + >> + SPDK_DEBUGLOG(vhost_blk, "unregister vq[%d]'s kickfd is %d\n", >> + vq_idx, vq->vring.kickfd); >> + spdk_interrupt_unregister(&vq->intr); >> } >> >> static int >> -vhost_blk_session_register_interrupts(struct spdk_vhost_blk_session >> *bvsession, >> - spdk_interrupt_fn fn, const char *name) >> +vhost_blk_session_register_vq_interrupts(struct spdk_vhost_blk_session >> *bvsession, >> + spdk_interrupt_fn fn, int vq_idx) >> { >> struct spdk_vhost_session *vsession = &bvsession->vsession; >> struct spdk_vhost_virtqueue *vq = NULL; >> - int i; >> - >> - SPDK_DEBUGLOG(vhost_blk, "Register virtqueues interrupt\n"); >> - for (i = 0; i < vsession->max_queues; i++) { >> - vq = &vsession->virtqueue[i]; >> - SPDK_DEBUGLOG(vhost_blk, "Register vq[%d]'s kickfd is %d\n", >> - i, vq->vring.kickfd); >> - >> - vq->intr = spdk_interrupt_register(vq->vring.kickfd, fn, vq, name); >> - if (vq->intr == NULL) { >> - SPDK_ERRLOG("Fail to register req notifier handler.\n"); >> - goto err; >> - } >> + >> + SPDK_DEBUGLOG(vhost_blk, "Register virtqueues %d interrupt\n", >> vq_idx); >> + >> + vq = &vsession->virtqueue[vq_idx]; >> + SPDK_DEBUGLOG(vhost_blk, "Register vq[%d]'s kickfd is %d\n", >> + vq_idx, vq->vring.kickfd); >> + >> + vq->intr = SPDK_INTERRUPT_REGISTER(vq->vring.kickfd, fn, vq); >> + if (vq->intr == NULL) { >> + SPDK_ERRLOG("Fail to register req notifier handler.\n"); >> + goto err; >> } >> >> return 0; >> >> err: >> - vhost_blk_session_unregister_interrupts(bvsession); >> - >> return -1; >> } >> >> static void >> -vhost_blk_poller_set_interrupt_mode(struct spdk_poller *poller, void *cb_arg, >> bool interrupt_mode) >> +vhost_blk_poller_set_vq_interrupt_mode(struct spdk_poller *poller, void >> *cb_arg, bool interrupt_mode) >> { >> - struct spdk_vhost_blk_session *bvsession = cb_arg; >> + struct spdk_vhost_virtqueue *vq = cb_arg; >> >> - vhost_session_set_interrupt_mode(&bvsession->vsession, >> interrupt_mode); >> + vhost_session_set_vq_interrupt_mode(vq, interrupt_mode); >> } >> >> static struct spdk_vhost_blk_dev * >> @@ -1127,35 +1099,44 @@ vhost_dev_bdev_remove_cpl_cb(struct >> spdk_vhost_dev *vdev, void *ctx) >> bvdev->bdev = NULL; >> } >> >> -static int >> -vhost_session_bdev_remove_cb(struct spdk_vhost_dev *vdev, >> +static int vq_bdev_remove_cb(struct spdk_vhost_dev *vdev, >> struct spdk_vhost_session *vsession, >> void *ctx) >> { >> - struct spdk_vhost_blk_session *bvsession; >> + struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> + unsigned long vq_idx = (unsigned long)ctx; >> int rc; >> >> - bvsession = to_blk_session(vsession); >> - if (bvsession->requestq_poller) { >> - spdk_poller_unregister(&bvsession->requestq_poller); >> - if (vsession->virtqueue[0].intr) { >> - vhost_blk_session_unregister_interrupts(bvsession); >> - rc = vhost_blk_session_register_interrupts(bvsession, >> no_bdev_vdev_vq_worker, >> - "no_bdev_vdev_vq_worker"); >> - if (rc) { >> - SPDK_ERRLOG("%s: Interrupt register failed\n", >> vsession->name); >> - return rc; >> - } >> - } >> + if (bvsession->requestq_poller[vq_idx]) { >> + spdk_poller_unregister(&bvsession->requestq_poller[vq_idx]); >> + } >> + >> + vhost_blk_session_unregister_vq_interrupts(bvsession, vq_idx); >> >> - bvsession->requestq_poller = >> SPDK_POLLER_REGISTER(no_bdev_vdev_worker, bvsession, 0); >> - spdk_poller_register_interrupt(bvsession->requestq_poller, >> vhost_blk_poller_set_interrupt_mode, >> - bvsession); >> + if (spdk_interrupt_mode_is_enabled()) { >> + rc = vhost_blk_session_register_vq_interrupts(bvsession, >> no_bdev_vdev_vq_worker, vq_idx); >> + if (rc) { >> + SPDK_ERRLOG("%s: Interrupt register failed\n", vsession- >>> name); >> + return rc; >> + } >> } >> >> + bvsession->requestq_poller[vq_idx] = >> SPDK_POLLER_REGISTER(no_bdev_vdev_vq_worker, bvsession, 0); >> + spdk_poller_register_interrupt(bvsession->requestq_poller[vq_idx], >> vhost_blk_poller_set_vq_interrupt_mode, >> + bvsession); >> + >> return 0; >> } >> >> +static int >> +vhost_session_bdev_remove_cb(struct spdk_vhost_dev *vdev, >> + struct spdk_vhost_session *vsession, >> + void *ctx) >> +{ >> + return vhost_session_send_event_mt(vsession, vq_bdev_remove_cb, >> + 3, "remove bdev"); >> +} >> + >> static void >> bdev_remove_cb(void *remove_ctx) >> { >> @@ -1194,156 +1175,143 @@ bdev_event_cb(enum spdk_bdev_event_type >> type, struct spdk_bdev *bdev, >> } >> >> static void >> -free_task_pool(struct spdk_vhost_blk_session *bvsession) >> +free_vq_task_pool(struct spdk_vhost_virtqueue *vq) >> { >> - struct spdk_vhost_session *vsession = &bvsession->vsession; >> - struct spdk_vhost_virtqueue *vq; >> - uint16_t i; >> - >> - for (i = 0; i < vsession->max_queues; i++) { >> - vq = &vsession->virtqueue[i]; >> - if (vq->tasks == NULL) { >> - continue; >> - } >> - >> - spdk_free(vq->tasks); >> - vq->tasks = NULL; >> + if (vq->tasks == NULL) { >> + return; >> } >> + >> + spdk_free(vq->tasks); >> + vq->tasks = NULL; >> + return; >> } >> >> static int >> -alloc_task_pool(struct spdk_vhost_blk_session *bvsession) >> +alloc_vq_task_pool(struct spdk_vhost_virtqueue *vq) >> { >> - struct spdk_vhost_session *vsession = &bvsession->vsession; >> - struct spdk_vhost_virtqueue *vq; >> + struct spdk_vhost_session *vsession = vq->vsession; >> + struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> struct spdk_vhost_blk_task *task; >> uint32_t task_cnt; >> - uint16_t i; >> uint32_t j; >> >> - for (i = 0; i < vsession->max_queues; i++) { >> - vq = &vsession->virtqueue[i]; >> - if (vq->vring.desc == NULL) { >> - continue; >> - } >> + if (vq->vring.desc == NULL) { >> + return 0; >> + } >> >> - task_cnt = vq->vring.size; >> - if (task_cnt > SPDK_VHOST_MAX_VQ_SIZE) { >> - /* sanity check */ >> - SPDK_ERRLOG("%s: virtuque %"PRIu16" is too big. (size >> = %"PRIu32", max = %"PRIu32")\n", >> - vsession->name, i, task_cnt, >> SPDK_VHOST_MAX_VQ_SIZE); >> - free_task_pool(bvsession); >> - return -1; >> - } >> - vq->tasks = spdk_zmalloc(sizeof(struct spdk_vhost_blk_task) * >> task_cnt, >> - SPDK_CACHE_LINE_SIZE, NULL, >> - SPDK_ENV_LCORE_ID_ANY, >> SPDK_MALLOC_DMA); >> - if (vq->tasks == NULL) { >> - SPDK_ERRLOG("%s: failed to allocate %"PRIu32" tasks for >> virtqueue %"PRIu16"\n", >> - vsession->name, task_cnt, i); >> - free_task_pool(bvsession); >> - return -1; >> - } >> + task_cnt = vq->vring.size; >> + if (task_cnt > SPDK_VHOST_MAX_VQ_SIZE) { >> + /* sanity check */ >> + SPDK_ERRLOG("%s: virtuque %"PRIu16" is too big. (size >> = %"PRIu32", max = %"PRIu32")\n", >> + vsession->name, vq->vring_idx, task_cnt, >> SPDK_VHOST_MAX_VQ_SIZE); >> + return -1; >> + } >> + vq->tasks = spdk_zmalloc(sizeof(struct spdk_vhost_blk_task) * task_cnt, >> + SPDK_CACHE_LINE_SIZE, NULL, >> + SPDK_ENV_LCORE_ID_ANY, >> SPDK_MALLOC_DMA); >> + if (vq->tasks == NULL) { >> + SPDK_ERRLOG("%s: failed to allocate %"PRIu32" tasks for >> virtqueue %"PRIu16"\n", >> + vsession->name, task_cnt, vq->vring_idx); >> + return -1; >> + } >> >> - for (j = 0; j < task_cnt; j++) { >> - task = &((struct spdk_vhost_blk_task *)vq->tasks)[j]; >> - task->bvsession = bvsession; >> - task->req_idx = j; >> - task->vq = vq; >> - } >> + for (j = 0; j < task_cnt; j++) { >> + task = &((struct spdk_vhost_blk_task *)vq->tasks)[j]; >> + task->bvsession = bvsession; >> + task->req_idx = j; >> + task->vq = vq; >> } >> >> return 0; >> } >> >> static int >> -vhost_blk_start_cb(struct spdk_vhost_dev *vdev, >> +vhost_blk_start_vq_cb(struct spdk_vhost_dev *vdev, >> struct spdk_vhost_session *vsession, void *unused) >> { >> struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> struct spdk_vhost_blk_dev *bvdev; >> - int i, rc = 0; >> + int rc = 0; >> + unsigned int vq_idx = (unsigned long)unused; >> + struct spdk_vhost_virtqueue *vq = &vsession->virtqueue[vq_idx]; >> >> bvdev = to_blk_dev(vdev); >> assert(bvdev != NULL); >> bvsession->bvdev = bvdev; >> >> - /* validate all I/O queues are in a contiguous index range */ >> - for (i = 0; i < vsession->max_queues; i++) { >> - /* vring.desc and vring.desc_packed are in a union struct >> - * so q->vring.desc can replace q->vring.desc_packed. >> - */ >> - if (vsession->virtqueue[i].vring.desc == NULL) { >> - SPDK_ERRLOG("%s: queue %"PRIu32" is empty\n", >> vsession->name, i); >> - rc = -1; >> - goto out; >> - } >> + assert(vq->vring_idx == vq_idx); >> + >> + /* vring.desc and vring.desc_packed are in a union struct >> + * so q->vring.desc can replace q->vring.desc_packed. >> + */ >> + if (vsession->virtqueue[vq_idx].vring.desc == NULL) { >> + SPDK_ERRLOG("%s: queue %"PRIu32" is empty\n", vsession- >>> name, vq_idx); >> + rc = -1; >> + goto out; >> } >> >> - rc = alloc_task_pool(bvsession); >> + rc = alloc_vq_task_pool(vq); >> if (rc != 0) { >> - SPDK_ERRLOG("%s: failed to alloc task pool.\n", vsession->name); >> + SPDK_ERRLOG("%s: failed to alloc %u task pool.\n", vsession- >>> name, vq_idx); >> goto out; >> } >> >> if (bvdev->bdev) { >> - bvsession->io_channel = spdk_bdev_get_io_channel(bvdev- >>> bdev_desc); >> - if (!bvsession->io_channel) { >> - free_task_pool(bvsession); >> - SPDK_ERRLOG("%s: I/O channel allocation failed\n", >> vsession->name); >> + bvsession->io_channel[vq_idx] = >> spdk_bdev_get_io_channel(bvdev->bdev_desc); >> + if (!bvsession->io_channel[vq_idx]) { >> + free_vq_task_pool(vq); >> + SPDK_ERRLOG("%s: I/O channel %u allocation failed\n", >> vsession->name, vq_idx); >> rc = -1; >> goto out; >> } >> } >> >> if (spdk_interrupt_mode_is_enabled()) { >> - if (bvdev->bdev) { >> - rc = vhost_blk_session_register_interrupts(bvsession, >> - vdev_vq_worker, >> - "vdev_vq_worker"); >> - } else { >> - rc = vhost_blk_session_register_interrupts(bvsession, >> - no_bdev_vdev_vq_worker, >> - "no_bdev_vdev_vq_worker"); >> - } >> - >> + rc = vhost_blk_session_register_vq_interrupts(bvsession, >> + bvdev->bdev ? vdev_vq_worker : >> no_bdev_vdev_vq_worker, vq_idx); >> if (rc) { >> - SPDK_ERRLOG("%s: Interrupt register failed\n", vsession- >>> name); >> + SPDK_ERRLOG("%s: Interrupt %u register failed\n", >> vsession->name, vq->vring_idx); >> goto out; >> } >> } >> >> - if (bvdev->bdev) { >> - bvsession->requestq_poller = >> SPDK_POLLER_REGISTER(vdev_worker, bvsession, 0); >> - } else { >> - bvsession->requestq_poller = >> SPDK_POLLER_REGISTER(no_bdev_vdev_worker, bvsession, 0); >> - } >> - SPDK_INFOLOG(vhost, "%s: started poller on lcore %d\n", >> - vsession->name, spdk_env_get_current_core()); >> + bvsession->requestq_poller[vq_idx] = SPDK_POLLER_REGISTER(bvdev- >>> bdev ? vdev_vq_worker : no_bdev_vdev_vq_worker, >> + vq, 0); >> + SPDK_INFOLOG(vhost, "%s.%u: started poller on lcore %d\n", >> + vsession->name, vq_idx, spdk_env_get_current_core()); >> + >> + spdk_poller_register_interrupt(bvsession->requestq_poller[vq_idx], >> vhost_blk_poller_set_vq_interrupt_mode, >> + vq); >> >> - spdk_poller_register_interrupt(bvsession->requestq_poller, >> vhost_blk_poller_set_interrupt_mode, >> - bvsession); >> >> out: >> - vhost_session_start_done(vsession, rc); >> + vhost_session_start_vq_done(vq, rc); >> return rc; >> } >> >> static int >> vhost_blk_start(struct spdk_vhost_session *vsession) >> { >> - return vhost_session_send_event(vsession, vhost_blk_start_cb, >> + int rc; >> + >> + rc = vhost_session_send_event_mt(vsession, vhost_blk_start_vq_cb, >> 3, "start session"); >> + if (rc) { >> + vhost_session_send_event_mt(vsession, vhost_blk_stop_vq_cb, >> + 3, "stop session"); >> + } >> + >> + return rc; >> } >> >> static int >> -destroy_session_poller_cb(void *arg) >> +destroy_session_vq_poller_cb(void *arg) >> { >> - struct spdk_vhost_blk_session *bvsession = arg; >> - struct spdk_vhost_session *vsession = &bvsession->vsession; >> - int i; >> + struct spdk_vhost_virtqueue *vq = arg; >> + struct spdk_vhost_session *vsession = vq->vsession; >> + struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> >> - if (vsession->task_cnt > 0) { >> + if (vq->task_cnt > 0) { >> return SPDK_POLLER_BUSY; >> } >> >> @@ -1351,48 +1319,46 @@ destroy_session_poller_cb(void *arg) >> return SPDK_POLLER_BUSY; >> } >> >> - for (i = 0; i < vsession->max_queues; i++) { >> - vsession->virtqueue[i].next_event_time = 0; >> - vhost_vq_used_signal(vsession, &vsession->virtqueue[i]); >> - } >> + vq->next_event_time = 0; >> + vhost_vq_used_signal(vsession, vq); >> >> SPDK_INFOLOG(vhost, "%s: stopping poller on lcore %d\n", >> vsession->name, spdk_env_get_current_core()); >> >> - if (bvsession->io_channel) { >> - spdk_put_io_channel(bvsession->io_channel); >> - bvsession->io_channel = NULL; >> + if (bvsession->io_channel[vq->vring_idx]) { >> + spdk_put_io_channel(bvsession->io_channel[vq->vring_idx]); >> + bvsession->io_channel[vq->vring_idx] = NULL; >> } >> >> - free_task_pool(bvsession); >> - spdk_poller_unregister(&bvsession->stop_poller); >> - vhost_session_stop_done(vsession, 0); >> + free_vq_task_pool(vq); >> + spdk_poller_unregister(&bvsession->stop_poller[vq->vring_idx]); >> + vhost_session_stop_vq_done(vq, 0); >> >> spdk_vhost_unlock(); >> return SPDK_POLLER_BUSY; >> } >> >> static int >> -vhost_blk_stop_cb(struct spdk_vhost_dev *vdev, >> +vhost_blk_stop_vq_cb(struct spdk_vhost_dev *vdev, >> struct spdk_vhost_session *vsession, void *unused) >> { >> struct spdk_vhost_blk_session *bvsession = to_blk_session(vsession); >> + unsigned long vq_idx = (unsigned long)unused; >> + struct spdk_vhost_virtqueue *vq = &vsession->virtqueue[vq_idx]; >> >> - spdk_poller_unregister(&bvsession->requestq_poller); >> - >> - if (vsession->virtqueue[0].intr) { >> - vhost_blk_session_unregister_interrupts(bvsession); >> + if (bvsession->requestq_poller[vq_idx]) { >> + spdk_poller_unregister(&bvsession->requestq_poller[vq_idx]); >> } >> >> - bvsession->stop_poller = >> SPDK_POLLER_REGISTER(destroy_session_poller_cb, >> - bvsession, 1000); >> + vhost_blk_session_unregister_vq_interrupts(bvsession, vq_idx); >> + bvsession->stop_poller[vq_idx] = >> SPDK_POLLER_REGISTER(destroy_session_vq_poller_cb, vq, 1000); >> return 0; >> } >> >> static int >> vhost_blk_stop(struct spdk_vhost_session *vsession) >> { >> - return vhost_session_send_event(vsession, vhost_blk_stop_cb, >> + return vhost_session_send_event_mt(vsession, vhost_blk_stop_vq_cb, >> 3, "stop session"); >> } >> >> diff --git a/lib/vhost/vhost_internal.h b/lib/vhost/vhost_internal.h >> index 36ab0c16f..92d096a8c 100644 >> --- a/lib/vhost/vhost_internal.h >> +++ b/lib/vhost/vhost_internal.h >> @@ -111,6 +111,14 @@ struct spdk_vhost_virtqueue { >> >> void *tasks; >> >> + int task_cnt; >> + >> + bool initialized; >> + bool started; >> + bool needs_restart; >> + bool forced_polling; >> + bool interrupt_mode; >> + >> /* Request count from last stats check */ >> uint32_t req_cnt; >> >> @@ -150,10 +158,12 @@ struct spdk_vhost_session { >> >> struct rte_vhost_memory *mem; >> >> - int task_cnt; >> - >> uint16_t max_queues; >> >> + uint16_t active_queues; >> + >> + int task_cnt; >> + >> uint64_t negotiated_features; >> >> /* Local copy of device coalescing settings. */ >> @@ -168,6 +178,8 @@ struct spdk_vhost_session { >> >> struct spdk_vhost_virtqueue virtqueue[SPDK_VHOST_MAX_VQUEUES]; >> >> + struct spdk_thread *thread[SPDK_VHOST_MAX_VQUEUES]; /* thread of >> data plane per vq */ >> + >> TAILQ_ENTRY(spdk_vhost_session) tailq; >> }; >> >> @@ -175,7 +187,7 @@ struct spdk_vhost_dev { >> char *name; >> char *path; >> >> - struct spdk_thread *thread; >> + struct spdk_thread *thread; /* if support mt, only run as control plane */ >> bool registered; >> >> uint64_t virtio_features; >> @@ -420,6 +432,7 @@ int vhost_destroy_connection_cb(int vid); >> * Set vhost session to run in interrupt or poll mode >> */ >> void vhost_session_set_interrupt_mode(struct spdk_vhost_session *vsession, >> bool interrupt_mode); >> +void vhost_session_set_vq_interrupt_mode(struct spdk_vhost_virtqueue *vq, >> bool interrupt_mode); >> >> /* >> * Memory registration functions used in start/stop device callbacks >> @@ -464,6 +477,10 @@ int vhost_session_send_event(struct >> spdk_vhost_session *vsession, >> spdk_vhost_session_fn cb_fn, unsigned timeout_sec, >> const char *errmsg); >> >> +int vhost_session_send_event_mt(struct spdk_vhost_session *vsession, >> + spdk_vhost_session_fn cb_fn, unsigned timeout_sec, >> + const char *errmsg); >> + >> /** >> * Finish a blocking spdk_vhost_session_send_event() call and finally >> * start the session. This must be called on the target lcore, which >> @@ -477,6 +494,8 @@ int vhost_session_send_event(struct spdk_vhost_session >> *vsession, >> */ >> void vhost_session_start_done(struct spdk_vhost_session *vsession, int >> response); >> >> +void vhost_session_start_vq_done(struct spdk_vhost_virtqueue *vq, int >> response); >> + >> /** >> * Finish a blocking spdk_vhost_session_send_event() call and finally >> * stop the session. This must be called on the session's lcore which >> @@ -493,6 +512,9 @@ void vhost_session_start_done(struct >> spdk_vhost_session *vsession, int response) >> */ >> void vhost_session_stop_done(struct spdk_vhost_session *vsession, int >> response); >> >> +void vhost_session_stop_vq_done(struct spdk_vhost_virtqueue *vq, int >> response); >> + >> + >> struct spdk_vhost_session *vhost_session_find_by_vid(int vid); >> void vhost_session_install_rte_compat_hooks(struct spdk_vhost_session >> *vsession); >> int vhost_register_unix_socket(const char *path, const char *ctrl_name, >> -- >> 2.14.1.40.g8e62ba1 >> _______________________________________________ >> SPDK mailing list -- spdk(a)lists.01.org >> To unsubscribe send an email to spdk-leave(a)lists.01.org > _______________________________________________ > SPDK mailing list -- spdk(a)lists.01.org > To unsubscribe send an email to spdk-leave(a)lists.01.org