analyze-initcall-debug.py

Hello,

this is the second version of the analyze-initcall-debug.py script, which can be used to analyze the kernel output when booting with initcall_debug to extract some human-readable data from it.

The first version ca be found at:
https://lore.kernel.org/linux-embedded/[email protected]/

Regards,

Francesco

---

Differences v1 -> v2:

- added a list of failed driver probes in the plain text output
- added a preliminary comparison on read lines to avoid false negatives in regex-based matches
- added the --body-only option to output only the raw HTML body
- use of dictionaries instead of filter() constructs

------ the analyze-initcall-debug.py script follows ------

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
This script can be used to analyze a Linux kernel dmesg output when the initcall_debug command line output is used. It extract various information, such as init durations, impact of probe deferral and so on.
"""

import argparse
import re
import sys

ERRCODE_PROBE_DEFER = 517

parser = argparse.ArgumentParser(description='Analyze a Linux kernel dmesg with the initcall_debug option enabled')

parser.add_argument('--dmesg', nargs='?', type=argparse.FileType('r'),
default=sys.stdin, help='The dmesg file to analyze (default: stdin)')
format_group = parser.add_mutually_exclusive_group() format_group.add_argument('--html', action='store_true',
help='Output analysis result as HTML table')
format_group.add_argument('--markdown', action='store_true',
help='Output analysis result as Markdown table')
parser.add_argument('--body-only', action='store_true',
help='Do not add header and footer to HTML output')
args = parser.parse_args()

class Run:
def __init__(self, start_time:int, end_time:int = -1, duration:int = 0, retval:int = 0):
self._start_time = start_time
self._end_time = end_time
self._duration = duration
self._retval = retval
self._ended = (end_time >= 0)

@property
def start_time(self) -> int:
return self._start_time

@property
def end_time(self) -> int:
return self._end_time

@end_time.setter
def end_time(self, time:int):
self._end_time = time
self._ended = True

@property
def duration(self) -> int:
return self._duration

@duration.setter
def duration(self, time:int):
self._duration = time

@property
def retval(self) -> int:
return self._retval

@retval.setter
def retval(self, val:int):
self._retval = val

@property
def running(self) -> bool:
return (not self._ended or abs(self.retval) == ERRCODE_PROBE_DEFER)

@property
def failed(self) -> bool:
return ((self.retval != 0) and not self.running)

class Entity:

    def __init__(self, name:str, start_time:int = 0, end_time:int = 0, duration:int = 0, retval:int = 0):
        self._name = name
        self._runs = [ Run(start_time, end_time, duration, retval) ]

    @property
    def name(self) -> str:
        return self._name

    @property
    def first_start_time(self) -> int:
        return self._runs[0].start_time

    @property
    def last_start_time(self) -> int:
        return self._runs[-1].start_time

    @property
    def first_end_time(self) -> int:
        return self._runs[0].end_time

    @property
    def last_end_time(self) -> int:
        return self._runs[-1].end_time

    @property
    def duration(self) -> int:
        return sum([ r.duration for r in self._runs ])

    @property
    def wasted_time(self) -> int:
        return sum([ r.duration for r in filter(lambda x: x.failed or abs(x.retval) == ERRCODE_PROBE_DEFER, self._runs) ])

    @property
    def retval(self) -> int:
        return self._runs[-1].retval

    @property
    def failed(self) -> bool:
        return self._runs[-1].failed

    @property
    def running(self) -> bool:
        return self._runs[-1].running

    @property
    def deferred_probe_pending(self) -> bool:
        return (abs(self._runs[-1].retval) == ERRCODE_PROBE_DEFER)

    @property
    def num_deferred_probes(self) -> int:
        return len(list(filter(lambda r: abs(r.retval) == ERRCODE_PROBE_DEFER, self._runs)))

    def addStart(self, start_time: int):
        self._runs.append( Run(start_time) )

    def addEnd(self, end_time:int = 0, duration:int = 0, retval:int = 0):
        self._runs[-1].end_time = end_time
        self._runs[-1].duration = duration
        self._runs[-1].retval = retval

    def addRun(self, start_time: int, end_time:int = 0, duration:int = 0, retval:int = 0):
        self._runs.append( Run(start_time, end_time, duration, retval) )

class Driver (Entity):
def __init__(self, name: str, start_time: int):
super().__init__(name, start_time)

def addInit(self, init_time: int):
self.addStart(init_time)

def addReturn(self, return_time: int, duration: int, retval: int):
self.addEnd(return_time, duration, retval)

class Probe (Entity):
def __init__(self, name: str, return_time: int, duration: int, retval: int):
super().__init__(name, return_time, return_time, duration, retval)

class Init (Entity):

    def __init__(self, name: str, start_time: int):
        super().__init__(name, start_time)

# Regex for 'calling' messages
# "[ 0.466115] calling pci_sysfs_init+0x0/0xa8 @ 1"
calling_sentinel = 'calling '
calling_prog = re.compile(r'\[([0-9\s]+\.[0-9]+)\] calling ([0-9a-zA-Z_]+)\+(0x[0-9a-fA-F]+\/0x[0-9a-fA-F]+)( \[[a-zA-Z0-9\-_]+\])? @ ([0-9]+)')

# Regex for 'initcall ... returned' messages
# "[ 0.466115] initcall pci_sysfs_init+0x0/0xa8 returned 0 after 5 usecs"
returned_sentinel = 'initcall '
returned_prog = re.compile(r'\[([0-9\s]+\.[0-9]+)\] initcall ([0-9a-zA-Z_]+)\+(0x[0-9a-fA-F]+\/0x[0-9a-fA-F]+)( \[[a-zA-Z0-9\-_]+\])? returned ([\-0-9]+) after ([0-9]+) usecs')

# Regex for 'probe ... returned' messages
# "[ 0.466115] probe of cpufreq-dt returned 517 after 140 usec"
probe_sentinel = 'probe of '
probe_prog = re.compile(r'\[([0-9\s]+\.[0-9]+)\] probe of ([0-9a-zA-Z_\-\.\:@]+) returned ([\-0-9]+) after ([0-9]+) usecs')

# Regex for
# "[ 1.060329] Run /sbin/init as init process"
init_sentinel = 'as init process'
init_prog = re.compile(r'\[([0-9\s]+\.[0-9]+)\] Run ([/0-9a-zA-Z_]+) as init process')

drivers = {}
probes = {}
init = None

# Extract data from dmesg
lineno = 0
for line in args.dmesg:
lineno += 1
if calling_sentinel in line:
match = calling_prog.match(line)
if match:
try:
time = int(float(match.group(1)) * 1000000.0)
name = str(match.group(2))
except Exception as e:
print(f'Failed parsing line {lineno}:"{line.rstrip()}" as call', file=sys.stderr) raise e
else:
if name not in drivers.keys():
drivers[name] = Driver(name, time)
else:
drivers[name].addInit(time)
finally:
continue
else:
print(f'Failed matching line {lineno}:"{line.rstrip()}" as call', file=sys.stderr)

if returned_sentinel in line:
match = returned_prog.match(line)
if match:
try:
time = int(float(match.group(1)) * 1000000.0)
name = str(match.group(2))
retval = int(match.group(5))
duration = int(match.group(6))
except:
print(f'Failed parsing line {lineno}:"{line.rstrip()}" as call return', file=sys.stderr)
else:
if name not in drivers.keys():
print(f'Detected return for driver {name}, for which an init was never recorded', file=sys.stderr)
else:
drivers[name].addReturn(time, duration, retval)
finally:
continue
else:
print(f'Failed matching line {lineno}:"{line.rstrip()}" as call return', file=sys.stderr)

if probe_sentinel in line:
match = probe_prog.match(line)
if match:
try:
time = int(float(match.group(1)) * 1000000.0)
name = str(match.group(2))
retval = int(match.group(3))
duration = int(match.group(4))
except:
print(f'Failed parsing line {lineno}:"{line.rstrip()}" as probe return', file=sys.stderr)
else:
if name not in probes.keys():
probes[name] = Probe(name, time, duration, retval)
else:
probes[name].addRun(time, time, duration, retval)
finally:
continue
else:
print(f'Failed matching line {lineno}:"{line.rstrip()}" as probe return', file=sys.stderr)

if not init and init_sentinel in line:
match = init_prog.match(line)
if match:
try:
time = int(float(match.group(1)) * 1000000.0)
name = str(match.group(2))
except:
print(f'Failed parsing line {lineno}:"{line.rstrip()}" as init', file=sys.stderr)
else:
init = Init(name, time)
finally:
continue
else:
print(f'Failed matching line {lineno}:"{line.rstrip()}" as init', file=sys.stderr)

# Print Markdown format
if args.markdown:

    print('| Name | Type [^1] | Status [^2] | Duration (us) | Before userspace | Num deferred probes | Time spent in failed inits (us) |')
    print('| -----| --------- | ----------- | ------------- | ---------------- | ------------------- | ------------------------------- |')

for d in sorted(filter(lambda k: k.duration > 0, list(drivers.values()) + list(probes.values())), key=lambda k: k.duration, reverse=True):
etype = 'I' if d in drivers else 'P'
run_status = 'D' if d.deferred_probe_pending else 'R' if d.running else f'F(-{abs(d.retval)})' if d.failed else 'K' before_init = 'Y' if d.last_start_time < init.last_start_time else 'N'
print(f'| {d.name} | {etype} | {run_status} | {d.duration} | {before_init} | {d.num_deferred_probes} | {d.wasted_time} |')
print('[^1]: I=Init function, P=device Probe')
print('[^2]: D=Deferred probe pending, R=still Running, F(N)=Failed (with errcode N), K=init oK')

# Print HTML format
elif args.html:
if not args.body_only:
print(
'''
<!DOCTYPE HTML>
<html>
<head>
<meta charset="utf-8" />
<title>Initcall analysis</title>
<meta name="viewport" content="width=device-width, initial-scale=1" /> <style>

    table, th, td {
      border: 1px solid;
    }

</style>
</head>
<body>''')

    print(
'''
  <table>
    <tr>
      <th>Name</th>
      <th>Type <a href="#note1">[1]</a></th>
      <th>Status <a href="#note2">[2]</a></th>
      <th>Duration (us)</th>
      <th>Before userspace</th>
      <th>Num deferred probes</th>
      <th>Time spent in failed inits (us)</th>
    </tr>''')

    for d in sorted(filter(lambda k: k.duration > 0, list(drivers.values()) + list(probes.values())), key=lambda k: k.duration, reverse=True):
        etype = 'I' if d in drivers else 'P'
        run_status = 'D' if d.deferred_probe_pending else 'R' if d.running else f'F({abs(d.retval)})' if d.failed else 'K'
        before_init = 'Y' if d.last_start_time < init.last_start_time else 'N'
        print(
f'''
    <tr>
      <td>{d.name}</td>
      <td>{etype}</td>
      <td>{run_status}</td>
      <td>{d.duration}</td>
      <td>{before_init}</td>
      <td>{d.num_deferred_probes}</td>
      <td>{d.wasted_time}</td>
    </tr>''')

    print(

'''
</table>
<p id="note1">[1] I=Init function, P=device Probe</p>
<p id="note2">[2] D=Deferred probe pending, R=still Running, F(N)=Failed (with errcode N), K=init oK</p>
''')

if not args.body_only:
print(
'''
</body>
</html>
''')

# Print plain text
else:
num_before_userspace = len(list(filter(lambda d: d.last_start_time <= init.last_start_time, drivers.values()))) num_after_userspace = len(list(filter(lambda d: d.last_start_time > init.last_start_time, drivers.values()))) num_deferred_probe_pending = len(list(filter(lambda d: d.deferred_probe_pending, list(drivers.values()) + list(probes.values())))) num_failed = len(list(filter(lambda d: d.failed, list(drivers.values()) + list(probes.values()))))

print('Summary:')
print(f' {len(drivers)} drivers have been initialized, of which {num_before_userspace} before userspace and {num_after_userspace} after') print(f' {len(probes)} probes happened outside of the initcall of their driver')
print(f' {num_deferred_probe_pending} deferred probes are pending')
print(f' {num_failed} initcalls/probes failed')

print('\n---\n')

print('Top 10 initcall/probe durations:')
for d in sorted(filter(lambda k: k.duration > 0, list(drivers.values()) + list(probes.values())), key=lambda k: k.duration, reverse=True)[0:10]:
print(f' * {d.name} -> {d.duration}us')

print('\n---\n')

print('Failed initcalls/probes:')
for d in filter(lambda k: k.failed, list(drivers.values()) + list(probes.values())):
print(f' * {d.name} -> ret = -{abs(d.retval)}')