diff --git a/bin/recon-mutil b/bin/recon-mutil new file mode 100755 index 0000000000..533f8130bb --- /dev/null +++ b/bin/recon-mutil @@ -0,0 +1,12 @@ +#!/bin/bash + +. `dirname $0`/../libexec/env.sh + +split_cli $@ + +export MALLOC_ARENA_MAX=1 + +java ${JAVA_OPTS-} -Xms10240m -XX:+UseParallelGC ${jvm_options[@]} \ + -cp ${COATJAVA_CLASSPATH:-''} \ + org.jlab.clas.reco.EngineMultiProcessor \ + ${class_options[@]} diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineMultiProcessor.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineMultiProcessor.java new file mode 100644 index 0000000000..d9e4c7e2c3 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/EngineMultiProcessor.java @@ -0,0 +1,209 @@ +package org.jlab.clas.reco; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.jlab.coda.jevio.EvioException; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.io.base.DataEvent; +import org.jlab.io.base.DataSource; +import org.jlab.io.evio.EvioDataEvent; +import org.jlab.io.evio.EvioSource; +import org.jlab.io.hipo.HipoDataEvent; +import org.jlab.io.hipo.HipoDataSource; +import org.jlab.io.hipo.HipoDataSync; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.benchmark.ProgressPrintout; +import org.jlab.utils.options.OptionParser; + +/** + * + * @author baltzell + */ +public class EngineMultiProcessor extends EngineProcessor { + + DataSource reader; + HipoDataSync writer; + CompletableFuture readerThread; + CompletableFuture writerThread; + ArrayList inputs = new ArrayList<>(); + ProgressPrintout progress = new ProgressPrintout(); + ConcurrentLinkedQueue procThreads = new ConcurrentLinkedQueue(); + ConcurrentLinkedQueue readQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue writeQueue = new ConcurrentLinkedQueue<>(); + + int threads; + int maxEvents = 0; + int maxEventsUser = 0; + int skipEvents = 0; + int readEvents = 0; + int writeEvents = 0; + + public EngineMultiProcessor(OptionParser parser) { + super(parser); + threads = parser.getOption("-t").intValue(); + maxEventsUser = parser.getOption("-n").intValue(); + skipEvents = parser.getOption("-s").intValue(); + } + + /** + * The thread launcher. + * @param output + * @param input + */ + public void process(String output, String... input) { + readerThread = CompletableFuture.runAsync(() -> { read(input); }); + writerThread = CompletableFuture.runAsync(() -> { write(output); }); + for (int i=0; i { process(j); })); + } + while (!writerThread.isDone()) { + for (CompletableFuture f : procThreads) + if (f.isDone()) procThreads.remove(f); + sleep(100); + } + } + + /** + * The reader thread. + * @param input input filenames + */ + void read(String... input) { + inputs.addAll(Arrays.asList(input)); + while (maxEvents < 1 || readEvents < maxEvents) { + if (reader != null && reader.hasEvent()) { + // sleep instead of overfilling the read queue: + if (readQueue.size() > 100*threads) sleep(100); + // read the next event: + else { + Benchmark.getInstance().resume("read"); + readEvents++; + Object o = null; + if (reader instanceof EvioSource evio) { + try { o = evio.getEventBuffer(readEvents, true); } + catch (EvioException ex) { ex.printStackTrace(); } + } + else o = reader.getNextEvent(); + if (skipEvents < 1 || readEvents > skipEvents) + if (o != null) readQueue.offer(o); + Benchmark.getInstance().pause("read"); + } + } + else if (inputs.isEmpty()) break; + else { + // open a new input file: + 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 { + // override maxEvents for EVIO: + int n = ((EvioSource)reader).getEventCount(); + maxEvents = maxEventsUser < n ? maxEventsUser : n; + } + readEvents = 0; + } + } + } + + /** + * The event processor thread. + * @param thread unique thread number + */ + void process(int thread) { + while (true) { + Object o = readQueue.poll(); + if (o == null) { + if (readerThread.isDone() && readQueue.isEmpty()) + break; + sleep(100); + } + else { + DataEvent event; + // decode if necessary: + if (o instanceof ByteBuffer bb) event = decode(bb); + else event = (HipoDataEvent)o; + // run it through the engine chain: + for (Map.Entry engine : processorEngines.entrySet()) { + Benchmark.getInstance().resume(engine.getValue().getName()); + try { engine.getValue().processDataEvent(event); } + catch (Exception ex) { ex.printStackTrace(); } + Benchmark.getInstance().pause(engine.getValue().getName()); + } + writeQueue.offer(event); + } + } + } + + /** + * The writer thread. + * @param output output filename + */ + void write(String output) { + writer = new HipoDataSync(); + writer.setCompressionType(2); + writer.open(output); + while (true) { + DataEvent e = writeQueue.poll(); + if (e == null) { + if (procThreads.isEmpty() && 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; + } + sleep(100); + } + else { + Benchmark.getInstance().resume("write"); + writer.writeEvent(e); + if (writeEvents > 100) progress.updateStatus(); + if (writeEvents == 101) Benchmark.getInstance().printTimer(10); + writeEvents++; + Benchmark.getInstance().pause("write"); + } + } + } + + /** + * Decoding. + * @param bytes EVIO byte buffer + * @return decoded event + */ + HipoDataEvent decode(ByteBuffer bytes) { + Benchmark.getInstance().resume("EVIO"); + EvioDataEvent evio = new EvioDataEvent(bytes.array(), ByteOrder.LITTLE_ENDIAN); + Benchmark.getInstance().pause("EVIO"); + Benchmark.getInstance().resume("DECO"); + HipoDataEvent hipo; + try { + CLASDecoder d = decoders.take(); + hipo = d.getDecodedDataEvenet(evio); + decoders.put(d); + } + catch (InterruptedException ex) { hipo = null; } + Benchmark.getInstance().pause("DECO"); + return hipo; + } + + void sleep(int milliseconds) { + try { Thread.sleep(milliseconds); } + catch (InterruptedException ex) {} + } + + public static void main(String[] args) { + OptionParser parser = EngineProcessor.getParser(); + parser.addOption("-t","4","number of threads"); + parser.parse(args); + EngineMultiProcessor proc = new EngineMultiProcessor(parser); + proc.process(parser.getOption("-o").stringValue(), parser.getOption("-i").stringValue()); + } +} 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..83e4f2dec6 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,11 +18,10 @@ 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.CLASDecoderPool; import org.jlab.io.evio.EvioDataEvent; import org.jlab.io.evio.EvioSource; -import org.jlab.io.hipo.HipoDataEvent; -import org.jlab.jnp.hipo4.data.Event; import org.jlab.jnp.hipo4.data.SchemaFactory; import org.json.JSONObject; import org.jlab.utils.ClaraYaml; @@ -36,16 +35,18 @@ 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 final CLASDecoderPool decoders = new CLASDecoderPool(64,"default",null); public EngineProcessor(){} + public EngineProcessor(OptionParser p) { init(p); } + private ReconstructionEngine findEngine(String clazz) { for (String k : processorEngines.keySet()) { if (processorEngines.get(k).getClass().getName().equals(clazz)) { @@ -90,7 +91,7 @@ private void setPreloadFiles(String filenames, boolean restream, boolean 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(); @@ -251,7 +252,8 @@ public void addEngine(String name, String clazz, String jsonConf) { } this.processorEngines.put(name == null ? engine.getName() : name, engine); } else { - LOGGER.log(Level.SEVERE, ">>>> ERROR: class is not a reconstruction engine : {0}", clazz); + LOGGER.log( clazz.contains("DecoderEngine") ? Level.INFO : Level.SEVERE, + "Class is not a reconstruction engine : {0}", clazz); } } catch (ClassNotFoundException | InstantiationException | IllegalAccessException ex) { @@ -338,9 +340,13 @@ public void processFile(EvioSource reader, HipoDataSync writer, int skipEvents, 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); + try { + CLASDecoder d = decoders.take(); + processEvent(d.getDecodedDataEvenet(evio), writer); + decoders.put(d); + } catch (InterruptedException ex) { + System.getLogger(EngineProcessor.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); + } } if (maxEvents > 0 && eventsRead > maxEvents+skipEvents) break; } catch (EvioException ex) { @@ -390,7 +396,6 @@ 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 +406,56 @@ 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(); + protected final void init(OptionParser p) { + p.syncLogLevel(LOGGER); - EngineProcessor proc = new EngineProcessor(); + if(p.getOption("-u").stringValue().contains("false")) updateDictionary = false; - 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); + // configure from YAML: + if(!p.getOption("-y").stringValue().equals("0")) { + parseYaml(p.getOption("-y").stringValue()); } - else if (config>0){ - if(config>2){ - proc.initCaloDebug(); - } else if(config==2){ - proc.initAll(); - } else { - proc.initDefault(); - } + // builtin engine list: + else if (p.getOption("-c").intValue() > 0) { + if (p.getOption("-c").intValue() > 2 ) initCaloDebug(); + else if (p.getOption("-c").intValue() == 2) initAll(); + else initDefault(); } + // command-line engine list: else { - for(String engine : services){ + 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/validation/advanced-tests/run-eb-tests.sh b/validation/advanced-tests/run-eb-tests.sh index 05c7470db5..b21c234a1c 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 6 -l FINE -i ${input_dir}/${stub}.hipo -o out_${stub}.hipo -c 2 # run EB tests: java -Xmx1536m -Xms1024m -cp $classPath -DINPUTFILE=out_${stub}.hipo eb.EBTwoTrackTest