xref: /openbmc/qemu/io/channel.c (revision ed75658a)
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 "block/aio-wait.h"
23 #include "io/channel.h"
24 #include "qapi/error.h"
25 #include "qemu/main-loop.h"
26 #include "qemu/module.h"
27 #include "qemu/iov.h"
28 
29 bool qio_channel_has_feature(QIOChannel *ioc,
30                              QIOChannelFeature feature)
31 {
32     return ioc->features & (1 << feature);
33 }
34 
35 
36 void qio_channel_set_feature(QIOChannel *ioc,
37                              QIOChannelFeature feature)
38 {
39     ioc->features |= (1 << feature);
40 }
41 
42 
43 void qio_channel_set_name(QIOChannel *ioc,
44                           const char *name)
45 {
46     g_free(ioc->name);
47     ioc->name = g_strdup(name);
48 }
49 
50 
51 ssize_t qio_channel_readv_full(QIOChannel *ioc,
52                                const struct iovec *iov,
53                                size_t niov,
54                                int **fds,
55                                size_t *nfds,
56                                int flags,
57                                Error **errp)
58 {
59     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
60 
61     if ((fds || nfds) &&
62         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
63         error_setg_errno(errp, EINVAL,
64                          "Channel does not support file descriptor passing");
65         return -1;
66     }
67 
68     if ((flags & QIO_CHANNEL_READ_FLAG_MSG_PEEK) &&
69         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_READ_MSG_PEEK)) {
70         error_setg_errno(errp, EINVAL,
71                          "Channel does not support peek read");
72         return -1;
73     }
74 
75     return klass->io_readv(ioc, iov, niov, fds, nfds, flags, errp);
76 }
77 
78 
79 ssize_t qio_channel_writev_full(QIOChannel *ioc,
80                                 const struct iovec *iov,
81                                 size_t niov,
82                                 int *fds,
83                                 size_t nfds,
84                                 int flags,
85                                 Error **errp)
86 {
87     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
88 
89     if (fds || nfds) {
90         if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
91             error_setg_errno(errp, EINVAL,
92                              "Channel does not support file descriptor passing");
93             return -1;
94         }
95         if (flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) {
96             error_setg_errno(errp, EINVAL,
97                              "Zero Copy does not support file descriptor passing");
98             return -1;
99         }
100     }
101 
102     if ((flags & QIO_CHANNEL_WRITE_FLAG_ZERO_COPY) &&
103         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
104         error_setg_errno(errp, EINVAL,
105                          "Requested Zero Copy feature is not available");
106         return -1;
107     }
108 
109     return klass->io_writev(ioc, iov, niov, fds, nfds, flags, errp);
110 }
111 
112 
113 int coroutine_mixed_fn qio_channel_readv_all_eof(QIOChannel *ioc,
114                                                  const struct iovec *iov,
115                                                  size_t niov,
116                                                  Error **errp)
117 {
118     return qio_channel_readv_full_all_eof(ioc, iov, niov, NULL, NULL, errp);
119 }
120 
121 int coroutine_mixed_fn qio_channel_readv_all(QIOChannel *ioc,
122                                              const struct iovec *iov,
123                                              size_t niov,
124                                              Error **errp)
125 {
126     return qio_channel_readv_full_all(ioc, iov, niov, NULL, NULL, errp);
127 }
128 
129 int coroutine_mixed_fn qio_channel_readv_full_all_eof(QIOChannel *ioc,
130                                                       const struct iovec *iov,
131                                                       size_t niov,
132                                                       int **fds, size_t *nfds,
133                                                       Error **errp)
134 {
135     int ret = -1;
136     struct iovec *local_iov = g_new(struct iovec, niov);
137     struct iovec *local_iov_head = local_iov;
138     unsigned int nlocal_iov = niov;
139     int **local_fds = fds;
140     size_t *local_nfds = nfds;
141     bool partial = false;
142 
143     if (nfds) {
144         *nfds = 0;
145     }
146 
147     if (fds) {
148         *fds = NULL;
149     }
150 
151     nlocal_iov = iov_copy(local_iov, nlocal_iov,
152                           iov, niov,
153                           0, iov_size(iov, niov));
154 
155     while ((nlocal_iov > 0) || local_fds) {
156         ssize_t len;
157         len = qio_channel_readv_full(ioc, local_iov, nlocal_iov, local_fds,
158                                      local_nfds, 0, errp);
159         if (len == QIO_CHANNEL_ERR_BLOCK) {
160             if (qemu_in_coroutine()) {
161                 qio_channel_yield(ioc, G_IO_IN);
162             } else {
163                 qio_channel_wait(ioc, G_IO_IN);
164             }
165             continue;
166         }
167 
168         if (len == 0) {
169             if (local_nfds && *local_nfds) {
170                 /*
171                  * Got some FDs, but no data yet. This isn't an EOF
172                  * scenario (yet), so carry on to try to read data
173                  * on next loop iteration
174                  */
175                 goto next_iter;
176             } else if (!partial) {
177                 /* No fds and no data - EOF before any data read */
178                 ret = 0;
179                 goto cleanup;
180             } else {
181                 len = -1;
182                 error_setg(errp,
183                            "Unexpected end-of-file before all data were read");
184                 /* Fallthrough into len < 0 handling */
185             }
186         }
187 
188         if (len < 0) {
189             /* Close any FDs we previously received */
190             if (nfds && fds) {
191                 size_t i;
192                 for (i = 0; i < (*nfds); i++) {
193                     close((*fds)[i]);
194                 }
195                 g_free(*fds);
196                 *fds = NULL;
197                 *nfds = 0;
198             }
199             goto cleanup;
200         }
201 
202         if (nlocal_iov) {
203             iov_discard_front(&local_iov, &nlocal_iov, len);
204         }
205 
206 next_iter:
207         partial = true;
208         local_fds = NULL;
209         local_nfds = NULL;
210     }
211 
212     ret = 1;
213 
214  cleanup:
215     g_free(local_iov_head);
216     return ret;
217 }
218 
219 int coroutine_mixed_fn qio_channel_readv_full_all(QIOChannel *ioc,
220                                                   const struct iovec *iov,
221                                                   size_t niov,
222                                                   int **fds, size_t *nfds,
223                                                   Error **errp)
224 {
225     int ret = qio_channel_readv_full_all_eof(ioc, iov, niov, fds, nfds, errp);
226 
227     if (ret == 0) {
228         error_setg(errp, "Unexpected end-of-file before all data were read");
229         return -1;
230     }
231     if (ret == 1) {
232         return 0;
233     }
234 
235     return ret;
236 }
237 
238 int coroutine_mixed_fn qio_channel_writev_all(QIOChannel *ioc,
239                                               const struct iovec *iov,
240                                               size_t niov,
241                                               Error **errp)
242 {
243     return qio_channel_writev_full_all(ioc, iov, niov, NULL, 0, 0, errp);
244 }
245 
246 int coroutine_mixed_fn qio_channel_writev_full_all(QIOChannel *ioc,
247                                                    const struct iovec *iov,
248                                                    size_t niov,
249                                                    int *fds, size_t nfds,
250                                                    int flags, Error **errp)
251 {
252     int ret = -1;
253     struct iovec *local_iov = g_new(struct iovec, niov);
254     struct iovec *local_iov_head = local_iov;
255     unsigned int nlocal_iov = niov;
256 
257     nlocal_iov = iov_copy(local_iov, nlocal_iov,
258                           iov, niov,
259                           0, iov_size(iov, niov));
260 
261     while (nlocal_iov > 0) {
262         ssize_t len;
263 
264         len = qio_channel_writev_full(ioc, local_iov, nlocal_iov, fds,
265                                             nfds, flags, errp);
266 
267         if (len == QIO_CHANNEL_ERR_BLOCK) {
268             if (qemu_in_coroutine()) {
269                 qio_channel_yield(ioc, G_IO_OUT);
270             } else {
271                 qio_channel_wait(ioc, G_IO_OUT);
272             }
273             continue;
274         }
275         if (len < 0) {
276             goto cleanup;
277         }
278 
279         iov_discard_front(&local_iov, &nlocal_iov, len);
280 
281         fds = NULL;
282         nfds = 0;
283     }
284 
285     ret = 0;
286  cleanup:
287     g_free(local_iov_head);
288     return ret;
289 }
290 
291 ssize_t qio_channel_readv(QIOChannel *ioc,
292                           const struct iovec *iov,
293                           size_t niov,
294                           Error **errp)
295 {
296     return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, 0, errp);
297 }
298 
299 
300 ssize_t qio_channel_writev(QIOChannel *ioc,
301                            const struct iovec *iov,
302                            size_t niov,
303                            Error **errp)
304 {
305     return qio_channel_writev_full(ioc, iov, niov, NULL, 0, 0, errp);
306 }
307 
308 
309 ssize_t qio_channel_read(QIOChannel *ioc,
310                          char *buf,
311                          size_t buflen,
312                          Error **errp)
313 {
314     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
315     return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, 0, errp);
316 }
317 
318 
319 ssize_t qio_channel_write(QIOChannel *ioc,
320                           const char *buf,
321                           size_t buflen,
322                           Error **errp)
323 {
324     struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
325     return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, 0, errp);
326 }
327 
328 
329 int coroutine_mixed_fn qio_channel_read_all_eof(QIOChannel *ioc,
330                                                 char *buf,
331                                                 size_t buflen,
332                                                 Error **errp)
333 {
334     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
335     return qio_channel_readv_all_eof(ioc, &iov, 1, errp);
336 }
337 
338 
339 int coroutine_mixed_fn qio_channel_read_all(QIOChannel *ioc,
340                                             char *buf,
341                                             size_t buflen,
342                                             Error **errp)
343 {
344     struct iovec iov = { .iov_base = buf, .iov_len = buflen };
345     return qio_channel_readv_all(ioc, &iov, 1, errp);
346 }
347 
348 
349 int coroutine_mixed_fn qio_channel_write_all(QIOChannel *ioc,
350                                              const char *buf,
351                                              size_t buflen,
352                                              Error **errp)
353 {
354     struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
355     return qio_channel_writev_all(ioc, &iov, 1, errp);
356 }
357 
358 
359 int qio_channel_set_blocking(QIOChannel *ioc,
360                               bool enabled,
361                               Error **errp)
362 {
363     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
364     return klass->io_set_blocking(ioc, enabled, errp);
365 }
366 
367 
368 void qio_channel_set_follow_coroutine_ctx(QIOChannel *ioc, bool enabled)
369 {
370     ioc->follow_coroutine_ctx = enabled;
371 }
372 
373 
374 int qio_channel_close(QIOChannel *ioc,
375                       Error **errp)
376 {
377     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
378     return klass->io_close(ioc, errp);
379 }
380 
381 
382 GSource *qio_channel_create_watch(QIOChannel *ioc,
383                                   GIOCondition condition)
384 {
385     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
386     GSource *ret = klass->io_create_watch(ioc, condition);
387 
388     if (ioc->name) {
389         g_source_set_name(ret, ioc->name);
390     }
391 
392     return ret;
393 }
394 
395 
396 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
397                                     AioContext *read_ctx,
398                                     IOHandler *io_read,
399                                     AioContext *write_ctx,
400                                     IOHandler *io_write,
401                                     void *opaque)
402 {
403     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
404 
405     klass->io_set_aio_fd_handler(ioc, read_ctx, io_read, write_ctx, io_write,
406             opaque);
407 }
408 
409 guint qio_channel_add_watch_full(QIOChannel *ioc,
410                                  GIOCondition condition,
411                                  QIOChannelFunc func,
412                                  gpointer user_data,
413                                  GDestroyNotify notify,
414                                  GMainContext *context)
415 {
416     GSource *source;
417     guint id;
418 
419     source = qio_channel_create_watch(ioc, condition);
420 
421     g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
422 
423     id = g_source_attach(source, context);
424     g_source_unref(source);
425 
426     return id;
427 }
428 
429 guint qio_channel_add_watch(QIOChannel *ioc,
430                             GIOCondition condition,
431                             QIOChannelFunc func,
432                             gpointer user_data,
433                             GDestroyNotify notify)
434 {
435     return qio_channel_add_watch_full(ioc, condition, func,
436                                       user_data, notify, NULL);
437 }
438 
439 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
440                                       GIOCondition condition,
441                                       QIOChannelFunc func,
442                                       gpointer user_data,
443                                       GDestroyNotify notify,
444                                       GMainContext *context)
445 {
446     GSource *source;
447     guint id;
448 
449     id = qio_channel_add_watch_full(ioc, condition, func,
450                                     user_data, notify, context);
451     source = g_main_context_find_source_by_id(context, id);
452     g_source_ref(source);
453     return source;
454 }
455 
456 
457 ssize_t qio_channel_pwritev(QIOChannel *ioc, const struct iovec *iov,
458                             size_t niov, off_t offset, Error **errp)
459 {
460     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
461 
462     if (!klass->io_pwritev) {
463         error_setg(errp, "Channel does not support pwritev");
464         return -1;
465     }
466 
467     if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_SEEKABLE)) {
468         error_setg_errno(errp, EINVAL, "Requested channel is not seekable");
469         return -1;
470     }
471 
472     return klass->io_pwritev(ioc, iov, niov, offset, errp);
473 }
474 
475 ssize_t qio_channel_pwrite(QIOChannel *ioc, char *buf, size_t buflen,
476                            off_t offset, Error **errp)
477 {
478     struct iovec iov = {
479         .iov_base = buf,
480         .iov_len = buflen
481     };
482 
483     return qio_channel_pwritev(ioc, &iov, 1, offset, errp);
484 }
485 
486 ssize_t qio_channel_preadv(QIOChannel *ioc, const struct iovec *iov,
487                            size_t niov, off_t offset, Error **errp)
488 {
489     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
490 
491     if (!klass->io_preadv) {
492         error_setg(errp, "Channel does not support preadv");
493         return -1;
494     }
495 
496     if (!qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_SEEKABLE)) {
497         error_setg_errno(errp, EINVAL, "Requested channel is not seekable");
498         return -1;
499     }
500 
501     return klass->io_preadv(ioc, iov, niov, offset, errp);
502 }
503 
504 ssize_t qio_channel_pread(QIOChannel *ioc, char *buf, size_t buflen,
505                           off_t offset, Error **errp)
506 {
507     struct iovec iov = {
508         .iov_base = buf,
509         .iov_len = buflen
510     };
511 
512     return qio_channel_preadv(ioc, &iov, 1, offset, errp);
513 }
514 
515 int qio_channel_shutdown(QIOChannel *ioc,
516                          QIOChannelShutdown how,
517                          Error **errp)
518 {
519     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
520 
521     if (!klass->io_shutdown) {
522         error_setg(errp, "Data path shutdown not supported");
523         return -1;
524     }
525 
526     return klass->io_shutdown(ioc, how, errp);
527 }
528 
529 
530 void qio_channel_set_delay(QIOChannel *ioc,
531                            bool enabled)
532 {
533     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
534 
535     if (klass->io_set_delay) {
536         klass->io_set_delay(ioc, enabled);
537     }
538 }
539 
540 
541 void qio_channel_set_cork(QIOChannel *ioc,
542                           bool enabled)
543 {
544     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
545 
546     if (klass->io_set_cork) {
547         klass->io_set_cork(ioc, enabled);
548     }
549 }
550 
551 
552 off_t qio_channel_io_seek(QIOChannel *ioc,
553                           off_t offset,
554                           int whence,
555                           Error **errp)
556 {
557     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
558 
559     if (!klass->io_seek) {
560         error_setg(errp, "Channel does not support random access");
561         return -1;
562     }
563 
564     return klass->io_seek(ioc, offset, whence, errp);
565 }
566 
567 int qio_channel_flush(QIOChannel *ioc,
568                                 Error **errp)
569 {
570     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
571 
572     if (!klass->io_flush ||
573         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
574         return 0;
575     }
576 
577     return klass->io_flush(ioc, errp);
578 }
579 
580 
581 static void qio_channel_restart_read(void *opaque)
582 {
583     QIOChannel *ioc = opaque;
584     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
585 
586     if (!co) {
587         return;
588     }
589 
590     /* Assert that aio_co_wake() reenters the coroutine directly */
591     assert(qemu_get_current_aio_context() ==
592            qemu_coroutine_get_aio_context(co));
593     aio_co_wake(co);
594 }
595 
596 static void qio_channel_restart_write(void *opaque)
597 {
598     QIOChannel *ioc = opaque;
599     Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
600 
601     if (!co) {
602         return;
603     }
604 
605     /* Assert that aio_co_wake() reenters the coroutine directly */
606     assert(qemu_get_current_aio_context() ==
607            qemu_coroutine_get_aio_context(co));
608     aio_co_wake(co);
609 }
610 
611 static void coroutine_fn
612 qio_channel_set_fd_handlers(QIOChannel *ioc, GIOCondition condition)
613 {
614     AioContext *ctx = ioc->follow_coroutine_ctx ?
615         qemu_coroutine_get_aio_context(qemu_coroutine_self()) :
616         iohandler_get_aio_context();
617     AioContext *read_ctx = NULL;
618     IOHandler *io_read = NULL;
619     AioContext *write_ctx = NULL;
620     IOHandler *io_write = NULL;
621 
622     if (condition == G_IO_IN) {
623         ioc->read_coroutine = qemu_coroutine_self();
624         ioc->read_ctx = ctx;
625         read_ctx = ctx;
626         io_read = qio_channel_restart_read;
627 
628         /*
629          * Thread safety: if the other coroutine is set and its AioContext
630          * matches ours, then there is mutual exclusion between read and write
631          * because they share a single thread and it's safe to set both read
632          * and write fd handlers here. If the AioContext does not match ours,
633          * then both threads may run in parallel but there is no shared state
634          * to worry about.
635          */
636         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
637             write_ctx = ctx;
638             io_write = qio_channel_restart_write;
639         }
640     } else if (condition == G_IO_OUT) {
641         ioc->write_coroutine = qemu_coroutine_self();
642         ioc->write_ctx = ctx;
643         write_ctx = ctx;
644         io_write = qio_channel_restart_write;
645         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
646             read_ctx = ctx;
647             io_read = qio_channel_restart_read;
648         }
649     } else {
650         abort();
651     }
652 
653     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
654             write_ctx, io_write, ioc);
655 }
656 
657 static void coroutine_fn
658 qio_channel_clear_fd_handlers(QIOChannel *ioc, GIOCondition condition)
659 {
660     AioContext *read_ctx = NULL;
661     IOHandler *io_read = NULL;
662     AioContext *write_ctx = NULL;
663     IOHandler *io_write = NULL;
664     AioContext *ctx;
665 
666     if (condition == G_IO_IN) {
667         ctx = ioc->read_ctx;
668         read_ctx = ctx;
669         io_read = NULL;
670         if (ioc->write_coroutine && ioc->write_ctx == ctx) {
671             write_ctx = ctx;
672             io_write = qio_channel_restart_write;
673         }
674     } else if (condition == G_IO_OUT) {
675         ctx = ioc->write_ctx;
676         write_ctx = ctx;
677         io_write = NULL;
678         if (ioc->read_coroutine && ioc->read_ctx == ctx) {
679             read_ctx = ctx;
680             io_read = qio_channel_restart_read;
681         }
682     } else {
683         abort();
684     }
685 
686     qio_channel_set_aio_fd_handler(ioc, read_ctx, io_read,
687             write_ctx, io_write, ioc);
688 }
689 
690 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
691                                     GIOCondition condition)
692 {
693     AioContext *ioc_ctx;
694 
695     assert(qemu_in_coroutine());
696     ioc_ctx = qemu_coroutine_get_aio_context(qemu_coroutine_self());
697 
698     if (condition == G_IO_IN) {
699         assert(!ioc->read_coroutine);
700     } else if (condition == G_IO_OUT) {
701         assert(!ioc->write_coroutine);
702     } else {
703         abort();
704     }
705     qio_channel_set_fd_handlers(ioc, condition);
706     qemu_coroutine_yield();
707     assert(in_aio_context_home_thread(ioc_ctx));
708 
709     /* Allow interrupting the operation by reentering the coroutine other than
710      * through the aio_fd_handlers. */
711     if (condition == G_IO_IN) {
712         assert(ioc->read_coroutine == NULL);
713     } else if (condition == G_IO_OUT) {
714         assert(ioc->write_coroutine == NULL);
715     }
716     qio_channel_clear_fd_handlers(ioc, condition);
717 }
718 
719 void qio_channel_wake_read(QIOChannel *ioc)
720 {
721     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
722     if (co) {
723         aio_co_wake(co);
724     }
725 }
726 
727 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
728                                           GIOCondition condition,
729                                           gpointer opaque)
730 {
731     GMainLoop *loop = opaque;
732 
733     g_main_loop_quit(loop);
734     return FALSE;
735 }
736 
737 
738 void qio_channel_wait(QIOChannel *ioc,
739                       GIOCondition condition)
740 {
741     GMainContext *ctxt = g_main_context_new();
742     GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
743     GSource *source;
744 
745     source = qio_channel_create_watch(ioc, condition);
746 
747     g_source_set_callback(source,
748                           (GSourceFunc)qio_channel_wait_complete,
749                           loop,
750                           NULL);
751 
752     g_source_attach(source, ctxt);
753 
754     g_main_loop_run(loop);
755 
756     g_source_unref(source);
757     g_main_loop_unref(loop);
758     g_main_context_unref(ctxt);
759 }
760 
761 
762 static void qio_channel_finalize(Object *obj)
763 {
764     QIOChannel *ioc = QIO_CHANNEL(obj);
765 
766     /* Must not have coroutines in qio_channel_yield() */
767     assert(!ioc->read_coroutine);
768     assert(!ioc->write_coroutine);
769 
770     g_free(ioc->name);
771 
772 #ifdef _WIN32
773     if (ioc->event) {
774         CloseHandle(ioc->event);
775     }
776 #endif
777 }
778 
779 static const TypeInfo qio_channel_info = {
780     .parent = TYPE_OBJECT,
781     .name = TYPE_QIO_CHANNEL,
782     .instance_size = sizeof(QIOChannel),
783     .instance_finalize = qio_channel_finalize,
784     .abstract = true,
785     .class_size = sizeof(QIOChannelClass),
786 };
787 
788 
789 static void qio_channel_register_types(void)
790 {
791     type_register_static(&qio_channel_info);
792 }
793 
794 
795 type_init(qio_channel_register_types);
796