1#include <glib.h>
2#include "qemu-common.h"
3#include "block/aio.h"
4#include "block/thread-pool.h"
5#include "block/block.h"
6#include "qemu/timer.h"
7#include "qemu/error-report.h"
8
9static AioContext *ctx;
10static ThreadPool *pool;
11static int active;
12
13typedef struct {
14 BlockAIOCB *aiocb;
15 int n;
16 int ret;
17} WorkerTestData;
18
19static int worker_cb(void *opaque)
20{
21 WorkerTestData *data = opaque;
22 return atomic_fetch_inc(&data->n);
23}
24
25static int long_cb(void *opaque)
26{
27 WorkerTestData *data = opaque;
28 atomic_inc(&data->n);
29 g_usleep(2000000);
30 atomic_inc(&data->n);
31 return 0;
32}
33
34static void done_cb(void *opaque, int ret)
35{
36 WorkerTestData *data = opaque;
37 g_assert(data->ret == -EINPROGRESS || data->ret == -ECANCELED);
38 data->ret = ret;
39 data->aiocb = NULL;
40
41
42 active--;
43}
44
45static void test_submit(void)
46{
47 WorkerTestData data = { .n = 0 };
48 thread_pool_submit(pool, worker_cb, &data);
49 while (data.n == 0) {
50 aio_poll(ctx, true);
51 }
52 g_assert_cmpint(data.n, ==, 1);
53}
54
55static void test_submit_aio(void)
56{
57 WorkerTestData data = { .n = 0, .ret = -EINPROGRESS };
58 data.aiocb = thread_pool_submit_aio(pool, worker_cb, &data,
59 done_cb, &data);
60
61
62 active = 1;
63 g_assert_cmpint(data.ret, ==, -EINPROGRESS);
64 while (data.ret == -EINPROGRESS) {
65 aio_poll(ctx, true);
66 }
67 g_assert_cmpint(active, ==, 0);
68 g_assert_cmpint(data.n, ==, 1);
69 g_assert_cmpint(data.ret, ==, 0);
70}
71
72static void co_test_cb(void *opaque)
73{
74 WorkerTestData *data = opaque;
75
76 active = 1;
77 data->n = 0;
78 data->ret = -EINPROGRESS;
79 thread_pool_submit_co(pool, worker_cb, data);
80
81
82
83 g_assert_cmpint(data->n, ==, 1);
84 data->ret = 0;
85 active--;
86
87
88}
89
90static void test_submit_co(void)
91{
92 WorkerTestData data;
93 Coroutine *co = qemu_coroutine_create(co_test_cb);
94
95 qemu_coroutine_enter(co, &data);
96
97
98
99 g_assert_cmpint(active, ==, 1);
100 g_assert_cmpint(data.ret, ==, -EINPROGRESS);
101
102
103
104 while (data.ret == -EINPROGRESS) {
105 aio_poll(ctx, true);
106 }
107
108
109
110 g_assert_cmpint(active, ==, 0);
111 g_assert_cmpint(data.ret, ==, 0);
112}
113
114static void test_submit_many(void)
115{
116 WorkerTestData data[100];
117 int i;
118
119
120 for (i = 0; i < 100; i++) {
121 data[i].n = 0;
122 data[i].ret = -EINPROGRESS;
123 thread_pool_submit_aio(pool, worker_cb, &data[i], done_cb, &data[i]);
124 }
125
126 active = 100;
127 while (active > 0) {
128 aio_poll(ctx, true);
129 }
130 for (i = 0; i < 100; i++) {
131 g_assert_cmpint(data[i].n, ==, 1);
132 g_assert_cmpint(data[i].ret, ==, 0);
133 }
134}
135
136static void do_test_cancel(bool sync)
137{
138 WorkerTestData data[100];
139 int num_canceled;
140 int i;
141
142
143
144
145 test_submit_many();
146
147
148 for (i = 0; i < 100; i++) {
149 data[i].n = 0;
150 data[i].ret = -EINPROGRESS;
151 data[i].aiocb = thread_pool_submit_aio(pool, long_cb, &data[i],
152 done_cb, &data[i]);
153 }
154
155
156
157
158 active = 100;
159 aio_notify(ctx);
160 aio_poll(ctx, false);
161
162
163
164
165 g_assert_cmpint(active, ==, 100);
166 g_usleep(1000000);
167 g_assert_cmpint(active, >, 50);
168
169
170 num_canceled = 0;
171 for (i = 0; i < 100; i++) {
172 if (atomic_cmpxchg(&data[i].n, 0, 3) == 0) {
173 data[i].ret = -ECANCELED;
174 if (sync) {
175 bdrv_aio_cancel(data[i].aiocb);
176 } else {
177 bdrv_aio_cancel_async(data[i].aiocb);
178 }
179 num_canceled++;
180 }
181 }
182 g_assert_cmpint(active, >, 0);
183 g_assert_cmpint(num_canceled, <, 100);
184
185 for (i = 0; i < 100; i++) {
186 if (data[i].aiocb && data[i].n != 3) {
187 if (sync) {
188
189 bdrv_aio_cancel(data[i].aiocb);
190 } else {
191 bdrv_aio_cancel_async(data[i].aiocb);
192 }
193 }
194 }
195
196
197 while (active > 0) {
198 aio_poll(ctx, true);
199 }
200 g_assert_cmpint(active, ==, 0);
201 for (i = 0; i < 100; i++) {
202 if (data[i].n == 3) {
203 g_assert_cmpint(data[i].ret, ==, -ECANCELED);
204 g_assert(data[i].aiocb == NULL);
205 } else {
206 g_assert_cmpint(data[i].n, ==, 2);
207 g_assert(data[i].ret == 0 || data[i].ret == -ECANCELED);
208 g_assert(data[i].aiocb == NULL);
209 }
210 }
211}
212
213static void test_cancel(void)
214{
215 do_test_cancel(true);
216}
217
218static void test_cancel_async(void)
219{
220 do_test_cancel(false);
221}
222
223int main(int argc, char **argv)
224{
225 int ret;
226 Error *local_error = NULL;
227
228 init_clocks();
229
230 ctx = aio_context_new(&local_error);
231 if (!ctx) {
232 error_report("Failed to create AIO Context: '%s'",
233 error_get_pretty(local_error));
234 error_free(local_error);
235 exit(1);
236 }
237 pool = aio_get_thread_pool(ctx);
238
239 g_test_init(&argc, &argv, NULL);
240 g_test_add_func("/thread-pool/submit", test_submit);
241 g_test_add_func("/thread-pool/submit-aio", test_submit_aio);
242 g_test_add_func("/thread-pool/submit-co", test_submit_co);
243 g_test_add_func("/thread-pool/submit-many", test_submit_many);
244 g_test_add_func("/thread-pool/cancel", test_cancel);
245 g_test_add_func("/thread-pool/cancel-async", test_cancel_async);
246
247 ret = g_test_run();
248
249 aio_context_unref(ctx);
250 return ret;
251}
252