3 # SPDX-License-Identifier: GPL-2.0-or-later
4 # Copyright (C) 2023-2024 Oracle. All rights reserved.
6 # Author: Darrick J. Wong <djwong@kernel.org>
8 # Run bcachefsck in parallel, but avoid thrashing.
19 from io import TextIOWrapper
20 from pathlib import Path
21 from datetime import timedelta
22 from datetime import datetime
23 from datetime import timezone
30 '''Return /dev/null in subprocess writable format.'''
32 from subprocess import DEVNULL
35 return open(os.devnull, 'wb')
38 '''Map mountpoints to physical disks.'''
39 def find_bcachefs_mounts(bdev, fs, lastdisk):
40 '''Attach all lastdisk to each fs found under bdev.'''
41 if bdev['fstype'] == 'bcachefs' and bdev['mountpoint'] is not None:
42 mnt = bdev['mountpoint']
44 fs[mnt].add(lastdisk.split(':'))
46 fs[mnt] = set(lastdisk.split(':'))
47 if 'children' not in bdev:
49 for child in bdev['children']:
50 find_bcachefs_mounts(child, fs, lastdisk)
53 cmd=['lsblk', '-o', 'NAME,KNAME,TYPE,FSTYPE,MOUNTPOINT', '-J']
54 result = subprocess.Popen(cmd, stdout=subprocess.PIPE)
56 if result.returncode != 0:
58 sarray = [x.decode(sys.stdout.encoding) for x in result.stdout.readlines()]
59 output = ' '.join(sarray)
60 bdevdata = json.loads(output)
62 # The lsblk output had better be in disks-then-partitions order
63 for bdev in bdevdata['blockdevices']:
64 lastdisk = bdev['kname']
65 find_bcachefs_mounts(bdev, fs, lastdisk)
70 '''Generator function that yields lines of a program's stdout.'''
71 p = subprocess.Popen(cmd, stdout = subprocess.PIPE)
72 for line in TextIOWrapper(p.stdout, encoding="utf-8"):
75 def remove_killfunc(killfuncs, fn):
76 '''Ensure fn is not in killfuncs.'''
82 class scrub_control(object):
83 '''Control object for bcachefsck.'''
88 '''Start scrub and wait for it to complete. Returns -1 if the
89 service was not started, 0 if it succeeded, or 1 if it
97 class scrub_subprocess(scrub_control):
98 '''Control object for bcachefsck subprocesses.'''
99 def __init__(self, mnt):
100 cmd = ['bcachefs', 'fsck']
101 cmd += '@bcachefsck_args@'.split()
107 '''Start bcachefsck and wait for it to complete. Returns -1 if
108 the service was not started, 0 if it succeeded, or 1 if it
113 print('run ', ' '.join(self.cmdline))
116 self.proc = subprocess.Popen(self.cmdline)
123 return proc.returncode
126 '''Stop bcachefsck.'''
130 print('kill ', ' '.join(self.cmdline))
131 if self.proc is not None:
132 self.proc.terminate()
134 def run_subprocess(mnt, killfuncs):
135 '''Run a killable program. Returns program retcode or -1 if we can't
138 p = scrub_subprocess(mnt)
139 killfuncs.add(p.stop)
141 remove_killfunc(killfuncs, p.stop)
146 # systemd doesn't like unit instance names with slashes in them, so it
147 # replaces them with dashes when it invokes the service. Filesystem paths
148 # need a special --path argument so that dashes do not get mangled.
149 def path_to_serviceunit(path):
150 '''Convert a pathname into a systemd service unit name.'''
152 svcname = 'bcachefsck@.service'
153 cmd = ['systemd-escape', '--template', svcname, '--path', path]
155 proc = subprocess.Popen(cmd, stdout = subprocess.PIPE)
157 for line in proc.stdout:
158 return line.decode(sys.stdout.encoding).strip()
160 def fibonacci(max_ret):
161 '''Yield fibonacci sequence up to but not including max_ret.'''
176 class scrub_service(scrub_control):
177 '''Control object for bcachefsck systemd service.'''
178 def __init__(self, mnt):
179 self.unitname = path_to_serviceunit(mnt)
185 '''Bind to the dbus proxy object for this service.'''
186 sysbus = dbus.SystemBus()
187 systemd1 = sysbus.get_object('org.freedesktop.systemd1',
188 '/org/freedesktop/systemd1')
189 manager = dbus.Interface(systemd1,
190 'org.freedesktop.systemd1.Manager')
191 path = manager.LoadUnit(self.unitname)
193 svc_obj = sysbus.get_object('org.freedesktop.systemd1', path)
194 self.prop = dbus.Interface(svc_obj,
195 'org.freedesktop.DBus.Properties')
196 self.unit = dbus.Interface(svc_obj,
197 'org.freedesktop.systemd1.Unit')
199 def __dbusrun(self, lambda_fn):
200 '''Call the lambda function to execute something on dbus. dbus
201 exceptions result in retries with Fibonacci backoff, and the
202 bindings will be rebuilt every time.'''
207 for i in fibonacci(30):
210 except dbus.exceptions.DBusException as e:
219 '''Retrieve the active state for a systemd service. As of
220 systemd 249, this is supposed to be one of the following:
221 "active", "reloading", "inactive", "failed", "activating",
222 or "deactivating". These strings are not localized.'''
225 l = lambda: self.prop.Get('org.freedesktop.systemd1.Unit',
228 return self.__dbusrun(l)
229 except Exception as e:
231 print(e, file = sys.stderr)
234 def wait(self, interval = 1):
235 '''Wait until the service finishes.'''
238 # Use a poll/sleep loop to wait for the service to finish.
239 # Avoid adding a dependency on python3 glib, which is required
240 # to use an event loop to receive a dbus signal.
242 while s not in ['failed', 'inactive']:
244 print('waiting %s %s' % (self.unitname, s))
248 print('waited %s %s' % (self.unitname, s))
254 '''Start the service and wait for it to complete. Returns -1
255 if the service was not started, 0 if it succeeded, or 1 if it
260 print('starting %s' % self.unitname)
263 self.__dbusrun(lambda: self.unit.Start('replace'))
265 except Exception as e:
266 print(e, file = sys.stderr)
270 '''Stop the service.'''
274 print('stopping %s' % self.unitname)
277 self.__dbusrun(lambda: self.unit.Stop('replace'))
279 except Exception as e:
280 print(e, file = sys.stderr)
283 def run_service(mnt, killfuncs):
284 '''Run scrub as a service.'''
286 svc = scrub_service(mnt)
290 killfuncs.add(svc.stop)
291 retcode = svc.start()
292 remove_killfunc(killfuncs, svc.stop)
295 def run_scrub(mnt, cond, running_devs, mntdevs, killfuncs):
296 '''Run a scrub process.'''
297 global retcode, terminate
299 print("Scrubbing %s..." % mnt)
306 # Run per-mount systemd bcachefsck service only if we ourselves
307 # are running as a systemd service.
308 if 'SERVICE_MODE' in os.environ:
309 ret = run_service(mnt, killfuncs)
310 if ret == 0 or ret == 1:
311 print("Scrubbing %s done, (err=%d)" % (mnt, ret))
319 # Invoke bcachefsck manually if we're running in the foreground.
320 # We also permit this if we're running as a cronjob where
321 # systemd services are unavailable.
322 ret = run_subprocess(mnt, killfuncs)
324 print("Scrubbing %s done, (err=%d)" % (mnt, ret))
332 print("Unable to start scrub tool.")
335 running_devs -= mntdevs
340 def signal_scrubs(signum, cond):
341 '''Handle termination signals by killing bcachefsck children.'''
342 global debug, terminate
345 print('Signal handler called with signal', signum)
353 def wait_for_termination(cond, killfuncs):
354 '''Wait for a child thread to terminate. Returns True if we should
355 abort the program, False otherwise.'''
356 global debug, terminate
359 print('waiting for threads to terminate')
365 except KeyboardInterrupt:
372 print("Terminating...")
374 while len(killfuncs) > 0:
379 def scan_interval(string):
380 '''Convert a textual scan interval argument into a time delta.'''
382 if string.endswith('y'):
383 year = timedelta(seconds = 31556952)
384 return year * float(string[:-1])
385 if string.endswith('q'):
386 return timedelta(days = 90 * float(string[:-1]))
387 if string.endswith('mo'):
388 return timedelta(days = 30 * float(string[:-2]))
389 if string.endswith('w'):
390 return timedelta(weeks = float(string[:-1]))
391 if string.endswith('d'):
392 return timedelta(days = float(string[:-1]))
393 if string.endswith('h'):
394 return timedelta(hours = float(string[:-1]))
395 if string.endswith('m'):
396 return timedelta(minutes = float(string[:-1]))
397 if string.endswith('s'):
398 return timedelta(seconds = float(string[:-1]))
399 return timedelta(seconds = int(string))
402 '''Create a representation of the time right now, in UTC.'''
404 dt = datetime.utcnow()
405 return dt.replace(tzinfo = timezone.utc)
408 '''Find mounts, schedule bcachefsck runs.'''
410 a = (mnt, cond, running_devs, devs, killfuncs)
411 thr = threading.Thread(target = run_scrub, args = a)
413 global retcode, terminate, debug
415 parser = argparse.ArgumentParser( \
416 description = "Scrub all mounted bcachefs filesystems.")
417 parser.add_argument("--debug", help = "Enabling debugging messages.", \
418 action = "store_true")
419 args = parser.parse_args()
426 # Schedule scrub jobs...
429 cond = threading.Condition()
431 signal.signal(signal.SIGINT, lambda s, f: signal_scrubs(s, cond))
432 signal.signal(signal.SIGTERM, lambda s, f: signal_scrubs(s, cond))
435 if len(running_devs) == 0:
436 mnt, devs = fs.popitem()
437 running_devs.update(devs)
444 if dev in running_devs:
448 running_devs.update(devs)
454 # Wait for one thread to finish
455 if wait_for_termination(cond, killfuncs):
458 # Wait for the rest of the threads to finish
459 while len(killfuncs) > 0:
460 wait_for_termination(cond, killfuncs)
462 # If we're being run as a service, the return code must fit the LSB
463 # init script action error guidelines, which is to say that we compress
464 # all errors to 1 ("generic or unspecified error", LSB 5.0 section
465 # 22.2) and hope the admin will scan the log for what actually
468 # We have to sleep 2 seconds here because journald uses the pid to
469 # connect our log messages to the systemd service. This is critical
470 # for capturing all the log messages if the scrub fails, because the
471 # fail service uses the service name to gather log messages for the
473 if 'SERVICE_MODE' in os.environ:
480 if __name__ == '__main__':