#!/usr/bin/env python3

def cli():

import os,shutil,argparse

cli = argparse.ArgumentParser(description='CLARA scaling test')

cli.add_argument('-y','--yaml', metavar='YAML',help='path to YAML file',required=True)

cli.add_argument('-c','--clara', metavar='DIR',help='CLARA_HOME path (default=$CLARA_HOME)',default=os.getenv('CLARA_HOME',None))

cli.add_argument('-t','--threads',metavar='#',help='threads (default=4,8)',default='4,8')

cli.add_argument('-e','--events', metavar='#',help='events per thread (default=555)',default=555,type=int)

cli.add_argument('-N','--numa', metavar='#',help='NUMA socket (default=None, choices=[0,1])',default=None,type=int,choices=[0,1])

cli.add_argument('datafile', help='input EVIO/HIPO data file')

cfg = cli.parse_args()

cfg.threads = cfg.threads.split(',')

if not cfg.clara:

cli.error('cannot find CLARA installation via -c or $CLARA_HOME')

cfg.run_clara = find_run_clara(cfg.clara)

if not cfg.run_clara:

cli.error('cannot find run-clara in $PATH or CLARA installation')

if cfg.numa is not None:

if not shutil.which('numactl'):

cli.error('numactl is not in $PATH, --numa option not supported')

if len(list(get_numa_cpus(cfg.numa))) == 0:

cli.error('invalid --numa node: {cfg.numa}')

cfg.numa = ','.join(list(get_numa_cpus(cfg.numa)))

return cfg

def find_run_clara(clara_home):

import os,shutil

if shutil.which('run-clara'):

return shutil.which('run-clara')

elif os.path.exists(clara_home + '/plugins/clas12/bin/run-clara'):

return clara_home + '/plugins/clas12/bin/run-clara'

def get_numa_cpus(node):

for line in run(['numactl','-H']):

if line.startswith(f'node {node} cpus:'):

for col in line.strip().split()[3:]:

yield col

def run(cmd):

import subprocess

print('run >>> '+' '.join(cmd))

p = subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,universal_newlines=True,encoding='latin-1')

for line in iter(p.stdout.readline, ''):

line = line.strip()

if len(line) > 0:

yield line

p.wait()

if p.returncode != 0:

pass

def benchmark(cfg, threads, log):

import collections

cmd = []

# use taskset:

if cfg.numa is not None:

cmd.extend(['taskset','-c',cfg.numa])

# add the run-clara command:

cmd.extend([cfg.run_clara,

'-c',cfg.clara,

'-n',str(cfg.events*int(threads)),

'-t',str(threads),

'-l',

'-y',cfg.yaml,

'-o',f'tmp-scaling-{threads}',

cfg.datafile])

exiting,benchmarks = False,collections.OrderedDict()

for line in run(cmd):

cols = line.split()

print(line)

log.write(line+'\n')

try:

if line.find('Benchmark results:') >= 0:

exiting = True

elif line.find('Processing is complete') >= 0:

exiting = False

elif len(cols) > 20:

if line.find('Processed') >= 0:

benchmarks['Event'] = float(cols[12])

elif exiting:

# catch-all for services:

if len(cols) > 14:

if 'Services' not in benchmarks:

benchmarks['Services'] = collections.OrderedDict()

benchmarks['Services'][cols[2]] = float(cols[14])

# FIXME: what are these, why don't they add up?

elif line.find('Average processing time') >= 0:

benchmarks['Avg'] = float(cols[6])

elif line.find('Total processing time') >= 0:

benchmarks['Total'] = float(cols[6])

elif line.find('Total orchestrator time') >= 0:

benchmarks['Orch'] = float(cols[6])

except ValueError:

pass

return benchmarks

def table(benchmarks):

table = []

header = [ 'Threads' ]

b = benchmarks[0][1]

header.extend([x for x in b if x != 'Services' and x != 'Event'])

if 'Services' in b:

header.extend(b['Services'].keys())

table.append(header)

for b in benchmarks:

threads,benchmark = b[0],b[1]

row = [threads]

for k in ['Avg','Total','Orch','Services']:

if k in benchmark:

if k == 'Services':

row.extend(benchmark[k].values())

else:

row.append(benchmark[k])

table.append(row)

return table

def show(benchmarks):

for row in table(benchmarks):

print(' '.join([str(x) for x in row]))

def save(benchmarks):

with open('scaling.txt','w') as f:

for row in table(benchmarks):

f.write(' '.join([str(x) for x in row])+'\n')

if __name__ == '__main__':

cfg = cli()

import os

benchmarks = []

for threads in cfg.threads:

os.makedirs('tmp-scaling-'+threads)

with open(f'tmp-scaling-{threads}/run-clara.log','w') as log:

benchmarks.append([threads, benchmark(cfg, threads, log)])

show(benchmarks)

save(benchmarks)