async-thread.c 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419
  1. // SPDX-License-Identifier: GPL-2.0
  2. /*
  3. * Copyright (C) 2007 Oracle. All rights reserved.
  4. * Copyright (C) 2014 Fujitsu. All rights reserved.
  5. */
  6. #include <linux/kthread.h>
  7. #include <linux/slab.h>
  8. #include <linux/list.h>
  9. #include <linux/spinlock.h>
  10. #include <linux/freezer.h>
  11. #include "async-thread.h"
  12. #include "ctree.h"
  13. enum {
  14. WORK_DONE_BIT,
  15. WORK_ORDER_DONE_BIT,
  16. WORK_HIGH_PRIO_BIT,
  17. };
  18. #define NO_THRESHOLD (-1)
  19. #define DFT_THRESHOLD (32)
  20. struct __btrfs_workqueue {
  21. struct workqueue_struct *normal_wq;
  22. /* File system this workqueue services */
  23. struct btrfs_fs_info *fs_info;
  24. /* List head pointing to ordered work list */
  25. struct list_head ordered_list;
  26. /* Spinlock for ordered_list */
  27. spinlock_t list_lock;
  28. /* Thresholding related variants */
  29. atomic_t pending;
  30. /* Up limit of concurrency workers */
  31. int limit_active;
  32. /* Current number of concurrency workers */
  33. int current_active;
  34. /* Threshold to change current_active */
  35. int thresh;
  36. unsigned int count;
  37. spinlock_t thres_lock;
  38. };
  39. struct btrfs_workqueue {
  40. struct __btrfs_workqueue *normal;
  41. struct __btrfs_workqueue *high;
  42. };
  43. struct btrfs_fs_info * __pure btrfs_workqueue_owner(const struct __btrfs_workqueue *wq)
  44. {
  45. return wq->fs_info;
  46. }
  47. struct btrfs_fs_info * __pure btrfs_work_owner(const struct btrfs_work *work)
  48. {
  49. return work->wq->fs_info;
  50. }
  51. bool btrfs_workqueue_normal_congested(const struct btrfs_workqueue *wq)
  52. {
  53. /*
  54. * We could compare wq->normal->pending with num_online_cpus()
  55. * to support "thresh == NO_THRESHOLD" case, but it requires
  56. * moving up atomic_inc/dec in thresh_queue/exec_hook. Let's
  57. * postpone it until someone needs the support of that case.
  58. */
  59. if (wq->normal->thresh == NO_THRESHOLD)
  60. return false;
  61. return atomic_read(&wq->normal->pending) > wq->normal->thresh * 2;
  62. }
  63. static struct __btrfs_workqueue *
  64. __btrfs_alloc_workqueue(struct btrfs_fs_info *fs_info, const char *name,
  65. unsigned int flags, int limit_active, int thresh)
  66. {
  67. struct __btrfs_workqueue *ret = kzalloc(sizeof(*ret), GFP_KERNEL);
  68. if (!ret)
  69. return NULL;
  70. ret->fs_info = fs_info;
  71. ret->limit_active = limit_active;
  72. atomic_set(&ret->pending, 0);
  73. if (thresh == 0)
  74. thresh = DFT_THRESHOLD;
  75. /* For low threshold, disabling threshold is a better choice */
  76. if (thresh < DFT_THRESHOLD) {
  77. ret->current_active = limit_active;
  78. ret->thresh = NO_THRESHOLD;
  79. } else {
  80. /*
  81. * For threshold-able wq, let its concurrency grow on demand.
  82. * Use minimal max_active at alloc time to reduce resource
  83. * usage.
  84. */
  85. ret->current_active = 1;
  86. ret->thresh = thresh;
  87. }
  88. if (flags & WQ_HIGHPRI)
  89. ret->normal_wq = alloc_workqueue("btrfs-%s-high", flags,
  90. ret->current_active, name);
  91. else
  92. ret->normal_wq = alloc_workqueue("btrfs-%s", flags,
  93. ret->current_active, name);
  94. if (!ret->normal_wq) {
  95. kfree(ret);
  96. return NULL;
  97. }
  98. INIT_LIST_HEAD(&ret->ordered_list);
  99. spin_lock_init(&ret->list_lock);
  100. spin_lock_init(&ret->thres_lock);
  101. trace_btrfs_workqueue_alloc(ret, name, flags & WQ_HIGHPRI);
  102. return ret;
  103. }
  104. static inline void
  105. __btrfs_destroy_workqueue(struct __btrfs_workqueue *wq);
  106. struct btrfs_workqueue *btrfs_alloc_workqueue(struct btrfs_fs_info *fs_info,
  107. const char *name,
  108. unsigned int flags,
  109. int limit_active,
  110. int thresh)
  111. {
  112. struct btrfs_workqueue *ret = kzalloc(sizeof(*ret), GFP_KERNEL);
  113. if (!ret)
  114. return NULL;
  115. ret->normal = __btrfs_alloc_workqueue(fs_info, name,
  116. flags & ~WQ_HIGHPRI,
  117. limit_active, thresh);
  118. if (!ret->normal) {
  119. kfree(ret);
  120. return NULL;
  121. }
  122. if (flags & WQ_HIGHPRI) {
  123. ret->high = __btrfs_alloc_workqueue(fs_info, name, flags,
  124. limit_active, thresh);
  125. if (!ret->high) {
  126. __btrfs_destroy_workqueue(ret->normal);
  127. kfree(ret);
  128. return NULL;
  129. }
  130. }
  131. return ret;
  132. }
  133. /*
  134. * Hook for threshold which will be called in btrfs_queue_work.
  135. * This hook WILL be called in IRQ handler context,
  136. * so workqueue_set_max_active MUST NOT be called in this hook
  137. */
  138. static inline void thresh_queue_hook(struct __btrfs_workqueue *wq)
  139. {
  140. if (wq->thresh == NO_THRESHOLD)
  141. return;
  142. atomic_inc(&wq->pending);
  143. }
  144. /*
  145. * Hook for threshold which will be called before executing the work,
  146. * This hook is called in kthread content.
  147. * So workqueue_set_max_active is called here.
  148. */
  149. static inline void thresh_exec_hook(struct __btrfs_workqueue *wq)
  150. {
  151. int new_current_active;
  152. long pending;
  153. int need_change = 0;
  154. if (wq->thresh == NO_THRESHOLD)
  155. return;
  156. atomic_dec(&wq->pending);
  157. spin_lock(&wq->thres_lock);
  158. /*
  159. * Use wq->count to limit the calling frequency of
  160. * workqueue_set_max_active.
  161. */
  162. wq->count++;
  163. wq->count %= (wq->thresh / 4);
  164. if (!wq->count)
  165. goto out;
  166. new_current_active = wq->current_active;
  167. /*
  168. * pending may be changed later, but it's OK since we really
  169. * don't need it so accurate to calculate new_max_active.
  170. */
  171. pending = atomic_read(&wq->pending);
  172. if (pending > wq->thresh)
  173. new_current_active++;
  174. if (pending < wq->thresh / 2)
  175. new_current_active--;
  176. new_current_active = clamp_val(new_current_active, 1, wq->limit_active);
  177. if (new_current_active != wq->current_active) {
  178. need_change = 1;
  179. wq->current_active = new_current_active;
  180. }
  181. out:
  182. spin_unlock(&wq->thres_lock);
  183. if (need_change) {
  184. workqueue_set_max_active(wq->normal_wq, wq->current_active);
  185. }
  186. }
  187. static void run_ordered_work(struct __btrfs_workqueue *wq,
  188. struct btrfs_work *self)
  189. {
  190. struct list_head *list = &wq->ordered_list;
  191. struct btrfs_work *work;
  192. spinlock_t *lock = &wq->list_lock;
  193. unsigned long flags;
  194. bool free_self = false;
  195. while (1) {
  196. spin_lock_irqsave(lock, flags);
  197. if (list_empty(list))
  198. break;
  199. work = list_entry(list->next, struct btrfs_work,
  200. ordered_list);
  201. if (!test_bit(WORK_DONE_BIT, &work->flags))
  202. break;
  203. /*
  204. * Orders all subsequent loads after reading WORK_DONE_BIT,
  205. * paired with the smp_mb__before_atomic in btrfs_work_helper
  206. * this guarantees that the ordered function will see all
  207. * updates from ordinary work function.
  208. */
  209. smp_rmb();
  210. /*
  211. * we are going to call the ordered done function, but
  212. * we leave the work item on the list as a barrier so
  213. * that later work items that are done don't have their
  214. * functions called before this one returns
  215. */
  216. if (test_and_set_bit(WORK_ORDER_DONE_BIT, &work->flags))
  217. break;
  218. trace_btrfs_ordered_sched(work);
  219. spin_unlock_irqrestore(lock, flags);
  220. work->ordered_func(work);
  221. /* now take the lock again and drop our item from the list */
  222. spin_lock_irqsave(lock, flags);
  223. list_del(&work->ordered_list);
  224. spin_unlock_irqrestore(lock, flags);
  225. if (work == self) {
  226. /*
  227. * This is the work item that the worker is currently
  228. * executing.
  229. *
  230. * The kernel workqueue code guarantees non-reentrancy
  231. * of work items. I.e., if a work item with the same
  232. * address and work function is queued twice, the second
  233. * execution is blocked until the first one finishes. A
  234. * work item may be freed and recycled with the same
  235. * work function; the workqueue code assumes that the
  236. * original work item cannot depend on the recycled work
  237. * item in that case (see find_worker_executing_work()).
  238. *
  239. * Note that different types of Btrfs work can depend on
  240. * each other, and one type of work on one Btrfs
  241. * filesystem may even depend on the same type of work
  242. * on another Btrfs filesystem via, e.g., a loop device.
  243. * Therefore, we must not allow the current work item to
  244. * be recycled until we are really done, otherwise we
  245. * break the above assumption and can deadlock.
  246. */
  247. free_self = true;
  248. } else {
  249. /*
  250. * We don't want to call the ordered free functions with
  251. * the lock held.
  252. */
  253. work->ordered_free(work);
  254. /* NB: work must not be dereferenced past this point. */
  255. trace_btrfs_all_work_done(wq->fs_info, work);
  256. }
  257. }
  258. spin_unlock_irqrestore(lock, flags);
  259. if (free_self) {
  260. self->ordered_free(self);
  261. /* NB: self must not be dereferenced past this point. */
  262. trace_btrfs_all_work_done(wq->fs_info, self);
  263. }
  264. }
  265. static void btrfs_work_helper(struct work_struct *normal_work)
  266. {
  267. struct btrfs_work *work = container_of(normal_work, struct btrfs_work,
  268. normal_work);
  269. struct __btrfs_workqueue *wq;
  270. int need_order = 0;
  271. /*
  272. * We should not touch things inside work in the following cases:
  273. * 1) after work->func() if it has no ordered_free
  274. * Since the struct is freed in work->func().
  275. * 2) after setting WORK_DONE_BIT
  276. * The work may be freed in other threads almost instantly.
  277. * So we save the needed things here.
  278. */
  279. if (work->ordered_func)
  280. need_order = 1;
  281. wq = work->wq;
  282. trace_btrfs_work_sched(work);
  283. thresh_exec_hook(wq);
  284. work->func(work);
  285. if (need_order) {
  286. /*
  287. * Ensures all memory accesses done in the work function are
  288. * ordered before setting the WORK_DONE_BIT. Ensuring the thread
  289. * which is going to executed the ordered work sees them.
  290. * Pairs with the smp_rmb in run_ordered_work.
  291. */
  292. smp_mb__before_atomic();
  293. set_bit(WORK_DONE_BIT, &work->flags);
  294. run_ordered_work(wq, work);
  295. } else {
  296. /* NB: work must not be dereferenced past this point. */
  297. trace_btrfs_all_work_done(wq->fs_info, work);
  298. }
  299. }
  300. void btrfs_init_work(struct btrfs_work *work, btrfs_func_t func,
  301. btrfs_func_t ordered_func, btrfs_func_t ordered_free)
  302. {
  303. work->func = func;
  304. work->ordered_func = ordered_func;
  305. work->ordered_free = ordered_free;
  306. INIT_WORK(&work->normal_work, btrfs_work_helper);
  307. INIT_LIST_HEAD(&work->ordered_list);
  308. work->flags = 0;
  309. }
  310. static inline void __btrfs_queue_work(struct __btrfs_workqueue *wq,
  311. struct btrfs_work *work)
  312. {
  313. unsigned long flags;
  314. work->wq = wq;
  315. thresh_queue_hook(wq);
  316. if (work->ordered_func) {
  317. spin_lock_irqsave(&wq->list_lock, flags);
  318. list_add_tail(&work->ordered_list, &wq->ordered_list);
  319. spin_unlock_irqrestore(&wq->list_lock, flags);
  320. }
  321. trace_btrfs_work_queued(work);
  322. queue_work(wq->normal_wq, &work->normal_work);
  323. }
  324. void btrfs_queue_work(struct btrfs_workqueue *wq,
  325. struct btrfs_work *work)
  326. {
  327. struct __btrfs_workqueue *dest_wq;
  328. if (test_bit(WORK_HIGH_PRIO_BIT, &work->flags) && wq->high)
  329. dest_wq = wq->high;
  330. else
  331. dest_wq = wq->normal;
  332. __btrfs_queue_work(dest_wq, work);
  333. }
  334. static inline void
  335. __btrfs_destroy_workqueue(struct __btrfs_workqueue *wq)
  336. {
  337. destroy_workqueue(wq->normal_wq);
  338. trace_btrfs_workqueue_destroy(wq);
  339. kfree(wq);
  340. }
  341. void btrfs_destroy_workqueue(struct btrfs_workqueue *wq)
  342. {
  343. if (!wq)
  344. return;
  345. if (wq->high)
  346. __btrfs_destroy_workqueue(wq->high);
  347. __btrfs_destroy_workqueue(wq->normal);
  348. kfree(wq);
  349. }
  350. void btrfs_workqueue_set_max(struct btrfs_workqueue *wq, int limit_active)
  351. {
  352. if (!wq)
  353. return;
  354. wq->normal->limit_active = limit_active;
  355. if (wq->high)
  356. wq->high->limit_active = limit_active;
  357. }
  358. void btrfs_set_work_high_priority(struct btrfs_work *work)
  359. {
  360. set_bit(WORK_HIGH_PRIO_BIT, &work->flags);
  361. }
  362. void btrfs_flush_workqueue(struct btrfs_workqueue *wq)
  363. {
  364. if (wq->high)
  365. flush_workqueue(wq->high->normal_wq);
  366. flush_workqueue(wq->normal->normal_wq);
  367. }