[thirdparty/xfsprogs-dev.git] / scrub / xfs_scrub_all.in

#!/usr/bin/python3

# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright (C) 2018-2024 Oracle.  All rights reserved.
#
# Author: Darrick J. Wong <djwong@kernel.org>

# Run online scrubbers in parallel, but avoid thrashing.

import subprocess
import json
import threading
import time
import sys
import os
import argparse
from io import TextIOWrapper

retcode = 0
terminate = False

def DEVNULL():
	'''Return /dev/null in subprocess writable format.'''
	try:
		from subprocess import DEVNULL
		return DEVNULL
	except ImportError:
		return open(os.devnull, 'wb')

def find_mounts():
	'''Map mountpoints to physical disks.'''
	def find_xfs_mounts(bdev, fs, lastdisk):
		'''Attach lastdisk to each fs found under bdev.'''
		if bdev['fstype'] == 'xfs' and bdev['mountpoint'] is not None:
			mnt = bdev['mountpoint']
			if mnt in fs:
				fs[mnt].add(lastdisk)
			else:
				fs[mnt] = set([lastdisk])
		if 'children' not in bdev:
			return
		for child in bdev['children']:
			find_xfs_mounts(child, fs, lastdisk)

	fs = {}
	cmd=['lsblk', '-o', 'NAME,KNAME,TYPE,FSTYPE,MOUNTPOINT', '-J']
	result = subprocess.Popen(cmd, stdout=subprocess.PIPE)
	result.wait()
	if result.returncode != 0:
		return fs
	sarray = [x.decode(sys.stdout.encoding) for x in result.stdout.readlines()]
	output = ' '.join(sarray)
	bdevdata = json.loads(output)

	# The lsblk output had better be in disks-then-partitions order
	for bdev in bdevdata['blockdevices']:
		lastdisk = bdev['kname']
		find_xfs_mounts(bdev, fs, lastdisk)

	return fs

def backtick(cmd):
	'''Generator function that yields lines of a program's stdout.'''
	p = subprocess.Popen(cmd, stdout = subprocess.PIPE)
	for line in TextIOWrapper(p.stdout, encoding="utf-8"):
		yield line.strip()

def remove_killfunc(killfuncs, fn):
	'''Ensure fn is not in killfuncs.'''
	try:
		killfuncs.remove(fn)
	except:
		pass

def run_killable(cmd, stdout, killfuncs):
	'''Run a killable program.  Returns program retcode or -1 if we can't
	start it.'''
	try:
		proc = subprocess.Popen(cmd, stdout = stdout)
		killfuncs.add(proc.terminate)
		proc.wait()
		remove_killfunc(killfuncs, proc.terminate)
		return proc.returncode
	except:
		return -1

# systemd doesn't like unit instance names with slashes in them, so it
# replaces them with dashes when it invokes the service.  Filesystem paths
# need a special --path argument so that dashes do not get mangled.
def path_to_serviceunit(path):
	'''Convert a pathname into a systemd service unit name.'''

	cmd = ['systemd-escape', '--template', '@scrub_svcname@',
	       '--path', path]
	try:
		proc = subprocess.Popen(cmd, stdout = subprocess.PIPE)
		proc.wait()
		for line in proc.stdout:
			return line.decode(sys.stdout.encoding).strip()
	except:
		return None

def systemctl_stop(unitname):
	'''Stop a systemd unit.'''
	cmd = ['systemctl', 'stop', unitname]
	x = subprocess.Popen(cmd)
	x.wait()

def systemctl_start(unitname, killfuncs):
	'''Start a systemd unit and wait for it to complete.'''
	stop_fn = None
	cmd = ['systemctl', 'start', unitname]
	try:
		proc = subprocess.Popen(cmd, stdout = DEVNULL())
		stop_fn = lambda: systemctl_stop(unitname)
		killfuncs.add(stop_fn)
		proc.wait()
		ret = proc.returncode
	except:
		if stop_fn is not None:
			remove_killfunc(killfuncs, stop_fn)
		return -1

	if ret != 1:
		remove_killfunc(killfuncs, stop_fn)
		return ret

	# If systemctl-start returns 1, it's possible that the service failed
	# or that dbus/systemd restarted and the client program lost its
	# connection -- according to the systemctl man page, 1 means "unit not
	# failed".
	#
	# Either way, we switch to polling the service status to try to wait
	# for the service to end.  As of systemd 249, the is-active command
	# returns any of the following states: active, reloading, inactive,
	# failed, activating, deactivating, or maintenance.  Apparently these
	# strings are not localized.
	while True:
		try:
			for l in backtick(['systemctl', 'is-active', unitname]):
				if l == 'failed':
					remove_killfunc(killfuncs, stop_fn)
					return 1
				if l == 'inactive':
					remove_killfunc(killfuncs, stop_fn)
					return 0
		except:
			remove_killfunc(killfuncs, stop_fn)
			return -1

		time.sleep(1)

