###############################################################################
# (c) Copyright 2000-2022 CERN for the benefit of the LHCb Collaboration #
# #
# This software is distributed under the terms of the GNU General Public #
# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING". #
# #
# In applying this licence, CERN does not waive the privileges and immunities #
# granted to it by virtue of its status as an Intergovernmental Organization #
# or submit itself to any jurisdiction. #
###############################################################################
import GaudiPython as GP
from GaudiConf.reading import decoder, unpack_rawevent, hlt_decisions
from Configurables import (ApplicationMgr, LHCbApp, IODataManager,
EventSelector, createODIN)
from GaudiConf import IOHelper
from PyConf.application import configured_ann_svc
import operator
from collections import Counter
import json
import re
import argparse
import csv
import os
import yaml
'''
Run snippet with 'python line-rates.py and [1] <MDF file name> [2] <TCK config file name> [3] <JSON file name specifying configuration>'
+ '--c' flag with 'production or 'wg' for production or WG stream configuration
When running production-stream config, returns:
Per line (in form of single HTML table):
1. Inclusive retention
2. Inclusive rate
3. Exclusive retention
4. Exclusive rate
5. Average DstData bank size
6. DstData bandwidth
7. Average event size (all banks in particular stream)
8. Bandwidth
Per stream in Turbo/Full/Turcal
1. Inclusive retention
2. Inclusive rate
3. Average DstData bank size
4. DstData bandwidth
5. Average event size (all banks in particular stream)
6. Bandwidth
When running wg-stream config, returns same figures as above (both per line and per stream)
'''
LHCb = GP.gbl.LHCb
RAW_BANK_TYPES = [(i, LHCb.RawBank.typeName(i))
for i in range(LHCb.RawBank.LastType)]
def parse_yaml(file_path):
with open(os.path.expandvars(file_path), 'r') as f:
return yaml.safe_load(f)
def rawbank_sizes(rawevent, lst):
"""Return (name, size) for each raw bank type."""
if rawevent:
def size(i):
return sum(bank.totalSize() for bank in rawevent.banks(i))
else:
def size(i):
return 0
return [(name, size(i)) for i, name in lst]
def processing_events_per_line_and_stream(evt_max, lines, process='Hlt2'):
'''
Returns, per line:
i) How many events triggered on
ii) How many unique events triggered on
iii) Average DstData size of all events
iv) Average size of all events
Returns, per stream:
i) How many events triggered on
ii) Average DstData size of all events
iii) Average size of all events
'''
# Per file (stream) information
events_file = 0
raw_size_all = 0
dst_size_all = 0
# Per line information
# Stores how many events each line fired on
event_stats = {
line: []
for line in [line + 'Decision' for line in list(lines)]
}
# Stores whole event size size
raw = {line: 0 for line in [line + 'Decision' for line in list(lines)]}
# Stores DstData bank size
dst = {line: 0 for line in [line + 'Decision' for line in list(lines)]}
exclusive = {}
# Loop over all events
analysed = 0
while analysed < evt_max:
analysed += 1
exclusive.update({analysed: 0})
# Run an event
appMgr.run(1)
report = evt['/Event/{}/DecReports'.format(process)]
rawevent = evt['/Event/DAQ/RawEvent']
evtsize = sum(
bank[1] for bank in rawbank_sizes(rawevent, RAW_BANK_TYPES))
dstsize = sum(
bank[1] for bank in rawbank_sizes(rawevent, [(60, 'DstData')]))
# Will quit running if there are no more events in the input file
if report:
# Count per file/stream
events_file += 1
raw_size_all += evtsize
dst_size_all += dstsize
for line in event_stats.keys():
# Count per line
if report.decReport(line):
if report.decReport(line).decision() == 1:
event_stats[line].append(analysed)
exclusive[analysed] += 1
raw[line] += evtsize
dst[line] += dstsize
else:
break
# First three variables per stream/file, last four for lines
return events_file, raw_size_all, dst_size_all, event_stats, exclusive, raw, dst
def rates_per_line(event_stats, exclusive, raw, dst, configname, streamname,
input_rate):
data = []
# Compute exclusive rate
sort = dict(
sorted(
{k: v
for (k, v) in exclusive.items() if v > 0}.items(),
key=operator.itemgetter(1),
reverse=True))
unique_events = [key for key, value in sort.items() if value == 1]
for line, val in event_stats.items():
events_all = val + unique_events
num_events = len(event_stats[line])
row_values = (
line,
num_events / LHCbApp().EvtMax * 100
if num_events else 0, # Inclusive Retention (expressed as %)
num_events / LHCbApp().EvtMax * input_rate
if num_events else 0, # Inclusive Rate (in kHz)
len([
key for key, value in Counter(events_all).items() if value > 1
]) / LHCbApp().EvtMax * 100
if num_events else 0, # Exclusive retention (expressed as %)
len([
key for key, value in Counter(events_all).items() if value > 1
]) / LHCbApp().EvtMax * input_rate
if num_events else 0, # Exclusive rate (in kHz)
raw[line] / num_events * 1e-3
if num_events else 0, # Average event size (in kB)
(num_events / LHCbApp().EvtMax * raw[line] / num_events) *
input_rate / 1e6 if num_events else 0, # Event bandwidth (in GB/s)
dst[line] / len(event_stats[line]) * 1e-3
if num_events else 0, # Average DstData size (in kB)
(num_events / LHCbApp().EvtMax * dst[line] / num_events) *
input_rate / 1e6 if num_events else 0
) # DstData Bandwidth (in GB/s)
data.append(row_values)
with open(
f'tmp/Output/Inter/rates-all-lines-{configname}-{streamname}.csv',
'w') as f:
csv_out = csv.writer(f)
for tup in data:
csv_out.writerow(tup)
return
def rates_per_stream(events, raw_size, dst_size, configname, streamname,
input_rate):
data = []
row_values = (
streamname,
events / LHCbApp().EvtMax * 100
if events else 0, # Inclusive Retention (expressed as %)
events / LHCbApp().EvtMax * input_rate
if events else 0, # Inclusive Rate (in kHz)
raw_size / events * 1e-3
if events else 0, # Average event size (in kB)
(events / LHCbApp().EvtMax * raw_size / events) * input_rate / 1e6
if events else 0, # Event bandwidth (in GB/s)
dst_size / events * 1e-3
if events else 0, # Average DstData size (in kB)
(events / LHCbApp().EvtMax * dst_size / events) * input_rate / 1e6
if events else 0) # DstData Bandwidth (in GB/s)
data.append(row_values)
with open(
f'tmp/Output/Inter/rates-per-stream-{configname}-{streamname}.csv',
'w') as f:
csv_out = csv.writer(f)
for tup in data:
csv_out.writerow(tup)
return
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Inspect Moore output')
parser.add_argument(
'-i', '--input', type=str, help='MDF input file', required=True)
parser.add_argument(
'-n',
'--events',
default=-1,
type=lambda x: int(round(float(x))),
help='nb of events to process',
required=True)
parser.add_argument(
'-c',
'--config',
type=str,
required=True,
help='Path to yaml config file defining the input.')
parser.add_argument(
'-r',
'--rate',
default=500, # kHz
type=float,
help='Input rate of the process in kHz')
parser.add_argument(
'-t',
'--tck',
type=str,
help='Manifest file for chosen MDF',
required=True)
parser.add_argument(
'-j',
'--json',
type=str,
help='Stream configuration specified as JSON',
required=True)
parser.add_argument(
'-s',
'--stream-config',
type=str,
help='Choose production or per-WG stream configuration',
choices=['production', 'wg'],
required=True)
parser.add_argument(
'-p',
'--process',
type=str,
help='Compute for Hlt2 or Sprucing lines',
choices=['Hlt2', 'Spruce'],
required=True)
args = parser.parse_args()
n_events = args.events
input_config = parse_yaml(args.config)
LHCbApp(
DataType="Upgrade",
Simulation=True,
DDDBtag="dddb-20171126",
CondDBtag="sim-20171127-vc-md100",
EvtMax=n_events)
EventSelector().PrintFreq = 10000
IODataManager(DisablePFNWarning=True)
# we have to configure the algorithms manually instead of `do_unpacking`
# because we need to set `input_process='Hlt2'` in `unpack_rawevent`
# to read MDF output from Sprucing
algs = []
unpack = unpack_rawevent(
bank_types=['ODIN', 'HltDecReports', 'DstData', 'HltRoutingBits'],
configurables=True)
hlt2 = [hlt_decisions(source="Hlt2", output_loc="/Event/Hlt2/DecReports")]
if args.process == 'Spruce':
spruce = [
hlt_decisions(
source="Spruce", output_loc="/Event/Spruce/DecReports")
]
else:
spruce = []
decoder = decoder(input_process=args.process)
algs = [unpack] + hlt2 + spruce + [decoder] + [createODIN(ODIN='myODIN')]
appMgr = ApplicationMgr(TopAlg=algs)
appMgr.ExtSvc += [configured_ann_svc(json_file=args.tck)]
file = args.input
# if no line is registered in one module,
# the output file won't be created.
# In this case we just create an empty file.
if not os.path.exists(file): os.system('touch {}'.format(file))
IOHelper("MDF").inputFiles([file])
with open(args.json) as f:
config = json.load(f)
if args.process == 'Hlt2':
# Two conditions for HLT2 run:
# Use production-stream config to compute rate/size/bandwidth per line and stream
# Use wg-stream config to compute rate/size/bandwidth per line and stream
if args.stream_config == 'production': configname = 'production'
elif args.stream_config == 'wg': configname = 'wg'
elif args.process == 'Spruce':
# Three conditions for Spruce run:
# Use wg-stream config to compute rate/size/bandwidth per line
# Use wg-stream config to compute rate/size/bandwidth per stream
if not args.stream_config == 'wg': exit()
configname = 'wg-stream'
stream = str(re.search("-(?!.*-)(.*).mdf", file).group(
1)) # Finds string between last - and .mdf suffix = stream identifier
lines = config[stream]
appMgr = GP.AppMgr()
evt = appMgr.evtsvc()
# Calculates retention, rate and bandwidth per line and stream (file)
evts_all, rawbanks_all, dst_all, event_stats, exclusive, raw, dst = processing_events_per_line_and_stream(
LHCbApp().EvtMax, lines, process=args.process)
rates_per_line(
event_stats,
exclusive,
raw,
dst,
configname,
stream,
input_rate=input_config['input_rate'])
rates_per_stream(
evts_all,
rawbanks_all,
dst_all,
configname,
stream,
input_rate=input_config['input_rate'])