Skip to content

Commit 6e56e49

Browse files
chiarazampollishahor02
chiarazampolli
authored and
shahor02
committed
Template of simple moving average.
1 parent 3c20540 commit 6e56e49

5 files changed

Lines changed: 97 additions & 19 deletions

File tree

‎Detectors/DCS/CMakeLists.txt‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@
88
# granted to it by virtue of its status as an Intergovernmental Organization or
99
# submit itself to any jurisdiction.
1010

11-
1211
o2_add_library(DetectorsDCS
12+
TARGETVARNAME targetName
1313
SOURCES src/Clock.cxx
1414
src/DataPointCompositeObject.cxx
1515
src/DataPointIdentifier.cxx
@@ -33,3 +33,8 @@ o2_add_executable(dcs-data-workflow
3333
SOURCES testWorkflow/dcs-data-workflow.cxx
3434
PUBLIC_LINK_LIBRARIES O2::Framework
3535
O2::DetectorsDCS)
36+
37+
if (OpenMP_CXX_FOUND)
38+
target_compile_definitions(${targetName} PRIVATE WITH_OPENMP)
39+
target_link_libraries(${targetName} PRIVATE OpenMP::OpenMP_CXX)
40+
endif()

‎Detectors/DCS/include/DetectorsDCS/DCSProcessor.h‎

Lines changed: 39 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@
2121
#include "DetectorsDCS/DataPointValue.h"
2222
#include "DetectorsDCS/DeliveryType.h"
2323

24+
//#ifdef WITH_OPENMP
25+
//#include <omp.h>
26+
//#endif
27+
2428
/// @brief Class to process DCS data points
2529

2630
namespace o2
@@ -68,6 +72,9 @@ class DCSProcessor
6872
template <typename T>
6973
int processArrayType(const std::vector<DPID>& array, DeliveryType type, const std::unordered_map<DPID, DPVAL>& map, std::vector<uint64_t>& latestTimeStamp, std::unordered_map<DPID, T>& destmap);
7074

75+
template <typename T>
76+
void doSimpleMovingAverage(int nelements, std::deque<T>& vect, float& avg, bool& isSMA);
77+
7178
virtual void processChars();
7279
virtual void processInts();
7380
virtual void processDoubles();
@@ -78,8 +85,6 @@ class DCSProcessor
7885
virtual void processBinaries();
7986
virtual uint64_t processFlag(uint64_t flag, const char* alias);
8087

81-
void doSimpleMovingAverage(int nelements, std::deque<int>& vect, float& avg, bool& isSMA);
82-
8388
DQChars& getVectorForAliasChar(const DPID& id) { return mDpscharsmap[id]; }
8489
DQInts& getVectorForAliasInt(const DPID& id) { return mDpsintsmap[id]; }
8590
DQDoubles& getVectorForAliasDouble(const DPID& id) { return mDpsdoublesmap[id]; }
@@ -88,9 +93,13 @@ class DCSProcessor
8893
DQStrings& getVectorForAliasString(const DPID& id) { return mDpsstringsmap[id]; }
8994
DQTimes& getVectorForAliasTime(const DPID& id) { return mDpstimesmap[id]; }
9095
DQBinaries& getVectorForAliasBinary(const DPID& id) { return mDpsbinariesmap[id]; }
91-
96+
97+
void setNThreads(int n);
98+
int getNThreads() const { return mNThreads; }
99+
92100
private:
93-
std::vector<float> mAvgTestInt; // moving average for DP named TestInt0
101+
std::vector<float> mAvgTestInt; // moving average for int DPs
102+
std::vector<float> mAvgTestDouble; // moving average for double DPs
94103
std::unordered_map<DPID, DQChars> mDpscharsmap;
95104
std::unordered_map<DPID, DQInts> mDpsintsmap;
96105
std::unordered_map<DPID, DQDoubles> mDpsdoublesmap;
@@ -115,7 +124,8 @@ class DCSProcessor
115124
std::vector<uint64_t> mLatestTimestampstrings;
116125
std::vector<uint64_t> mLatestTimestamptimes;
117126
std::vector<uint64_t> mLatestTimestampbinaries;
118-
127+
int mNThreads = 1; // number of threads
128+
119129
ClassDefNV(DCSProcessor, 0);
120130
};
121131