def run_scrub(mnt, cond, running_devs, mntdevs, killfuncs):
	'''Run a scrub process.'''
	global retcode, terminate

	print("Scrubbing %s..." % mnt)
	sys.stdout.flush()

	try:
		if terminate:
			return

		# Try it the systemd way
		unitname = path_to_serviceunit(path)
		if unitname is not None:
			ret = systemctl_start(unitname, killfuncs)
			if ret == 0 or ret == 1:
				print("Scrubbing %s done, (err=%d)" % (mnt, ret))
				sys.stdout.flush()
				retcode |= ret
				return

			if terminate:
				return

		# Invoke xfs_scrub manually
		cmd = ['@sbindir@/xfs_scrub']
		cmd += '@scrub_args@'.split()
		cmd += [mnt]
		ret = run_killable(cmd, None, killfuncs)
		if ret >= 0:
			print("Scrubbing %s done, (err=%d)" % (mnt, ret))
			sys.stdout.flush()
			retcode |= ret
			return

		if terminate:
			return

		print("Unable to start scrub tool.")
		sys.stdout.flush()
	finally:
		running_devs -= mntdevs
		cond.acquire()
		cond.notify()
		cond.release()

def main():
	'''Find mounts, schedule scrub runs.'''
	def thr(mnt, devs):
		a = (mnt, cond, running_devs, devs, killfuncs)
		thr = threading.Thread(target = run_scrub, args = a)
		thr.start()
	global retcode, terminate

	parser = argparse.ArgumentParser( \
			description = "Scrub all mounted XFS filesystems.")
	parser.add_argument("-V", help = "Report version and exit.", \
			action = "store_true")
	args = parser.parse_args()

	if args.V:
		print("xfs_scrub_all version @pkg_version@")
		sys.exit(0)

	fs = find_mounts()

	# Tail the journal if we ourselves aren't a service...
	journalthread = None
	if 'SERVICE_MODE' not in os.environ:
		try:
			cmd=['journalctl', '--no-pager', '-q', '-S', 'now', \
					'-f', '-u', 'xfs_scrub@*', '-o', \
					'cat']
			journalthread = subprocess.Popen(cmd)
		except:
			pass

	# Schedule scrub jobs...
	running_devs = set()
	killfuncs = set()
	cond = threading.Condition()
	while len(fs) > 0:
		if len(running_devs) == 0:
			mnt, devs = fs.popitem()
			running_devs.update(devs)
			thr(mnt, devs)
		poppers = set()
		for mnt in fs:
			devs = fs[mnt]
			can_run = True
			for dev in devs:
				if dev in running_devs:
					can_run = False
					break
			if can_run:
				running_devs.update(devs)
				poppers.add(mnt)
				thr(mnt, devs)
		for p in poppers:
			fs.pop(p)
		cond.acquire()
		try:
			cond.wait()
		except KeyboardInterrupt:
			terminate = True
			print("Terminating...")
			sys.stdout.flush()
			while len(killfuncs) > 0:
				fn = killfuncs.pop()
				fn()
			fs = []
		cond.release()

	if journalthread is not None:
		journalthread.terminate()

	# See the service mode comments in xfs_scrub.c for why we do this.
	if 'SERVICE_MODE' in os.environ:
		time.sleep(2)
		if retcode != 0:
			retcode = 1

	sys.exit(retcode)

if __name__ == '__main__':
	main()
Commit	Line	Data
9d50331a	1	#!/usr/bin/python3
f1dca11c	2
8d318d62	3	# SPDX-License-Identifier: GPL-2.0-or-later
52520522	4	# Copyright (C) 2018-2024 Oracle. All rights reserved.
f1dca11c	5	#
8d318d62	6	# Author: Darrick J. Wong <djwong@kernel.org>
959ef981 DC	7
959ef981 DC	8	# Run online scrubbers in parallel, but avoid thrashing.
f1dca11c DW	9
	10	import subprocess
	11	import json
	12	import threading
	13	import time
	14	import sys
824b5807	15	import os
3dd91472	16	import argparse
3abc6a0c	17	from io import TextIOWrapper
f1dca11c DW	18
	19	retcode = 0
	20	terminate = False
	21
824b5807 DW	22	def DEVNULL():
	23	'''Return /dev/null in subprocess writable format.'''
	24	try:
	25	from subprocess import DEVNULL
	26	return DEVNULL
	27	except ImportError:
	28	return open(os.devnull, 'wb')
	29
