Changeset 15514
- Timestamp:
- 05/06/13 20:10:43 (11 years ago)
- File:
-
- 1 edited
-
trunk/FACT++/src/EventBuilderWrapper.h (modified) (13 diffs)
Legend:
- Unmodified
- Added
- Removed
-
trunk/FACT++/src/EventBuilderWrapper.h
r15477 r15514 12 12 #include <boost/filesystem.hpp> 13 13 #include <boost/date_time/posix_time/posix_time_types.hpp> 14 #include <condition_variable> 14 15 15 16 #include "DimWriteStatistics.h" … … 26 27 #include "DataWriteFits2.h" 27 28 29 #include "queue.h" 30 28 31 namespace ba = boost::asio; 29 32 namespace bs = boost::system; … … 55 58 private: 56 59 boost::thread fThreadMain; 57 boost::thread fThreadDimQueue1;58 boost::thread fThreadDimQueue2;59 60 bool fThreadDimQueueStop;61 62 deque<pair<Time,GUI_STAT>> fDimQueue1;63 deque<tuple<Time,bool,FAD::EventHeader>> fDimQueue2;64 65 mutex fMutexDimQueue1;66 mutex fMutexDimQueue2;67 68 60 69 61 enum CommandStates_t // g_runStat … … 117 109 DimDescribedService fDimIncomplete; 118 110 111 Queue<pair<Time,GUI_STAT>> fDimQueue1; 112 Queue<tuple<Time,bool,FAD::EventHeader>> fDimQueue2; 113 119 114 bool fDebugStream; 120 115 bool fDebugRead; … … 188 183 public: 189 184 EventBuilderWrapper(MessageImp &imp) : fMsg(imp), 190 fThreadDimQueueStop(false),191 185 fFileFormat(FAD::kNone), fMaxRun(0), fLastOpened(0), fLastClosed(0), 192 186 fDimWriteStats ("FAD_CONTROL", imp), … … 272 266 fDimFileFormat("FAD_CONTROL/FILE_FORMAT", "S:1", "|format[int]:Current file format"), 273 267 fDimIncomplete("FAD_CONTROL/INCOMPLETE", "X:1", "|incomplete[bits]:One bit per board"), 268 // It is important to instantiate them after the DimServices 269 fDimQueue1(std::bind(&EventBuilderWrapper::factStatSend, this, placeholders::_1)), 270 fDimQueue2(std::bind(&EventBuilderWrapper::procHeader, this, placeholders::_1)), 274 271 fDebugStream(false), fDebugRead(false), fDebugLog(false), fNightAsInt(0) 275 272 { … … 284 281 for (size_t i=0; i<40; i++) 285 282 ConnectSlot(i, tcp::endpoint()); 286 287 fThreadDimQueue1 = boost::thread(boost::bind(&EventBuilderWrapper::SendStat, this));288 fThreadDimQueue2 = boost::thread(boost::bind(&EventBuilderWrapper::ProcHeader, this));289 283 } 290 284 … … 292 286 { 293 287 Abort(); 294 fThreadDimQueueStop = true;295 288 296 289 // FIXME: Used timed_join and abort afterwards 297 290 // What's the maximum time the eb need to abort? 298 291 fThreadMain.join(); 299 fThreadDimQueue1.join();300 fThreadDimQueue2.join();301 292 302 293 //ffMsg.Info("EventBuilder stopped."); … … 367 358 368 359 g_runStat = kModeRun; 369 g_maxProc = 3;360 g_maxProc = 1; 370 361 371 362 fMsg.Message("Starting EventBuilder thread"); … … 1104 1095 case 0: 1105 1096 SendRawData(fadhd, event); 1106 return 1; 1097 return 100; 1098 /* 1107 1099 case 1: 1108 1100 SendFeedbackData(fadhd, event); 1109 return 2; 1101 return 2;*/ 1110 1102 } 1111 1103 return 100; … … 1261 1253 } 1262 1254 1263 void SendStat() 1264 { 1265 // Performance could be increased signaling the thread to wake up 1266 while (!fThreadDimQueueStop) 1267 { 1268 if (fDimQueue1.size()==0) 1269 { 1270 usleep(1000); 1271 continue; 1272 } 1273 1274 fMutexDimQueue1.lock(); 1275 const pair<Time,GUI_STAT> stat = fDimQueue1.front(); 1276 fDimQueue1.pop_front(); 1277 fMutexDimQueue1.unlock(); 1278 1279 fDimStatistics1.setData(&stat.second, sizeof(GUI_STAT)); 1280 fDimStatistics1.Update(stat.first); 1281 } 1255 void factStatSend(const pair<Time,GUI_STAT> &stat) 1256 { 1257 fDimStatistics1.setData(&stat.second, sizeof(GUI_STAT)); 1258 fDimStatistics1.Update(stat.first); 1282 1259 } 1283 1260 1284 1261 void factStat(const GUI_STAT &stat) 1285 1262 { 1286 const lock_guard<mutex> guard(fMutexDimQueue1); 1287 fDimQueue1.push_back(make_pair(Time(), stat)); 1263 fDimQueue1.post(make_pair(Time(), stat)); 1288 1264 } 1289 1265 … … 1404 1380 vector<uint> fNumConnected; 1405 1381 1406 void ProcHeader() 1407 { 1408 // Performance could be increased signaling the thread to wake up 1409 while (!fThreadDimQueueStop) 1410 { 1411 if (fDimQueue2.size()==0) 1412 { 1413 usleep(100); 1414 continue; 1415 } 1416 1417 fMutexDimQueue2.lock(); 1418 const auto dat = fDimQueue2.front(); 1419 fDimQueue2.pop_front(); 1420 fMutexDimQueue2.unlock(); 1421 1422 const Time &t = get<0>(dat); 1423 const bool changed = get<1>(dat); 1424 const FAD::EventHeader &h = get<2>(dat); 1425 1426 const FAD::EventHeader old = fVecHeader[h.Id()]; 1427 fVecHeader[h.Id()] = h; 1428 1429 if (old.fVersion != h.fVersion || changed) 1430 { 1431 const array<uint16_t,42> ver = Compare(&fVecHeader[0], &fVecHeader[0].fVersion); 1432 1433 array<float,42> data; 1434 for (int i=0; i<42; i++) 1382 void procHeader(const tuple<Time,bool,FAD::EventHeader> &dat) 1383 { 1384 const Time &t = get<0>(dat); 1385 const bool changed = get<1>(dat); 1386 const FAD::EventHeader &h = get<2>(dat); 1387 1388 const FAD::EventHeader old = fVecHeader[h.Id()]; 1389 fVecHeader[h.Id()] = h; 1390 1391 if (old.fVersion != h.fVersion || changed) 1392 { 1393 const array<uint16_t,42> ver = Compare(&fVecHeader[0], &fVecHeader[0].fVersion); 1394 1395 array<float,42> data; 1396 for (int i=0; i<42; i++) 1397 { 1398 ostringstream str; 1399 str << (ver[i]>>8) << '.' << (ver[i]&0xff); 1400 data[i] = stof(str.str()); 1401 } 1402 Update(fDimFwVersion, data, t); 1403 } 1404 1405 if (old.fRunNumber != h.fRunNumber || changed) 1406 { 1407 const array<uint32_t,42> run = Compare(&fVecHeader[0], &fVecHeader[0].fRunNumber); 1408 fDimRunNumber.setData(&run[0], 42*sizeof(uint32_t)); 1409 fDimRunNumber.Update(t); 1410 } 1411 1412 if (old.fTriggerGeneratorPrescaler != h.fTriggerGeneratorPrescaler || changed) 1413 { 1414 const array<uint16_t,42> pre = Compare(&fVecHeader[0], &fVecHeader[0].fTriggerGeneratorPrescaler); 1415 fDimPrescaler.setData(&pre[0], 42*sizeof(uint16_t)); 1416 fDimPrescaler.Update(t); 1417 } 1418 1419 if (old.fDNA != h.fDNA || changed) 1420 { 1421 const array<uint64_t,42> dna = Compare(&fVecHeader[0], &fVecHeader[0].fDNA); 1422 Update(fDimDNA, dna, t, 40); 1423 } 1424 1425 if (old.fStatus != h.fStatus || changed) 1426 { 1427 const array<uint16_t,42> sts = CompareBits(&fVecHeader[0], &fVecHeader[0].fStatus); 1428 Update(fDimStatus, sts, t); 1429 } 1430 1431 if (memcmp(old.fDac, h.fDac, sizeof(h.fDac)) || changed) 1432 { 1433 array<uint16_t, FAD::kNumDac*42> dacs; 1434 1435 for (int i=0; i<FAD::kNumDac; i++) 1436 { 1437 const array<uint16_t, 42> dac = Compare(&fVecHeader[0], &fVecHeader[0].fDac[i]); 1438 memcpy(&dacs[i*42], &dac[0], sizeof(uint16_t)*42); 1439 } 1440 1441 Update(fDimDac, dacs, t); 1442 } 1443 1444 // ----------- 1445 1446 static Time oldt(boost::date_time::neg_infin); 1447 Time newt; 1448 1449 if (newt>oldt+boost::posix_time::seconds(1)) 1450 { 1451 oldt = newt; 1452 1453 // --- RefClock 1454 1455 const array<uint32_t,42> clk = Compare(&fVecHeader[0], &fVecHeader[0].fFreqRefClock); 1456 Update(fDimRefClock, clk, t); 1457 1458 // --- Temperatures 1459 1460 const array<int16_t,42> tmp[4] = 1461 { 1462 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[0]), // 0-39:val, 40:min, 41:max 1463 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[1]), // 0-39:val, 40:min, 41:max 1464 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[2]), // 0-39:val, 40:min, 41:max 1465 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[3]) // 0-39:val, 40:min, 41:max 1466 }; 1467 1468 vector<int16_t> data; 1469 data.reserve(82); 1470 data.push_back(tmp[0][40]); // min: 0 1471 data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 1-40 1472 data.push_back(tmp[0][41]); // max: 41 1473 data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 42-81 1474 1475 for (int j=1; j<=3; j++) 1476 { 1477 const array<int16_t,42> &ref = tmp[j]; 1478 1479 // Gloabl min 1480 if (ref[40]<data[0]) // 40=min 1481 data[0] = ref[40]; 1482 1483 // Global max 1484 if (ref[41]>data[41]) // 41=max 1485 data[41] = ref[41]; 1486 1487 for (int i=0; i<40; i++) 1435 1488 { 1436 ostringstream str; 1437 str << (ver[i]>>8) << '.' << (ver[i]&0xff); 1438 data[i] = stof(str.str()); 1489 // min per board 1490 if (ref[i]<data[i+1]) // data: 1-40 1491 data[i+1] = ref[i]; // ref: 0-39 1492 1493 // max per board 1494 if (ref[i]>data[i+42]) // data: 42-81 1495 data[i+42] = ref[i]; // ref: 0-39 1439 1496 } 1440 Update(fDimFwVersion, data, t); 1441 } 1442 1443 if (old.fRunNumber != h.fRunNumber || changed) 1444 { 1445 const array<uint32_t,42> run = Compare(&fVecHeader[0], &fVecHeader[0].fRunNumber); 1446 fDimRunNumber.setData(&run[0], 42*sizeof(uint32_t)); 1447 fDimRunNumber.Update(t); 1448 } 1449 1450 if (old.fTriggerGeneratorPrescaler != h.fTriggerGeneratorPrescaler || changed) 1451 { 1452 const array<uint16_t,42> pre = Compare(&fVecHeader[0], &fVecHeader[0].fTriggerGeneratorPrescaler); 1453 fDimPrescaler.setData(&pre[0], 42*sizeof(uint16_t)); 1454 fDimPrescaler.Update(t); 1455 } 1456 1457 if (old.fDNA != h.fDNA || changed) 1458 { 1459 const array<uint64_t,42> dna = Compare(&fVecHeader[0], &fVecHeader[0].fDNA); 1460 Update(fDimDNA, dna, t, 40); 1461 } 1462 1463 if (old.fStatus != h.fStatus || changed) 1464 { 1465 const array<uint16_t,42> sts = CompareBits(&fVecHeader[0], &fVecHeader[0].fStatus); 1466 Update(fDimStatus, sts, t); 1467 } 1468 1469 if (memcmp(old.fDac, h.fDac, sizeof(h.fDac)) || changed) 1470 { 1471 array<uint16_t, FAD::kNumDac*42> dacs; 1472 1473 for (int i=0; i<FAD::kNumDac; i++) 1474 { 1475 const array<uint16_t, 42> dac = Compare(&fVecHeader[0], &fVecHeader[0].fDac[i]); 1476 memcpy(&dacs[i*42], &dac[0], sizeof(uint16_t)*42); 1477 } 1478 1479 Update(fDimDac, dacs, t); 1480 } 1481 1482 // ----------- 1483 1484 static Time oldt(boost::date_time::neg_infin); 1485 Time newt; 1486 1487 if (newt>oldt+boost::posix_time::seconds(1)) 1488 { 1489 oldt = newt; 1490 1491 // --- RefClock 1492 1493 const array<uint32_t,42> clk = Compare(&fVecHeader[0], &fVecHeader[0].fFreqRefClock); 1494 Update(fDimRefClock, clk, t); 1495 1496 // --- Temperatures 1497 1498 const array<int16_t,42> tmp[4] = 1499 { 1500 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[0]), // 0-39:val, 40:min, 41:max 1501 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[1]), // 0-39:val, 40:min, 41:max 1502 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[2]), // 0-39:val, 40:min, 41:max 1503 Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[3]) // 0-39:val, 40:min, 41:max 1504 }; 1505 1506 vector<int16_t> data; 1507 data.reserve(82); 1508 data.push_back(tmp[0][40]); // min: 0 1509 data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 1-40 1510 data.push_back(tmp[0][41]); // max: 41 1511 data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 42-81 1512 1513 for (int j=1; j<=3; j++) 1514 { 1515 const array<int16_t,42> &ref = tmp[j]; 1516 1517 // Gloabl min 1518 if (ref[40]<data[0]) // 40=min 1519 data[0] = ref[40]; 1520 1521 // Global max 1522 if (ref[41]>data[41]) // 41=max 1523 data[41] = ref[41]; 1524 1525 for (int i=0; i<40; i++) 1526 { 1527 // min per board 1528 if (ref[i]<data[i+1]) // data: 1-40 1529 data[i+1] = ref[i]; // ref: 0-39 1530 1531 // max per board 1532 if (ref[i]>data[i+42]) // data: 42-81 1533 data[i+42] = ref[i]; // ref: 0-39 1534 } 1535 } 1536 1537 vector<float> deg(82); // 0: global min, 1-40: min 1538 for (int i=0; i<82; i++) // 41: global max, 42-81: max 1539 deg[i] = data[i]/16.; 1540 1541 fDimTemperature.setData(deg.data(), 82*sizeof(float)); 1542 fDimTemperature.Update(t); 1543 } 1544 1497 } 1498 1499 vector<float> deg(82); // 0: global min, 1-40: min 1500 for (int i=0; i<82; i++) // 41: global max, 42-81: max 1501 deg[i] = data[i]/16.; 1502 1503 fDimTemperature.setData(deg.data(), 82*sizeof(float)); 1504 fDimTemperature.Update(t); 1545 1505 } 1546 1506 } … … 1561 1521 fNumConnected = con; 1562 1522 1563 const lock_guard<mutex> guard(fMutexDimQueue2); 1564 fDimQueue2.push_back(make_tuple(Time(), changed, h)); 1523 fDimQueue2.post(make_tuple(Time(), changed, h)); 1524 1525 //const lock_guard<mutex> guard(fMutexDimQueue2); 1526 //fDimQueue2.push_back(make_tuple(Time(), changed, h)); 1565 1527 } 1566 1528 };
Note:
See TracChangeset
for help on using the changeset viewer.
