]>
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 | ||
520d8b40 KW |
78 | void vhost_user_server_ref(VuServer *server) |
79 | { | |
80 | assert(!server->wait_idle); | |
81 | server->refcount++; | |
82 | } | |
83 | ||
84 | void vhost_user_server_unref(VuServer *server) | |
85 | { | |
86 | server->refcount--; | |
87 | if (server->wait_idle && !server->refcount) { | |
88 | aio_co_wake(server->co_trip); | |
89 | } | |
90 | } | |
91 | ||
70eb2c07 CX |
92 | static bool coroutine_fn |
93 | vu_message_read(VuDev *vu_dev, int conn_fd, VhostUserMsg *vmsg) | |
94 | { | |
95 | struct iovec iov = { | |
96 | .iov_base = (char *)vmsg, | |
97 | .iov_len = VHOST_USER_HDR_SIZE, | |
98 | }; | |
99 | int rc, read_bytes = 0; | |
100 | Error *local_err = NULL; | |
70eb2c07 | 101 | const size_t max_fds = G_N_ELEMENTS(vmsg->fds); |
70eb2c07 CX |
102 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); |
103 | QIOChannel *ioc = server->ioc; | |
104 | ||
8c7f7cbc | 105 | vmsg->fd_num = 0; |
70eb2c07 CX |
106 | if (!ioc) { |
107 | error_report_err(local_err); | |
108 | goto fail; | |
109 | } | |
110 | ||
111 | assert(qemu_in_coroutine()); | |
112 | do { | |
8c7f7cbc SH |
113 | size_t nfds = 0; |
114 | int *fds = NULL; | |
115 | ||
70eb2c07 CX |
116 | /* |
117 | * qio_channel_readv_full may have short reads, keeping calling it | |
118 | * until getting VHOST_USER_HDR_SIZE or 0 bytes in total | |
119 | */ | |
84615a19 | 120 | rc = qio_channel_readv_full(ioc, &iov, 1, &fds, &nfds, 0, &local_err); |
70eb2c07 CX |
121 | if (rc < 0) { |
122 | if (rc == QIO_CHANNEL_ERR_BLOCK) { | |
8c7f7cbc | 123 | assert(local_err == NULL); |
70eb2c07 CX |
124 | qio_channel_yield(ioc, G_IO_IN); |
125 | continue; | |
126 | } else { | |
127 | error_report_err(local_err); | |
8c7f7cbc | 128 | goto fail; |
70eb2c07 CX |
129 | } |
130 | } | |
8c7f7cbc SH |
131 | |
132 | if (nfds > 0) { | |
133 | if (vmsg->fd_num + nfds > max_fds) { | |
70eb2c07 CX |
134 | error_report("A maximum of %zu fds are allowed, " |
135 | "however got %zu fds now", | |
8c7f7cbc SH |
136 | max_fds, vmsg->fd_num + nfds); |
137 | g_free(fds); | |
70eb2c07 CX |
138 | goto fail; |
139 | } | |
8c7f7cbc SH |
140 | memcpy(vmsg->fds + vmsg->fd_num, fds, nfds * sizeof(vmsg->fds[0])); |
141 | vmsg->fd_num += nfds; | |
142 | g_free(fds); | |
70eb2c07 | 143 | } |
8c7f7cbc SH |
144 | |
145 | if (rc == 0) { /* socket closed */ | |
146 | goto fail; | |
70eb2c07 | 147 | } |
70eb2c07 | 148 | |
8c7f7cbc SH |
149 | iov.iov_base += rc; |
150 | iov.iov_len -= rc; | |
151 | read_bytes += rc; | |
152 | } while (read_bytes != VHOST_USER_HDR_SIZE); | |
153 | ||
70eb2c07 CX |
154 | /* qio_channel_readv_full will make socket fds blocking, unblock them */ |
155 | vmsg_unblock_fds(vmsg); | |
156 | if (vmsg->size > sizeof(vmsg->payload)) { | |
157 | error_report("Error: too big message request: %d, " | |
158 | "size: vmsg->size: %u, " | |
159 | "while sizeof(vmsg->payload) = %zu", | |
160 | vmsg->request, vmsg->size, sizeof(vmsg->payload)); | |
161 | goto fail; | |
162 | } | |
163 | ||
164 | struct iovec iov_payload = { | |
165 | .iov_base = (char *)&vmsg->payload, | |
166 | .iov_len = vmsg->size, | |
167 | }; | |
168 | if (vmsg->size) { | |
169 | rc = qio_channel_readv_all_eof(ioc, &iov_payload, 1, &local_err); | |
edaf6205 SH |
170 | if (rc != 1) { |
171 | if (local_err) { | |
172 | error_report_err(local_err); | |
173 | } | |
70eb2c07 CX |
174 | goto fail; |
175 | } | |
176 | } | |
177 | ||
178 | return true; | |
179 | ||
180 | fail: | |
181 | vmsg_close_fds(vmsg); | |
182 | ||
183 | return false; | |
184 | } | |
185 | ||
70eb2c07 CX |
186 | static coroutine_fn void vu_client_trip(void *opaque) |
187 | { | |
188 | VuServer *server = opaque; | |
7185c857 | 189 | VuDev *vu_dev = &server->vu_dev; |
70eb2c07 | 190 | |
7185c857 SH |
191 | while (!vu_dev->broken && vu_dispatch(vu_dev)) { |
192 | /* Keep running */ | |
70eb2c07 CX |
193 | } |
194 | ||
520d8b40 KW |
195 | if (server->refcount) { |
196 | /* Wait for requests to complete before we can unmap the memory */ | |
197 | server->wait_idle = true; | |
198 | qemu_coroutine_yield(); | |
199 | server->wait_idle = false; | |
200 | } | |
201 | assert(server->refcount == 0); | |
202 | ||
7185c857 | 203 | vu_deinit(vu_dev); |
70eb2c07 | 204 | |
7185c857 SH |
205 | /* vu_deinit() should have called remove_watch() */ |
206 | assert(QTAILQ_EMPTY(&server->vu_fd_watches)); | |
207 | ||
208 | object_unref(OBJECT(server->sioc)); | |
209 | server->sioc = NULL; | |
210 | ||
211 | object_unref(OBJECT(server->ioc)); | |
212 | server->ioc = NULL; | |
213 | ||
214 | server->co_trip = NULL; | |
215 | if (server->restart_listener_bh) { | |
216 | qemu_bh_schedule(server->restart_listener_bh); | |
217 | } | |
218 | aio_wait_kick(); | |
70eb2c07 CX |
219 | } |
220 | ||
221 | /* | |
222 | * a wrapper for vu_kick_cb | |
223 | * | |
224 | * since aio_dispatch can only pass one user data pointer to the | |
225 | * callback function, pack VuDev and pvt into a struct. Then unpack it | |
226 | * and pass them to vu_kick_cb | |
227 | */ | |
228 | static void kick_handler(void *opaque) | |
229 | { | |
230 | VuFdWatch *vu_fd_watch = opaque; | |
7185c857 | 231 | VuDev *vu_dev = vu_fd_watch->vu_dev; |
70eb2c07 | 232 | |
7185c857 SH |
233 | vu_fd_watch->cb(vu_dev, 0, vu_fd_watch->pvt); |
234 | ||
235 | /* Stop vu_client_trip() if an error occurred in vu_fd_watch->cb() */ | |
236 | if (vu_dev->broken) { | |
237 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); | |
238 | ||
239 | qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); | |
240 | } | |
241 | } | |
70eb2c07 CX |
242 | |
243 | static VuFdWatch *find_vu_fd_watch(VuServer *server, int fd) | |
244 | { | |
245 | ||
246 | VuFdWatch *vu_fd_watch, *next; | |
247 | QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) { | |
248 | if (vu_fd_watch->fd == fd) { | |
249 | return vu_fd_watch; | |
250 | } | |
251 | } | |
252 | return NULL; | |
253 | } | |
254 | ||
255 | static void | |
256 | set_watch(VuDev *vu_dev, int fd, int vu_evt, | |
257 | vu_watch_cb cb, void *pvt) | |
258 | { | |
259 | ||
260 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); | |
261 | g_assert(vu_dev); | |
262 | g_assert(fd >= 0); | |
263 | g_assert(cb); | |
264 | ||
265 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
266 | ||
267 | if (!vu_fd_watch) { | |
268 | VuFdWatch *vu_fd_watch = g_new0(VuFdWatch, 1); | |
269 | ||
270 | QTAILQ_INSERT_TAIL(&server->vu_fd_watches, vu_fd_watch, next); | |
271 | ||
272 | vu_fd_watch->fd = fd; | |
273 | vu_fd_watch->cb = cb; | |
ff5927ba | 274 | qemu_socket_set_nonblock(fd); |
70eb2c07 | 275 | aio_set_fd_handler(server->ioc->ctx, fd, true, kick_handler, |
826cc324 | 276 | NULL, NULL, NULL, vu_fd_watch); |
70eb2c07 CX |
277 | vu_fd_watch->vu_dev = vu_dev; |
278 | vu_fd_watch->pvt = pvt; | |
279 | } | |
280 | } | |
281 | ||
282 | ||
283 | static void remove_watch(VuDev *vu_dev, int fd) | |
284 | { | |
285 | VuServer *server; | |
286 | g_assert(vu_dev); | |
287 | g_assert(fd >= 0); | |
288 | ||
289 | server = container_of(vu_dev, VuServer, vu_dev); | |
290 | ||
291 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
292 | ||
293 | if (!vu_fd_watch) { | |
294 | return; | |
295 | } | |
826cc324 SH |
296 | aio_set_fd_handler(server->ioc->ctx, fd, true, |
297 | NULL, NULL, NULL, NULL, NULL); | |
70eb2c07 CX |
298 | |
299 | QTAILQ_REMOVE(&server->vu_fd_watches, vu_fd_watch, next); | |
300 | g_free(vu_fd_watch); | |
301 | } | |
302 | ||
303 | ||
304 | static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc, | |
305 | gpointer opaque) | |
306 | { | |
307 | VuServer *server = opaque; | |
308 | ||
309 | if (server->sioc) { | |
310 | warn_report("Only one vhost-user client is allowed to " | |
311 | "connect the server one time"); | |
312 | return; | |
313 | } | |
314 | ||
315 | if (!vu_init(&server->vu_dev, server->max_queues, sioc->fd, panic_cb, | |
316 | vu_message_read, set_watch, remove_watch, server->vu_iface)) { | |
317 | error_report("Failed to initialize libvhost-user"); | |
318 | return; | |
319 | } | |
320 | ||
321 | /* | |
322 | * Unset the callback function for network listener to make another | |
323 | * vhost-user client keeping waiting until this client disconnects | |
324 | */ | |
325 | qio_net_listener_set_client_func(server->listener, | |
326 | NULL, | |
327 | NULL, | |
328 | NULL); | |
329 | server->sioc = sioc; | |
330 | /* | |
331 | * Increase the object reference, so sioc will not freed by | |
332 | * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc)) | |
333 | */ | |
334 | object_ref(OBJECT(server->sioc)); | |
335 | qio_channel_set_name(QIO_CHANNEL(sioc), "vhost-user client"); | |
336 | server->ioc = QIO_CHANNEL(sioc); | |
337 | object_ref(OBJECT(server->ioc)); | |
7185c857 SH |
338 | |
339 | /* TODO vu_message_write() spins if non-blocking! */ | |
46a096c8 | 340 | qio_channel_set_blocking(server->ioc, false, NULL); |
70eb2c07 | 341 | |
7185c857 SH |
342 | server->co_trip = qemu_coroutine_create(vu_client_trip, server); |
343 | ||
344 | aio_context_acquire(server->ctx); | |
345 | vhost_user_server_attach_aio_context(server, server->ctx); | |
346 | aio_context_release(server->ctx); | |
347 | } | |
70eb2c07 CX |
348 | |
349 | void vhost_user_server_stop(VuServer *server) | |
350 | { | |
7185c857 SH |
351 | aio_context_acquire(server->ctx); |
352 | ||
353 | qemu_bh_delete(server->restart_listener_bh); | |
354 | server->restart_listener_bh = NULL; | |
355 | ||
70eb2c07 | 356 | if (server->sioc) { |
7185c857 SH |
357 | VuFdWatch *vu_fd_watch; |
358 | ||
359 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { | |
360 | aio_set_fd_handler(server->ctx, vu_fd_watch->fd, true, | |
826cc324 | 361 | NULL, NULL, NULL, NULL, vu_fd_watch); |
7185c857 SH |
362 | } |
363 | ||
364 | qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); | |
365 | ||
366 | AIO_WAIT_WHILE(server->ctx, server->co_trip); | |
70eb2c07 CX |
367 | } |
368 | ||
7185c857 SH |
369 | aio_context_release(server->ctx); |
370 | ||
70eb2c07 CX |
371 | if (server->listener) { |
372 | qio_net_listener_disconnect(server->listener); | |
373 | object_unref(OBJECT(server->listener)); | |
374 | } | |
7185c857 SH |
375 | } |
376 | ||
377 | /* | |
378 | * Allow the next client to connect to the server. Called from a BH in the main | |
379 | * loop. | |
380 | */ | |
381 | static void restart_listener_bh(void *opaque) | |
382 | { | |
383 | VuServer *server = opaque; | |
70eb2c07 | 384 | |
7185c857 SH |
385 | qio_net_listener_set_client_func(server->listener, vu_accept, server, |
386 | NULL); | |
70eb2c07 CX |
387 | } |
388 | ||
7185c857 SH |
389 | /* Called with ctx acquired */ |
390 | void vhost_user_server_attach_aio_context(VuServer *server, AioContext *ctx) | |
70eb2c07 | 391 | { |
7185c857 | 392 | VuFdWatch *vu_fd_watch; |
70eb2c07 | 393 | |
7185c857 | 394 | server->ctx = ctx; |
70eb2c07 CX |
395 | |
396 | if (!server->sioc) { | |
70eb2c07 CX |
397 | return; |
398 | } | |
399 | ||
7185c857 SH |
400 | qio_channel_attach_aio_context(server->ioc, ctx); |
401 | ||
402 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { | |
403 | aio_set_fd_handler(ctx, vu_fd_watch->fd, true, kick_handler, NULL, | |
826cc324 | 404 | NULL, NULL, vu_fd_watch); |
70eb2c07 CX |
405 | } |
406 | ||
7185c857 SH |
407 | aio_co_schedule(ctx, server->co_trip); |
408 | } | |
409 | ||
410 | /* Called with server->ctx acquired */ | |
411 | void vhost_user_server_detach_aio_context(VuServer *server) | |
412 | { | |
413 | if (server->sioc) { | |
414 | VuFdWatch *vu_fd_watch; | |
415 | ||
416 | QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) { | |
417 | aio_set_fd_handler(server->ctx, vu_fd_watch->fd, true, | |
826cc324 | 418 | NULL, NULL, NULL, NULL, vu_fd_watch); |
70eb2c07 | 419 | } |
7185c857 SH |
420 | |
421 | qio_channel_detach_aio_context(server->ioc); | |
70eb2c07 | 422 | } |
70eb2c07 | 423 | |
7185c857 SH |
424 | server->ctx = NULL; |
425 | } | |
70eb2c07 CX |
426 | |
427 | bool vhost_user_server_start(VuServer *server, | |
428 | SocketAddress *socket_addr, | |
429 | AioContext *ctx, | |
430 | uint16_t max_queues, | |
70eb2c07 CX |
431 | const VuDevIface *vu_iface, |
432 | Error **errp) | |
433 | { | |
7185c857 | 434 | QEMUBH *bh; |
90fc91d5 SH |
435 | QIONetListener *listener; |
436 | ||
437 | if (socket_addr->type != SOCKET_ADDRESS_TYPE_UNIX && | |
438 | socket_addr->type != SOCKET_ADDRESS_TYPE_FD) { | |
439 | error_setg(errp, "Only socket address types 'unix' and 'fd' are supported"); | |
440 | return false; | |
441 | } | |
442 | ||
443 | listener = qio_net_listener_new(); | |
70eb2c07 CX |
444 | if (qio_net_listener_open_sync(listener, socket_addr, 1, |
445 | errp) < 0) { | |
446 | object_unref(OBJECT(listener)); | |
447 | return false; | |
448 | } | |
449 | ||
7185c857 SH |
450 | bh = qemu_bh_new(restart_listener_bh, server); |
451 | ||
1d787456 | 452 | /* zero out unspecified fields */ |
70eb2c07 CX |
453 | *server = (VuServer) { |
454 | .listener = listener, | |
7185c857 | 455 | .restart_listener_bh = bh, |
70eb2c07 CX |
456 | .vu_iface = vu_iface, |
457 | .max_queues = max_queues, | |
458 | .ctx = ctx, | |
70eb2c07 CX |
459 | }; |
460 | ||
461 | qio_net_listener_set_name(server->listener, "vhost-user-backend-listener"); | |
462 | ||
463 | qio_net_listener_set_client_func(server->listener, | |
464 | vu_accept, | |
465 | server, | |
466 | NULL); | |
467 | ||
468 | QTAILQ_INIT(&server->vu_fd_watches); | |
469 | return true; | |
470 | } |