diff --git a/bin/recon-mutil b/bin/recon-mutil new file mode 100755 index 0000000000..c495c84120 --- /dev/null +++ b/bin/recon-mutil @@ -0,0 +1,13 @@ +#!/bin/bash + +. `dirname $0`/../libexec/env.sh + +split_cli $@ + +export MALLOC_ARENA_MAX=1 + +java ${JAVA_OPTS-} -Xms10240m -XX:+UseSerialGC ${jvm_options[@]} \ + -cp ${COATJAVA_CLASSPATH:-''} \ + org.jlab.clas.reco.EngineMultiProcessor \ + ${class_options[@]} + diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoder.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoder.java index 45af4c9293..9755da4a86 100644 --- a/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoder.java +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoder.java @@ -797,8 +797,7 @@ public Event getDecodedEvent(EvioDataEvent rawEvent) { return getDecodedEvent(rawEvent, -1, -1, null, null); } - public HipoDataEvent getDecodedDataEvenet(EvioDataEvent rawEvent) { + public HipoDataEvent getDecodedDataEvent(EvioDataEvent rawEvent) { return new HipoDataEvent(getDecodedEvent(rawEvent), schemaFactory); } - -} \ No newline at end of file +} diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/DecoderPool.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/DecoderPool.java new file mode 100644 index 0000000000..7fd7eeeb90 --- /dev/null +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/DecoderPool.java @@ -0,0 +1,37 @@ +package org.jlab.detector.decode; + +import java.util.concurrent.ArrayBlockingQueue; + +/** + * + * @author baltzell + */ +public class DecoderPool { + + ArrayBlockingQueue pool; + int constantsShared = 64; + + public DecoderPool(int size, String variation, String timestamp) { + pool = new ArrayBlockingQueue<>(size); + CLASDecoder d0 = null; + for (int i=0; i pool; - int constantsShared = 64; + DecoderPool pool; public DecoderEngine() { schema = new SchemaFactory(); @@ -57,22 +55,10 @@ public void destroy() {} @Override public EngineData configure(EngineData ed) { - JSONObject json = new JSONObject(ed.getData()); - pool = new ArrayBlockingQueue<>(POOL_SIZE); - CLASDecoder d0 = null; - for (int i=0; i { read(input); }); + writerThread = CompletableFuture.runAsync(() -> { write(output); }); + for (int i=0; i { process(j); })); + } + while (!writerThread.isDone()) + try { Thread.sleep(100); } catch (InterruptedException ex) {} + } + + DataSource reader; + HipoDataSync writer; + CompletableFuture readerThread; + CompletableFuture writerThread; + + int threads; + int maxEvents = 0; + int maxEventsUser = 0; + int skipEvents = 0; + int readEvents = 0; + int writeEvents = 0; + + ArrayList inputs = new ArrayList<>(); + + ConcurrentLinkedQueue procThreads = new ConcurrentLinkedQueue(); + ConcurrentLinkedQueue evioQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue hipoQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue writeQueue = new ConcurrentLinkedQueue<>(); + + ProgressPrintout progress = new ProgressPrintout(); + + void read(String... input) { + inputs.addAll(Arrays.asList(input)); + while (maxEvents < 1 || readEvents < maxEvents) { + if (reader != null && reader.hasEvent()) { + if (evioQueue.size()+hipoQueue.size() > 100*threads) { + try { Thread.sleep(100); } + catch (InterruptedException ex) {} + } + else { + Benchmark.getInstance().resume("read"); + readEvents++; + if (reader instanceof EvioSource evio) { + try { evioQueue.offer(evio.getEventBuffer(readEvents, true)); } + catch (EvioException ex) { ex.printStackTrace(); } + } + else { + DataEvent event = reader.getNextEvent(); + if (skipEvents < 1 || readEvents > skipEvents) + hipoQueue.offer(event); + } + Benchmark.getInstance().pause("read"); + } + } + else if (inputs.isEmpty()) break; + else { + if (inputs.get(0).endsWith(".hipo")) reader = new HipoDataSource(); + else reader = new EvioSource(); + reader.open(inputs.remove(0)); + maxEvents = maxEventsUser; + if (reader instanceof HipoDataSource hipo) + updateDictionary(hipo, writer); + else { + int n = ((EvioSource)reader).getEventCount(); + maxEvents = maxEventsUser < n ? maxEventsUser : n; + } + readEvents = 0; + } + } + } + + EvioDataEvent process(ByteBuffer bytes) { + Benchmark.getInstance().resume("EVIO"); + EvioDataEvent e = new EvioDataEvent(bytes.array(), ByteOrder.LITTLE_ENDIAN); + Benchmark.getInstance().pause("EVIO"); + return e; + } + + HipoDataEvent process(EvioDataEvent event) { + Benchmark.getInstance().resume("DECO"); + HipoDataEvent e; + try { + CLASDecoder d = decoders.take(); + e = d.getDecodedDataEvent(event); + decoders.put(d); + } + catch (InterruptedException ex) { e = null; } + Benchmark.getInstance().pause("DECO"); + return e; + } + + void process(HipoDataEvent event) { + for (Map.Entry engine : this.processorEngines.entrySet()) { + Benchmark.getInstance().resume(engine.getValue().getName()); + try { engine.getValue().processDataEvent(event); } + catch (Exception ex) { ex.printStackTrace(); } + Benchmark.getInstance().pause(engine.getValue().getName()); + } + } + + void process(int thread) { + while (true) { + if (evioQueue.isEmpty() && hipoQueue.isEmpty()) { + if (readerThread.isDone()) + if (evioQueue.isEmpty() && hipoQueue.isEmpty()) break; + try { Thread.sleep(100); } + catch (InterruptedException ex) {} + } + else { + DataEvent event; + if (!evioQueue.isEmpty()) event = process(process(evioQueue.poll())); + else if (!hipoQueue.isEmpty()) event = hipoQueue.poll(); + else continue; + process((HipoDataEvent)event); + writeQueue.offer(event); + } + } + } + + void write(String output) { + writer = new HipoDataSync(); + writer.setCompressionType(2); + writer.open(output); + while (true) { + if (writeQueue.isEmpty()) { + for (CompletableFuture f : procThreads) + if (f.isDone()) procThreads.remove(f); + if (procThreads.isEmpty()) { + if (writeQueue.isEmpty()) { + writer.close(); + System.out.println(Benchmark.getInstance()); + System.out.println(String.format("recon-mutil::::: Read/Write/Diff = %d/%d/%d", + readEvents, writeEvents, readEvents-writeEvents)); + break; + } + } + try { Thread.sleep(100); } + catch (InterruptedException ex) {} + } + else write(writeQueue.poll()); + } + } + + void write(DataEvent event) { + Benchmark.getInstance().resume("write"); + writer.writeEvent(writeQueue.poll()); + if (writeEvents > 100) progress.updateStatus(); + if (writeEvents == 101) Benchmark.getInstance().printTimer(10); + writeEvents++; + Benchmark.getInstance().pause("write"); + } + +} 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 b0b4a98542..0f08c2db3b 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 @@ -18,7 +18,8 @@ import org.jlab.clara.engine.EngineDataType; import java.util.Arrays; import org.jlab.coda.jevio.EvioException; -import org.jlab.detector.decode.CLASDecoder4; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.detector.decode.DecoderPool; import org.jlab.io.evio.EvioDataEvent; import org.jlab.io.evio.EvioSource; import org.jlab.io.hipo.HipoDataEvent; @@ -32,20 +33,27 @@ * @author gavalian, kenjo, baltzell */ public class EngineProcessor { - + public static final String ENGINE_CLASS_BG = "org.jlab.service.bg.BackgroundEngine"; public static final String ENGINE_CLASS_PP = "org.jlab.service.postproc.PostprocEngine"; - private final Map processorEngines = new LinkedHashMap<>(); + protected final Map processorEngines = new LinkedHashMap<>(); private static final Logger LOGGER = Logger.getLogger(EngineProcessor.class.getPackage().getName()); - private boolean updateDictionary = true; private SchemaFactory banksToKeep = null; private final List schemaExempt = Arrays.asList("RUN::config","DC::tdc"); - - private CLASDecoder4 decoder = new CLASDecoder4(); - + + protected boolean updateDictionary = true; + + protected DecoderPool decoders = new DecoderPool(64,"default",null); + + int eventsRead = 0; + public EngineProcessor(){} - + + public EngineProcessor(OptionParser parser) { + init(parser); + } + private ReconstructionEngine findEngine(String clazz) { for (String k : processorEngines.keySet()) { if (processorEngines.get(k).getClass().getName().equals(clazz)) { @@ -54,7 +62,7 @@ private ReconstructionEngine findEngine(String clazz) { } return null; } - + public void parseYaml(String filename) { ClaraYaml yaml = new ClaraYaml(filename); if (yaml.schemaDirectory() != null) { @@ -69,7 +77,7 @@ public void parseYaml(String filename) { } } } - + private void setBackgroundFiles(String filenames) { if (findEngine(ENGINE_CLASS_BG) == null) { LOGGER.info("Adding BackgroundEngine for -B option."); @@ -78,8 +86,8 @@ private void setBackgroundFiles(String filenames) { findEngine(ENGINE_CLASS_BG).engineConfigMap.put("filename", filenames); findEngine(ENGINE_CLASS_BG).init(); } - - private void setPreloadFiles(String filenames, boolean restream, boolean rebuild) { + + protected void setPreloadFiles(String filenames, boolean restream, boolean rebuild) { if (findEngine(ENGINE_CLASS_PP) == null) { LOGGER.info("Adding PostprocEngine for -P option."); addEngine("BG",ENGINE_CLASS_PP); @@ -89,8 +97,8 @@ private void setPreloadFiles(String filenames, boolean restream, boolean rebuild findEngine(ENGINE_CLASS_PP).engineConfigMap.put("rebuild", String.valueOf(rebuild)); findEngine(ENGINE_CLASS_PP).init(); } - - private void updateDictionary(HipoDataSource source, HipoDataSync sync){ + + protected void updateDictionary(HipoDataSource source, HipoDataSync sync){ SchemaFactory fsync = sync.getWriter().getSchemaFactory(); SchemaFactory fsrc = source.getReader().getSchemaFactory(); List schemaList = fsync.getSchemaKeys(); @@ -99,13 +107,13 @@ private void updateDictionary(HipoDataSource source, HipoDataSync sync){ fsrc.remove(schemaList.get(s)); fsrc.addSchema(fsync.getSchema(schemaList.get(s)).getCopy()); } else { - LOGGER.log(Level.INFO, "[dictrionary-update] schema {0} is not being updated\n", + LOGGER.log(Level.INFO, "[dictrionary-update] schema {0} is not being updated\n", schemaList.get(s)); } } } - - private void setBanksToKeep(String schemaDirectory) { + + protected void setBanksToKeep(String schemaDirectory) { if (!Files.isDirectory((new File(schemaDirectory)).toPath())) { LOGGER.log(Level.SEVERE, "Invalid schema directory, aborting: "+schemaDirectory); System.exit(1); @@ -114,7 +122,7 @@ private void setBanksToKeep(String schemaDirectory) { banksToKeep = new SchemaFactory(); banksToKeep.initFromDirectory(schemaDirectory); } - + private void removeBanks(DataEvent event) { if (banksToKeep != null) { for (String bankName : event.getBankList()) { @@ -124,15 +132,15 @@ private void removeBanks(DataEvent event) { } } } - + public void initDefault(){ - + String[] names = new String[]{ "MAGFIELDS", "DCCR","DCHB","FTOFHB","EC","HTCC","EBHB", "DCTB","FTOFTB","EBTB","VTX" }; - + String[] services = new String[]{ "org.jlab.clas.swimtools.MagFieldsEngine", "org.jlab.service.dc.DCHBClustering", @@ -146,13 +154,13 @@ public void initDefault(){ "org.jlab.service.eb.EBTBEngine", "org.jlab.rec.service.vtx.VTXEngine" }; - + for(int i = 0; i < names.length; i++){ this.addEngine(names[i], services[i]); } } public void initAll(){ - + String[] names = new String[]{ "MAGFIELDS", "FTCAL", "FTHODO", "FTTRK", "FTEB", @@ -162,7 +170,7 @@ public void initAll(){ "DCTB","FMT","FTOFTB","CVT","EBTB", "RICHEB","RTPC","AHDC","ATOF","ALERT", "MC","VTX" }; - + String[] services = new String[]{ "org.jlab.clas.swimtools.MagFieldsEngine", "org.jlab.rec.ft.cal.FTCALEngine", @@ -201,23 +209,23 @@ public void initAll(){ this.addEngine(names[i], services[i]); } } - - public void initCaloDebug(){ - + + public void initCaloDebug(){ + String[] names = new String[]{ "EC","EB" }; - + String[] services = new String[]{ "org.jlab.service.ec.ECEngine", "org.jlab.service.eb.EBEngine", }; - + for(int i = 0; i < names.length; i++){ this.addEngine(names[i], services[i]); } } - + /** * add a reconstruction engine to the chain * @param name name of the engine in the chain @@ -227,7 +235,7 @@ public void addEngine(String name, ReconstructionEngine engine){ engine.init(); this.processorEngines.put(name, engine); } - + /** * Adding engine to the map the order of the services matters, since they will * be executed in order added. @@ -239,7 +247,7 @@ public void addEngine(String name, String clazz, String jsonConf) { Class c; try { c = Class.forName(clazz); - if( ReconstructionEngine.class.isAssignableFrom(c)==true){ + if (ReconstructionEngine.class.isAssignableFrom(c)==true) { ReconstructionEngine engine = (ReconstructionEngine) c.newInstance(); if(jsonConf != null && !jsonConf.equals("null")) { EngineData input = new EngineData(); @@ -250,15 +258,19 @@ public void addEngine(String name, String clazz, String jsonConf) { engine.init(); } this.processorEngines.put(name == null ? engine.getName() : name, engine); - } else { - LOGGER.log(Level.SEVERE, ">>>> ERROR: class is not a reconstruction engine : {0}", clazz); } - + else if (clazz.endsWith("DecoderEngine")) { + + } + else { + LOGGER.log(clazz.endsWith("DecoderEngine") ? Level.INFO : Level.SEVERE, + ">>>> class is not a reconstruction engine : {0}", clazz); + } } catch (ClassNotFoundException | InstantiationException | IllegalAccessException ex) { LOGGER.log(Level.SEVERE, null, ex); } } - + /** * Adding engine to the map the order of the services matters, since they will * be executed in order added. @@ -268,7 +280,7 @@ public void addEngine(String name, String clazz, String jsonConf) { public void addEngine(String name, String clazz) { this.addEngine(name, clazz, null); } - + /** * Add reconstruction engine to the chain * @param clazz Engine class. @@ -276,11 +288,11 @@ public void addEngine(String name, String clazz) { public void addEngine(String clazz) { this.addEngine(null, clazz, null); } - + /** * Initialize all the engines in the chain. */ - public void init(){ + public final void init(){ System.out.println("\n\n\n "); for(Map.Entry entry : this.processorEngines.entrySet()){ System.out.println(String.format(" >>>>>> (*) initializing : %8s : %s",entry.getKey(), @@ -289,12 +301,21 @@ public void init(){ } System.out.println("\n\n"); } - + /** * process a single event through the chain. * @param event */ - public void processEvent(DataEvent event){ + public DataEvent processEvent(DataEvent event) { + if (event instanceof EvioDataEvent evio) { + try { + CLASDecoder d = decoders.take(); + Event hipo = d.getDecodedEvent(evio, -1, ++eventsRead, null, null); + event = new HipoDataEvent(hipo, d.getSchemaFactory()); + decoders.put(d); + } + catch (InterruptedException ex) { ex.printStackTrace(); return null; } + } for(Map.Entry engine : this.processorEngines.entrySet()){ try { engine.getValue().processDataEvent(event); @@ -303,22 +324,25 @@ public void processEvent(DataEvent event){ e.printStackTrace(); } } + return event; } - + public void processFile(String file, String output){ this.processFile(file, output, -1, -1); } - + public void processEvent(DataEvent event, HipoDataSync writer) { processEvent(event); - removeBanks(event); - writer.writeEvent(event); + if (event instanceof HipoDataEvent) { + removeBanks(event); + writer.writeEvent(event); + } } - + public void processFile(HipoDataSource reader, HipoDataSync writer, int skipEvents, int maxEvents) { if (updateDictionary==true) updateDictionary(reader, writer); ProgressPrintout progress = new ProgressPrintout(); - int eventsRead = 0; + eventsRead = 0; while (reader.hasEvent()) { DataEvent event = reader.getNextEvent(); eventsRead++; @@ -328,19 +352,17 @@ public void processFile(HipoDataSource reader, HipoDataSync writer, int skipEven } progress.showStatus(); } - + public void processFile(EvioSource reader, HipoDataSync writer, int skipEvents, int maxEvents) { ProgressPrintout progress = new ProgressPrintout(); - int eventsRead = 0; + eventsRead = 0; while (reader.hasEvent()) { eventsRead++; try { ByteBuffer bb = reader.getEventBuffer(eventsRead, true); if (skipEvents <= 0 || eventsRead > skipEvents) { EvioDataEvent evio = new EvioDataEvent(bb.array(), ByteOrder.LITTLE_ENDIAN); - Event hipo = decoder.getDecodedEvent(evio, -1, eventsRead, null, null); - HipoDataEvent hipo2 = new HipoDataEvent(hipo, decoder.getSchemaFactory()); - processEvent(hipo2, writer); + processEvent(evio, writer); } if (maxEvents > 0 && eventsRead > maxEvents+skipEvents) break; } catch (EvioException ex) { @@ -350,7 +372,7 @@ public void processFile(EvioSource reader, HipoDataSync writer, int skipEvents, } progress.showStatus(); } - + /**} * process entire file through engine chain. * @param input input file name to process @@ -375,7 +397,7 @@ public void processFile(String input, String output, int nskip, int nevents) { } writer.close(); } - + /** * display services registered with the processor. */ @@ -385,12 +407,11 @@ public void show(){ System.out.println(String.format("%-24s | %s", entry.getKey(),entry.getValue().getClass().getName())); } } - + 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","-1","number of events to skip"); parser.addOption("-n","-1","number of events to process"); @@ -401,65 +422,67 @@ protected static OptionParser getParser() { 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; } - - public static void main(String[] args){ - - OptionParser parser = EngineProcessor.getParser(); - parser.parse(args); - parser.syncLogLevel(LOGGER); - - List services = parser.getInputList(); - - String inputFile = parser.getOption("-i").stringValue(); - String outputFile = parser.getOption("-o").stringValue(); - - EngineProcessor proc = new EngineProcessor(); - - int config = parser.getOption("-c").intValue(); - int nskip = parser.getOption("-s").intValue(); - int nevents = parser.getOption("-n").intValue(); - String yamlFileName = parser.getOption("-y").stringValue(); - - String update = parser.getOption("-u").stringValue(); - if(update.contains("false")==true) proc.updateDictionary = false; - - if(!yamlFileName.equals("0")) { - proc.parseYaml(yamlFileName); - } - else if (config>0){ - if(config>2){ - proc.initCaloDebug(); - } else if(config==2){ - proc.initAll(); - } else { - proc.initDefault(); - } + + final void init(int config) { + if (config > 0) { + if (config > 2) initCaloDebug(); + else if(config == 2) initAll(); + else initDefault(); } - else { - for(String engine : services){ + } + + final void init(OptionParser p) { + + p.syncLogLevel(LOGGER); + + if (p.getOption("-u").stringValue().contains("false")) + updateDictionary = false; + + // CLARA yaml file: + if (!p.getOption("-y").isDefault()) + parseYaml(p.getOption("-y").stringValue()); + + // builtin configuration: + else if (!p.getOption("-c").isDefault()) + init(p.getOption("-c").intValue()); + + // user-defined services: + else + for (String engine : p.getInputList()) { System.out.println("Adding reconstruction engine " + engine); - proc.addEngine(engine); + addEngine(engine); } - } - + // command-line schema overrides YAML: - if (parser.getOption("-S").stringValue() != null) - proc.setBanksToKeep(parser.getOption("-S").stringValue()); - + if (p.getOption("-S").stringValue() != null) + setBanksToKeep(p.getOption("-S").stringValue()); + // command-line filename for background merging overrides YAML: - if (parser.getOption("-B").stringValue() != null) - proc.setBackgroundFiles(parser.getOption("-B").stringValue()); + if (p.getOption("-B").stringValue() != null) + setBackgroundFiles(p.getOption("-B").stringValue()); // command-line filename for post-processing overrides YAML: - if (parser.getOption("-P").stringValue() != null) { - proc.setPreloadFiles(parser.getOption("-P").stringValue(), - parser.getOption("-H").intValue()!=0, - parser.getOption("-R").intValue()!=0); + if (p.getOption("-P").stringValue() != null) { + setPreloadFiles(p.getOption("-P").stringValue(), + p.getOption("-H").intValue()!=0, + p.getOption("-R").intValue()!=0); } - - proc.processFile(inputFile,outputFile,nskip,nevents); } - + + public static void main(String[] args) { + + OptionParser parser = EngineProcessor.getParser(); + parser.parse(args); + + EngineProcessor proc = new EngineProcessor(parser); + + proc.processFile(parser.getOption("-i").stringValue(), + parser.getOption("-o").stringValue(), + parser.getOption("-s").intValue(), + parser.getOption("-n").intValue()); + } + } diff --git a/common-tools/clas-utils/pom.xml b/common-tools/clas-utils/pom.xml index 98a561ce92..a8cec71ad2 100644 --- a/common-tools/clas-utils/pom.xml +++ b/common-tools/clas-utils/pom.xml @@ -24,6 +24,10 @@ clas-logging 14.1.3-SNAPSHOT + + junit + junit + 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 a85aa443d8..96ef935a8b 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 @@ -6,6 +6,8 @@ import java.util.Map; import java.util.Timer; import java.util.TimerTask; +import org.jlab.utils.benchmark.BenchmarkTimer.BenchmarkMultiTimer; +import org.jlab.utils.benchmark.BenchmarkTimer.BenchmarkTimerTotal; /** * @@ -14,7 +16,7 @@ public class Benchmark { private static final Benchmark benchmarkInstance = new Benchmark(); - private final Map timerStore = new LinkedHashMap<>(); + private final Map timerStore = new LinkedHashMap<>(); private Timer updateTimer = null; private Benchmark() {} @@ -39,24 +41,37 @@ public void reset(){ public void addTimer(String name){ if (!timerStore.containsKey(name)) - timerStore.put(name, new BenchmarkTimer(name)); + timerStore.put(name, new BenchmarkMultiTimer(name)); else System.err.println("[Benchmark] -----> error. timer with name ("+ name + ") already exists"); } public void pause(String name){ if (!timerStore.containsKey(name)) - timerStore.put(name, new BenchmarkTimer(name)); + timerStore.put(name, new BenchmarkMultiTimer(name)); else timerStore.get(name).pause(); } public void resume(String name){ if (!timerStore.containsKey(name)) - timerStore.put(name, new BenchmarkTimer(name)); + timerStore.put(name, new BenchmarkMultiTimer(name)); timerStore.get(name).resume(); } + public void pause(int thread, String name){ + if (!timerStore.containsKey(name)) + timerStore.put(name, new BenchmarkMultiTimer(name)); + else + timerStore.get(name).pause(thread); + } + + public void resume(int thread, String name){ + if (!timerStore.containsKey(name)) + timerStore.put(name, new BenchmarkMultiTimer(name)); + timerStore.get(name).resume(thread); + } + public BenchmarkTimer getTimer(String name){ return timerStore.getOrDefault(name, null); } @@ -71,7 +86,7 @@ public BenchmarkTimer getTotal(String name) { @Override public String toString(){ StringBuilder s = new StringBuilder(); - Collection timers = timerStore.values(); + Collection timers = timerStore.values(); if (!timers.isEmpty()) { int len = timers.iterator().next().toString().length(); char[] asterix = new char[len+8]; 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 4e0d12f2e3..803a754d8b 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,17 +1,57 @@ package org.jlab.utils.benchmark; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + /** * * @author gavalian */ public class BenchmarkTimer { + public static class BenchmarkMultiTimer extends BenchmarkTimer { + HashMap timeAtResume = new HashMap<>(); + HashMap isPaused = new HashMap<>(); + public BenchmarkMultiTimer(String name) { super(name); } + public void resume(int thread) { + if (!isPaused.containsKey(thread) || isPaused.get(thread)) { + timeAtResume.put(thread, System.nanoTime()); + isPaused.put(thread, false); + } + } + public void pause(int thread) { + if (!isPaused.get(thread)) { + numberOfCalls.incrementAndGet(); + totalTime.addAndGet(System.nanoTime() - timeAtResume.get(thread)); + isPaused.put(thread, true); + } + } + @Override + public void reset(){ + super.reset(); + timeAtResume.clear(); + isPaused.clear(); + } + } + + public static class BenchmarkTimerTotal extends BenchmarkMultiTimer { + ArrayList benchmarks = new ArrayList<>(); + public BenchmarkTimerTotal(String name) { super(name); } + public void add(BenchmarkTimer b) { + benchmarks.add(b); + totalTime.addAndGet(b.totalTime.get()); + numberOfCalls.addAndGet(b.numberOfCalls.get()); + } + } + private String timerName = "generic"; private long timeAtResume = 0; private Boolean isPaused = true; - protected int numberOfCalls = 0; - protected long totalTime = 0; + AtomicInteger numberOfCalls = new AtomicInteger(0); + AtomicLong totalTime = new AtomicLong(0); public BenchmarkTimer() {} @@ -32,33 +72,32 @@ public void resume(){ public void pause(){ if(isPaused==false){ - long timeAtPause = System.nanoTime(); - totalTime += (timeAtPause - timeAtResume); - numberOfCalls++; + totalTime.addAndGet(System.nanoTime() - timeAtResume); + numberOfCalls.incrementAndGet(); isPaused = true; } } public void reset(){ - totalTime = 0; + totalTime.set(0); + numberOfCalls.set(0); timeAtResume = 0; - numberOfCalls = 0; isPaused = true; } public double getMiliseconds(){ - return totalTime/(1.0e6); + return totalTime.get() / 1.0e6; } public double getSeconds(){ - return totalTime/(1.0e9); + return totalTime.get() / 1.0e9; } @Override public String toString() { double timePerCall = 0.0; - if (numberOfCalls != 0) timePerCall = getMiliseconds() / numberOfCalls; + if (numberOfCalls.get() != 0) timePerCall = getMiliseconds() / numberOfCalls.get(); return String.format("%-15s : #Calls %12d, Total = %12.2f sec, Unit = %12.3f msec", - getName(), numberOfCalls, getSeconds(), timePerCall); + getName(), numberOfCalls.get(), getSeconds(), timePerCall); } } diff --git a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimerTotal.java b/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimerTotal.java deleted file mode 100644 index b16551893d..0000000000 --- a/common-tools/clas-utils/src/main/java/org/jlab/utils/benchmark/BenchmarkTimerTotal.java +++ /dev/null @@ -1,30 +0,0 @@ -package org.jlab.utils.benchmark; - -import java.util.ArrayList; - -/** - * - * @author baltzell - */ -public class BenchmarkTimerTotal extends BenchmarkTimer { - - ArrayList benchmarks = new ArrayList<>(); - - public BenchmarkTimerTotal(String name) { - super(name); - } - - @Override - public String toString() { - double timePerCall = 0.0; - if (numberOfCalls != 0) timePerCall = getMiliseconds() / numberOfCalls * benchmarks.size(); - return String.format("%-15s : #Calls %12d, Total = %12.2f sec, Unit = %12.3f msec", - getName(), numberOfCalls, getSeconds(), timePerCall); - } - - public void add(BenchmarkTimer b) { - benchmarks.add(b); - totalTime += b.totalTime; - numberOfCalls += b.numberOfCalls; - } -} diff --git a/common-tools/clas-utils/src/test/java/org/jlab/utils/benchmark/BenchmarkTest.java b/common-tools/clas-utils/src/test/java/org/jlab/utils/benchmark/BenchmarkTest.java new file mode 100644 index 0000000000..1063a2e0ba --- /dev/null +++ b/common-tools/clas-utils/src/test/java/org/jlab/utils/benchmark/BenchmarkTest.java @@ -0,0 +1,17 @@ +package org.jlab.utils.benchmark; + +import org.junit.Test; + +public class BenchmarkTest { + + @Test + public void multi() throws InterruptedException { + Benchmark.getInstance().resume(1,"test"); + Benchmark.getInstance().resume(2,"test"); + Thread.sleep(1000); + Benchmark.getInstance().pause(1,"test"); + Benchmark.getInstance().pause(2,"test"); + System.out.println(Benchmark.getInstance()); + } + +} diff --git a/validation/advanced-tests/run-eb-tests.sh b/validation/advanced-tests/run-eb-tests.sh index 05c7470db5..a996fb638e 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 FINE -i ${input_dir}/${stub}.hipo -o out_${stub}.hipo -c 2 -- -Xmx25000m -Xms3000m # run EB tests: java -Xmx1536m -Xms1024m -cp $classPath -DINPUTFILE=out_${stub}.hipo eb.EBTwoTrackTest