Skip to content

Commit b5cbe3b

Browse files
baltzellbaltzell
committed
working state (with yaml)
1 parent 37f0766 commit b5cbe3b

2 files changed

Lines changed: 94 additions & 61 deletions

File tree

‎common-tools/clas-detector/src/main/java/org/jlab/detector/serial/SerialHoncho.java‎

Lines changed: 72 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,12 @@
11
package org.jlab.detector.serial;
22

3+
import java.util.Arrays;
4+
import java.util.Iterator;
5+
import java.util.List;
6+
import java.util.ListIterator;
37
import java.util.TreeMap;
48
import java.util.TreeSet;
9+
import java.util.stream.Collectors;
510
import org.jlab.detector.calib.utils.ConstantsManager;
611
import org.jlab.detector.decode.CLASDecoder;
712
import org.jlab.detector.helicity.HelicityBit;
@@ -12,6 +17,7 @@
1217
import org.jlab.detector.scalers.DaqScalersSequence;
1318
import org.jlab.jnp.hipo4.data.Bank;
1419
import org.jlab.jnp.hipo4.data.Event;
20+
import org.jlab.jnp.hipo4.data.Schema;
1521
import org.jlab.jnp.hipo4.data.SchemaFactory;
1622
import org.jlab.jnp.hipo4.io.HipoWriterSorted;
1723

