make recon-util multi-threaded

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[@]}

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<String> inputs = new ArrayList<>();
+    ProgressPrintout progress = new ProgressPrintout();
+    ConcurrentLinkedQueue<CompletableFuture> procThreads = new ConcurrentLinkedQueue();
+    ConcurrentLinkedQueue<Object> readQueue = new ConcurrentLinkedQueue<>();
+    ConcurrentLinkedQueue<DataEvent> 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<threads; i++) {
+            final int j = i;
+            procThreads.offer(CompletableFuture.runAsync(() -> { 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<String,ReconstructionEngine> 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());
+    }
+}

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<String,ReconstructionEngine>  processorEngines = new LinkedHashMap<>();
+    protected final Map<String,ReconstructionEngine> 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<String> 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<String> 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<String> 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());
+    }
 }