]>
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" | |
11 | #include "qemu/main-loop.h" | |
12 | #include "vhost-user-server.h" | |
13 | ||
14 | static void vmsg_close_fds(VhostUserMsg *vmsg) | |
15 | { | |
16 | int i; | |
17 | for (i = 0; i < vmsg->fd_num; i++) { | |
18 | close(vmsg->fds[i]); | |
19 | } | |
20 | } | |
21 | ||
22 | static void vmsg_unblock_fds(VhostUserMsg *vmsg) | |
23 | { | |
24 | int i; | |
25 | for (i = 0; i < vmsg->fd_num; i++) { | |
26 | qemu_set_nonblock(vmsg->fds[i]); | |
27 | } | |
28 | } | |
29 | ||
30 | static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc, | |
31 | gpointer opaque); | |
32 | ||
33 | static void close_client(VuServer *server) | |
34 | { | |
35 | /* | |
36 | * Before closing the client | |
37 | * | |
38 | * 1. Let vu_client_trip stop processing new vhost-user msg | |
39 | * | |
40 | * 2. remove kick_handler | |
41 | * | |
42 | * 3. wait for the kick handler to be finished | |
43 | * | |
44 | * 4. wait for the current vhost-user msg to be finished processing | |
45 | */ | |
46 | ||
47 | QIOChannelSocket *sioc = server->sioc; | |
48 | /* When this is set vu_client_trip will stop new processing vhost-user message */ | |
49 | server->sioc = NULL; | |
50 | ||
70eb2c07 CX |
51 | while (server->processing_msg) { |
52 | if (server->ioc->read_coroutine) { | |
53 | server->ioc->read_coroutine = NULL; | |
54 | qio_channel_set_aio_fd_handler(server->ioc, server->ioc->ctx, NULL, | |
55 | NULL, server->ioc); | |
56 | server->processing_msg = false; | |
57 | } | |
58 | } | |
59 | ||
60 | vu_deinit(&server->vu_dev); | |
dad4f194 SH |
61 | |
62 | /* vu_deinit() should have called remove_watch() */ | |
63 | assert(QTAILQ_EMPTY(&server->vu_fd_watches)); | |
64 | ||
70eb2c07 CX |
65 | object_unref(OBJECT(sioc)); |
66 | object_unref(OBJECT(server->ioc)); | |
67 | } | |
68 | ||
69 | static void panic_cb(VuDev *vu_dev, const char *buf) | |
70 | { | |
71 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); | |
72 | ||
73 | /* avoid while loop in close_client */ | |
74 | server->processing_msg = false; | |
75 | ||
76 | if (buf) { | |
77 | error_report("vu_panic: %s", buf); | |
78 | } | |
79 | ||
80 | if (server->sioc) { | |
81 | close_client(server); | |
82 | } | |
83 | ||
70eb2c07 CX |
84 | /* |
85 | * Set the callback function for network listener so another | |
86 | * vhost-user client can connect to this server | |
87 | */ | |
88 | qio_net_listener_set_client_func(server->listener, | |
89 | vu_accept, | |
90 | server, | |
91 | NULL); | |
92 | } | |
93 | ||
94 | static bool coroutine_fn | |
95 | vu_message_read(VuDev *vu_dev, int conn_fd, VhostUserMsg *vmsg) | |
96 | { | |
97 | struct iovec iov = { | |
98 | .iov_base = (char *)vmsg, | |
99 | .iov_len = VHOST_USER_HDR_SIZE, | |
100 | }; | |
101 | int rc, read_bytes = 0; | |
102 | Error *local_err = NULL; | |
70eb2c07 | 103 | const size_t max_fds = G_N_ELEMENTS(vmsg->fds); |
70eb2c07 CX |
104 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); |
105 | QIOChannel *ioc = server->ioc; | |
106 | ||
8c7f7cbc | 107 | vmsg->fd_num = 0; |
70eb2c07 CX |
108 | if (!ioc) { |
109 | error_report_err(local_err); | |
110 | goto fail; | |
111 | } | |
112 | ||
113 | assert(qemu_in_coroutine()); | |
114 | do { | |
8c7f7cbc SH |
115 | size_t nfds = 0; |
116 | int *fds = NULL; | |
117 | ||
70eb2c07 CX |
118 | /* |
119 | * qio_channel_readv_full may have short reads, keeping calling it | |
120 | * until getting VHOST_USER_HDR_SIZE or 0 bytes in total | |
121 | */ | |
8c7f7cbc | 122 | rc = qio_channel_readv_full(ioc, &iov, 1, &fds, &nfds, &local_err); |
70eb2c07 CX |
123 | if (rc < 0) { |
124 | if (rc == QIO_CHANNEL_ERR_BLOCK) { | |
8c7f7cbc | 125 | assert(local_err == NULL); |
70eb2c07 CX |
126 | qio_channel_yield(ioc, G_IO_IN); |
127 | continue; | |
128 | } else { | |
129 | error_report_err(local_err); | |
8c7f7cbc | 130 | goto fail; |
70eb2c07 CX |
131 | } |
132 | } | |
8c7f7cbc SH |
133 | |
134 | if (nfds > 0) { | |
135 | if (vmsg->fd_num + nfds > max_fds) { | |
70eb2c07 CX |
136 | error_report("A maximum of %zu fds are allowed, " |
137 | "however got %zu fds now", | |
8c7f7cbc SH |
138 | max_fds, vmsg->fd_num + nfds); |
139 | g_free(fds); | |
70eb2c07 CX |
140 | goto fail; |
141 | } | |
8c7f7cbc SH |
142 | memcpy(vmsg->fds + vmsg->fd_num, fds, nfds * sizeof(vmsg->fds[0])); |
143 | vmsg->fd_num += nfds; | |
144 | g_free(fds); | |
70eb2c07 | 145 | } |
8c7f7cbc SH |
146 | |
147 | if (rc == 0) { /* socket closed */ | |
148 | goto fail; | |
70eb2c07 | 149 | } |
70eb2c07 | 150 | |
8c7f7cbc SH |
151 | iov.iov_base += rc; |
152 | iov.iov_len -= rc; | |
153 | read_bytes += rc; | |
154 | } while (read_bytes != VHOST_USER_HDR_SIZE); | |
155 | ||
70eb2c07 CX |
156 | /* qio_channel_readv_full will make socket fds blocking, unblock them */ |
157 | vmsg_unblock_fds(vmsg); | |
158 | if (vmsg->size > sizeof(vmsg->payload)) { | |
159 | error_report("Error: too big message request: %d, " | |
160 | "size: vmsg->size: %u, " | |
161 | "while sizeof(vmsg->payload) = %zu", | |
162 | vmsg->request, vmsg->size, sizeof(vmsg->payload)); | |
163 | goto fail; | |
164 | } | |
165 | ||
166 | struct iovec iov_payload = { | |
167 | .iov_base = (char *)&vmsg->payload, | |
168 | .iov_len = vmsg->size, | |
169 | }; | |
170 | if (vmsg->size) { | |
171 | rc = qio_channel_readv_all_eof(ioc, &iov_payload, 1, &local_err); | |
172 | if (rc == -1) { | |
173 | error_report_err(local_err); | |
174 | goto fail; | |
175 | } | |
176 | } | |
177 | ||
178 | return true; | |
179 | ||
180 | fail: | |
181 | vmsg_close_fds(vmsg); | |
182 | ||
183 | return false; | |
184 | } | |
185 | ||
186 | ||
187 | static void vu_client_start(VuServer *server); | |
188 | static coroutine_fn void vu_client_trip(void *opaque) | |
189 | { | |
190 | VuServer *server = opaque; | |
191 | ||
192 | while (!server->aio_context_changed && server->sioc) { | |
193 | server->processing_msg = true; | |
194 | vu_dispatch(&server->vu_dev); | |
195 | server->processing_msg = false; | |
196 | } | |
197 | ||
198 | if (server->aio_context_changed && server->sioc) { | |
199 | server->aio_context_changed = false; | |
200 | vu_client_start(server); | |
201 | } | |
202 | } | |
203 | ||
204 | static void vu_client_start(VuServer *server) | |
205 | { | |
206 | server->co_trip = qemu_coroutine_create(vu_client_trip, server); | |
207 | aio_co_enter(server->ctx, server->co_trip); | |
208 | } | |
209 | ||
210 | /* | |
211 | * a wrapper for vu_kick_cb | |
212 | * | |
213 | * since aio_dispatch can only pass one user data pointer to the | |
214 | * callback function, pack VuDev and pvt into a struct. Then unpack it | |
215 | * and pass them to vu_kick_cb | |
216 | */ | |
217 | static void kick_handler(void *opaque) | |
218 | { | |
219 | VuFdWatch *vu_fd_watch = opaque; | |
220 | vu_fd_watch->processing = true; | |
221 | vu_fd_watch->cb(vu_fd_watch->vu_dev, 0, vu_fd_watch->pvt); | |
222 | vu_fd_watch->processing = false; | |
223 | } | |
224 | ||
225 | ||
226 | static VuFdWatch *find_vu_fd_watch(VuServer *server, int fd) | |
227 | { | |
228 | ||
229 | VuFdWatch *vu_fd_watch, *next; | |
230 | QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) { | |
231 | if (vu_fd_watch->fd == fd) { | |
232 | return vu_fd_watch; | |
233 | } | |
234 | } | |
235 | return NULL; | |
236 | } | |
237 | ||
238 | static void | |
239 | set_watch(VuDev *vu_dev, int fd, int vu_evt, | |
240 | vu_watch_cb cb, void *pvt) | |
241 | { | |
242 | ||
243 | VuServer *server = container_of(vu_dev, VuServer, vu_dev); | |
244 | g_assert(vu_dev); | |
245 | g_assert(fd >= 0); | |
246 | g_assert(cb); | |
247 | ||
248 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
249 | ||
250 | if (!vu_fd_watch) { | |
251 | VuFdWatch *vu_fd_watch = g_new0(VuFdWatch, 1); | |
252 | ||
253 | QTAILQ_INSERT_TAIL(&server->vu_fd_watches, vu_fd_watch, next); | |
254 | ||
255 | vu_fd_watch->fd = fd; | |
256 | vu_fd_watch->cb = cb; | |
257 | qemu_set_nonblock(fd); | |
258 | aio_set_fd_handler(server->ioc->ctx, fd, true, kick_handler, | |
259 | NULL, NULL, vu_fd_watch); | |
260 | vu_fd_watch->vu_dev = vu_dev; | |
261 | vu_fd_watch->pvt = pvt; | |
262 | } | |
263 | } | |
264 | ||
265 | ||
266 | static void remove_watch(VuDev *vu_dev, int fd) | |
267 | { | |
268 | VuServer *server; | |
269 | g_assert(vu_dev); | |
270 | g_assert(fd >= 0); | |
271 | ||
272 | server = container_of(vu_dev, VuServer, vu_dev); | |
273 | ||
274 | VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd); | |
275 | ||
276 | if (!vu_fd_watch) { | |
277 | return; | |
278 | } | |
279 | aio_set_fd_handler(server->ioc->ctx, fd, true, NULL, NULL, NULL, NULL); | |
280 | ||
281 | QTAILQ_REMOVE(&server->vu_fd_watches, vu_fd_watch, next); | |
282 | g_free(vu_fd_watch); | |
283 | } | |
284 | ||
285 | ||
286 | static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc, | |
287 | gpointer opaque) | |
288 | { | |
289 | VuServer *server = opaque; | |
290 | ||
291 | if (server->sioc) { | |
292 | warn_report("Only one vhost-user client is allowed to " | |
293 | "connect the server one time"); | |
294 | return; | |
295 | } | |
296 | ||
297 | if (!vu_init(&server->vu_dev, server->max_queues, sioc->fd, panic_cb, | |
298 | vu_message_read, set_watch, remove_watch, server->vu_iface)) { | |
299 | error_report("Failed to initialize libvhost-user"); | |
300 | return; | |
301 | } | |
302 | ||
303 | /* | |
304 | * Unset the callback function for network listener to make another | |
305 | * vhost-user client keeping waiting until this client disconnects | |
306 | */ | |
307 | qio_net_listener_set_client_func(server->listener, | |
308 | NULL, | |
309 | NULL, | |
310 | NULL); | |
311 | server->sioc = sioc; | |
312 | /* | |
313 | * Increase the object reference, so sioc will not freed by | |
314 | * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc)) | |
315 | */ | |
316 | object_ref(OBJECT(server->sioc)); | |
317 | qio_channel_set_name(QIO_CHANNEL(sioc), "vhost-user client"); | |
318 | server->ioc = QIO_CHANNEL(sioc); | |
319 | object_ref(OBJECT(server->ioc)); | |
320 | qio_channel_attach_aio_context(server->ioc, server->ctx); | |
46a096c8 | 321 | qio_channel_set_blocking(server->ioc, false, NULL); |
70eb2c07 CX |
322 | vu_client_start(server); |
323 | } | |
324 | ||
325 | ||
326 | void vhost_user_server_stop(VuServer *server) | |
327 | { | |
328 | if (server->sioc) { | |
329 | close_client(server); | |
330 | } | |
331 | ||
332 | if (server->listener) { | |
333 | qio_net_listener_disconnect(server->listener); | |
334 | object_unref(OBJECT(server->listener)); | |
335 | } | |
336 | ||
337 | } | |
338 | ||
339 | void vhost_user_server_set_aio_context(VuServer *server, AioContext *ctx) | |
340 | { | |
341 | VuFdWatch *vu_fd_watch, *next; | |
342 | void *opaque = NULL; | |
343 | IOHandler *io_read = NULL; | |
344 | bool attach; | |
345 | ||
346 | server->ctx = ctx ? ctx : qemu_get_aio_context(); | |
347 | ||
348 | if (!server->sioc) { | |
349 | /* not yet serving any client*/ | |
350 | return; | |
351 | } | |
352 | ||
353 | if (ctx) { | |
354 | qio_channel_attach_aio_context(server->ioc, ctx); | |
355 | server->aio_context_changed = true; | |
356 | io_read = kick_handler; | |
357 | attach = true; | |
358 | } else { | |
359 | qio_channel_detach_aio_context(server->ioc); | |
360 | /* server->ioc->ctx keeps the old AioConext */ | |
361 | ctx = server->ioc->ctx; | |
362 | attach = false; | |
363 | } | |
364 | ||
365 | QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) { | |
366 | if (vu_fd_watch->cb) { | |
367 | opaque = attach ? vu_fd_watch : NULL; | |
368 | aio_set_fd_handler(ctx, vu_fd_watch->fd, true, | |
369 | io_read, NULL, NULL, | |
370 | opaque); | |
371 | } | |
372 | } | |
373 | } | |
374 | ||
375 | ||
376 | bool vhost_user_server_start(VuServer *server, | |
377 | SocketAddress *socket_addr, | |
378 | AioContext *ctx, | |
379 | uint16_t max_queues, | |
70eb2c07 CX |
380 | const VuDevIface *vu_iface, |
381 | Error **errp) | |
382 | { | |
383 | QIONetListener *listener = qio_net_listener_new(); | |
384 | if (qio_net_listener_open_sync(listener, socket_addr, 1, | |
385 | errp) < 0) { | |
386 | object_unref(OBJECT(listener)); | |
387 | return false; | |
388 | } | |
389 | ||
1d787456 | 390 | /* zero out unspecified fields */ |
70eb2c07 CX |
391 | *server = (VuServer) { |
392 | .listener = listener, | |
393 | .vu_iface = vu_iface, | |
394 | .max_queues = max_queues, | |
395 | .ctx = ctx, | |
70eb2c07 CX |
396 | }; |
397 | ||
398 | qio_net_listener_set_name(server->listener, "vhost-user-backend-listener"); | |
399 | ||
400 | qio_net_listener_set_client_func(server->listener, | |
401 | vu_accept, | |
402 | server, | |
403 | NULL); | |
404 | ||
405 | QTAILQ_INIT(&server->vu_fd_watches); | |
406 | return true; | |
407 | } |