diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c72c5b07a1..c38f661b6e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -247,12 +247,10 @@ jobs: tar xzf clara.tar.gz tar xzf coatjava.tar.gz - run: ls - - name: run test - run: ./bin/run-clara -y ./etc/services/rgd-clarode.yml -t 4 -n 500 -c ./clara -o ./tmp ./clas_018779.evio.00001 - - name: ls tmp - run: ls -lhtr tmp - - name: rename - run: mv -v tmp/rec_clas_018779.evio.00001.hipo rec.hipo + - name: run clara + run: ./coatjava/bin/run-clara -y ./etc/services/rgd-clarode.yml -t 4 -n 100 -c ./clara -o ./tmp ./clas_018779.evio.00001 + - name: run mutil + run: ./coatjava/bin/recon-mutil -y ./etc/services/rgd-clarode.yml -t 4 -o rec.hipo ./clas_018779.evio.00001 - uses: actions/upload-artifact@v7 with: name: test_clara_result diff --git a/.gitignore b/.gitignore index 9f0aad0fba..51862428ed 100644 --- a/.gitignore +++ b/.gitignore @@ -58,3 +58,6 @@ hs_err_pid* bin/evio2hipotest export.sh + +*.txt + diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index d42ca40d5f..8e4a12eb63 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -175,13 +175,24 @@ clara: - run-clara -v -c $CLARA_HOME -t 4 -y ./etc/services/rgd-clarode.yml -n 30 -o out $EVIOFILE - mv out/rec_$EVIOFILE.hipo claroded.hipo +recon-mutil: + stage: test + needs: [build,download] + dependencies: [build,download] + script: + - tar -xzf coatjava.tar.gz + - recon-mutil -t 4 -y ./etc/services/rgd-clarode.yml -n 30 -o rec_$EVIOFILE.hipo $EVIOFILE + profile: extends: .clon allow_failure: true script: - - ./clara/plugins/clas12/libexec/profile -e cpu run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp1 -t 16 -l $EVIOFILE - - ./clara/plugins/clas12/libexec/profile -e alloc run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp2 -t 16 -l $EVIOFILE - - ./clara/plugins/clas12/libexec/profile -e lock run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp3 -t 16 -l $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e cpu -o _clara run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp1 -t 16 -l $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e alloc -o _clara run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp2 -t 16 -l $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e lock -o _clara run-clara -c ./clara -y ./etc/services/rgd-clarode.yml -o tmp3 -t 16 -l $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e cpu -o _mutil recon-mutil -y ./etc/services/rgd-clarode.yml -o mutil1.hipo -t 16 -l SEVERE $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e alloc -o _mutil recon-mutil -y ./etc/services/rgd-clarode.yml -o mutil2.hipo -t 16 -l SEVERE $EVIOFILE + - ./clara/plugins/clas12/libexec/profile -e lock -o _mutil recon-mutil -y ./etc/services/rgd-clarode.yml -o mutil3.hipo -t 16 -l SEVERE $EVIOFILE artifacts: when: always expire_in: 1 week @@ -197,7 +208,7 @@ scaling: when: always expire_in: 1 week paths: - - scaling.svg + - scaling-clara.svg pages: stage: deploy @@ -207,7 +218,7 @@ pages: - cp -r pages public - mkdir public/perf - cp asprof*.html public/perf - - cp scaling.svg public/perf + - cp scaling*.svg public/perf pages: true artifacts: expire_in: 1 week diff --git a/bin/reco-mutil b/bin/reco-mutil new file mode 100755 index 0000000000..8d061f4c45 --- /dev/null +++ b/bin/reco-mutil @@ -0,0 +1,18 @@ +#!/bin/bash + +. `dirname $0`/../libexec/env.sh + +split_cli $@ + +export MALLOC_ARENA_MAX=1 + +#threads=$(get_threads) +#node=$(((RANDOM % 2))) +#cpus=$(get_all_numa_cpus $node) +#cpus=$(get_numa_cpus $node) +#taskset -c $cpus \ + + java ${JAVA_OPTS-} -XX:MaxRAMPercentage=40.0 -Xms10240m ${jvm_options[@]} \ + -cp ${COATJAVA_CLASSPATH:-''} \ + org.jlab.clas.reco.RecoMutil \ + ${class_options[@]} diff --git a/bin/recon-mutil b/bin/recon-mutil new file mode 100755 index 0000000000..f604441243 --- /dev/null +++ b/bin/recon-mutil @@ -0,0 +1,16 @@ +#!/bin/bash + +. `dirname $0`/../libexec/env.sh + +split_cli $@ + +export MALLOC_ARENA_MAX=12 + +threads=$(get_threads) + + #java ${JAVA_OPTS-} -Xmx256g -Xms10240m ${jvm_options[@]} \ + #java ${JAVA_OPTS-} -Xms10240m -XX:+UseParallelGC ${jvm_options[@]} \ + java ${JAVA_OPTS-} -XX:MaxRAMPercentage=40.0 -Xms10240m ${jvm_options[@]} \ + -cp ${COATJAVA_CLASSPATH:-''} \ + org.jlab.clas.reco.ReconMutil \ + ${class_options[@]} diff --git a/bin/recon-util b/bin/recon-util index 90f5a6324f..32843f637f 100755 --- a/bin/recon-util +++ b/bin/recon-util @@ -4,7 +4,7 @@ split_cli $@ -export MALLOC_ARENA_MAX=1 +export MALLOC_ARENA_MAX=12 java ${JAVA_OPTS-} -Xmx1536m -Xms1024m -XX:+UseSerialGC ${jvm_options[@]} \ -cp ${COATJAVA_CLASSPATH:-''} \ diff --git a/common-tools/clara-io/pom.xml b/common-tools/clara-io/pom.xml index 1c3a50f9b6..e192efa6b7 100644 --- a/common-tools/clara-io/pom.xml +++ b/common-tools/clara-io/pom.xml @@ -48,12 +48,6 @@ 14.2.1-SNAPSHOT - - org.jlab.clas - clas-utils - 14.2.1-SNAPSHOT - - org.apache.commons commons-text diff --git a/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java b/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java index c262aa3d2b..8b2385f7f3 100644 --- a/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java +++ b/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java @@ -1,17 +1,13 @@ package org.jlab.io.clara; -import java.io.File; import java.nio.file.Path; -import java.util.List; import org.jlab.clara.std.services.EventWriterException; import org.jlab.detector.calib.utils.ConstantsManager; -import org.jlab.detector.helicity.HelicitySequenceDelayed; import org.jlab.detector.serial.SerialHoncho; -import org.jlab.detector.serial.PostProcessor; +import org.jlab.detector.serial.Occupancer; import org.jlab.jnp.hipo4.data.Bank; import org.jlab.jnp.hipo4.data.Event; import org.jlab.jnp.hipo4.data.SchemaFactory; -import org.jlab.jnp.hipo4.io.HipoReader; import org.jlab.jnp.hipo4.io.HipoWriterSorted; import org.jlab.jnp.utils.file.FileUtils; import org.json.JSONObject; @@ -21,18 +17,17 @@ * 1. Copies certain banks on-the-fly to new tag-1 events * 2. Caches helicity states, scaler readouts, and unix time * 3. Writes HEL::flip, RUN/HEL::scaler, and RUN::unix to new tag-1 events - * 4. Runs post-processing, writing tag-1 information to all events * 5. Adds .hipo to the output filename, if necessary * 6. Outputs a parallel file of select trigger bits with a raw schema * @author baltzell */ public class Clas12Writer extends HipoToHipoWriter { + Occupancer occupancer; SerialHoncho serial; - Bank runConfig; ConstantsManager conman; SchemaFactory fullSchema; - boolean postprocess; + Bank runConfig; HipoWriterSorted paraWriter; long paraCount; @@ -40,13 +35,13 @@ public class Clas12Writer extends HipoToHipoWriter { int paraTriggerPrescale; private void init(JSONObject opts) { + occupancer = new Occupancer(); fullSchema = new SchemaFactory(); fullSchema.initFromDirectory(FileUtils.getEnvironmentPath("CLAS12DIR","etc/bankdefs/hipo4")); serial = new SerialHoncho(fullSchema); - runConfig = new Bank(fullSchema.getSchema("RUN::config")); conman = new ConstantsManager(); + runConfig = new Bank(fullSchema.getSchema("RUN::config")); conman.init("/runcontrol/hwp","/runcontrol/helicity"); - postprocess = opts.optBoolean("postprocess", false); paraTriggerMask = opts.optLong("paraTriggerMask", 0); paraTriggerPrescale = opts.optInt("paraTriggerPrescale", 0); if (opts.has("variation")) conman.setVariation(opts.getString("variation")); @@ -78,6 +73,7 @@ protected HipoWriterSorted createWriter(Path file, JSONObject opts) throws Event protected void writeEvent(Object event) throws EventWriterException { Event t = serial.read((Event)event); if (!t.isEmpty()) writer.addEvent(t, 1); + occupancer.process(((Event)event)); super.writeEvent(event); writeRaw((Event)event, t); } @@ -96,48 +92,8 @@ void writeRaw(Event physics, Event tagged) { protected void closeWriter() { serial.closure(writer); super.closeWriter(); - if (postprocess) postprocess(); serial.clear(); if (paraTriggerMask > 0) paraWriter.close(); } - - /** - * Get the first valid run number from a RUN::config bank. - * @return run - */ - private int getRunNumber() { - Event e = new Event(); - HipoReader r = new HipoReader(); - r.open(filename); - while (r.hasNext()) { - r.nextEvent(e); - e.read(runConfig); - if (runConfig.getRows()>0 && runConfig.getInt("run",0)>0) - return runConfig.getInt("run",0); - } - return 0; - } - - /** - * Copy helicity/charge tag-1 information to all events. - */ - private void postprocess() { - int d = conman.getConstants(getRunNumber(), "/runcontrol/helicity").getIntValue("delay",0,0,0); - HelicitySequenceDelayed helicity = new HelicitySequenceDelayed(d); - helicity.addStream(serial.getHelicities()); - PostProcessor p = new PostProcessor(List.of(filename), fullSchema, helicity, serial.getScalers()); - HipoReader r = new HipoReader(); - r.open(filename); - Event e = new Event(); - writer.open("pp_"+filename); - while (r.hasNext()) { - r.nextEvent(e); - p.processEvent(e); - HipoToHipoWriter.writeEvent(writer, e, schemaBankList); - } - writer.close(); - new File(filename).delete(); - new File("pp_"+filename).renameTo(new File(filename)); - } } diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/Occupancer.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/Occupancer.java new file mode 100644 index 0000000000..92e50b85b1 --- /dev/null +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/Occupancer.java @@ -0,0 +1,183 @@ +package org.jlab.detector.serial; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Map; +import org.jlab.detector.banks.RawBank; +import org.jlab.utils.groups.IndexedTable; +import org.jlab.utils.groups.IndexedTable.IndexedEntry; +import org.jlab.detector.banks.RawBank.OrderGroups; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.Event; +import org.jlab.jnp.hipo4.data.Schema; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.utils.system.ClasUtilsFile; + +/** + * Occupancy bookkeeper based on IndexedTable, with I/O helpers for indexed banks. + * + * @author baltzell + */ +public class Occupancer extends ArrayList { + + static final String BANKDIR = ClasUtilsFile.getResourceDir("CLAS12DIR","etc/bankdefs/hipo4/singles/occupancy"); + static final SchemaFactory FULL_SCHEMA = new SchemaFactory(); + static { FULL_SCHEMA.initFromDirectory(ClasUtilsFile.getResourceDir("CLAS12DIR","etc/bankdefs/hipo4")); } + + int prescale = 1000; + int nevents = 0; + + public Occupancer() { + super(); + init(); + } + + public Occupancer(int prescale) { + super(); + this.prescale = prescale; + init(); + } + + private void init() { + try { + // Initialize OccupanceTable list, by reading occupancy bank filenames + // and stripping ".json" suffix and "OCC::" prefix to get hit bank name: + addAll(Files.list(Paths.get(BANKDIR)) + .filter(Files::isRegularFile) + .map(p -> p.getFileName().toString()) + .map(s -> s.substring(0, s.length()-5)) + .map(s -> s.substring(5, s.length())) + .map(OccupanceTable::new).toList()); + } catch (IOException ex) { + System.getLogger(Occupancer.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); + } + } + + public boolean process(Event event) { + forEach(t -> { t.fill(event, false); }); + if (++nevents % prescale == 0) { + forEach(t -> { + if (t.getTable().getRowCount() > 0) { + event.write(t.create(nevents, event)); + } + t.reset(); + }); + nevents = 0; + } + return true; + } + + public void reset() { + forEach(t -> { t.reset(); nevents = 0; }); + } + + public static class OccupanceTable { + + Schema occSchema; + Schema hitSchema; + IndexedTable table; + + /** + * A 3-index table, e.g., sector/layer/component. + * @param hitBank name of the hit bank + */ + public OccupanceTable(String hitBank) { + hitSchema = FULL_SCHEMA.getSchema(hitBank); + occSchema = FULL_SCHEMA.getSchema("OCC::" + hitBank); + table = new IndexedTable(occSchema.hasEntry("order")?4:3, new String[]{"occ/F"}); + } + + /** + * An N-index table. + * @param hitBank name of the hit bank + * @param indexCount number of inidices in the hit bank + */ + public OccupanceTable(String hitBank, int indexCount) { + hitSchema = FULL_SCHEMA.getSchema(hitBank); + occSchema = FULL_SCHEMA.getSchema("OCC::" + hitBank); + table = new IndexedTable(indexCount, new String[]{"occ/F"}); + } + + public final IndexedTable getTable() { return table; } + + /** + * Zero the occupancy table. + */ + public final void reset() { + table = new IndexedTable(table.getList().getIndexSize(), new String[]{"occ/F"}); + } + + /** + * Get the occupancy table, normalized by number of events. + * @param events + * @return + */ + public final IndexedTable getOccupancy(long events) { + IndexedTable t = new IndexedTable(table.getList().getIndexSize(), new String[]{"occ/F"}); + for (long hash : ((Map)table.getList().getMap()).keySet()) { + t.addEntry(IndexedTable.DEFAULT_GENERATOR.getIndices(hash, table.getList().getIndexSize())); + t.setDoubleValueByHash((table.getDoubleValueByHash(0, hash))/events, 0, hash); + } + return t; + } + + /** + * Fill the occupancy table. + * @param weight + * @param index + */ + public final void fill(float weight, int... index) { + for (int i=0; i m = table.getList().getMap(); + int i = 0; + for (long hash : m.keySet()) { + int[] idx = IndexedTable.DEFAULT_GENERATOR.getIndices(hash, table.getList().getIndexSize()); + for (int j=0; j eventUnix; volatile HelicitySequence helicitySequence; @@ -48,6 +51,9 @@ public SerialHoncho(SchemaFactory schema) { recEvent = schema.getSchema("REC::Event"); helicityAdc = schema.getSchema("HEL::adc"); helScaler = schema.getSchema("HEL::scaler"); + runScaler = schema.getSchema("RUN::scaler"); + rawScaler = schema.getSchema("RAW::scaler"); + epics = schema.getSchema("RAW::epics"); scalers = new DaqScalersSequence(schema); helicities = new TreeSet<>(); eventUnix = new TreeMap<>(); @@ -56,6 +62,10 @@ public SerialHoncho(SchemaFactory schema) { tag1banks[i] = schema.getSchema(TAG1BANKS[i]); } + public boolean containsSerial(Event e) { + return e.isEmpty() ? false : e.hasBank(rawScaler) || e.hasBank(epics); + } + /** * Register an event's serial data and return a (possibly empty) tag-1 event. * @param event @@ -182,16 +192,16 @@ void prune() { * @param runConfig * @return */ - Event getUnixEvent(Bank config) { + Event getUnixEvent(Bank runConfig) { Bank unix = new Bank(schema.getSchema("RUN::unix"), eventUnix.size()); int row = 0; for (int evno : eventUnix.keySet()) { unix.putInt("event", row, evno); - unix.putInt("unixtime",row, eventUnix.get(evno)); + unix.putInt("unixtime", row, eventUnix.get(evno)); row++; } Event e = new Event(); - e.write(config); + e.write(runConfig); e.write(unix); return e; } diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineProcessor.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineProcessor.java index dfa428f7ec..5c920e8d9b 100644 --- a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineProcessor.java +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineProcessor.java @@ -19,6 +19,7 @@ import java.util.Arrays; import org.jlab.coda.jevio.EvioException; import org.jlab.detector.decode.CLASDecoder4; +import org.jlab.detector.serial.Occupancer; import org.jlab.io.evio.EvioDataEvent; import org.jlab.io.evio.EvioSource; import org.jlab.io.hipo.HipoDataEvent; @@ -42,7 +43,8 @@ public class EngineProcessor { private SchemaFactory banksToKeep = null; private final List schemaExempt = Arrays.asList("RUN::config","DC::tdc"); - private CLASDecoder4 decoder = new CLASDecoder4(); + private final CLASDecoder4 decoder = new CLASDecoder4(); + private final Occupancer occupancer = new Occupancer(); public EngineProcessor(){} @@ -134,7 +136,7 @@ public void initDefault(){ String[] names = new String[]{ "MAGFIELDS", "DCCR","DCHB","FTOFHB","EC","HTCC","EBHB", - "DCTB","FTOFTB","EBTB","VTX" + "DCTB","FTOFTB","EBTB","OCC" }; String[] services = new String[]{ @@ -148,7 +150,7 @@ public void initDefault(){ "org.jlab.service.dc.DCTBEngine", "org.jlab.service.ftof.FTOFTBEngine", "org.jlab.service.eb.EBTBEngine", - "org.jlab.rec.service.vtx.VTXEngine" + "org.jlab.calibration.service.OccupanceEngine", }; for(int i = 0; i < names.length; i++){ @@ -164,7 +166,7 @@ public void initAll(){ "CVTFP","CTOF","CND","BAND", "HTCC","LTCC","EBHB", "DCTB","FMT","FTOFTB","CVT","EBTB", - "RICHEB","RTPC","AHDC","ATOF","ALERT", "MC","VTX" + "RICHEB","RTPC","AHDC","ATOF","ALERT", "MC","VTX","OCC" }; String[] services = new String[]{ @@ -197,7 +199,8 @@ public void initAll(){ "org.jlab.service.atof.ATOFEngine", "org.jlab.service.alert.ALERTEngine", "org.jlab.service.mc.TruthMatch", - "org.jlab.rec.service.vtx.VTXEngine" + "org.jlab.rec.service.vtx.VTXEngine", + "org.jlab.calibration.service.OccupanceEngine", }; if(names.length!=services.length) LOGGER.log(Level.SEVERE, "initAll : the list of services does not match the list of service names... "); @@ -316,6 +319,7 @@ public void processFile(String file, String output){ public void processEvent(DataEvent event, HipoDataSync writer) { processEvent(event); removeBanks(event); + occupancer.process(((HipoDataEvent)event).getHipoEvent()); writer.writeEvent(event); } @@ -390,24 +394,6 @@ public void show(){ } } - protected static OptionParser getParser() { - OptionParser parser = new OptionParser("recon-util"); - parser.addRequired("-o","output.hipo"); - parser.addRequired("-i","input.evio/hipo"); - parser.setRequiresInputList(false); - parser.addOption("-c","0","use default configuration [0 - no, 1 - yes/default, 2 - all services] "); - parser.addOption("-s","0","number of events to skip"); - parser.addOption("-n","0","number of events to process"); - parser.addOption("-y","0","yaml file"); - parser.addOption("-u","true","update dictionary from writer ? "); - parser.addOption("-S",null,"schema directory"); - parser.addOption("-B",null,"background file"); - parser.addOption("-P",null,"preload file for post-processing"); - parser.addOption("-R","0","rebuild scalers"); - parser.addOption("-H","0","restream helicity"); - return parser; - } - protected final void init(OptionParser p) { p.syncLogLevel(LOGGER); @@ -449,7 +435,7 @@ else if (p.getOption("-c").intValue() > 0) { } public static void main(String[] args) { - OptionParser parser = EngineProcessor.getParser(); + OptionParser parser = ReconUtil.getParser(); parser.parse(args); EngineProcessor proc = new EngineProcessor(parser); proc.processFile(parser.getOption("-i").stringValue(), diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/Porch.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/Porch.java new file mode 100644 index 0000000000..4b2e31fae6 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/Porch.java @@ -0,0 +1,303 @@ +package org.jlab.clas.reco; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TimerTask; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jlab.io.hipo.HipoDataEvent; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.benchmark.BenchmarkTimer; +import org.jlab.utils.benchmark.ProgressPrintout; + +/** + * Parallel orchestrator. + * + */ +public abstract class Porch { + + abstract Object open(String filename); + abstract Object read(); + abstract HipoDataEvent[] decode(int thread, Object event); + abstract void process(int thread, HipoDataEvent event); + abstract void write(HipoDataEvent event); + abstract void readerExit(); + abstract void decoderExit(int thread); + abstract void writerExit(); + abstract void serial(); + + static final String[] BENCHMARK_NAMES = new String[]{"evio","deco","serial","post","write"}; + + // Performance parameters: + final int BENCH_SECONDS = 30; + final int EVENTS_PER_CHUNK = 50; + + // Static parameters: + int maxEvents; + int skipEvents; + String taskset; + + // Threads: + CompletableFuture readerThread; + CompletableFuture writerThread; + ConcurrentLinkedQueue decoThreads = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue procThreads = new ConcurrentLinkedQueue<>(); + + // Queues: + ConcurrentLinkedQueue> decoQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> procQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> writeQueue = new ConcurrentLinkedQueue<>(); + + // Progress counters: + volatile int readEvents; + volatile int failEvents; + volatile int fileEvents; + volatile int maxFileEvents; + volatile int writeEvents; + volatile AtomicInteger taggedEvents = new AtomicInteger(); + volatile ProgressPrintout progress = new ProgressPrintout(); + volatile Benchmark benchmark = null; + TimerTask statsShow = new TimerTask() { @Override public void run() { show(); } }; + + // Event reader/source: + Object reader; + + /** + * The thread launcher and collector. + * @param threads number of threads + * @param input names of input files to read + */ + public void launch(int[] threads, String... input) { + + reset(); + + System.out.println(String.format("recon-mutil:: spawning 2*%d+2 threads",threads[0])); + + // start a period status printout: + ReconUtil.runPeriodic(10, statsShow); + + // spawn all the threads: + readerThread = CompletableFuture.runAsync(() -> { reader(threads[0], input); }); + for (int i=0; i { decoder(j); })); + } + ReconUtil.launch(CompletableFuture.runAsync(() -> { serial(); })); + ReconUtil.sleep(1000); + for (int i=0; i { processor(j); })); + } + writerThread = ReconUtil.launch(CompletableFuture.runAsync(() -> { writer(); })); + + // perform scaling test: + if (threads.length > 1) + ReconUtil.launch(CompletableFuture.runAsync(() -> { rethreader(threads); })).join(); + + // wait for finish: + writerThread.join(); + } + + /** + * The reader thread. + * @param threads + * @param input input filenames + */ + final void reader(int threads, String... input) { + List inputs = new ArrayList<>(Arrays.asList(input)); + List output = new ArrayList<>(EVENTS_PER_CHUNK); + while ( (maxEvents < 1 || writeEvents < maxEvents+taggedEvents.get()) && + (maxFileEvents < 1 || fileEvents < maxFileEvents) ) { + if (reader != null) { + // sleep instead of overfilling the queue (100K events, ~2GB): + if (decoQueue.size()+procQueue.size() > 1e5) ReconUtil.sleep(1000); + else { + Object o = read(); + if (o != null && (skipEvents < 1 || readEvents > skipEvents)) { + output.add(o); + if (output.size() >= EVENTS_PER_CHUNK) { + decoQueue.offer(output); + readEvents += output.size(); + output = new ArrayList<>(EVENTS_PER_CHUNK); + } + } + } + } + else if (inputs.isEmpty()) break; + else reader = open(inputs.removeFirst()); + } + // write leftover, partial chunk: + if (!output.isEmpty()) { + readEvents += output.size(); + decoQueue.offer(output); + } + readerExit(); + } + + /** + * The decoder thread. + * @param thread thread number + */ + final void decoder(int thread) { + while (true) { + List input = decoQueue.poll(); + if (input == null) { + if (decoQueue.isEmpty() && readerThread.isDone() && decoQueue.isEmpty()) + break; + ReconUtil.sleep(100); + } + else { + List output = new ArrayList<>(input.size()); + for (int i=0; i 0 && writeEvents > maxEvents+taggedEvents.get()) { + stop(); + break; + } + List input = procQueue.poll(); + if (input == null) { + if (procQueue.isEmpty() && decoThreads.isEmpty() && procQueue.isEmpty()) { + if (writeEvents+skipEvents+failEvents >= readEvents+taggedEvents.get()) + break; + } + ReconUtil.sleep(100); + } + else { + List output = new ArrayList<>(input.size()); + for (int i=0; i e = writeQueue.poll(); + if (e == null) { + if (decoThreads.isEmpty() && procThreads.isEmpty() && procQueue.isEmpty() && writeQueue.isEmpty()) + break; + ReconUtil.sleep(1000); + } + else { + for (int i=0; i benches = new LinkedHashMap<>(); + Map progs = new LinkedHashMap<>(); + while (!ReconUtil.isDone(decoThreads)) ReconUtil.sleep(100); + System.out.println("recon-mutil:: ~~~~~~~~~ rethreading primed ~~~~~~~~~"); + for (int thread : threads) { + if (taskset.equals("-")) ReconUtil.taskset(0, thread); + BenchmarkTimer.WARMUP_CALLS = 10*thread+90; + benchmark = new Benchmark(thread+" Threads Scaling ",BENCHMARK_NAMES); + progress = new ProgressPrintout(10*thread); + for (int j=0; j { processor(k); })); + } + while (!progress.warmedUp() && !procQueue.isEmpty()) ReconUtil.sleep(100); + ReconUtil.sleep(BENCH_SECONDS *1000); + System.out.printf("\nrecon-mutil:: ~~~~~~~~~ rethreading timed %d ~~~~~~~~~\n%n",thread); + System.out.println(progress.getUpdateString()); + System.out.println(benchmark); + benches.put(thread, benchmark); + progs.put(thread, progress); + procThreads.forEach(p -> p.cancel(true)); + } + String fcsv = String.format("scaling-mutil%s.txt",taskset); + ReconUtil.writeFile(fcsv, ReconUtil.toCSV(progs, benches)); + ReconUtil.gnuplotScaling(fcsv, String.format("scaling-mutil%s.svg",taskset)); + stop(); + } + + /** + * Print the thread, queue, and event states. + */ + void show() { + String s1 = String.format("threads(r/d/p/w)=(%b/%b:%d/%b:%d/%b)", + readerThread != null ? !readerThread.isDone() : false, + !ReconUtil.isDone(decoThreads), + decoThreads.size(), + !ReconUtil.isDone(procThreads), + procThreads.size(), + writerThread != null ? !writerThread.isDone() : false); + String s2 = String.format(" queues(d/p/w)=(%d/%d/%d)", + decoQueue.size()*EVENTS_PER_CHUNK, + procQueue.size()*EVENTS_PER_CHUNK, + writeQueue.size()*EVENTS_PER_CHUNK); + String s3 = String.format(" events(r/w/t/f)=(%d/%d/%d/%d)", + readEvents, writeEvents, taggedEvents.get(), failEvents); + Logger.getLogger(ReconMutil.class.getName()).log(Level.CONFIG, () -> s1+" "+s2+" "+s3); + } + + /** + * Cancel threads, empty queues, and reset counters. + */ + void reset() { + //serialPause.set(true); + stop(); + if (writerThread != null) { + //if (writer != null) writer.close(); + writerThread.cancel(true); + //if (!parser.getOption("-t").stringValue().contains(",")) + // if (benchmark != null) System.out.println(benchmark); + } + readEvents = 0; + writeEvents = 0; + failEvents = 0; + taggedEvents.set(0); + } + + /** + * Cancel threads and empty the queues. + */ + void stop() { + if (readerThread != null) readerThread.cancel(true); + for (CompletableFuture cf : decoThreads) cf.cancel(true); + for (CompletableFuture cf : procThreads) cf.cancel(true); + decoQueue.clear(); + procQueue.clear(); + writeQueue.clear(); + } +} diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/RecoMutil.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/RecoMutil.java new file mode 100644 index 0000000000..fca6785750 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/RecoMutil.java @@ -0,0 +1,339 @@ +package org.jlab.clas.reco; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jlab.coda.jevio.EvioException; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.detector.decode.CLASDecoderPool; +import org.jlab.detector.serial.Occupancer; +import org.jlab.detector.serial.SerialHoncho; +import org.jlab.io.evio.EvioDataEvent; +import org.jlab.io.evio.EvioSource; +import org.jlab.io.hipo.HipoDataEvent; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.Event; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.jnp.hipo4.io.HipoReader; +import org.jlab.jnp.hipo4.io.HipoWriterSorted; +import org.jlab.utils.ClaraYaml; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.options.OptionParser; +import org.jlab.utils.options.OptionValue; +import org.jlab.utils.system.ClasUtilsFile; +import org.json.JSONObject; + +public class RecoMutil extends Porch { + + // Static parameters: + ClaraYaml yaml; + double[] fields = null; + + // I/O + HipoWriterSorted writer; + List schemaBankList; + static final SchemaFactory fullSchema = new SchemaFactory(); + static { fullSchema.initFromDirectory(ClasUtilsFile.getResourceDir("CLAS12DIR","etc/bankdefs/hipo4")); } + + // Processors: + SerialHoncho serial; + Occupancer occupancer = new Occupancer(); + CLASDecoderPool decoders = new CLASDecoderPool(64,"default",null); + Map engines = new LinkedHashMap<>(); + + // Control flags: + final Object serialLock = new Object(); + AtomicInteger serialTrigger = new AtomicInteger(0); + + public RecoMutil(OptionParser parser) { + init(parser); + } + + @Override + Object open(String filename) { + if (reader != null && reader instanceof EvioSource) + ((EvioSource)reader).close(); + fileEvents = 0; + if (filename.endsWith(".hipo")) { + reader = new HipoReader(); + ((HipoReader)reader).open(filename); + maxFileEvents = ((HipoReader)reader).getEventCount(); + } + else { + reader = new EvioSource(); + ((EvioSource)reader).open(filename); + maxFileEvents = ((EvioSource)reader).getEventCount(); + } + return reader; + } + + @Override + Object read() { + if (benchmark != null) benchmark.resume("read"); + Object o = null; + if (reader instanceof EvioSource evio) { + try { o = evio.getEventBuffer(++fileEvents, true); } + catch (EvioException ex) { + failEvents++; + ex.printStackTrace(); + } + } + else { + Event event = new Event(); + o = ((HipoReader)reader).getEvent(event, fileEvents++); + } + if (benchmark != null) benchmark.resume("pause"); + return o; + } + + @Override + void readerExit() { + if (reader != null && reader instanceof EvioSource) + ((EvioSource)reader).close(); + } + + @Override + HipoDataEvent[] decode(int thread, Object input) { + HipoDataEvent event = input instanceof ByteBuffer + ? decode(thread, (ByteBuffer)input) + : new HipoDataEvent(((Event)input), fullSchema); + if (benchmark != null) benchmark.resume(thread, "serial"); + Event tag; + synchronized (serialLock) { + tag = serial.read(event.getHipoEvent()); + } + if (!tag.isEmpty()) { + if (serial.containsSerial(tag)) serialTrigger.incrementAndGet(); + taggedEvents.incrementAndGet(); + } + if (benchmark != null) benchmark.pause(thread, "serial"); + return tag.isEmpty() ? + new HipoDataEvent[]{event} : + new HipoDataEvent[]{event, new HipoDataEvent(tag, fullSchema)}; + } + + @Override + void decoderExit(int thread) { + serialTrigger.incrementAndGet(); + } + + @Override + void process(int thread, HipoDataEvent event) { + for (Map.Entry engine : engines.entrySet()) { + if (benchmark != null) benchmark.resume(thread, engine.getKey()); + try { engine.getValue().processDataEvent(event); } + catch (Exception ex) { ex.printStackTrace(); } + if (benchmark != null) benchmark.pause(thread, engine.getKey()); + } + } + + @Override + void write(HipoDataEvent dataEvent) { + Event event = dataEvent.getHipoEvent(); + if (benchmark != null) benchmark.resume("post"); + synchronized (serialLock) { + serial.process(dataEvent.getHipoEvent()); + } + if (benchmark != null) { + benchmark.pause("post"); + benchmark.resume("write"); + } + if (writer != null) { + if (event.getEventTag() > 0 || schemaBankList.isEmpty()) + writer.addEvent(event, event.getEventTag()); + else + writer.addEvent(event.reduceEvent(schemaBankList), event.getEventTag()); + } + occupancer.process(event); + if (benchmark != null) benchmark.pause("write"); + } + + @Override + void writerExit() { + if (writer != null) { + serial.closure(writer); + writer.close(); + } + } + + /** + * Do some serial stuff, when triggered. + */ + @Override + void serial() { + boolean first = true; + while (true) { + if (serialTrigger.get() > 0) { + first = false; + // sleep to collect more triggers: + ReconUtil.sleep(1000); + // get current number of triggers: + int t = serialTrigger.get(); + // do the protected stuff: + synchronized (serialLock) { serial.updateHelicitySequence(); } + // remove the same number of triggers: + for (int i=0; i Integer.parseInt(s)).sorted().boxed().map(i -> String.valueOf(i)).toList()); + threads += suffix; + } + return threads; + } + + static int[] getThreads(String threadlist) { + return Arrays.stream(threadlist.replace("+","").replace("-","").split(",")) + .mapToInt(s -> Integer.parseInt(s)).sorted().toArray(); + } + + String taskset(String threads) { + taskset = ""; + if (threads.endsWith("+") || threads.endsWith("-")) { + taskset = String.valueOf(threads.charAt(threads.length()-1)); + if (threads.contains(",")) { + if (taskset.equals("+")) + ReconUtil.taskset(0, 0); + else + ReconUtil.taskset(0, Arrays.stream(getThreads(threads)).min().getAsInt()); + } + else if (taskset.equals("-")) + ReconUtil.taskset(0, Integer.parseInt(String.valueOf(threads.charAt(0)))); + else if (taskset.equals("+")) + ReconUtil.taskset(0, 0); + } + return threads; + } + + /** + * Initialize ReconMutil. + * @param parser + */ + final void init(OptionParser parser) { + + taskset(parser.getOption("-t").stringValue()); + + parser.syncLogLevel(Logger.getLogger(ReconMutil.class.getPackage().getName())); + + maxEvents = parser.getOption("-n").intValue(); + skipEvents = parser.getOption("-s").intValue(); + serial = new SerialHoncho(fullSchema); + engines = new LinkedHashMap<>(); + + if (!parser.getOption("-y").isDefault()) { + yaml = new ClaraYaml(parser.getOption("-y").stringValue()); + for (JSONObject service : yaml.services()) { + JSONObject cfg = yaml.filter(service.getString("name")); + if (cfg.length() > 0) ReconUtil.addEngine(engines, service.getString("name"), service.getString("class"), cfg); + else ReconUtil.addEngine(engines, service.getString("name"), service.getString("class"), null); + } + } + else if (!parser.getOption("-c").isDefault()) { + for (String clazz : parser.getOption("-c").stringValue().split(",")) + ReconUtil.addEngine(engines, null, clazz, null); + } + else { + for (String line : ReconUtil.readResourceLines("org/jlab/clas/reco/services.txt")) + ReconUtil.addEngine(engines, line.split(" ")[0], line.split(" ")[1], null); + } + if (!parser.getOption("-B").isDefault()) { + ReconstructionEngine bg = ReconUtil.addEngine(engines, "BG", "org.jlab.service.bg.BackgroundEngine", null); + bg.engineConfigMap.put("filename",parser.getOption("-B").stringValue()); + } + if (!parser.getOption("-f").isDefault()) { + try { + fields = Arrays.stream(parser.getOption("-f").stringValue() + .split(",")).mapToDouble(s -> Double.parseDouble(s)).toArray(); + } + catch (Exception e) { + Logger.getLogger(ReconMutil.class.getName()).log(Level.SEVERE, () -> "invalid field option: -f "+parser.getOption("-f").stringValue()); + System.exit(22); + } + } + if (!parser.getOption("-b").isDefault()) + benchmark = new Benchmark("ReconMutil",BENCHMARK_NAMES); + } + + /** + * The command-line entry-point known as "recon-mutil". + * @param args command-line arguments + */ + public static void main(String[] args) { + OptionParser opt = new OptionParser("reco-mutil"){ + @Override + public void parse(String[] args) { + super.parse(args); + getOption("-t").setValue(sortThreads(getOption("-t").stringValue())); + } + }; + opt.addOption("-t","4","number of threads"); + opt.addOption("-s","0","number of events to skip"); + opt.addOption("-n","0","number of events to process"); + opt.addOption("-y","0","yaml file"); + opt.addOption("-u","true","update dictionary from writer"); + opt.addOption("-o",null,"output file name"); + opt.addOption("-S",null,"schema directory"); + opt.addOption("-B",null,"background files, comma-separated"); + opt.addOption("-c",null,"comma-separated engine list"); + opt.addOption("-f",null,"field scales for torus and solenoid, comma-separated (T,S)"); + opt.addOption("-b","false","enable benchmarking"); + opt.setRequiresInputList(true); + opt.parse(args); + + RecoMutil f = new RecoMutil(opt); + + if (!opt.getOption("-o").isDefault()) + f.openWriter(opt.getOption("-S"), opt.getOption("-o").stringValue()); + + f.launch(getThreads(opt.getOption("-t").stringValue()), + opt.getInputList().stream().toArray(String[]::new)); + } + +} diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java new file mode 100644 index 0000000000..0f4250ec01 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java @@ -0,0 +1,627 @@ +package org.jlab.clas.reco; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.TimerTask; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jlab.coda.jevio.EvioException; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.detector.decode.CLASDecoderPool; +import org.jlab.detector.serial.Occupancer; +import org.jlab.detector.serial.SerialHoncho; +import org.jlab.io.evio.EvioDataEvent; +import org.jlab.io.evio.EvioSource; +import org.jlab.io.hipo.HipoDataEvent; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.Event; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.jnp.hipo4.io.HipoReader; +import org.jlab.jnp.hipo4.io.HipoWriterSorted; +import org.jlab.utils.ClaraYaml; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.benchmark.BenchmarkTimer; +import org.jlab.utils.benchmark.ProgressPrintout; +import org.jlab.utils.options.OptionParser; +import org.jlab.utils.system.ClasUtilsFile; +import org.json.JSONObject; + +/** + * + * @author baltzell + */ +final class ReconMutil { + + static final boolean COLLECT_GARBAGE = false; + + static final String[] BENCHMARK_NAMES = new String[]{"evio","deco","serial","post","write"}; + + // Performance parameters: + final int BENCH_SECONDS = 30; + final int EVENTS_PER_CHUNK = 100; + + // Static parameters: + int maxEvents; + int skipEvents; + ClaraYaml yaml; + OptionParser parser; + double[] fields = null; + String taskset; + + // File I/O: + Object reader; + HipoWriterSorted writer; + List schemaBankList = new ArrayList<>(); + SchemaFactory fullSchema; + + // Processors: + SerialHoncho serial; + Occupancer occupancer = new Occupancer(); + CLASDecoderPool decoders = new CLASDecoderPool(64,"default",null); + Map engines = new LinkedHashMap<>(); + + // Threads: + CompletableFuture readerThread; + CompletableFuture writerThread; + ConcurrentLinkedQueue decoThreads = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue procThreads = new ConcurrentLinkedQueue<>(); + + // Queues: + ConcurrentLinkedQueue> decoQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> procQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> writeQueue = new ConcurrentLinkedQueue<>(); + + // Progress counters: + volatile int readEvents; + volatile int failEvents; + volatile int fileEvents; + volatile int maxFileEvents; + volatile int writeEvents; + volatile AtomicInteger taggedEvents = new AtomicInteger(); + volatile ProgressPrintout progress = new ProgressPrintout(); + volatile Benchmark benchmark = null; + TimerTask statsShow = new TimerTask() { @Override public void run() { show(); } }; + + // Control flags: + final Object serialLock = new Object(); + AtomicInteger serialTrigger = new AtomicInteger(0); + + ReconMutil(OptionParser parser) { + init(parser); + } + + /** + * The thread launcher and collector. + * @param threads number of threads + * @param output name of output file to write + * @param input names of input files to read + */ + void launch(int[] threads, String output, String... input) { + + reset(); + + System.out.println(String.format("recon-mutil:: spawning 2*%d+2 threads",threads[0])); + + // start a period status printout: + ReconUtil.runPeriodic(10, statsShow); + + // one reader thread: + readerThread = ReconUtil.launch(CompletableFuture.runAsync(() -> { reader(threads[0], input); })); + + // some decoder threads: + for (int i=0; i { decoder(j); })); + ReconUtil.sleep(100); + } + + // one serial thread: + ReconUtil.launch(CompletableFuture.runAsync(this::serial)); + + // one writer thread: + writerThread = ReconUtil.launch(CompletableFuture.runAsync(() -> { writer(output); })); + + // let decoding get warmed up: + ReconUtil.sleep(10000); + + // perform scaling test: + if (threads.length > 1) + ReconUtil.launch(CompletableFuture.runAsync(() -> { rethreader(threads); })).join(); + + // spawn engine threads: + else for (int i=0; i { processor(j); })); + ReconUtil.sleep(100); + } + + // wait for finish: + writerThread.join(); + } + + /** + * The reader thread. + * @param input input filenames + */ + void reader(int threads, String... input) { + + // convert input filenames to a list: + List inputs = new ArrayList<>(Arrays.asList(input)); + + // initialize the event chunk: + List output = new ArrayList<>(EVENTS_PER_CHUNK); + + // loop over input events: + while ( (maxEvents < 1 || writeEvents < maxEvents+taggedEvents.get()) && + (maxFileEvents < 1 || fileEvents < maxFileEvents) ) { + + if (reader != null) { + + // sleep instead of overfilling the queue (100K events, ~2GB): + if (procQueue.size()+decoQueue.size() > 1e5) ReconUtil.sleep(1000); + + // read next event into chunk, and fill queue if chunk full: + else output = read(output); + } + + // open the next input file: + else if (!inputs.isEmpty()) open(inputs.removeFirst()); + + // no more events to read: + else break; + } + + // write leftover, partial chunk: + if (!output.isEmpty()) { + readEvents += output.size(); + decoQueue.offer(output); + } + + if (reader instanceof EvioSource evio) evio.close(); + } + + /** + * The decoder thread. + * @param thread thread number + */ + void decoder(int thread) { + while (true) { + List input = decoQueue.poll(); + if (input == null) { + if (decoQueue.isEmpty() && readerThread.isDone() && decoQueue.isEmpty()) + break; + ReconUtil.sleep(100); + } + else { + List output = new ArrayList<>(input.size()); + for (int i=0; i 0) { + first = false; + // sleep to collect more triggers: + ReconUtil.sleep(1000); + // get current number of triggers: + int t = serialTrigger.get(); + // do the protected stuff: + synchronized (serialLock) { serial.updateHelicitySequence(); } + // remove the same number of triggers: + for (int i=0; i 0 && writeEvents > maxEvents+taggedEvents.get()) { + stopProcessing(); + break; + } + List input = procQueue.poll(); + if (input == null) { + if (procQueue.isEmpty() && decoThreads.isEmpty() && procQueue.isEmpty()) { + if (writeEvents+skipEvents+failEvents >= readEvents+taggedEvents.get()) + break; + } + ReconUtil.sleep(100); + } + else { + List output = new ArrayList<>(input.size()); + for (int i=0; i engine : engines.entrySet()) { + if (benchmark != null) benchmark.resume(thread, engine.getKey()); + try { engine.getValue().processDataEvent(input.get(i)); } + catch (Exception ex) { ex.printStackTrace(); } + if (benchmark != null) benchmark.pause(thread, engine.getKey()); + } + } + output.add(input.get(i).getHipoEvent()); + progress.updateStatus(); + } + writeQueue.offer(output); + } + } + } + + /** + * The writer thread. + * @param output output filename + */ + void writer(String output) { + if (output != null) writer = open(output, yaml); + while (true) { + List e = writeQueue.poll(); + if (e == null) { + if (decoThreads.isEmpty() && procThreads.isEmpty() && procQueue.isEmpty() && writeQueue.isEmpty()) + break; + ReconUtil.sleep(1000); + } + else { + for (int i=0; i 0 || schemaBankList.isEmpty()) + writer.addEvent(e.get(i), e.get(i).getEventTag()); + else + writer.addEvent(e.get(i).reduceEvent(schemaBankList), e.get(i).getEventTag()); + occupancer.process(e.get(i)); + if (benchmark != null) benchmark.pause("write"); + } + if (++writeEvents % 1000 == 0 && COLLECT_GARBAGE) System.gc(); + } + } + } + close(); + } + + /** + * The rethreader thread. + * + * @param threads thread counts to use + */ + void rethreader(int... threads) { + System.out.println("recon-mutil:: ~~~~~~~~~ rethreading launched ~~~~~~~~~"); + Map benches = new LinkedHashMap<>(); + Map progs = new LinkedHashMap<>(); + while (!ReconUtil.isDone(decoThreads)) ReconUtil.sleep(100); + System.out.println("recon-mutil:: ~~~~~~~~~ rethreading primed ~~~~~~~~~"); + for (int thread : threads) { + if (taskset.equals("-")) ReconUtil.taskset(0, thread); + BenchmarkTimer.WARMUP_CALLS = 10*thread+90; + benchmark = new Benchmark(thread+" Threads Scaling ",BENCHMARK_NAMES); + progress = new ProgressPrintout(10*thread); + for (int j=0; j { processor(k); })); + } + while (!progress.warmedUp() && !procQueue.isEmpty()) ReconUtil.sleep(100); + ReconUtil.sleep(BENCH_SECONDS *1000); + System.out.printf("\nrecon-mutil:: ~~~~~~~~~ rethreading timed %d ~~~~~~~~~\n%n",thread); + System.out.println(progress.getUpdateString()); + System.out.println(benchmark); + benches.put(thread, benchmark); + progs.put(thread, progress); + procThreads.forEach(p -> p.cancel(true)); + } + String fcsv = String.format("scaling-mutil%s.txt",taskset); + ReconUtil.writeFile(fcsv, ReconUtil.toCSV(progs, benches)); + ReconUtil.gnuplotScaling(fcsv, String.format("scaling-mutil%s.svg",taskset)); + stopProcessing(); + } + + /** + * Decode an event. + * @param bytes the EVIO byte buffer + * @return decoded event + */ + HipoDataEvent decode(int thread, ByteBuffer bytes) { + if (benchmark != null) benchmark.resume(thread, "evio"); + EvioDataEvent evio = new EvioDataEvent(bytes.array(), ByteOrder.LITTLE_ENDIAN); + if (benchmark != null) { + benchmark.pause(thread, "evio"); + benchmark.resume(thread, "deco"); + } + CLASDecoder d = decoders.poll(); + HipoDataEvent hipo = fields == null ? + d.getDecodedDataEvent(evio) : + d.getDecodedDataEvent(evio, fields[0], fields[1]); + decoders.offer(d); + if (benchmark != null) benchmark.pause(thread, "deco"); + return hipo; + } + + /** + * Open a new input HIPO/EVIO event file. + * @param filename + */ + void open(String filename) { + if (reader != null && reader instanceof EvioSource) + ((EvioSource)reader).close(); + fileEvents = 0; + if (filename.endsWith(".hipo")) { + reader = new HipoReader(); + ((HipoReader)reader).open(filename); + maxFileEvents = ((HipoReader)reader).getEventCount(); + } + else { + reader = new EvioSource(); + ((EvioSource)reader).open(filename); + maxFileEvents = ((EvioSource)reader).getEventCount(); + } + } + + /** + * Open a new writer and initialize its schema. + * @param filename output filename + * @param yaml the configuration + */ + HipoWriterSorted open(String filename, ClaraYaml yaml) { + HipoWriterSorted writer = new HipoWriterSorted(); + writer.setCompressionType(2); + SchemaFactory s = ReconUtil.getSchemaFactory(parser.getOption("-S"), yaml); + writer.getSchemaFactory().copy(s); + schemaBankList = ReconUtil.getBankList(s, yaml); + writer.open(filename); + return writer; + } + + /** + * Read the next event into the chunk, and, if it's full, queue the chunk + * and make a new one. + * @param chunk + * @return modified chunk + */ + List read(List chunk) { + if (benchmark != null) benchmark.resume("read"); + Object o = null; + if (reader instanceof EvioSource evio) { + try { o = evio.getEventBuffer(++fileEvents, true); } + catch (EvioException ex) { + failEvents++; + ex.printStackTrace(); + } + } + else { + Event event = new Event(); + o = ((HipoReader)reader).getEvent(event, fileEvents++); + } + if (o != null && (skipEvents < 1 || readEvents > skipEvents)) { + chunk.add(o); + if (chunk.size() >= EVENTS_PER_CHUNK) { + decoQueue.offer(chunk); + readEvents += chunk.size(); + chunk = new ArrayList<>(EVENTS_PER_CHUNK); + } + } + if (benchmark != null) benchmark.pause("read"); + return chunk; + } + + /** + * Close the output file. + */ + void close() { + serial.closure(writer); + writer.close(); + if (parser.getOption("-t").stringValue().split(",").length == 1) + if (benchmark != null) System.out.println(benchmark); + System.out.println(String.format("recon-mutil :: read/write/tagged/diff = %d/%d/%d/%d", + readEvents, writeEvents, taggedEvents.get(), writeEvents-readEvents-taggedEvents.get())); + } + + /** + * Forcefully shutdown all threads, close files, and reset queues and counters. + */ + void reset() { + for (CompletableFuture cf : decoThreads) cf.cancel(true); + for (CompletableFuture cf : procThreads) cf.cancel(true); + if (readerThread != null) readerThread.cancel(true); + if (writerThread != null) { + writerThread.cancel(true); + close(); + } + decoQueue = new ConcurrentLinkedQueue<>(); + writeQueue = new ConcurrentLinkedQueue<>(); + procThreads = new ConcurrentLinkedQueue(); + readEvents = 0; + writeEvents = 0; + failEvents = 0; + taggedEvents.set(0); + } + + /** + * Stop processing, somewhat cleanly. + */ + void stopProcessing() { + if (readerThread != null) readerThread.cancel(true); + for (CompletableFuture cf : decoThreads) cf.cancel(true); + for (CompletableFuture cf : procThreads) cf.cancel(true); + decoQueue.clear(); + procQueue.clear(); + writeQueue.clear(); + } + + /** + * Sort and preserve +/- suffix. + * @param threads + * @return + */ + static String sortThreads(String threads) { + if (threads.contains(",")) { + String suffix = ""; + if (threads.endsWith("+") || threads.endsWith("-")) + suffix = String.valueOf(threads.charAt(threads.length()-1)); + threads = String.join(",",Arrays.stream(threads.replace("+","").replace("-","").split(",")) + .mapToInt(s -> Integer.parseInt(s)).sorted().boxed().map(i -> String.valueOf(i)).toList()); + threads += suffix; + } + return threads; + } + + String taskset(String threads) { + taskset = ""; + if (threads.endsWith("+") || threads.endsWith("-")) { + taskset = String.valueOf(threads.charAt(threads.length()-1)); + if (threads.contains(",")) { + if (taskset.equals("+")) + ReconUtil.taskset(0, 0); + else + ReconUtil.taskset(0, Arrays.stream(getThreads(threads)).min().getAsInt()); + } + else if (taskset.equals("-")) + ReconUtil.taskset(0, Integer.parseInt(String.valueOf(threads.charAt(0)))); + else if (taskset.equals("+")) + ReconUtil.taskset(0, 0); + } + return threads; + } + + static int[] getThreads(String threadlist) { + return Arrays.stream(threadlist.replace("+","").replace("-","").split(",")) + .mapToInt(s -> Integer.parseInt(s)).sorted().toArray(); + } + + /** + * Initialize ReconMutil. + * @param parser + */ + void init(OptionParser parser) { + + parser.getOption("-t").setValue(sortThreads(parser.getOption("-t").stringValue())); + + taskset(parser.getOption("-t").stringValue()); + + fullSchema = new SchemaFactory(); + fullSchema.initFromDirectory(ClasUtilsFile.getResourceDir("CLAS12DIR","etc/bankdefs/hipo4")); + this.parser = parser; + parser.syncLogLevel(Logger.getLogger(ReconMutil.class.getPackage().getName())); + maxEvents = parser.getOption("-n").intValue(); + skipEvents = parser.getOption("-s").intValue(); + serial = new SerialHoncho(fullSchema); + engines = new LinkedHashMap<>(); + if (!parser.getOption("-y").isDefault()) { + yaml = new ClaraYaml(parser.getOption("-y").stringValue()); + for (JSONObject service : yaml.services()) { + JSONObject cfg = yaml.filter(service.getString("name")); + if (cfg.length() > 0) ReconUtil.addEngine(engines, service.getString("name"), service.getString("class"), cfg); + else ReconUtil.addEngine(engines, service.getString("name"), service.getString("class"), null); + } + } + else if (!parser.getOption("-c").isDefault()) { + for (String clazz : parser.getOption("-c").stringValue().split(",")) + ReconUtil.addEngine(engines, null, clazz, null); + } + else { + for (String line : ReconUtil.readResourceLines("org/jlab/clas/reco/services.txt")) + ReconUtil.addEngine(engines, line.split(" ")[0], line.split(" ")[1], null); + } + if (!parser.getOption("-B").isDefault()) { + ReconstructionEngine bg = ReconUtil.addEngine(engines, "BG", "org.jlab.service.bg.BackgroundEngine", null); + bg.engineConfigMap.put("filename",parser.getOption("-B").stringValue()); + } + if (!parser.getOption("-f").isDefault()) { + try { + fields = Arrays.stream(parser.getOption("-f").stringValue() + .split(",")).mapToDouble(s -> Double.parseDouble(s)).toArray(); + } + catch (Exception e) { + Logger.getLogger(ReconMutil.class.getName()).log(Level.SEVERE, () -> "invalid field option: -f "+parser.getOption("-f").stringValue()); + System.exit(22); + } + } + if (!parser.getOption("-b").isDefault()) + benchmark = new Benchmark("ReconMutil",BENCHMARK_NAMES); + } + + /** + * Print the thread, queue, and event states. + */ + void show() { + String s1 = String.format("threads(r/d/p/w)=(%b/%b:%d/%b:%d/%b)", + readerThread != null ? !readerThread.isDone() : false, + !ReconUtil.isDone(decoThreads), + decoThreads.size(), + !ReconUtil.isDone(procThreads), + procThreads.size(), + writerThread != null ? !writerThread.isDone() : false); + String s2 = String.format(" queues(d/p/w)=(%d/%d/%d)", + decoQueue.size()*EVENTS_PER_CHUNK, + procQueue.size()*EVENTS_PER_CHUNK, + writeQueue.size()*EVENTS_PER_CHUNK); + String s3 = String.format(" events(r/w/t/f)=(%d/%d/%d/%d)", + readEvents, writeEvents, taggedEvents.get(), failEvents); + Logger.getLogger(ReconMutil.class.getName()).log(Level.CONFIG, () -> s1+" "+s2+" "+s3); + } + + /** + * The command-line entry-point known as "recon-mutil". + * @param args command-line arguments + */ + public static void main(String[] args) { + OptionParser o = ReconUtil.getParser(); + o.removeOption("-i"); + o.removeOption("-o"); + o.removeOption("-c"); + o.removeOption("-P"); + o.addOption("-t","4","number of threads, suffixed by +/- to taskset node/cpus)"); + o.addOption("-o", null, "output file name"); + o.addOption("-c","2","comma-separated engine list"); + o.addOption("-f",null,"field scales for torus and solenoid, comma-separated (T,S)"); + o.addOption("-b","false","enable benchmarking"); + o.setRequiresInputList(true); + o.parse(args); + ReconMutil r = new ReconMutil(o); + r.launch(getThreads(o.getOption("-t").stringValue()), + o.getOption("-o").stringValue(), + o.getInputList().stream().toArray(String[]::new)); + } + +} diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconUtil.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconUtil.java new file mode 100644 index 0000000000..2aa9497b90 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconUtil.java @@ -0,0 +1,330 @@ +package org.jlab.clas.reco; + +import java.io.BufferedReader; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jlab.clara.engine.EngineData; +import org.jlab.clara.engine.EngineDataType; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.utils.ClaraYaml; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.benchmark.ProgressPrintout; +import org.jlab.utils.options.OptionParser; +import org.jlab.utils.options.OptionValue; +import org.jlab.utils.system.ClasUtilsFile; +import org.json.JSONObject; + +/** + * Static utility methods. + * + */ +public class ReconUtil { + + static OptionParser getParser() { + OptionParser parser = new OptionParser("recon-util"); + parser.addRequired("-o","output.hipo"); + parser.addRequired("-i","input.evio/hipo"); + parser.addOption("-c","0","use default configuration [0 - no, 1 - yes/default, 2 - all services] "); + parser.addOption("-s","0","number of events to skip"); + parser.addOption("-n","0","number of events to process"); + parser.addOption("-y","0","yaml file"); + parser.addOption("-u","true","update dictionary from writer ? "); + parser.addOption("-S",null,"schema directory"); + parser.addOption("-B",null,"background file"); + parser.addOption("-P",null,"preload file for post-processing"); + parser.addOption("-R","0","rebuild scalers"); + parser.addOption("-H","0","restream helicity"); + parser.setRequiresInputList(false); + return parser; + } + + /** + * Get a new schema factory, potentially filtered based on yaml. + * @param schema + * @param yaml + * @return + */ + static SchemaFactory getSchemaFactory(OptionValue schema, ClaraYaml yaml) { + SchemaFactory ret = new SchemaFactory(); + SchemaFactory stock = new SchemaFactory(); + stock.initFromDirectory(getSchemaDirectory(schema, yaml)); + if (yaml == null) { + ret.copy(stock); + } else { + JSONObject json = yaml.filter("writer"); + if (json.has("wildcard")) { + ret.copy(stock.reduce(json.getString("wildcard"))); + } else { + ret.copy(stock); + } + } + return ret; + } + + /** + * Choose schema directory, prioritizing command-line, yaml, and then default. + * @param parser + * @param yaml + * @return the chosen schema directory + */ + static String getSchemaDirectory(OptionValue opt, ClaraYaml yaml) { + String d = ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4"); + if (yaml != null && yaml.getSchemaDirectory() != null) + d = yaml.getSchemaDirectory(); + if (opt.stringValue() != null) + d = opt.stringValue(); + return d; + } + + /** + * + * @param schema + * @param yaml + * @return + */ + static List getBankList(SchemaFactory schema, ClaraYaml yaml) { + List banks = new ArrayList<>(); + if (yaml != null) { + JSONObject json = yaml.filter("writer"); + if (json.has("wildcard")) { + if (json.optBoolean("schema_filter", true)) { + int schemaSize = schema.getSchemaList().size(); + for (int i = 0; i < schemaSize; i++) { + banks.add(new Bank(schema.getSchemaList().get(i))); + } + } + } + } + return banks; + } + + /** + * Add a new engine to the list. + * @param label display name + * @param clazz full class name + * @param cfg engine configuration + * @return + */ + static ReconstructionEngine addEngine(Map engines, String label, String clazz, JSONObject cfg) { + ReconstructionEngine engine = null; + try { + Class c = Class.forName(clazz); + if (ReconstructionEngine.class.isAssignableFrom(c) == true) { + engine = (ReconstructionEngine) c.newInstance(); + if (cfg != null && !cfg.toString().equals("null")) { + EngineData input = new EngineData(); + input.setData(EngineDataType.JSON.mimeType(), cfg.toString()); + engine.configure(input); + } else { + engine.init(); + } + engines.put(label == null ? engine.getName() : label, engine); + } else { + Logger.getLogger(ReconMutil.class.getPackage().getName()).log(clazz.contains("DecoderEngine") ? Level.INFO : Level.SEVERE, "Class is not a reconstruction engine : {0}", clazz); + } + } catch (ClassNotFoundException | InstantiationException | IllegalAccessException ex) { + Logger.getLogger(ReconMutil.class.getPackage().getName()).log(Level.SEVERE, null, ex); + } + return engine; + } + + /** + * Catch interruptions in sleep. + * @param milliseconds + */ + static void sleep(int milliseconds) { + try { Thread.sleep(milliseconds); } + catch (InterruptedException ex) {} + } + + /** + * Just get the contents of a text file resource. + * @param resource + * @return + */ + static List readResourceLines(String resource) { + List lines = new ArrayList<>(); + InputStream is = ReconMutil.class.getClassLoader().getResourceAsStream(resource); + BufferedReader br = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8)); + try { + for (String line; (line = br.readLine()) != null;) lines.add(line); + } catch (IOException ex) { + System.getLogger(ReconMutil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); + } + return lines; + } + + /** + * Write/overwrite a new file. + * @param filename + * @param content file contents + */ + static void writeFile(String filename, String content) { + Path p = Path.of(filename); + try { + Files.writeString(p, content); + } catch (IOException ex) { + System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); + } + + } + + /** + * Check whether any of the given futures are still running. + * @param queue + * @return whether any are still running + */ + static boolean isDone(ConcurrentLinkedQueue queue) { + return queue.stream().filter(f -> !f.isDone()).toList().isEmpty(); + } + + /** + * Print exception when complete. + * @param future + * @return + */ + static CompletableFuture launch(CompletableFuture future) { + return future.whenCompleteAsync((result,exception) -> { + if (exception != null && !(exception instanceof CancellationException)) + ((Exception)exception).printStackTrace(); + }); + } + + /** + * Add a thread to the queue and make it remove itself when done. + * @param queue + * @param future + */ + static void addAndRemove(ConcurrentLinkedQueue queue, CompletableFuture future) { + launch(future).whenCompleteAsync((result,exception) -> { queue.remove(future); }); + queue.offer(future); + } + + /** + * Periodically run the print the thread, queue, and event states. + * @param seconds + * @param method + * @return + */ + static Timer runPeriodic(double seconds, TimerTask task) { + Timer t = new Timer("timer", true); + t.scheduleAtFixedRate(task, 0, (int)(1000*seconds)); + return t; + } + + /** + * Generate a CSV table for thread-mapped benchmarks. + * @param benches + * @return + */ + static String toCSV(Map progs, Map benches) { + List csv = new ArrayList<>(); + for (Benchmark b : benches.values()) b.sortByName(); + String head = (new ArrayList<>(benches.values())).get(0).toCSV()[0]; + csv.add("threads,rate," + head); + for (int threads : benches.keySet()) + csv.add(String.format("%d,%.3f,%s", threads, + progs.get(threads).getHz(), benches.get(threads).toCSV()[1])); + return String.join("\n",csv); + } + + /** + * Run gnpulot coatjava/libexec/scaling.gpl + * @csvfilename input CSV filename + * @svgfilename output SVG filename + * @return exit code + */ + static int gnuplotScaling(String csvfilename, String svgfilename) { + ProcessBuilder pb = new ProcessBuilder("gnuplot","-c", + ClasUtilsFile.getResourceDir("CLAS12DIR", "libexec/scaling.gpl"), + "-m",csvfilename); + pb.redirectError(ProcessBuilder.Redirect.INHERIT); + pb.redirectOutput(new File(svgfilename)); + try { + Process proc = pb.start(); + proc.waitFor(); + return proc.exitValue(); + } + catch (IOException ex) { System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); } + catch (InterruptedException ex) {} + return 42; + } + + /** + * Get the cpus from a node from the "numactl" command. + * @param numaNode numa node index + * @param cpuCount maximum number of cpus to get + * @return taskset-formatted cpu string + */ + static String getTasksetCpus(int numaNode, int cpuCount) { + ProcessBuilder pb = new ProcessBuilder("numactl","-H"); + pb.redirectError(ProcessBuilder.Redirect.INHERIT); + Process proc; + try { + proc = pb.start(); + proc.waitFor(); + for (String line : (new String(proc.getInputStream().readAllBytes(), StandardCharsets.UTF_8)).split("\n")) { + if (line.startsWith("node "+numaNode+" cpus:")) { + List cols = Arrays.asList(line.split(" ")); + List cpus = cols.subList(3, cols.size()); + if (cpuCount > 0 && cpuCount < cpus.size()) + cpus = cpus.subList(0, cpuCount); + return String.join(",",cpus); + } + } + } + catch (InterruptedException ex) {} + catch (IOException ex) { System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); } + return null; + } + + /** + * Run the system "taskset" command on this JVM and all its threads. + * @param cpus cpu list for taskset + * @return taskset exit code + */ + static int taskset(String cpus) { + long pid = ProcessHandle.current().pid(); + String[] cmd = new String[]{"taskset","-a","-p","-c",cpus,String.valueOf(pid)}; + System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.INFO, () -> "running \""+String.join(" ",cmd)+"\""); + ProcessBuilder pb = new ProcessBuilder(cmd); + try { + Process proc = pb.start(); + proc.waitFor(); + System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.INFO, () -> "taskset exit code: "+proc.exitValue()); + return proc.exitValue(); + } + catch (IOException ex) { System.getLogger(ReconUtil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); } + catch (InterruptedException ex) {} + return 42; + } + + /** + * Run the system "taskset" command on this JVM, using "numactl" to pick cpus. + * @param numaNode numa node index + * @param cpuCount maximum number of cpus to get + * @return taskset exit code + */ + static int taskset(int numaNode, int cpuCount) { + String cpus = getTasksetCpus(numaNode, cpuCount); + return cpus != null ? taskset(cpus) : 42; + } + +} diff --git a/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt b/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt new file mode 100644 index 0000000000..8a81bedfb6 --- /dev/null +++ b/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt @@ -0,0 +1,31 @@ +MAGFIELDS org.jlab.clas.swimtools.MagFieldsEngine +DENOISE org.jlab.service.ai.DCDenoiseEngine +FTCAL org.jlab.rec.ft.cal.FTCALEngine +FTHODO org.jlab.rec.ft.hodo.FTHODOEngine +FTTRK org.jlab.rec.ft.trk.FTTRKEngine +FTEB org.jlab.rec.ft.FTEBEngine +URWT org.jlab.service.urwt.URWTEngine +DCCR org.jlab.service.dc.DCHBClustering +DCHB org.jlab.service.dc.DCHBPostClusterConv +FTOFHB org.jlab.service.ftof.FTOFHBEngine +EC org.jlab.service.ec.ECEngine +RASTER org.jlab.service.raster.RasterEngine +CVT org.jlab.rec.cvt.services.CVTEngine +CTOF org.jlab.service.ctof.CTOFEngine +CND org.jlab.service.cnd.CNDCalibrationEngine +BAND org.jlab.service.band.BANDEngine +HTCC org.jlab.service.htcc.HTCCReconstructionService +LTCC org.jlab.service.ltcc.LTCCEngine +EBHB org.jlab.service.eb.EBHBEngine +DCTB org.jlab.service.dc.DCTBEngine +FMT org.jlab.service.fmt.FMTEngine +FTOFTB org.jlab.service.ftof.FTOFTBEngine +CVT org.jlab.rec.cvt.services.CVTSecondPassEngine +EBTB org.jlab.service.eb.EBTBEngine +RICHEB org.jlab.rec.rich.RICHEBEngine +RTPC org.jlab.service.rtpc.RTPCEngine +AHDC org.jlab.service.ahdc.AHDCEngine +ATOF org.jlab.service.atof.ATOFEngine +ALERT org.jlab.service.alert.ALERTEngine +MC org.jlab.service.mc.TruthMatch +VTX org.jlab.rec.service.vtx.VTXEngine diff --git a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/Benchmark.java b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/Benchmark.java index 48495aee4b..8e38457611 100644 --- a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/Benchmark.java +++ b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/Benchmark.java @@ -28,7 +28,7 @@ public Benchmark(String name) { this.name = name; } - public Benchmark(String name, String[] specials) { + public Benchmark(String name, String... specials) { this.name = name; this.specials.addAll(Arrays.asList(specials)); } diff --git a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimer.java b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimer.java index aa07bb2841..996196e40e 100644 --- a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimer.java +++ b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimer.java @@ -1,6 +1,5 @@ package org.jlab.utils.benchmark; -import java.util.ArrayList; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -11,7 +10,7 @@ */ public class BenchmarkTimer { - private static final int WARMUP_CALLS = 100; + public static int WARMUP_CALLS = 100; public static class BenchmarkMultiTimer extends BenchmarkTimer { ConcurrentHashMap timeAtResume = new ConcurrentHashMap<>(); diff --git a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/ProgressPrintout.java b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/ProgressPrintout.java index 2964159283..f5f91928a6 100644 --- a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/ProgressPrintout.java +++ b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/ProgressPrintout.java @@ -11,7 +11,7 @@ */ public class ProgressPrintout { - private final int WARMUP_CALLS = 100; + public int WARMUP_CALLS = 100; private TreeMap items = new TreeMap<>(); private Long previousPrintoutTime = (long) 0; @@ -19,12 +19,21 @@ public class ProgressPrintout { private double printoutIntervalSeconds = 10.0; private String printoutLeadingString = ">>>>> progress : "; private Integer numberOfCalls = 0; - + private int warmupCalls = 100; + + public boolean warmedUp() { return numberOfCalls > warmupCalls; } + public ProgressPrintout(){ this.previousPrintoutTime = System.currentTimeMillis(); this.startPrintoutTime = System.currentTimeMillis(); } + public ProgressPrintout(int warmupCalls){ + this.warmupCalls = warmupCalls; + this.previousPrintoutTime = System.currentTimeMillis(); + this.startPrintoutTime = System.currentTimeMillis(); + } + public ProgressPrintout(String name){ this.printoutLeadingString = name; this.previousPrintoutTime = System.currentTimeMillis(); @@ -58,8 +67,8 @@ public void showStatus(){ System.out.println("\n\n"); } - public void updateStatus(){ - if (++this.numberOfCalls < WARMUP_CALLS){ + public synchronized void updateStatus(){ + if (++this.numberOfCalls < warmupCalls){ this.previousPrintoutTime = System.currentTimeMillis(); this.startPrintoutTime = this.previousPrintoutTime; } diff --git a/docs/zensical/docs/index.md b/docs/zensical/docs/index.md index 56daa84698..4cc6977f31 100644 --- a/docs/zensical/docs/index.md +++ b/docs/zensical/docs/index.md @@ -13,7 +13,11 @@ ## Reports - [Jacoco Coverage](jacoco/index.html) -- [Scaling Test](perf/scaling.svg) -- [Cpu Profiling](perf/asprof_cpu.html) -- [Alloc Profiling](perf/asprof_alloc.html) -- [Lock Profiling](perf/asprof_lock.html) +- [Scaling Test - CLARA](perf/scaling.svg) +- [Cpu Profiling - CLARA](perf/asprof_cpu_clara.html) +- [Alloc Profiling - CLARA](perf/asprof_alloc_clara.html) +- [Lock Profiling - CLARA](perf/asprof_lock_clara.html) +- [Scaling Test - MUTIL](perf/scaling_mutil.svg) +- [Cpu Profiling - MUTIL](perf/asprof_cpu_mutil.html) +- [Alloc Profiling - MUTIL](perf/asprof_alloc_mutil.html) +- [Lock Profiling - MUTIL](perf/asprof_lock_mutil.html) diff --git a/etc/bankdefs/hipo4/occupancy.json b/etc/bankdefs/hipo4/occupancy.json new file mode 100644 index 0000000000..ec02d1efdb --- /dev/null +++ b/etc/bankdefs/hipo4/occupancy.json @@ -0,0 +1,272 @@ +[ + { + "name": "OCC::DC::tdc", + "group": 500, + "item": 10, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::DC::tot", + "group": 500, + "item": 11, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::ECAL::adc", + "group": 500, + "item": 12, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::ECAL::tdc", + "group": 500, + "item": 13, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::FTOF::adc", + "group": 500, + "item": 14, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::FTOF::tdc", + "group": 500, + "item": 15, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::CTOF::adc", + "group": 500, + "item": 16, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::CTOF::tdc", + "group": 500, + "item": 17, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::HTCC::adc", + "group": 500, + "item": 18, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::HTCC::tdc", + "group": 500, + "item": 19, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::LTCC::adc", + "group": 500, + "item": 20, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::LTCC::tdc", + "group": 500, + "item": 21, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::BST::adc", + "group": 500, + "item": 22, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::BMT::adc", + "group": 500, + "item": 23, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::FTCAL::adc", + "group": 500, + "item": 24, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::FTHODO::adc", + "group": 500, + "item": 25, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::FTTRK::adc", + "group": 500, + "item": 26, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::RICH::tdc", + "group": 500, + "item": 27, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::BAND::adc", + "group": 500, + "item": 28, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::BAND::tdc", + "group": 500, + "item": 29, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::CND::adc", + "group": 500, + "item": 30, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + }, + { + "name": "OCC::CND::tdc", + "group": 500, + "item": 31, + "info": "Occupancy Bank", + "entries": [ + { "name":"sector" , "type":"B", "info":"sector"}, + { "name":"layer" , "type":"B", "info":"layer"}, + { "name":"component" , "type":"S", "info":"component"}, + { "name":"order" , "type":"B", "info":"order"}, + { "name":"hits" , "type":"F", "info":"number of hits"} + ] + } +] diff --git a/etc/bankdefs/util/bankSplit.py b/etc/bankdefs/util/bankSplit.py index a8449ecf42..3c21c0dc00 100755 --- a/etc/bankdefs/util/bankSplit.py +++ b/etc/bankdefs/util/bankSplit.py @@ -1,8 +1,9 @@ #!/usr/bin/env python -import json -import sys import os +import sys +import json +import glob # print usage if len(sys.argv)<2: @@ -54,6 +55,9 @@ def create(dirname, banklist): print("Single json files saved in " + workdirectory + singledirectory) +# occupancy banks: +occ = [os.path.basename(x)[:-5] for x in glob.glob('./singles/occupancy/*.json')] + # these should *always* be kept: mc = ["MC::Event", "MC::GenMatch", "MC::Header", "MC::Lund", "MC::Particle", "MC::RecMatch", "MC::True"] tag1 = ["RUN::config", "RAW::epics", "RAW::scaler", "RUN::scaler", "HEL::scaler", "COAT::config", "HEL::flip", "HEL::online", "HEL::decoder"] @@ -83,7 +87,7 @@ def create(dirname, banklist): trig = ["RAW::vtp","HTCC::rec","ECAL::adc","ECAL::calib","ECAL::clusters","ECAL::hits","ECAL::moments","ECAL::peaks","ECAL::tdc","ECAL::trigger"] # accumulate all the DST banks: -dst = rectbai + rectb + mc + tag1 + dets +dst = rectbai + rectb + mc + tag1 + occ + dets dsthb = dst + rechbai + rechb # generate the calib and mon schema: @@ -112,8 +116,6 @@ def create(dirname, banklist): level3 = list(dst) level3.extend(["DC::tdc", "DC::tot", "ECAL::adc", "ECAL::clusters", "FTOF::tdc", "FTOF::adc", "HitBasedTrkg::HBClusters", "HitBasedTrkg::HBTracks", "HTCC::adc", "RF::adc", "RF::tdc", "RUN::rf", "TimeBasedTrkg::TBClusters", "TimeBasedTrkg::TBTracks"]) -import glob - raw = glob.glob('./singles/data/*.json') raw.extend(glob.glob('./singles/header/*.json')) raw.extend(glob.glob('./singles/tag1/*.json')) diff --git a/etc/services/rgd-clarode.yml b/etc/services/rgd-clarode.yml index 1f5e638fb1..a99a9c3dc2 100644 --- a/etc/services/rgd-clarode.yml +++ b/etc/services/rgd-clarode.yml @@ -1,5 +1,3 @@ -mime-types: - - binary/data-hipo configuration: global: dcrbJitter: "true" @@ -14,24 +12,10 @@ configuration: MAGFIELDS: magfieldSolenoidMap: Symm_solenoid_r601_phi1_z1201_13June2018.dat magfieldTorusMap: Full_torus_r251_phi181_z251_25Jan2021.dat - DCCR: - rawBankGroup: "NOISE1" - MLTD: - run: "18305" - DCHAI: - outputBankPrefix: "HB" -io-services: - reader: - class: org.jlab.io.clara.Clas12Reader - name: Clas12Reader - writer: - class: org.jlab.io.clara.Clas12Writer - name: Clas12Writer + services: - class: org.jlab.clas.reco.DecoderEngine name: DECO - - class: org.jlab.service.ai.DCDenoiseEngine - name: DCDN - class: org.jlab.clas.swimtools.MagFieldsEngine name: MAGFIELDS - class: org.jlab.rec.ft.cal.FTCALEngine @@ -44,12 +28,18 @@ services: name: FTEB - class: org.jlab.service.raster.RasterEngine name: RASTER +# DC denoising + - class: org.jlab.service.ai.DCDenoiseEngine + name: DCDN +# DC clustering: nominal algoritm, filled the cluster bank for ai and conventional tracking - class: org.jlab.service.dc.DCHBClustering name: DCCR - - class: org.jlab.service.mltn.MLTDEngine - name: MLTD - - class: org.jlab.service.dc.DCHBPostClusterAI - name: DCHAI +# This is where the AI plugin would go. THe AI code reads the HB clusters. + - class: org.jlab.service.ai.DCClsComboEngine + name: DCCC +# DC using clustering on AI hits: NN assisted HB tracking + - class: org.jlab.service.dc.DCHBTrackingAI + name: DCHTAI - class: org.jlab.service.ftof.FTOFHBEngine name: FTOFHB - class: org.jlab.service.ec.ECEngine @@ -86,3 +76,13 @@ services: name: VTX - class: org.jlab.calibration.service.CalibBanksEngine name: CALIB +io-services: + reader: + class: org.jlab.io.clara.Clas12Reader + name: Clas12Reader + writer: + class: org.jlab.io.clara.Clas12Writer + name: Clas12Writer + +mime-types: + - binary/data-hipo diff --git a/etc/services/rgl-clarode.yml b/etc/services/rgl-clarode.yml deleted file mode 100644 index 9c92b0a1d2..0000000000 --- a/etc/services/rgl-clarode.yml +++ /dev/null @@ -1,66 +0,0 @@ -io-services: - reader: - class: org.jlab.io.clara.Clas12Reader - name: Clas12Reader - writer: - class: org.jlab.io.clara.Clas12Writer - name: Clas12Writer -services: - - class: org.jlab.clas.reco.DecoderEngine - name: DECO - - class: org.jlab.clas.swimtools.MagFieldsEngine - name: MAGFIELDS - - class: org.jlab.rec.ft.cal.FTCALEngine - name: FTCAL - - class: org.jlab.rec.ft.hodo.FTHODOEngine - name: FTHODO - - class: org.jlab.rec.ft.FTEBEngine - name: FTEB - - class: org.jlab.service.ai.DCDenoiseEngine - name: DCDN - - class: org.jlab.service.dc.DCHBClustering - name: DCCR - - class: org.jlab.service.dc.DCHBPostClusterConv - name: DCHB - - class: org.jlab.service.ftof.FTOFHBEngine - name: FTOFHB - - class: org.jlab.service.ec.ECEngine - name: EC - - class: org.jlab.service.ctof.CTOFEngine - name: CTOF - - class: org.jlab.service.cnd.CNDCalibrationEngine - name: CND - - class: org.jlab.service.band.BANDEngine - name: BAND - - class: org.jlab.service.htcc.HTCCReconstructionService - name: HTCC - - class: org.jlab.service.ltcc.LTCCEngine - name: LTCC - - class: org.jlab.service.eb.EBHBEngine - name: EBHB - - class: org.jlab.service.dc.DCTBEngine - name: DCTB - - class: org.jlab.service.ftof.FTOFTBEngine - name: FTOFTB - - class: org.jlab.service.eb.EBTBEngine - name: EBTB - - class: org.jlab.rec.rich.RICHEBEngine - name: RICH - - class: org.jlab.service.atof.ATOFEngine - name: ATOF - - class: org.jlab.service.ahdc.AHDCEngine - name: AHDC -configuration: - global: - variation: rgl_spring2025 - io-services: - writer: - schema_dir: dst - services: - MAGFIELDS: - magfieldSolenoidMap: Symm_solenoid_r601_phi1_z1201_13June2018.dat - magfieldTorusMap: Full_torus_r251_phi181_z251_25Jan2021.dat - AHDC: - Mode: "AI_Track_Finding" -mime-types: - - binary/data-hipo diff --git a/libexec/env.sh b/libexec/env.sh index 543a1f81bd..ae387d19ad 100644 --- a/libexec/env.sh +++ b/libexec/env.sh @@ -52,6 +52,31 @@ if [ $# -ge 1 ]; then fi fi +# get the number of threads from the command line options: +function get_threads() { + # look for a "-t" threading option: + for ((i=0; i<${#class_options[@]}; i++)); do + if [[ "${class_options[$i]}" == "-t" ]]; then + # found it; if it contains commas, split and get the max: + # FIXME: breaks for threads suffixed by +/- for numa tasksetting + let i=i+1 + threads=$(echo ${class_options[$i]} | awk -F, 'NR==1 {max=$1} { for(i=1; i<=NF; i++) { if($i > max) max=$i } } END {print max}') + break + fi + done + [ -z ${threads+x} ] && echo 0 || echo $threads +} + +# get a cpu list for taskset, of all cpus on a given NUMA node: +function get_all_numa_cpus() { + echo $(numactl -H | grep "^node $1 cpus:" | awk '{for(i=4;i<=NF;++i)print$i}') | sed 's/ /,/g' +} + +# get a cpu list for tasket, of the first N cpus on a given NUMA node: +function get_numa_cpus() { + echo $(numactl -H | grep "^node $1 cpus:" | awk -v T="$2" '{for(i=4;i<=T;++i)print$i}') | sed 's/ /,/g' +} + function split_cli { jvm_options=() class_options=() diff --git a/libexec/profile b/libexec/profile index a16b534935..dc6088a323 100755 --- a/libexec/profile +++ b/libexec/profile @@ -25,7 +25,7 @@ done shift $((OPTIND-1)) executable=$1 arguments=${@:2} -logfile=asprof$stub.log +logfile=asprof_$event$stub.log which asprof >& /dev/null [ "$?" -ne 0 ] && echo 'ERROR: asprof is not in $PATH.' && echo -e $usage2 && exit 3 @@ -42,13 +42,17 @@ set -e $executable $arguments >> $logfile 2>&1 & set +e +# figure out the process id to profile: pid_bash=$! echo info: pid bash=$pid_bash, getting java ... sleep 1 pid_java=$(pgrep -P $pid_bash) -sleep 1 -# go 2 deep for run-clara: -pid_java=$(pgrep -P $pid_java | head -n 1) +if [[ "$executable" == *clara* ]] +then + # go 2 deep for run-clara: + sleep 1 + pid_java=$(pgrep -P $pid_java | head -n 1) +fi echo info: pid java=$pid_java tail -f $logfile & pid_tail=$! diff --git a/libexec/scaling b/libexec/scaling index 6cad8e3450..01e714d2af 100755 --- a/libexec/scaling +++ b/libexec/scaling @@ -7,9 +7,9 @@ def cli(): 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=8,16,24,34,44)',default='8,16,24,34,44') cli.add_argument('-e','--events', metavar='#',help='events per thread (default=555)',default=555,type=int) - cli.add_argument('-N','--numa', metavar='#',help='restrict to a NUMA socket (choices=[0,1])',choices=[0,1]) + cli.add_argument('-N','--numa', metavar='#',help='restrict to a NUMA socket (choices=[0,1])',choices=[0,1], type=int) cli.add_argument('-C','--cpus', metavar='#',help='restrict to a CPU list (e.g. 0,1,2,3)') - cli.add_argument('-T','--cputhr', metavar='#',help='restrict to same number of CPUs as threads') + cli.add_argument('-T','--cputhr', help='restrict to same number of CPUs as threads',action='store_true') cli.add_argument('datafile', help='input EVIO/HIPO data file') cfg = cli.parse_args() cfg.threads = cfg.threads.split(',') @@ -17,8 +17,8 @@ def cli(): cli.error('cannot find CLARA installation via -c or $CLARA_HOME') elif shutil.which('run-clara'): cfg.run_clara = shutil.which('run-clara') - elif os.path.exists(clara_home + '/plugins/clas12/bin/run-clara'): - cfg.run_clara = clara_home + '/plugins/clas12/bin/run-clara' + elif os.path.exists(cfg.clara + '/plugins/clas12/bin/run-clara'): + cfg.run_clara = cfg.clara + '/plugins/clas12/bin/run-clara' else: cli.error('cannot find run-clara in $PATH or $CLARA_HOME') if cfg.cpus: @@ -55,7 +55,7 @@ def benchmark(cfg, threads, log): # use taskset for thread affinity: if cfg.cpus: if cfg.cputhr: - cmd.extend(['taskset','-c',','.join(cfg.cpus.split(',')[:threads])]) + cmd.extend(['taskset','-c',','.join(cfg.cpus.split(',')[:int(threads)])]) else: cmd.extend(['taskset','-c',cfg.cpus]) # add the run-clara command: @@ -122,7 +122,7 @@ def show(benchmarks): print(' '.join([str(x) for x in row])) def save(benchmarks): - with open('scaling.txt','w') as f: + with open('scaling-clara.txt','w') as f: for row in table(benchmarks): f.write(' '.join([str(x) for x in row])+'\n') @@ -130,11 +130,11 @@ def plot(): import os,shutil,subprocess if shutil.which('gnuplot'): g = os.path.dirname(os.path.abspath(__file__))+'/scaling.gpl' - r = subprocess.run(['gnuplot',g], capture_output=True, text=True) + r = subprocess.run(['gnuplot',g,'scaling-clara.txt'], capture_output=True, text=True) if r.returncode == 0: - with open('scaling.svg','w') as f: + with open('scaling-clara.svg','w') as f: f.write(r.stdout) - print('Info: created scaling.svg from scaling.txt and scaling.gpl!') + print('Info: created scaling-clara.svg from scaling-clara.txt and coatjava/libexec/scaling.gpl!') else: print(r.stdout) print(r.stderr) diff --git a/libexec/scaling.gpl b/libexec/scaling.gpl index 3c5598d0d4..a827688ce6 100644 --- a/libexec/scaling.gpl +++ b/libexec/scaling.gpl @@ -1,4 +1,20 @@ set terminal svg size 1000,700 enhanced font 'Verdana, 14' + +# parse command line arguments: +# (usage: scaling.gpl [-m] [datafile]) +datafile = 'scaling.txt' +if (ARGC > 0) { + if (ARGV[1] eq "-m") { + recoMutil = 1 + set datafile separator "," + if (ARGC > 1) { + datafile = ARGV[ARGC] + } + } else { + datafile = ARGV[ARGC] + } +} + set palette maxcolors 8 set palette defined ( \ 0 '#A6CEE3',\ @@ -42,23 +58,10 @@ set style line 36 pt 6 lt 1 lc rgb '#E31A1C' # dark red set style line 37 pt 6 lt 1 lc rgb '#FDBF6F' # light orange set style line 38 pt 6 lt 1 lc rgb '#FF7F00' # dark orange -# parse command line arguments: -datafile = 'scaling.txt' -if (ARGC > 0) { - if (ARGV[1] eq "-u") { - uberMode = 1 - if (ARGC > 1) { - datafile = ARGV[ARGC] - } - } else { - datafile = ARGV[ARGC] - } -} - set multiplot layout 2,2 set datafile columnheaders +stats datafile nooutput set yrange [0:] - set xlabel 'Threads' set ylabel 'Event Rate [Hz]' set size 0.4,0.45 @@ -67,11 +70,17 @@ set key nobox inside top left f(x) = m*x m = 2 -set fit quiet -set fit nolog -fit [0:24] f(x) datafile using 1:(1/$2*1e3) via m -rate = sprintf('%.1f Hz/CPU',m) -plot datafile using 1:(1/$2*1e3) pt 7 notitle, f(x) title rate +set fit quiet nolog + +if (exists("recoMutil")) { + fit [0:24] f(x) datafile using 1:($1/column($#)*1e3) via m + rate = sprintf('%.1f Hz/Thread',m) + plot datafile using 1:($1/column($#)*1e3) pt 7 notitle, f(x) title rate, '' using 1:2 pt 9 with points +} else { + fit [0:24] f(x) datafile using 1:(1/$2*1e3) via m + rate = sprintf('%.1f Hz/Thread',m) + plot datafile using 1:(1/$2*1e3) pt 7 notitle, f(x) title rate +} set xlabel 'Threads' set ylabel 'Event Time [ms]' @@ -79,25 +88,28 @@ set key nobox outside top center horizontal set size 0.4,0.55 set origin 0.0,0.0 -plot datafile \ +set logscale y +if (exists("recoMutil")) { + plot datafile \ + using 1:((column($#))) pt 7 with points title 'Total' ,\ + '' using 1:((column("write"))) pt 11 with points title 'Writer' ,\ + '' using 1:((column("read"))) pt 9 with points title 'Reader' ,\ + +} else { + plot datafile \ using 1:((column($#))) pt 7 with points title 'Total' ,\ '' using 1:(($5*$1)) pt 7 with points title 'N-Readers' ,\ '' using 1:($5) pt 9 with points title 'Reader' ,\ '' using 1:((column($#-1))) pt 11 with points title 'Writer' ,\ +} +unset logscale y set size 0.6,1.0 set origin 0.4,0.0 -set key outside right vertical +set key outside right vertical font ",12" -if (exists("uberMode")) { - plot datafile \ - using 1:($6) pt 7 with points title columnhead(6) ,\ - '' using 1:($7) pt 7 with points title columnhead(7) ,\ - '' using 1:($8) pt 7 with points title columnhead(8) ,\ - '' using 1:($9) pt 7 with points title columnhead(9) ,\ - '' using 1:($10) pt 7 with points title columnhead(10) ,\ - '' using 1:($11) pt 7 with points title columnhead(11) ,\ - '' using 1:($12) pt 7 with points title columnhead(12) ,\ +if (exists("recoMutil")) { + plot for [i=3:STATS_columns-1] datafile using 1:i ls i-1 with points title columnhead(i) } else { plot datafile \ using 1:($6) ls 1 with points title columnhead(6) ,\ diff --git a/validation/advanced-tests/run-advanced-tests.sh b/validation/advanced-tests/run-advanced-tests.sh index c96599669a..d4ead657d0 100755 --- a/validation/advanced-tests/run-advanced-tests.sh +++ b/validation/advanced-tests/run-advanced-tests.sh @@ -4,35 +4,23 @@ # and input data files at ./data # set up environment -CLARA_HOME=$PWD/../../clara/ ; export CLARA_HOME -COAT=$CLARA_HOME/plugins/clas12/ - -# source coatjava environment -source $COAT/libexec/env.sh +source $PWD/../../coatjava/libexec/env.sh classPath="${COATJAVA_CLASSPATH}:../lib/*:src/" -# download test files -#wget --no-check-certificate http://clasweb.jlab.org/clas12offline/distribution/coatjava/validation_files/twoTrackEvents_809_raw.evio.tar.gz -#[ $? -ne 0 ] && echo "wget validation files failure" && exit 2 -#tar -zxvf twoTrackEvents_809_raw.evio.tar.gz - -# run decoder -$COAT/bin/decoder -t -0.5 -s 0.0 -i ./data/twoTrackEvents_809_raw.evio -o ./twoTrackEvents_809.hipo -c 2 -[ $? -ne 0 ] && echo "decoder failure" && exit 3 +# run reconstruction +recon-mutil -f -0.5,0 -t 4 -y $CLAS12DIR/etc/services/kpp.yaml -o rec_twoTrackEvents_809.hipo data/twoTrackEvents_809_raw.evio +[ $? -ne 0 ] && echo "recon-mutil failure" && exit 1 # take a peek -$COAT/bin/hipo-utils -stats ./twoTrackEvents_809.hipo - -# run clara -$COAT/bin/run-clara -y $COAT/etc/services/kpp.yaml ./twoTrackEvents_809.hipo -[ $? -ne 0 ] && echo "reconstruction with clara failure" && exit 4 +hipo-utils -stats ./rec_twoTrackEvents_809.hipo +[ $? -ne 0 ] && echo "recon-utils failure" && exit 2 # compile test codes javac -cp $classPath src/kpptracking/KppTrackingTest.java -[ $? -ne 0 ] && echo "KppTrackingTest compilation failure" && exit 5 +[ $? -ne 0 ] && echo "KppTrackingTest compilation failure" && exit 3 # run KppTracking junit tests -java -DCLAS12DIR="$COAT" -Xmx1536m -Xms1024m -cp $classPath org.junit.runner.JUnitCore kpptracking.KppTrackingTest -[ $? -ne 0 ] && echo "KppTracking unit test failure" && exit 6 +java -Xmx1536m -Xms1024m -cp $classPath org.junit.runner.JUnitCore kpptracking.KppTrackingTest +[ $? -ne 0 ] && echo "KppTracking unit test failure" && exit 4 echo "KppTracking passed unit tests" diff --git a/validation/advanced-tests/run-eb-tests.sh b/validation/advanced-tests/run-eb-tests.sh index 05c7470db5..1bac999bfe 100755 --- a/validation/advanced-tests/run-eb-tests.sh +++ b/validation/advanced-tests/run-eb-tests.sh @@ -49,7 +49,7 @@ if [ $? != 0 ] ; then echo "EBTwoTrackTest compilation failure" ; exit 1 ; fi # run reconstruction: rm -f out_${stub}.hipo -../../coatjava/bin/recon-util -l FINE -i ${input_dir}/${stub}.hipo -o out_${stub}.hipo -c 2 +../../coatjava/bin/recon-mutil -t 4 -l SEVERE -o out_${stub}.hipo ${input_dir}/${stub}.hipo # run EB tests: java -Xmx1536m -Xms1024m -cp $classPath -DINPUTFILE=out_${stub}.hipo eb.EBTwoTrackTest