]> git.proxmox.com Git - mirror_qemu.git/blame - util/vhost-user-server.c
util/vhost-user-server: fix memory leak in vu_message_read()
[mirror_qemu.git] / util / vhost-user-server.c
CommitLineData
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
14static 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
22static 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
30static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc,
31 gpointer opaque);
32
33static 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
69static 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
94static bool coroutine_fn
95vu_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
180fail:
181 vmsg_close_fds(vmsg);
182
183 return false;
184}
185
186
187static void vu_client_start(VuServer *server);
188static 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
204static 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 */
217static 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
226static 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
238static void
239set_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
266static 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
286static 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
326void 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
339void 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
376bool 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}