1 /* 2 * Copyright (C) 2007 Oracle. All rights reserved. 3 * Copyright (C) 2014 Fujitsu. All rights reserved. 4 * 5 * This program is free software; you can redistribute it and/or 6 * modify it under the terms of the GNU General Public 7 * License v2 as published by the Free Software Foundation. 8 * 9 * This program is distributed in the hope that it will be useful, 10 * but WITHOUT ANY WARRANTY; without even the implied warranty of 11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 12 * General Public License for more details. 13 * 14 * You should have received a copy of the GNU General Public 15 * License along with this program; if not, write to the 16 * Free Software Foundation, Inc., 59 Temple Place - Suite 330, 17 * Boston, MA 021110-1307, USA. 18 */ 19 20 #include <linux/kthread.h> 21 #include <linux/slab.h> 22 #include <linux/list.h> 23 #include <linux/spinlock.h> 24 #include <linux/freezer.h> 25 #include "async-thread.h" 26 #include "ctree.h" 27 28 #define WORK_DONE_BIT 0 29 #define WORK_ORDER_DONE_BIT 1 30 #define WORK_HIGH_PRIO_BIT 2 31 32 #define NO_THRESHOLD (-1) 33 #define DFT_THRESHOLD (32) 34 35 struct __btrfs_workqueue { 36 struct workqueue_struct *normal_wq; 37 38 /* File system this workqueue services */ 39 struct btrfs_fs_info *fs_info; 40 41 /* List head pointing to ordered work list */ 42 struct list_head ordered_list; 43 44 /* Spinlock for ordered_list */ 45 spinlock_t list_lock; 46 47 /* Thresholding related variants */ 48 atomic_t pending; 49 50 /* Up limit of concurrency workers */ 51 int limit_active; 52 53 /* Current number of concurrency workers */ 54 int current_active; 55 56 /* Threshold to change current_active */ 57 int thresh; 58 unsigned int count; 59 spinlock_t thres_lock; 60 }; 61 62 struct btrfs_workqueue { 63 struct __btrfs_workqueue *normal; 64 struct __btrfs_workqueue *high; 65 }; 66 67 static void normal_work_helper(struct btrfs_work *work); 68 69 #define BTRFS_WORK_HELPER(name) \ 70 void btrfs_##name(struct work_struct *arg) \ 71 { \ 72 struct btrfs_work *work = container_of(arg, struct btrfs_work, \ 73 normal_work); \ 74 normal_work_helper(work); \ 75 } 76 77 struct btrfs_fs_info * 78 btrfs_workqueue_owner(struct __btrfs_workqueue *wq) 79 { 80 return wq->fs_info; 81 } 82 83 struct btrfs_fs_info * 84 btrfs_work_owner(struct btrfs_work *work) 85 { 86 return work->wq->fs_info; 87 } 88 89 bool btrfs_workqueue_normal_congested(struct btrfs_workqueue *wq) 90 { 91 /* 92 * We could compare wq->normal->pending with num_online_cpus() 93 * to support "thresh == NO_THRESHOLD" case, but it requires 94 * moving up atomic_inc/dec in thresh_queue/exec_hook. Let's 95 * postpone it until someone needs the support of that case. 96 */ 97 if (wq->normal->thresh == NO_THRESHOLD) 98 return false; 99 100 return atomic_read(&wq->normal->pending) > wq->normal->thresh * 2; 101 } 102 103 BTRFS_WORK_HELPER(worker_helper); 104 BTRFS_WORK_HELPER(delalloc_helper); 105 BTRFS_WORK_HELPER(flush_delalloc_helper); 106 BTRFS_WORK_HELPER(cache_helper); 107 BTRFS_WORK_HELPER(submit_helper); 108 BTRFS_WORK_HELPER(fixup_helper); 109 BTRFS_WORK_HELPER(endio_helper); 110 BTRFS_WORK_HELPER(endio_meta_helper); 111 BTRFS_WORK_HELPER(endio_meta_write_helper); 112 BTRFS_WORK_HELPER(endio_raid56_helper); 113 BTRFS_WORK_HELPER(endio_repair_helper); 114 BTRFS_WORK_HELPER(rmw_helper); 115 BTRFS_WORK_HELPER(endio_write_helper); 116 BTRFS_WORK_HELPER(freespace_write_helper); 117 BTRFS_WORK_HELPER(delayed_meta_helper); 118 BTRFS_WORK_HELPER(readahead_helper); 119 BTRFS_WORK_HELPER(qgroup_rescan_helper); 120 BTRFS_WORK_HELPER(extent_refs_helper); 121 BTRFS_WORK_HELPER(scrub_helper); 122 BTRFS_WORK_HELPER(scrubwrc_helper); 123 BTRFS_WORK_HELPER(scrubnc_helper); 124 BTRFS_WORK_HELPER(scrubparity_helper); 125 126 static struct __btrfs_workqueue * 127 __btrfs_alloc_workqueue(struct btrfs_fs_info *fs_info, const char *name, 128 unsigned int flags, int limit_active, int thresh) 129 { 130 struct __btrfs_workqueue *ret = kzalloc(sizeof(*ret), GFP_KERNEL); 131 132 if (!ret) 133 return NULL; 134 135 ret->fs_info = fs_info; 136 ret->limit_active = limit_active; 137 atomic_set(&ret->pending, 0); 138 if (thresh == 0) 139 thresh = DFT_THRESHOLD; 140 /* For low threshold, disabling threshold is a better choice */ 141 if (thresh < DFT_THRESHOLD) { 142 ret->current_active = limit_active; 143 ret->thresh = NO_THRESHOLD; 144 } else { 145 /* 146 * For threshold-able wq, let its concurrency grow on demand. 147 * Use minimal max_active at alloc time to reduce resource 148 * usage. 149 */ 150 ret->current_active = 1; 151 ret->thresh = thresh; 152 } 153 154 if (flags & WQ_HIGHPRI) 155 ret->normal_wq = alloc_workqueue("%s-%s-high", flags, 156 ret->current_active, "btrfs", 157 name); 158 else 159 ret->normal_wq = alloc_workqueue("%s-%s", flags, 160 ret->current_active, "btrfs", 161 name); 162 if (!ret->normal_wq) { 163 kfree(ret); 164 return NULL; 165 } 166 167 INIT_LIST_HEAD(&ret->ordered_list); 168 spin_lock_init(&ret->list_lock); 169 spin_lock_init(&ret->thres_lock); 170 trace_btrfs_workqueue_alloc(ret, name, flags & WQ_HIGHPRI); 171 return ret; 172 } 173 174 static inline void 175 __btrfs_destroy_workqueue(struct __btrfs_workqueue *wq); 176 177 struct btrfs_workqueue *btrfs_alloc_workqueue(struct btrfs_fs_info *fs_info, 178 const char *name, 179 unsigned int flags, 180 int limit_active, 181 int thresh) 182 { 183 struct btrfs_workqueue *ret = kzalloc(sizeof(*ret), GFP_KERNEL); 184 185 if (!ret) 186 return NULL; 187 188 ret->normal = __btrfs_alloc_workqueue(fs_info, name, 189 flags & ~WQ_HIGHPRI, 190 limit_active, thresh); 191 if (!ret->normal) { 192 kfree(ret); 193 return NULL; 194 } 195 196 if (flags & WQ_HIGHPRI) { 197 ret->high = __btrfs_alloc_workqueue(fs_info, name, flags, 198 limit_active, thresh); 199 if (!ret->high) { 200 __btrfs_destroy_workqueue(ret->normal); 201 kfree(ret); 202 return NULL; 203 } 204 } 205 return ret; 206 } 207 208 /* 209 * Hook for threshold which will be called in btrfs_queue_work. 210 * This hook WILL be called in IRQ handler context, 211 * so workqueue_set_max_active MUST NOT be called in this hook 212 */ 213 static inline void thresh_queue_hook(struct __btrfs_workqueue *wq) 214 { 215 if (wq->thresh == NO_THRESHOLD) 216 return; 217 atomic_inc(&wq->pending); 218 } 219 220 /* 221 * Hook for threshold which will be called before executing the work, 222 * This hook is called in kthread content. 223 * So workqueue_set_max_active is called here. 224 */ 225 static inline void thresh_exec_hook(struct __btrfs_workqueue *wq) 226 { 227 int new_current_active; 228 long pending; 229 int need_change = 0; 230 231 if (wq->thresh == NO_THRESHOLD) 232 return; 233 234 atomic_dec(&wq->pending); 235 spin_lock(&wq->thres_lock); 236 /* 237 * Use wq->count to limit the calling frequency of 238 * workqueue_set_max_active. 239 */ 240 wq->count++; 241 wq->count %= (wq->thresh / 4); 242 if (!wq->count) 243 goto out; 244 new_current_active = wq->current_active; 245 246 /* 247 * pending may be changed later, but it's OK since we really 248 * don't need it so accurate to calculate new_max_active. 249 */ 250 pending = atomic_read(&wq->pending); 251 if (pending > wq->thresh) 252 new_current_active++; 253 if (pending < wq->thresh / 2) 254 new_current_active--; 255 new_current_active = clamp_val(new_current_active, 1, wq->limit_active); 256 if (new_current_active != wq->current_active) { 257 need_change = 1; 258 wq->current_active = new_current_active; 259 } 260 out: 261 spin_unlock(&wq->thres_lock); 262 263 if (need_change) { 264 workqueue_set_max_active(wq->normal_wq, wq->current_active); 265 } 266 } 267 268 static void run_ordered_work(struct __btrfs_workqueue *wq) 269 { 270 struct list_head *list = &wq->ordered_list; 271 struct btrfs_work *work; 272 spinlock_t *lock = &wq->list_lock; 273 unsigned long flags; 274 275 while (1) { 276 spin_lock_irqsave(lock, flags); 277 if (list_empty(list)) 278 break; 279 work = list_entry(list->next, struct btrfs_work, 280 ordered_list); 281 if (!test_bit(WORK_DONE_BIT, &work->flags)) 282 break; 283 284 /* 285 * we are going to call the ordered done function, but 286 * we leave the work item on the list as a barrier so 287 * that later work items that are done don't have their 288 * functions called before this one returns 289 */ 290 if (test_and_set_bit(WORK_ORDER_DONE_BIT, &work->flags)) 291 break; 292 trace_btrfs_ordered_sched(work); 293 spin_unlock_irqrestore(lock, flags); 294 work->ordered_func(work); 295 296 /* now take the lock again and drop our item from the list */ 297 spin_lock_irqsave(lock, flags); 298 list_del(&work->ordered_list); 299 spin_unlock_irqrestore(lock, flags); 300 301 /* 302 * we don't want to call the ordered free functions 303 * with the lock held though 304 */ 305 work->ordered_free(work); 306 trace_btrfs_all_work_done(work); 307 } 308 spin_unlock_irqrestore(lock, flags); 309 } 310 311 static void normal_work_helper(struct btrfs_work *work) 312 { 313 struct __btrfs_workqueue *wq; 314 int need_order = 0; 315 316 /* 317 * We should not touch things inside work in the following cases: 318 * 1) after work->func() if it has no ordered_free 319 * Since the struct is freed in work->func(). 320 * 2) after setting WORK_DONE_BIT 321 * The work may be freed in other threads almost instantly. 322 * So we save the needed things here. 323 */ 324 if (work->ordered_func) 325 need_order = 1; 326 wq = work->wq; 327 328 trace_btrfs_work_sched(work); 329 thresh_exec_hook(wq); 330 work->func(work); 331 if (need_order) { 332 set_bit(WORK_DONE_BIT, &work->flags); 333 run_ordered_work(wq); 334 } 335 if (!need_order) 336 trace_btrfs_all_work_done(work); 337 } 338 339 void btrfs_init_work(struct btrfs_work *work, btrfs_work_func_t uniq_func, 340 btrfs_func_t func, 341 btrfs_func_t ordered_func, 342 btrfs_func_t ordered_free) 343 { 344 work->func = func; 345 work->ordered_func = ordered_func; 346 work->ordered_free = ordered_free; 347 INIT_WORK(&work->normal_work, uniq_func); 348 INIT_LIST_HEAD(&work->ordered_list); 349 work->flags = 0; 350 } 351 352 static inline void __btrfs_queue_work(struct __btrfs_workqueue *wq, 353 struct btrfs_work *work) 354 { 355 unsigned long flags; 356 357 work->wq = wq; 358 thresh_queue_hook(wq); 359 if (work->ordered_func) { 360 spin_lock_irqsave(&wq->list_lock, flags); 361 list_add_tail(&work->ordered_list, &wq->ordered_list); 362 spin_unlock_irqrestore(&wq->list_lock, flags); 363 } 364 trace_btrfs_work_queued(work); 365 queue_work(wq->normal_wq, &work->normal_work); 366 } 367 368 void btrfs_queue_work(struct btrfs_workqueue *wq, 369 struct btrfs_work *work) 370 { 371 struct __btrfs_workqueue *dest_wq; 372 373 if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags) && wq->high) 374 dest_wq = wq->high; 375 else 376 dest_wq = wq->normal; 377 __btrfs_queue_work(dest_wq, work); 378 } 379 380 static inline void 381 __btrfs_destroy_workqueue(struct __btrfs_workqueue *wq) 382 { 383 destroy_workqueue(wq->normal_wq); 384 trace_btrfs_workqueue_destroy(wq); 385 kfree(wq); 386 } 387 388 void btrfs_destroy_workqueue(struct btrfs_workqueue *wq) 389 { 390 if (!wq) 391 return; 392 if (wq->high) 393 __btrfs_destroy_workqueue(wq->high); 394 __btrfs_destroy_workqueue(wq->normal); 395 kfree(wq); 396 } 397 398 void btrfs_workqueue_set_max(struct btrfs_workqueue *wq, int limit_active) 399 { 400 if (!wq) 401 return; 402 wq->normal->limit_active = limit_active; 403 if (wq->high) 404 wq->high->limit_active = limit_active; 405 } 406 407 void btrfs_set_work_high_priority(struct btrfs_work *work) 408 { 409 set_bit(WORK_HIGH_PRIO_BIT, &work->flags); 410 } 411