diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0ca4817d84..a20b1924bc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -227,12 +227,12 @@ jobs: tar xzf clara.tar.gz tar xzf coatjava.tar.gz - run: ls - - name: run test - run: ./bin/run-clara -y ./etc/services/rgd-clarode.yml -t 4 -n 500 -c ./clara -o ./tmp ./clas_018779.evio.00001 + - name: run clara + run: ./coatjava/bin/run-clara -y ./etc/services/rgd-clarode.yml -t 4 -n 100 -c ./clara -o ./tmp ./clas_018779.evio.00001 + - name: run mutil + run: ./coatjava/bin/recon-mutil -y ./etc/services/rgd-clarode.yml -t 4 -n 30 -o rec.hipo ./clas_018779.evio.00001 - name: ls tmp - run: ls -lhtr tmp - - name: rename - run: mv -v tmp/rec_clas_018779.evio.00001.hipo rec.hipo + run: ls -lhtr . tmp - uses: actions/upload-artifact@v7 with: name: test_clara_result diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index d42ca40d5f..a74a9cbd3c 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -175,6 +175,14 @@ clara: - run-clara -v -c $CLARA_HOME -t 4 -y ./etc/services/rgd-clarode.yml -n 30 -o out $EVIOFILE - mv out/rec_$EVIOFILE.hipo claroded.hipo +recon-mutil: + stage: test + needs: [build,download] + dependencies: [build,download] + script: + - tar -xzf coatjava.tar.gz + - recon-mutil -t 4 -y ./etc/services/rgd-clarode.yml -n 30 -o rec_$EVIOFILE.hipo $EVIOFILE + profile: extends: .clon allow_failure: true diff --git a/bin/recon-mutil b/bin/recon-mutil new file mode 100755 index 0000000000..3aaec2480b --- /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.ReconMutil \ + ${class_options[@]} diff --git a/common-tools/clara-io/pom.xml b/common-tools/clara-io/pom.xml index f1c3f6f0e1..3c9ddb5948 100644 --- a/common-tools/clara-io/pom.xml +++ b/common-tools/clara-io/pom.xml @@ -36,12 +36,6 @@ jnp-hipo4 - - org.jlab.clas - clas-io - 14.2.0-SNAPSHOT - - org.jlab.clas clas-detector @@ -50,7 +44,7 @@ org.jlab.clas - clas-utils + clas-io 14.2.0-SNAPSHOT diff --git a/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java b/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java index e2d5aaa7a0..5d5a917add 100644 --- a/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java +++ b/common-tools/clara-io/src/main/java/org/jlab/io/clara/Clas12Writer.java @@ -1,22 +1,11 @@ package org.jlab.io.clara; -import java.io.File; import java.nio.file.Path; -import java.util.List; -import java.util.TreeMap; -import java.util.TreeSet; import org.jlab.clara.std.services.EventWriterException; import org.jlab.detector.calib.utils.ConstantsManager; -import org.jlab.detector.decode.CLASDecoder4; -import org.jlab.detector.helicity.HelicitySequence; -import org.jlab.detector.helicity.HelicitySequenceDelayed; -import org.jlab.detector.helicity.HelicityState; -import org.jlab.detector.scalers.DaqScalersSequence; -import org.jlab.detector.serial.PostProcessor; -import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.detector.serial.SerialHoncho; import org.jlab.jnp.hipo4.data.Event; import org.jlab.jnp.hipo4.data.SchemaFactory; -import org.jlab.jnp.hipo4.io.HipoReader; import org.jlab.jnp.hipo4.io.HipoWriterSorted; import org.jlab.jnp.utils.file.FileUtils; import org.json.JSONObject; @@ -26,41 +15,24 @@ * 1. Copies certain banks on-the-fly to new tag-1 events * 2. Caches helicity states, scaler readouts, and unix time * 3. Writes HEL::flip, RUN/HEL::scaler, and RUN::unix to new tag-1 events - * 4. Runs post-processing, writing tag-1 information to all events * 5. Adds .hipo to the output filename, if necessary * * @author baltzell */ public class Clas12Writer extends HipoToHipoWriter { - static final String[] TAG1BANKS = {"RUN::scaler","HEL::scaler","RAW::scaler","RAW::epics","HEL::flip","COAT::config"}; - - Bank[] tag1banks; - Bank runConfig; - Bank helicityAdc; + SerialHoncho serial; ConstantsManager conman; - TreeMap eventUnix; - TreeSet helicities; - DaqScalersSequence scalers; SchemaFactory fullSchema; - boolean postprocess; private void init(JSONObject opts) { fullSchema = new SchemaFactory(); fullSchema.initFromDirectory(FileUtils.getEnvironmentPath("CLAS12DIR","etc/bankdefs/hipo4")); - runConfig = new Bank(fullSchema.getSchema("RUN::config")); - helicityAdc = new Bank(fullSchema.getSchema("HEL::adc")); - helicities = new TreeSet<>(); - scalers = new DaqScalersSequence(fullSchema); + serial = new SerialHoncho(fullSchema); conman = new ConstantsManager(); - eventUnix = new TreeMap<>(); conman.init("/runcontrol/hwp","/runcontrol/helicity"); - postprocess = opts.optBoolean("postprocess", false); if (opts.has("variation")) conman.setVariation(opts.getString("variation")); if (opts.has("timestamp")) conman.setTimeStamp(opts.getString("timestamp")); - tag1banks = new Bank[TAG1BANKS.length]; - for (int i=0; i 0) { - int unix = runConfig.getInt("unixtime",0); - int evno = runConfig.getInt("event",0); - if (unix > 0 && evno > 0) eventUnix.put(evno, unix); - } - helicities.add(HelicityState.createFromFadcBank(helicityAdc, runConfig, conman)); - Event t = CLASDecoder4.createTaggedEvent((Event)event, runConfig, tag1banks); + Event t = serial.read((Event)event); if (!t.isEmpty()) writer.addEvent(t, 1); super.writeEvent(event); } @Override protected void closeWriter() { - HelicitySequence.writeFlips(fullSchema, writer, helicities); - writer.addEvent(getUnixEvent(runConfig),1); + serial.finish(writer); super.closeWriter(); - if (postprocess) postprocess(); - // keep the latest helicity/scaler reading for the next file: - while (helicities.size() > 60) helicities.pollFirst(); - scalers.clear(10); - } - - /** - * Get the first valid run number from a RUN::config bank. - * @return run - */ - private int getRunNumber() { - Event e = new Event(); - HipoReader r = new HipoReader(); - r.open(filename); - while (r.hasNext()) { - r.nextEvent(e); - e.read(runConfig); - if (runConfig.getRows()>0 && runConfig.getInt("run",0)>0) - return runConfig.getInt("run",0); - } - return 0; + serial.clear(); } - - /** - * Get a new event with a RUN::unix bank containing event-timestamp mapping, - * and the latest RUN::config bank. - * @param config - * @return - */ - private Event getUnixEvent(Bank config) { - Bank unix = new Bank(fullSchema.getSchema("RUN::unix")); - unix.setRows(eventUnix.size()); - int row = 0; - for (int evno : eventUnix.keySet()) { - unix.putInt("event", row, evno); - unix.putInt("unixtime",row, eventUnix.get(evno)); - row++; - } - Event e = new Event(); - e.write(config); - e.write(unix); - return e; - } - - /** - * Copy helicity/charge tag-1 information to all events. - */ - private void postprocess() { - int d = conman.getConstants(getRunNumber(), "/runcontrol/helicity").getIntValue("delay",0,0,0); - HelicitySequenceDelayed helicity = new HelicitySequenceDelayed(d); - helicity.addStream(helicities); - PostProcessor p = new PostProcessor(List.of(filename), fullSchema, helicity, scalers); - HipoReader r = new HipoReader(); - r.open(filename); - Event e = new Event(); - writer.open("pp_"+filename); - while (r.hasNext()) { - r.nextEvent(e); - p.processEvent(e); - HipoToHipoWriter.writeEvent(writer, e, schemaBankList); - } - writer.close(); - new File(filename).delete(); - new File("pp_"+filename).renameTo(new File(filename)); - } - } diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoderPool.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoderPool.java index 80430583ee..a28135d906 100644 --- a/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoderPool.java +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/decode/CLASDecoderPool.java @@ -34,11 +34,11 @@ public CLASDecoderPool(int size, String variation, String timestamp) { } } - public CLASDecoder take() throws InterruptedException { + public CLASDecoder take() { return pool.poll(); } - public void put(CLASDecoder decoder) throws InterruptedException { + public void put(CLASDecoder decoder) { pool.offer(decoder); } diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/PostProcessor.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/PostProcessor.java index 0d0f1b96bc..a245029e17 100644 --- a/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/PostProcessor.java +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/PostProcessor.java @@ -231,6 +231,22 @@ public void processEvent(Event event) { } } + public void processFile(String input, String output) { + Event event = new Event(); + HipoReader r = new HipoReader(); + r.open(input); + HipoWriterSorted w = new HipoWriterSorted(); + w.getSchemaFactory().initFromDirectory(ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4")); + w.setCompressionType(2); + w.open(output); + while (r.hasNext()) { + r.nextEvent(event); + processEvent(event); + if (w != null) w.addEvent(event); + } + r.close(); + } + /** * The "postprocess" program. * @param args diff --git a/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/SerialHoncho.java b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/SerialHoncho.java new file mode 100644 index 0000000000..6fe28171ef --- /dev/null +++ b/common-tools/clas-detector/src/main/java/org/jlab/detector/serial/SerialHoncho.java @@ -0,0 +1,182 @@ +package org.jlab.detector.serial; + +import java.util.TreeMap; +import java.util.TreeSet; +import org.jlab.detector.calib.utils.ConstantsManager; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.detector.helicity.HelicityBit; +import org.jlab.detector.helicity.HelicitySequence; +import org.jlab.detector.helicity.HelicitySequenceDelayed; +import org.jlab.detector.helicity.HelicityState; +import org.jlab.detector.scalers.DaqScalers; +import org.jlab.detector.scalers.DaqScalersSequence; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.Event; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.jnp.hipo4.io.HipoWriterSorted; + +/** + * + * @author baltzell + */ +public class SerialHoncho { + + static final String[] TAG1BANKS = {"RUN::scaler","HEL::scaler","RAW::scaler","RAW::epics","HEL::flip","COAT::config"}; + SchemaFactory schema; + Bank[] tag1banks; + Bank runConfig; // FIXME: store Schema for banks; + Bank helicityAdc; + ConstantsManager conman; + TreeMap eventUnix; + HelicitySequence helicitySequence; + TreeSet helicities; + DaqScalersSequence scalers; + int run; + + public SerialHoncho(SchemaFactory schema) { + this.schema = schema; + conman = new ConstantsManager(); + conman.init("/runcontrol/hwp","/runcontrol/helicity"); + runConfig = new Bank(schema.getSchema("RUN::config")); + helicityAdc = new Bank(schema.getSchema("HEL::adc")); + scalers = new DaqScalersSequence(schema); + helicities = new TreeSet<>(); + eventUnix = new TreeMap<>(); + tag1banks = new Bank[TAG1BANKS.length]; + for (int i=0; i 0) { + if (run <= 0 && runConfig.getInt("run", 0) > 0) { + run = runConfig.getInt("run",0); + helicitySequence = new HelicitySequenceDelayed( + conman.getConstants(run, "/runcontrol/helicity").getIntValue("delay",0,0,0)); + } + int unix = runConfig.getInt("unixtime",0); + int evno = runConfig.getInt("event",0); + if (unix > 0 && evno > 0) eventUnix.put(evno, unix); + } + if (helicitySequence != null) { + HelicityState state = HelicityState.createFromFadcBank(helicityAdc, runConfig, conman); + helicities.add(state); + helicitySequence.addState(state); + } + return CLASDecoder.createTaggedEvent(event, runConfig, tag1banks); + } + + public void process(Event event) { + Bank cfg = new Bank(schema.getSchema("RUN::config")); + Bank evt = new Bank(schema.getSchema("REC::Event")); + event.read(cfg); + event.read(evt); + if (cfg.getRows() > 0) { + processEventUnix(event, cfg); + if (evt.getRows() > 0) { + event.remove(evt.getSchema()); + processHelicity(event, cfg, evt); + processScalers(cfg, evt); + event.write(evt); + } + } + } + + public void finish(HipoWriterSorted writer) { + writer.addEvent(getUnixEvent(runConfig),1); + // FIXME: mark written flips and don't write them again + helicitySequence.writeFlips(writer, 1); + } + + public void clear() { + eventUnix.clear(); + helicities.clear(); + scalers.clear(); + } + + public DaqScalersSequence getScalers() { + return scalers; + } + + public HelicitySequence getHelicitySequence() { + return helicitySequence; + } + + public ConstantsManager getConstantsManager() { + return conman; + } + + public SchemaFactory getSchemaFactory() { + return schema; + } + + public TreeSet getHelicities() { + return helicities; + } + + HelicitySequence createHelicitySequence() { + HelicitySequence seq = new HelicitySequenceDelayed( + conman.getConstants(run, "/runcontrol/helicity").getIntValue("delay",0,0,0)); + seq.addStream(helicities); + return seq; + } + + Event getUnixEvent(Bank config) { + Bank unix = new Bank(schema.getSchema("RUN::unix")); + unix.setRows(eventUnix.size()); + int row = 0; + for (int evno : eventUnix.keySet()) { + unix.putInt("event", row, evno); + unix.putInt("unixtime",row, eventUnix.get(evno)); + row++; + } + Event e = new Event(); + e.write(config); + e.write(unix); + return e; + } + + int getUnixTime(Bank runConfig) { + if (runConfig.getRows() < 1) { + Integer key = eventUnix.floorKey(runConfig.getInt("event",0)); + if (key != null) { + Integer unix = eventUnix.get(key); + if (unix != null) return unix; + } + } + return 0; + } + + void processEventUnix(Event event, Bank runConfig) { + int ut = getUnixTime(runConfig); + event.remove(runConfig.getSchema()); + runConfig.putInt("unixtime", 0, ut); + event.write(runConfig); + } + + void processScalers(Bank runConfig, Bank recEvent) { + DaqScalers ds = scalers.get(runConfig.getLong("timestamp", 0)); + if (ds != null) { + recEvent.putFloat("beamCharge",0, (float) ds.dsc2.getBeamChargeGated()); + recEvent.putDouble("liveTime",0,ds.dsc2.getLivetime()); + } + } + + void processHelicity(Event event, Bank runConfig, Bank recEvent) { + HelicityBit hb = helicitySequence.search(runConfig.getLong("timestamp", 0)); + HelicityBit hbraw = helicitySequence.getHalfWavePlate() ? HelicityBit.getFlipped(hb) : hb; + recEvent.putByte("helicity",0,hb.value()); + recEvent.putByte("helicityRaw",0,hbraw.value()); + Bank helScaler = new Bank(schema.getSchema("HEL::scaler")); + event.read(helScaler); + if (helScaler.getRows()>0) { + event.remove(schema.getSchema("HEL::scaler")); + SerialUtil.assignScalerHelicity(runConfig.getLong("timestamp",0), helScaler, helicitySequence); + event.write(helScaler); + } + } + +} diff --git a/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java new file mode 100644 index 0000000000..1ae7d07684 --- /dev/null +++ b/common-tools/clas-reco/src/main/java/org/jlab/clas/reco/ReconMutil.java @@ -0,0 +1,541 @@ +package org.jlab.clas.reco; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jlab.clara.engine.EngineData; +import org.jlab.clara.engine.EngineDataType; +import org.jlab.coda.jevio.EvioException; +import org.jlab.detector.decode.CLASDecoder; +import org.jlab.detector.decode.CLASDecoderPool; +import org.jlab.detector.serial.SerialHoncho; +import org.jlab.io.evio.EvioDataEvent; +import org.jlab.io.evio.EvioSource; +import org.jlab.io.hipo.HipoDataEvent; +import org.jlab.jnp.hipo4.data.Bank; +import org.jlab.jnp.hipo4.data.Event; +import org.jlab.jnp.hipo4.data.SchemaFactory; +import org.jlab.jnp.hipo4.io.HipoReader; +import org.jlab.jnp.hipo4.io.HipoWriterSorted; +import org.jlab.utils.ClaraYaml; +import org.jlab.utils.benchmark.Benchmark; +import org.jlab.utils.benchmark.ProgressPrintout; +import org.jlab.utils.options.OptionParser; +import org.jlab.utils.system.ClasUtilsFile; +import org.json.JSONObject; + +/** + * FIXME: add tagged bank counter for completino decision + * + * @author baltzell + */ +final class ReconMutil { + + // Performance parameters: + final int BENCH_SECONDS = 30; + final int CHUNKS_PER_QUEUE = 100; + final int EVENTS_PER_CHUNK = 100; + + // File I/O: + Object reader; + HipoWriterSorted writer; + List schemaBankList; + static final SchemaFactory schema = new SchemaFactory(); + static { schema.initFromDirectory(ClasUtilsFile.getResourceDir("CLAS12DIR","etc/bankdefs/hipo4")); } + + // Processors: + SerialHoncho serial; + CLASDecoderPool decoders = new CLASDecoderPool(64,"default",null); + Map engines = new LinkedHashMap<>(); + + // Threads: + CompletableFuture readerThread; + CompletableFuture writerThread; + CompletableFuture rethreadThread; + ConcurrentLinkedQueue decoThreads = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue procThreads = new ConcurrentLinkedQueue<>(); + + // Queues: + ConcurrentLinkedQueue> readQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> procQueue = new ConcurrentLinkedQueue<>(); + ConcurrentLinkedQueue> writeQueue = new ConcurrentLinkedQueue<>(); + boolean paused = false; + + // Static parameters: + int maxEvents; + int skipEvents; + ClaraYaml yaml; + OptionParser parser; + + // Progress counters: + int readEvents; + int writeEvents; + int failEvents; + int fileEvents; + int maxFileEvents; + AtomicInteger taggedEvents = new AtomicInteger(); + ProgressPrintout progress = new ProgressPrintout(); + + ReconMutil(OptionParser parser) { + init(parser); + } + + /** + * The thread launcher and collector. + * @param threads number of threads + * @param output name of output file to write + * @param input names of input files to read + */ + void launch(int[] threads, String output, String... input) { + + reset(); + + System.out.println(String.format("recon-mutil:: Spawning %d+++ Threads...",threads[0])); + + // spawn all the threads: + readerThread = CompletableFuture.runAsync(() -> { read(threads[0], input); }); + writerThread = CompletableFuture.runAsync(() -> { write(output); }); + for (int i=0; i { decode(j); })); + procThreads.offer(CompletableFuture.runAsync(() -> { process(j); })); + } + + // wait for the writer to be done: + while (!writerThread.isDone()) { + sleep(1000); + + //System.out.println(String.format("recon-mutil:: read(%b)/[deco(%d)]/proc(%d)/tag/write(%b)", + // readerThread.isDone(), decoThreads.size(), procThreads.size(), writerThread.isDone())); + //System.out.println(String.format("recon-util:: %d-%d/%d/%d/%d", readEvents, + // readQueue.size(), procQueue.size(), taggedEvents.get(), writeQueue.size())); + + // cleanup completed parallel threads: + for (CompletableFuture f : decoThreads) + if (f.isDone()) decoThreads.remove(f); + for (CompletableFuture f : procThreads) + if (f.isDone()) procThreads.remove(f); + + // perform scaling test: + if (threads.length > 1 && rethreadThread == null && writeEvents > 100) { + rethreadThread = CompletableFuture.runAsync(() -> { rethread(BENCH_SECONDS,threads); }); + rethreadThread.join(); + reset(); + } + } + } + + /** + * The reader thread. + * @param input input filenames + */ + void read(int threads, String... input) { + + // convert input filenames to a list: + List inputs = new ArrayList<>(Arrays.asList(input)); + + // initialize the event chunk: + List output = new ArrayList<>(EVENTS_PER_CHUNK); + + // loop over input events: + while ( (maxEvents < 1 || readEvents < maxEvents) && + (maxFileEvents < 1 || fileEvents < maxFileEvents) ) { + + if (reader != null) { + + // sleep instead of overfilling the read queue (100K events, ~2GB): + if (readEvents > 1e5) sleep(1000); + + // read next event into chunk, and fill queue if chunk full: + else output = read(output); + } + + // open the next input file: + else if (!inputs.isEmpty()) open(inputs.removeFirst()); + + // no more events to read: + else break; + } + + // write leftover, partial chunk: + if (!output.isEmpty()) { + readEvents += output.size(); + readQueue.offer(output); + } + + if (reader instanceof EvioSource evio) evio.close(); + } + + /** + * The decoder thread. + * @param thread thread number + */ + void decode(int thread) { + while (true) { + List input = readQueue.poll(); + if (input == null) { + if (readerThread.isDone() && readQueue.isEmpty() && + writeEvents+skipEvents+failEvents >= readEvents) break; + sleep(100); + } + else { + List output = new ArrayList<>(input.size()); + for (int i=0; i input = procQueue.poll(); + if (input == null) { + if (decoThreads.isEmpty() && procQueue.isEmpty() && + writeEvents+skipEvents+failEvents >= readEvents) break; + sleep(100); + } + else { + //if (rethreadThread != null && !rethreadThread.isDone()) readQueue.offer(o); + List output = new ArrayList<>(input.size()); + for (int i=0; i engine : engines.entrySet()) { + Benchmark.getInstance().resume(engine.getValue().getName()); + try { engine.getValue().processDataEvent(input.get(i)); } + catch (Exception ex) { ex.printStackTrace(); } + Benchmark.getInstance().pause(engine.getValue().getName()); + } + Event e = input.get(i).getHipoEvent(); + output.add(e); + } + writeQueue.offer(output); + } + } + } + + /** + * The writer thread. + * @param output output filename + */ + void write(String output) { + if (output != null) writer = open(output, yaml); + while (true) { + List e = writeQueue.poll(); + if (e == null) { + if (readerThread.isDone() && procThreads.isEmpty() && writeQueue.isEmpty()) { + close(); + break; + } + sleep(1000); + } + else { + for (int i=0; i 0 || schemaBankList.isEmpty()) + writer.addEvent(e.get(i), e.get(i).getEventTag()); + else + writer.addEvent(e.get(i).reduceEvent(schemaBankList), e.get(i).getEventTag()); + } + Benchmark.getInstance().pause("write"); + progress.updateStatus(); + } + writeEvents += e.size(); + } + } + } + + /** + * The rethreader thread. + * @param seconds delay before switching to next thread count + * @param threads thread counts to use + */ + void rethread(int seconds, int... threads) { + System.out.println("~~~~~~~~~ Rethreading Initiated ~~~~~~~~~"); + for (int i=0; i { process(k); })); + } + while (progress.getNumberOfCalls() < 100) sleep(1000); + sleep(seconds*1000); + System.out.println(String.format("\n~~~~~~~~~ Rethreading Count: %d ~~~~~~~~~\n",threads[i])); + System.out.println(progress.getUpdateString()); + System.out.println(Benchmark.getInstance()); + } + } + + /** + * Decode an event. + * @param bytes the 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"); + CLASDecoder d = decoders.take(); + HipoDataEvent hipo = d.getDecodedDataEvenet(evio); + decoders.put(d); + Benchmark.getInstance().pause("deco"); + return hipo; + } + + /** + * Open a new input HIPO/EVIO event file. + * @param filename + */ + void open(String filename) { + fileEvents = 0; + if (filename.endsWith(".hipo")) { + reader = new HipoReader(); + ((HipoReader)reader).open(filename); + maxFileEvents = ((HipoReader)reader).getEventCount(); + } + else { + reader = new EvioSource(); + ((EvioSource)reader).open(filename); + maxFileEvents = ((EvioSource)reader).getEventCount(); + } + } + + /** + * Open a new writer and initialize its schema. + * @param filename output filename + * @param yaml the configuration + */ + HipoWriterSorted open(String filename, ClaraYaml yaml) { + HipoWriterSorted w = new HipoWriterSorted(); + w.setCompressionType(2); + String d = ClasUtilsFile.getResourceDir("CLAS12DIR", "etc/bankdefs/hipo4"); + if (yaml != null && yaml.getSchemaDirectory() != null) d = yaml.getSchemaDirectory(); + if (!parser.getOption("-S").isDefault()) d = parser.getOption("-S").stringValue(); + SchemaFactory s = new SchemaFactory(); + s.initFromDirectory(d); + if (yaml != null) { + JSONObject json = yaml.filter("writer"); + if (json.has("wildcard")) { + SchemaFactory s2 = s.reduce(json.getString("wildcard")); + w.getSchemaFactory().copy(s2); + } + else w.getSchemaFactory().copy(s); + schemaBankList = new ArrayList<>(); + if (json.has("wildcard")) { + if (json.optBoolean("schema_filter",true)) { + int schemaSize = w.getSchemaFactory().getSchemaList().size(); + for (int i=0; i read(List chunk) { + Benchmark.getInstance().resume("read"); + Object o = null; + if (reader instanceof EvioSource evio) { + try { o = evio.getEventBuffer(++fileEvents, true); } + catch (EvioException ex) { + failEvents++; + ex.printStackTrace(); + } + } + else { + Event event = new Event(); + o = ((HipoReader)reader).getEvent(event, fileEvents); + } + if (o != null && (skipEvents < 1 || readEvents > skipEvents)) { + chunk.add(o); + if (chunk.size() >= EVENTS_PER_CHUNK) { + readQueue.offer(chunk); + readEvents += chunk.size(); + chunk = new ArrayList<>(EVENTS_PER_CHUNK); + } + } + Benchmark.getInstance().pause("read"); + return chunk; + } + + /** + * Close the output file. + */ + void close() { + serial.finish(writer); + writer.close(); + System.out.println(Benchmark.getInstance()); + System.out.println(String.format("recon-mutil :: read/write/tagged/diff = %d/%d/%d/%d", + readEvents, writeEvents, taggedEvents.get(), writeEvents-readEvents)); + } + + /** + * Forcefully shutdown all threads, close files, and reset queues and counters. + */ + void reset() { + for (CompletableFuture f : procThreads) f.cancel(true); + if (readerThread != null) readerThread.cancel(true); + if (writerThread != null) { + writerThread.cancel(true); + close(); + } + readQueue = new ConcurrentLinkedQueue<>(); + writeQueue = new ConcurrentLinkedQueue<>(); + procThreads = new ConcurrentLinkedQueue(); + readEvents = 0; + writeEvents = 0; + failEvents = 0; + taggedEvents.set(0); + } + + /** + * Add a new engine to the list. + * @param label display name + * @param clazz full class name + * @param cfg engine configuration + * @return + */ + ReconstructionEngine addEngine(String label, String clazz, JSONObject cfg) { + ReconstructionEngine engine = null; + try { + Class c = Class.forName(clazz); + if (ReconstructionEngine.class.isAssignableFrom(c)==true){ + engine = (ReconstructionEngine) c.newInstance(); + if (cfg != null && !cfg.toString().equals("null")) { + EngineData input = new EngineData(); + input.setData(EngineDataType.JSON.mimeType(), cfg.toString()); + engine.configure(input); + } + else engine.init(); + engines.put(label == null ? engine.getName() : label, engine); + } + else Logger.getLogger(ReconMutil.class.getPackage().getName()) + .log(clazz.contains("DecoderEngine") ? Level.INFO : Level.SEVERE, + "Class is not a reconstruction engine : {0}", clazz); + } catch (ClassNotFoundException | InstantiationException | IllegalAccessException ex) { + Logger.getLogger(ReconMutil.class.getPackage().getName()).log(Level.SEVERE, null, ex); + } + return engine; + } + + /** + * Catch interruptions in sleep. + * @param milliseconds + */ + void sleep(int milliseconds) { + try { Thread.sleep(milliseconds); } + catch (InterruptedException ex) {} + } + + void init(OptionParser parser) { + this.parser = parser; + parser.syncLogLevel(Logger.getLogger(ReconMutil.class.getPackage().getName())); + maxEvents = parser.getOption("-n").intValue(); + skipEvents = parser.getOption("-s").intValue(); + serial = new SerialHoncho(schema); + engines = new LinkedHashMap<>(); + if (!parser.getOption("-y").isDefault()) { + yaml = new ClaraYaml(parser.getOption("-y").stringValue()); + for (JSONObject service : yaml.services()) { + JSONObject cfg = yaml.filter(service.getString("name")); + if (cfg.length() > 0) addEngine(service.getString("name"), service.getString("class"), cfg); + else addEngine(service.getString("name"), service.getString("class"), null); + } + } + else if (!parser.getOption("-c").isDefault()) { + for (String s : parser.getOption("-c").stringValue().split(",")) + addEngine(null, s, null); + } + else { + InputStream is = ReconMutil.class.getClassLoader().getResourceAsStream("org/jlab/clas/reco/services.txt"); + BufferedReader br = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8)); + try { + for (String line; (line=br.readLine()) != null;) + addEngine(line.split(" ")[0],line.split(" ")[1],null); + } catch (IOException ex) { + System.getLogger(ReconMutil.class.getName()).log(System.Logger.Level.ERROR, (String) null, ex); + } + } + if (!parser.getOption("-B").isDefault()) { + ReconstructionEngine bg = addEngine("BG","org.jlab.service.bg.BackgroundEngine",null); + bg.engineConfigMap.put("filename",parser.getOption("-B").stringValue()); + } + if (!parser.getOption("-S").isDefault()) { + } + } + + /** + * The command-line entry-point known as "recon-mutil". + * @param args command-line arguments + */ + public static void main(String[] args) { + OptionParser o = EngineProcessor.getParser(); + o.removeOption("-i"); + o.removeOption("-o"); + o.removeOption("-c"); + o.addOption("-t","4","number of threads"); + o.addOption("-o", null, "output file name"); + o.addOption("-c","2","comma-separated engine list"); + o.setRequiresInputList(true); + o.parse(args); + ReconMutil r = new ReconMutil(o); + r.launch(Arrays.stream(o.getOption("-t").stringValue().split(",")).mapToInt(Integer::parseInt).toArray(), + o.getOption("-o").stringValue(), + o.getInputList().stream().toArray(String[]::new)); + } + +} diff --git a/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt b/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt new file mode 100644 index 0000000000..8a81bedfb6 --- /dev/null +++ b/common-tools/clas-reco/src/main/resources/org/jlab/clas/reco/services.txt @@ -0,0 +1,31 @@ +MAGFIELDS org.jlab.clas.swimtools.MagFieldsEngine +DENOISE org.jlab.service.ai.DCDenoiseEngine +FTCAL org.jlab.rec.ft.cal.FTCALEngine +FTHODO org.jlab.rec.ft.hodo.FTHODOEngine +FTTRK org.jlab.rec.ft.trk.FTTRKEngine +FTEB org.jlab.rec.ft.FTEBEngine +URWT org.jlab.service.urwt.URWTEngine +DCCR org.jlab.service.dc.DCHBClustering +DCHB org.jlab.service.dc.DCHBPostClusterConv +FTOFHB org.jlab.service.ftof.FTOFHBEngine +EC org.jlab.service.ec.ECEngine +RASTER org.jlab.service.raster.RasterEngine +CVT org.jlab.rec.cvt.services.CVTEngine +CTOF org.jlab.service.ctof.CTOFEngine +CND org.jlab.service.cnd.CNDCalibrationEngine +BAND org.jlab.service.band.BANDEngine +HTCC org.jlab.service.htcc.HTCCReconstructionService +LTCC org.jlab.service.ltcc.LTCCEngine +EBHB org.jlab.service.eb.EBHBEngine +DCTB org.jlab.service.dc.DCTBEngine +FMT org.jlab.service.fmt.FMTEngine +FTOFTB org.jlab.service.ftof.FTOFTBEngine +CVT org.jlab.rec.cvt.services.CVTSecondPassEngine +EBTB org.jlab.service.eb.EBTBEngine +RICHEB org.jlab.rec.rich.RICHEBEngine +RTPC org.jlab.service.rtpc.RTPCEngine +AHDC org.jlab.service.ahdc.AHDCEngine +ATOF org.jlab.service.atof.ATOFEngine +ALERT org.jlab.service.alert.ALERTEngine +MC org.jlab.service.mc.TruthMatch +VTX org.jlab.rec.service.vtx.VTXEngine diff --git a/validation/advanced-tests/run-eb-tests.sh b/validation/advanced-tests/run-eb-tests.sh index 05c7470db5..a21ad11c92 100755 --- a/validation/advanced-tests/run-eb-tests.sh +++ b/validation/advanced-tests/run-eb-tests.sh @@ -49,7 +49,8 @@ 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 INFO -o out_${stub}.hipo ${input_dir}/${stub}.hipo +exit # run EB tests: java -Xmx1536m -Xms1024m -cp $classPath -DINPUTFILE=out_${stub}.hipo eb.EBTwoTrackTest