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
);
130 qio_channel_yield(ioc
, G_IO_IN
);
133 error_report_err(local_err
);
139 if (vmsg
->fd_num
+ nfds
> max_fds
) {
140 error_report("A maximum of %zu fds are allowed, "
141 "however got %zu fds now",
142 max_fds
, vmsg
->fd_num
+ nfds
);
146 memcpy(vmsg
->fds
+ vmsg
->fd_num
, fds
, nfds
* sizeof(vmsg
->fds
[0]));
147 vmsg
->fd_num
+= nfds
;
151 if (rc
== 0) { /* socket closed */
158 } while (read_bytes
!= VHOST_USER_HDR_SIZE
);
160 /* qio_channel_readv_full will make socket fds blocking, unblock them */
161 vmsg_unblock_fds(vmsg
);
162 if (vmsg
->size
> sizeof(vmsg
->payload
)) {
163 error_report("Error: too big message request: %d, "
164 "size: vmsg->size: %u, "
165 "while sizeof(vmsg->payload) = %zu",
166 vmsg
->request
, vmsg
->size
, sizeof(vmsg
->payload
));
170 struct iovec iov_payload
= {
171 .iov_base
= (char *)&vmsg
->payload
,
172 .iov_len
= vmsg
->size
,
175 rc
= qio_channel_readv_all_eof(ioc
, &iov_payload
, 1, &local_err
);
178 error_report_err(local_err
);
187 vmsg_close_fds(vmsg
);
192 static coroutine_fn
void vu_client_trip(void *opaque
)
194 VuServer
*server
= opaque
;
195 VuDev
*vu_dev
= &server
->vu_dev
;
197 while (!vu_dev
->broken
&& vu_dispatch(vu_dev
)) {
201 if (vhost_user_server_has_in_flight(server
)) {
202 /* Wait for requests to complete before we can unmap the memory */
203 server
->wait_idle
= true;
204 qemu_coroutine_yield();
205 server
->wait_idle
= false;
207 assert(!vhost_user_server_has_in_flight(server
));
211 /* vu_deinit() should have called remove_watch() */
212 assert(QTAILQ_EMPTY(&server
->vu_fd_watches
));
214 object_unref(OBJECT(server
->sioc
));
217 object_unref(OBJECT(server
->ioc
));
220 server
->co_trip
= NULL
;
221 if (server
->restart_listener_bh
) {
222 qemu_bh_schedule(server
->restart_listener_bh
);
228 * a wrapper for vu_kick_cb
230 * since aio_dispatch can only pass one user data pointer to the
231 * callback function, pack VuDev and pvt into a struct. Then unpack it
232 * and pass them to vu_kick_cb
234 static void kick_handler(void *opaque
)
236 VuFdWatch
*vu_fd_watch
= opaque
;
237 VuDev
*vu_dev
= vu_fd_watch
->vu_dev
;
239 vu_fd_watch
->cb(vu_dev
, 0, vu_fd_watch
->pvt
);
241 /* Stop vu_client_trip() if an error occurred in vu_fd_watch->cb() */
242 if (vu_dev
->broken
) {
243 VuServer
*server
= container_of(vu_dev
, VuServer
, vu_dev
);
245 qio_channel_shutdown(server
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
249 static VuFdWatch
*find_vu_fd_watch(VuServer
*server
, int fd
)
252 VuFdWatch
*vu_fd_watch
, *next
;
253 QTAILQ_FOREACH_SAFE(vu_fd_watch
, &server
->vu_fd_watches
, next
, next
) {
254 if (vu_fd_watch
->fd
== fd
) {
262 set_watch(VuDev
*vu_dev
, int fd
, int vu_evt
,
263 vu_watch_cb cb
, void *pvt
)
266 VuServer
*server
= container_of(vu_dev
, VuServer
, vu_dev
);
271 VuFdWatch
*vu_fd_watch
= find_vu_fd_watch(server
, fd
);
274 VuFdWatch
*vu_fd_watch
= g_new0(VuFdWatch
, 1);
276 QTAILQ_INSERT_TAIL(&server
->vu_fd_watches
, vu_fd_watch
, next
);
278 vu_fd_watch
->fd
= fd
;
279 vu_fd_watch
->cb
= cb
;
280 qemu_socket_set_nonblock(fd
);
281 aio_set_fd_handler(server
->ioc
->ctx
, fd
, kick_handler
,
282 NULL
, NULL
, NULL
, vu_fd_watch
);
283 vu_fd_watch
->vu_dev
= vu_dev
;
284 vu_fd_watch
->pvt
= pvt
;
289 static void remove_watch(VuDev
*vu_dev
, int fd
)
295 server
= container_of(vu_dev
, VuServer
, vu_dev
);
297 VuFdWatch
*vu_fd_watch
= find_vu_fd_watch(server
, fd
);
302 aio_set_fd_handler(server
->ioc
->ctx
, fd
, NULL
, NULL
, NULL
, NULL
, NULL
);
304 QTAILQ_REMOVE(&server
->vu_fd_watches
, vu_fd_watch
, next
);
309 static void vu_accept(QIONetListener
*listener
, QIOChannelSocket
*sioc
,
312 VuServer
*server
= opaque
;
315 warn_report("Only one vhost-user client is allowed to "
316 "connect the server one time");
320 if (!vu_init(&server
->vu_dev
, server
->max_queues
, sioc
->fd
, panic_cb
,
321 vu_message_read
, set_watch
, remove_watch
, server
->vu_iface
)) {
322 error_report("Failed to initialize libvhost-user");
327 * Unset the callback function for network listener to make another
328 * vhost-user client keeping waiting until this client disconnects
330 qio_net_listener_set_client_func(server
->listener
,
336 * Increase the object reference, so sioc will not freed by
337 * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc))
339 object_ref(OBJECT(server
->sioc
));
340 qio_channel_set_name(QIO_CHANNEL(sioc
), "vhost-user client");
341 server
->ioc
= QIO_CHANNEL(sioc
);
342 object_ref(OBJECT(server
->ioc
));
344 /* TODO vu_message_write() spins if non-blocking! */
345 qio_channel_set_blocking(server
->ioc
, false, NULL
);
347 server
->co_trip
= qemu_coroutine_create(vu_client_trip
, server
);
349 aio_context_acquire(server
->ctx
);
350 vhost_user_server_attach_aio_context(server
, server
->ctx
);
351 aio_context_release(server
->ctx
);
354 /* server->ctx acquired by caller */
355 void vhost_user_server_stop(VuServer
*server
)
357 qemu_bh_delete(server
->restart_listener_bh
);
358 server
->restart_listener_bh
= NULL
;
361 VuFdWatch
*vu_fd_watch
;
363 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
364 aio_set_fd_handler(server
->ctx
, vu_fd_watch
->fd
,
365 NULL
, NULL
, NULL
, NULL
, vu_fd_watch
);
368 qio_channel_shutdown(server
->ioc
, QIO_CHANNEL_SHUTDOWN_BOTH
, NULL
);
370 AIO_WAIT_WHILE(server
->ctx
, server
->co_trip
);
373 if (server
->listener
) {
374 qio_net_listener_disconnect(server
->listener
);
375 object_unref(OBJECT(server
->listener
));
380 * Allow the next client to connect to the server. Called from a BH in the main
383 static void restart_listener_bh(void *opaque
)
385 VuServer
*server
= opaque
;
387 qio_net_listener_set_client_func(server
->listener
, vu_accept
, server
,
391 /* Called with ctx acquired */
392 void vhost_user_server_attach_aio_context(VuServer
*server
, AioContext
*ctx
)
394 VuFdWatch
*vu_fd_watch
;
402 qio_channel_attach_aio_context(server
->ioc
, ctx
);
404 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
405 aio_set_fd_handler(ctx
, vu_fd_watch
->fd
, kick_handler
, NULL
,
406 NULL
, NULL
, vu_fd_watch
);
409 aio_co_schedule(ctx
, server
->co_trip
);
412 /* Called with server->ctx acquired */
413 void vhost_user_server_detach_aio_context(VuServer
*server
)
416 VuFdWatch
*vu_fd_watch
;
418 QTAILQ_FOREACH(vu_fd_watch
, &server
->vu_fd_watches
, next
) {
419 aio_set_fd_handler(server
->ctx
, vu_fd_watch
->fd
,
420 NULL
, NULL
, NULL
, NULL
, vu_fd_watch
);
423 qio_channel_detach_aio_context(server
->ioc
);
429 bool vhost_user_server_start(VuServer
*server
,
430 SocketAddress
*socket_addr
,
433 const VuDevIface
*vu_iface
,
437 QIONetListener
*listener
;
439 if (socket_addr
->type
!= SOCKET_ADDRESS_TYPE_UNIX
&&
440 socket_addr
->type
!= SOCKET_ADDRESS_TYPE_FD
) {
441 error_setg(errp
, "Only socket address types 'unix' and 'fd' are supported");
445 listener
= qio_net_listener_new();
446 if (qio_net_listener_open_sync(listener
, socket_addr
, 1,
448 object_unref(OBJECT(listener
));
452 bh
= qemu_bh_new(restart_listener_bh
, server
);
454 /* zero out unspecified fields */
455 *server
= (VuServer
) {
456 .listener
= listener
,
457 .restart_listener_bh
= bh
,
458 .vu_iface
= vu_iface
,
459 .max_queues
= max_queues
,
463 qio_net_listener_set_name(server
->listener
, "vhost-user-backend-listener");
465 qio_net_listener_set_client_func(server
->listener
,
470 QTAILQ_INIT(&server
->vu_fd_watches
);