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. */
 37static void
 38xfs_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 */
 59int
 60xfs_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. */
 87void
 88xfs_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. */
 99int
100xfs_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 */
112void
113xfs_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 */
125unsigned int
126xfs_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_queue) ? 2 : 1;
136}