@@ -23,87 +29,98 @@ public class SerialHoncho {
2329

2430
static final String[] TAG1BANKS = {"RUN::scaler","HEL::scaler","RAW::scaler","RAW::epics","HEL::flip","COAT::config"};
2531
SchemaFactory schema;
26-
Bank[] tag1banks;
27-
Bank runConfig; // FIXME: store Schema for banks;
28-
Bank helicityAdc;
32+
Schema[] tag1banks;
33+
Schema runConfig;
34+
Schema recEvent;
35+
Schema helScaler;
36+
Schema helicityAdc;
2937
ConstantsManager conman;
3038
TreeMap<Integer,Integer> eventUnix;
3139
HelicitySequence helicitySequence;
3240
TreeSet<HelicityState> helicities;
3341
DaqScalersSequence scalers;
34-
int run;
42+
int run = 0;
3543

3644
public SerialHoncho(SchemaFactory schema) {
3745
this.schema = schema;
3846
conman = new ConstantsManager();
3947
conman.init("/runcontrol/hwp","/runcontrol/helicity");
40-
runConfig = new Bank(schema.getSchema("RUN::config"));
41-
helicityAdc = new Bank(schema.getSchema("HEL::adc"));
48+
runConfig = schema.getSchema("RUN::config");
49+
recEvent = schema.getSchema("REC::Event");
50+
helicityAdc = schema.getSchema("HEL::adc");
51+
helScaler = schema.getSchema("HEL::scaler");
4252
scalers = new DaqScalersSequence(schema);
4353
helicities = new TreeSet<>();
4454
eventUnix = new TreeMap<>();
45-
tag1banks = new Bank[TAG1BANKS.length];
55+
tag1banks = new Schema[TAG1BANKS.length];
4656
for (int i=0; i<tag1banks.length; ++i)
47-
tag1banks[i] = new Bank(schema.getSchema(TAG1BANKS[i]));
57+
tag1banks[i] = schema.getSchema(TAG1BANKS[i]);
4858
}
4959

5060
public synchronized Event read(Event event) {
61+
Bank cfg = new Bank(runConfig);
62+
Bank hel = new Bank(helicityAdc);
5163
scalers.add(event);
52-
event.read(runConfig);
53-
event.read(helicityAdc);
54-
if (runConfig.getRows() > 0) {
55-
if (run <= 0 && runConfig.getInt("run", 0) > 0) {
56-
run = runConfig.getInt("run",0);
57-
helicitySequence = new HelicitySequenceDelayed(
58-
conman.getConstants(run, "/runcontrol/helicity").getIntValue("delay",0,0,0));
59-
}
60-
int unix = runConfig.getInt("unixtime",0);
61-
int evno = runConfig.getInt("event",0);
64+
event.read(cfg);
65+
event.read(hel);
66+
if (cfg.getRows() > 0) {
67+
if (run <= 0 && cfg.getInt("run", 0) > 0)
68+
run = cfg.getInt("run",0);
69+
int unix = cfg.getInt("unixtime",0);
70+
int evno = cfg.getInt("event",0);
6271
if (unix > 0 && evno > 0) eventUnix.put(evno, unix);
6372
}
64-
if (helicitySequence != null) {
65-
HelicityState state = HelicityState.createFromFadcBank(helicityAdc, runConfig, conman);
66-
helicities.add(state);
67-
helicitySequence.addState(state);
68-
}
69-
return CLASDecoder.createTaggedEvent(event, runConfig, tag1banks);
73+
helicities.add(HelicityState.createFromFadcBank(hel, cfg, conman));
74+
return CLASDecoder.createTaggedEvent(event, cfg, createTaggedBanks(tag1banks));
7075
}
71-
76+
7277
public void process(Event event) {
73-
Bank cfg = new Bank(schema.getSchema("RUN::config"));
74-
Bank evt = new Bank(schema.getSchema("REC::Event"));
78+
Bank cfg = new Bank(runConfig);
79+
Bank evt = new Bank(recEvent);
7580
event.read(cfg);
7681
event.read(evt);
7782
if (cfg.getRows() > 0) {
7883
processEventUnix(event, cfg);
7984
if (evt.getRows() > 0) {
8085
event.remove(evt.getSchema());
81-
processHelicity(event, cfg, evt);
86+
//processHelicity(event, cfg, evt);
8287
processScalers(cfg, evt);
8388
event.write(evt);
8489
}
8590
}
8691
}
8792

93+
public void prune() {
94+
// remove identical helicities in the stream:
95+
HelicityState prev = null;
96+
Iterator<HelicityState> iter = (ListIterator)helicities.iterator();
97+
while (iter.hasNext()) {
98+
HelicityState next = iter.next();
99+
if (prev != null && prev == next)
100+
helicities.remove(next);
101+
}
102+
scalers.clear((int)1e5);
103+
// trim helicities to 100 million events, ~1 run, ~1 GB:
104+
//while (helicities.size() < 1e8) helicities.pollFirst();
105+
}
106+
88107
public void finish(HipoWriterSorted writer) {
89-
writer.addEvent(getUnixEvent(runConfig),1);
90-
// FIXME: mark written flips and don't write them again
108+
Bank cfg = new Bank(runConfig, 1);
109+
cfg.putInt("run",0,run);
110+
writer.addEvent(getUnixEvent(cfg),1);
91111
helicitySequence.writeFlips(writer, 1);
92112
}
93113

94114
public void clear() {
95115
eventUnix.clear();
96116
helicities.clear();
97117
scalers.clear();
118+
helicitySequence = null;
98119
}
99120

100121
public DaqScalersSequence getScalers() {
101122
return scalers;
102123
}
103-
104-
public HelicitySequence getHelicitySequence() {
105-
return helicitySequence;
106-
}
107124

108125
public ConstantsManager getConstantsManager() {
109126
return conman;
@@ -117,14 +134,13 @@ public TreeSet<HelicityState> getHelicities() {
117134
return helicities;
118135
}
119136

120-
HelicitySequence createHelicitySequence() {
121-
HelicitySequence seq = new HelicitySequenceDelayed(
122-
conman.getConstants(run, "/runcontrol/helicity").getIntValue("delay",0,0,0));
123-
seq.addStream(helicities);
124-
return seq;
137+
public void updateHelicitySequence() {
138+
helicitySequence = new HelicitySequenceDelayed(
139+
conman.getConstants(run, "/runcontrol/helicity").getIntValue("delay",0,0,0));
140+
helicitySequence.addStream(helicities);
125141
}
126142

127-
Event getUnixEvent(Bank config) {
143+
Event getUnixEvent(Bank runConfig) {
128144
Bank unix = new Bank(schema.getSchema("RUN::unix"));
129145
unix.setRows(eventUnix.size());
130146
int row = 0;
@@ -134,7 +150,7 @@ Event getUnixEvent(Bank config) {
134150
row++;
135151
}
136152
Event e = new Event();
137-
e.write(config);
153+
e.write(runConfig);
138154
e.write(unix);
139155
return e;
140156
}
@@ -170,13 +186,22 @@ void processHelicity(Event event, Bank runConfig, Bank recEvent) {
170186
HelicityBit hbraw = helicitySequence.getHalfWavePlate() ? HelicityBit.getFlipped(hb) : hb;
171187
recEvent.putByte("helicity",0,hb.value());
172188
recEvent.putByte("helicityRaw",0,hbraw.value());
173-
Bank helScaler = new Bank(schema.getSchema("HEL::scaler"));
174-
event.read(helScaler);
175-
if (helScaler.getRows()>0) {
176-
event.remove(schema.getSchema("HEL::scaler"));
177-
SerialUtil.assignScalerHelicity(runConfig.getLong("timestamp",0), helScaler, helicitySequence);
178-
event.write(helScaler);
189+
Bank scaler = new Bank(helScaler);
190+
event.read(scaler);
191+
if (scaler.getRows()>0) {
192+
event.remove(helScaler);
193+
SerialUtil.assignScalerHelicity(runConfig.getLong("timestamp",0), scaler, helicitySequence);
194+
event.write(scaler);
179195
}
180196
}
181197

198+
static Bank[] createTaggedBanks(Schema[] tag1banks) {
199+
List<Bank> lbank = Arrays.asList(tag1banks).stream().map(s -> new Bank(s)).collect(Collectors.toList());
200+
ListIterator<Bank> ibank = lbank.listIterator();
201+
Bank[] banks = new Bank[lbank.size()];
202+
while (ibank.hasNext())
203+
banks[ibank.nextIndex()] = ibank.next();
204+
return banks;
205+
}
206+
182207
}

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

Lines changed: 22 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,8 @@
4444
*/
4545
final class ReconMutil {
4646

47+
boolean DEBUG = true;
48+
4749
// Performance parameters:
4850
final int BENCH_SECONDS = 30;
4951
final int EVENTS_PER_CHUNK = 100;
@@ -68,7 +70,7 @@ final class ReconMutil {
6870
ConcurrentLinkedQueue<CompletableFuture> procThreads = new ConcurrentLinkedQueue<>();
6971

7072
// Queues:
71-
ConcurrentLinkedQueue<List<Object>> readQueue = new ConcurrentLinkedQueue<>();
73+
ConcurrentLinkedQueue<List<Object>> decoQueue = new ConcurrentLinkedQueue<>();
7274
ConcurrentLinkedQueue<List<HipoDataEvent>> procQueue = new ConcurrentLinkedQueue<>();
7375
ConcurrentLinkedQueue<List<Event>> writeQueue = new ConcurrentLinkedQueue<>();
7476
boolean paused = false;
@@ -117,10 +119,12 @@ void launch(int[] threads, String output, String... input) {
117119
while (!writerThread.isDone()) {
118120
sleep(1000);
119121

120-
//System.out.println(String.format("recon-mutil:: read(%b)/[deco(%d)]/proc(%d)/tag/write(%b)",
121-
// readerThread.isDone(), decoThreads.size(), procThreads.size(), writerThread.isDone()));
122-
//System.out.println(String.format("recon-util:: %d-%d/%d/%d/%d", readEvents,
123-
// readQueue.size(), procQueue.size(), taggedEvents.get(), writeQueue.size()));
122+
if (DEBUG){
123+
System.out.println(String.format("recon-mutil:: read(%b)/[deco(%d)]/proc(%d)/tag/write(%b)",
124+
readerThread.isDone(), decoThreads.size(), procThreads.size(), writerThread.isDone()));
125+
System.out.println(String.format("recon-util:: %d-%d/%d/%d/%d", readEvents,
126+
decoQueue.size(), procQueue.size(), taggedEvents.get(), writeQueue.size()));
127+
}
124128

125129
// cleanup completed parallel threads:
126130
for (CompletableFuture f : decoThreads)
@@ -172,7 +176,7 @@ void read(int threads, String... input) {
172176
// write leftover, partial chunk:
173177
if (!output.isEmpty()) {
174178
readEvents += output.size();
175-
readQueue.offer(output);
179+
decoQueue.offer(output);
176180
}
177181

178182
if (reader instanceof EvioSource evio) evio.close();
@@ -184,10 +188,9 @@ void read(int threads, String... input) {
184188
*/
185189
void decode(int thread) {
186190
while (true) {
187-
List<Object> input = readQueue.poll();
191+
List<Object> input = decoQueue.poll();
188192
if (input == null) {
189-
if (readerThread.isDone() && readQueue.isEmpty() &&
190-
writeEvents+skipEvents+failEvents >= readEvents) break;
193+
if (decoQueue.isEmpty() && readerThread.isDone() && decoQueue.isEmpty()) break;
191194
sleep(100);
192195
}
193196
else {
@@ -220,6 +223,10 @@ void process(int thread) {
220223
sleep(100);
221224
continue;
222225
}
226+
if (procQueue.isEmpty() && decoThreads.isEmpty() && procQueue.isEmpty()) {
227+
if (writeEvents+skipEvents+failEvents >= readEvents) break;
228+
sleep(100);
229+
}
223230
List<HipoDataEvent> input = procQueue.poll();
224231
if (input == null) {
225232
if (decoThreads.isEmpty() && procQueue.isEmpty() &&
@@ -237,6 +244,10 @@ void process(int thread) {
237244
Benchmark.getInstance().pause(engine.getValue().getName());
238245
}
239246
Event e = input.get(i).getHipoEvent();
247+
Benchmark.getInstance().resume("post");
248+
serial.process(e);
249+
serial.process(e);
250+
Benchmark.getInstance().pause("post");
240251
output.add(e);
241252
}
242253
writeQueue.offer(output);
@@ -261,9 +272,6 @@ void write(String output) {
261272
}
262273
else {
263274
for (int i=0; i<e.size(); i++) {
264-
Benchmark.getInstance().resume("post");
265-
serial.process(e.get(i));
266-
Benchmark.getInstance().pause("post");
267275
Benchmark.getInstance().resume("write");
268276
if (writer != null) {
269277
if (e.get(i).getEventTag() > 0 || schemaBankList.isEmpty())
@@ -401,7 +409,7 @@ List<Object> read(List<Object> chunk) {
401409
if (o != null && (skipEvents < 1 || readEvents > skipEvents)) {
402410
chunk.add(o);
403411
if (chunk.size() >= EVENTS_PER_CHUNK) {
404-
readQueue.offer(chunk);
412+
decoQueue.offer(chunk);
405413
readEvents += chunk.size();
406414
chunk = new ArrayList<>(EVENTS_PER_CHUNK);
407415
}
@@ -431,7 +439,7 @@ void reset() {
431439
writerThread.cancel(true);
432440
close();
433441
}
434-
readQueue = new ConcurrentLinkedQueue<>();
442+
decoQueue = new ConcurrentLinkedQueue<>();
435443
writeQueue = new ConcurrentLinkedQueue<>();
436444
procThreads = new ConcurrentLinkedQueue();
437445
readEvents = 0;

0 commit comments

Comments
 (0)