2 * Sharing QEMU devices via vhost-user protocol
4 * Copyright (c) Coiby Xu <coiby.xu@gmail.com>.
5 * Copyright (c) 2020 Red Hat, Inc.
7 * This work is licensed under the terms of the GNU GPL, version 2 or
8 * later. See the COPYING file in the top-level directory.
10 #include "qemu/osdep.h"
11 #include "qemu/error-report.h"
12 #include "qemu/main-loop.h"
13 #include "qemu/vhost-user-server.h"
14 #include "block/aio-wait.h"
17 * Theory of operation:
19 * VuServer is started and stopped by vhost_user_server_start() and
20 * vhost_user_server_stop() from the main loop thread. Starting the server
21 * opens a vhost-user UNIX domain socket and listens for incoming connections.
22 * Only one connection is allowed at a time.
24 * The connection is handled by the vu_client_trip() coroutine in the
25 * VuServer->ctx AioContext. The coroutine consists of a vu_dispatch() loop
26 * where libvhost-user calls vu_message_read() to receive the next vhost-user
27 * protocol messages over the UNIX domain socket.
29 * When virtqueues are set up libvhost-user calls set_watch() to monitor kick
30 * fds. These fds are also handled in the VuServer->ctx AioContext.
32 * Both vu_client_trip() and kick fd monitoring can be stopped by shutting down
33 * the socket connection. Shutting down the socket connection causes
34 * vu_message_read() to fail since no more data can be received from the socket.
35 * After vu_dispatch() fails, vu_client_trip() calls vu_deinit() to stop
36 * libvhost-user before terminating the coroutine. vu_deinit() calls
37 * remove_watch() to stop monitoring kick fds and this stops virtqueue
40 * When vu_client_trip() has finished cleaning up it schedules a BH in the main
41 * loop thread to accept the next client connection.
43 * When libvhost-user detects an error it calls panic_cb() and sets the
44 * dev->broken flag. Both vu_client_trip() and kick fd processing stop when
45 * the dev->broken flag is set.
47 * It is possible to switch AioContexts using
48 * vhost_user_server_detach_aio_context() and
49 * vhost_user_server_attach_aio_context(). They stop monitoring fds in the old
50 * AioContext and resume monitoring in the new AioContext. The vu_client_trip()
51 * coroutine remains in a yielded state during the switch. This is made
52 * possible by QIOChannel's support for spurious coroutine re-entry in
53 * qio_channel_yield(). The coroutine will restart I/O when re-entered from the
57 static void vmsg_close_fds(VhostUserMsg
*vmsg
)
60 for (i
= 0; i
< vmsg
->fd_num
; i
++) {
65 static void vmsg_unblock_fds(VhostUserMsg
*vmsg
)
68 for (i
= 0; i
< vmsg
->fd_num
; i
++) {
69 qemu_socket_set_nonblock(vmsg
->fds
[i
]);
73 static void panic_cb(VuDev
*vu_dev
, const char *buf
)
75 error_report("vu_panic: %s", buf
);
78 void vhost_user_server_inc_in_flight(VuServer
*server
)
80 assert(!server
->wait_idle
);
81 qatomic_inc(&server
->in_flight
);
84 void vhost_user_server_dec_in_flight(VuServer
*server
)
86 if (qatomic_fetch_dec(&server
->in_flight
) == 1) {
87 if (server
->wait_idle
) {
88 aio_co_wake(server
->co_trip
);
93 bool vhost_user_server_has_in_flight(VuServer
*server
)
95 return qatomic_load_acquire(&server
->in_flight
) > 0;
98 static bool coroutine_fn
99 vu_message_read(VuDev
*vu_dev
, int conn_fd
, VhostUserMsg
*vmsg
)
102 .iov_base
= (char *)vmsg
,
103 .iov_len
= VHOST_USER_HDR_SIZE
,
105 int rc
, read_bytes
= 0;
106 Error
*local_err
= NULL
;
107 const size_t max_fds
= G_N_ELEMENTS(vmsg
->fds
);
108 VuServer
*server
= container_of(vu_dev
, VuServer
, vu_dev
);
109 QIOChannel
*ioc
= server
->ioc
;
113 error_report_err(local_err
);
117 assert(qemu_in_coroutine());
123 * qio_channel_readv_full may have short reads, keeping calling it
124 * until getting VHOST_USER_HDR_SIZE or 0 bytes in total
126 rc
= qio_channel_readv_full(ioc
, &iov
, 1, &fds
, &nfds
, 0, &local_err
);
128 if (rc
== QIO_CHANNEL_ERR_BLOCK
) {
129 assert(local_err
== NULL
);
131 server
->in_qio_channel_yield
= true;
132 qio_channel_yield(ioc
, G_IO_IN
);
133 server
->in_qio_channel_yield
= false;
135 /* Wait until attached to an AioContext again */
136 qemu_coroutine_yield();
140 error_report_err(local_err
);
146 if (vmsg
->fd_num
+ nfds
> max_fds
) {
147 error_report("A maximum of %zu fds are allowed, "
148 "however got %zu fds now",
149 max_fds
, vmsg
->fd_num
+ nfds
);
153 memcpy(vmsg
->fds
+ vmsg
->fd_num
, fds
, nfds
* sizeof(vmsg
->fds
[0]));
154 vmsg
->fd_num
+= nfds
;
158 if (rc
== 0) { /* socket closed */
165 } while (read_bytes
!= VHOST_USER_HDR_SIZE
);
167 /* qio_channel_readv_full will make socket fds blocking, unblock them */
168 vmsg_unblock_fds(vmsg
);
169 if (vmsg
->size
> sizeof(vmsg
->payload
)) {
170 error_report("Error: too big message request: %d, "
171 "size: vmsg->size: %u, "
172 "while sizeof(vmsg->payload) = %zu",
173 vmsg
->request
, vmsg
->size
, sizeof(vmsg
->payload
));
177 struct iovec iov_payload
= {
178 .iov_base
= (char *)&vmsg
->payload
,
179 .iov_len
= vmsg
->size
,
182 rc
= qio_channel_readv_all_eof(ioc
, &iov_payload
, 1, &local_err
);
185 error_report_err(local_err
);
194 vmsg_close_fds(vmsg
);
199 static coroutine_fn
void vu_client_trip(void *opaque
)
201 VuServer
*server
= opaque
;
202 VuDev
*vu_dev
= &server
->vu_dev
;
204 while (!vu_dev
->broken
&& vu_dispatch(vu_dev
)) {
208 if (vhost_user_server_has_in_flight(server
)) {
209 /* Wait for requests to complete before we can unmap the memory */
210 server
->wait_idle
= true;
211 qemu_coroutine_yield();
212 server
->wait_idle
= false;
214 assert(!vhost_user_server_has_in_flight(server
));
218 /* vu_deinit() should have called remove_watch() */
219 assert(QTAILQ_EMPTY(&server
->vu_fd_watches
));
221 object_unref(OBJECT(server
->sioc
));
224 object_unref(OBJECT(server
->ioc
));
227 server
->co_trip
= NULL
;
228 if (server
->restart_listener_bh
) {
229 qemu_bh_schedule(server
->restart_listener_bh
);
235 * a wrapper for vu_kick_cb
237 * since aio_dispatch can only pass one user data pointer to the
238 * callback function, pack VuDev and pvt into a struct. Then unpack it
239 * and pass them to vu_kick_cb
241 static void kick_handler(void *opaque
)
243 VuFdWatch
*vu_fd_watch
= opaque
;
244 VuDev
*vu_dev
= vu_fd_watch
->vu_dev
;
246 vu_fd_watch
->cb(vu_dev
, 0, vu_fd_watch
->pvt
);
248 /* Stop vu_client_trip() if an error occurred in vu_fd_watch->cb() */
249 if (vu_dev
->broken
) {
250 VuServer
*server
= container_of(vu_dev
, VuServer
, vu_dev
);
252 qio_channel_shutdown(server
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
256 static VuFdWatch
*find_vu_fd_watch(VuServer
*server
, int fd
)
259 VuFdWatch
*vu_fd_watch
, *next
;
260 QTAILQ_FOREACH_SAFE(vu_fd_watch
, &server
->vu_fd_watches
, next
, next
) {
261 if (vu_fd_watch
->fd
== fd
) {
269 set_watch(VuDev
*vu_dev
, int fd
, int vu_evt
,
270 vu_watch_cb cb
, void *pvt
)
273 VuServer
*server
= container_of(vu_dev
, VuServer
, vu_dev
);
278 VuFdWatch
*vu_fd_watch
= find_vu_fd_watch(server
, fd
);
281 vu_fd_watch
= g_new0(VuFdWatch
, 1);
283 QTAILQ_INSERT_TAIL(&server
->vu_fd_watches
, vu_fd_watch
, next
);
285 vu_fd_watch
->fd
= fd
;
286 vu_fd_watch
->cb
= cb
;
287 qemu_socket_set_nonblock(fd
);
288 aio_set_fd_handler(server
->ctx
, fd
, kick_handler
,
289 NULL
, NULL
, NULL
, vu_fd_watch
);
290 vu_fd_watch
->vu_dev
= vu_dev
;
291 vu_fd_watch
->pvt
= pvt
;
296 static void remove_watch(VuDev
*vu_dev
, int fd
)
302 server
= container_of(vu_dev
, VuServer
, vu_dev
);
304 VuFdWatch
*vu_fd_watch
= find_vu_fd_watch(server
, fd
);
309 aio_set_fd_handler(server
->ctx
, fd
, NULL
, NULL
, NULL
, NULL
, NULL
);
311 QTAILQ_REMOVE(&server
->vu_fd_watches
, vu_fd_watch
, next
);
316 static void vu_accept(QIONetListener
*listener
, QIOChannelSocket
*sioc
,
319 VuServer
*server
= opaque
;
322 warn_report("Only one vhost-user client is allowed to "
323 "connect the server one time");
327 if (!vu_init(&server
->vu_dev
, server
->max_queues
, sioc
->fd
, panic_cb
,
328 vu_message_read
, set_watch
, remove_watch
, server
->vu_iface
)) {
329 error_report("Failed to initialize libvhost-user");
334 * Unset the callback function for network listener to make another
335 * vhost-user client keeping waiting until this client disconnects
337 qio_net_listener_set_client_func(server
->listener
,
343 * Increase the object reference, so sioc will not freed by
344 * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc))
346 object_ref(OBJECT(server
->sioc
));
347 qio_channel_set_name(QIO_CHANNEL(sioc
), "vhost-user client");
348 server
->ioc
= QIO_CHANNEL(sioc
);
349 object_ref(OBJECT(server
->ioc
));
351 /* TODO vu_message_write() spins if non-blocking! */
352 qio_channel_set_blocking(server
->ioc
, false, NULL
);
354 qio_channel_set_follow_coroutine_ctx(server
->ioc
, true);
356 server
->co_trip
= qemu_coroutine_create(vu_client_trip
, server
);
358 aio_context_acquire(server
->ctx
);
359 vhost_user_server_attach_aio_context(server
, server
->ctx
);
360 aio_context_release(server
->ctx
);
363 /* server->ctx acquired by caller */
364 void vhost_user_server_stop(VuServer
*server
)
366 qemu_bh_delete(server
->restart_listener_bh
);
367 server
->restart_listener_bh
= NULL
;
370 VuFdWatch
*vu_fd_watch
;
372 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
373 aio_set_fd_handler(server
->ctx
, vu_fd_watch
->fd
,
374 NULL
, NULL
, NULL
, NULL
, vu_fd_watch
);
377 qio_channel_shutdown(server
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
379 AIO_WAIT_WHILE(server
->ctx
, server
->co_trip
);
382 if (server
->listener
) {
383 qio_net_listener_disconnect(server
->listener
);
384 object_unref(OBJECT(server
->listener
));
389 * Allow the next client to connect to the server. Called from a BH in the main
392 static void restart_listener_bh(void *opaque
)
394 VuServer
*server
= opaque
;
396 qio_net_listener_set_client_func(server
->listener
, vu_accept
, server
,
400 /* Called with ctx acquired */
401 void vhost_user_server_attach_aio_context(VuServer
*server
, AioContext
*ctx
)
403 VuFdWatch
*vu_fd_watch
;
411 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
412 aio_set_fd_handler(ctx
, vu_fd_watch
->fd
, kick_handler
, NULL
,
413 NULL
, NULL
, vu_fd_watch
);
416 assert(!server
->in_qio_channel_yield
);
417 aio_co_schedule(ctx
, server
->co_trip
);
420 /* Called with server->ctx acquired */
421 void vhost_user_server_detach_aio_context(VuServer
*server
)
424 VuFdWatch
*vu_fd_watch
;
426 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
427 aio_set_fd_handler(server
->ctx
, vu_fd_watch
->fd
,
428 NULL
, NULL
, NULL
, NULL
, vu_fd_watch
);
435 if (server
->in_qio_channel_yield
) {
436 /* Stop receiving the next vhost-user message */
437 qio_channel_wake_read(server
->ioc
);
442 bool vhost_user_server_start(VuServer
*server
,
443 SocketAddress
*socket_addr
,
446 const VuDevIface
*vu_iface
,
450 QIONetListener
*listener
;
452 if (socket_addr
->type
!= SOCKET_ADDRESS_TYPE_UNIX
&&
453 socket_addr
->type
!= SOCKET_ADDRESS_TYPE_FD
) {
454 error_setg(errp
, "Only socket address types 'unix' and 'fd' are supported");
458 listener
= qio_net_listener_new();
459 if (qio_net_listener_open_sync(listener
, socket_addr
, 1,
461 object_unref(OBJECT(listener
));
465 bh
= qemu_bh_new(restart_listener_bh
, server
);
467 /* zero out unspecified fields */
468 *server
= (VuServer
) {
469 .listener
= listener
,
470 .restart_listener_bh
= bh
,
471 .vu_iface
= vu_iface
,
472 .max_queues
= max_queues
,
476 qio_net_listener_set_name(server
->listener
, "vhost-user-backend-listener");
478 qio_net_listener_set_client_func(server
->listener
,
483 QTAILQ_INIT(&server
->vu_fd_watches
);