xref: /qemu/io/channel.c (revision 78f314cf) 
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 int qio_channel_close(QIOChannel *ioc,
369                       Error **errp)
370 {
371     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
372     return klass->io_close(ioc, errp);
373 }
374 
375 
376 GSource *qio_channel_create_watch(QIOChannel *ioc,
377                                   GIOCondition condition)
378 {
379     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
380     GSource *ret = klass->io_create_watch(ioc, condition);
381 
382     if (ioc->name) {
383         g_source_set_name(ret, ioc->name);
384     }
385 
386     return ret;
387 }
388 
389 
390 void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
391                                     AioContext *ctx,
392                                     IOHandler *io_read,
393                                     IOHandler *io_write,
394                                     void *opaque)
395 {
396     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
397 
398     klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
399 }
400 
401 guint qio_channel_add_watch_full(QIOChannel *ioc,
402                                  GIOCondition condition,
403                                  QIOChannelFunc func,
404                                  gpointer user_data,
405                                  GDestroyNotify notify,
406                                  GMainContext *context)
407 {
408     GSource *source;
409     guint id;
410 
411     source = qio_channel_create_watch(ioc, condition);
412 
413     g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
414 
415     id = g_source_attach(source, context);
416     g_source_unref(source);
417 
418     return id;
419 }
420 
421 guint qio_channel_add_watch(QIOChannel *ioc,
422                             GIOCondition condition,
423                             QIOChannelFunc func,
424                             gpointer user_data,
425                             GDestroyNotify notify)
426 {
427     return qio_channel_add_watch_full(ioc, condition, func,
428                                       user_data, notify, NULL);
429 }
430 
431 GSource *qio_channel_add_watch_source(QIOChannel *ioc,
432                                       GIOCondition condition,
433                                       QIOChannelFunc func,
434                                       gpointer user_data,
435                                       GDestroyNotify notify,
436                                       GMainContext *context)
437 {
438     GSource *source;
439     guint id;
440 
441     id = qio_channel_add_watch_full(ioc, condition, func,
442                                     user_data, notify, context);
443     source = g_main_context_find_source_by_id(context, id);
444     g_source_ref(source);
445     return source;
446 }
447 
448 
449 int qio_channel_shutdown(QIOChannel *ioc,
450                          QIOChannelShutdown how,
451                          Error **errp)
452 {
453     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
454 
455     if (!klass->io_shutdown) {
456         error_setg(errp, "Data path shutdown not supported");
457         return -1;
458     }
459 
460     return klass->io_shutdown(ioc, how, errp);
461 }
462 
463 
464 void qio_channel_set_delay(QIOChannel *ioc,
465                            bool enabled)
466 {
467     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
468 
469     if (klass->io_set_delay) {
470         klass->io_set_delay(ioc, enabled);
471     }
472 }
473 
474 
475 void qio_channel_set_cork(QIOChannel *ioc,
476                           bool enabled)
477 {
478     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
479 
480     if (klass->io_set_cork) {
481         klass->io_set_cork(ioc, enabled);
482     }
483 }
484 
485 
486 off_t qio_channel_io_seek(QIOChannel *ioc,
487                           off_t offset,
488                           int whence,
489                           Error **errp)
490 {
491     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
492 
493     if (!klass->io_seek) {
494         error_setg(errp, "Channel does not support random access");
495         return -1;
496     }
497 
498     return klass->io_seek(ioc, offset, whence, errp);
499 }
500 
501 int qio_channel_flush(QIOChannel *ioc,
502                                 Error **errp)
503 {
504     QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
505 
506     if (!klass->io_flush ||
507         !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_WRITE_ZERO_COPY)) {
508         return 0;
509     }
510 
511     return klass->io_flush(ioc, errp);
512 }
513 
514 
515 static void qio_channel_restart_read(void *opaque)
516 {
517     QIOChannel *ioc = opaque;
518     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
519 
520     if (!co) {
521         return;
522     }
523 
524     /* Assert that aio_co_wake() reenters the coroutine directly */
525     assert(qemu_get_current_aio_context() ==
526            qemu_coroutine_get_aio_context(co));
527     aio_co_wake(co);
528 }
529 
530 static void qio_channel_restart_write(void *opaque)
531 {
532     QIOChannel *ioc = opaque;
533     Coroutine *co = qatomic_xchg(&ioc->write_coroutine, NULL);
534 
535     if (!co) {
536         return;
537     }
538 
539     /* Assert that aio_co_wake() reenters the coroutine directly */
540     assert(qemu_get_current_aio_context() ==
541            qemu_coroutine_get_aio_context(co));
542     aio_co_wake(co);
543 }
544 
545 static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
546 {
547     IOHandler *rd_handler = NULL, *wr_handler = NULL;
548     AioContext *ctx;
549 
550     if (ioc->read_coroutine) {
551         rd_handler = qio_channel_restart_read;
552     }
553     if (ioc->write_coroutine) {
554         wr_handler = qio_channel_restart_write;
555     }
556 
557     ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
558     qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
559 }
560 
561 void qio_channel_attach_aio_context(QIOChannel *ioc,
562                                     AioContext *ctx)
563 {
564     assert(!ioc->read_coroutine);
565     assert(!ioc->write_coroutine);
566     ioc->ctx = ctx;
567 }
568 
569 void qio_channel_detach_aio_context(QIOChannel *ioc)
570 {
571     ioc->read_coroutine = NULL;
572     ioc->write_coroutine = NULL;
573     qio_channel_set_aio_fd_handlers(ioc);
574     ioc->ctx = NULL;
575 }
576 
577 void coroutine_fn qio_channel_yield(QIOChannel *ioc,
578                                     GIOCondition condition)
579 {
580     AioContext *ioc_ctx = ioc->ctx ?: qemu_get_aio_context();
581 
582     assert(qemu_in_coroutine());
583     assert(in_aio_context_home_thread(ioc_ctx));
584 
585     if (condition == G_IO_IN) {
586         assert(!ioc->read_coroutine);
587         ioc->read_coroutine = qemu_coroutine_self();
588     } else if (condition == G_IO_OUT) {
589         assert(!ioc->write_coroutine);
590         ioc->write_coroutine = qemu_coroutine_self();
591     } else {
592         abort();
593     }
594     qio_channel_set_aio_fd_handlers(ioc);
595     qemu_coroutine_yield();
596     assert(in_aio_context_home_thread(ioc_ctx));
597 
598     /* Allow interrupting the operation by reentering the coroutine other than
599      * through the aio_fd_handlers. */
600     if (condition == G_IO_IN) {
601         assert(ioc->read_coroutine == NULL);
602         qio_channel_set_aio_fd_handlers(ioc);
603     } else if (condition == G_IO_OUT) {
604         assert(ioc->write_coroutine == NULL);
605         qio_channel_set_aio_fd_handlers(ioc);
606     }
607 }
608 
609 void qio_channel_wake_read(QIOChannel *ioc)
610 {
611     Coroutine *co = qatomic_xchg(&ioc->read_coroutine, NULL);
612     if (co) {
613         aio_co_wake(co);
614     }
615 }
616 
617 static gboolean qio_channel_wait_complete(QIOChannel *ioc,
618                                           GIOCondition condition,
619                                           gpointer opaque)
620 {
621     GMainLoop *loop = opaque;
622 
623     g_main_loop_quit(loop);
624     return FALSE;
625 }
626 
627 
628 void qio_channel_wait(QIOChannel *ioc,
629                       GIOCondition condition)
630 {
631     GMainContext *ctxt = g_main_context_new();
632     GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
633     GSource *source;
634 
635     source = qio_channel_create_watch(ioc, condition);
636 
637     g_source_set_callback(source,
638                           (GSourceFunc)qio_channel_wait_complete,
639                           loop,
640                           NULL);
641 
642     g_source_attach(source, ctxt);
643 
644     g_main_loop_run(loop);
645 
646     g_source_unref(source);
647     g_main_loop_unref(loop);
648     g_main_context_unref(ctxt);
649 }
650 
651 
652 static void qio_channel_finalize(Object *obj)
653 {
654     QIOChannel *ioc = QIO_CHANNEL(obj);
655 
656     g_free(ioc->name);
657 
658 #ifdef _WIN32
659     if (ioc->event) {
660         CloseHandle(ioc->event);
661     }
662 #endif
663 }
664 
665 static const TypeInfo qio_channel_info = {
666     .parent = TYPE_OBJECT,
667     .name = TYPE_QIO_CHANNEL,
668     .instance_size = sizeof(QIOChannel),
669     .instance_finalize = qio_channel_finalize,
670     .abstract = true,
671     .class_size = sizeof(QIOChannelClass),
672 };
673 
674 
675 static void qio_channel_register_types(void)
676 {
677     type_register_static(&qio_channel_info);
678 }
679 
680 
681 type_init(qio_channel_register_types);
682