Merge remote-tracking branch 'remotes/mst/tags/for_upstream' into staging
[qemu.git] / io / channel.c
1 /*
2 * QEMU I/O channels
3 *
4 * Copyright (c) 2015 Red Hat, Inc.
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, see <http://www.gnu.org/licenses/>.
18 *
19 */
20
21 #include "qemu/osdep.h"
22 #include "io/channel.h"
23 #include "qapi/error.h"
24 #include "qemu/main-loop.h"
25 #include "qemu/module.h"
26 #include "qemu/iov.h"
27
28 bool qio_channel_has_feature(QIOChannel *ioc,
29 QIOChannelFeature feature)
30 {
31 return ioc->features & (1 << feature);
32 }
33
34
35 void qio_channel_set_feature(QIOChannel *ioc,
36 QIOChannelFeature feature)
37 {
38 ioc->features |= (1 << feature);
39 }
40
41
42 void qio_channel_set_name(QIOChannel *ioc,
43 const char *name)
44 {
45 g_free(ioc->name);
46 ioc->name = g_strdup(name);
47 }
48
49
50 ssize_t qio_channel_readv_full(QIOChannel *ioc,
51 const struct iovec *iov,
52 size_t niov,
53 int **fds,
54 size_t *nfds,
55 Error **errp)
56 {
57 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
58
59 if ((fds || nfds) &&
60 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
61 error_setg_errno(errp, EINVAL,
62 "Channel does not support file descriptor passing");
63 return -1;
64 }
65
66 return klass->io_readv(ioc, iov, niov, fds, nfds, errp);
67 }
68
69
70 ssize_t qio_channel_writev_full(QIOChannel *ioc,
71 const struct iovec *iov,
72 size_t niov,
73 int *fds,
74 size_t nfds,
75 Error **errp)
76 {
77 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
78
79 if ((fds || nfds) &&
80 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
81 error_setg_errno(errp, EINVAL,
82 "Channel does not support file descriptor passing");
83 return -1;
84 }
85
86 return klass->io_writev(ioc, iov, niov, fds, nfds, errp);
87 }
88
89
90 int qio_channel_readv_all_eof(QIOChannel *ioc,
91 const struct iovec *iov,
92 size_t niov,
93 Error **errp)
94 {
95 return qio_channel_readv_full_all_eof(ioc, iov, niov, NULL, NULL, errp);
96 }
97
98 int qio_channel_readv_all(QIOChannel *ioc,
99 const struct iovec *iov,
100 size_t niov,
101 Error **errp)
102 {
103 return qio_channel_readv_full_all(ioc, iov, niov, NULL, NULL, errp);
104 }
105
106 int qio_channel_readv_full_all_eof(QIOChannel *ioc,
107 const struct iovec *iov,
108 size_t niov,
109 int **fds, size_t *nfds,
110 Error **errp)
111 {
112 int ret = -1;
113 struct iovec *local_iov = g_new(struct iovec, niov);
114 struct iovec *local_iov_head = local_iov;
115 unsigned int nlocal_iov = niov;
116 int **local_fds = fds;
117 size_t *local_nfds = nfds;
118 bool partial = false;
119
120 if (nfds) {
121 *nfds = 0;
122 }
123
124 if (fds) {
125 *fds = NULL;
126 }
127
128 nlocal_iov = iov_copy(local_iov, nlocal_iov,
129 iov, niov,
130 0, iov_size(iov, niov));
131
132 while ((nlocal_iov > 0) || local_fds) {
133 ssize_t len;
134 len = qio_channel_readv_full(ioc, local_iov, nlocal_iov, local_fds,
135 local_nfds, errp);
136 if (len == QIO_CHANNEL_ERR_BLOCK) {
137 if (qemu_in_coroutine()) {
138 qio_channel_yield(ioc, G_IO_IN);
139 } else {
140 qio_channel_wait(ioc, G_IO_IN);
141 }
142 continue;
143 }
144
145 if (len == 0) {
146 if (local_nfds && *local_nfds) {
147 /*
148 * Got some FDs, but no data yet. This isn't an EOF
149 * scenario (yet), so carry on to try to read data
150 * on next loop iteration
151 */
152 goto next_iter;
153 } else if (!partial) {
154 /* No fds and no data - EOF before any data read */
155 ret = 0;
156 goto cleanup;
157 } else {
158 len = -1;
159 error_setg(errp,
160 "Unexpected end-of-file before all data were read");
161 /* Fallthrough into len < 0 handling */
162 }
163 }
164
165 if (len < 0) {
166 /* Close any FDs we previously received */
167 if (nfds && fds) {
168 size_t i;
169 for (i = 0; i < (*nfds); i++) {
170 close((*fds)[i]);
171 }
172 g_free(*fds);
173 *fds = NULL;
174 *nfds = 0;
175 }
176 goto cleanup;
177 }
178
179 if (nlocal_iov) {
180 iov_discard_front(&local_iov, &nlocal_iov, len);
181 }
182
183 next_iter:
184 partial = true;
185 local_fds = NULL;
186 local_nfds = NULL;
187 }
188
189 ret = 1;
190
191 cleanup:
192 g_free(local_iov_head);
193 return ret;
194 }
195
196 int qio_channel_readv_full_all(QIOChannel *ioc,
197 const struct iovec *iov,
198 size_t niov,
199 int **fds, size_t *nfds,
200 Error **errp)
201 {
202 int ret = qio_channel_readv_full_all_eof(ioc, iov, niov, fds, nfds, errp);
203
204 if (ret == 0) {
205 error_setg(errp, "Unexpected end-of-file before all data were read");
206 return -1;
207 }
208 if (ret == 1) {
209 return 0;
210 }
211
212 return ret;
213 }
214
215 int qio_channel_writev_all(QIOChannel *ioc,
216 const struct iovec *iov,
217 size_t niov,
218 Error **errp)
219 {
220 return qio_channel_writev_full_all(ioc, iov, niov, NULL, 0, errp);
221 }
222
223 int qio_channel_writev_full_all(QIOChannel *ioc,
224 const struct iovec *iov,
225 size_t niov,
226 int *fds, size_t nfds,
227 Error **errp)
228 {
229 int ret = -1;
230 struct iovec *local_iov = g_new(struct iovec, niov);
231 struct iovec *local_iov_head = local_iov;
232 unsigned int nlocal_iov = niov;
233
234 nlocal_iov = iov_copy(local_iov, nlocal_iov,
235 iov, niov,
236 0, iov_size(iov, niov));
237
238 while (nlocal_iov > 0) {
239 ssize_t len;
240 len = qio_channel_writev_full(ioc, local_iov, nlocal_iov, fds, nfds,
241 errp);
242 if (len == QIO_CHANNEL_ERR_BLOCK) {
243 if (qemu_in_coroutine()) {
244 qio_channel_yield(ioc, G_IO_OUT);
245 } else {
246 qio_channel_wait(ioc, G_IO_OUT);
247 }
248 continue;
249 }
250 if (len < 0) {
251 goto cleanup;
252 }
253
254 iov_discard_front(&local_iov, &nlocal_iov, len);
255
256 fds = NULL;
257 nfds = 0;
258 }
259
260 ret = 0;
261 cleanup:
262 g_free(local_iov_head);
263 return ret;
264 }
265
266 ssize_t qio_channel_readv(QIOChannel *ioc,
267 const struct iovec *iov,
268 size_t niov,
269 Error **errp)
270 {
271 return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, errp);
272 }
273
274
275 ssize_t qio_channel_writev(QIOChannel *ioc,
276 const struct iovec *iov,
277 size_t niov,
278 Error **errp)
279 {
280 return qio_channel_writev_full(ioc, iov, niov, NULL, 0, errp);
281 }
282
283
284 ssize_t qio_channel_read(QIOChannel *ioc,
285 char *buf,
286 size_t buflen,
287 Error **errp)
288 {
289 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
290 return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, errp);
291 }
292
293
294 ssize_t qio_channel_write(QIOChannel *ioc,
295 const char *buf,
296 size_t buflen,
297 Error **errp)
298 {
299 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
300 return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, errp);
301 }
302
303
304 int qio_channel_read_all_eof(QIOChannel *ioc,
305 char *buf,
306 size_t buflen,
307 Error **errp)
308 {
309 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
310 return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
311 }
312
313
314 int qio_channel_read_all(QIOChannel *ioc,
315 char *buf,
316 size_t buflen,
317 Error **errp)
318 {
319 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
320 return qio_channel_readv_all(ioc, &iov, 1, errp);
321 }
322
323
324 int qio_channel_write_all(QIOChannel *ioc,
325 const char *buf,
326 size_t buflen,
327 Error **errp)
328 {
329 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
330 return qio_channel_writev_all(ioc, &iov, 1, errp);
331 }
332
333
334 int qio_channel_set_blocking(QIOChannel *ioc,
335 bool enabled,
336 Error **errp)
337 {
338 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
339 return klass->io_set_blocking(ioc, enabled, errp);
340 }
341
342
343 int qio_channel_close(QIOChannel *ioc,
344 Error **errp)
345 {
346 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
347 return klass->io_close(ioc, errp);
348 }
349
350
351 GSource *qio_channel_create_watch(QIOChannel *ioc,
352 GIOCondition condition)
353 {
354 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
355 GSource *ret = klass->io_create_watch(ioc, condition);
356
357 if (ioc->name) {
358 g_source_set_name(ret, ioc->name);
359 }
360
361 return ret;
362 }
363
364
365 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
366 AioContext *ctx,
367 IOHandler *io_read,
368 IOHandler *io_write,
369 void *opaque)
370 {
371 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
372
373 klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
374 }
375
376 guint qio_channel_add_watch_full(QIOChannel *ioc,
377 GIOCondition condition,
378 QIOChannelFunc func,
379 gpointer user_data,
380 GDestroyNotify notify,
381 GMainContext *context)
382 {
383 GSource *source;
384 guint id;
385
386 source = qio_channel_create_watch(ioc, condition);
387
388 g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
389
390 id = g_source_attach(source, context);
391 g_source_unref(source);
392
393 return id;
394 }
395
396 guint qio_channel_add_watch(QIOChannel *ioc,
397 GIOCondition condition,
398 QIOChannelFunc func,
399 gpointer user_data,
400 GDestroyNotify notify)
401 {
402 return qio_channel_add_watch_full(ioc, condition, func,
403 user_data, notify, NULL);
404 }
405
406 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
407 GIOCondition condition,
408 QIOChannelFunc func,
409 gpointer user_data,
410 GDestroyNotify notify,
411 GMainContext *context)
412 {
413 GSource *source;
414 guint id;
415
416 id = qio_channel_add_watch_full(ioc, condition, func,
417 user_data, notify, context);
418 source = g_main_context_find_source_by_id(context, id);
419 g_source_ref(source);
420 return source;
421 }
422
423
424 int qio_channel_shutdown(QIOChannel *ioc,
425 QIOChannelShutdown how,
426 Error **errp)
427 {
428 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
429
430 if (!klass->io_shutdown) {
431 error_setg(errp, "Data path shutdown not supported");
432 return -1;
433 }
434
435 return klass->io_shutdown(ioc, how, errp);
436 }
437
438
439 void qio_channel_set_delay(QIOChannel *ioc,
440 bool enabled)
441 {
442 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
443
444 if (klass->io_set_delay) {
445 klass->io_set_delay(ioc, enabled);
446 }
447 }
448
449
450 void qio_channel_set_cork(QIOChannel *ioc,
451 bool enabled)
452 {
453 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
454
455 if (klass->io_set_cork) {
456 klass->io_set_cork(ioc, enabled);
457 }
458 }
459
460
461 off_t qio_channel_io_seek(QIOChannel *ioc,
462 off_t offset,
463 int whence,
464 Error **errp)
465 {
466 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
467
468 if (!klass->io_seek) {
469 error_setg(errp, "Channel does not support random access");
470 return -1;
471 }
472
473 return klass->io_seek(ioc, offset, whence, errp);
474 }
475
476
477 static void qio_channel_restart_read(void *opaque)
478 {
479 QIOChannel *ioc = opaque;
480 Coroutine *co = ioc->read_coroutine;
481
482 /* Assert that aio_co_wake() reenters the coroutine directly */
483 assert(qemu_get_current_aio_context() ==
484 qemu_coroutine_get_aio_context(co));
485 aio_co_wake(co);
486 }
487
488 static void qio_channel_restart_write(void *opaque)
489 {
490 QIOChannel *ioc = opaque;
491 Coroutine *co = ioc->write_coroutine;
492
493 /* Assert that aio_co_wake() reenters the coroutine directly */
494 assert(qemu_get_current_aio_context() ==
495 qemu_coroutine_get_aio_context(co));
496 aio_co_wake(co);
497 }
498
499 static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
500 {
501 IOHandler *rd_handler = NULL, *wr_handler = NULL;
502 AioContext *ctx;
503
504 if (ioc->read_coroutine) {
505 rd_handler = qio_channel_restart_read;
506 }
507 if (ioc->write_coroutine) {
508 wr_handler = qio_channel_restart_write;
509 }
510
511 ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
512 qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
513 }
514
515 void qio_channel_attach_aio_context(QIOChannel *ioc,
516 AioContext *ctx)
517 {
518 assert(!ioc->read_coroutine);
519 assert(!ioc->write_coroutine);
520 ioc->ctx = ctx;
521 }
522
523 void qio_channel_detach_aio_context(QIOChannel *ioc)
524 {
525 ioc->read_coroutine = NULL;
526 ioc->write_coroutine = NULL;
527 qio_channel_set_aio_fd_handlers(ioc);
528 ioc->ctx = NULL;
529 }
530
531 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
532 GIOCondition condition)
533 {
534 assert(qemu_in_coroutine());
535 if (condition == G_IO_IN) {
536 assert(!ioc->read_coroutine);
537 ioc->read_coroutine = qemu_coroutine_self();
538 } else if (condition == G_IO_OUT) {
539 assert(!ioc->write_coroutine);
540 ioc->write_coroutine = qemu_coroutine_self();
541 } else {
542 abort();
543 }
544 qio_channel_set_aio_fd_handlers(ioc);
545 qemu_coroutine_yield();
546
547 /* Allow interrupting the operation by reentering the coroutine other than
548 * through the aio_fd_handlers. */
549 if (condition == G_IO_IN && ioc->read_coroutine) {
550 ioc->read_coroutine = NULL;
551 qio_channel_set_aio_fd_handlers(ioc);
552 } else if (condition == G_IO_OUT && ioc->write_coroutine) {
553 ioc->write_coroutine = NULL;
554 qio_channel_set_aio_fd_handlers(ioc);
555 }
556 }
557
558
559 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
560 GIOCondition condition,
561 gpointer opaque)
562 {
563 GMainLoop *loop = opaque;
564
565 g_main_loop_quit(loop);
566 return FALSE;
567 }
568
569
570 void qio_channel_wait(QIOChannel *ioc,
571 GIOCondition condition)
572 {
573 GMainContext *ctxt = g_main_context_new();
574 GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
575 GSource *source;
576
577 source = qio_channel_create_watch(ioc, condition);
578
579 g_source_set_callback(source,
580 (GSourceFunc)qio_channel_wait_complete,
581 loop,
582 NULL);
583
584 g_source_attach(source, ctxt);
585
586 g_main_loop_run(loop);
587
588 g_source_unref(source);
589 g_main_loop_unref(loop);
590 g_main_context_unref(ctxt);
591 }
592
593
594 static void qio_channel_finalize(Object *obj)
595 {
596 QIOChannel *ioc = QIO_CHANNEL(obj);
597
598 g_free(ioc->name);
599
600 #ifdef _WIN32
601 if (ioc->event) {
602 CloseHandle(ioc->event);
603 }
604 #endif
605 }
606
607 static const TypeInfo qio_channel_info = {
608 .parent = TYPE_OBJECT,
609 .name = TYPE_QIO_CHANNEL,
610 .instance_size = sizeof(QIOChannel),
611 .instance_finalize = qio_channel_finalize,
612 .abstract = true,
613 .class_size = sizeof(QIOChannelClass),
614 };
615
616
617 static void qio_channel_register_types(void)
618 {
619 type_register_static(&qio_channel_info);
620 }
621
622
623 type_init(qio_channel_register_types);