1
2
3
4
5
6
7
8
9
10#include "qemu/osdep.h"
11#include "qemu/main-loop.h"
12#include "qemu/vhost-user-server.h"
13#include "block/aio-wait.h"
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56static void vmsg_close_fds(VhostUserMsg *vmsg)
57{
58 int i;
59 for (i = 0; i < vmsg->fd_num; i++) {
60 close(vmsg->fds[i]);
61 }
62}
63
64static void vmsg_unblock_fds(VhostUserMsg *vmsg)
65{
66 int i;
67 for (i = 0; i < vmsg->fd_num; i++) {
68 qemu_socket_set_nonblock(vmsg->fds[i]);
69 }
70}
71
72static void panic_cb(VuDev *vu_dev, const char *buf)
73{
74 error_report("vu_panic: %s", buf);
75}
76
77void vhost_user_server_ref(VuServer *server)
78{
79 assert(!server->wait_idle);
80 server->refcount++;
81}
82
83void vhost_user_server_unref(VuServer *server)
84{
85 server->refcount--;
86 if (server->wait_idle && !server->refcount) {
87 aio_co_wake(server->co_trip);
88 }
89}
90
91static bool coroutine_fn
92vu_message_read(VuDev *vu_dev, int conn_fd, VhostUserMsg *vmsg)
93{
94 struct iovec iov = {
95 .iov_base = (char *)vmsg,
96 .iov_len = VHOST_USER_HDR_SIZE,
97 };
98 int rc, read_bytes = 0;
99 Error *local_err = NULL;
100 const size_t max_fds = G_N_ELEMENTS(vmsg->fds);
101 VuServer *server = container_of(vu_dev, VuServer, vu_dev);
102 QIOChannel *ioc = server->ioc;
103
104 vmsg->fd_num = 0;
105 if (!ioc) {
106 error_report_err(local_err);
107 goto fail;
108 }
109
110 assert(qemu_in_coroutine());
111 do {
112 size_t nfds = 0;
113 int *fds = NULL;
114
115
116
117
118
119 rc = qio_channel_readv_full(ioc, &iov, 1, &fds, &nfds, &local_err);
120 if (rc < 0) {
121 if (rc == QIO_CHANNEL_ERR_BLOCK) {
122 assert(local_err == NULL);
123 qio_channel_yield(ioc, G_IO_IN);
124 continue;
125 } else {
126 error_report_err(local_err);
127 goto fail;
128 }
129 }
130
131 if (nfds > 0) {
132 if (vmsg->fd_num + nfds > max_fds) {
133 error_report("A maximum of %zu fds are allowed, "
134 "however got %zu fds now",
135 max_fds, vmsg->fd_num + nfds);
136 g_free(fds);
137 goto fail;
138 }
139 memcpy(vmsg->fds + vmsg->fd_num, fds, nfds * sizeof(vmsg->fds[0]));
140 vmsg->fd_num += nfds;
141 g_free(fds);
142 }
143
144 if (rc == 0) {
145 goto fail;
146 }
147
148 iov.iov_base += rc;
149 iov.iov_len -= rc;
150 read_bytes += rc;
151 } while (read_bytes != VHOST_USER_HDR_SIZE);
152
153
154 vmsg_unblock_fds(vmsg);
155 if (vmsg->size > sizeof(vmsg->payload)) {
156 error_report("Error: too big message request: %d, "
157 "size: vmsg->size: %u, "
158 "while sizeof(vmsg->payload) = %zu",
159 vmsg->request, vmsg->size, sizeof(vmsg->payload));
160 goto fail;
161 }
162
163 struct iovec iov_payload = {
164 .iov_base = (char *)&vmsg->payload,
165 .iov_len = vmsg->size,
166 };
167 if (vmsg->size) {
168 rc = qio_channel_readv_all_eof(ioc, &iov_payload, 1, &local_err);
169 if (rc != 1) {
170 if (local_err) {
171 error_report_err(local_err);
172 }
173 goto fail;
174 }
175 }
176
177 return true;
178
179fail:
180 vmsg_close_fds(vmsg);
181
182 return false;
183}
184
185static coroutine_fn void vu_client_trip(void *opaque)
186{
187 VuServer *server = opaque;
188 VuDev *vu_dev = &server->vu_dev;
189
190 while (!vu_dev->broken && vu_dispatch(vu_dev)) {
191
192 }
193
194 if (server->refcount) {
195
196 server->wait_idle = true;
197 qemu_coroutine_yield();
198 server->wait_idle = false;
199 }
200 assert(server->refcount == 0);
201
202 vu_deinit(vu_dev);
203
204
205 assert(QTAILQ_EMPTY(&server->vu_fd_watches));
206
207 object_unref(OBJECT(server->sioc));
208 server->sioc = NULL;
209
210 object_unref(OBJECT(server->ioc));
211 server->ioc = NULL;
212
213 server->co_trip = NULL;
214 if (server->restart_listener_bh) {
215 qemu_bh_schedule(server->restart_listener_bh);
216 }
217 aio_wait_kick();
218}
219
220
221
222
223
224
225
226
227static void kick_handler(void *opaque)
228{
229 VuFdWatch *vu_fd_watch = opaque;
230 VuDev *vu_dev = vu_fd_watch->vu_dev;
231
232 vu_fd_watch->cb(vu_dev, 0, vu_fd_watch->pvt);
233
234
235 if (vu_dev->broken) {
236 VuServer *server = container_of(vu_dev, VuServer, vu_dev);
237
238 qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
239 }
240}
241
242static VuFdWatch *find_vu_fd_watch(VuServer *server, int fd)
243{
244
245 VuFdWatch *vu_fd_watch, *next;
246 QTAILQ_FOREACH_SAFE(vu_fd_watch, &server->vu_fd_watches, next, next) {
247 if (vu_fd_watch->fd == fd) {
248 return vu_fd_watch;
249 }
250 }
251 return NULL;
252}
253
254static void
255set_watch(VuDev *vu_dev, int fd, int vu_evt,
256 vu_watch_cb cb, void *pvt)
257{
258
259 VuServer *server = container_of(vu_dev, VuServer, vu_dev);
260 g_assert(vu_dev);
261 g_assert(fd >= 0);
262 g_assert(cb);
263
264 VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd);
265
266 if (!vu_fd_watch) {
267 VuFdWatch *vu_fd_watch = g_new0(VuFdWatch, 1);
268
269 QTAILQ_INSERT_TAIL(&server->vu_fd_watches, vu_fd_watch, next);
270
271 vu_fd_watch->fd = fd;
272 vu_fd_watch->cb = cb;
273 qemu_socket_set_nonblock(fd);
274 aio_set_fd_handler(server->ioc->ctx, fd, true, kick_handler,
275 NULL, NULL, NULL, vu_fd_watch);
276 vu_fd_watch->vu_dev = vu_dev;
277 vu_fd_watch->pvt = pvt;
278 }
279}
280
281
282static void remove_watch(VuDev *vu_dev, int fd)
283{
284 VuServer *server;
285 g_assert(vu_dev);
286 g_assert(fd >= 0);
287
288 server = container_of(vu_dev, VuServer, vu_dev);
289
290 VuFdWatch *vu_fd_watch = find_vu_fd_watch(server, fd);
291
292 if (!vu_fd_watch) {
293 return;
294 }
295 aio_set_fd_handler(server->ioc->ctx, fd, true,
296 NULL, NULL, NULL, NULL, NULL);
297
298 QTAILQ_REMOVE(&server->vu_fd_watches, vu_fd_watch, next);
299 g_free(vu_fd_watch);
300}
301
302
303static void vu_accept(QIONetListener *listener, QIOChannelSocket *sioc,
304 gpointer opaque)
305{
306 VuServer *server = opaque;
307
308 if (server->sioc) {
309 warn_report("Only one vhost-user client is allowed to "
310 "connect the server one time");
311 return;
312 }
313
314 if (!vu_init(&server->vu_dev, server->max_queues, sioc->fd, panic_cb,
315 vu_message_read, set_watch, remove_watch, server->vu_iface)) {
316 error_report("Failed to initialize libvhost-user");
317 return;
318 }
319
320
321
322
323
324 qio_net_listener_set_client_func(server->listener,
325 NULL,
326 NULL,
327 NULL);
328 server->sioc = sioc;
329
330
331
332
333 object_ref(OBJECT(server->sioc));
334 qio_channel_set_name(QIO_CHANNEL(sioc), "vhost-user client");
335 server->ioc = QIO_CHANNEL(sioc);
336 object_ref(OBJECT(server->ioc));
337
338
339 qio_channel_set_blocking(server->ioc, false, NULL);
340
341 server->co_trip = qemu_coroutine_create(vu_client_trip, server);
342
343 aio_context_acquire(server->ctx);
344 vhost_user_server_attach_aio_context(server, server->ctx);
345 aio_context_release(server->ctx);
346}
347
348void vhost_user_server_stop(VuServer *server)
349{
350 aio_context_acquire(server->ctx);
351
352 qemu_bh_delete(server->restart_listener_bh);
353 server->restart_listener_bh = NULL;
354
355 if (server->sioc) {
356 VuFdWatch *vu_fd_watch;
357
358 QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) {
359 aio_set_fd_handler(server->ctx, vu_fd_watch->fd, true,
360 NULL, NULL, NULL, NULL, vu_fd_watch);
361 }
362
363 qio_channel_shutdown(server->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
364
365 AIO_WAIT_WHILE(server->ctx, server->co_trip);
366 }
367
368 aio_context_release(server->ctx);
369
370 if (server->listener) {
371 qio_net_listener_disconnect(server->listener);
372 object_unref(OBJECT(server->listener));
373 }
374}
375
376
377
378
379
380static void restart_listener_bh(void *opaque)
381{
382 VuServer *server = opaque;
383
384 qio_net_listener_set_client_func(server->listener, vu_accept, server,
385 NULL);
386}
387
388
389void vhost_user_server_attach_aio_context(VuServer *server, AioContext *ctx)
390{
391 VuFdWatch *vu_fd_watch;
392
393 server->ctx = ctx;
394
395 if (!server->sioc) {
396 return;
397 }
398
399 qio_channel_attach_aio_context(server->ioc, ctx);
400
401 QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) {
402 aio_set_fd_handler(ctx, vu_fd_watch->fd, true, kick_handler, NULL,
403 NULL, NULL, vu_fd_watch);
404 }
405
406 aio_co_schedule(ctx, server->co_trip);
407}
408
409
410void vhost_user_server_detach_aio_context(VuServer *server)
411{
412 if (server->sioc) {
413 VuFdWatch *vu_fd_watch;
414
415 QTAILQ_FOREACH(vu_fd_watch, &server->vu_fd_watches, next) {
416 aio_set_fd_handler(server->ctx, vu_fd_watch->fd, true,
417 NULL, NULL, NULL, NULL, vu_fd_watch);
418 }
419
420 qio_channel_detach_aio_context(server->ioc);
421 }
422
423 server->ctx = NULL;
424}
425
426bool vhost_user_server_start(VuServer *server,
427 SocketAddress *socket_addr,
428 AioContext *ctx,
429 uint16_t max_queues,
430 const VuDevIface *vu_iface,
431 Error **errp)
432{
433 QEMUBH *bh;
434 QIONetListener *listener;
435
436 if (socket_addr->type != SOCKET_ADDRESS_TYPE_UNIX &&
437 socket_addr->type != SOCKET_ADDRESS_TYPE_FD) {
438 error_setg(errp, "Only socket address types 'unix' and 'fd' are supported");
439 return false;
440 }
441
442 listener = qio_net_listener_new();
443 if (qio_net_listener_open_sync(listener, socket_addr, 1,
444 errp) < 0) {
445 object_unref(OBJECT(listener));
446 return false;
447 }
448
449 bh = qemu_bh_new(restart_listener_bh, server);
450
451
452 *server = (VuServer) {
453 .listener = listener,
454 .restart_listener_bh = bh,
455 .vu_iface = vu_iface,
456 .max_queues = max_queues,
457 .ctx = ctx,
458 };
459
460 qio_net_listener_set_name(server->listener, "vhost-user-backend-listener");
461
462 qio_net_listener_set_client_func(server->listener,
463 vu_accept,
464 server,
465 NULL);
466
467 QTAILQ_INIT(&server->vu_fd_watches);
468 return true;
469}
470