#!/usr/bin/env python3# SPDX-License-Identifier: GPL-2.0## Run a perf script command multiple times in parallel, using perf script# options --cpu and --time so that each job processes a different chunk# of the data.## Copyright (c) 2024, Intel Corporation. import subprocessimport argparseimport pathlibimport shleximport timeimport copyimport sysimport osimport re glb_prog_name = "parallel-perf.py"glb_min_interval = 10.0glb_min_samples = 64 class Verbosity(): 	def __init__(self, quiet=False, verbose=False, debug=False):		self.normal    = True		self.verbose   = verbose		self.debug     = debug		self.self_test = True		if self.debug:			self.verbose = True		if self.verbose:			quiet = False		if quiet:			self.normal = False # Manage work (Start/Wait/Kill), as represented by a subprocess.Popen commandclass Work(): 	def __init__(self, cmd, pipe_to, output_dir="."):		self.popen = None		self.consumer = None		self.cmd = cmd		self.pipe_to = pipe_to		self.output_dir = output_dir		self.cmdout_name = f"{output_dir}/cmd.txt"		self.stdout_name = f"{output_dir}/out.txt"		self.stderr_name = f"{output_dir}/err.txt" 	def Command(self):		sh_cmd = [ shlex.quote(x) for x in self.cmd ]		return " ".join(self.cmd) 	def Stdout(self):		return open(self.stdout_name, "w") 	def Stderr(self):		return open(self.stderr_name, "w") 	def CreateOutputDir(self):		pathlib.Path(self.output_dir).mkdir(parents=True, exist_ok=True) 	def Start(self):		if self.popen:			return		self.CreateOutputDir()		with open(self.cmdout_name, "w") as f:			f.write(self.Command())			f.write("\n")		stdout = self.Stdout()		stderr = self.Stderr()		if self.pipe_to:			self.popen = subprocess.Popen(self.cmd, stdout=subprocess.PIPE, stderr=stderr)			args = shlex.split(self.pipe_to)			self.consumer = subprocess.Popen(args, stdin=self.popen.stdout, stdout=stdout, stderr=stderr)		else:			self.popen = subprocess.Popen(self.cmd, stdout=stdout, stderr=stderr) 	def RemoveEmptyErrFile(self):		if os.path.exists(self.stderr_name):			if os.path.getsize(self.stderr_name) == 0:				os.unlink(self.stderr_name) 	def Errors(self):		if os.path.exists(self.stderr_name):			if os.path.getsize(self.stderr_name) != 0:				return [ f"Non-empty error file {self.stderr_name}" ]		return [] 	def TidyUp(self):		self.RemoveEmptyErrFile() 	def RawPollWait(self, p, wait):		if wait:			return p.wait()		return p.poll() 	def Poll(self, wait=False):		if not self.popen:			return None		result = self.RawPollWait(self.popen, wait)		if self.consumer:			res = result			result = self.RawPollWait(self.consumer, wait)			if result != None and res == None:				self.popen.kill()				result = None			elif result == 0 and res != None and res != 0:				result = res		if result != None:			self.TidyUp()		return result 	def Wait(self):		return self.Poll(wait=True) 	def Kill(self):		if not self.popen:			return		self.popen.kill()		if self.consumer:			self.consumer.kill() def KillWork(worklist, verbosity):	for w in worklist:		w.Kill()	for w in worklist:		w.Wait() def NumberOfCPUs():	return os.sysconf("SC_NPROCESSORS_ONLN") def NanoSecsToSecsStr(x):	if x == None:		return ""	x = str(x)	if len(x) < 10:		x = "0" * (10 - len(x)) + x	return x[:len(x) - 9] + "." + x[-9:] def InsertOptionAfter(cmd, option, after):	try:		pos = cmd.index(after)		cmd.insert(pos + 1, option)	except:		cmd.append(option) def CreateWorkList(cmd, pipe_to, output_dir, cpus, time_ranges_by_cpu):	max_len = len(str(cpus[-1]))	cpu_dir_fmt = f"cpu-%.{max_len}u"	worklist = []	pos = 0	for cpu in cpus:		if cpu >= 0:			cpu_dir = os.path.join(output_dir, cpu_dir_fmt % cpu)			cpu_option = f"--cpu={cpu}"		else:			cpu_dir = output_dir			cpu_option = None 		tr_dir_fmt = "time-range" 		if len(time_ranges_by_cpu) > 1:			time_ranges = time_ranges_by_cpu[pos]			tr_dir_fmt += f"-{pos}"			pos += 1		else:			time_ranges = time_ranges_by_cpu[0] 		max_len = len(str(len(time_ranges)))		tr_dir_fmt += f"-%.{max_len}u" 		i = 0		for r in time_ranges:			if r == [None, None]:				time_option = None				work_output_dir = cpu_dir			else:				time_option = "--time=" + NanoSecsToSecsStr(r[0]) + "," + NanoSecsToSecsStr(r[1])				work_output_dir = os.path.join(cpu_dir, tr_dir_fmt % i)				i += 1			work_cmd = list(cmd)			if time_option != None:				InsertOptionAfter(work_cmd, time_option, "script")			if cpu_option != None:				InsertOptionAfter(work_cmd, cpu_option, "script")			w = Work(work_cmd, pipe_to, work_output_dir)			worklist.append(w)	return worklist def DoRunWork(worklist, nr_jobs, verbosity):	nr_to_do = len(worklist)	not_started = list(worklist)	running = []	done = []	chg = False	while True:		nr_done = len(done)		if chg and verbosity.normal:			nr_run = len(running)			print(f"\rThere are {nr_to_do} jobs: {nr_done} completed, {nr_run} running", flush=True, end=" ")			if verbosity.verbose:				print()			chg = False		if nr_done == nr_to_do:			break		while len(running) < nr_jobs and len(not_started):			w = not_started.pop(0)			running.append(w)			if verbosity.verbose:				print("Starting:", w.Command())			w.Start()			chg = True		if len(running):			time.sleep(0.1)		finished = []		not_finished = []		while len(running):			w = running.pop(0)			r = w.Poll()			if r == None:				not_finished.append(w)				continue			if r == 0:				if verbosity.verbose:					print("Finished:", w.Command())				finished.append(w)				chg = True				continue			if verbosity.normal and not verbosity.verbose:				print()			print("Job failed!\n    return code:", r, "\n    command:    ", w.Command())			if w.pipe_to:				print("    piped to:   ", w.pipe_to)			print("Killing outstanding jobs")			KillWork(not_finished, verbosity)			KillWork(running, verbosity)			return False		running = not_finished		done += finished	errorlist = []	for w in worklist:		errorlist += w.Errors()	if len(errorlist):		print("Errors:")		for e in errorlist:			print(e)	elif verbosity.normal:		print("\r"," "*50, "\rAll jobs finished successfully", flush=True)	return True def RunWork(worklist, nr_jobs=NumberOfCPUs(), verbosity=Verbosity()):	try:		return DoRunWork(worklist, nr_jobs, verbosity)	except:		for w in worklist:			w.Kill()		raise	return True def ReadHeader(perf, file_name):	return subprocess.Popen([perf, "script", "--header-only", "--input", file_name], stdout=subprocess.PIPE).stdout.read().decode("utf-8") def ParseHeader(hdr):	result = {}	lines = hdr.split("\n")	for line in lines:		if ":" in line and line[0] == "#":			pos = line.index(":")			name = line[1:pos-1].strip()			value = line[pos+1:].strip()			if name in result:				orig_name = name				nr = 2				while True:					name = f"{orig_name} {nr}"					if name not in result:						break					nr += 1			result[name] = value	return result def HeaderField(hdr_dict, hdr_fld):	if hdr_fld not in hdr_dict:		raise Exception(f"'{hdr_fld}' missing from header information")	return hdr_dict[hdr_fld] # Represent the position of an option within a command string# and provide the option value and/or remove the optionclass OptPos(): 	def Init(self, opt_element=-1, value_element=-1, opt_pos=-1, value_pos=-1, error=None):		self.opt_element = opt_element		# list element that contains option		self.value_element = value_element	# list element that contains option value		self.opt_pos = opt_pos			# string position of option		self.value_pos = value_pos		# string position of value		self.error = error			# error message string 	def __init__(self, args, short_name, long_name, default=None):		self.args = list(args)		self.default = default		n = 2 + len(long_name)		m = len(short_name)		pos = -1		for opt in args:			pos += 1			if m and opt[:2] == f"-{short_name}":				if len(opt) == 2:					if pos + 1 < len(args):						self.Init(pos, pos + 1, 0, 0)					else:						self.Init(error = f"-{short_name} option missing value")				else:					self.Init(pos, pos, 0, 2)				return			if opt[:n] == f"--{long_name}":				if len(opt) == n:					if pos + 1 < len(args):						self.Init(pos, pos + 1, 0, 0)					else:						self.Init(error = f"--{long_name} option missing value")				elif opt[n] == "=":					self.Init(pos, pos, 0, n + 1)				else:					self.Init(error = f"--{long_name} option expected '='")				return			if m and opt[:1] == "-" and opt[:2] != "--" and short_name in opt:				ipos = opt.index(short_name)				if "-" in opt[1:]:					hpos = opt[1:].index("-")					if hpos < ipos:						continue				if ipos + 1 == len(opt):					if pos + 1 < len(args):						self.Init(pos, pos + 1, ipos, 0)					else:						self.Init(error = f"-{short_name} option missing value")				else:					self.Init(pos, pos, ipos, ipos + 1)				return		self.Init() 	def Value(self):		if self.opt_element >= 0:			if self.opt_element != self.value_element:				return self.args[self.value_element]			else:				return self.args[self.value_element][self.value_pos:]		return self.default 	def Remove(self, args):		if self.opt_element == -1:			return		if self.opt_element != self.value_element:			del args[self.value_element]		if self.opt_pos:			args[self.opt_element] = args[self.opt_element][:self.opt_pos]		else:			del args[self.opt_element] def DetermineInputFileName(cmd):	p = OptPos(cmd, "i", "input", "perf.data")	if p.error:		raise Exception(f"perf command {p.error}")	file_name = p.Value()	if not os.path.exists(file_name):		raise Exception(f"perf command input file '{file_name}' not found")	return file_name def ReadOption(args, short_name, long_name, err_prefix, remove=False):	p = OptPos(args, short_name, long_name)	if p.error:		raise Exception(f"{err_prefix}{p.error}")	value = p.Value()	if remove:		p.Remove(args)	return value def ExtractOption(args, short_name, long_name, err_prefix):	return ReadOption(args, short_name, long_name, err_prefix, True) def ReadPerfOption(args, short_name, long_name):	return ReadOption(args, short_name, long_name, "perf command ") def ExtractPerfOption(args, short_name, long_name):	return ExtractOption(args, short_name, long_name, "perf command ") def PerfDoubleQuickCommands(cmd, file_name):	cpu_str = ReadPerfOption(cmd, "C", "cpu")	time_str = ReadPerfOption(cmd, "", "time")	# Use double-quick sampling to determine trace data density	times_cmd = ["perf", "script", "--ns", "--input", file_name, "--itrace=qqi"]	if cpu_str != None and cpu_str != "":		times_cmd.append(f"--cpu={cpu_str}")	if time_str != None and time_str != "":		times_cmd.append(f"--time={time_str}")	cnts_cmd = list(times_cmd)	cnts_cmd.append("-Fcpu")	times_cmd.append("-Fcpu,time")	return cnts_cmd, times_cmd class CPUTimeRange():	def __init__(self, cpu):		self.cpu = cpu		self.sample_cnt = 0		self.time_ranges = None		self.interval = 0		self.interval_remaining = 0		self.remaining = 0		self.tr_pos = 0 def CalcTimeRangesByCPU(line, cpu, cpu_time_ranges, max_time):	cpu_time_range = cpu_time_ranges[cpu]	cpu_time_range.remaining -= 1	cpu_time_range.interval_remaining -= 1	if cpu_time_range.remaining == 0:		cpu_time_range.time_ranges[cpu_time_range.tr_pos][1] = max_time		return	if cpu_time_range.interval_remaining == 0:		time = TimeVal(line[1][:-1], 0)		time_ranges = cpu_time_range.time_ranges		time_ranges[cpu_time_range.tr_pos][1] = time - 1		time_ranges.append([time, max_time])		cpu_time_range.tr_pos += 1		cpu_time_range.interval_remaining = cpu_time_range.interval def CountSamplesByCPU(line, cpu, cpu_time_ranges):	try:		cpu_time_ranges[cpu].sample_cnt += 1	except:		print("exception")		print("cpu", cpu)		print("len(cpu_time_ranges)", len(cpu_time_ranges))		raise def ProcessCommandOutputLines(cmd, per_cpu, fn, *x):	# Assume CPU number is at beginning of line and enclosed by []	pat = re.compile(r"\s*\[[0-9]+\]")	p = subprocess.Popen(cmd, stdout=subprocess.PIPE)	while True:		if line := p.stdout.readline():			line = line.decode("utf-8")			if pat.match(line):				line = line.split()				if per_cpu:					# Assumes CPU number is enclosed by []					cpu = int(line[0][1:-1])				else:					cpu = 0				fn(line, cpu, *x)		else:			break	p.wait() def IntersectTimeRanges(new_time_ranges, time_ranges):	pos = 0	new_pos = 0	# Can assume len(time_ranges) != 0 and len(new_time_ranges) != 0	# Note also, there *must* be at least one intersection.	while pos < len(time_ranges) and new_pos < len(new_time_ranges):		# new end < old start => no intersection, remove new		if new_time_ranges[new_pos][1] < time_ranges[pos][0]:			del new_time_ranges[new_pos]			continue		# new start > old end => no intersection, check next		if new_time_ranges[new_pos][0] > time_ranges[pos][1]:			pos += 1			if pos < len(time_ranges):				continue			# no next, so remove remaining			while new_pos < len(new_time_ranges):				del new_time_ranges[new_pos]			return		# Found an intersection		# new start < old start => adjust new start = old start		if new_time_ranges[new_pos][0] < time_ranges[pos][0]:			new_time_ranges[new_pos][0] = time_ranges[pos][0]		# new end > old end => keep the overlap, insert the remainder		if new_time_ranges[new_pos][1] > time_ranges[pos][1]:			r = [ time_ranges[pos][1] + 1, new_time_ranges[new_pos][1] ]			new_time_ranges[new_pos][1] = time_ranges[pos][1]			new_pos += 1			new_time_ranges.insert(new_pos, r)			continue		# new [start, end] is within old [start, end]		new_pos += 1 def SplitTimeRangesByTraceDataDensity(time_ranges, cpus, nr, cmd, file_name, per_cpu, min_size, min_interval, verbosity):	if verbosity.normal:		print("\rAnalyzing...", flush=True, end=" ")		if verbosity.verbose:			print()	cnts_cmd, times_cmd = PerfDoubleQuickCommands(cmd, file_name) 	nr_cpus = cpus[-1] + 1 if per_cpu else 1	if per_cpu:		nr_cpus = cpus[-1] + 1		cpu_time_ranges = [ CPUTimeRange(cpu) for cpu in range(nr_cpus) ]	else:

Lines 1–500 of 988. Load more