1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25#include "qemu/osdep.h"
26#include "qapi/error.h"
27#include "qemu-common.h"
28#include "block/aio.h"
29#include "block/thread-pool.h"
30#include "qemu/main-loop.h"
31#include "qemu/atomic.h"
32#include "block/raw-aio.h"
33
34
35
36
37struct QEMUBH {
38 AioContext *ctx;
39 QEMUBHFunc *cb;
40 void *opaque;
41 QEMUBH *next;
42 bool scheduled;
43 bool idle;
44 bool deleted;
45};
46
47void aio_bh_schedule_oneshot(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
48{
49 QEMUBH *bh;
50 bh = g_new(QEMUBH, 1);
51 *bh = (QEMUBH){
52 .ctx = ctx,
53 .cb = cb,
54 .opaque = opaque,
55 };
56 qemu_mutex_lock(&ctx->bh_lock);
57 bh->next = ctx->first_bh;
58 bh->scheduled = 1;
59 bh->deleted = 1;
60
61 smp_wmb();
62 ctx->first_bh = bh;
63 qemu_mutex_unlock(&ctx->bh_lock);
64 aio_notify(ctx);
65}
66
67QEMUBH *aio_bh_new(AioContext *ctx, QEMUBHFunc *cb, void *opaque)
68{
69 QEMUBH *bh;
70 bh = g_new(QEMUBH, 1);
71 *bh = (QEMUBH){
72 .ctx = ctx,
73 .cb = cb,
74 .opaque = opaque,
75 };
76 qemu_mutex_lock(&ctx->bh_lock);
77 bh->next = ctx->first_bh;
78
79 smp_wmb();
80 ctx->first_bh = bh;
81 qemu_mutex_unlock(&ctx->bh_lock);
82 return bh;
83}
84
85void aio_bh_call(QEMUBH *bh)
86{
87 bh->cb(bh->opaque);
88}
89
90
91int aio_bh_poll(AioContext *ctx)
92{
93 QEMUBH *bh, **bhp, *next;
94 int ret;
95
96 ctx->walking_bh++;
97
98 ret = 0;
99 for (bh = ctx->first_bh; bh; bh = next) {
100
101 smp_read_barrier_depends();
102 next = bh->next;
103
104
105
106
107
108
109 if (atomic_xchg(&bh->scheduled, 0)) {
110
111 if (!bh->idle) {
112 ret = 1;
113 }
114 bh->idle = 0;
115 aio_bh_call(bh);
116 }
117 }
118
119 ctx->walking_bh--;
120
121
122 if (!ctx->walking_bh) {
123 qemu_mutex_lock(&ctx->bh_lock);
124 bhp = &ctx->first_bh;
125 while (*bhp) {
126 bh = *bhp;
127 if (bh->deleted && !bh->scheduled) {
128 *bhp = bh->next;
129 g_free(bh);
130 } else {
131 bhp = &bh->next;
132 }
133 }
134 qemu_mutex_unlock(&ctx->bh_lock);
135 }
136
137 return ret;
138}
139
140void qemu_bh_schedule_idle(QEMUBH *bh)
141{
142 bh->idle = 1;
143
144
145
146 atomic_mb_set(&bh->scheduled, 1);
147}
148
149void qemu_bh_schedule(QEMUBH *bh)
150{
151 AioContext *ctx;
152
153 ctx = bh->ctx;
154 bh->idle = 0;
155
156
157
158
159
160
161 if (atomic_xchg(&bh->scheduled, 1) == 0) {
162 aio_notify(ctx);
163 }
164}
165
166
167
168
169void qemu_bh_cancel(QEMUBH *bh)
170{
171 bh->scheduled = 0;
172}
173
174
175
176
177void qemu_bh_delete(QEMUBH *bh)
178{
179 bh->scheduled = 0;
180 bh->deleted = 1;
181}
182
183int64_t
184aio_compute_timeout(AioContext *ctx)
185{
186 int64_t deadline;
187 int timeout = -1;
188 QEMUBH *bh;
189
190 for (bh = ctx->first_bh; bh; bh = bh->next) {
191 if (bh->scheduled) {
192 if (bh->idle) {
193
194
195 timeout = 10000000;
196 } else {
197
198
199 return 0;
200 }
201 }
202 }
203
204 deadline = timerlistgroup_deadline_ns(&ctx->tlg);
205 if (deadline == 0) {
206 return 0;
207 } else {
208 return qemu_soonest_timeout(timeout, deadline);
209 }
210}
211
212static gboolean
213aio_ctx_prepare(GSource *source, gint *timeout)
214{
215 AioContext *ctx = (AioContext *) source;
216
217 atomic_or(&ctx->notify_me, 1);
218
219
220 *timeout = qemu_timeout_ns_to_ms(aio_compute_timeout(ctx));
221
222 if (aio_prepare(ctx)) {
223 *timeout = 0;
224 }
225
226 return *timeout == 0;
227}
228
229static gboolean
230aio_ctx_check(GSource *source)
231{
232 AioContext *ctx = (AioContext *) source;
233 QEMUBH *bh;
234
235 atomic_and(&ctx->notify_me, ~1);
236 aio_notify_accept(ctx);
237
238 for (bh = ctx->first_bh; bh; bh = bh->next) {
239 if (bh->scheduled) {
240 return true;
241 }
242 }
243 return aio_pending(ctx) || (timerlistgroup_deadline_ns(&ctx->tlg) == 0);
244}
245
246static gboolean
247aio_ctx_dispatch(GSource *source,
248 GSourceFunc callback,
249 gpointer user_data)
250{
251 AioContext *ctx = (AioContext *) source;
252
253 assert(callback == NULL);
254 aio_dispatch(ctx);
255 return true;
256}
257
258static void
259aio_ctx_finalize(GSource *source)
260{
261 AioContext *ctx = (AioContext *) source;
262
263 thread_pool_free(ctx->thread_pool);
264
265#ifdef CONFIG_LINUX_AIO
266 if (ctx->linux_aio) {
267 laio_detach_aio_context(ctx->linux_aio, ctx);
268 laio_cleanup(ctx->linux_aio);
269 ctx->linux_aio = NULL;
270 }
271#endif
272
273 qemu_mutex_lock(&ctx->bh_lock);
274 while (ctx->first_bh) {
275 QEMUBH *next = ctx->first_bh->next;
276
277
278 assert(ctx->first_bh->deleted);
279
280 g_free(ctx->first_bh);
281 ctx->first_bh = next;
282 }
283 qemu_mutex_unlock(&ctx->bh_lock);
284
285 aio_set_event_notifier(ctx, &ctx->notifier, false, NULL);
286 event_notifier_cleanup(&ctx->notifier);
287 qemu_rec_mutex_destroy(&ctx->lock);
288 qemu_mutex_destroy(&ctx->bh_lock);
289 timerlistgroup_deinit(&ctx->tlg);
290}
291
292static GSourceFuncs aio_source_funcs = {
293 aio_ctx_prepare,
294 aio_ctx_check,
295 aio_ctx_dispatch,
296 aio_ctx_finalize
297};
298
299GSource *aio_get_g_source(AioContext *ctx)
300{
301 g_source_ref(&ctx->source);
302 return &ctx->source;
303}
304
305ThreadPool *aio_get_thread_pool(AioContext *ctx)
306{
307 if (!ctx->thread_pool) {
308 ctx->thread_pool = thread_pool_new(ctx);
309 }
310 return ctx->thread_pool;
311}
312
313#ifdef CONFIG_LINUX_AIO
314LinuxAioState *aio_get_linux_aio(AioContext *ctx)
315{
316 if (!ctx->linux_aio) {
317 ctx->linux_aio = laio_init();
318 laio_attach_aio_context(ctx->linux_aio, ctx);
319 }
320 return ctx->linux_aio;
321}
322#endif
323
324void aio_notify(AioContext *ctx)
325{
326
327
328
329 smp_mb();
330 if (ctx->notify_me) {
331 event_notifier_set(&ctx->notifier);
332 atomic_mb_set(&ctx->notified, true);
333 }
334}
335
336void aio_notify_accept(AioContext *ctx)
337{
338 if (atomic_xchg(&ctx->notified, false)) {
339 event_notifier_test_and_clear(&ctx->notifier);
340 }
341}
342
343static void aio_timerlist_notify(void *opaque)
344{
345 aio_notify(opaque);
346}
347
348static void event_notifier_dummy_cb(EventNotifier *e)
349{
350}
351
352AioContext *aio_context_new(Error **errp)
353{
354 int ret;
355 AioContext *ctx;
356
357 ctx = (AioContext *) g_source_new(&aio_source_funcs, sizeof(AioContext));
358 aio_context_setup(ctx);
359
360 ret = event_notifier_init(&ctx->notifier, false);
361 if (ret < 0) {
362 error_setg_errno(errp, -ret, "Failed to initialize event notifier");
363 goto fail;
364 }
365 g_source_set_can_recurse(&ctx->source, true);
366 aio_set_event_notifier(ctx, &ctx->notifier,
367 false,
368 (EventNotifierHandler *)
369 event_notifier_dummy_cb);
370#ifdef CONFIG_LINUX_AIO
371 ctx->linux_aio = NULL;
372#endif
373 ctx->thread_pool = NULL;
374 qemu_mutex_init(&ctx->bh_lock);
375 qemu_rec_mutex_init(&ctx->lock);
376 timerlistgroup_init(&ctx->tlg, aio_timerlist_notify, ctx);
377
378 return ctx;
379fail:
380 g_source_destroy(&ctx->source);
381 return NULL;
382}
383
384void aio_context_ref(AioContext *ctx)
385{
386 g_source_ref(&ctx->source);
387}
388
389void aio_context_unref(AioContext *ctx)
390{
391 g_source_unref(&ctx->source);
392}
393
394void aio_context_acquire(AioContext *ctx)
395{
396 qemu_rec_mutex_lock(&ctx->lock);
397}
398
399void aio_context_release(AioContext *ctx)
400{
401 qemu_rec_mutex_unlock(&ctx->lock);
402}
403