diff --git a/bin/recon-mutil b/bin/recon-mutil new file mode 100755 index 0000000000..622579382b --- /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-} -Xms2048m -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/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(); })); + while (!writerThread.isDone()) + try { Thread.sleep(100); } catch (InterruptedException ex) {} + } + + DataSource reader; + HipoDataSync writer; + CompletableFuture readerThread; + CompletableFuture writerThread; + + int threads; + int maxEvents = 0; + int skipEvents = 0; + int readEvents = 0; + + ArrayList procThreads = new ArrayList<>(); + ArrayList inputs = new ArrayList<>(); + ProgressPrintout progress = new ProgressPrintout(); + ConcurrentLinkedQueue evioQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue hipoQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue writeQueue = new ConcurrentLinkedQueue<>(); + + 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 { + 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); + } + } + } + else if (inputs.isEmpty()) break; + else { + readEvents = 0; + if (inputs.get(0).endsWith(".hipo")) reader = new HipoDataSource(); + else reader = new EvioSource(); + reader.open(inputs.remove(0)); + if (reader instanceof HipoDataSource) + updateDictionary((HipoDataSource)reader, writer); + else + maxEvents = ((EvioSource)reader).getEventCount(); + } + } + } + + void write(String output) { + writer = new HipoDataSync(); + writer.setCompressionType(2); + writer.open(output); + while (true) { + if (writeQueue.isEmpty()) { + if (procThreads.stream().filter(x->!x.isDone()).collect(Collectors.toList()).isEmpty()) { + if (writeQueue.isEmpty()) { + progress.showStatus(); + writer.close(); + break; + } + } + try { Thread.sleep(100); } + catch (InterruptedException ex) {} + } + else { + writer.writeEvent(writeQueue.poll()); + if (readEvents > 20) progress.updateStatus(); + } + } + } + + void process() { + 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 = new EvioDataEvent(evioQueue.poll().array(), ByteOrder.LITTLE_ENDIAN); + else if (!hipoQueue.isEmpty()) + event = hipoQueue.poll(); + else continue; + processEvent(event); + writeQueue.offer(event); + } + } + } +} 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 5f17e9b6d4..46d38862f7 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; @@ -38,14 +39,21 @@ public class EngineProcessor { private 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)) { @@ -55,7 +63,22 @@ private ReconstructionEngine findEngine(String clazz) { return null; } - private void setBackgroundFiles(String filenames) { + protected void parseYaml(String filename) { + ClaraYaml yaml = new ClaraYaml(filename); + if (yaml.schemaDirectory() != null) { + setBanksToKeep(yaml.schemaDirectory()); + } + for (JSONObject service : yaml.services()) { + JSONObject cfg = yaml.filter(service.getString("name")); + if (cfg.length() > 0) { + addEngine(service.getString("name"),service.getString("class"),cfg.toString()); + } else { + addEngine(service.getString("name"),service.getString("class")); + } + } + } + + protected void setBackgroundFiles(String filenames) { if (findEngine(ENGINE_CLASS_BG) == null) { LOGGER.info("Adding BackgroundEngine for -B option."); addEngine("BG",ENGINE_CLASS_BG); @@ -64,7 +87,7 @@ private void setBackgroundFiles(String 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); @@ -75,7 +98,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(); @@ -90,7 +113,7 @@ private void updateDictionary(HipoDataSource source, HipoDataSync sync){ } } - 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); @@ -187,7 +210,7 @@ public void initAll(){ } } - public void initCaloDebug(){ + public void initCaloDebug(){ String[] names = new String[]{ "EC","EB" @@ -265,7 +288,7 @@ public void addEngine(String clazz) { /** * 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(), @@ -279,7 +302,16 @@ public void init(){ * 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); @@ -288,6 +320,7 @@ public void processEvent(DataEvent event){ e.printStackTrace(); } } + return event; } public void processFile(String file, String output){ @@ -296,14 +329,16 @@ public void processFile(String file, String output){ 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++; @@ -316,16 +351,14 @@ public void processFile(HipoDataSource reader, HipoDataSync writer, int skipEven 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) { @@ -371,11 +404,16 @@ public void show(){ } } - public static void main(String[] args){ + protected final void init(int config) { + if (config > 2) initCaloDebug(); + else if(config == 2) initAll(); + else initDefault(); + } + + protected static OptionParser parser() { 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"); @@ -386,71 +424,58 @@ public static void main(String[] args){ 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; + } - parser.parse(args); + protected final void init(OptionParser parser) { parser.syncLogLevel(LOGGER); + if (parser.getOption("-u").stringValue().contains("false")) + updateDictionary = false; - 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")) { - ClaraYaml yaml = new ClaraYaml(yamlFileName); - if (yaml.schemaDirectory() != null) { - proc.setBanksToKeep(yaml.schemaDirectory()); - } - for (JSONObject service : yaml.services()) { - JSONObject cfg = yaml.filter(service.getString("name")); - if (cfg.length() > 0) { - proc.addEngine(service.getString("name"),service.getString("class"),cfg.toString()); - } else { - proc.addEngine(service.getString("name"),service.getString("class")); - } - } + // read services and schema from YAML: + if (!parser.getOption("-y").stringValue().equals("0")) { + parseYaml(parser.getOption("-y").stringValue()); } - else if (config>0){ - if(config>2){ - proc.initCaloDebug(); - } else if(config==2){ - proc.initAll(); - } else { - proc.initDefault(); - } + // builtin configuration: + else if (parser.getOption("-c").intValue() > 0) { + init(parser.getOption("-c").intValue()); } + // user-defined services: else { - for(String engine : services){ + for(String engine : parser.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()); + setBanksToKeep(parser.getOption("-S").stringValue()); // command-line filename for background merging overrides YAML: if (parser.getOption("-B").stringValue() != null) - proc.setBackgroundFiles(parser.getOption("-B").stringValue()); + setBackgroundFiles(parser.getOption("-B").stringValue()); // command-line filename for post-processing overrides YAML: if (parser.getOption("-P").stringValue() != null) { - proc.setPreloadFiles(parser.getOption("-P").stringValue(), + setPreloadFiles(parser.getOption("-P").stringValue(), parser.getOption("-H").intValue()!=0, parser.getOption("-R").intValue()!=0); } + } + + public static void main(String[] args) { + + OptionParser parser = EngineProcessor.parser(); + parser.parse(args); + + EngineProcessor proc = new EngineProcessor(parser); - proc.processFile(inputFile,outputFile,nskip,nevents); + 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..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