1 // SPDX-License-Identifier: GPL-2.0-or-later 2 /* 3 * Copyright (C) 2019 Oracle. All Rights Reserved. 4 * Author: Darrick J. Wong <darrick.wong@oracle.com> 5 */ 6 #include "xfs.h" 7 #include "xfs_fs.h" 8 #include "xfs_shared.h" 9 #include "xfs_format.h" 10 #include "xfs_log_format.h" 11 #include "xfs_trans_resv.h" 12 #include "xfs_mount.h" 13 #include "xfs_trace.h" 14 #include "xfs_sysctl.h" 15 #include "xfs_pwork.h" 16 #include <linux/nmi.h> 17 18 /* 19 * Parallel Work Queue 20 * =================== 21 * 22 * Abstract away the details of running a large and "obviously" parallelizable 23 * task across multiple CPUs. Callers initialize the pwork control object with 24 * a desired level of parallelization and a work function. Next, they embed 25 * struct xfs_pwork in whatever structure they use to pass work context to a 26 * worker thread and queue that pwork. The work function will be passed the 27 * pwork item when it is run (from process context) and any returned error will 28 * be recorded in xfs_pwork_ctl.error. Work functions should check for errors 29 * and abort if necessary; the non-zeroness of xfs_pwork_ctl.error does not 30 * stop workqueue item processing. 31 * 32 * This is the rough equivalent of the xfsprogs workqueue code, though we can't 33 * reuse that name here. 34 */ 35 36 /* Invoke our caller's function. */ 37 static void 38 xfs_pwork_work( 39 struct work_struct *work) 40 { 41 struct xfs_pwork *pwork; 42 struct xfs_pwork_ctl *pctl; 43 int error; 44 45 pwork = container_of(work, struct xfs_pwork, work); 46 pctl = pwork->pctl; 47 error = pctl->work_fn(pctl->mp, pwork); 48 if (error && !pctl->error) 49 pctl->error = error; 50 if (atomic_dec_and_test(&pctl->nr_work)) 51 wake_up(&pctl->poll_wait); 52 } 53 54 /* 55 * Set up control data for parallel work. @work_fn is the function that will 56 * be called. @tag will be written into the kernel threads. @nr_threads is 57 * the level of parallelism desired, or 0 for no limit. 58 */ 59 int 60 xfs_pwork_init( 61 struct xfs_mount *mp, 62 struct xfs_pwork_ctl *pctl, 63 xfs_pwork_work_fn work_fn, 64 const char *tag, 65 unsigned int nr_threads) 66 { 67 #ifdef DEBUG 68 if (xfs_globals.pwork_threads >= 0) 69 nr_threads = xfs_globals.pwork_threads; 70 #endif 71 trace_xfs_pwork_init(mp, nr_threads, current->pid); 72 73 pctl->wq = alloc_workqueue("%s-%d", WQ_FREEZABLE, nr_threads, tag, 74 current->pid); 75 if (!pctl->wq) 76 return -ENOMEM; 77 pctl->work_fn = work_fn; 78 pctl->error = 0; 79 pctl->mp = mp; 80 atomic_set(&pctl->nr_work, 0); 81 init_waitqueue_head(&pctl->poll_wait); 82 83 return 0; 84 } 85 86 /* Queue some parallel work. */ 87 void 88 xfs_pwork_queue( 89 struct xfs_pwork_ctl *pctl, 90 struct xfs_pwork *pwork) 91 { 92 INIT_WORK(&pwork->work, xfs_pwork_work); 93 pwork->pctl = pctl; 94 atomic_inc(&pctl->nr_work); 95 queue_work(pctl->wq, &pwork->work); 96 } 97 98 /* Wait for the work to finish and tear down the control structure. */ 99 int 100 xfs_pwork_destroy( 101 struct xfs_pwork_ctl *pctl) 102 { 103 destroy_workqueue(pctl->wq); 104 pctl->wq = NULL; 105 return pctl->error; 106 } 107 108 /* 109 * Wait for the work to finish by polling completion status and touch the soft 110 * lockup watchdog. This is for callers such as mount which hold locks. 111 */ 112 void 113 xfs_pwork_poll( 114 struct xfs_pwork_ctl *pctl) 115 { 116 while (wait_event_timeout(pctl->poll_wait, 117 atomic_read(&pctl->nr_work) == 0, HZ) == 0) 118 touch_softlockup_watchdog(); 119 } 120 121 /* 122 * Return the amount of parallelism that the data device can handle, or 0 for 123 * no limit. 124 */ 125 unsigned int 126 xfs_pwork_guess_datadev_parallelism( 127 struct xfs_mount *mp) 128 { 129 struct xfs_buftarg *btp = mp->m_ddev_targp; 130 131 /* 132 * For now we'll go with the most conservative setting possible, 133 * which is two threads for an SSD and 1 thread everywhere else. 134 */ 135 return blk_queue_nonrot(btp->bt_bdev->bd_disk->queue) ? 2 : 1; 136 } 137