]>
Commit | Line | Data |
---|---|---|
70eb2c07 CX |
1 | /* |
2 | * Sharing QEMU devices via vhost-user protocol | |
3 | * | |
4 | * Copyright (c) Coiby Xu <coiby.xu@gmail.com>. | |
5 | * Copyright (c) 2020 Red Hat, Inc. | |
6 | * | |
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. | |
9 | */ | |
10 | #include "qemu/osdep.h" | |
5feed38c | 11 | #include "qemu/error-report.h" |
70eb2c07 | 12 | #include "qemu/main-loop.h" |
80a06cc5 | 13 | #include "qemu/vhost-user-server.h" |
7185c857 | 14 | #include "block/aio-wait.h" |
70eb2c07 | 15 | |
7185c857 SH |
16 | /* |
17 | * Theory of operation: | |
18 | * | |
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. | |
23 | * | |
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. | |
28 | * | |
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. | |
31 | * | |
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 | |
38 | * processing. | |
39 | * | |
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. | |
42 | * | |
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. | |
46 | * | |
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 | |
54 | * new AioContext. | |
55 | */ | |
56 | ||
70eb2c07 CX |
57 | static void vmsg_close_fds(VhostUserMsg *vmsg) |
58 | { | |
59 | int i; | |
60 | for (i = 0; i < vmsg->fd_num; i++) { | |
61 | close(vmsg->fds[i]); | |
62 | } | |
63 | } | |
64 | ||
65 | static void vmsg_unblock_fds(VhostUserMsg *vmsg) | |
66 | { | |
67 | int i; | |
68 | for (i = 0; i < vmsg->fd_num; i++) { | |
ff5927ba | 69 | qemu_socket_set_nonblock(vmsg->fds[i]); |
70eb2c07 CX |
70 | } |
71 | } | |
72 | ||
70eb2c07 CX |
73 | static void panic_cb(VuDev *vu_dev, const char *buf) |
74 | { | |
7185c857 | 75 | error_report("vu_panic: %s", buf); |
70eb2c07 CX |
76 | } |
77 | ||
75d33e85 | 78 | void vhost_user_server_inc_in_flight(VuServer *server) |
520d8b40 KW |
79 | { |
80 | assert(!server->wait_idle); | |
8f5e9a8e | 81 | qatomic_inc(&server->in_flight); |
520d8b40 KW |
82 | } |
83 | ||
75d33e85 | 84 | void vhost_user_server_dec_in_flight(VuServer *server) |
520d8b40 | 85 | { |
8f5e9a8e SH |
86 | if (qatomic_fetch_dec(&server->in_flight) == 1) { |
87 | if (server->wait_idle) { | |
88 | aio_co_wake(server->co_trip); | |
89 | } | |
520d8b40 KW |
90 | } |
91 | } | |
92 | ||
8f5e9a8e SH |
93 | bool vhost_user_server_has_in_flight(VuServer *server) |
94 | { | |
95 | return qatomic_load_acquire(&server->in_flight) > 0; | |
96 | } | |
97 | ||
70eb2c07 CX |
98 | static bool coroutine_fn |
99 | vu_message_read(VuDev *vu_dev, int conn_fd, VhostUserMsg *vmsg) | |
100 | { | |
101 | struct iovec iov = { | |
102 | .iov_base = (char *)vmsg, | |
103 | .iov_len = VHOST_USER_HDR_SIZE, | |
104 | }; | |
105 | int rc, read_bytes = 0; | |
106 | Error *local_err = NULL; | |
70eb2c07 | 107 | const size_t max_fds = G_N_ELEMENTS(vmsg->fds); |
70eb2c07 CX |
108 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); |
109 | QIOChannel *ioc = server->ioc; | |
110 | ||
8c7f7cbc | 111 | vmsg->fd_num = 0; |
70eb2c07 CX |
112 | if (!ioc) { |
113 | error_report_err(local_err); | |
114 | goto fail; | |
115 | } | |
116 | ||
117 | assert(qemu_in_coroutine()); | |
118 | do { | |
8c7f7cbc SH |
119 | size_t nfds = 0; |
120 | int *fds = NULL; | |
121 | ||
70eb2c07 CX |
122 | /* |
123 | * qio_channel_readv_full may have short reads, keeping calling it | |
124 | * until getting VHOST_USER_HDR_SIZE or 0 bytes in total | |
125 | */ | |
84615a19 | 126 | rc = qio_channel_readv_full(ioc, &iov, 1, &fds, &nfds, 0, &local_err); |
70eb2c07 CX |
127 | if (rc < 0) { |
128 | if (rc == QIO_CHANNEL_ERR_BLOCK) { | |
8c7f7cbc | 129 | assert(local_err == NULL); |
06e0f098 SH |
130 | if (server->ctx) { |
131 | server->in_qio_channel_yield = true; | |
132 | qio_channel_yield(ioc, G_IO_IN); | |
133 | server->in_qio_channel_yield = false; | |
134 | } else { | |
135 | /* Wait until attached to an AioContext again */ | |
136 | qemu_coroutine_yield(); | |
137 | } | |
70eb2c07 CX |
138 | continue; |
139 | } else { | |
140 | error_report_err(local_err); | |
8c7f7cbc | 141 | goto fail; |
70eb2c07 CX |
142 | } |
143 | } | |
8c7f7cbc SH |
144 | |
145 | if (nfds > 0) { | |
146 | if (vmsg->fd_num + nfds > max_fds) { | |
70eb2c07 CX |
147 | error_report("A maximum of %zu fds are allowed, " |
148 | "however got %zu fds now", | |
8c7f7cbc SH |
149 | max_fds, vmsg->fd_num + nfds); |
150 | g_free(fds); | |
70eb2c07 CX |
151 | goto fail; |
152 | } | |
8c7f7cbc SH |
153 | memcpy(vmsg->fds + vmsg->fd_num, fds, nfds * sizeof(vmsg->fds[0])); |
154 | vmsg->fd_num += nfds; | |
155 | g_free(fds); | |
70eb2c07 | 156 | } |
8c7f7cbc SH |
157 | |
158 | if (rc == 0) { /* socket closed */ | |
159 | goto fail; | |
70eb2c07 | 160 | } |
70eb2c07 | 161 | |
8c7f7cbc SH |
162 | iov.iov_base += rc; |
163 | iov.iov_len -= rc; | |
164 | read_bytes += rc; | |
165 | } while (read_bytes != VHOST_USER_HDR_SIZE); | |
166 | ||
70eb2c07 CX |
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)); | |
174 | goto fail; | |
175 | } | |
176 | ||
177 | struct iovec iov_payload = { | |
178 | .iov_base = (char *)&vmsg->payload, | |
179 | .iov_len = vmsg->size, | |
180 | }; | |
181 | if (vmsg->size) { | |
182 | rc = qio_channel_readv_all_eof(ioc, &iov_payload, 1, &local_err); | |
edaf6205 SH |
183 | if (rc != 1) { |
184 | if (local_err) { | |
185 | error_report_err(local_err); | |
186 | } | |
70eb2c07 CX |
187 | goto fail; |
188 | } | |
189 | } | |
190 | ||
191 | return true; | |
192 | ||
193 | fail: | |
194 | vmsg_close_fds(vmsg); | |
195 | ||
196 | return false; | |
197 | } | |
198 | ||
70eb2c07 CX |
199 | static coroutine_fn void vu_client_trip(void *opaque) |
200 | { | |
201 | VuServer *server = opaque; | |
7185c857 | 202 | VuDev *vu_dev = &server->vu_dev; |
70eb2c07 | 203 | |
7185c857 SH |
204 | while (!vu_dev->broken && vu_dispatch(vu_dev)) { |
205 | /* Keep running */ | |
70eb2c07 CX |
206 | } |
207 | ||
8f5e9a8e | 208 | if (vhost_user_server_has_in_flight(server)) { |
520d8b40 KW |
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; | |
213 | } | |
8f5e9a8e | 214 | assert(!vhost_user_server_has_in_flight(server)); |
520d8b40 | 215 | |
7185c857 | 216 | vu_deinit(vu_dev); |
70eb2c07 | 217 | |
7185c857 SH |
218 | /* vu_deinit() should have called remove_watch() */ |
219 | assert(QTAILQ_EMPTY(&server->vu_fd_watches)); | |
220 | ||
221 | object_unref(OBJECT(server->sioc)); | |
222 | server->sioc = NULL; | |
223 | ||
224 | object_unref(OBJECT(server->ioc)); | |
225 | server->ioc = NULL; | |
226 | ||
227 | server->co_trip = NULL; | |
228 | if (server->restart_listener_bh) { | |
229 | qemu_bh_schedule(server->restart_listener_bh); | |
230 | } | |
231 | aio_wait_kick(); | |
70eb2c07 CX |
232 | } |
233 | ||
234 | /* | |
235 | * a wrapper for vu_kick_cb | |
236 | * | |
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 | |
240 | */ | |
241 | static void kick_handler(void *opaque) | |
242 | { | |
243 | VuFdWatch *vu_fd_watch = opaque; | |
7185c857 | 244 | VuDev *vu_dev = vu_fd_watch->vu_dev; |
70eb2c07 | 245 | |
7185c857 SH |
246 | vu_fd_watch->cb(vu_dev, 0, vu_fd_watch->pvt); |
247 | ||
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); | |
251 | ||
252 | qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); | |
253 | } | |
254 | } | |
70eb2c07 CX |
255 | |
256 | static VuFdWatch *find_vu_fd_watch(VuServer *server, int fd) | |
257 | { | |
258 | ||
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) { | |
262 | return vu_fd_watch; | |
263 | } | |
264 | } | |
265 | return NULL; | |
266 | } | |
267 | ||
268 | static void | |
269 | set_watch(VuDev *vu_dev, int fd, int vu_evt, | |
270 | vu_watch_cb cb, void *pvt) | |
271 | { | |
272 | ||
273 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); | |
274 | g_assert(vu_dev); | |
275 | g_assert(fd >= 0); | |
276 | g_assert(cb); | |
277 | ||
278 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
279 | ||
280 | if (!vu_fd_watch) { | |
fbf58f21 | 281 | vu_fd_watch = g_new0(VuFdWatch, 1); |
70eb2c07 CX |
282 | |
283 | QTAILQ_INSERT_TAIL(&server->vu_fd_watches, vu_fd_watch, next); | |
284 | ||
285 | vu_fd_watch->fd = fd; | |
286 | vu_fd_watch->cb = cb; | |
ff5927ba | 287 | qemu_socket_set_nonblock(fd); |
06e0f098 | 288 | aio_set_fd_handler(server->ctx, fd, kick_handler, |
826cc324 | 289 | NULL, NULL, NULL, vu_fd_watch); |
70eb2c07 CX |
290 | vu_fd_watch->vu_dev = vu_dev; |
291 | vu_fd_watch->pvt = pvt; | |
292 | } | |
293 | } | |
294 | ||
295 | ||
296 | static void remove_watch(VuDev *vu_dev, int fd) | |
297 | { | |
298 | VuServer *server; | |
299 | g_assert(vu_dev); | |
300 | g_assert(fd >= 0); | |
301 | ||
302 | server = container_of(vu_dev, VuServer, vu_dev); | |
303 | ||
304 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
305 | ||
306 | if (!vu_fd_watch) { | |
307 | return; | |
308 | } | |
06e0f098 | 309 | aio_set_fd_handler(server->ctx, fd, NULL, NULL, NULL, NULL, NULL); |
70eb2c07 CX |
310 | |
311 | QTAILQ_REMOVE(&server->vu_fd_watches, vu_fd_watch, next); | |
312 | g_free(vu_fd_watch); | |
313 | } | |
314 | ||
315 | ||
316 | static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc, | |
317 | gpointer opaque) | |
318 | { | |
319 | VuServer *server = opaque; | |
320 | ||
321 | if (server->sioc) { | |
322 | warn_report("Only one vhost-user client is allowed to " | |
323 | "connect the server one time"); | |
324 | return; | |
325 | } | |
326 | ||
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"); | |
330 | return; | |
331 | } | |
332 | ||
333 | /* | |
334 | * Unset the callback function for network listener to make another | |
335 | * vhost-user client keeping waiting until this client disconnects | |
336 | */ | |
337 | qio_net_listener_set_client_func(server->listener, | |
338 | NULL, | |
339 | NULL, | |
340 | NULL); | |
341 | server->sioc = sioc; | |
342 | /* | |
343 | * Increase the object reference, so sioc will not freed by | |
344 | * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc)) | |
345 | */ | |
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)); | |
7185c857 SH |
350 | |
351 | /* TODO vu_message_write() spins if non-blocking! */ | |
46a096c8 | 352 | qio_channel_set_blocking(server->ioc, false, NULL); |
70eb2c07 | 353 | |
06e0f098 SH |
354 | qio_channel_set_follow_coroutine_ctx(server->ioc, true); |
355 | ||
7185c857 SH |
356 | server->co_trip = qemu_coroutine_create(vu_client_trip, server); |
357 | ||
358 | aio_context_acquire(server->ctx); | |
359 | vhost_user_server_attach_aio_context(server, server->ctx); | |
360 | aio_context_release(server->ctx); | |
361 | } | |
70eb2c07 | 362 | |
2957dc40 | 363 | /* server->ctx acquired by caller */ |
70eb2c07 CX |
364 | void vhost_user_server_stop(VuServer *server) |
365 | { | |
7185c857 SH |
366 | qemu_bh_delete(server->restart_listener_bh); |
367 | server->restart_listener_bh = NULL; | |
368 | ||
70eb2c07 | 369 | if (server->sioc) { |
7185c857 SH |
370 | VuFdWatch *vu_fd_watch; |
371 | ||
372 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { | |
60f782b6 | 373 | aio_set_fd_handler(server->ctx, vu_fd_watch->fd, |
826cc324 | 374 | NULL, NULL, NULL, NULL, vu_fd_watch); |
7185c857 SH |
375 | } |
376 | ||
377 | qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); | |
378 | ||
379 | AIO_WAIT_WHILE(server->ctx, server->co_trip); | |
70eb2c07 CX |
380 | } |
381 | ||
382 | if (server->listener) { | |
383 | qio_net_listener_disconnect(server->listener); | |
384 | object_unref(OBJECT(server->listener)); | |
385 | } | |
7185c857 SH |
386 | } |
387 | ||
388 | /* | |
389 | * Allow the next client to connect to the server. Called from a BH in the main | |
390 | * loop. | |
391 | */ | |
392 | static void restart_listener_bh(void *opaque) | |
393 | { | |
394 | VuServer *server = opaque; | |
70eb2c07 | 395 | |
7185c857 SH |
396 | qio_net_listener_set_client_func(server->listener, vu_accept, server, |
397 | NULL); | |
70eb2c07 CX |
398 | } |
399 | ||
7185c857 SH |
400 | /* Called with ctx acquired */ |
401 | void vhost_user_server_attach_aio_context(VuServer *server, AioContext *ctx) | |
70eb2c07 | 402 | { |
7185c857 | 403 | VuFdWatch *vu_fd_watch; |
70eb2c07 | 404 | |
7185c857 | 405 | server->ctx = ctx; |
70eb2c07 CX |
406 | |
407 | if (!server->sioc) { | |
70eb2c07 CX |
408 | return; |
409 | } | |
410 | ||
7185c857 | 411 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { |
60f782b6 | 412 | aio_set_fd_handler(ctx, vu_fd_watch->fd, kick_handler, NULL, |
826cc324 | 413 | NULL, NULL, vu_fd_watch); |
70eb2c07 CX |
414 | } |
415 | ||
06e0f098 | 416 | assert(!server->in_qio_channel_yield); |
7185c857 SH |
417 | aio_co_schedule(ctx, server->co_trip); |
418 | } | |
419 | ||
420 | /* Called with server->ctx acquired */ | |
421 | void vhost_user_server_detach_aio_context(VuServer *server) | |
422 | { | |
423 | if (server->sioc) { | |
424 | VuFdWatch *vu_fd_watch; | |
425 | ||
426 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { | |
60f782b6 | 427 | aio_set_fd_handler(server->ctx, vu_fd_watch->fd, |
826cc324 | 428 | NULL, NULL, NULL, NULL, vu_fd_watch); |
70eb2c07 CX |
429 | } |
430 | } | |
70eb2c07 | 431 | |
7185c857 | 432 | server->ctx = NULL; |
06e0f098 SH |
433 | |
434 | if (server->ioc) { | |
435 | if (server->in_qio_channel_yield) { | |
436 | /* Stop receiving the next vhost-user message */ | |
437 | qio_channel_wake_read(server->ioc); | |
438 | } | |
439 | } | |
7185c857 | 440 | } |
70eb2c07 CX |
441 | |
442 | bool vhost_user_server_start(VuServer *server, | |
443 | SocketAddress *socket_addr, | |
444 | AioContext *ctx, | |
445 | uint16_t max_queues, | |
70eb2c07 CX |
446 | const VuDevIface *vu_iface, |
447 | Error **errp) | |
448 | { | |
7185c857 | 449 | QEMUBH *bh; |
90fc91d5 SH |
450 | QIONetListener *listener; |
451 | ||
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"); | |
455 | return false; | |
456 | } | |
457 | ||
458 | listener = qio_net_listener_new(); | |
70eb2c07 CX |
459 | if (qio_net_listener_open_sync(listener, socket_addr, 1, |
460 | errp) < 0) { | |
461 | object_unref(OBJECT(listener)); | |
462 | return false; | |
463 | } | |
464 | ||
7185c857 SH |
465 | bh = qemu_bh_new(restart_listener_bh, server); |
466 | ||
1d787456 | 467 | /* zero out unspecified fields */ |
70eb2c07 CX |
468 | *server = (VuServer) { |
469 | .listener = listener, | |
7185c857 | 470 | .restart_listener_bh = bh, |
70eb2c07 CX |
471 | .vu_iface = vu_iface, |
472 | .max_queues = max_queues, | |
473 | .ctx = ctx, | |
70eb2c07 CX |
474 | }; |
475 | ||
476 | qio_net_listener_set_name(server->listener, "vhost-user-backend-listener"); | |
477 | ||
478 | qio_net_listener_set_client_func(server->listener, | |
479 | vu_accept, | |
480 | server, | |
481 | NULL); | |
482 | ||
483 | QTAILQ_INIT(&server->vu_fd_watches); | |
484 | return true; | |
485 | } |