util/vhost-user-server: drop unused DevicePanicNotifier
[qemu.git] / util / vhost-user-server.c
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
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);
61
62 /* vu_deinit() should have called remove_watch() */
63 assert(QTAILQ_EMPTY(&server->vu_fd_watches));
64
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
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;
103 /*
104 * Store fds/nfds returned from qio_channel_readv_full into
105 * temporary variables.
106 *
107 * VhostUserMsg is a packed structure, gcc will complain about passing
108 * pointer to a packed structure member if we pass &VhostUserMsg.fd_num
109 * and &VhostUserMsg.fds directly when calling qio_channel_readv_full,
110 * thus two temporary variables nfds and fds are used here.
111 */
112 size_t nfds = 0, nfds_t = 0;
113 const size_t max_fds = G_N_ELEMENTS(vmsg->fds);
114 int *fds_t = NULL;
115 VuServer *server = container_of(vu_dev, VuServer, vu_dev);
116 QIOChannel *ioc = server->ioc;
117
118 if (!ioc) {
119 error_report_err(local_err);
120 goto fail;
121 }
122
123 assert(qemu_in_coroutine());
124 do {
125 /*
126 * qio_channel_readv_full may have short reads, keeping calling it
127 * until getting VHOST_USER_HDR_SIZE or 0 bytes in total
128 */
129 rc = qio_channel_readv_full(ioc, &iov, 1, &fds_t, &nfds_t, &local_err);
130 if (rc < 0) {
131 if (rc == QIO_CHANNEL_ERR_BLOCK) {
132 qio_channel_yield(ioc, G_IO_IN);
133 continue;
134 } else {
135 error_report_err(local_err);
136 return false;
137 }
138 }
139 read_bytes += rc;
140 if (nfds_t > 0) {
141 if (nfds + nfds_t > max_fds) {
142 error_report("A maximum of %zu fds are allowed, "
143 "however got %zu fds now",
144 max_fds, nfds + nfds_t);
145 goto fail;
146 }
147 memcpy(vmsg->fds + nfds, fds_t,
148 nfds_t *sizeof(vmsg->fds[0]));
149 nfds += nfds_t;
150 g_free(fds_t);
151 }
152 if (read_bytes == VHOST_USER_HDR_SIZE || rc == 0) {
153 break;
154 }
155 iov.iov_base = (char *)vmsg + read_bytes;
156 iov.iov_len = VHOST_USER_HDR_SIZE - read_bytes;
157 } while (true);
158
159 vmsg->fd_num = nfds;
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));
167 goto fail;
168 }
169
170 struct iovec iov_payload = {
171 .iov_base = (char *)&vmsg->payload,
172 .iov_len = vmsg->size,
173 };
174 if (vmsg->size) {
175 rc = qio_channel_readv_all_eof(ioc, &iov_payload, 1, &local_err);
176 if (rc == -1) {
177 error_report_err(local_err);
178 goto fail;
179 }
180 }
181
182 return true;
183
184 fail:
185 vmsg_close_fds(vmsg);
186
187 return false;
188 }
189
190
191 static void vu_client_start(VuServer *server);
192 static coroutine_fn void vu_client_trip(void *opaque)
193 {
194 VuServer *server = opaque;
195
196 while (!server->aio_context_changed && server->sioc) {
197 server->processing_msg = true;
198 vu_dispatch(&server->vu_dev);
199 server->processing_msg = false;
200 }
201
202 if (server->aio_context_changed && server->sioc) {
203 server->aio_context_changed = false;
204 vu_client_start(server);
205 }
206 }
207
208 static void vu_client_start(VuServer *server)
209 {
210 server->co_trip = qemu_coroutine_create(vu_client_trip, server);
211 aio_co_enter(server->ctx, server->co_trip);
212 }
213
214 /*
215 * a wrapper for vu_kick_cb
216 *
217 * since aio_dispatch can only pass one user data pointer to the
218 * callback function, pack VuDev and pvt into a struct. Then unpack it
219 * and pass them to vu_kick_cb
220 */
221 static void kick_handler(void *opaque)
222 {
223 VuFdWatch *vu_fd_watch = opaque;
224 vu_fd_watch->processing = true;
225 vu_fd_watch->cb(vu_fd_watch->vu_dev, 0, vu_fd_watch->pvt);
226 vu_fd_watch->processing = false;
227 }
228
229
230 static VuFdWatch *find_vu_fd_watch(VuServer *server, int fd)
231 {
232
233 VuFdWatch *vu_fd_watch, *next;
234 QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) {
235 if (vu_fd_watch->fd == fd) {
236 return vu_fd_watch;
237 }
238 }
239 return NULL;
240 }
241
242 static void
243 set_watch(VuDev *vu_dev, int fd, int vu_evt,
244 vu_watch_cb cb, void *pvt)
245 {
246
247 VuServer *server = container_of(vu_dev, VuServer, vu_dev);
248 g_assert(vu_dev);
249 g_assert(fd >= 0);
250 g_assert(cb);
251
252 VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd);
253
254 if (!vu_fd_watch) {
255 VuFdWatch *vu_fd_watch = g_new0(VuFdWatch, 1);
256
257 QTAILQ_INSERT_TAIL(&server->vu_fd_watches, vu_fd_watch, next);
258
259 vu_fd_watch->fd = fd;
260 vu_fd_watch->cb = cb;
261 qemu_set_nonblock(fd);
262 aio_set_fd_handler(server->ioc->ctx, fd, true, kick_handler,
263 NULL, NULL, vu_fd_watch);
264 vu_fd_watch->vu_dev = vu_dev;
265 vu_fd_watch->pvt = pvt;
266 }
267 }
268
269
270 static void remove_watch(VuDev *vu_dev, int fd)
271 {
272 VuServer *server;
273 g_assert(vu_dev);
274 g_assert(fd >= 0);
275
276 server = container_of(vu_dev, VuServer, vu_dev);
277
278 VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd);
279
280 if (!vu_fd_watch) {
281 return;
282 }
283 aio_set_fd_handler(server->ioc->ctx, fd, true, NULL, NULL, NULL, NULL);
284
285 QTAILQ_REMOVE(&server->vu_fd_watches, vu_fd_watch, next);
286 g_free(vu_fd_watch);
287 }
288
289
290 static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc,
291 gpointer opaque)
292 {
293 VuServer *server = opaque;
294
295 if (server->sioc) {
296 warn_report("Only one vhost-user client is allowed to "
297 "connect the server one time");
298 return;
299 }
300
301 if (!vu_init(&server->vu_dev, server->max_queues, sioc->fd, panic_cb,
302 vu_message_read, set_watch, remove_watch, server->vu_iface)) {
303 error_report("Failed to initialize libvhost-user");
304 return;
305 }
306
307 /*
308 * Unset the callback function for network listener to make another
309 * vhost-user client keeping waiting until this client disconnects
310 */
311 qio_net_listener_set_client_func(server->listener,
312 NULL,
313 NULL,
314 NULL);
315 server->sioc = sioc;
316 /*
317 * Increase the object reference, so sioc will not freed by
318 * qio_net_listener_channel_func which will call object_unref(OBJECT(sioc))
319 */
320 object_ref(OBJECT(server->sioc));
321 qio_channel_set_name(QIO_CHANNEL(sioc), "vhost-user client");
322 server->ioc = QIO_CHANNEL(sioc);
323 object_ref(OBJECT(server->ioc));
324 qio_channel_attach_aio_context(server->ioc, server->ctx);
325 qio_channel_set_blocking(server->ioc, false, NULL);
326 vu_client_start(server);
327 }
328
329
330 void vhost_user_server_stop(VuServer *server)
331 {
332 if (server->sioc) {
333 close_client(server);
334 }
335
336 if (server->listener) {
337 qio_net_listener_disconnect(server->listener);
338 object_unref(OBJECT(server->listener));
339 }
340
341 }
342
343 void vhost_user_server_set_aio_context(VuServer *server, AioContext *ctx)
344 {
345 VuFdWatch *vu_fd_watch, *next;
346 void *opaque = NULL;
347 IOHandler *io_read = NULL;
348 bool attach;
349
350 server->ctx = ctx ? ctx : qemu_get_aio_context();
351
352 if (!server->sioc) {
353 /* not yet serving any client*/
354 return;
355 }
356
357 if (ctx) {
358 qio_channel_attach_aio_context(server->ioc, ctx);
359 server->aio_context_changed = true;
360 io_read = kick_handler;
361 attach = true;
362 } else {
363 qio_channel_detach_aio_context(server->ioc);
364 /* server->ioc->ctx keeps the old AioConext */
365 ctx = server->ioc->ctx;
366 attach = false;
367 }
368
369 QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) {
370 if (vu_fd_watch->cb) {
371 opaque = attach ? vu_fd_watch : NULL;
372 aio_set_fd_handler(ctx, vu_fd_watch->fd, true,
373 io_read, NULL, NULL,
374 opaque);
375 }
376 }
377 }
378
379
380 bool vhost_user_server_start(VuServer *server,
381 SocketAddress *socket_addr,
382 AioContext *ctx,
383 uint16_t max_queues,
384 const VuDevIface *vu_iface,
385 Error **errp)
386 {
387 QIONetListener *listener = qio_net_listener_new();
388 if (qio_net_listener_open_sync(listener, socket_addr, 1,
389 errp) < 0) {
390 object_unref(OBJECT(listener));
391 return false;
392 }
393
394 /* zero out unspecified fields */
395 *server = (VuServer) {
396 .listener = listener,
397 .vu_iface = vu_iface,
398 .max_queues = max_queues,
399 .ctx = ctx,
400 };
401
402 qio_net_listener_set_name(server->listener, "vhost-user-backend-listener");
403
404 qio_net_listener_set_client_func(server->listener,
405 vu_accept,
406 server,
407 NULL);
408
409 QTAILQ_INIT(&server->vu_fd_watches);
410 return true;
411 }