1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16#include "qemu/osdep.h"
17#include "qemu-common.h"
18#include "block/block.h"
19#include "qemu/rcu_queue.h"
20#include "qemu/sockets.h"
21#include "qemu/cutils.h"
22#include "trace.h"
23#ifdef CONFIG_EPOLL_CREATE1
24#include <sys/epoll.h>
25#endif
26
27struct AioHandler
28{
29 GPollFD pfd;
30 IOHandler *io_read;
31 IOHandler *io_write;
32 AioPollFn *io_poll;
33 IOHandler *io_poll_begin;
34 IOHandler *io_poll_end;
35 int deleted;
36 void *opaque;
37 bool is_external;
38 QLIST_ENTRY(AioHandler) node;
39};
40
41#ifdef CONFIG_EPOLL_CREATE1
42
43
44#define EPOLL_ENABLE_THRESHOLD 64
45
46static void aio_epoll_disable(AioContext *ctx)
47{
48 ctx->epoll_enabled = false;
49 if (!ctx->epoll_available) {
50 return;
51 }
52 ctx->epoll_available = false;
53 close(ctx->epollfd);
54}
55
56static inline int epoll_events_from_pfd(int pfd_events)
57{
58 return (pfd_events & G_IO_IN ? EPOLLIN : 0) |
59 (pfd_events & G_IO_OUT ? EPOLLOUT : 0) |
60 (pfd_events & G_IO_HUP ? EPOLLHUP : 0) |
61 (pfd_events & G_IO_ERR ? EPOLLERR : 0);
62}
63
64static bool aio_epoll_try_enable(AioContext *ctx)
65{
66 AioHandler *node;
67 struct epoll_event event;
68
69 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
70 int r;
71 if (node->deleted || !node->pfd.events) {
72 continue;
73 }
74 event.events = epoll_events_from_pfd(node->pfd.events);
75 event.data.ptr = node;
76 r = epoll_ctl(ctx->epollfd, EPOLL_CTL_ADD, node->pfd.fd, &event);
77 if (r) {
78 return false;
79 }
80 }
81 ctx->epoll_enabled = true;
82 return true;
83}
84
85static void aio_epoll_update(AioContext *ctx, AioHandler *node, bool is_new)
86{
87 struct epoll_event event;
88 int r;
89 int ctl;
90
91 if (!ctx->epoll_enabled) {
92 return;
93 }
94 if (!node->pfd.events) {
95 ctl = EPOLL_CTL_DEL;
96 } else {
97 event.data.ptr = node;
98 event.events = epoll_events_from_pfd(node->pfd.events);
99 ctl = is_new ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
100 }
101
102 r = epoll_ctl(ctx->epollfd, ctl, node->pfd.fd, &event);
103 if (r) {
104 aio_epoll_disable(ctx);
105 }
106}
107
108static int aio_epoll(AioContext *ctx, GPollFD *pfds,
109 unsigned npfd, int64_t timeout)
110{
111 AioHandler *node;
112 int i, ret = 0;
113 struct epoll_event events[128];
114
115 assert(npfd == 1);
116 assert(pfds[0].fd == ctx->epollfd);
117 if (timeout > 0) {
118 ret = qemu_poll_ns(pfds, npfd, timeout);
119 }
120 if (timeout <= 0 || ret > 0) {
121 ret = epoll_wait(ctx->epollfd, events,
122 ARRAY_SIZE(events),
123 timeout);
124 if (ret <= 0) {
125 goto out;
126 }
127 for (i = 0; i < ret; i++) {
128 int ev = events[i].events;
129 node = events[i].data.ptr;
130 node->pfd.revents = (ev & EPOLLIN ? G_IO_IN : 0) |
131 (ev & EPOLLOUT ? G_IO_OUT : 0) |
132 (ev & EPOLLHUP ? G_IO_HUP : 0) |
133 (ev & EPOLLERR ? G_IO_ERR : 0);
134 }
135 }
136out:
137 return ret;
138}
139
140static bool aio_epoll_enabled(AioContext *ctx)
141{
142
143 return !aio_external_disabled(ctx) && ctx->epoll_enabled;
144}
145
146static bool aio_epoll_check_poll(AioContext *ctx, GPollFD *pfds,
147 unsigned npfd, int64_t timeout)
148{
149 if (!ctx->epoll_available) {
150 return false;
151 }
152 if (aio_epoll_enabled(ctx)) {
153 return true;
154 }
155 if (npfd >= EPOLL_ENABLE_THRESHOLD) {
156 if (aio_epoll_try_enable(ctx)) {
157 return true;
158 } else {
159 aio_epoll_disable(ctx);
160 }
161 }
162 return false;
163}
164
165#else
166
167static void aio_epoll_update(AioContext *ctx, AioHandler *node, bool is_new)
168{
169}
170
171static int aio_epoll(AioContext *ctx, GPollFD *pfds,
172 unsigned npfd, int64_t timeout)
173{
174 assert(false);
175}
176
177static bool aio_epoll_enabled(AioContext *ctx)
178{
179 return false;
180}
181
182static bool aio_epoll_check_poll(AioContext *ctx, GPollFD *pfds,
183 unsigned npfd, int64_t timeout)
184{
185 return false;
186}
187
188#endif
189
190static AioHandler *find_aio_handler(AioContext *ctx, int fd)
191{
192 AioHandler *node;
193
194 QLIST_FOREACH(node, &ctx->aio_handlers, node) {
195 if (node->pfd.fd == fd)
196 if (!node->deleted)
197 return node;
198 }
199
200 return NULL;
201}
202
203void aio_set_fd_handler(AioContext *ctx,
204 int fd,
205 bool is_external,
206 IOHandler *io_read,
207 IOHandler *io_write,
208 AioPollFn *io_poll,
209 void *opaque)
210{
211 AioHandler *node;
212 bool is_new = false;
213 bool deleted = false;
214 int poll_disable_change;
215
216 qemu_lockcnt_lock(&ctx->list_lock);
217
218 node = find_aio_handler(ctx, fd);
219
220
221 if (!io_read && !io_write && !io_poll) {
222 if (node == NULL) {
223 qemu_lockcnt_unlock(&ctx->list_lock);
224 return;
225 }
226
227
228
229
230
231
232 if (!g_source_is_destroyed(&ctx->source)) {
233 g_source_remove_poll(&ctx->source, &node->pfd);
234 }
235
236
237 if (qemu_lockcnt_count(&ctx->list_lock)) {
238 node->deleted = 1;
239 node->pfd.revents = 0;
240 } else {
241
242
243
244
245 QLIST_REMOVE(node, node);
246 deleted = true;
247 }
248 poll_disable_change = -!node->io_poll;
249 } else {
250 poll_disable_change = !io_poll - (node && !node->io_poll);
251 if (node == NULL) {
252
253 node = g_new0(AioHandler, 1);
254 node->pfd.fd = fd;
255 QLIST_INSERT_HEAD_RCU(&ctx->aio_handlers, node, node);
256
257 g_source_add_poll(&ctx->source, &node->pfd);
258 is_new = true;
259 }
260
261
262 node->io_read = io_read;
263 node->io_write = io_write;
264 node->io_poll = io_poll;
265 node->opaque = opaque;
266 node->is_external = is_external;
267
268 node->pfd.events = (io_read ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0);
269 node->pfd.events |= (io_write ? G_IO_OUT | G_IO_ERR : 0);
270 }
271
272
273
274
275
276
277
278 atomic_set(&ctx->poll_disable_cnt,
279 atomic_read(&ctx->poll_disable_cnt) + poll_disable_change);
280
281 aio_epoll_update(ctx, node, is_new);
282 qemu_lockcnt_unlock(&ctx->list_lock);
283 aio_notify(ctx);
284
285 if (deleted) {
286 g_free(node);
287 }
288}
289
290void aio_set_fd_poll(AioContext *ctx, int fd,
291 IOHandler *io_poll_begin,
292 IOHandler *io_poll_end)
293{
294 AioHandler *node = find_aio_handler(ctx, fd);
295
296 if (!node) {
297 return;
298 }
299
300 node->io_poll_begin = io_poll_begin;
301 node->io_poll_end = io_poll_end;
302}
303
304void aio_set_event_notifier(AioContext *ctx,
305 EventNotifier *notifier,
306 bool is_external,
307 EventNotifierHandler *io_read,
308 AioPollFn *io_poll)
309{
310 aio_set_fd_handler(ctx, event_notifier_get_fd(notifier), is_external,
311 (IOHandler *)io_read, NULL, io_poll, notifier);
312}
313
314void aio_set_event_notifier_poll(AioContext *ctx,
315 EventNotifier *notifier,
316 EventNotifierHandler *io_poll_begin,
317 EventNotifierHandler *io_poll_end)
318{
319 aio_set_fd_poll(ctx, event_notifier_get_fd(notifier),
320 (IOHandler *)io_poll_begin,
321 (IOHandler *)io_poll_end);
322}
323
324static void poll_set_started(AioContext *ctx, bool started)
325{
326 AioHandler *node;
327
328 if (started == ctx->poll_started) {
329 return;
330 }
331
332 ctx->poll_started = started;
333
334 qemu_lockcnt_inc(&ctx->list_lock);
335 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
336 IOHandler *fn;
337
338 if (node->deleted) {
339 continue;
340 }
341
342 if (started) {
343 fn = node->io_poll_begin;
344 } else {
345 fn = node->io_poll_end;
346 }
347
348 if (fn) {
349 fn(node->opaque);
350 }
351 }
352 qemu_lockcnt_dec(&ctx->list_lock);
353}
354
355
356bool aio_prepare(AioContext *ctx)
357{
358
359 poll_set_started(ctx, false);
360
361 return false;
362}
363
364bool aio_pending(AioContext *ctx)
365{
366 AioHandler *node;
367 bool result = false;
368
369
370
371
372
373 qemu_lockcnt_inc(&ctx->list_lock);
374
375 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
376 int revents;
377
378 revents = node->pfd.revents & node->pfd.events;
379 if (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR) && node->io_read &&
380 aio_node_check(ctx, node->is_external)) {
381 result = true;
382 break;
383 }
384 if (revents & (G_IO_OUT | G_IO_ERR) && node->io_write &&
385 aio_node_check(ctx, node->is_external)) {
386 result = true;
387 break;
388 }
389 }
390 qemu_lockcnt_dec(&ctx->list_lock);
391
392 return result;
393}
394
395static bool aio_dispatch_handlers(AioContext *ctx)
396{
397 AioHandler *node, *tmp;
398 bool progress = false;
399
400 QLIST_FOREACH_SAFE_RCU(node, &ctx->aio_handlers, node, tmp) {
401 int revents;
402
403 revents = node->pfd.revents & node->pfd.events;
404 node->pfd.revents = 0;
405
406 if (!node->deleted &&
407 (revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)) &&
408 aio_node_check(ctx, node->is_external) &&
409 node->io_read) {
410 node->io_read(node->opaque);
411
412
413 if (node->opaque != &ctx->notifier) {
414 progress = true;
415 }
416 }
417 if (!node->deleted &&
418 (revents & (G_IO_OUT | G_IO_ERR)) &&
419 aio_node_check(ctx, node->is_external) &&
420 node->io_write) {
421 node->io_write(node->opaque);
422 progress = true;
423 }
424
425 if (node->deleted) {
426 if (qemu_lockcnt_dec_if_lock(&ctx->list_lock)) {
427 QLIST_REMOVE(node, node);
428 g_free(node);
429 qemu_lockcnt_inc_and_unlock(&ctx->list_lock);
430 }
431 }
432 }
433
434 return progress;
435}
436
437void aio_dispatch(AioContext *ctx)
438{
439 qemu_lockcnt_inc(&ctx->list_lock);
440 aio_bh_poll(ctx);
441 aio_dispatch_handlers(ctx);
442 qemu_lockcnt_dec(&ctx->list_lock);
443
444 timerlistgroup_run_timers(&ctx->tlg);
445}
446
447
448
449
450
451
452
453
454
455
456
457
458static __thread GPollFD *pollfds;
459static __thread AioHandler **nodes;
460static __thread unsigned npfd, nalloc;
461static __thread Notifier pollfds_cleanup_notifier;
462
463static void pollfds_cleanup(Notifier *n, void *unused)
464{
465 g_assert(npfd == 0);
466 g_free(pollfds);
467 g_free(nodes);
468 nalloc = 0;
469}
470
471static void add_pollfd(AioHandler *node)
472{
473 if (npfd == nalloc) {
474 if (nalloc == 0) {
475 pollfds_cleanup_notifier.notify = pollfds_cleanup;
476 qemu_thread_atexit_add(&pollfds_cleanup_notifier);
477 nalloc = 8;
478 } else {
479 g_assert(nalloc <= INT_MAX);
480 nalloc *= 2;
481 }
482 pollfds = g_renew(GPollFD, pollfds, nalloc);
483 nodes = g_renew(AioHandler *, nodes, nalloc);
484 }
485 nodes[npfd] = node;
486 pollfds[npfd] = (GPollFD) {
487 .fd = node->pfd.fd,
488 .events = node->pfd.events,
489 };
490 npfd++;
491}
492
493static bool run_poll_handlers_once(AioContext *ctx, int64_t *timeout)
494{
495 bool progress = false;
496 AioHandler *node;
497
498 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
499 if (!node->deleted && node->io_poll &&
500 aio_node_check(ctx, node->is_external) &&
501 node->io_poll(node->opaque)) {
502 *timeout = 0;
503 if (node->opaque != &ctx->notifier) {
504 progress = true;
505 }
506 }
507
508
509 }
510
511 return progress;
512}
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527static bool run_poll_handlers(AioContext *ctx, int64_t max_ns, int64_t *timeout)
528{
529 bool progress;
530 int64_t start_time, elapsed_time;
531
532 assert(ctx->notify_me);
533 assert(qemu_lockcnt_count(&ctx->list_lock) > 0);
534
535 trace_run_poll_handlers_begin(ctx, max_ns, *timeout);
536
537 start_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
538 do {
539 progress = run_poll_handlers_once(ctx, timeout);
540 elapsed_time = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time;
541 } while (!progress && elapsed_time < max_ns
542 && !atomic_read(&ctx->poll_disable_cnt));
543
544
545
546
547 if (*timeout != -1) {
548 *timeout -= MIN(*timeout, elapsed_time);
549 }
550
551 trace_run_poll_handlers_end(ctx, progress, *timeout);
552 return progress;
553}
554
555
556
557
558
559
560
561
562
563
564
565
566static bool try_poll_mode(AioContext *ctx, int64_t *timeout)
567{
568
569 int64_t max_ns = MIN((uint64_t)*timeout, (uint64_t)ctx->poll_ns);
570
571 if (max_ns && !atomic_read(&ctx->poll_disable_cnt)) {
572 poll_set_started(ctx, true);
573
574 if (run_poll_handlers(ctx, max_ns, timeout)) {
575 return true;
576 }
577 }
578
579 poll_set_started(ctx, false);
580
581
582
583
584 return run_poll_handlers_once(ctx, timeout);
585}
586
587bool aio_poll(AioContext *ctx, bool blocking)
588{
589 AioHandler *node;
590 int i;
591 int ret = 0;
592 bool progress;
593 int64_t timeout;
594 int64_t start = 0;
595
596
597
598
599
600
601
602
603 if (blocking) {
604 assert(in_aio_context_home_thread(ctx));
605 atomic_add(&ctx->notify_me, 2);
606 }
607
608 qemu_lockcnt_inc(&ctx->list_lock);
609
610 if (ctx->poll_max_ns) {
611 start = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
612 }
613
614 timeout = blocking ? aio_compute_timeout(ctx) : 0;
615 progress = try_poll_mode(ctx, &timeout);
616 assert(!(timeout && progress));
617
618
619
620
621 if (timeout || atomic_read(&ctx->poll_disable_cnt)) {
622 assert(npfd == 0);
623
624
625
626 if (!aio_epoll_enabled(ctx)) {
627 QLIST_FOREACH_RCU(node, &ctx->aio_handlers, node) {
628 if (!node->deleted && node->pfd.events
629 && aio_node_check(ctx, node->is_external)) {
630 add_pollfd(node);
631 }
632 }
633 }
634
635
636 if (aio_epoll_check_poll(ctx, pollfds, npfd, timeout)) {
637 AioHandler epoll_handler;
638
639 epoll_handler.pfd.fd = ctx->epollfd;
640 epoll_handler.pfd.events = G_IO_IN | G_IO_OUT | G_IO_HUP | G_IO_ERR;
641 npfd = 0;
642 add_pollfd(&epoll_handler);
643 ret = aio_epoll(ctx, pollfds, npfd, timeout);
644 } else {
645 ret = qemu_poll_ns(pollfds, npfd, timeout);
646 }
647 }
648
649 if (blocking) {
650 atomic_sub(&ctx->notify_me, 2);
651 aio_notify_accept(ctx);
652 }
653
654
655 if (ctx->poll_max_ns) {
656 int64_t block_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start;
657
658 if (block_ns <= ctx->poll_ns) {
659
660 } else if (block_ns > ctx->poll_max_ns) {
661
662 int64_t old = ctx->poll_ns;
663
664 if (ctx->poll_shrink) {
665 ctx->poll_ns /= ctx->poll_shrink;
666 } else {
667 ctx->poll_ns = 0;
668 }
669
670 trace_poll_shrink(ctx, old, ctx->poll_ns);
671 } else if (ctx->poll_ns < ctx->poll_max_ns &&
672 block_ns < ctx->poll_max_ns) {
673
674 int64_t old = ctx->poll_ns;
675 int64_t grow = ctx->poll_grow;
676
677 if (grow == 0) {
678 grow = 2;
679 }
680
681 if (ctx->poll_ns) {
682 ctx->poll_ns *= grow;
683 } else {
684 ctx->poll_ns = 4000;
685 }
686
687 if (ctx->poll_ns > ctx->poll_max_ns) {
688 ctx->poll_ns = ctx->poll_max_ns;
689 }
690
691 trace_poll_grow(ctx, old, ctx->poll_ns);
692 }
693 }
694
695
696 if (ret > 0) {
697 for (i = 0; i < npfd; i++) {
698 nodes[i]->pfd.revents = pollfds[i].revents;
699 }
700 }
701
702 npfd = 0;
703
704 progress |= aio_bh_poll(ctx);
705
706 if (ret > 0) {
707 progress |= aio_dispatch_handlers(ctx);
708 }
709
710 qemu_lockcnt_dec(&ctx->list_lock);
711
712 progress |= timerlistgroup_run_timers(&ctx->tlg);
713
714 return progress;
715}
716
717void aio_context_setup(AioContext *ctx)
718{
719#ifdef CONFIG_EPOLL_CREATE1
720 assert(!ctx->epollfd);
721 ctx->epollfd = epoll_create1(EPOLL_CLOEXEC);
722 if (ctx->epollfd == -1) {
723 fprintf(stderr, "Failed to create epoll instance: %s", strerror(errno));
724 ctx->epoll_available = false;
725 } else {
726 ctx->epoll_available = true;
727 }
728#endif
729}
730
731void aio_context_destroy(AioContext *ctx)
732{
733#ifdef CONFIG_EPOLL_CREATE1
734 aio_epoll_disable(ctx);
735#endif
736}
737
738void aio_context_set_poll_params(AioContext *ctx, int64_t max_ns,
739 int64_t grow, int64_t shrink, Error **errp)
740{
741
742
743
744 ctx->poll_max_ns = max_ns;
745 ctx->poll_ns = 0;
746 ctx->poll_grow = grow;
747 ctx->poll_shrink = shrink;
748
749 aio_notify(ctx);
750}
751