1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
26bool qio_channel_has_feature(QIOChannel *ioc,
27 QIOChannelFeature feature)
28{
29 return ioc->features & (1 << feature);
30}
31
32
33void qio_channel_set_feature(QIOChannel *ioc,
34 QIOChannelFeature feature)
35{
36 ioc->features |= (1 << feature);
37}
38
39
40void qio_channel_set_name(QIOChannel *ioc,
41 const char *name)
42{
43 g_free(ioc->name);
44 ioc->name = g_strdup(name);
45}
46
47
48ssize_t qio_channel_readv_full(QIOChannel *ioc,
49 const struct iovec *iov,
50 size_t niov,
51 int **fds,
52 size_t *nfds,
53 Error **errp)
54{
55 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
56
57 if ((fds || nfds) &&
58 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
59 error_setg_errno(errp, EINVAL,
60 "Channel does not support file descriptor passing");
61 return -1;
62 }
63
64 return klass->io_readv(ioc, iov, niov, fds, nfds, errp);
65}
66
67
68ssize_t qio_channel_writev_full(QIOChannel *ioc,
69 const struct iovec *iov,
70 size_t niov,
71 int *fds,
72 size_t nfds,
73 Error **errp)
74{
75 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
76
77 if ((fds || nfds) &&
78 !qio_channel_has_feature(ioc, QIO_CHANNEL_FEATURE_FD_PASS)) {
79 error_setg_errno(errp, EINVAL,
80 "Channel does not support file descriptor passing");
81 return -1;
82 }
83
84 return klass->io_writev(ioc, iov, niov, fds, nfds, errp);
85}
86
87
88ssize_t qio_channel_readv(QIOChannel *ioc,
89 const struct iovec *iov,
90 size_t niov,
91 Error **errp)
92{
93 return qio_channel_readv_full(ioc, iov, niov, NULL, NULL, errp);
94}
95
96
97ssize_t qio_channel_writev(QIOChannel *ioc,
98 const struct iovec *iov,
99 size_t niov,
100 Error **errp)
101{
102 return qio_channel_writev_full(ioc, iov, niov, NULL, 0, errp);
103}
104
105
106ssize_t qio_channel_read(QIOChannel *ioc,
107 char *buf,
108 size_t buflen,
109 Error **errp)
110{
111 struct iovec iov = { .iov_base = buf, .iov_len = buflen };
112 return qio_channel_readv_full(ioc, &iov, 1, NULL, NULL, errp);
113}
114
115
116ssize_t qio_channel_write(QIOChannel *ioc,
117 const char *buf,
118 size_t buflen,
119 Error **errp)
120{
121 struct iovec iov = { .iov_base = (char *)buf, .iov_len = buflen };
122 return qio_channel_writev_full(ioc, &iov, 1, NULL, 0, errp);
123}
124
125
126int qio_channel_set_blocking(QIOChannel *ioc,
127 bool enabled,
128 Error **errp)
129{
130 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
131 return klass->io_set_blocking(ioc, enabled, errp);
132}
133
134
135int qio_channel_close(QIOChannel *ioc,
136 Error **errp)
137{
138 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
139 return klass->io_close(ioc, errp);
140}
141
142
143GSource *qio_channel_create_watch(QIOChannel *ioc,
144 GIOCondition condition)
145{
146 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
147 GSource *ret = klass->io_create_watch(ioc, condition);
148
149 if (ioc->name) {
150 g_source_set_name(ret, ioc->name);
151 }
152
153 return ret;
154}
155
156
157void qio_channel_set_aio_fd_handler(QIOChannel *ioc,
158 AioContext *ctx,
159 IOHandler *io_read,
160 IOHandler *io_write,
161 void *opaque)
162{
163 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
164
165 klass->io_set_aio_fd_handler(ioc, ctx, io_read, io_write, opaque);
166}
167
168guint qio_channel_add_watch(QIOChannel *ioc,
169 GIOCondition condition,
170 QIOChannelFunc func,
171 gpointer user_data,
172 GDestroyNotify notify)
173{
174 GSource *source;
175 guint id;
176
177 source = qio_channel_create_watch(ioc, condition);
178
179 g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
180
181 id = g_source_attach(source, NULL);
182 g_source_unref(source);
183
184 return id;
185}
186
187
188int qio_channel_shutdown(QIOChannel *ioc,
189 QIOChannelShutdown how,
190 Error **errp)
191{
192 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
193
194 if (!klass->io_shutdown) {
195 error_setg(errp, "Data path shutdown not supported");
196 return -1;
197 }
198
199 return klass->io_shutdown(ioc, how, errp);
200}
201
202
203void qio_channel_set_delay(QIOChannel *ioc,
204 bool enabled)
205{
206 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
207
208 if (klass->io_set_delay) {
209 klass->io_set_delay(ioc, enabled);
210 }
211}
212
213
214void qio_channel_set_cork(QIOChannel *ioc,
215 bool enabled)
216{
217 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
218
219 if (klass->io_set_cork) {
220 klass->io_set_cork(ioc, enabled);
221 }
222}
223
224
225off_t qio_channel_io_seek(QIOChannel *ioc,
226 off_t offset,
227 int whence,
228 Error **errp)
229{
230 QIOChannelClass *klass = QIO_CHANNEL_GET_CLASS(ioc);
231
232 if (!klass->io_seek) {
233 error_setg(errp, "Channel does not support random access");
234 return -1;
235 }
236
237 return klass->io_seek(ioc, offset, whence, errp);
238}
239
240
241static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc);
242
243static void qio_channel_restart_read(void *opaque)
244{
245 QIOChannel *ioc = opaque;
246 Coroutine *co = ioc->read_coroutine;
247
248 ioc->read_coroutine = NULL;
249 qio_channel_set_aio_fd_handlers(ioc);
250 aio_co_wake(co);
251}
252
253static void qio_channel_restart_write(void *opaque)
254{
255 QIOChannel *ioc = opaque;
256 Coroutine *co = ioc->write_coroutine;
257
258 ioc->write_coroutine = NULL;
259 qio_channel_set_aio_fd_handlers(ioc);
260 aio_co_wake(co);
261}
262
263static void qio_channel_set_aio_fd_handlers(QIOChannel *ioc)
264{
265 IOHandler *rd_handler = NULL, *wr_handler = NULL;
266 AioContext *ctx;
267
268 if (ioc->read_coroutine) {
269 rd_handler = qio_channel_restart_read;
270 }
271 if (ioc->write_coroutine) {
272 wr_handler = qio_channel_restart_write;
273 }
274
275 ctx = ioc->ctx ? ioc->ctx : iohandler_get_aio_context();
276 qio_channel_set_aio_fd_handler(ioc, ctx, rd_handler, wr_handler, ioc);
277}
278
279void qio_channel_attach_aio_context(QIOChannel *ioc,
280 AioContext *ctx)
281{
282 assert(!ioc->read_coroutine);
283 assert(!ioc->write_coroutine);
284 ioc->ctx = ctx;
285}
286
287void qio_channel_detach_aio_context(QIOChannel *ioc)
288{
289 ioc->read_coroutine = NULL;
290 ioc->write_coroutine = NULL;
291 qio_channel_set_aio_fd_handlers(ioc);
292 ioc->ctx = NULL;
293}
294
295void coroutine_fn qio_channel_yield(QIOChannel *ioc,
296 GIOCondition condition)
297{
298 assert(qemu_in_coroutine());
299 if (condition == G_IO_IN) {
300 assert(!ioc->read_coroutine);
301 ioc->read_coroutine = qemu_coroutine_self();
302 } else if (condition == G_IO_OUT) {
303 assert(!ioc->write_coroutine);
304 ioc->write_coroutine = qemu_coroutine_self();
305 } else {
306 abort();
307 }
308 qio_channel_set_aio_fd_handlers(ioc);
309 qemu_coroutine_yield();
310}
311
312
313static gboolean qio_channel_wait_complete(QIOChannel *ioc,
314 GIOCondition condition,
315 gpointer opaque)
316{
317 GMainLoop *loop = opaque;
318
319 g_main_loop_quit(loop);
320 return FALSE;
321}
322
323
324void qio_channel_wait(QIOChannel *ioc,
325 GIOCondition condition)
326{
327 GMainContext *ctxt = g_main_context_new();
328 GMainLoop *loop = g_main_loop_new(ctxt, TRUE);
329 GSource *source;
330
331 source = qio_channel_create_watch(ioc, condition);
332
333 g_source_set_callback(source,
334 (GSourceFunc)qio_channel_wait_complete,
335 loop,
336 NULL);
337
338 g_source_attach(source, ctxt);
339
340 g_main_loop_run(loop);
341
342 g_source_unref(source);
343 g_main_loop_unref(loop);
344 g_main_context_unref(ctxt);
345}
346
347
348static void qio_channel_finalize(Object *obj)
349{
350 QIOChannel *ioc = QIO_CHANNEL(obj);
351
352 g_free(ioc->name);
353
354#ifdef _WIN32
355 if (ioc->event) {
356 CloseHandle(ioc->event);
357 }
358#endif
359}
360
361static const TypeInfo qio_channel_info = {
362 .parent = TYPE_OBJECT,
363 .name = TYPE_QIO_CHANNEL,
364 .instance_size = sizeof(QIOChannel),
365 .instance_finalize = qio_channel_finalize,
366 .abstract = true,
367 .class_size = sizeof(QIOChannelClass),
368};
369
370
371static void qio_channel_register_types(void)
372{
373 type_register_static(&qio_channel_info);
374}
375
376
377type_init(qio_channel_register_types);
378