@@ -140,6 +150,10 @@ int DCSProcessor::processArrayType(const std::vector<DPID>& array, DeliveryType
140150
int found = 0;
141151
auto s = array.size();
142152
if (s > 0) {
153+
//#ifdef WITH_OPENMP
154+
//omp_set_num_threads(mNThreads);
155+
//#pragma omp parallel for schedule(dynamic)
156+
//#endif
143157
for (size_t i = 0; i != s; ++i) {
144158
auto it = processAlias(array[i], type, map);
145159
if (it == map.end()) {
@@ -170,6 +184,26 @@ int DCSProcessor::processArrayType(const std::vector<DCSProcessor::DPID>& array,
170184
template <>
171185
int DCSProcessor::processArrayType(const std::vector<DCSProcessor::DPID>& array, DeliveryType type, const std::unordered_map<DCSProcessor::DPID, DCSProcessor::DPVAL>& map, std::vector<uint64_t>& latestTimeStamp, std::unordered_map<DCSProcessor::DPID, DCSProcessor::DQBinaries>& destmap);
172186

187+
template <typename T>
188+
void DCSProcessor::doSimpleMovingAverage(int nelements, std::deque<T>& vect, float& avg, bool& isSMA) {
189+
190+
// Do simple moving average on vector of type T
191+
192+
if (vect.size() < nelements) {
193+
avg += vect[vect.size() - 1];
194+
return;
195+
}
196+
if (vect.size() == nelements) {
197+
avg += vect[vect.size() - 1];
198+
avg /= nelements;
199+
isSMA = true;
200+
return;
201+
}
202+
avg += (vect[vect.size() - 1] - vect[0]) / nelements;
203+
vect.pop_front();
204+
isSMA = true;
205+
}
206+
173207
} // namespace dcs
174208
} // namespace o2
175209

‎Detectors/DCS/src/DCSProcessor.cxx‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -349,6 +349,7 @@ void DCSProcessor::processInts()
349349
}
350350
bool isSMA = false;
351351
LOG(DEBUG) << "get alias = " << id.get_alias();
352+
// I do the moving average always of the last 2 points, no matter if it was updated or not
352353
doSimpleMovingAverage(2, vint, mAvgTestInt[i], isSMA);
353354
LOG(DEBUG) << "Moving average = " << mAvgTestInt[i];
354355
if (isSMA) {
@@ -358,7 +359,7 @@ void DCSProcessor::processInts()
358359
}
359360

360361
//______________________________________________________________________
361-
362+
/*
362363
void DCSProcessor::doSimpleMovingAverage(int nelements, std::deque<int>& vect, float& avg, bool& isSMA)
363364
{
364365
@@ -377,7 +378,7 @@ void DCSProcessor::doSimpleMovingAverage(int nelements, std::deque<int>& vect, f
377378
vect.pop_front();
378379
isSMA = true;
379380
}
380-
381+
*/
381382
//______________________________________________________________________
382383

383384
void DCSProcessor::processDoubles()
@@ -485,3 +486,17 @@ uint64_t DCSProcessor::processFlag(const uint64_t flags, const char* alias)
485486

486487
return 0;
487488
}
489+
490+
//______________________________________________________________________
491+
492+
void DCSProcessor::setNThreads(int n) {
493+
494+
// to set number of threads used to process the DPs
495+
496+
#ifdef WITH_OPENMP
497+
mNThreads = n > 0 ? n : 1;
498+
#else
499+
LOG(WARNING) << " Multithreading is not supported, imposing single thread";
500+
mNThreads = 1;
501+
#endif
502+
}

‎Detectors/DCS/testWorkflow/DCSDataGeneratorSpec.h‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#include <unistd.h>
1717
#include <TRandom.h>
1818
#include <TDatime.h>
19+
#include <TStopwatch.h>
1920
#include "DetectorsDCS/DataPointIdentifier.h"
2021
#include "DetectorsDCS/DataPointValue.h"
2122
#include "DetectorsDCS/DataPointCompositeObject.h"
@@ -87,10 +88,11 @@ class DCSDataGenerator : public o2::framework::Task
8788
void run(o2::framework::ProcessingContext& pc) final
8889
{
8990

91+
TStopwatch s;
9092
uint64_t tfid;
9193
for (auto& input : pc.inputs()) {
9294
tfid = header::get<o2::framework::DataProcessingHeader*>(input.header)->startTime;
93-
LOG(INFO) << "tfid = " << tfid;
95+
LOG(DEBUG) << "tfid = " << tfid;
9496
if (tfid >= mMaxTF) {
9597
LOG(INFO) << "Data generator reached TF " << tfid << ", stopping";
9698
pc.services().get<o2::framework::ControlService>().endOfStream();
@@ -99,6 +101,8 @@ class DCSDataGenerator : public o2::framework::Task
99101
break; // we break because one input is enough to get the TF ID
100102
}
101103

104+
LOG(INFO) << "TF: " << tfid << " --> building binary blob...";
105+
s.Start();
102106
uint16_t flags = 0;
103107
uint16_t milliseconds = 0;
104108
TDatime currentTime;
@@ -113,16 +117,16 @@ class DCSDataGenerator : public o2::framework::Task
113117
DPVAL valdouble(flags, milliseconds + tfid * 10, seconds + tfid, payload, mtypedouble);
114118
DPVAL valstring(flags, milliseconds + tfid * 10, seconds + tfid, payload, mtypestring);
115119

116-
LOG(INFO) << "Value used for char DPs:";
117-
LOG(INFO) << valchar << " --> " << (char)valchar.payload_pt1;
118-
LOG(INFO) << "Value used for int DPs:";
119-
LOG(INFO) << valint << " --> " << (int)valint.payload_pt1;
120-
LOG(INFO) << "Value used for double DPs:";
121-
LOG(INFO) << valdouble << " --> " << (double)valdouble.payload_pt1;
120+
LOG(DEBUG) << "Value used for char DPs:";
121+
LOG(DEBUG) << valchar << " --> " << (char)valchar.payload_pt1;
122+
LOG(DEBUG) << "Value used for int DPs:";
123+
LOG(DEBUG) << valint << " --> " << (int)valint.payload_pt1;
124+
LOG(DEBUG) << "Value used for double DPs:";
125+
LOG(DEBUG) << valdouble << " --> " << (double)valdouble.payload_pt1;
122126
char tt[56];
123127
memcpy(&tt[0], &valstring.payload_pt1, 56);
124-
LOG(INFO) << "Value used for string DPs:";
125-
LOG(INFO) << valstring << " --> " << tt;
128+
LOG(DEBUG) << "Value used for string DPs:";
129+
LOG(DEBUG) << valstring << " --> " << tt;
126130

127131
std::vector<DPCOM> dpcomVect;
128132
for (int i = 0; i < mNumDPschar; i++) {
@@ -139,7 +143,7 @@ class DCSDataGenerator : public o2::framework::Task
139143
}
140144

141145
auto svect = dpcomVect.size();
142-
LOG(INFO) << "dpcomVect has size " << svect;
146+
LOG(DEBUG) << "dpcomVect has size " << svect;
143147
for (int i = 0; i < svect; i++) {
144148
LOG(DEBUG) << "i = " << i << ", DPCOM = " << dpcomVect[i];
145149
}
@@ -150,7 +154,13 @@ class DCSDataGenerator : public o2::framework::Task
150154
}
151155
auto sbuff = buff.size();
152156
LOG(DEBUG) << "size of output buffer = " << sbuff;
157+
s.Stop();
158+
LOG(INFO) << "TF: " << tfid << " --> ...binary blob prepared: realTime = " << s.RealTime() << ", cpuTime = " << s.CpuTime();
159+
LOG(INFO) << "TF: " << tfid << " --> sending snapshot...";
160+
s.Start();
153161
pc.outputs().snapshot(Output{"DCS", "DATAPOINTS", 0, Lifetime::Timeframe}, buff.data(), sbuff);
162+
s.Stop();
163+
LOG(INFO) << "TF: " << tfid << " --> ...snapshot sent: realTime = " << s.RealTime() << ", cpuTime = " << s.CpuTime();
154164

155165
/*
156166
LOG(INFO) << "Reading back";

‎Detectors/DCS/testWorkflow/DCSDataProcessorSpec.h‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
#include <unistd.h>
1818
#include <TRandom.h>
19+
#include <TStopwatch.h>
1920
#include "DetectorsDCS/DataPointIdentifier.h"
2021
#include "DetectorsDCS/DataPointValue.h"
2122
#include "DetectorsDCS/DataPointCompositeObject.h"
@@ -75,21 +76,34 @@ class DCSDataProcessor : public o2::framework::Task
7576

7677
void run(o2::framework::ProcessingContext& pc) final
7778
{
78-
auto tfcounter = o2::header::get<o2::framework::DataProcessingHeader*>(pc.inputs().get("input").header)->startTime;
79+
auto tfid = o2::header::get<o2::framework::DataProcessingHeader*>(pc.inputs().get("input").header)->startTime;
80+
TStopwatch s;
81+
LOG(INFO) << "TF: " << tfid << " --> receiving binary data...";
82+
s.Start();
7983
auto rawchar = pc.inputs().get<const char*>("input");
84+
s.Stop();
85+
LOG(INFO) << "TF: " << tfid << " --> ...binary data received: realTime = " << s.RealTime() << ", cpuTime = " << s.CpuTime();
8086
const auto* dh = o2::header::get<o2::header::DataHeader*>(pc.inputs().get("input").header);
8187
auto sz = dh->payloadSize;
8288
int nDPs = sz / sizeof(DPCOM);
8389
LOG(INFO) << "Number of DPs received = " << nDPs;
8490
std::unordered_map<DPID, DPVAL> dcsmap;
8591
DPCOM dptmp;
92+
LOG(INFO) << "TF: " << tfid << " --> building unordered_map...";
93+
s.Start();
8694
for (int i = 0; i < nDPs; i++) {
8795
memcpy(&dptmp, rawchar + i * sizeof(DPCOM), sizeof(DPCOM));
8896
dcsmap[dptmp.id] = dptmp.data;
8997
LOG(DEBUG) << "Reading from generator: i = " << i << ", DPCOM = " << dptmp;
9098
LOG(DEBUG) << "Reading from generator: i = " << i << ", DPID = " << dptmp.id;
9199
}
100+
s.Stop();
101+
LOG(INFO) << "TF: " << tfid << " --> ...unordered_map built = " << s.RealTime() << ", cpuTime = " << s.CpuTime();
102+
LOG(INFO) << "TF: " << tfid << " --> starting processing...";
103+
s.Start();
92104
mDCSproc.process(dcsmap);
105+
s.Stop();
106+
LOG(INFO) << "TF: " << tfid << " --> ...processing done: realTime = " << s.RealTime() << ", cpuTime = " << s.CpuTime();
93107
}
94108

95109
private:

0 commit comments

Comments
 (0)