1 /*
2 * Aio tasks loops
3 *
4 * Copyright (c) 2019 Virtuozzo International GmbH.
5 *
6 * Permission is hereby granted, free of charge, to any person obtaining a copy
7 * of this software and associated documentation files (the "Software"), to deal
8 * in the Software without restriction, including without limitation the rights
9 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10 * copies of the Software, and to permit persons to whom the Software is
11 * furnished to do so, subject to the following conditions:
12 *
13 * The above copyright notice and this permission notice shall be included in
14 * all copies or substantial portions of the Software.
15 *
16 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
19 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
22 * THE SOFTWARE.
23 */
24
25 #include "qemu/osdep.h"
26 #include "block/aio.h"
27 #include "block/aio_task.h"
28
29 struct AioTaskPool {
30 Coroutine *main_co;
31 int status;
32 int max_busy_tasks;
33 int busy_tasks;
34 bool waiting;
35 };
36
aio_task_co(void * opaque)37 static void coroutine_fn aio_task_co(void *opaque)
38 {
39 AioTask *task = opaque;
40 AioTaskPool *pool = task->pool;
41
42 assert(pool->busy_tasks < pool->max_busy_tasks);
43 pool->busy_tasks++;
44
45 task->ret = task->func(task);
46
47 pool->busy_tasks--;
48
49 if (task->ret < 0 && pool->status == 0) {
50 pool->status = task->ret;
51 }
52
53 g_free(task);
54
55 if (pool->waiting) {
56 pool->waiting = false;
57 aio_co_wake(pool->main_co);
58 }
59 }
60
aio_task_pool_wait_one(AioTaskPool * pool)61 void coroutine_fn aio_task_pool_wait_one(AioTaskPool *pool)
62 {
63 assert(pool->busy_tasks > 0);
64 assert(qemu_coroutine_self() == pool->main_co);
65
66 pool->waiting = true;
67 qemu_coroutine_yield();
68
69 assert(!pool->waiting);
70 assert(pool->busy_tasks < pool->max_busy_tasks);
71 }
72
aio_task_pool_wait_slot(AioTaskPool * pool)73 void coroutine_fn aio_task_pool_wait_slot(AioTaskPool *pool)
74 {
75 if (pool->busy_tasks < pool->max_busy_tasks) {
76 return;
77 }
78
79 aio_task_pool_wait_one(pool);
80 }
81
aio_task_pool_wait_all(AioTaskPool * pool)82 void coroutine_fn aio_task_pool_wait_all(AioTaskPool *pool)
83 {
84 while (pool->busy_tasks > 0) {
85 aio_task_pool_wait_one(pool);
86 }
87 }
88
aio_task_pool_start_task(AioTaskPool * pool,AioTask * task)89 void coroutine_fn aio_task_pool_start_task(AioTaskPool *pool, AioTask *task)
90 {
91 aio_task_pool_wait_slot(pool);
92
93 task->pool = pool;
94 qemu_coroutine_enter(qemu_coroutine_create(aio_task_co, task));
95 }
96
aio_task_pool_new(int max_busy_tasks)97 AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks)
98 {
99 AioTaskPool *pool = g_new0(AioTaskPool, 1);
100
101 assert(max_busy_tasks > 0);
102
103 pool->main_co = qemu_coroutine_self();
104 pool->max_busy_tasks = max_busy_tasks;
105
106 return pool;
107 }
108
aio_task_pool_free(AioTaskPool * pool)109 void aio_task_pool_free(AioTaskPool *pool)
110 {
111 g_free(pool);
112 }
113
aio_task_pool_status(AioTaskPool * pool)114 int aio_task_pool_status(AioTaskPool *pool)
115 {
116 if (!pool) {
117 return 0; /* Sugar for lazy allocation of aio pool */
118 }
119
120 return pool->status;
121 }
122
aio_task_pool_empty(AioTaskPool * pool)123 bool aio_task_pool_empty(AioTaskPool *pool)
124 {
125 return pool->busy_tasks == 0;
126 }
127