Merge tag 'for-upstream' of git://repo.or.cz/qemu/kevin 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 int flags,
76 Error **errp)
77 {
78 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
79
80 if (fds || nfds) {
81 if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
82 error_setg_errno(errp, EINVAL,
83 "Channel does not support file descriptor passing");
84 return -1;
85 }
86 if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
87 error_setg_errno(errp, EINVAL,
88 "Zero Copy does not support file descriptor passing");
89 return -1;
90 }
91 }
92
93 if ((flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) &&
94 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
95 error_setg_errno(errp, EINVAL,
96 "Requested Zero Copy feature is not available");
97 return -1;
98 }
99
100 return klass->io_writev(ioc, iov, niov, fds, nfds, flags, errp);
101 }
102
103
104 int qio_channel_readv_all_eof(QIOChannel *ioc,
105 const struct iovec *iov,
106 size_t niov,
107 Error **errp)
108 {
109 return qio_channel_readv_full_all_eof(ioc, iov, niov, NULL, NULL, errp);
110 }
111
112 int qio_channel_readv_all(QIOChannel *ioc,
113 const struct iovec *iov,
114 size_t niov,
115 Error **errp)
116 {
117 return qio_channel_readv_full_all(ioc, iov, niov, NULL, NULL, errp);
118 }
119
120 int qio_channel_readv_full_all_eof(QIOChannel *ioc,
121 const struct iovec *iov,
122 size_t niov,
123 int **fds, size_t *nfds,
124 Error **errp)
125 {
126 int ret = -1;
127 struct iovec *local_iov = g_new(struct iovec, niov);
128 struct iovec *local_iov_head = local_iov;
129 unsigned int nlocal_iov = niov;
130 int **local_fds = fds;
131 size_t *local_nfds = nfds;
132 bool partial = false;
133
134 if (nfds) {
135 *nfds = 0;
136 }
137
138 if (fds) {
139 *fds = NULL;
140 }
141
142 nlocal_iov = iov_copy(local_iov, nlocal_iov,
143 iov, niov,
144 0, iov_size(iov, niov));
145
146 while ((nlocal_iov > 0) || local_fds) {
147 ssize_t len;
148 len = qio_channel_readv_full(ioc, local_iov, nlocal_iov, local_fds,
149 local_nfds, errp);
150 if (len == QIO_CHANNEL_ERR_BLOCK) {
151 if (qemu_in_coroutine()) {
152 qio_channel_yield(ioc, G_IO_IN);
153 } else {
154 qio_channel_wait(ioc, G_IO_IN);
155 }
156 continue;
157 }
158
159 if (len == 0) {
160 if (local_nfds && *local_nfds) {
161 /*
162 * Got some FDs, but no data yet. This isn't an EOF
163 * scenario (yet), so carry on to try to read data
164 * on next loop iteration
165 */
166 goto next_iter;
167 } else if (!partial) {
168 /* No fds and no data - EOF before any data read */
169 ret = 0;
170 goto cleanup;
171 } else {
172 len = -1;
173 error_setg(errp,
174 "Unexpected end-of-file before all data were read");
175 /* Fallthrough into len < 0 handling */
176 }
177 }
178
179 if (len < 0) {
180 /* Close any FDs we previously received */
181 if (nfds && fds) {
182 size_t i;
183 for (i = 0; i < (*nfds); i++) {
184 close((*fds)[i]);
185 }
186 g_free(*fds);
187 *fds = NULL;
188 *nfds = 0;
189 }
190 goto cleanup;
191 }
192
193 if (nlocal_iov) {
194 iov_discard_front(&local_iov, &nlocal_iov, len);
195 }
196
197 next_iter:
198 partial = true;
199 local_fds = NULL;
200 local_nfds = NULL;
201 }
202
203 ret = 1;
204
205 cleanup:
206 g_free(local_iov_head);
207 return ret;
208 }
209
210 int qio_channel_readv_full_all(QIOChannel *ioc,
211 const struct iovec *iov,
212 size_t niov,
213 int **fds, size_t *nfds,
214 Error **errp)
215 {
216 int ret = qio_channel_readv_full_all_eof(ioc, iov, niov, fds, nfds, errp);
217
218 if (ret == 0) {
219 error_setg(errp, "Unexpected end-of-file before all data were read");
220 return -1;
221 }
222 if (ret == 1) {
223 return 0;
224 }
225
226 return ret;
227 }
228
229 int qio_channel_writev_all(QIOChannel *ioc,
230 const struct iovec *iov,
231 size_t niov,
232 Error **errp)
233 {
234 return qio_channel_writev_full_all(ioc, iov, niov, NULL, 0, 0, errp);
235 }
236
237 int qio_channel_writev_full_all(QIOChannel *ioc,
238 const struct iovec *iov,
239 size_t niov,
240 int *fds, size_t nfds,
241 int flags, Error **errp)
242 {
243 int ret = -1;
244 struct iovec *local_iov = g_new(struct iovec, niov);
245 struct iovec *local_iov_head = local_iov;
246 unsigned int nlocal_iov = niov;
247
248 nlocal_iov = iov_copy(local_iov, nlocal_iov,
249 iov, niov,
250 0, iov_size(iov, niov));
251
252 while (nlocal_iov > 0) {
253 ssize_t len;
254
255 len = qio_channel_writev_full(ioc, local_iov, nlocal_iov, fds,
256 nfds, flags, errp);
257
258 if (len == QIO_CHANNEL_ERR_BLOCK) {
259 if (qemu_in_coroutine()) {
260 qio_channel_yield(ioc, G_IO_OUT);
261 } else {
262 qio_channel_wait(ioc, G_IO_OUT);
263 }
264 continue;
265 }
266 if (len < 0) {
267 goto cleanup;
268 }
269
270 iov_discard_front(&local_iov, &nlocal_iov, len);
271
272 fds = NULL;
273 nfds = 0;
274 }
275
276 ret = 0;
277 cleanup:
278 g_free(local_iov_head);
279 return ret;
280 }
281
282 ssize_t qio_channel_readv(QIOChannel *ioc,
283 const struct iovec *iov,
284 size_t niov,
285 Error **errp)
286 {
287 return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, errp);
288 }
289
290
291 ssize_t qio_channel_writev(QIOChannel *ioc,
292 const struct iovec *iov,
293 size_t niov,
294 Error **errp)
295 {
296 return qio_channel_writev_full(ioc, iov, niov, NULL, 0, 0, errp);
297 }
298
299
300 ssize_t qio_channel_read(QIOChannel *ioc,
301 char *buf,
302 size_t buflen,
303 Error **errp)
304 {
305 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
306 return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, errp);
307 }
308
309
310 ssize_t qio_channel_write(QIOChannel *ioc,
311 const char *buf,
312 size_t buflen,
313 Error **errp)
314 {
315 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
316 return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, 0, errp);
317 }
318
319
320 int qio_channel_read_all_eof(QIOChannel *ioc,
321 char *buf,
322 size_t buflen,
323 Error **errp)
324 {
325 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
326 return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
327 }
328
329
330 int qio_channel_read_all(QIOChannel *ioc,
331 char *buf,
332 size_t buflen,
333 Error **errp)
334 {
335 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
336 return qio_channel_readv_all(ioc, &iov, 1, errp);
337 }
338
339
340 int qio_channel_write_all(QIOChannel *ioc,
341 const char *buf,
342 size_t buflen,
343 Error **errp)
344 {
345 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
346 return qio_channel_writev_all(ioc, &iov, 1, errp);
347 }
348
349
350 int qio_channel_set_blocking(QIOChannel *ioc,
351 bool enabled,
352 Error **errp)
353 {
354 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
355 return klass->io_set_blocking(ioc, enabled, errp);
356 }
357
358
359 int qio_channel_close(QIOChannel *ioc,
360 Error **errp)
361 {
362 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
363 return klass->io_close(ioc, errp);
364 }
365
366
367 GSource *qio_channel_create_watch(QIOChannel *ioc,
368 GIOCondition condition)
369 {
370 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
371 GSource *ret = klass->io_create_watch(ioc, condition);
372
373 if (ioc->name) {
374 g_source_set_name(ret, ioc->name);
375 }
376
377 return ret;
378 }
379
380
381 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
382 AioContext *ctx,
383 IOHandler *io_read,
384 IOHandler *io_write,
385 void *opaque)
386 {
387 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
388
389 klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
390 }
391
392 guint qio_channel_add_watch_full(QIOChannel *ioc,
393 GIOCondition condition,
394 QIOChannelFunc func,
395 gpointer user_data,
396 GDestroyNotify notify,
397 GMainContext *context)
398 {
399 GSource *source;
400 guint id;
401
402 source = qio_channel_create_watch(ioc, condition);
403
404 g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
405
406 id = g_source_attach(source, context);
407 g_source_unref(source);
408
409 return id;
410 }
411
412 guint qio_channel_add_watch(QIOChannel *ioc,
413 GIOCondition condition,
414 QIOChannelFunc func,
415 gpointer user_data,
416 GDestroyNotify notify)
417 {
418 return qio_channel_add_watch_full(ioc, condition, func,
419 user_data, notify, NULL);
420 }
421
422 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
423 GIOCondition condition,
424 QIOChannelFunc func,
425 gpointer user_data,
426 GDestroyNotify notify,
427 GMainContext *context)
428 {
429 GSource *source;
430 guint id;
431
432 id = qio_channel_add_watch_full(ioc, condition, func,
433 user_data, notify, context);
434 source = g_main_context_find_source_by_id(context, id);
435 g_source_ref(source);
436 return source;
437 }
438
439
440 int qio_channel_shutdown(QIOChannel *ioc,
441 QIOChannelShutdown how,
442 Error **errp)
443 {
444 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
445
446 if (!klass->io_shutdown) {
447 error_setg(errp, "Data path shutdown not supported");
448 return -1;
449 }
450
451 return klass->io_shutdown(ioc, how, errp);
452 }
453
454
455 void qio_channel_set_delay(QIOChannel *ioc,
456 bool enabled)
457 {
458 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
459
460 if (klass->io_set_delay) {
461 klass->io_set_delay(ioc, enabled);
462 }
463 }
464
465
466 void qio_channel_set_cork(QIOChannel *ioc,
467 bool enabled)
468 {
469 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
470
471 if (klass->io_set_cork) {
472 klass->io_set_cork(ioc, enabled);
473 }
474 }
475
476
477 off_t qio_channel_io_seek(QIOChannel *ioc,
478 off_t offset,
479 int whence,
480 Error **errp)
481 {
482 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
483
484 if (!klass->io_seek) {
485 error_setg(errp, "Channel does not support random access");
486 return -1;
487 }
488
489 return klass->io_seek(ioc, offset, whence, errp);
490 }
491
492 int qio_channel_flush(QIOChannel *ioc,
493 Error **errp)
494 {
495 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
496
497 if (!klass->io_flush ||
498 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
499 return 0;
500 }
501
502 return klass->io_flush(ioc, errp);
503 }
504
505
506 static void qio_channel_restart_read(void *opaque)
507 {
508 QIOChannel *ioc = opaque;
509 Coroutine *co = ioc->read_coroutine;
510
511 /* Assert that aio_co_wake() reenters the coroutine directly */
512 assert(qemu_get_current_aio_context() ==
513 qemu_coroutine_get_aio_context(co));
514 aio_co_wake(co);
515 }
516
517 static void qio_channel_restart_write(void *opaque)
518 {
519 QIOChannel *ioc = opaque;
520 Coroutine *co = ioc->write_coroutine;
521
522 /* Assert that aio_co_wake() reenters the coroutine directly */
523 assert(qemu_get_current_aio_context() ==
524 qemu_coroutine_get_aio_context(co));
525 aio_co_wake(co);
526 }
527
528 static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
529 {
530 IOHandler *rd_handler = NULL, *wr_handler = NULL;
531 AioContext *ctx;
532
533 if (ioc->read_coroutine) {
534 rd_handler = qio_channel_restart_read;
535 }
536 if (ioc->write_coroutine) {
537 wr_handler = qio_channel_restart_write;
538 }
539
540 ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
541 qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
542 }
543
544 void qio_channel_attach_aio_context(QIOChannel *ioc,
545 AioContext *ctx)
546 {
547 assert(!ioc->read_coroutine);
548 assert(!ioc->write_coroutine);
549 ioc->ctx = ctx;
550 }
551
552 void qio_channel_detach_aio_context(QIOChannel *ioc)
553 {
554 ioc->read_coroutine = NULL;
555 ioc->write_coroutine = NULL;
556 qio_channel_set_aio_fd_handlers(ioc);
557 ioc->ctx = NULL;
558 }
559
560 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
561 GIOCondition condition)
562 {
563 assert(qemu_in_coroutine());
564 if (condition == G_IO_IN) {
565 assert(!ioc->read_coroutine);
566 ioc->read_coroutine = qemu_coroutine_self();
567 } else if (condition == G_IO_OUT) {
568 assert(!ioc->write_coroutine);
569 ioc->write_coroutine = qemu_coroutine_self();
570 } else {
571 abort();
572 }
573 qio_channel_set_aio_fd_handlers(ioc);
574 qemu_coroutine_yield();
575
576 /* Allow interrupting the operation by reentering the coroutine other than
577 * through the aio_fd_handlers. */
578 if (condition == G_IO_IN && ioc->read_coroutine) {
579 ioc->read_coroutine = NULL;
580 qio_channel_set_aio_fd_handlers(ioc);
581 } else if (condition == G_IO_OUT && ioc->write_coroutine) {
582 ioc->write_coroutine = NULL;
583 qio_channel_set_aio_fd_handlers(ioc);
584 }
585 }
586
587
588 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
589 GIOCondition condition,
590 gpointer opaque)
591 {
592 GMainLoop *loop = opaque;
593
594 g_main_loop_quit(loop);
595 return FALSE;
596 }
597
598
599 void qio_channel_wait(QIOChannel *ioc,
600 GIOCondition condition)
601 {
602 GMainContext *ctxt = g_main_context_new();
603 GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
604 GSource *source;
605
606 source = qio_channel_create_watch(ioc, condition);
607
608 g_source_set_callback(source,
609 (GSourceFunc)qio_channel_wait_complete,
610 loop,
611 NULL);
612
613 g_source_attach(source, ctxt);
614
615 g_main_loop_run(loop);
616
617 g_source_unref(source);
618 g_main_loop_unref(loop);
619 g_main_context_unref(ctxt);
620 }
621
622
623 static void qio_channel_finalize(Object *obj)
624 {
625 QIOChannel *ioc = QIO_CHANNEL(obj);
626
627 g_free(ioc->name);
628
629 #ifdef _WIN32
630 if (ioc->event) {
631 CloseHandle(ioc->event);
632 }
633 #endif
634 }
635
636 static const TypeInfo qio_channel_info = {
637 .parent = TYPE_OBJECT,
638 .name = TYPE_QIO_CHANNEL,
639 .instance_size = sizeof(QIOChannel),
640 .instance_finalize = qio_channel_finalize,
641 .abstract = true,
642 .class_size = sizeof(QIOChannelClass),
643 };
644
645
646 static void qio_channel_register_types(void)
647 {
648 type_register_static(&qio_channel_info);
649 }
650
651
652 type_init(qio_channel_register_types);