feat: add decoder engine

common-tools/clara-io/src/main/java/org/jlab/io/clara/EvioToEvioReader.java

@@ -39,7 +39,7 @@ public int readEventCount() throws EventReaderException {
 
     @Override
     public ByteOrder readByteOrder() throws EventReaderException {
-        return reader.getFileByteOrder();
+        return ByteOrder.LITTLE_ENDIAN; //reader.getFileByteOrder();
     }
 
     @Override

common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoder.java

@@ -15,8 +15,6 @@
 
 import org.jlab.io.base.DataEvent;
 import org.jlab.io.evio.EvioDataEvent;
-import org.jlab.io.hipo.HipoDataEvent;
-import org.jlab.io.hipo.HipoDataSync;
 
 import org.jlab.jnp.hipo4.data.Bank;
 import org.jlab.jnp.hipo4.data.Event;
@@ -38,27 +36,28 @@ public class CLASDecoder {
     protected SchemaFactory           schemaFactory = new SchemaFactory();
     private CodaEventDecoder          codaDecoder = null;
     private List<DetectorDataDgtz>       dataList = new ArrayList<>();
-    private HipoDataSync                   writer = null;
-    private HipoDataEvent               hipoEvent = null;
     private boolean              isRunNumberFixed = false;
     private int                  decoderDebugMode = 0;
     private ModeAHDC                ahdcExtractor = new ModeAHDC();
-    private RCDBManager               rcdbManager = new RCDBManager();
+    private static RCDBManager        rcdbManager = new RCDBManager();
 
     public CLASDecoder(boolean development){
         codaDecoder = new CodaEventDecoder();
         detectorDecoder = new DetectorEventDecoder(development);
-        writer = new HipoDataSync();
-        hipoEvent = (HipoDataEvent) writer.createEvent();
         String dir = ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4");
         schemaFactory.initFromDirectory(dir);
     }
 
     public CLASDecoder(){
         codaDecoder = new CodaEventDecoder();
         detectorDecoder = new DetectorEventDecoder();
-        writer = new HipoDataSync();
-        hipoEvent = (HipoDataEvent) writer.createEvent();
+        String dir = ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4");
+        schemaFactory.initFromDirectory(dir);
+    }
+
+    public CLASDecoder(CLASDecoder d) {
+        codaDecoder = new CodaEventDecoder();
+        detectorDecoder = new DetectorEventDecoder(d.detectorDecoder);
         String dir = ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4");
         schemaFactory.initFromDirectory(dir);
     }
@@ -792,4 +791,8 @@ public Event getDecodedEvent(EvioDataEvent rawEvent, int run, int counter, Doubl
 
         return decodedEvent;
     }
+
+    public Event getDecodedEvent(EvioDataEvent rawEvent) {
+        return getDecodedEvent(rawEvent, -1, -1, null, null);
+    }
 }

common-tools/clas-detector/src/main/java/org/jlab/detector/decode/DetectorEventDecoder.java

@@ -18,6 +18,7 @@
  */
 public class DetectorEventDecoder {
 
+    boolean sharedManagers = false;
     ConstantsManager translationManager = new ConstantsManager();
     ConstantsManager fitterManager      = new ConstantsManager();
     ConstantsManager scalerManager      = new ConstantsManager();
@@ -45,6 +46,18 @@ public DetectorEventDecoder(boolean development){
         }
     }
 
+    public DetectorEventDecoder(){
+        this.initDecoder();
+    }
+
+    public DetectorEventDecoder(DetectorEventDecoder d) {
+        translationManager = d.translationManager;
+        fitterManager = d.fitterManager;
+        scalerManager = d.scalerManager;
+        sharedManagers = true;
+        initDecoder();
+    }
+
     public void setTimestamp(String timestamp) {
         translationManager.setTimeStamp(timestamp);
         fitterManager.setTimeStamp(timestamp);
@@ -80,10 +93,6 @@ public float getRcdbSolenoidScale() {
                 getValue()).floatValue();
     }
 
-    public DetectorEventDecoder(){
-        this.initDecoder();
-    }
-
     public final void initDecoderDev(){
         keysTrans = Arrays.asList(new DetectorType[]{ DetectorType.HTCC,DetectorType.BST,DetectorType.RTPC} );
         tablesTrans = Arrays.asList(new String[]{ "/daq/tt/clasdev/htcc","/daq/tt/clasdev/svt","/daq/tt/clasdev/rtpc" });
@@ -109,7 +118,6 @@ public final void initDecoder(){
             "/daq/tt/rf","/daq/tt/bmt","/daq/tt/fmt","/daq/tt/rich2","/daq/tt/hel","/daq/tt/band","/daq/tt/rtpc",
             "/daq/tt/raster","/daq/tt/atof","/daq/tt/ahdc"
         });
-        translationManager.init(tablesTrans);
 
         // ADC waveform fitter translation table
         keysFitter   = Arrays.asList(new DetectorType[]{DetectorType.FTCAL,DetectorType.FTHODO,DetectorType.FTTRK,DetectorType.FTOF,DetectorType.LTCC,
@@ -122,17 +130,20 @@ public final void initDecoder(){
             "/daq/config/fmt","/daq/fadc/hel","/daq/fadc/rf","/daq/fadc/band","/daq/fadc/raster",
             "/daq/config/ahdc"
         });
-        fitterManager.init(tablesFitter);
 
         // Data filter list
         keysFilter   = Arrays.asList(new DetectorType[]{DetectorType.DC});
-
-        scalerManager.init(Arrays.asList(new String[]{"/runcontrol/fcup","/runcontrol/slm","/runcontrol/hwp",
-                                                      "/runcontrol/helicity","/daq/config/scalers/dsc1"}));
-
+
         keysMicromega = Arrays.asList(new DetectorType[]{DetectorType.BMT,DetectorType.FMT,DetectorType.FTTRK});
 
-        checkTables();
+        if (!sharedManagers) {
+            translationManager.init(tablesTrans);
+            fitterManager.init(tablesFitter);
+            scalerManager.init(Arrays.asList(new String[]{"/runcontrol/fcup","/runcontrol/slm","/runcontrol/hwp",
+                                                          "/runcontrol/helicity","/daq/config/scalers/dsc1"}));
+            checkTables();
+        }
+
     }
 
     public void checkTables() {

common-tools/clas-reco/src/main/java/org/jlab/clas/reco/DecoderEngine.java

@@ -0,0 +1,107 @@
+package org.jlab.clas.reco;
+
+import java.util.Set;
+import java.util.HashSet;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import org.jlab.clara.base.ClaraUtil;
+import org.jlab.clara.engine.Engine;
+import org.jlab.clara.engine.EngineData;
+import org.jlab.clara.engine.EngineDataType;
+import org.jlab.clara.engine.EngineStatus;
+import org.jlab.detector.decode.CLASDecoder;
+import org.jlab.io.evio.EvioDataEvent;
+import org.jlab.io.hipo.HipoDataEvent;
+import org.jlab.jnp.hipo4.data.SchemaFactory;
+import org.json.JSONObject;
+
+/**
+ *
+ * @author baltzell
+ */
+public class DecoderEngine implements Engine {
+
+    static final int POOL_SIZE = 64;
+    static final Set<EngineDataType> ED_TYPES = ClaraUtil.buildDataTypes(
+        Clas12Types.EVIO,Clas12Types.HIPO,EngineDataType.JSON,EngineDataType.STRING);
+
+    SchemaFactory schema;
+    BlockingQueue<CLASDecoder> pool;
+    int constantsShared = 16;
+
+    public DecoderEngine() {
+        schema = new SchemaFactory();
+        schema.initFromDirectory(System.getenv("CLAS12DIR") + "/etc/bankdefs/hipo4");
+    }
+
+    @Override
+    public Set<EngineDataType> getInputDataTypes() { return ED_TYPES; }
+    @Override
+    public Set<EngineDataType> getOutputDataTypes() { return ED_TYPES; }
+    @Override
+    public EngineData executeGroup(Set<EngineData> set) { return null; }
+    @Override
+    public Set<String> getStates() { return new HashSet<>(); }
+    @Override
+    public String getDescription() { return "decoder engine"; }
+    @Override
+    public String getVersion() { return "1.0"; }
+    @Override
+    public String getAuthor() { return "baltzell"; }
+    @Override
+    public void reset() {}
+    @Override
+    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<POOL_SIZE; i++) {
+            CLASDecoder d;
+            if (i % constantsShared == 0) {
+                d0 = new CLASDecoder();
+                if (json.has("variation")) d0.setVariation(json.getString("variation"));
+                if (json.has("timestamp")) d0.setVariation(json.getString("timestamp"));
+                d = d0;
+            }
+            else {
+                d = new CLASDecoder(d0);
+            }
+            pool.add(d);
+        }
+        return ed;
+    }
+
+    @Override
+    public EngineData execute(EngineData input) {
+        EngineData output = input;
+        EvioDataEvent evio;
+        HipoDataEvent hipo;
+        try {
+            ByteBuffer bb = (ByteBuffer) input.getData();
+            //evio = new EvioDataEvent(bb.array(), bb.order());
+            evio = new EvioDataEvent(bb.array(), ByteOrder.LITTLE_ENDIAN);
+        } catch (Exception e) {
+            String msg = String.format("Error reading input event%n%n%s", ClaraUtil.reportException(e));
+            output.setStatus(EngineStatus.ERROR);
+            output.setDescription(msg);
+            return output;
+        }
+        try {
+            CLASDecoder d = pool.take();
+            hipo = new HipoDataEvent(d.getDecodedEvent(evio),schema);
+            pool.put(d);
+            output.setData("binary/data-hipo", hipo.getHipoEvent());
+        } catch (Exception e) {
+            String msg = String.format("Error processing input event%n%n%s", ClaraUtil.reportException(e));
+            output.setStatus(EngineStatus.ERROR);
+            output.setDescription(msg);
+            return output;
+        }
+        return output;
+    }
+}

