1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18#include "qemu/osdep.h"
19#include "block/block.h"
20#include "qemu/main-loop.h"
21#include "qemu/queue.h"
22#include "qemu/sockets.h"
23#include "qapi/error.h"
24#include "qemu/rcu_queue.h"
25
26struct AioHandler {
27 EventNotifier *e;
28 IOHandler *io_read;
29 IOHandler *io_write;
30 EventNotifierHandler *io_notify;
31 GPollFD pfd;
32 int deleted;
33 void *opaque;
34 bool is_external;
35 QLIST_ENTRY(AioHandler) node;
36};
37
38static void aio_remove_fd_handler(AioContext *ctx, AioHandler *node)
39{
40
41
42
43
44
45
46 if (!g_source_is_destroyed(&ctx->source)) {
47 g_source_remove_poll(&ctx->source, &node->pfd);
48 }
49
50
51 if (qemu_lockcnt_count(&ctx->list_lock)) {
52 node->deleted = 1;
53 node->pfd.revents = 0;
54 } else {
55
56
57
58
59 QLIST_REMOVE(node, node);
60 g_free(node);
61 }
62}
63
64void aio_set_fd_handler(AioContext *ctx,
65 int fd,
66 bool is_external,
67 IOHandler *io_read,
68 IOHandler *io_write,
69 AioPollFn *io_poll,
70 IOHandler *io_poll_ready,
71 void *opaque)
72{
73
74 AioHandler *old_node;
75 AioHandler *node = NULL;
76
77 qemu_lockcnt_lock(&ctx->list_lock);
78 QLIST_FOREACH(old_node, &ctx->aio_handlers, node) {
79 if (old_node->pfd.fd == fd && !old_node->deleted) {
80 break;
81 }
82 }
83
84 if (io_read || io_write) {
85 HANDLE event;
86 long bitmask = 0;
87
88
89 node = g_new0(AioHandler, 1);
90 node->pfd.fd = fd;
91
92 node->pfd.events = 0;
93 if (node->io_read) {
94 node->pfd.events |= G_IO_IN;
95 }
96 if (node->io_write) {
97 node->pfd.events |= G_IO_OUT;
98 }
99
100 node->e = &ctx->notifier;
101
102
103 node->opaque = opaque;
104 node->io_read = io_read;
105 node->io_write = io_write;
106 node->is_external = is_external;
107
108 if (io_read) {
109 bitmask |= FD_READ | FD_ACCEPT | FD_CLOSE;
110 }
111
112 if (io_write) {
113 bitmask |= FD_WRITE | FD_CONNECT;
114 }
115
116 QLIST_INSERT_HEAD_RCU(&ctx->aio_handlers, node, node);
117 event = event_notifier_get_handle(&ctx->notifier);
118 WSAEventSelect(node->pfd.fd, event, bitmask);
119 }
120 if (old_node) {
121 aio_remove_fd_handler(ctx, old_node);
122 }
123
124 qemu_lockcnt_unlock(&ctx->list_lock);
125 aio_notify(ctx);
126}
127
128void aio_set_fd_poll(AioContext *ctx, int fd,
129 IOHandler *io_poll_begin,
130 IOHandler *io_poll_end)
131{
132
133}
134
135void aio_set_event_notifier(AioContext *ctx,
136 EventNotifier *e,
137 bool is_external,
138 EventNotifierHandler *io_notify,
139 AioPollFn *io_poll,
140 EventNotifierHandler *io_poll_ready)
141{
142 AioHandler *node;
143
144 qemu_lockcnt_lock(&ctx->list_lock);
145 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
146 if (node->e == e && !node->deleted) {
147 break;
148 }
149 }
150
151
152 if (!io_notify) {
153 if (node) {
154 aio_remove_fd_handler(ctx, node);
155 }
156 } else {
157 if (node == NULL) {
158
159 node = g_new0(AioHandler, 1);
160 node->e = e;
161 node->pfd.fd = (uintptr_t)event_notifier_get_handle(e);
162 node->pfd.events = G_IO_IN;
163 node->is_external = is_external;
164 QLIST_INSERT_HEAD_RCU(&ctx->aio_handlers, node, node);
165
166 g_source_add_poll(&ctx->source, &node->pfd);
167 }
168
169 node->io_notify = io_notify;
170 }
171
172 qemu_lockcnt_unlock(&ctx->list_lock);
173 aio_notify(ctx);
174}
175
176void aio_set_event_notifier_poll(AioContext *ctx,
177 EventNotifier *notifier,
178 EventNotifierHandler *io_poll_begin,
179 EventNotifierHandler *io_poll_end)
180{
181
182}
183
184bool aio_prepare(AioContext *ctx)
185{
186 static struct timeval tv0;
187 AioHandler *node;
188 bool have_select_revents = false;
189 fd_set rfds, wfds;
190
191
192
193
194
195 qemu_lockcnt_inc(&ctx->list_lock);
196
197
198 FD_ZERO(&rfds);
199 FD_ZERO(&wfds);
200 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
201 if (node->io_read) {
202 FD_SET ((SOCKET)node->pfd.fd, &rfds);
203 }
204 if (node->io_write) {
205 FD_SET ((SOCKET)node->pfd.fd, &wfds);
206 }
207 }
208
209 if (select(0, &rfds, &wfds, NULL, &tv0) > 0) {
210 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
211 node->pfd.revents = 0;
212 if (FD_ISSET(node->pfd.fd, &rfds)) {
213 node->pfd.revents |= G_IO_IN;
214 have_select_revents = true;
215 }
216
217 if (FD_ISSET(node->pfd.fd, &wfds)) {
218 node->pfd.revents |= G_IO_OUT;
219 have_select_revents = true;
220 }
221 }
222 }
223
224 qemu_lockcnt_dec(&ctx->list_lock);
225 return have_select_revents;
226}
227
228bool aio_pending(AioContext *ctx)
229{
230 AioHandler *node;
231 bool result = false;
232
233
234
235
236
237 qemu_lockcnt_inc(&ctx->list_lock);
238 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
239 if (node->pfd.revents && node->io_notify) {
240 result = true;
241 break;
242 }
243
244 if ((node->pfd.revents & G_IO_IN) && node->io_read) {
245 result = true;
246 break;
247 }
248 if ((node->pfd.revents & G_IO_OUT) && node->io_write) {
249 result = true;
250 break;
251 }
252 }
253
254 qemu_lockcnt_dec(&ctx->list_lock);
255 return result;
256}
257
258static bool aio_dispatch_handlers(AioContext *ctx, HANDLE event)
259{
260 AioHandler *node;
261 bool progress = false;
262 AioHandler *tmp;
263
264
265
266
267
268 QLIST_FOREACH_SAFE_RCU(node, &ctx->aio_handlers, node, tmp) {
269 int revents = node->pfd.revents;
270
271 if (!node->deleted &&
272 (revents || event_notifier_get_handle(node->e) == event) &&
273 node->io_notify) {
274 node->pfd.revents = 0;
275 node->io_notify(node->e);
276
277
278 if (node->e != &ctx->notifier) {
279 progress = true;
280 }
281 }
282
283 if (!node->deleted &&
284 (node->io_read || node->io_write)) {
285 node->pfd.revents = 0;
286 if ((revents & G_IO_IN) && node->io_read) {
287 node->io_read(node->opaque);
288 progress = true;
289 }
290 if ((revents & G_IO_OUT) && node->io_write) {
291 node->io_write(node->opaque);
292 progress = true;
293 }
294
295
296 if (event == event_notifier_get_handle(&ctx->notifier)) {
297 WSANETWORKEVENTS ev;
298 WSAEnumNetworkEvents(node->pfd.fd, event, &ev);
299 if (ev.lNetworkEvents) {
300 progress = true;
301 }
302 }
303 }
304
305 if (node->deleted) {
306 if (qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {
307 QLIST_REMOVE(node, node);
308 g_free(node);
309 qemu_lockcnt_inc_and_unlock(&ctx->list_lock);
310 }
311 }
312 }
313
314 return progress;
315}
316
317void aio_dispatch(AioContext *ctx)
318{
319 qemu_lockcnt_inc(&ctx->list_lock);
320 aio_bh_poll(ctx);
321 aio_dispatch_handlers(ctx, INVALID_HANDLE_VALUE);
322 qemu_lockcnt_dec(&ctx->list_lock);
323 timerlistgroup_run_timers(&ctx->tlg);
324}
325
326bool aio_poll(AioContext *ctx, bool blocking)
327{
328 AioHandler *node;
329 HANDLE events[MAXIMUM_WAIT_OBJECTS + 1];
330 bool progress, have_select_revents, first;
331 int count;
332 int timeout;
333
334
335
336
337
338
339
340
341
342
343 assert(in_aio_context_home_thread(ctx == iohandler_get_aio_context() ?
344 qemu_get_aio_context() : ctx));
345 progress = false;
346
347
348
349
350
351
352
353
354 if (blocking) {
355 qatomic_set(&ctx->notify_me, qatomic_read(&ctx->notify_me) + 2);
356
357
358
359
360
361 smp_mb();
362 }
363
364 qemu_lockcnt_inc(&ctx->list_lock);
365 have_select_revents = aio_prepare(ctx);
366
367
368 count = 0;
369 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
370 if (!node->deleted && node->io_notify
371 && aio_node_check(ctx, node->is_external)) {
372 events[count++] = event_notifier_get_handle(node->e);
373 }
374 }
375
376 first = true;
377
378
379 assert(count > 0);
380
381
382
383
384
385 do {
386 HANDLE event;
387 int ret;
388
389 timeout = blocking && !have_select_revents
390 ? qemu_timeout_ns_to_ms(aio_compute_timeout(ctx)) : 0;
391 ret = WaitForMultipleObjects(count, events, FALSE, timeout);
392 if (blocking) {
393 assert(first);
394 qatomic_store_release(&ctx->notify_me,
395 qatomic_read(&ctx->notify_me) - 2);
396 aio_notify_accept(ctx);
397 }
398
399 if (first) {
400 progress |= aio_bh_poll(ctx);
401 first = false;
402 }
403
404
405 event = NULL;
406 if ((DWORD) (ret - WAIT_OBJECT_0) < count) {
407 event = events[ret - WAIT_OBJECT_0];
408 events[ret - WAIT_OBJECT_0] = events[--count];
409 } else if (!have_select_revents) {
410 break;
411 }
412
413 have_select_revents = false;
414 blocking = false;
415
416 progress |= aio_dispatch_handlers(ctx, event);
417 } while (count > 0);
418
419 qemu_lockcnt_dec(&ctx->list_lock);
420
421 progress |= timerlistgroup_run_timers(&ctx->tlg);
422 return progress;
423}
424
425void aio_context_setup(AioContext *ctx)
426{
427}
428
429void aio_context_destroy(AioContext *ctx)
430{
431}
432
433void aio_context_use_g_source(AioContext *ctx)
434{
435}
436
437void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
438 int64_t grow, int64_t shrink, Error **errp)
439{
440 if (max_ns) {
441 error_setg(errp, "AioContext polling is not implemented on Windows");
442 }
443}
444
445void aio_context_set_aio_params(AioContext *ctx, int64_t max_batch,
446 Error **errp)
447{
448}
449