Lines Matching full:wq
23 #include "io-wq.h"
37 IO_WQ_BIT_EXIT = 0, /* wq exiting */
46 * One for each thread in a wq pool
54 struct io_wq *wq; member
147 static bool create_io_worker(struct io_wq *wq, struct io_wq_acct *acct);
149 static bool io_acct_cancel_pending_work(struct io_wq *wq,
153 static void io_wq_cancel_tw_create(struct io_wq *wq);
176 static inline struct io_wq_acct *io_get_acct(struct io_wq *wq, bool bound) in io_get_acct() argument
178 return &wq->acct[bound ? IO_WQ_ACCT_BOUND : IO_WQ_ACCT_UNBOUND]; in io_get_acct()
181 static inline struct io_wq_acct *io_work_get_acct(struct io_wq *wq, in io_work_get_acct() argument
184 return io_get_acct(wq, !(work_flags & IO_WQ_WORK_UNBOUND)); in io_work_get_acct()
192 static void io_worker_ref_put(struct io_wq *wq) in io_worker_ref_put() argument
194 if (atomic_dec_and_test(&wq->worker_refs)) in io_worker_ref_put()
195 complete(&wq->worker_done); in io_worker_ref_put()
205 return test_bit(IO_WQ_BIT_EXIT, &worker->wq->state); in io_wq_worker_stopped()
211 struct io_wq *wq = worker->wq; in io_worker_cancel_cb() local
217 io_worker_ref_put(wq); in io_worker_cancel_cb()
234 struct io_wq *wq = worker->wq; in io_worker_exit() local
238 struct callback_head *cb = task_work_cancel_match(wq->task, in io_worker_exit()
263 io_worker_ref_put(wq); in io_worker_exit()
323 static bool io_wq_create_worker(struct io_wq *wq, struct io_wq_acct *acct) in io_wq_create_worker() argument
330 pr_warn_once("io-wq is not configured for unbound workers"); in io_wq_create_worker()
340 atomic_inc(&wq->worker_refs); in io_wq_create_worker()
341 return create_io_worker(wq, acct); in io_wq_create_worker()
354 struct io_wq *wq; in create_worker_cb() local
360 wq = worker->wq; in create_worker_cb()
377 create_io_worker(wq, acct); in create_worker_cb()
381 io_worker_ref_put(wq); in create_worker_cb()
391 struct io_wq *wq = worker->wq; in io_queue_worker_create() local
394 if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) in io_queue_worker_create()
408 atomic_inc(&wq->worker_refs); in io_queue_worker_create()
410 if (!task_work_add(wq->task, &worker->create_work, TWA_SIGNAL)) { in io_queue_worker_create()
414 * now set. wq exit does that too, but we can have added this in io_queue_worker_create()
417 if (test_bit(IO_WQ_BIT_EXIT, &wq->state)) in io_queue_worker_create()
418 io_wq_cancel_tw_create(wq); in io_queue_worker_create()
419 io_worker_ref_put(wq); in io_queue_worker_create()
422 io_worker_ref_put(wq); in io_queue_worker_create()
428 io_worker_ref_put(wq); in io_queue_worker_create()
459 struct io_wq *wq = worker->wq; in io_wq_dec_running() local
477 atomic_inc(&wq->worker_refs); in io_wq_dec_running()
507 static bool io_wait_on_hash(struct io_wq *wq, unsigned int hash) in io_wait_on_hash() argument
511 spin_lock_irq(&wq->hash->wait.lock); in io_wait_on_hash()
512 if (list_empty(&wq->wait.entry)) { in io_wait_on_hash()
513 __add_wait_queue(&wq->hash->wait, &wq->wait); in io_wait_on_hash()
514 if (!test_bit(hash, &wq->hash->map)) { in io_wait_on_hash()
516 list_del_init(&wq->wait.entry); in io_wait_on_hash()
520 spin_unlock_irq(&wq->hash->wait.lock); in io_wait_on_hash()
525 struct io_wq *wq) in io_get_next_work() argument
547 tail = wq->hash_tail[hash]; in io_get_next_work()
550 if (!test_and_set_bit(hash, &wq->hash->map)) { in io_get_next_work()
551 wq->hash_tail[hash] = NULL; in io_get_next_work()
570 unstalled = io_wait_on_hash(wq, stall_hash); in io_get_next_work()
574 if (wq_has_sleeper(&wq->hash->wait)) in io_get_next_work()
575 wake_up(&wq->hash->wait); in io_get_next_work()
602 struct io_wq *wq = worker->wq; in io_worker_handle_work() local
605 bool do_kill = test_bit(IO_WQ_BIT_EXIT, &wq->state); in io_worker_handle_work()
615 work = io_get_next_work(acct, wq); in io_worker_handle_work()
663 io_wq_enqueue(wq, linked); in io_worker_handle_work()
667 spin_lock_irq(&wq->hash->wait.lock); in io_worker_handle_work()
668 clear_bit(hash, &wq->hash->map); in io_worker_handle_work()
670 spin_unlock_irq(&wq->hash->wait.lock); in io_worker_handle_work()
671 if (wq_has_sleeper(&wq->hash->wait)) in io_worker_handle_work()
672 wake_up(&wq->hash->wait); in io_worker_handle_work()
686 struct io_wq *wq = worker->wq; in io_wq_worker() local
693 snprintf(buf, sizeof(buf), "iou-wrk-%d", wq->task->pid); in io_wq_worker()
696 while (!test_bit(IO_WQ_BIT_EXIT, &wq->state)) { in io_wq_worker()
711 * or if someone modified our affinity. If wq is marked in io_wq_worker()
713 * keeping io-wq workers around for tasks that no longer have in io_wq_worker()
717 test_bit(IO_WQ_BIT_EXIT_ON_IDLE, &wq->state)) { in io_wq_worker()
739 wq->cpu_mask); in io_wq_worker()
743 if (test_bit(IO_WQ_BIT_EXIT, &wq->state) && io_acct_run_queue(acct)) in io_wq_worker()
786 static void io_init_new_worker(struct io_wq *wq, struct io_wq_acct *acct, struct io_worker *worker, in io_init_new_worker() argument
791 set_cpus_allowed_ptr(tsk, wq->cpu_mask); in io_init_new_worker()
845 struct io_wq *wq; in create_worker_cont() local
850 wq = worker->wq; in create_worker_cont()
854 io_init_new_worker(wq, acct, worker, tsk); in create_worker_cont()
868 while (io_acct_cancel_pending_work(wq, acct, &match)) in create_worker_cont()
873 io_worker_ref_put(wq); in create_worker_cont()
893 static bool create_io_worker(struct io_wq *wq, struct io_wq_acct *acct) in create_io_worker() argument
907 io_worker_ref_put(wq); in create_io_worker()
912 worker->wq = wq; in create_io_worker()
919 io_init_new_worker(wq, acct, worker, tsk); in create_io_worker()
956 static void io_wq_for_each_worker(struct io_wq *wq, in io_wq_for_each_worker() argument
961 if (io_acct_for_each_worker(&wq->acct[i], func, data)) in io_wq_for_each_worker()
972 void io_wq_set_exit_on_idle(struct io_wq *wq, bool enable) in io_wq_set_exit_on_idle() argument
974 if (!wq->task) in io_wq_set_exit_on_idle()
978 clear_bit(IO_WQ_BIT_EXIT_ON_IDLE, &wq->state); in io_wq_set_exit_on_idle()
982 if (test_and_set_bit(IO_WQ_BIT_EXIT_ON_IDLE, &wq->state)) in io_wq_set_exit_on_idle()
986 io_wq_for_each_worker(wq, io_wq_worker_wake, NULL); in io_wq_set_exit_on_idle()
990 static void io_run_cancel(struct io_wq_work *work, struct io_wq *wq) in io_run_cancel() argument
999 static void io_wq_insert_work(struct io_wq *wq, struct io_wq_acct *acct, in io_wq_insert_work() argument
1012 tail = wq->hash_tail[hash]; in io_wq_insert_work()
1013 wq->hash_tail[hash] = work; in io_wq_insert_work()
1025 void io_wq_enqueue(struct io_wq *wq, struct io_wq_work *work) in io_wq_enqueue() argument
1028 struct io_wq_acct *acct = io_work_get_acct(wq, work_flags); in io_wq_enqueue()
1037 * If io-wq is exiting for this task, or if the request has explicitly in io_wq_enqueue()
1040 if (test_bit(IO_WQ_BIT_EXIT, &wq->state) || in io_wq_enqueue()
1042 io_run_cancel(work, wq); in io_wq_enqueue()
1047 io_wq_insert_work(wq, acct, work, work_flags); in io_wq_enqueue()
1059 did_create = io_wq_create_worker(wq, acct); in io_wq_enqueue()
1071 io_acct_cancel_pending_work(wq, acct, &match); in io_wq_enqueue()
1116 static inline void io_wq_remove_pending(struct io_wq *wq, in io_wq_remove_pending() argument
1124 if (io_wq_is_hashed(work) && work == wq->hash_tail[hash]) { in io_wq_remove_pending()
1129 wq->hash_tail[hash] = prev_work; in io_wq_remove_pending()
1131 wq->hash_tail[hash] = NULL; in io_wq_remove_pending()
1136 static bool io_acct_cancel_pending_work(struct io_wq *wq, in io_acct_cancel_pending_work() argument
1148 io_wq_remove_pending(wq, acct, work, prev); in io_acct_cancel_pending_work()
1150 io_run_cancel(work, wq); in io_acct_cancel_pending_work()
1160 static void io_wq_cancel_pending_work(struct io_wq *wq, in io_wq_cancel_pending_work() argument
1166 struct io_wq_acct *acct = io_get_acct(wq, i == 0); in io_wq_cancel_pending_work()
1168 if (io_acct_cancel_pending_work(wq, acct, match)) { in io_wq_cancel_pending_work()
1184 static void io_wq_cancel_running_work(struct io_wq *wq, in io_wq_cancel_running_work() argument
1190 io_acct_cancel_running_work(&wq->acct[i], match); in io_wq_cancel_running_work()
1195 enum io_wq_cancel io_wq_cancel_cb(struct io_wq *wq, work_cancel_fn *cancel, in io_wq_cancel_cb() argument
1217 io_wq_cancel_pending_work(wq, &match); in io_wq_cancel_cb()
1221 io_wq_cancel_running_work(wq, &match); in io_wq_cancel_cb()
1235 struct io_wq *wq = container_of(wait, struct io_wq, wait); in io_wq_hash_wake() local
1242 struct io_wq_acct *acct = &wq->acct[i]; in io_wq_hash_wake()
1254 struct io_wq *wq; in io_wq_create() local
1259 wq = kzalloc_obj(struct io_wq); in io_wq_create()
1260 if (!wq) in io_wq_create()
1264 wq->hash = data->hash; in io_wq_create()
1268 if (!alloc_cpumask_var(&wq->cpu_mask, GFP_KERNEL)) in io_wq_create()
1270 cpuset_cpus_allowed(data->task, wq->cpu_mask); in io_wq_create()
1271 wq->acct[IO_WQ_ACCT_BOUND].max_workers = bounded; in io_wq_create()
1272 wq->acct[IO_WQ_ACCT_UNBOUND].max_workers = in io_wq_create()
1274 INIT_LIST_HEAD(&wq->wait.entry); in io_wq_create()
1275 wq->wait.func = io_wq_hash_wake; in io_wq_create()
1277 struct io_wq_acct *acct = &wq->acct[i]; in io_wq_create()
1289 wq->task = get_task_struct(data->task); in io_wq_create()
1290 atomic_set(&wq->worker_refs, 1); in io_wq_create()
1291 init_completion(&wq->worker_done); in io_wq_create()
1292 ret = cpuhp_state_add_instance_nocalls(io_wq_online, &wq->cpuhp_node); in io_wq_create()
1294 put_task_struct(wq->task); in io_wq_create()
1298 return wq; in io_wq_create()
1301 free_cpumask_var(wq->cpu_mask); in io_wq_create()
1302 kfree(wq); in io_wq_create()
1313 return worker->wq == data; in io_task_work_match()
1316 void io_wq_exit_start(struct io_wq *wq) in io_wq_exit_start() argument
1318 set_bit(IO_WQ_BIT_EXIT, &wq->state); in io_wq_exit_start()
1321 static void io_wq_cancel_tw_create(struct io_wq *wq) in io_wq_cancel_tw_create() argument
1325 while ((cb = task_work_cancel_match(wq->task, io_task_work_match, wq)) != NULL) { in io_wq_cancel_tw_create()
1339 static void io_wq_exit_workers(struct io_wq *wq) in io_wq_exit_workers() argument
1343 if (!wq->task) in io_wq_exit_workers()
1346 io_wq_cancel_tw_create(wq); in io_wq_exit_workers()
1349 io_wq_for_each_worker(wq, io_wq_worker_wake, NULL); in io_wq_exit_workers()
1351 io_worker_ref_put(wq); in io_wq_exit_workers()
1359 * io-wq items can easily trigger the hung task timeout. Only sleep in io_wq_exit_workers()
1368 if (wait_for_completion_timeout(&wq->worker_done, timeout)) in io_wq_exit_workers()
1373 spin_lock_irq(&wq->hash->wait.lock); in io_wq_exit_workers()
1374 list_del_init(&wq->wait.entry); in io_wq_exit_workers()
1375 spin_unlock_irq(&wq->hash->wait.lock); in io_wq_exit_workers()
1377 put_task_struct(wq->task); in io_wq_exit_workers()
1378 wq->task = NULL; in io_wq_exit_workers()
1381 static void io_wq_destroy(struct io_wq *wq) in io_wq_destroy() argument
1388 cpuhp_state_remove_instance_nocalls(io_wq_online, &wq->cpuhp_node); in io_wq_destroy()
1389 io_wq_cancel_pending_work(wq, &match); in io_wq_destroy()
1390 free_cpumask_var(wq->cpu_mask); in io_wq_destroy()
1391 io_wq_put_hash(wq->hash); in io_wq_destroy()
1392 kfree(wq); in io_wq_destroy()
1395 void io_wq_put_and_exit(struct io_wq *wq) in io_wq_put_and_exit() argument
1397 WARN_ON_ONCE(!test_bit(IO_WQ_BIT_EXIT, &wq->state)); in io_wq_put_and_exit()
1399 io_wq_exit_workers(wq); in io_wq_put_and_exit()
1400 io_wq_destroy(wq); in io_wq_put_and_exit()
1413 cpumask_set_cpu(od->cpu, worker->wq->cpu_mask); in io_wq_worker_affinity()
1415 cpumask_clear_cpu(od->cpu, worker->wq->cpu_mask); in io_wq_worker_affinity()
1419 static int __io_wq_cpu_online(struct io_wq *wq, unsigned int cpu, bool online) in __io_wq_cpu_online() argument
1427 io_wq_for_each_worker(wq, io_wq_worker_affinity, &od); in __io_wq_cpu_online()
1434 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); in io_wq_cpu_online() local
1436 return __io_wq_cpu_online(wq, cpu, true); in io_wq_cpu_online()
1441 struct io_wq *wq = hlist_entry_safe(node, struct io_wq, cpuhp_node); in io_wq_cpu_offline() local
1443 return __io_wq_cpu_online(wq, cpu, false); in io_wq_cpu_offline()
1477 int io_wq_max_workers(struct io_wq *wq, int *new_count) in io_wq_max_workers() argument
1498 acct = &wq->acct[i]; in io_wq_max_workers()
1517 ret = cpuhp_setup_state_multi(CPUHP_AP_ONLINE_DYN, "io-wq/online", in io_wq_init()