etc/services/rgd-clarode.yml

@@ -19,12 +19,14 @@ configuration:
       outputBankPrefix: "HB"
 io-services:
   reader:
-    class: org.jlab.io.clara.DecoderReader
-    name: DecoderReader
+    class: org.jlab.io.clara.EvioToEvioReader
+    name: EvioReader
   writer:
     class: org.jlab.io.clara.DecoderWriter
     name: DecoderWriter
 services:
+  - class: org.jlab.clas.reco.DecoderEngine
+    name: DECO
   - class: org.jlab.service.ai.DCDenoiseEngine
     name: DCDN
   - class: org.jlab.clas.swimtools.MagFieldsEngine

libexec/scaling

@@ -6,7 +6,7 @@ def cli():
     cli.add_argument('-y','--yaml',   metavar='YAML',help='path to YAML file',required=True)
     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=4,8)',default='4,8')
-    cli.add_argument('-e','--events', metavar='#',help='events per thread (default=555)',default=555,type=int)
+    cli.add_argument('-e','--events', metavar='#',help='events per threads (default=2550)',default=555,type=int)
     cli.add_argument('-N','--numa',   metavar='#',help='NUMA socket (default=None, choices=[0,1])',default=None,type=int,choices=[0,1])
     cli.add_argument('datafile', help='input EVIO/HIPO data file')
     cfg = cli.parse_args()
@@ -58,7 +58,7 @@ def benchmark(cfg, threads, log):
     # add the run-clara command:
     cmd.extend([cfg.run_clara,
             '-c',cfg.clara,
-            '-n',str(cfg.events*int(threads)),
+            '-n',str(cfg.events),
             '-t',str(threads),
             '-l',
             '-y',cfg.yaml,

libexec/scaling.gpl

@@ -67,6 +67,7 @@ set key nobox inside top left
 f(x) = m*x
 m = 2
 
+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