f1dca11c DW	30	def find_mounts():
f1dca11c DW	31	'''Map mountpoints to physical disks.'''
ab11d016 DW	32	def find_xfs_mounts(bdev, fs, lastdisk):
	33	'''Attach lastdisk to each fs found under bdev.'''
	34	if bdev['fstype'] == 'xfs' and bdev['mountpoint'] is not None:
	35	mnt = bdev['mountpoint']
	36	if mnt in fs:
	37	fs[mnt].add(lastdisk)
	38	else:
	39	fs[mnt] = set([lastdisk])
	40	if 'children' not in bdev:
	41	return
	42	for child in bdev['children']:
	43	find_xfs_mounts(child, fs, lastdisk)
f1dca11c DW	44
f1dca11c DW	45	fs = {}
ab11d016	46	cmd=['lsblk', '-o', 'NAME,KNAME,TYPE,FSTYPE,MOUNTPOINT', '-J']
f1dca11c DW	47	result = subprocess.Popen(cmd, stdout=subprocess.PIPE)
	48	result.wait()
	49	if result.returncode != 0:
	50	return fs
74aed9c8	51	sarray = [x.decode(sys.stdout.encoding) for x in result.stdout.readlines()]
f1dca11c DW	52	output = ' '.join(sarray)
f1dca11c DW	53	bdevdata = json.loads(output)
ab11d016	54
f1dca11c DW	55	# The lsblk output had better be in disks-then-partitions order
f1dca11c DW	56	for bdev in bdevdata['blockdevices']:
ab11d016 DW	57	lastdisk = bdev['kname']
	58	find_xfs_mounts(bdev, fs, lastdisk)
	59
f1dca11c DW	60	return fs
f1dca11c DW	61
3abc6a0c DW	62	def backtick(cmd):
	63	'''Generator function that yields lines of a program's stdout.'''
	64	p = subprocess.Popen(cmd, stdout = subprocess.PIPE)
	65	for line in TextIOWrapper(p.stdout, encoding="utf-8"):
	66	yield line.strip()
	67
	68	def remove_killfunc(killfuncs, fn):
	69	'''Ensure fn is not in killfuncs.'''
	70	try:
	71	killfuncs.remove(fn)
	72	except:
	73	pass
824b5807	74
0c22427f DW	75	def run_killable(cmd, stdout, killfuncs):
	76	'''Run a killable program. Returns program retcode or -1 if we can't
	77	start it.'''
f1dca11c DW	78	try:
f1dca11c DW	79	proc = subprocess.Popen(cmd, stdout = stdout)
0c22427f	80	killfuncs.add(proc.terminate)
f1dca11c	81	proc.wait()
0c22427f	82	remove_killfunc(killfuncs, proc.terminate)
f1dca11c DW	83	return proc.returncode
	84	except:
	85	return -1
	86
07c6fd59	87	# systemd doesn't like unit instance names with slashes in them, so it
595874f2 DW	88	# replaces them with dashes when it invokes the service. Filesystem paths
	89	# need a special --path argument so that dashes do not get mangled.
	90	def path_to_serviceunit(path):
	91	'''Convert a pathname into a systemd service unit name.'''
	92
	93	cmd = ['systemd-escape', '--template', '@scrub_svcname@',
	94	'--path', path]
07c6fd59 DW	95	try:
	96	proc = subprocess.Popen(cmd, stdout = subprocess.PIPE)
	97	proc.wait()
	98	for line in proc.stdout:
595874f2	99	return line.decode(sys.stdout.encoding).strip()
07c6fd59	100	except:
7c4b91c5	101	return None
07c6fd59	102
3abc6a0c DW	103	def systemctl_stop(unitname):
	104	'''Stop a systemd unit.'''
	105	cmd = ['systemctl', 'stop', unitname]
	106	x = subprocess.Popen(cmd)
	107	x.wait()
	108
	109	def systemctl_start(unitname, killfuncs):
	110	'''Start a systemd unit and wait for it to complete.'''
	111	stop_fn = None
	112	cmd = ['systemctl', 'start', unitname]
	113	try:
	114	proc = subprocess.Popen(cmd, stdout = DEVNULL())
	115	stop_fn = lambda: systemctl_stop(unitname)
	116	killfuncs.add(stop_fn)
	117	proc.wait()
	118	ret = proc.returncode
	119	except:
	120	if stop_fn is not None:
	121	remove_killfunc(killfuncs, stop_fn)
	122	return -1
	123
	124	if ret != 1:
	125	remove_killfunc(killfuncs, stop_fn)
	126	return ret
	127
	128	# If systemctl-start returns 1, it's possible that the service failed
	129	# or that dbus/systemd restarted and the client program lost its
	130	# connection -- according to the systemctl man page, 1 means "unit not
	131	# failed".
	132	#
	133	# Either way, we switch to polling the service status to try to wait
	134	# for the service to end. As of systemd 249, the is-active command
	135	# returns any of the following states: active, reloading, inactive,
	136	# failed, activating, deactivating, or maintenance. Apparently these
	137	# strings are not localized.
	138	while True:
	139	try:
	140	for l in backtick(['systemctl', 'is-active', unitname]):
	141	if l == 'failed':
	142	remove_killfunc(killfuncs, stop_fn)
	143	return 1
	144	if l == 'inactive':
	145	remove_killfunc(killfuncs, stop_fn)
	146	return 0
	147	except:
	148	remove_killfunc(killfuncs, stop_fn)
	149	return -1
	150
	151	time.sleep(1)
	152
