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