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 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 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 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 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 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 97 AioTaskPool *coroutine_fn aio_task_pool_new(int max_busy_tasks) 98 { 99 AioTaskPool *pool = g_new0(AioTaskPool, 1); 100 101 pool->main_co = qemu_coroutine_self(); 102 pool->max_busy_tasks = max_busy_tasks; 103 104 return pool; 105 } 106 107 void aio_task_pool_free(AioTaskPool *pool) 108 { 109 g_free(pool); 110 } 111 112 int aio_task_pool_status(AioTaskPool *pool) 113 { 114 if (!pool) { 115 return 0; /* Sugar for lazy allocation of aio pool */ 116 } 117 118 return pool->status; 119 } 120 121 bool aio_task_pool_empty(AioTaskPool *pool) 122 { 123 return pool->busy_tasks == 0; 124 } 125