f1dca11c DW	153	def run_scrub(mnt, cond, running_devs, mntdevs, killfuncs):
	154	'''Run a scrub process.'''
	155	global retcode, terminate
	156
	157	print("Scrubbing %s..." % mnt)
	158	sys.stdout.flush()
	159
	160	try:
	161	if terminate:
	162	return
	163
824b5807	164	# Try it the systemd way
595874f2 DW	165	unitname = path_to_serviceunit(path)
595874f2 DW	166	if unitname is not None:
3abc6a0c	167	ret = systemctl_start(unitname, killfuncs)
7c4b91c5 DW	168	if ret == 0 or ret == 1:
	169	print("Scrubbing %s done, (err=%d)" % (mnt, ret))
	170	sys.stdout.flush()
	171	retcode \|= ret
	172	return
	173
	174	if terminate:
	175	return
824b5807	176
f1dca11c	177	# Invoke xfs_scrub manually
27df677a DW	178	cmd = ['@sbindir@/xfs_scrub']
	179	cmd += '@scrub_args@'.split()
	180	cmd += [mnt]
0c22427f	181	ret = run_killable(cmd, None, killfuncs)
f1dca11c DW	182	if ret >= 0:
	183	print("Scrubbing %s done, (err=%d)" % (mnt, ret))
	184	sys.stdout.flush()
	185	retcode \|= ret
	186	return
	187
	188	if terminate:
	189	return
	190
	191	print("Unable to start scrub tool.")
	192	sys.stdout.flush()
	193	finally:
	194	running_devs -= mntdevs
	195	cond.acquire()
	196	cond.notify()
	197	cond.release()
	198
	199	def main():
	200	'''Find mounts, schedule scrub runs.'''
	201	def thr(mnt, devs):
	202	a = (mnt, cond, running_devs, devs, killfuncs)
	203	thr = threading.Thread(target = run_scrub, args = a)
	204	thr.start()
	205	global retcode, terminate
	206
3dd91472 DW	207	parser = argparse.ArgumentParser( \
	208	description = "Scrub all mounted XFS filesystems.")
	209	parser.add_argument("-V", help = "Report version and exit.", \
	210	action = "store_true")
	211	args = parser.parse_args()
	212
	213	if args.V:
	214	print("xfs_scrub_all version @pkg_version@")
	215	sys.exit(0)
	216
f1dca11c DW	217	fs = find_mounts()
f1dca11c DW	218
824b5807 DW	219	# Tail the journal if we ourselves aren't a service...
	220	journalthread = None
	221	if 'SERVICE_MODE' not in os.environ:
	222	try:
	223	cmd=['journalctl', '--no-pager', '-q', '-S', 'now', \
	224	'-f', '-u', 'xfs_scrub@*', '-o', \
	225	'cat']
	226	journalthread = subprocess.Popen(cmd)
	227	except:
	228	pass
	229
f1dca11c DW	230	# Schedule scrub jobs...
	231	running_devs = set()
	232	killfuncs = set()
	233	cond = threading.Condition()
	234	while len(fs) > 0:
	235	if len(running_devs) == 0:
	236	mnt, devs = fs.popitem()
	237	running_devs.update(devs)
	238	thr(mnt, devs)
	239	poppers = set()
	240	for mnt in fs:
	241	devs = fs[mnt]
	242	can_run = True
	243	for dev in devs:
	244	if dev in running_devs:
	245	can_run = False
	246	break
	247	if can_run:
	248	running_devs.update(devs)
	249	poppers.add(mnt)
	250	thr(mnt, devs)
	251	for p in poppers:
	252	fs.pop(p)
	253	cond.acquire()
	254	try:
	255	cond.wait()
	256	except KeyboardInterrupt:
	257	terminate = True
	258	print("Terminating...")
	259	sys.stdout.flush()
	260	while len(killfuncs) > 0:
	261	fn = killfuncs.pop()
	262	fn()
	263	fs = []
	264	cond.release()
	265
824b5807 DW	266	if journalthread is not None:
	267	journalthread.terminate()
	268
	269	# See the service mode comments in xfs_scrub.c for why we do this.
	270	if 'SERVICE_MODE' in os.environ:
	271	time.sleep(2)
	272	if retcode != 0:
	273	retcode = 1
	274
f1dca11c DW	275	sys.exit(retcode)
	276
	277	if __name__ == '__main__':
	278	main()