参照ソースと行番号
資料へ戻る。ローカルスナップショットから生成。ソースは変更していません。
AmQStrTdcData.h
SHA-256: 5f2cd72198629999094012bcae745e050d1e8ec30293260b38a4e4064e2845bb
1 #ifndef AmQStrTdcData_h 2 #define AmQStrTdcData_h 3 4 #include <cstdint> 5 6 namespace AmQStrTdc::Data { 7 8 struct Word { 9 uint8_t d[8]; 10 }; 11 12 namespace v0 { 13 14 struct Bits { 15 union { 16 uint8_t d[8]; 17 18 struct { 19 uint64_t raw : 64; 20 }; 21 22 // common 23 struct { 24 uint64_t com_rsv : 58; // [57:0] 25 uint64_t head : 6; // [63:58] 26 }; 27 28 // lrtdc 29 struct { 30 uint64_t zero_t1 : 16; // [15:0] 31 uint64_t tdc : 19; // [34:16] 32 uint64_t tot : 16; // [50:35] 33 uint64_t ch : 7; // [57:51] 34 uint64_t dtype : 6; // [63:58] 35 }; 36 37 // hrtdc 38 struct { 39 uint64_t hrtdc : 29; // [28:0] 40 uint64_t hrtot : 22; // [50:29] 41 uint64_t hrch : 7; // [57:51] 42 uint64_t hrdtype : 6; // [63:58] 43 }; 44 45 // heartbeat 46 struct { 47 uint64_t zero_h : 24; 48 uint64_t hbframe : 16; //[39,24] 49 uint64_t hbspilln : 8; //[47,40] 50 uint64_t hbflag : 10; //[57,48] 51 uint64_t htype : 6; //[63,58] 52 }; 53 54 // spill 55 struct { 56 uint64_t zero_s : 24; 57 uint64_t hbfrco : 16; //[39,24] 58 uint64_t spilln : 8; //[47,40] 59 uint64_t spflag : 10; //[57,48] 60 uint64_t stype : 6; //[63,58] 61 }; 62 63 }; 64 }; 65 66 enum HeadTypes { 67 Data = 0x0B, 68 Heartbeat = 0x1C, 69 Trailer = 0x0D, 70 // ErrorRecovery = 0xE, 71 SpillOn = 0x18, 72 SpillEnd = 0x14 73 }; 74 75 } // namespace v0 76 77 78 inline namespace v1 { 79 80 struct Bits { 81 union { 82 uint8_t d[8]; 83 84 struct { 85 uint64_t raw : 64; 86 }; 87 88 // common 89 struct { 90 uint64_t com_rsv : 58; // [57:0] 91 uint64_t head : 6; // [63:58] 92 }; 93 94 // lrtdc 95 struct { 96 uint64_t zero_t1 : 15; // [14:0] 97 uint64_t tdc : 19; // [33:15] 98 uint64_t tot : 16; // [49:34] 99 uint64_t ch : 8; // [57:50] 100 uint64_t dtype : 6; // [63:58] 101 }; 102 103 // hrtdc 104 struct { 105 uint64_t hrtdc : 29; // [28:0] 106 uint64_t hrtot : 22; // [50:29] 107 uint64_t hrch : 7; // [57:51] 108 uint64_t hrdtype : 6; // [63:58] 109 }; 110 111 // 1st heartbeat 112 struct { 113 uint64_t hbframe : 24; // [23,0] 114 uint64_t toffset : 16; // [39,24] 115 uint64_t hbflag : 16; // [55,40] 116 uint64_t reserve1 : 2; // [57,56] 117 uint64_t hbtype1 : 6; // [63,58] 118 }; 119 120 // 2nd heartbeat 121 struct { 122 uint64_t transSize : 20; 123 uint64_t geneSize : 20; // [39,20] 124 uint64_t userReg : 16; // [55,40] 125 uint64_t reserve2 : 2; // [57,56] 126 uint64_t hbtype2 : 6; // [63,58] 127 }; 128 129 }; 130 }; 131 132 enum HeadTypes { 133 Data = 0x0B, 134 Heartbeat = 0x1C, 135 Heartbeat2nd = 0x1E, 136 Trailer = 0x0D, 137 ThrottlingT1Start = 0x19, 138 ThrottlingT1End = 0x11, 139 ThrottlingT2Start = 0x1A, 140 ThrottlingT2End = 0x12, 141 142 SpillOn = 0x18, // for compatibility 143 SpillEnd = 0x14 // for compatibility 144 }; 145 146 } // namespace v1 147 148 149 } // namespace AmQStrTdc::Data 150 151 152 153 #endif
AmQStrTdcSampler.cxx
SHA-256: dc56c89345c6735ca1809bd21eb3478027b7b1d069ac19d14b6a4fe0d381ad79
1 #include <vector> 2 #include <chrono> 3 #include <algorithm> 4 #include <iostream> 5 6 #include <fairmq/runDevice.h> 7 8 #include "utility/MessageUtil.h" 9 #include "AmQStrTdcSampler.h" 10 11 #include "utility/network.hh" 12 #include "utility/rbcp.h" 13 #include "utility/UDPRBCP.hh" 14 15 namespace bpo = boost::program_options; 16 17 //______________________________________________________________________________ 18 void 19 addCustomOptions(bpo::options_description& options) 20 { 21 using opt = AmQStrTdcSampler::OptionKey; 22 options.add_options() 23 (opt::IpSiTCP.data(), bpo::value<std::string>()->default_value("0"), "SiTCP IP (xxx.yyy.zzz.aaa)") 24 (opt::OutputChannelName.data(), bpo::value<std::string>()->default_value("out"), "Name of the output channel"); 25 // ("TdcType", bpo::value<std::string>()->default_value("2"), "TDC type: HR=2, LR=1"); 26 } 27 28 //______________________________________________________________________________ 29 std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions&) 30 { 31 return std::make_unique<AmQStrTdcSampler>(); 32 } 33 34 //______________________________________________________________________________ 35 36 AmQStrTdcSampler::AmQStrTdcSampler() 37 : FairMQDevice() 38 {} 39 40 //______________________________________________________________________________ 41 bool AmQStrTdcSampler::ConditionalRun() 42 { 43 using namespace std::chrono_literals; 44 45 // int n_word = 0; 46 int recv_status = 0; 47 int num_recieved_bytes = 0; 48 int n_word = 0; 49 uint8_t* buffer = new uint8_t[kOutBufByte*fnWordPerCycle] {}; 50 51 while( -1 == ( recv_status = Event_Cycle(buffer, num_recieved_bytes))) { 52 if (NewStatePending()) { 53 break; 54 } else { 55 continue; 56 } 57 } 58 59 n_word = num_recieved_bytes/optnByte; 60 61 #if 0 62 if(recv_status == -4) { 63 for(int i = 0; i<fnWordPerCycle; ++i) { 64 65 // if((buffer[fnByte*i+4] & 0xff) == 0x50){ 66 //if((buffer[optnByte*i + header_pos] & 0xff) == 0x50) { 67 // if((buffer[optnByte*i + header_pos] & 0xf0) == 0x50) { 68 if(true 69 &&(buffer[optnByte*i + 0] & 0xff) == 0x0 70 &&(buffer[optnByte*i + 1] & 0xff) == 0x0 71 &&(buffer[optnByte*i + 2] & 0xff) == 0x0 72 &&(buffer[optnByte*i + 3] & 0xff) == 0x0 73 &&(buffer[optnByte*i + 4] & 0xff) == 0x0 74 &&(buffer[optnByte*i + 5] & 0xff) == 0x0 75 &&(buffer[optnByte*i + 6] & 0xff) == 0x0 76 &&(buffer[optnByte*i + 7] & 0xff) == 0x0 77 ) { 78 LOG(debug) << "#D: Empty data: "; 79 LOG(debug) << " n_word: " << n_word << ", array index:" << i; 80 81 // n_word = i+1; 82 // n_word = i+2; 83 break; 84 } 85 }// For(i) 86 } 87 #endif 88 89 // if(n_word == -4){ 90 // n_word = remain; 91 // remain = 0; 92 // } 93 94 if (recv_status <= 0 && recv_status != -4) { 95 delete[] buffer; 96 return true; 97 } 98 99 #if 0 100 if(n_word == -4) { 101 for(int i = 0; i<fnWordPerCycle; ++i) { 102 103 // if((buffer[fnByte*i+4] & 0xff) == 0x50){ 104 //if((buffer[optnByte*i + header_pos] & 0xff) == 0x50) { 105 if((buffer[optnByte*i + header_pos] & 0xf0) == 0x50) { 106 printf("\n#D : Spill End is detected\n"); 107 // n_word = i+1; 108 n_word = i+2; 109 break; 110 } 111 }// For(i) 112 } 113 114 if (n_word<=0) { 115 delete[] buffer; 116 return true; 117 } 118 #endif 119 120 FairMQMessagePtr msg(NewMessage((char*)buffer, 121 //fnByte*fnWordPerCycle, 122 kOutBufByte*n_word, 123 [](void* object, void*) 124 { 125 delete [] static_cast<uint8_t*>(object); 126 } 127 ) 128 ); 129 130 // while (Send(msg, fOutputChannelName) == -2); 131 132 Send(msg, fOutputChannelName); 133 134 // ++fNumIterations; 135 // if ((fMaxIterations >0) && (fNumIterations >= fMaxIterations)) { 136 // LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state."; 137 // return false; 138 // } 139 // } 140 // else { 141 // LOG(warn) << "failed to send a message."; 142 // } 143 144 // LOG(warn) << "You should swim!"; 145 // std::this_thread::sleep_for(std::chrono::seconds(1)); 146 147 return true; 148 } 149 150 //______________________________________________________________________________ 151 void AmQStrTdcSampler::Init() 152 { 153 } 154 //_____________________________________________________________________________ 155 void AmQStrTdcSampler::SendFEMInfo() 156 { 157 158 { 159 uint64_t fFEMId = 0; 160 std::string sFEMId(fIpSiTCP); 161 std::istringstream istrst(sFEMId); 162 std::string token; 163 int ik = 3; 164 while(std::getline(istrst, token, '.')) { 165 if(ik < 0) break; 166 167 uint32_t ipv = (std::stoul(token) & 0xff) << (8*ik); 168 fFEMId |= ipv; 169 --ik; 170 } 171 LOG(debug) << "FEM ID " << std::hex << fFEMId << std::dec; 172 LOG(debug) << "FEM Magic " << std::hex << fem_info_.magic << std::dec; 173 174 fem_info_.FEMId = fFEMId; 175 fem_info_.FEMType = fTdcType; 176 } 177 178 179 // bit set-up for module information 180 unsigned char* fbuf = new uint8_t[sizeof(fem_info_)]; 181 182 uint8_t buf[8] = {0}; 183 for(int i = 0; i < 8; i++) { 184 buf[7-i] = (fem_info_.magic >> 8*(7-i) ) & 0xff; 185 } 186 memcpy(fbuf, &buf, sizeof(char)*8); 187 188 for(int i = 0; i < 8; i++) { 189 if(i < 4) { 190 buf[7-i] = (fem_info_.FEMId >> 8*(3-i)) & 0xff; 191 } else { 192 buf[7-i] = (fem_info_.FEMType >> 8*(7-i)) & 0xff; 193 } 194 } 195 memcpy(&fbuf[8], &buf, sizeof(char)*8); 196 197 uint8_t resv[8] = {0}; 198 memcpy(&fbuf[16], &resv, sizeof(char)*8); 199 //memcpy(fbuf, &fem_info_, sizeof(fem_info_)); 200 201 FairMQMessagePtr initmsg( 202 NewMessage((char*)fbuf, 203 kOutBufByte*3, 204 [](void* object, void*) { 205 delete [] static_cast<uint8_t*>(object); 206 } 207 ) 208 ); 209 210 LOG(info) << "Sending FEMInfo \"" << sizeof(fem_info_) << "\""; 211 212 213 int count=0; 214 while(true) { 215 216 if (Send(initmsg, fOutputChannelName) < 0) { 217 LOG(warn) << " Fail to send FEMInfo : " << count; 218 count++; 219 } else { 220 LOG(debug) << " Send FEMInfo : " << count; 221 break; 222 } 223 } 224 225 } 226 227 //______________________________________________________________________________ 228 void AmQStrTdcSampler::InitTask() 229 { 230 231 using opt = OptionKey; 232 233 fIpSiTCP = fConfig->GetProperty<std::string>("msiTcpIp"); 234 // fIpSiTCP = fConfig->GetProperty<std::string>(opt::IpSiTCP.data()); 235 LOG(info) << "TPC IP: " << fIpSiTCP; 236 // fIpSiTCP = fConfig->GetProperty<std::string>(opt::IpSiTCP.data()); 237 fOutputChannelName = fConfig->GetProperty<std::string>(opt::OutputChannelName.data()); 238 239 fTdcType = std::stoi(fConfig->GetProperty<std::string>("TdcType")); 240 LOG(info) << "TDC Type: " << fTdcType << std::endl; 241 242 LOG(info) << fIpSiTCP; 243 SendFEMInfo(); 244 245 header_pos = 7; 246 optnByte = 8; 247 248 249 /* 250 rbcp_header rbcpHeader; 251 rbcpHeader.type = UDPRBCP::rbcp_ver_; 252 rbcpHeader.id = 0; 253 254 FPGAModule fModule(fIpSiTCP.c_str(), 4660, &rbcpHeader, 0); 255 LOG(info) << std::hex << fModule.ReadModule(hul_strtdc::BCT::addr_Version, 4) << std::dec; 256 */ 257 } 258 259 //______________________________________________________________________________ 260 void AmQStrTdcSampler::PreRun() 261 { 262 263 if(-1 == (fAmqSocket = ConnectSocket(fIpSiTCP.c_str()))) return; 264 LOG(info) << "TCP connected"; 265 266 /* 267 rbcp_header rbcpHeader; 268 rbcpHeader.type = UDPRBCP::rbcp_ver_; 269 rbcpHeader.id = 0; 270 271 FPGAModule fModule(fIpSiTCP.c_str(), 4660, &rbcpHeader, 0); 272 fModule.WriteModule(ODP::addr_en_filter, fTotFilterEn, 1); 273 fModule.WriteModule(DCT::addr_gate, 1, 1); 274 */ 275 LOG(info) << "Start DAQ"; 276 } 277 278 //______________________________________________________________________________ 279 void AmQStrTdcSampler::PostRun() 280 { 281 /* 282 using namespace hul_strtdc; 283 284 rbcp_header rbcpHeader; 285 rbcpHeader.type = UDPRBCP::rbcp_ver_; 286 rbcpHeader.id = 0; 287 288 FPGAModule fModule(fIpSiTCP.c_str(), 4660, &rbcpHeader, 0); 289 fModule.WriteModule(DCT::addr_gate, 0, 1); 290 */ 291 292 close(fAmqSocket); 293 LOG(info) << "Socket close"; 294 LOG(info) << "End DAQ"; 295 } 296 297 //______________________________________________________________________________ 298 void AmQStrTdcSampler::ResetTask() 299 { 300 // close(fAmqSocket); 301 302 } 303 304 //______________________________________________________________________________ 305 int AmQStrTdcSampler::ConnectSocket(const char* ip) 306 { 307 struct sockaddr_in SiTCP_ADDR; 308 unsigned int port = 24; 309 310 int sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); 311 SiTCP_ADDR.sin_family = AF_INET; 312 SiTCP_ADDR.sin_port = htons((unsigned short int) port); 313 SiTCP_ADDR.sin_addr.s_addr = inet_addr(ip); 314 315 struct timeval tv; 316 //tv.tv_sec = 0; 317 //tv.tv_usec = 250000; 318 tv.tv_sec = 1; 319 // tv.tv_sec = 5; 320 tv.tv_usec = 0; 321 setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (char*)&tv, sizeof(tv)); 322 323 int flag = 1; 324 setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char*)&flag, sizeof(flag)); 325 326 if(0 > connect(sock, (struct sockaddr*)&SiTCP_ADDR, sizeof(SiTCP_ADDR))) { 327 LOG(error) << "TCP connection error"; 328 close(sock); 329 return -1; 330 } 331 332 return sock; 333 } 334 335 // Event Cycle ------------------------------------------------------------ 336 int AmQStrTdcSampler::Event_Cycle(uint8_t* buffer, int& num_recieved_bytes) 337 { 338 // data read --------------------------------------------------------- 339 // static const unsigned int sizeData = fnByte*fnWordPerCycle*sizeof(uint8_t); 340 static const unsigned int sizeData = optnByte*fnWordPerCycle*sizeof(uint8_t); 341 int ret = receive(fAmqSocket, (char*)buffer, sizeData, num_recieved_bytes); 342 343 return ret; 344 // if(ret <= 0) return ret; 345 // return fnWordPerCycle; 346 } 347 348 // receive ---------------------------------------------------------------- 349 int AmQStrTdcSampler::receive(int sock, char* data_buf, unsigned int length, int& num_recieved_bytes) 350 { 351 unsigned int revd_size = 0; 352 int tmp_ret = 0; 353 354 while(revd_size < length) { 355 tmp_ret = recv(sock, data_buf + revd_size, length -revd_size, 0); 356 // LOG(debug) << "tmp_ret: " << tmp_ret; 357 358 if(tmp_ret == 0) break; 359 if(tmp_ret < 0) { 360 int errbuf = errno; 361 perror("TCP receive"); 362 if(errbuf == EAGAIN) { 363 // this is time out 364 // std::cout << "#D : TCP recv time out" << std::endl; 365 // remain = revd_size; 366 LOG(debug) << "#D: recv time out, recieved byte size: " << revd_size; 367 num_recieved_bytes = revd_size; 368 return -4; 369 } else { 370 // something wrong 371 std::cerr << "TCP error : " << errbuf << std::endl; 372 } 373 374 revd_size = tmp_ret; 375 break; 376 } 377 378 revd_size += tmp_ret; 379 } 380 381 num_recieved_bytes = revd_size; 382 return revd_size; 383 } 384
AmQStrTdcSampler.h
SHA-256: 60f08e80f9af505d6638f94b11028b279a64822fa1e64b397c55207c058ed7bc
1 #ifndef AmQStrTdcSampler_h 2 #define AmQStrTdcSampler_h 3 4 #include <cstdint> 5 #include <string> 6 #include <string_view> 7 #include <thread> 8 #include <atomic> 9 #include <memory> 10 11 #include <fairmq/Device.h> 12 13 // O F N I - M E F 14 constexpr uint64_t Magic {0x4f464e492d4d4546}; 15 constexpr int kOutBufByte {8}; 16 // const int fnWordPerCycle {16384*10}; 17 //constexpr int fnWordPerCycle {16384}; 18 constexpr int fnWordPerCycle {1024}; 19 20 struct FEMInfo { 21 uint64_t magic {Magic}; 22 uint32_t FEMId {0}; 23 uint32_t FEMType {0}; 24 uint64_t reserved {0}; 25 }; 26 27 struct lrWord { 28 uint8_t d[5]; 29 }; 30 31 class AmQStrTdcSampler : public fair::mq::Device 32 { 33 public: 34 struct OptionKey { 35 static constexpr std::string_view IpSiTCP {"sitcp-ip"}; 36 static constexpr std::string_view OutputChannelName {"out-chan-name"}; 37 }; 38 39 AmQStrTdcSampler(); 40 AmQStrTdcSampler(const AmQStrTdcSampler&) = delete; 41 AmQStrTdcSampler& operator=(const AmQStrTdcSampler&) = delete; 42 ~AmQStrTdcSampler() = default; 43 44 int ConnectSocket(const char* ip); 45 int Event_Cycle(uint8_t* buffer, int& num_recieved_bytes); 46 int receive(int sock, char* data_buf, unsigned int length, int& num_recived_bytes); 47 48 void SendFEMInfo(); 49 50 protected: 51 bool ConditionalRun() override; 52 void Init() override; 53 void InitTask() override; 54 void PreRun() override; 55 void PostRun() override; 56 void ResetTask() override; 57 58 int fAmqSocket {0}; 59 std::string fIpSiTCP {"0"}; 60 std::string fOutputChannelName; 61 62 63 64 65 FEMInfo fem_info_; 66 int fTdcType {2}; 67 int header_pos {0}; 68 int optnByte {0}; 69 70 int remain {0}; 71 }; 72 73 #endif
FileSink.cxx
SHA-256: 8e146bc3b5e27afa39cc6be8c8befbf82125271307ddf9c7f99e6b1998b21dad
1 #include <functional> 2 #include <sstream> 3 4 #include <boost/algorithm/string/case_conv.hpp> 5 6 #include <fairmq/runDevice.h> 7 8 #include "utility/CompressHelper.h" 9 #include "utility/Compressor.h" 10 #include "utility/HexDump.h" 11 #include "utility/program_options.h" 12 #include "utility/MessageUtil.h" 13 #include "utility/TaskProcessorMT.h" 14 #include "FileSink.h" 15 #include "FileSinkHeader.h" 16 #include "FileSinkTrailer.h" 17 18 using namespace nestdaq; 19 using namespace std::string_literals; 20 21 namespace bpo = boost::program_options; 22 using opt = nestdaq::FileSink::OptionKey; 23 24 //============================================================================= 25 void addCustomOptions(bpo::options_description &options) 26 { 27 { // FileSink's options ------------------------------ 28 options.add_options() 29 // 30 (opt::InputDataChannelName.data(), 31 bpo::value<std::string>()->default_value("in"), 32 "Name of input channel") 33 // 34 (opt::Multipart.data(), bpo::value<std::string>()->default_value("true"), "Handle multipart message") 35 // 36 (opt::MergeMessage.data(), bpo::value<std::string>()->default_value("true"), 37 "Merge multipart message to a single message before write") 38 // 39 (opt::RunNumber.data(), bpo::value<std::string>(), "Run number (integer) given by DAQ plugin") 40 // 41 (opt::MQTimeout.data(), bpo::value<std::string>()->default_value("1000"), 42 "Timeout of receive messages in milliseconds") 43 // 44 (opt::Discard.data(), bpo::value<std::string>()->default_value("true"), "Discard data received in PostRun") 45 // 46 (opt::NThreads.data(), bpo::value<std::string>()->default_value("0"), 47 "Number of threads for data compression. (if zero or negative, a value obtained by " 48 "std::thread::hardware_concurrency() will be used.") 49 // 50 (opt::InProcMQLength.data(), bpo::value<std::string>()->default_value("1"), "in-process message queue length.") 51 // 52 (opt::MaxIteration.data(), bpo::value<std::string>()->default_value("0"), 53 "Number of iterations (if zero, no limitation is set)") 54 (opt::WriteSleepInMilliSec.data(), bpo::value<std::string>()->default_value("0"), 55 "Write Sleep in msec") 56 // 57 (opt::DQMChannelName.data(), bpo::value<std::string>()->default_value("dqm"), 58 "Name of DQM channel") 59 ; 60 } 61 62 { // FileUtil's options ------------------------------ 63 using fu = nestdaq::FileUtil; 64 //using fopt = fu::OptionKey; 65 //using sopt = fu::SplitOption; 66 //using oopt = fu::OpenmodeOption; 67 auto &desc = fu::GetDescriptions(); 68 69 fu::AddOptions(options); 70 } 71 } 72 73 //______________________________________________________________________________ 74 std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions &config) 75 { 76 (void)config; 77 78 return std::make_unique<nestdaq::FileSink>(); 79 } 80 81 //============================================================================= 82 namespace nestdaq { 83 84 //______________________________________________________________________________ 85 bool FileSink::CompressData(FairMQMessagePtr &rawMsg, int index) 86 { 87 if (fStopRequested) { 88 return false; 89 } 90 // LOG(debug) << __LINE__ << ":" << __func__ << " index = " << index; 91 auto first = reinterpret_cast<char *>(rawMsg->GetData()); 92 fSize += rawMsg->GetSize(); 93 auto compressed = std::make_unique<std::vector<char>>(fCompress(first, rawMsg->GetSize())); 94 auto compressedMsg = MessageUtil::NewMessage(*fWorker, std::move(compressed)); 95 fCompressedSize += compressedMsg->GetSize(); 96 fWorker->fWorkerOutputChannels[index]->Send(compressedMsg, fMQTimeoutMS); 97 return true; 98 } 99 100 //______________________________________________________________________________ 101 bool FileSink::CompressMultipartData(FairMQParts &rawMsgParts, int index) 102 { 103 if (fStopRequested) { 104 return false; 105 } 106 FairMQParts compressedMsgParts; 107 if (fMergeMessage) { 108 auto len = MessageUtil::TotalLength(rawMsgParts); 109 auto buf = fWorker->NewMessage(len); 110 auto bufFirst = reinterpret_cast<char *>(buf->GetData()); 111 auto bufLast = bufFirst + len; 112 auto itr = bufFirst; 113 114 for (const auto &m : rawMsgParts) { 115 auto first = reinterpret_cast<char *>(m->GetData()); 116 auto last = first + m->GetSize(); 117 // std::for_each(first, last, HexDump()); 118 std::copy(first, last, itr); 119 itr += m->GetSize(); 120 } 121 if (bufLast != itr) { 122 LOG(error) << __LINE__ << ":" << __func__ << " index = " << index << ": failed to copy buffer"; 123 } 124 auto compressed = std::make_unique<std::vector<char>>(fCompress(bufFirst, len)); 125 fSize += len; 126 fCompressedSize += compressed->size(); 127 compressedMsgParts.AddPart(MessageUtil::NewMessage(*fWorker, std::move(compressed))); 128 } else { 129 for (const auto &m : rawMsgParts) { 130 fSize += m->GetSize(); 131 auto first = reinterpret_cast<char *>(m->GetData()); 132 auto compressed = std::make_unique<std::vector<char>>(fCompress(first, m->GetSize())); 133 fCompressedSize += compressed->size(); 134 compressedMsgParts.AddPart(MessageUtil::NewMessage(*fWorker, std::move(compressed))); 135 } 136 } 137 138 // LOG(debug) << __LINE__ << ":" << __func__ << " index = " << index << " worker output channels send"; 139 fWorker->fWorkerOutputChannels[index]->Send(compressedMsgParts, fMQTimeoutMS); 140 // LOG(debug) << __LINE__ << ":" << __func__ << " index = " << index << " worker output channels send done"; 141 return true; 142 } 143 144 //______________________________________________________________________________ 145 bool FileSink::HandleData(FairMQMessagePtr &msg, int index) 146 { 147 if (fStopRequested) { 148 return false; 149 } 150 // LOG(info) << __LINE__ << ":" << __func__ << " index = " << index << " n-received = " << fNReceived << ", n-write = 151 // " << fNWrite; 152 ++fNReceived; 153 154 // send data to data quality monitor 155 #if 1 156 FairMQParts dqmParts; 157 if (int nSubChan = fChannels.count(fDQMChannelName)) { 158 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 159 msgCopy->Copy(*msg); 160 dqmParts.AddPart(std::move(msgCopy)); 161 if (Send(dqmParts,fDQMChannelName) < 0) { 162 if (NewStatePending()) { 163 LOG(info) << "Device is not RUNNING"; 164 } else { 165 LOG(error) << "Failed to enqueue time frame slice (DQM) " << std::endl; 166 } 167 } 168 } 169 #endif 170 return WriteData(msg, index); 171 } 172 173 //______________________________________________________________________________ 174 bool FileSink::HandleDataMT(FairMQMessagePtr &msg, int index) 175 { 176 (void)index; 177 178 if (fStopRequested) { 179 return false; 180 } 181 auto i = fNReceived % fNThreads; 182 // LOG(info) << __LINE__ << ":" << __func__ << " index = " << index << " i = " << i << " n-received = " << fNReceived 183 // << ", n-write = " << fNWrite; 184 ++fNReceived; 185 while (fWorker->fSplitterChannels[i]->Send(msg, fMQTimeoutMS) < 0) { 186 LOG(warn) << __func__ << " (local mq) failed to send"; 187 if (fStopRequested || NewStatePending()) { 188 return false; 189 } 190 } 191 192 return true; 193 } 194 195 //______________________________________________________________________________ 196 bool FileSink::HandleMultipartData(FairMQParts &msgParts, int index) 197 { 198 if (fStopRequested) { 199 return false; 200 } 201 // LOG(info) << __LINE__ << ":" << __func__ << " index = " << index << " n-received = " << fNReceived << ", n-write = 202 // " << fNWrite; 203 ++fNReceived; 204 auto nSubChan = GetNumSubChannels(fDQMChannelName); 205 206 for (auto iSubChan = 0; iSubChan < nSubChan; ++iSubChan) { 207 fair::mq::Parts partsCopy = MessageUtil::Copy(*this, msgParts); 208 if (Send(partsCopy,fDQMChannelName,iSubChan) < 0) { 209 if (NewStatePending()) { 210 LOG(info) << "Device is not RUNNING"; 211 } else { 212 LOG(error) << "Failed to enqueue time frame slice (DQM) " << std::endl; 213 } 214 } 215 } 216 217 218 return WriteMultipartData(msgParts, index); 219 } 220 221 //______________________________________________________________________________ 222 bool FileSink::HandleMultipartDataMT(FairMQParts &msgParts, int index) 223 { 224 (void)index; 225 226 if (fStopRequested) { 227 return false; 228 } 229 auto i = fNReceived % fNThreads; 230 // LOG(info) << __LINE__ << ":" << __func__ << " index = " << index << " i = " << i << " n-received = " << 231 // fNReceived << ", n-write = " << fNWrite; 232 ++fNReceived; 233 while (fWorker->fSplitterChannels[i]->Send(msgParts, fMQTimeoutMS) < 0) { 234 LOG(warn) << __func__ << " (local mq) failed to send"; 235 } 236 237 return true; 238 } 239 240 //______________________________________________________________________________ 241 void FileSink::Init() {} 242 243 //______________________________________________________________________________ 244 void FileSink::InitTask() 245 { 246 fStopRequested = false; 247 248 auto get = [this](auto name) -> std::string { 249 if (fConfig->Count(name.data()) < 1) { 250 LOG(debug) << " variable: " << name << " not found"; 251 return ""; 252 } 253 return fConfig->GetProperty<std::string>(name.data()); 254 }; 255 256 auto checkFlag = [this, &get](auto name) { 257 std::string s = get(name); 258 s = boost::to_lower_copy(s); 259 return (s == "1") || (s == "true") || (s == "yes"); 260 }; 261 262 fInputDataChannelName = get(opt::InputDataChannelName); 263 fDQMChannelName = get(opt::DQMChannelName); 264 265 fNThreads = std::stoi(get(opt::NThreads)); 266 if (fNThreads < 1) { 267 fNThreads = std::thread::hardware_concurrency(); 268 } 269 LOG(debug) << __func__ << " n threads = " << fNThreads; 270 fInProcMQLength = std::stoi(get(opt::InProcMQLength)); 271 fMultipart = checkFlag(opt::Multipart); 272 fWriteSleepInMilliSec = std::stoi(get(opt::WriteSleepInMilliSec)); 273 LOG(info) << __func__ << " fWriteSleepInMilliSec = " << fWriteSleepInMilliSec; 274 275 fFile = std::make_unique<FileUtil>(); 276 try { 277 fFile->Init(fConfig->GetVarMap()); 278 } catch (...) { 279 LOG(error) << __func__ << ", Invalid File property..."; 280 //ChangeStateOrThrow(fair::mq::Transition::Stop); 281 //ChangeState(fair::mq::Transition::Stop); 282 fStopRequested = true; 283 return; 284 } 285 fFile->Print(); 286 fFileExtension = fFile->GetExtension(); 287 auto compressFormat = Compressor::ExtToFormat(fFileExtension); 288 LOG(debug) << " compress format = " << fFileExtension << " " << static_cast<int>(compressFormat); 289 fCompress = [this, compressFormat](const char *src, int n) -> std::vector<char> { 290 return std::move(Compressor::Compress(src, n, compressFormat, fFile->GetCompBufSize())); 291 }; 292 fWorker.reset(); 293 if (fMultipart) { 294 LOG(debug) << " register data handler for multipart message"; 295 if ((fNThreads > 1) && (compressFormat != Compressor::Format::none)) { 296 OnData(fInputDataChannelName, &FileSink::HandleMultipartDataMT); 297 fWorker = std::make_unique<TaskProcessorMT>(*this); 298 LOG(debug) << " use multi-thread for compression"; 299 } else { 300 OnData(fInputDataChannelName, &FileSink::HandleMultipartData); 301 LOG(debug) << " single thread"; 302 } 303 } else { 304 LOG(debug) << " register data handler for single message"; 305 if ((fNThreads > 1) && (compressFormat != Compressor::Format::none)) { 306 OnData(fInputDataChannelName, &FileSink::HandleDataMT); 307 fWorker = std::make_unique<TaskProcessorMT>(*this); 308 LOG(debug) << " use multi-thread for compression"; 309 } else { 310 OnData(fInputDataChannelName, &FileSink::HandleData); 311 LOG(debug) << " single thread"; 312 } 313 } 314 315 fMaxIteration = std::stoull(get(opt::MaxIteration)); 316 317 fMQTimeoutMS = std::stoi(get(opt::MQTimeout)); 318 fDiscard = checkFlag(opt::Discard); 319 fMergeMessage = checkFlag(opt::MergeMessage); 320 } 321 322 //______________________________________________________________________________ 323 void FileSink::PostRun() 324 { 325 if (fStopRequested) { 326 LOG(error) << __LINE__ << ":" << __func__ << ", Stop requested. skip PostRun."; 327 return; 328 } 329 330 LOG(info) << __LINE__ << ":" << __func__; 331 if (fWorker) { 332 fWorker->Join(); 333 } 334 if (fChannels.count(fInputDataChannelName) > 0) { 335 auto n = fChannels.count(fInputDataChannelName); 336 for (auto i = 0u; i < n; ++i) { 337 for (auto itry = 0; itry < 10; ++itry) { 338 std::this_thread::sleep_for(std::chrono::milliseconds(200)); 339 if (fMultipart) { 340 FairMQParts part; 341 if (Receive(part, fInputDataChannelName, i, fMQTimeoutMS) > 0) { 342 // if (!fDiscard) { 343 // HandleMultipartData(part, i); 344 //} 345 } else { 346 break; 347 } 348 } else { 349 FairMQMessagePtr msg(NewMessage()); 350 if (Receive(msg, fInputDataChannelName, i, fMQTimeoutMS) > 0) { 351 // if (!fDiscard) { 352 // HandleData(msg, i); 353 //} 354 } else { 355 break; 356 } 357 } 358 } 359 } 360 } 361 362 fFileSinkTrailer.magic = FileSinkTrailer::MAGIC; 363 //fFileSinkTrailer.size = sizeof(FileSinkTrailer::Trailer); 364 fFileSinkTrailer.length = sizeof(FileSinkTrailer::Trailer); 365 fFileSinkTrailer.hLength = sizeof(FileSinkTrailer::Trailer); 366 fFileSinkTrailer.type = 0; 367 fFileSinkTrailer.fairMQDeviceType = fFileSinkHeader.fairMQDeviceType; 368 fFileSinkTrailer.runNumber = fFileSinkHeader.runNumber; 369 fFileSinkTrailer.startUnixtime = fFileSinkHeader.startUnixtime; 370 fFileSinkTrailer.stopUnixtime = time(0); 371 strcpy(fFileSinkTrailer.comments, run_comment.c_str()); 372 LOG(debug) << "FileSink::Trailer.magic : " << fFileSinkTrailer.magic; 373 LOG(debug) << "FileSink::Trailer.size : " << fFileSinkTrailer.size; 374 LOG(debug) << "FileSink::Trailer.fairMQDeviceType : " << fFileSinkTrailer.fairMQDeviceType; 375 LOG(debug) << "FileSink::Trailer.runNumber : " << fFileSinkTrailer.runNumber; 376 LOG(debug) << "FileSink::Trailer.startUnixtime : " << fFileSinkTrailer.startUnixtime; 377 LOG(debug) << "FileSink::Trailer.stopUnixtime : " << fFileSinkTrailer.stopUnixtime; 378 LOG(debug) << "FileSink::Trailer.comments : " << fFileSinkTrailer.comments; 379 fFile->Write(reinterpret_cast<char *>(&fFileSinkTrailer), sizeof(FileSinkTrailer::Trailer)); 380 381 fFile->Close(); 382 int64_t n = -1; 383 std::ostringstream ss; 384 ss << " nwrite = " << fNWrite << "\n"; 385 if (fFilePath.find("/dev/null") == std::string::npos) { 386 const auto &files = fFile->GetBranchFilePath(); 387 const auto &rawSize = fFile->GetBranchRawSize(); 388 const auto &compressedSize = fFile->GetBranchCompressedSize(); 389 const auto &numIteration = fFile->GetBranchNumIteration(); 390 391 n = 0; 392 if (!fWorker) { 393 fSize = 0; 394 fCompressedSize = 0; 395 } 396 for (auto i = 0u; i < files.size(); ++i) { 397 ss << " file: " << files[i] // 398 << ", n = " << numIteration[i] // 399 << ", compressed = " << compressedSize[i]; // 400 401 if (!fWorker) { 402 ss << ", raw = " << rawSize[i]; 403 fSize += rawSize[i]; 404 fCompressedSize += compressedSize[i]; 405 } 406 407 ss << "\n"; 408 n += stdfs::file_size(files[i]); 409 } 410 } 411 double r = (fSize > 0) ? static_cast<double>(fCompressedSize) / fSize : 0.0; 412 // double realMB = n/1e6; 413 // double rawMB = fSize/1e6; 414 // ss << " raw size = " << fSize << " Bytes, compressed size = " << fCompressedSize << " Bytes processed. r = " << r; 415 ss << " comperssion ratio (after/before) = " << r; 416 ss << "\n file size (sum) = " << n; 417 LOG(info) << " run : " << fFile->GetRunNumber() << " stopped." << ss.str(); 418 419 fFile->ClearBranch(); 420 LOG(info) << __LINE__ << ":" << __func__ << " done"; 421 } 422 //______________________________________________________________________________ 423 void FileSink::PreRun() 424 { 425 if (fStopRequested) { 426 LOG(error) << __LINE__ << ":" << __func__ << ", Stop requested. skip PreRun."; 427 return; 428 } 429 430 LOG(info) << __LINE__ << ":" << __func__; 431 fNReceived = 0; 432 fNWrite = 0; 433 fStopRequested = false; 434 fSize = 0; 435 fCompressedSize = 0; 436 437 if (fConfig->Count(opt::RunNumber.data())) { 438 try { 439 fRunNumber = std::stoll(fConfig->GetProperty<std::string>(opt::RunNumber.data())); 440 } catch (const std::exception &e) { 441 LOG(error) << "Invalid Run number: " << e.what(); 442 //ChangeState(fair::mq::Transition::Stop); 443 fStopRequested = true; 444 } catch (...) { 445 LOG(error) << "Invalid Run number: " << fConfig->GetProperty<std::string>(opt::RunNumber.data()); 446 //ChangeState(fair::mq::Transition::Stop); 447 fStopRequested = true; 448 } 449 LOG(debug) << " Run number: " << fRunNumber; 450 } 451 if (fConfig->Count("registry-uri")) { 452 std::string registryUri = fConfig->GetProperty<std::string>("registry-uri"); 453 LOG(debug) << "registryUri: " << registryUri; 454 if (!registryUri.empty()) { 455 fClient = std::make_shared<sw::redis::Redis>(registryUri); 456 std::string key; 457 fClient->keys("run_info:run_comment", &key); 458 if (key.length()>0) { 459 auto run_comment_ptr = fClient->get("run_info:run_comment"); 460 run_comment = *run_comment_ptr; 461 LOG(debug) << "run comment: " << run_comment; 462 }else{ 463 LOG(debug) << "There is no key run_info:run_comment"; 464 } 465 } 466 } 467 if (run_comment.length() > 255) { 468 run_comment.resize(255); // The max bytes of the comment are 256 bytes including a null charactor of 1 bytes. 469 } 470 471 fFile->SetRunNumber(fRunNumber); 472 fFile->ClearBranch(); 473 fFile->Open(); 474 fFilePath = fFile->GetFilePath(); 475 if (fWorker) { 476 // FileUtil writes data compressed by worker threads of FileSink. 477 fFile->SetCompressFunc([this](const char *src, int n) -> std::vector<char> { 478 return std::move(Compressor::Compress(src, n, Compressor::Format::none, fFile->GetCompBufSize())); 479 }); 480 LOG(debug) << " setup threads: n = " << fNThreads; 481 LOG(debug) << " file extension = " << fFileExtension << ", !compress = " << (!fCompress); 482 fWorker->InitChannels(fNThreads, fInProcMQLength, fMQTimeoutMS); 483 484 if (!fWorker->fHandleInput && !fMultipart) { 485 LOG(debug) << " set single message handler for input data"; 486 fWorker->fHandleInput = [this](auto &msg, auto index) { 487 return CompressData(msg, index); 488 }; 489 } 490 if (!fWorker->fHandleInputMultipart) { 491 LOG(debug) << " set multipart message handler for input data"; 492 fWorker->fHandleInputMultipart = [this](auto &msgParts, auto index) { 493 return CompressMultipartData(msgParts, index); 494 }; 495 } 496 if (!fWorker->fHandleOutput && !fMultipart) { 497 LOG(debug) << " set single message handler for output data"; 498 fWorker->fHandleOutput = [this](auto &msg, auto index) { 499 return WriteData(msg, index); 500 }; 501 } 502 if (!fWorker->fHandleOutputMultipart) { 503 LOG(debug) << " set multipart message handler for output data"; 504 fWorker->fHandleOutputMultipart = [this](auto &msgParts, auto index) { 505 return WriteMultipartData(msgParts, index); 506 }; 507 } 508 fWorker->Run(); 509 } 510 511 fFileSinkHeader.magic = FileSinkHeader::MAGIC; 512 //fFileSinkHeader.size = sizeof(FileSinkHeader::Header); 513 fFileSinkHeader.length = sizeof(FileSinkHeader::Header); 514 fFileSinkHeader.hLength = sizeof(FileSinkHeader::Header); 515 fFileSinkHeader.type = 0; 516 fFileSinkHeader.fairMQDeviceType = 99; 517 fFileSinkHeader.runNumber = fRunNumber; 518 fFileSinkHeader.startUnixtime = time(0); 519 fFileSinkHeader.stopUnixtime = 0; 520 strcpy(fFileSinkHeader.comments, run_comment.c_str()); 521 LOG(debug) << "FileSink::Header.magic : " << fFileSinkHeader.magic; 522 //LOG(debug) << "FileSink::Header.size : " << fFileSinkHeader.size; 523 LOG(debug) << "FileSink::Header.hLength : " << fFileSinkHeader.hLength; 524 LOG(debug) << "FileSink::Header.fairMQDeviceType : " << fFileSinkHeader.fairMQDeviceType; 525 LOG(debug) << "FileSink::Header.runNumber : " << fFileSinkHeader.runNumber; 526 LOG(debug) << "FileSink::Header.startUnixtime : " << fFileSinkHeader.startUnixtime; 527 LOG(debug) << "FileSink::Header.stopUnixtime : " << fFileSinkHeader.stopUnixtime; 528 LOG(debug) << "FileSink::Header.comments : " << fFileSinkHeader.comments; 529 fFile->Write(reinterpret_cast<char *>(&fFileSinkHeader), sizeof(FileSinkHeader::Header)); 530 } 531 532 //______________________________________________________________________________ 533 bool FileSink::WriteData(FairMQMessagePtr &msg, int index) 534 { 535 (void)index; 536 537 if (fStopRequested) { 538 return false; 539 } 540 // LOG(info) << __LINE__ << ":" << __func__ << " index = " << index << " n-received = " << fNReceived << ", n-write = 541 // " << fNWrite; 542 fFile->Write(reinterpret_cast<char *>(msg->GetData()), msg->GetSize()); 543 ++fNWrite; 544 // LOG(info) << __LINE__ << ":" << __func__ << " : done : n-received = " << fNReceived << ", n-write = " << fNWrite; 545 if (fWriteSleepInMilliSec > 0) { 546 std::this_thread::sleep_for(std::chrono::milliseconds(fWriteSleepInMilliSec)); 547 LOG(info) << __LINE__ << ":" << __func__ << " : fWriteSleepInMilliSec = " << fWriteSleepInMilliSec << " msec"; 548 } 549 550 if ((fMaxIteration > 0) && (fNWrite == fMaxIteration)) { 551 LOG(info) << " number of WriteData() reached the max iteration. n-write = " << fNWrite 552 << " max iteration = " << fMaxIteration << ". state transition : stop"; 553 // ChangeState(fair::mq::Transition::Stop); 554 fStopRequested = true; 555 } 556 557 return !fStopRequested; 558 } 559 560 //______________________________________________________________________________ 561 bool FileSink::WriteMultipartData(FairMQParts &msgParts, int index) 562 { 563 (void)index; 564 565 if (fStopRequested) { 566 return false; 567 } 568 auto len = MessageUtil::TotalLength(msgParts); 569 // LOG(info) << __LINE__ << ":" << __func__ 570 // << " index = " << index 571 // << " n-received = " << fNReceived 572 // << " parts.Sze() = " << msgParts.Size() 573 // << " total length = " << len << " bytes" 574 // << ", n-write = " << fNWrite; 575 576 #if 0 577 for (const auto &m : msgParts) { 578 auto first = reinterpret_cast<const char *>(m->GetData()); 579 auto last = first + m->GetSize(); 580 LOG(debug) << "message nbytes = " << m->GetSize(); 581 std::for_each(first, last, HexDump())); 582 } 583 #endif 584 585 if (fMergeMessage) { 586 std::vector<char> v; 587 v.reserve(len); 588 for (auto &msg : msgParts) { 589 auto pbegin = reinterpret_cast<char *>(msg->GetData()); 590 auto pend = pbegin + msg->GetSize(); 591 v.insert(v.end(), std::make_move_iterator(pbegin), std::make_move_iterator(pend)); 592 } 593 fFile->Write(v.data(), v.size()); 594 ++fNWrite; 595 if (fWriteSleepInMilliSec > 0) { 596 std::this_thread::sleep_for(std::chrono::milliseconds(fWriteSleepInMilliSec)); 597 LOG(info) << __LINE__ << ":" << __func__ << " : fWriteSleepInMilliSec = " << fWriteSleepInMilliSec << " msec"; 598 } 599 } else { 600 for (auto &msg : msgParts) { 601 fFile->Write(reinterpret_cast<char *>(msg->GetData()), msg->GetSize()); 602 ++fNWrite; 603 if (fWriteSleepInMilliSec > 0) { 604 std::this_thread::sleep_for(std::chrono::milliseconds(fWriteSleepInMilliSec)); 605 LOG(info) << __LINE__ << ":" << __func__ << " : fWriteSleepInMilliSec = " << fWriteSleepInMilliSec << " msec"; 606 } 607 } 608 } 609 610 if ((fMaxIteration > 0) && (fNWrite == fMaxIteration)) { 611 LOG(info) << " number of WriteData() reached the max iteration. n-write = " << fNWrite 612 << " max iteration = " << fMaxIteration << ". state transition : stop"; 613 // ChangeState(fair::mq::Transition::Stop); 614 fStopRequested = true; 615 } 616 return !fStopRequested; 617 } 618 } // namespace nestdaq
FileSinkHeader.h
SHA-256: f75e6781e1482ff20be73d60941c0e3056d460459583cfa0c4d954b069d85f7b
1 #ifndef FileSinkHeader_h 2 #define FileSinkHeader_h 3 4 #include <ctime> 5 #include <cstdint> 6 7 #pragma pack(4) 8 namespace FileSinkHeader { 9 10 namespace v0 { 11 constexpr uint64_t MAGIC = {0x444145482d534640}; /* "@FS-HEAD" */ 12 13 struct Header { /* Total size: 304 bytes */ 14 uint64_t magic {MAGIC}; /* 64 bits = 8 bytes */ 15 uint64_t size {0}; /* 64 bits = 8 bytes */ 16 uint64_t fairMQDeviceType {0}; /* 64 bits = 8 bytes */ 17 uint64_t runNumber {0}; /* 64 bits = 8 bytes */ 18 time_t startUnixtime {0}; /* 64 bits = 8 bytes */ 19 time_t stopUnixtime {0}; /* 64 bits = 8 bytes */ 20 char comments[256]; /* 8 bits x 256 = 256 bytes*/ 21 }; 22 } //namespace v0 23 24 inline namespace v1 { 25 constexpr uint64_t MAGIC = {0x004b4e53454c4946}; /* "FILESNK " */ 26 27 struct Header { /* Total size: 304 bytes */ 28 uint64_t magic {MAGIC}; /* 64 bits = 8 bytes */ 29 union { 30 uint64_t size ; /* 64 bits = 8 bytes */ 31 struct { 32 uint32_t length {0x130}; 33 uint16_t hLength {0x130}; 34 uint16_t type {0}; 35 }; 36 }; 37 uint64_t fairMQDeviceType {0}; /* 64 bits = 8 bytes */ 38 uint64_t runNumber {0}; /* 64 bits = 8 bytes */ 39 time_t startUnixtime {0}; /* 64 bits = 8 bytes */ 40 time_t stopUnixtime {0}; /* 64 bits = 8 bytes */ 41 char comments[256]; /* 8 bits x 256 = 256 bytes*/ 42 }; 43 } //namespace v0 44 45 46 } //namespace FileSinkHeader 47 #pragma pack() 48 49 #endif
FileSinkTrailer.h
SHA-256: 29d2b5b18bd7fc673891ab45720e1b8d67fca9b4aa0d3c23a1f5c54c0052d6e1
1 #ifndef FileSinkTrailer_h 2 #define FileSinkTrailer_h 3 4 #include <ctime> 5 #include <cstdint> 6 7 namespace FileSinkTrailer { 8 9 namespace v0 { 10 constexpr uint64_t MAGIC = {0x494152542d534640}; /* "@FS-TRAI" */ 11 12 struct Trailer { /* Total size: 304 bytes */ 13 uint64_t magic {MAGIC}; /* 64 bits = 8 bytes */ 14 uint64_t size {0}; /* 64 bits = 8 bytes */ 15 uint64_t fairMQDeviceType {0}; /* 64 bits = 8 bytes */ 16 uint64_t runNumber {0}; /* 64 bits = 8 bytes */ 17 time_t startUnixtime {0}; /* 64 bits = 8 bytes */ 18 time_t stopUnixtime {0}; /* 64 bits = 8 bytes */ 19 char comments[256]; /* 8 bits x 256 = 256 bytes*/ 20 }; 21 } 22 23 inline namespace v1 { 24 constexpr uint64_t MAGIC = {0x004c5254454c4946}; /* "FILETRL " */ 25 26 struct Trailer { /* Total size: 304 bytes */ 27 uint64_t magic {MAGIC}; /* 64 bits = 8 bytes */ 28 union { 29 uint64_t size ; /* 64 bits = 8 bytes */ 30 struct { 31 uint32_t length {0x130}; /* 32 bits = 8 bytes */ 32 uint16_t hLength {0x130}; /* 32 bits = 8 bytes */ 33 uint16_t type {0}; /* 32 bits = 8 bytes */ 34 }; 35 }; 36 uint64_t fairMQDeviceType {0}; /* 64 bits = 8 bytes */ 37 uint64_t runNumber {0}; /* 64 bits = 8 bytes */ 38 time_t startUnixtime {0}; /* 64 bits = 8 bytes */ 39 time_t stopUnixtime {0}; /* 64 bits = 8 bytes */ 40 char comments[256]; /* 8 bits x 256 = 256 bytes*/ 41 }; 42 } 43 44 } // namespace FileSinkTrailer 45 46 #endif
FilterHeader.h
SHA-256: 1d3643ad036ce38aa06f437679a4d1addf9c0ed3b8e0c6989c8478109f2851a9
1 #ifndef FilterHeader_h 2 #define FilterHeader_h 3 4 #include <cstdint> 5 #include <sys/time.h> 6 7 namespace Filter { 8 namespace v0 { 9 // "NIOC-TLF" : little endian of "FLT-COIN" 10 constexpr uint64_t MAGIC {0x4e494f43'2d544c46}; 11 12 struct Header { 13 uint64_t magic {MAGIC}; 14 uint64_t length {0}; 15 uint32_t numTrigs {0}; 16 uint32_t workerId {0}; 17 uint32_t elapseTime {0}; 18 //struct timeval processTime {0, 0}; 19 uint64_t timeSec {0}; 20 uint64_t timeUSec {0}; 21 }; 22 } //namespace v0 23 24 inline namespace v1 { 25 #pragma pack(4) 26 27 // " IGOLTLF" : little endian of "FLTLOGI " 28 static constexpr uint64_t MAGIC {0x0049474f'4c544c46}; 29 30 struct Header { 31 uint64_t magic {MAGIC}; 32 uint32_t length {0}; 33 uint16_t hLength {56}; 34 uint16_t type {0}; 35 uint32_t timeFrameId {0}; 36 uint32_t numTrigs {0}; 37 uint32_t workerId {0}; 38 uint32_t numMessages {0}; 39 uint32_t elapseTime {0}; 40 uint32_t reserve {0xff00ff00}; 41 //struct timeval processTime {0, 0}; 42 uint64_t timeSec {0}; 43 uint64_t timeUSec {0}; 44 }; 45 46 // " EMITGRT" 47 static constexpr uint64_t TDC_MAGIC {0x00454d49'54475254}; 48 struct TrgTimeHeader { 49 union { 50 uint32_t u32data[4]; 51 struct { 52 uint64_t magic {TDC_MAGIC}; 53 uint32_t length {0}; 54 uint16_t hLength {16}; 55 uint16_t type {0}; 56 //uint32_t timeFrameId {0}; 57 }; 58 }; 59 }; 60 61 constexpr unsigned char FLT_TDC_TYPE {0xaa}; 62 struct TrgTime { 63 union { 64 struct { 65 uint32_t time; 66 uint32_t type; 67 }; 68 unsigned char cdata[8] {0, 0, 0, 0, 0, 0, 0, FLT_TDC_TYPE}; 69 }; 70 }; 71 72 #pragma pack() 73 } //namespace v1 74 75 } // namespace Filter 76 77 #endif
FilterTimeFrameSliceABC.icxx
SHA-256: 7c63b2c5b335265be1001787061141340c01f4e778c5260f1791c66e96e0dd2c
1 #include "FilterTimeFrameSliceABC.h" 2 #include "utility/MessageUtil.h" 3 #include "UnpackTdc.h" 4 #include <chrono> 5 6 #define Max_Wiremap 850 7 #define DEBUG 1 8 #define OUTPUT 1 9 10 using nestdaq::FilterTimeFrameSliceABC; 11 namespace bpo = boost::program_options; 12 13 FilterTimeFrameSliceABC::FilterTimeFrameSliceABC() 14 : fKTimer(1,KTimer(1000)) 15 { 16 } 17 18 int FilterTimeFrameSliceABC::geoToIndex(uint64_t geo) { 19 // change here ! Detector geometry 20 std::vector<uint64_t> geoValues = { 21 0xc0a802a1, 0xc0a802a2, 0xc0a802a3, 0xc0a802a4, 22 0xc0a802a5, 0xc0a802a6, 0xc0a802a7, 0xc0a802a8 23 }; 24 25 auto it = std::find(geoValues.begin(), geoValues.end(), geo); 26 if (it != geoValues.end()) { 27 return std::distance(geoValues.begin(), it); 28 } 29 return -1; 30 } 31 32 void FilterTimeFrameSliceABC::InitTask() 33 { 34 using opt = OptionKey; 35 36 fInputChannelName = fConfig->GetValue<std::string>(opt::InputChannelName.data()); 37 fOutputChannelName = fConfig->GetValue<std::string>(opt::OutputChannelName.data()); 38 fName = fConfig->GetProperty<std::string>("id"); 39 std::istringstream ss(fName.substr(fName.rfind("-") + 1)); 40 ss >> fId; 41 42 fNumDestination = GetNumSubChannels(fOutputChannelName); 43 fPollTimeoutMS = std::stoi(fConfig->GetProperty<std::string>(opt::PollTimeout.data())); 44 45 // Initialize Wire_map array 46 wireMapArray.fill(std::array<Wire_map, maxCh + 1>{}); 47 48 std::ifstream file("vdc_amaneq_gr.map");// Loading VDC_Wire_map 49 if (!file) { 50 std::cerr << "Cannot open file " << std::endl; 51 return; 52 } 53 54 std::string line; 55 std::getline(file, line); // Skip the first line (header). 56 57 while (std::getline(file, line)) { 58 std::istringstream iss(line); 59 Wire_map record; 60 if (!(iss >> record.catid >> record.id >> record.sh >> record.fp >> record.det >> std::hex >> record.geo >> std::dec >> record.ch)) { 61 std::cerr << "Data read error: " << line << std::endl; 62 continue; 63 } 64 65 int geoIndex = geoToIndex(record.geo); 66 #if DEBUG 67 std::cout << "Debug for loading Vdc Channnel - geoindex: " << geoIndex << std::endl; 68 std::cout << "Debug for record.geo outside of function: " << record.geo << std::endl; 69 #endif 70 if (geoIndex != -1 && record.ch <= maxCh) { 71 wireMapArray[geoIndex][record.ch] = record; 72 } else { 73 std::cerr << "Invalid geo or ch value: " << line << std::endl; 74 } 75 } 76 #if DEBUG 77 std::cout << "Printing wireMapArray contents after loading the file:" << std::endl; 78 for (size_t i = 0; i < wireMapArray.size(); ++i) { 79 for (size_t j = 0; j < wireMapArray[i].size(); ++j) { 80 const Wire_map& wire = wireMapArray[i][j]; 81 if (wire.catid != -1) { 82 std::cout << "geoIndex: " << i << ", ch: " << j 83 << " - catid: " << wire.catid 84 << ", id: " << wire.id 85 << ", sh: " << wire.sh 86 << ", fp: " << wire.fp 87 << ", det: " << wire.det 88 << ", geo: " << std::hex << wire.geo << std::dec 89 << ", ch: " << wire.ch << std::endl; 90 } 91 } 92 } 93 #endif 94 } 95 96 void FilterTimeFrameSliceABC::PreRun() 97 { 98 } 99 100 bool FilterTimeFrameSliceABC::ConditionalRun() 101 { 102 using namespace std::chrono; 103 std::cout << "count Number " << std::endl; 104 //Timer start 105 auto start = high_resolution_clock::now(); 106 107 #if DEBUG 108 //debug for loading wire-map 109 for (size_t i = 0; i < wireMapArray.size(); ++i) { 110 for (size_t j = 0; j < wireMapArray[i].size(); ++j) { 111 const Wire_map& wire = wireMapArray[i][j]; 112 if (wire.catid != -1) { 113 std::cout << "geoIndex: " << i << ", ch: " << j << " - catid: " << wire.catid << ", id: " << wire.id << ", geo: " << std::hex << wire.geo << std::dec << std::endl; 114 } 115 } 116 } 117 #endif 118 119 fDoCheck = fKTimer[0].Check(); 120 for (auto& tf : fTFs) { 121 tf.Clear(); 122 } 123 fTFs.clear(); 124 125 FairMQParts inParts; 126 FairMQParts outParts; 127 if (Receive(inParts, fInputChannelName,0,1) <= 0) return true; 128 129 // Calculate the input data size 130 size_t inputDataSize = 0; 131 for (uint32_t i = 0; i < inParts.Size(); i++) { 132 inputDataSize += inParts[i].GetSize(); 133 std::cout << "inputDataSize: " << inputDataSize << ", inParts: " << inParts[i].GetSize() << std::endl; 134 } 135 136 if (fDoCheck) { 137 LOG(info) << "ConditionalRun"; 138 } 139 140 ParseMessages(inParts); 141 142 using copyUnit = uint32_t; 143 144 auto outdataptr = std::make_unique<std::vector<copyUnit>>(); 145 auto outdata = outdataptr.get(); 146 for (uint64_t ifrm = 0, nfrm = fTFs.size(); ifrm < nfrm; ++ifrm) { 147 auto& tf = fTFs[ifrm]; 148 bool doKeep = false; 149 if (tf.GetHeader()->type == TimeFrame::META) { 150 doKeep = true; 151 } else { 152 doKeep = ProcessSlice(tf); 153 } 154 if (!doKeep) continue; 155 tf.GetHeader()->length = tf.GetRealLength(); 156 tf.CopyAllTo<copyUnit>(outdata); 157 } 158 159 FairMQMessagePtr msgall(MessageUtil::NewMessage(*this, std::move(outdataptr))); 160 outParts.AddPart(std::move(msgall)); 161 162 if (fDoCheck) { 163 } 164 165 auto poller = NewPoller(fOutputChannelName); 166 while (!NewStatePending()) { 167 auto direction = (fDirection++) % fNumDestination; 168 poller->Poll(fPollTimeoutMS); 169 if (poller->CheckOutput(fOutputChannelName, direction)) { 170 if (Send(outParts, fOutputChannelName, direction) > 0) { 171 break; 172 } else { 173 LOG(error) << "Failed to queue output-channel"; 174 } 175 } 176 if (fNumDestination == 1) { 177 std::this_thread::sleep_for(std::chrono::milliseconds(1)); 178 } 179 } 180 181 #if OUTPUT 182 // Obtains processing end time and calculates processing time and data volume 183 auto end = high_resolution_clock::now(); 184 auto duration = duration_cast<microseconds>(end - start).count(); // processing time (μs) 185 186 // Get the amount of data processed (e.g., size of outdata) and calculate the amount of data processed per unit time 187 double throughput = (inputDataSize / 1024.0) / (duration / 1000.0); // KB/s 188 189 std::cout << "Duration: " << duration << " μs, DataSize: " << inputDataSize << " bytes, Throughput: " << throughput << " KB/s" << std::endl; 190 191 // Open the TXT File 192 std::ofstream outFile("FilterPerformance.txt", std::ios::app); // 追記モード 193 if (outFile.is_open()) { 194 outFile << "Duration: " << duration << " μs, DataSize: " << inputDataSize 195 << " bytes, Throughput: " << throughput << " KB/s" << std::endl; 196 outFile.close(); 197 } else { 198 std::cerr << "Error opening file for writing" << std::endl; 199 } 200 #endif 201 202 return true; 203 } 204 205 void FilterTimeFrameSliceABC::PostRun() 206 { 207 } 208 209 bool FilterTimeFrameSliceABC::ParseMessages(FairMQParts& inParts) 210 { 211 // parse all the messages 212 int stfIdx = 0; // array index for stf 213 int hbfIdx = 0; // array index for hbf 214 int fltIdx = 0; // array index for flt 215 TTF *timeFrame = nullptr; 216 Header<struct TimeFrame::Header> tfh; 217 Header<struct Filter::Header> lfh; 218 Header<struct Filter::TrgTimeHeader> tth; 219 for (uint32_t i = 0, n = inParts.Size(); i < n; ++i) { 220 auto& part = inParts[i]; 221 char* dataptr = reinterpret_cast<char*>(part.GetData()); 222 uint64_t nw = part.GetSize(); 223 char* dataend = dataptr + nw; 224 225 while (dataptr < dataend) { 226 auto magic = *reinterpret_cast<uint64_t*>(dataptr); 227 if (magic == TimeFrame::MAGIC) { 228 // add new timeframe 229 fTFs.resize(fTFs.size() + 1); 230 timeFrame = &(fTFs[fTFs.size() - 1]); 231 timeFrame->SetHeader(dataptr); 232 stfIdx = -1; // reset stf index 233 auto ns = timeFrame->GetHeader()->numSource; 234 if (timeFrame->size() < ns) { 235 timeFrame->resize(ns); 236 for (decltype(ns) is = 0; is < ns; ++is) { 237 if ((*timeFrame)[is] == nullptr) { 238 (*timeFrame)[is] = new TSTF; 239 } 240 } 241 } 242 dataptr += timeFrame->GetHeader()->hLength; 243 } else if (magic == Filter::MAGIC) { 244 // do nothing 245 lfh.SetHeader(dataptr); 246 dataptr += lfh.GetHeader()->hLength; 247 #if 0 248 fLF.SetHeader(part.GetData()); 249 auto nm = fLF.GetHeader()->numMessages - 1; 250 if (fLF.size() < nm) { 251 fLF.resize(nm); 252 for (decltype(nm) im = 0; im < nm; ++im) { 253 if (fLF[im] == nullptr) fLF[im] = new TTT; 254 } 255 } 256 #endif 257 } else if (magic == Filter::TDC_MAGIC) { 258 // do nothing 259 tth.SetHeader(dataptr); 260 dataptr += tth.GetHeader()->length; 261 #if 0 262 auto& tdcf = *(fLF[fltIdx]); 263 tdcf.Set(part.GetData()); 264 fltIdx++; 265 #endif 266 } else if (magic == SubTimeFrame::MAGIC) { 267 hbfIdx = 0; // reset hbf index 268 stfIdx++; 269 auto& stf = *((*timeFrame)[stfIdx]); 270 stf.SetHeader(dataptr); 271 dataptr += stf.GetHeader()->hLength; 272 } else if (magic == HeartbeatFrame::MAGIC) { 273 auto& stf = *((*timeFrame)[stfIdx]); 274 if (stf.size() < hbfIdx + 1) { 275 stf.push_back(new THBF); 276 } 277 auto& hbf = *((*timeFrame)[stfIdx]->at(hbfIdx)); 278 hbf.Set(dataptr); 279 hbfIdx++; 280 dataptr += hbf.GetHeader()->length; 281 } else { 282 //LOG(info) << "Error with magic" << std::hex << magic; 283 return false; 284 } 285 } 286 } 287 return true; 288 }
FilterTimeFrameSliceBySomething.cxx
SHA-256: 9b98848c2fc4899fdf489cb2a676d4bfa5cd07bd734c592f43b28bc66ea21663
1 /* 2 * @file FilterTimeFrameSliceBySomething 3 * @brief Slice Timeframe by Logic timing for NestDAQ 4 * @date Created : 2024-05-04 12:31:55 JST 5 * Last Modified : 2024-05-23 07:12:22 JST 6 * 7 * @author Shinsuke OTA <ota@rcnp.osaka-u.ac.jp> 8 * 9 */ 10 #include "FilterTimeFrameSliceBySomething.h" 11 #include "FilterTimeFrameSliceABC.icxx" 12 #include "fairmq/runDevice.h" 13 14 #include "utility/MessageUtil.h" 15 #include "UnpackTdc.h" 16 17 #define DEBUG 0 18 19 using nestdaq::FilterTimeFrameSliceBySomething; 20 namespace bpo = boost::program_options; 21 22 23 24 25 FilterTimeFrameSliceBySomething::FilterTimeFrameSliceBySomething() 26 { 27 } 28 29 bool FilterTimeFrameSliceBySomething::ProcessSlice(TTF& tf) 30 { 31 #if 0 32 int doKeep = false; 33 34 if (tf[0]->GetHeader()->femid == /* some module */) { 35 auto& hbf = tf[0]->at(0)); // 36 uint64_t nData = hbf->GetNumData(); 37 for (int i = 0; i < nData; ++i) { 38 // unpack data 39 // Unpack(hbf->UncheckedAt(i),tdc); 40 // 41 } 42 // judge 43 doKeep = true; 44 } 45 46 if (!doKeep) { 47 return false; 48 } 49 #endif 50 return true; 51 } 52 53 54 55 //////////////////////////////////////////////////// 56 // override runDevice 57 //////////////////////////////////////////////////// 58 59 void addCustomOptions(bpo::options_description& options) 60 { 61 using opt = FilterTimeFrameSliceBySomething::OptionKey; 62 63 options.add_options() 64 (opt::InputChannelName.data(), 65 bpo::value<std::string>()->default_value("in"), 66 "Name of the input channel") 67 (opt::OutputChannelName.data(), 68 bpo::value<std::string>()->default_value("out"), 69 "Name of the output channel") 70 (opt::DQMChannelName.data(), 71 bpo::value<std::string>()->default_value("dqm"), 72 "Name of the data quality monitoring channel") 73 (opt::PollTimeout.data(), 74 bpo::value<std::string>()->default_value("1"), 75 "Timeout of polling (in msec)") 76 (opt::SplitMethod.data(), 77 bpo::value<std::string>()->default_value("1"), 78 "STF split method") 79 ; 80 81 } 82 83 84 85 std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /*config*/) 86 { 87 return std::make_unique<FilterTimeFrameSliceBySomething>(); 88 }
FrameContainer.h
SHA-256: b93c752d4f804c1d6df2ba917a51cd9b71989473bf43bacac508b484bf7e148a
1 /** 2 * @file FrameContainer.h 3 * @brief Frame Structure Container 4 * @date Created : 2024-05-04 12:27:57 JST 5 * Last Modified : 2024-06-30 18:56:56 JST 6 * 7 * @author Shinsuke OTA <ota@rcnp.osaka-u.ac.jp> 8 * 9 */ 10 #ifndef NESTDAQ_FRAMECONTAINEW_H 11 #define NESTDAQ_FRAMECONTAINEW_H 12 13 #include "SubTimeFrameHeader.h" 14 #include "TimeFrameHeader.h" 15 #include "FilterHeader.h" 16 #include "HeartbeatFrameHeader.h" 17 #include <iostream> 18 19 namespace nestdaq { 20 template <typename frametype> class Header { 21 public: 22 void SetHeader(void* data) { 23 this->fHeader = static_cast<frametype*>(data); 24 } 25 frametype* GetHeader() { return this->fHeader; } 26 template<typename unit> 27 void CopyHeaderTo(std::vector<unit>* output) { 28 for (uint32_t i = 0, n = sizeof(frametype)/sizeof(unit); i < n; ++i) { 29 output->push_back(*((unit*)fHeader+i)); 30 } 31 } 32 protected: 33 frametype *fHeader; 34 }; 35 36 template <class frametype, class datatype> 37 class DataFrame : public Header<frametype> { 38 public: 39 void Set(void* data) { 40 this->SetHeader(data); 41 fData = reinterpret_cast<datatype*>((char*)data + sizeof(frametype)); 42 auto len = this->GetHeader()->length; 43 auto hlen = this->GetHeader()->hLength; 44 fNumData = (len > hlen) ? (len - hlen )/sizeof(datatype) : 0; 45 } 46 47 datatype UncheckedAt(uint64_t idx) { return *(fData+idx); } 48 auto GetNumData() const { return fNumData; } 49 auto GetData() { return fData; } 50 template <typename unit> 51 void CopyDataTo(std::vector<unit>* output) { 52 auto nj = sizeof(datatype)/sizeof(unit); 53 for (uint32_t i =0, n = fNumData; i < n; ++i) { 54 for (uint32_t j = 0 ; j < nj; ++j) { 55 output->push_back(*((unit*)(fData+i) + j)); 56 } 57 } 58 } 59 60 template <typename unit> 61 void CopyDataTo(std::vector<unit>* output, uint64_t idx) { 62 auto nj = sizeof(datatype)/sizeof(unit); 63 for (uint32_t j = 0; j < nj; ++j) { 64 output->push_back(*((unit*)(fData+idx)+j)); 65 } 66 } 67 68 template <typename unit> 69 void CopyAllTo(std::vector<unit>* output) { 70 //std::cout << std::hex << this->GetHeader()->magic << " n data = " << this->GetNumData() << std::endl; 71 this->template CopyHeaderTo<unit>(output); 72 this->template CopyDataTo<unit>(output); 73 } 74 75 uint64_t GetRealLength() { 76 return sizeof(frametype) + fNumData * sizeof(datatype); 77 } 78 79 void Clear() { 80 ; 81 } 82 83 protected: 84 datatype* fData; 85 uint64_t fNumData; 86 }; 87 88 89 template <class frametype, class child> 90 class ContainerFrame : public Header<frametype>, public std::vector<child*> { 91 public: 92 uint64_t GetRealLength() { 93 uint64_t length = sizeof(frametype); 94 for (int i = 0, n = this->size(); i<n; ++i) { 95 length += this->at(i)->GetRealLength(); 96 } 97 return length; 98 } 99 template <typename unit> void CopyAllTo(std::vector<unit>* output) { 100 this->template CopyHeaderTo<unit>(output); 101 //std::cout << std::hex << this->GetHeader()->magic << std::endl; 102 for (int i = 0,n = this->size(); i < n; ++i) { 103 this->at(i)->template CopyAllTo<unit>(output); 104 } 105 } 106 107 void Clear() { 108 std::for_each(this->begin(),this->end(), 109 [] (child* c) { 110 c->Clear(); 111 delete c; 112 }); 113 this->std::vector<child*>::clear(); 114 } 115 116 protected: 117 void clear(); 118 }; 119 120 121 using THBF = DataFrame<struct HeartbeatFrame::Header,uint64_t>; 122 using TSTF = ContainerFrame<struct SubTimeFrame::Header,THBF>; 123 124 using TTT = DataFrame<struct Filter::TrgTimeHeader,struct Filter::TrgTime>; 125 using TLF = ContainerFrame<struct Filter::Header,TTT>; 126 using TTF = ContainerFrame<struct TimeFrame::Header,TSTF>; 127 128 } 129 #endif // NESTDAQ_FRAMECONTAINEW_H
HeartbeatFrameHeader.h
SHA-256: 93a95ef00b0323ab5bba3d3eee565f2308d6606b7183d7d5da0891b3d5ca1a38
1 #ifndef HBFHeader_H 2 #define HBFHeader_H 3 4 namespace HeartbeatFrame 5 { 6 // "HRTBEAT\0" -> "\0TAEBTRH" 7 constexpr uint64_t MAGIC = 0x0054414542545248; 8 #pragma pack(4) 9 struct Header { 10 uint64_t magic {MAGIC}; 11 uint32_t length {0}; 12 uint16_t hLength {16}; 13 uint16_t type {0}; 14 }; 15 #pragma pack() 16 } 17 18 #endif
LogicFilter/LogicFilter.cxx
SHA-256: d98184e39032f29ee9d096d0e112e76eea1b0cce7d25dad9f1fbd7b2147cf8f3
1 /******************************************************************************** 2 * Copyright (C) 2014 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH * 3 * * 4 * This software is distributed under the terms of the * 5 * GNU Lesser General Public Licence (LGPL) version 3, * 6 * copied verbatim in the file "LICENSE" * 7 ********************************************************************************/ 8 9 #include <fairmq/Device.h> 10 #include <fairmq/runDevice.h> 11 12 #include <iostream> 13 #include <iomanip> 14 #include <string> 15 #include <chrono> 16 17 #include <unordered_map> 18 #include <unordered_set> 19 20 #include <cstring> 21 22 #include "../utility/MessageUtil.h" 23 24 #include "../SubTimeFrameHeader.h" 25 #include "../TimeFrameHeader.h" 26 #include "../HeartbeatFrameHeader.h" 27 #include "../FilterHeader.h" 28 29 #include "SignalParser.cxx" 30 #include "Trigger.cxx" 31 #include "../KTimer.cxx" 32 33 34 //std::atomic<int> gQdepth = 0; 35 36 namespace bpo = boost::program_options; 37 38 struct LogicFilter : fair::mq::Device 39 { 40 struct OptionKey { 41 static constexpr std::string_view InputChannelName {"in-chan-name"}; 42 static constexpr std::string_view OutputChannelName {"out-chan-name"}; 43 static constexpr std::string_view DQMChannelName {"dqm-chan-name"}; 44 static constexpr std::string_view DataSuppress {"data-suppress"}; 45 static constexpr std::string_view RemoveHB {"remove-hb"}; 46 static constexpr std::string_view PollTimeout {"poll-timeout"}; 47 static constexpr std::string_view SplitMethod {"split"}; 48 49 static constexpr std::string_view TriggerSignals {"trigger-signals"}; 50 static constexpr std::string_view TriggerExpression {"trigger-expression"}; 51 static constexpr std::string_view TriggerWidth {"trigger-width"}; 52 }; 53 54 struct DataBlock { 55 uint32_t femId; 56 uint32_t Type; 57 uint32_t timeFrameId; 58 int HBFrame; 59 bool is_HB; 60 int msg_index; 61 int nTrig; 62 }; 63 64 LogicFilter() 65 { 66 // register a handler for data arriving on "data" channel 67 //OnData("in", &LogicFilter::HandleData); 68 69 fTrig = new Trigger(); 70 fKt1 = new KTimer(1000); 71 fKt2 = new KTimer(1000); 72 fKt3 = new KTimer(1000); 73 fKt4 = new KTimer(1000); 74 } 75 76 void InitTask() override; 77 #if 0 78 void InitTask() override 79 { 80 using opt = OptionKey; 81 82 // Get the fMaxIterations value from the command line options (via fConfig) 83 fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations"); 84 fInputChannelName = fConfig->GetValue<std::string>(opt::InputChannelName.data()); 85 fOutputChannelName = fConfig->GetValue<std::string>(opt::OutputChannelName.data()); 86 LOG(info) << "InitTask: Input Channel : " << fInputChannelName 87 << " Output Channel : " << fOutputChannelName; 88 89 //OnData(fInputChannelName, &LogicFilter::HandleData); 90 } 91 #endif 92 93 #if 0 94 bool HandleData(fair::mq::MessagePtr& msg, int) 95 { 96 LOG(info) << "Received: \"" 97 << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) 98 << "\""; 99 100 if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) { 101 LOG(info) << "Configured maximum number of iterations reached." 102 << " Leaving RUNNING state."; 103 return false; 104 } 105 106 // return true if you want the handler to be called again 107 // (otherwise return false go to the Ready state) 108 return true; 109 } 110 #endif 111 112 113 bool ConditionalRun() override; 114 void PostRun() override; 115 116 bool CheckData(fair::mq::MessagePtr&); 117 118 private: 119 int IsHartBeat(uint64_t, uint32_t); 120 int RemoveData(fair::mq::Parts &, int); 121 void CheckMultiPart(FairMQParts &); 122 int MakeBlockMap(FairMQParts &, 123 std::vector< std::vector<struct DataBlock> > &); 124 int MakeBlockMap2(FairMQParts &, 125 std::vector< std::vector<struct DataBlock> > &); 126 int BuildHBF( 127 std::vector<struct HBFIndex> &, 128 std::vector< std::vector<struct DataBlock> > &, 129 FairMQParts &, 130 int); 131 int AddFilterMessage( 132 FairMQParts &, 133 std::vector< std::vector<uint32_t> > &, 134 uint32_t, 135 uint32_t); 136 137 int MarkFlagSending( 138 std::vector< std::vector<struct DataBlock> > &block_map, 139 int i, 140 int nhits, 141 std::vector<bool> &flag_sending); 142 143 std::string fInputChannelName; 144 std::string fOutputChannelName; 145 std::string fDQMChannelName; 146 std::string fName; 147 148 int fNumDestination {0}; 149 uint32_t fDirection {0}; 150 int fPollTimeoutMS {0}; 151 int fSplitMethod {2}; 152 153 uint32_t fId {0}; 154 Trigger *fTrig; 155 bool fIsDataSuppress = true; 156 bool fIsRemoveHB = false; 157 158 int fHBflag = 0; 159 160 KTimer *fKt1; 161 KTimer *fKt2; 162 KTimer *fKt3; 163 KTimer *fKt4; 164 }; 165 166 167 void LogicFilter::InitTask() 168 { 169 using opt = OptionKey; 170 171 fInputChannelName = fConfig->GetValue<std::string>(opt::InputChannelName.data()); 172 fOutputChannelName = fConfig->GetValue<std::string>(opt::OutputChannelName.data()); 173 fDQMChannelName = fConfig->GetValue<std::string>(opt::DQMChannelName.data()); 174 175 LOG(info) << "InitTask: Input Channel : " << fInputChannelName 176 << " Output Channel : " << fOutputChannelName; 177 178 fNumDestination = GetNumSubChannels(fOutputChannelName); 179 fPollTimeoutMS = std::stoi(fConfig->GetProperty<std::string>(opt::PollTimeout.data())); 180 181 fName = fConfig->GetProperty<std::string>("id"); 182 std::istringstream ss(fName.substr(fName.rfind("-") + 1)); 183 ss >> fId; 184 185 fSplitMethod = std::stoi( 186 fConfig->GetProperty<std::string>(opt::SplitMethod.data())); 187 LOG(info) << "InitTask: SplitMethod : " << fSplitMethod; 188 189 std::string sIsDataSuppress = fConfig->GetValue<std::string>(opt::DataSuppress.data()); 190 if (sIsDataSuppress == "true") { 191 fIsDataSuppress = true; 192 } else { 193 fIsDataSuppress = false; 194 } 195 LOG(info) << "InitTask: DataSuppress : " << fIsDataSuppress; 196 197 std::string sIsRemoveHB = fConfig->GetValue<std::string>(opt::RemoveHB.data()); 198 if (sIsRemoveHB == "true") { 199 fIsRemoveHB = true; 200 } else { 201 fIsRemoveHB = false; 202 } 203 LOG(info) << "InitTask: RemoveHB : " << fIsRemoveHB; 204 205 206 fTrig->SetTimeRegion(1024 * 128); 207 fTrig->ClearEntry(); 208 209 std::string str_signals = fConfig->GetProperty<std::string>(opt::TriggerSignals.data()); 210 std::string expression = fConfig->GetProperty<std::string>(opt::TriggerExpression.data()); 211 int window_width = std::stoi(fConfig->GetProperty<std::string>(opt::TriggerWidth.data())); 212 213 std::vector< std::vector<uint32_t> > signals = SignalParser::Parsing(str_signals); 214 int i = 0; 215 for (auto &v : signals) { 216 if (v.size() >= 3) { 217 fTrig->Entry(v[0], v[1], static_cast<int>(v[2])); 218 union ipval { 219 uint32_t u32; 220 char c[4]; 221 }; 222 ipval mid; mid.u32 = v[0]; 223 //LOG(info) << "Module: " << std::hex << v[0] << ", Channel: " << v[1] << ", Offset: " << v[2]; 224 LOG(info) << std::setw(4) << i << ": " 225 << "M_id: " << static_cast<unsigned int>(mid.c[3] & 0xff) 226 << "." << static_cast<unsigned int>(mid.c[2] & 0xff) 227 << "." << static_cast<unsigned int>(mid.c[1] & 0xff) 228 << "." << static_cast<unsigned int>(mid.c[0] & 0xff) 229 << ", Ch.: " << std::setw(3) << v[1] << ", Offset: " << std::setw(6) << static_cast<int>(v[2]); 230 i++; 231 } 232 } 233 234 LOG(info) << "Expression: " << expression; 235 fTrig->MakeTable(expression); 236 237 LOG(info) << "Trigger windows width: " << window_width; 238 fTrig->SetMarkLen(window_width); 239 240 } 241 242 bool LogicFilter::CheckData(fair::mq::MessagePtr &msg) 243 { 244 unsigned int msize = msg->GetSize(); 245 unsigned char *pdata = reinterpret_cast<unsigned char *>(msg->GetData()); 246 uint64_t msg_magic = *(reinterpret_cast<uint64_t *>(pdata)); 247 248 static int fe_type = 0; 249 250 std::cout << "#Msg Top(8B): " << std::hex << msg_magic 251 << " Size: " << std::dec << msize << std::endl; 252 253 if (msg_magic == TimeFrame::MAGIC) { 254 TimeFrame::Header *ptf = reinterpret_cast<TimeFrame::Header *>(pdata); 255 std::cout << "#TF Header " 256 << std::hex << std::setw(16) << std::setfill('0') << ptf->magic 257 << " id: " << std::setw(8) << std::setfill('0') << ptf->timeFrameId 258 << " Nsource: " << std::setw(8) << std::setfill('0') << ptf->numSource 259 << " len: " << std::dec << ptf->length 260 << std::endl; 261 #if 0 262 for (unsigned int i = 0 ; i < ptf->length ; i++) { 263 if ((i % 16) == 0) { 264 std::cout << std::endl 265 << "#" << std::setw(8) << std::setfill('0') 266 << i << " : "; 267 } 268 std::cout << " " 269 << std::hex << std::setw(2) << std::setfill('0') 270 << static_cast<unsigned int>(pdata[i]); 271 } 272 std::cout << std::endl; 273 #endif 274 275 } else if (msg_magic == SubTimeFrame::MAGIC) { 276 SubTimeFrame::Header *pstf 277 = reinterpret_cast<SubTimeFrame::Header *>(pdata); 278 std::cout << "#STF Header " 279 << std::hex << std::setw(8) << std::setfill('0') << pstf->magic 280 << " id: " << std::setw(8) << std::setfill('0') << pstf->timeFrameId 281 << " Type: " << std::setw(8) << std::setfill('0') << pstf->femType 282 << " FE: " << std::setw(8) << std::setfill('0') << pstf->femId 283 << std::endl << "# " 284 << " len: " << std::dec << pstf->length 285 << " nMsg: " << std::dec << pstf->numMessages 286 << std::endl << "# " 287 << " Ts: " << std::dec << pstf->timeSec 288 << " Tus: " << std::dec << pstf->timeUSec 289 << std::endl; 290 291 fe_type = pstf->femType; 292 293 } else { 294 #if 1 295 for (unsigned int j = 0 ; j < msize ; j += 8) { 296 std::cout << "# " << std::setw(8) << j << " : " 297 << std::hex << std::setw(2) << std::setfill('0') 298 << std::setw(2) << static_cast<unsigned int>(pdata[j + 7]) << " " 299 << std::setw(2) << static_cast<unsigned int>(pdata[j + 6]) << " " 300 << std::setw(2) << static_cast<unsigned int>(pdata[j + 5]) << " " 301 << std::setw(2) << static_cast<unsigned int>(pdata[j + 4]) << " " 302 << std::setw(2) << static_cast<unsigned int>(pdata[j + 3]) << " " 303 << std::setw(2) << static_cast<unsigned int>(pdata[j + 2]) << " " 304 << std::setw(2) << static_cast<unsigned int>(pdata[j + 1]) << " " 305 << std::setw(2) << static_cast<unsigned int>(pdata[j + 0]) << " : "; 306 307 if ((pdata[j + 7] & 0xfc) == (TDC64H::T_TDC << 2)) { 308 std::cout << "TDC "; 309 uint64_t *dword = reinterpret_cast<uint64_t *>(&(pdata[j])); 310 if (fe_type == SubTimeFrame::TDC64H) { 311 struct TDC64H::tdc64 tdc; 312 TDC64H::Unpack(*dword, &tdc); 313 std::cout << "H :" 314 << " CH: " << std::dec << std::setw(3) << tdc.ch 315 << " TDC: " << std::setw(7) << tdc.tdc << std::endl; 316 } else 317 if (fe_type == SubTimeFrame::TDC64L) { 318 struct TDC64L::tdc64 tdc; 319 TDC64L::Unpack(*dword, &tdc); 320 std::cout << "L :" 321 << " CH: " << std::dec << std::setw(3) << tdc.ch 322 << " TDC: " << std::setw(7) << tdc.tdc << std::endl; 323 } else { 324 std::cout << "UNKNOWN Type :0x" << std::hex << fe_type << std::endl; 325 } 326 327 } else if ((pdata[j + 7] & 0xfc) == (TDC64H::T_HB << 2)) { 328 std::cout << "Hart beat" << std::endl; 329 } else if ((pdata[j + 7] & 0xfc) == (TDC64H::T_SPL_START << 2)) { 330 std::cout << "SPILL Start" << std::endl; 331 } else if ((pdata[j + 7] & 0xfc) == (TDC64H::T_SPL_END << 2)) { 332 std::cout << "SPILL End" << std::endl; 333 } else { 334 std::cout << std::endl; 335 } 336 } 337 #else 338 std::cout << "#Unknown Header " << std::hex << msg_magic << std::endl; 339 #endif 340 } 341 342 343 return true; 344 } 345 346 int LogicFilter::IsHartBeat(uint64_t val, uint32_t type) 347 { 348 int hbflag = -1; 349 int hbframe = -1; 350 if (type == SubTimeFrame::TDC64H) { 351 struct TDC64H::tdc64 tdc; 352 if (TDC64H::Unpack(val, &tdc) == TDC64H::T_HB) { 353 hbframe = tdc.hartbeat; 354 hbflag = tdc.flag; 355 } 356 } else if (type == SubTimeFrame::TDC64L) { 357 struct TDC64L::tdc64 tdc; 358 if (TDC64L::Unpack(val, &tdc) == TDC64L::T_HB) { 359 hbframe = tdc.hartbeat; 360 hbflag = tdc.flag; 361 } 362 } else if (type == SubTimeFrame::TDC64L_V1) { 363 struct TDC64L::tdc64 tdc; 364 if (TDC64L::v1::Unpack(val, &tdc) == TDC64L::v1::T_HB) { 365 hbframe = tdc.hartbeat; 366 hbflag = tdc.flag; 367 } 368 } else if (type == SubTimeFrame::TDC64H_V3) { 369 struct TDC64H_V3::tdc64 tdc; 370 if (TDC64H_V3::Unpack(val, &tdc) == TDC64H_V3::T_HB1) { 371 hbframe = tdc.hartbeat; 372 hbflag = tdc.flag; 373 } 374 } else if (type == SubTimeFrame::TDC64L_V3) { 375 struct TDC64L_V3::tdc64 tdc; 376 if (TDC64L_V3::Unpack(val, &tdc) == TDC64L_V3::T_HB1) { 377 hbframe = tdc.hartbeat; 378 hbflag = tdc.flag; 379 } 380 } else { 381 std::cout << "Unknown device : " << std::hex << type << std::endl; 382 const uint64_t* data = reinterpret_cast<const uint64_t*>(&val); 383 for (size_t i = 0; i < sizeof(val) / sizeof(uint64_t); ++i) { 384 std::cout << "Data[" << i << "]: " << std::hex << data[i] << std::endl; 385 } 386 } 387 388 if (hbflag > 0) fHBflag |= hbflag; 389 if (fKt4->Check()) { 390 if (fHBflag > 0) { 391 if ((fHBflag & 0x200) == 0x200) LOG(warn) << "HB Data lost"; 392 if ((fHBflag & 0x100) == 0x100) LOG(warn) << "HB Data confiliction"; 393 if ((fHBflag & 0x080) == 0x080) LOG(warn) << "HB LFN mismatch"; 394 if ((fHBflag & 0x040) == 0x040) LOG(warn) << "HB GFN mismatch"; 395 } 396 fHBflag = 0; 397 } 398 399 return hbframe; 400 } 401 402 int LogicFilter::RemoveData(fair::mq::Parts &parts, int index) 403 { 404 const int HB_SIZE = sizeof(uint64_t) * 2; 405 const uint64_t HB_HEAD = 0x7000'0000'0000'0000; 406 407 int rval = 0; 408 409 fair::mq::Message & msg = parts[index]; 410 //fair::mq::MessagePtr & msg = parts.At(index); 411 //auto & msg = parts[index]; 412 char *pdata = reinterpret_cast<char *>(msg.GetData()); 413 uint64_t *pdata64 = reinterpret_cast<uint64_t *>(msg.GetData()); 414 unsigned int msize = msg.GetSize(); 415 416 if (msize > HB_SIZE) { 417 418 for (unsigned int i = 0 ; i < (msize / sizeof(uint64_t)) ; i++) { 419 std::cout << std::hex << pdata64[i] << std::endl; 420 } 421 422 #if 0 423 auto hbp = std::make_unique<char[]>(HB_SIZE); 424 strncpy(hbp.get(), pdata + (msize - HB_SIZE), HB_SIZE); 425 auto hbmsg = MessageUtil::NewMessage(*this, std::move(hbp)); 426 #endif 427 428 #if 1 429 auto hb = std::make_unique< std::vector<char> > (HB_SIZE); 430 strncpy(hb->data(), pdata + (msize - HB_SIZE), HB_SIZE); 431 auto hbmsg = MessageUtil::NewMessage(*this, std::move(hb)); 432 #endif 433 434 #if 0 435 auto hb = new char[HB_SIZE]; 436 fair::mq::MessagePtr hbmsg( 437 NewMessage( 438 reinterpret_cast<char *>(hb), 439 sizeof(hb), 440 [](void* ppdata, void*) { 441 char *p = reinterpret_cast<char *>(ppdata); 442 delete p; 443 }, 444 nullptr) 445 ); 446 auto phbchar = reinterpret_cast<char *>(hbmsg->GetData()); 447 strncpy(phbchar, pdata + (msize - HB_SIZE), HB_SIZE); 448 #endif 449 450 uint64_t *v = reinterpret_cast<uint64_t *>(hbmsg->GetData()); 451 for (unsigned int i = 0 ; i < (HB_SIZE / sizeof(uint64_t)) ; i++) { 452 if ((v[i] & 0xfc00'0000'0000'0000) != HB_HEAD) { 453 rval = -1; 454 std::cout << "#E not HB Tail " << v[i]; 455 } 456 } 457 458 auto it = parts.fParts.begin(); 459 std::advance(it, index); 460 parts.fParts.insert(it, std::move(hbmsg)); 461 parts.fParts.erase(it + 1); 462 463 #if 1 464 for (int i = 0 ; i < parts.Size() ; i++) { 465 auto phead = reinterpret_cast<uint64_t *>(parts[i].GetData()); 466 std::cout << "#D " << i 467 << " " << std::hex << phead[0] << " " << std::hex << phead[1] 468 << " " << std::dec << parts[i].GetSize() << std::endl; 469 } 470 #endif 471 472 } 473 474 return rval; 475 } 476 477 void LogicFilter::CheckMultiPart(FairMQParts &inParts) 478 { 479 std::cout << "#Nmsg: " << std::dec << inParts.Size() << std::endl; 480 std::cout << "# " << std::dec; 481 for(int i = 0 ; i < inParts.Size() ; i++) { 482 uint64_t *top = reinterpret_cast<uint64_t *>(inParts[i].GetData()); 483 struct TimeFrame::Header *tbh 484 = reinterpret_cast<TimeFrame::Header *>(inParts[i].GetData()); 485 struct SubTimeFrame::Header *stbh 486 = reinterpret_cast<SubTimeFrame::Header *>(inParts[i].GetData()); 487 struct HeartbeatFrame::Header *hbh 488 = reinterpret_cast<HeartbeatFrame::Header *>(inParts[i].GetData()); 489 if (tbh->magic == TimeFrame::MAGIC) { 490 std::cout << "TF Id: " 491 << tbh->timeFrameId << " Nsrc: " << tbh->numSource; 492 } else 493 if (stbh->magic == SubTimeFrame::MAGIC) { 494 std::cout << std::endl << "* STF: TFid: 0x" << std::hex 495 << std::setw(8) << stbh->timeFrameId << " :"; 496 } else 497 if (hbh->magic == HeartbeatFrame::MAGIC) { 498 std::cout << " HB:" << std::dec 499 << std::setw(8) << inParts[i].GetSize() << " :"; 500 } else { 501 std::cout << "UK " << std::setw(8) << std::setfill('0') 502 << IsHartBeat(top[0], SubTimeFrame::TDC64H); 503 } 504 } 505 std::cout << std::dec << std::endl; 506 507 return; 508 } 509 510 int LogicFilter::MakeBlockMap( 511 FairMQParts &inParts, 512 std::vector< std::vector<struct DataBlock> > &block_map) 513 { 514 uint64_t femid = 0; 515 uint64_t time_frame_id = 0; 516 uint64_t devtype = 0; 517 int ifem = 0; 518 //std::vector<bool> flag_sending; 519 std::vector<struct DataBlock> blocks; 520 521 //struct TimeFrame::Header *i_tfHeader = nullptr; 522 523 for(int i = 0 ; i < inParts.Size() ; i++) { 524 //flag_sending.push_back(true); 525 #if 0 526 CheckData(inParts.At(i)); 527 #endif 528 529 auto tfHeader = reinterpret_cast<struct TimeFrame::Header *>(inParts[i].GetData()); 530 auto stfHeader = reinterpret_cast<struct SubTimeFrame::Header *>(inParts[i].GetData()); 531 struct DataBlock dblock; 532 533 if (tfHeader->magic == TimeFrame::MAGIC) { 534 ifem = -1; 535 //stf.clear(); 536 //stf.resize(0); 537 //i_tfHeader = tfHeader; 538 dblock.is_HB = false; 539 dblock.msg_index = 0; 540 dblock.timeFrameId = tfHeader->timeFrameId; 541 } else 542 if (stfHeader->magic == SubTimeFrame::MAGIC) { 543 femid = stfHeader->femId; 544 devtype = stfHeader->femType; 545 time_frame_id = stfHeader->timeFrameId; 546 if (blocks.size() > 0) block_map.push_back(blocks); 547 //stf.push_back(*stfHeader); 548 dblock.is_HB = false; 549 dblock.msg_index = -1; 550 dblock.timeFrameId = time_frame_id; 551 blocks.clear(); 552 blocks.resize(0); 553 ifem++; 554 555 #if 0 556 std::cout << "#D STF Nmsg: " << stfHeader->numMessages 557 << " Tid: " << stfHeader->timeFrameId << std::endl; 558 #endif 559 560 } else { 561 // make block map; 562 563 uint64_t *data = reinterpret_cast<uint64_t *>(inParts[i].GetData()); 564 int hbframe = IsHartBeat(data[0], devtype); 565 566 #if 0 567 std::cout << "#DDD msg" << std::dec << std::setw(3) << i << ":" 568 << " HBFrame:" << std::setw(6) << hbframe 569 << " blocks.size(): " << blocks.size() << std::endl; 570 #endif 571 572 //if (hbframe < 0) { 573 if ((data[0] == HeartbeatFrame::MAGIC) 574 || (hbframe < 0)) { 575 dblock.femId = femid; 576 dblock.Type = devtype; 577 dblock.is_HB = false; 578 dblock.msg_index = i; 579 dblock.nTrig = 0; 580 dblock.HBFrame = 0; 581 dblock.timeFrameId = time_frame_id; 582 } else { 583 if (fSplitMethod == 1) { 584 //data ga nakattatokimo push_back dummy 585 if ( (blocks.size() == 0) 586 || ((blocks.size() > 0) 587 && (blocks.back().is_HB == true)) ) { 588 589 #if 0 590 std::cout << "#W no data frame Msg:" 591 << std::dec << i 592 << " " << inParts.Size() 593 << " " << blocks.size() 594 << " is_HB:" << hbframe 595 << " " << std::hex << data[0]; 596 if (blocks.size() > 0) { 597 std::cout << " p_is_HB:" 598 << blocks.back().is_HB; 599 } 600 std::cout << std::endl; 601 //assert(0); 602 #endif 603 604 dblock.femId = femid; 605 dblock.Type = SubTimeFrame::NULDEV; 606 dblock.is_HB = false; 607 dblock.msg_index = -2; 608 dblock.nTrig = 0; 609 dblock.HBFrame = 0; 610 dblock.timeFrameId = time_frame_id; 611 blocks.push_back(dblock); 612 } 613 } 614 615 dblock.HBFrame = hbframe; 616 dblock.femId = femid; 617 dblock.Type = devtype; 618 dblock.is_HB = true; 619 dblock.msg_index = i; 620 dblock.nTrig = 0; 621 dblock.timeFrameId = time_frame_id; 622 } 623 blocks.push_back(dblock); 624 } 625 } /// end of the for loop 626 block_map.push_back(blocks); 627 628 #if 0 //BlockMap check 629 if (fKt2->Check()) { 630 std::cout << "#D block_map.size: " << std::dec << block_map.size() << std::endl; 631 for (auto& blk : block_map) { 632 std::cout << "#D block " << std::setw(2) << blk.size() << " /"; 633 for (auto& b : blk) { 634 std::cout << " " << std::setw(3) << b.msg_index << " HB:" << b.is_HB; 635 } 636 std::cout << std::endl; 637 } 638 std::cout << std::endl; 639 } 640 #endif 641 642 size_t bsize_min = block_map[0].size(); 643 for (auto& blk : block_map) { 644 if (blk.size() < bsize_min) { 645 LOG(warn) << "Unmatched number of stf in TF " 646 << bsize_min << " " << blk.size(); 647 bsize_min = blk.size(); 648 } 649 } 650 651 #if 0 652 std::cout << "blocks: " << blocks.size() << std::endl; 653 #endif 654 655 return 0; 656 } 657 658 int LogicFilter::MakeBlockMap2( 659 FairMQParts &inParts, 660 std::vector< std::vector<struct DataBlock> > &block_map) 661 { 662 uint64_t femid = 0; 663 uint64_t time_frame_id = 0; 664 uint64_t devtype = 0; 665 int ifem = 0; 666 //std::vector<bool> flag_sending; 667 std::vector<struct DataBlock> blocks; 668 669 //struct TimeFrame::Header *i_tfHeader = nullptr; 670 671 for(int i = 0 ; i < inParts.Size() ; i++) { 672 //flag_sending.push_back(true); 673 #if 0 674 CheckData(inParts.At(i)); 675 #endif 676 677 auto tfHeader = reinterpret_cast<struct TimeFrame::Header *>(inParts[i].GetData()); 678 auto stfHeader = reinterpret_cast<struct SubTimeFrame::Header *>(inParts[i].GetData()); 679 struct DataBlock dblock; 680 681 if (tfHeader->magic == TimeFrame::MAGIC) { 682 ifem = -1; 683 //stf.clear(); 684 //stf.resize(0); 685 //i_tfHeader = tfHeader; 686 dblock.is_HB = false; 687 dblock.msg_index = 0; 688 dblock.timeFrameId = tfHeader->timeFrameId; 689 } else 690 if (stfHeader->magic == SubTimeFrame::MAGIC) { 691 femid = stfHeader->femId; 692 devtype = stfHeader->femType; 693 time_frame_id = stfHeader->timeFrameId; 694 if (blocks.size() > 0) block_map.push_back(blocks); 695 //stf.push_back(*stfHeader); 696 dblock.is_HB = false; 697 dblock.msg_index = -1; 698 dblock.timeFrameId = time_frame_id; 699 blocks.clear(); 700 blocks.resize(0); 701 ifem++; 702 703 #if 0 704 std::cout << "#D STF Nmsg: " << stfHeader->numMessages 705 << " Tid: " << stfHeader->timeFrameId << std::endl; 706 #endif 707 708 } else { 709 // make block map; 710 711 uint64_t *data = reinterpret_cast<uint64_t *>(inParts[i].GetData()); 712 int hbframe = IsHartBeat(data[0], devtype); 713 714 #if 0 715 std::cout << "#DDD msg" << std::dec << std::setw(3) << i << ":" 716 << " HBFrame:" << std::setw(6) << hbframe 717 << " blocks.size(): " << blocks.size() << std::endl; 718 #endif 719 720 if ((data[0] == HeartbeatFrame::MAGIC) 721 || (hbframe < 0)) { 722 dblock.femId = femid; 723 dblock.Type = devtype; 724 dblock.is_HB = false; 725 dblock.msg_index = i; 726 dblock.nTrig = 0; 727 dblock.HBFrame = 0; 728 dblock.timeFrameId = time_frame_id; 729 } else { 730 if (fSplitMethod == 1) { 731 //data ga nakattatokimo push_back dummy 732 if ( (blocks.size() == 0) 733 || ((blocks.size() > 0) 734 && (blocks.back().is_HB == true)) ) { 735 736 #if 0 737 std::cout << "#W no data frame Msg:" 738 << std::dec << i 739 << " " << inParts.Size() 740 << " " << blocks.size() 741 << " is_HB:" << hbframe 742 << " " << std::hex << data[0]; 743 if (blocks.size() > 0) { 744 std::cout << " p_is_HB:" 745 << blocks.back().is_HB; 746 } 747 std::cout << std::endl; 748 //assert(0); 749 #endif 750 751 dblock.femId = femid; 752 dblock.Type = SubTimeFrame::NULDEV; 753 dblock.is_HB = false; 754 dblock.msg_index = -2; 755 dblock.nTrig = 0; 756 dblock.HBFrame = 0; 757 dblock.timeFrameId = time_frame_id; 758 blocks.push_back(dblock); 759 } 760 } 761 762 dblock.HBFrame = hbframe; 763 dblock.femId = femid; 764 dblock.Type = devtype; 765 dblock.is_HB = true; 766 dblock.msg_index = i; 767 dblock.nTrig = 0; 768 dblock.timeFrameId = time_frame_id; 769 } 770 blocks.push_back(dblock); 771 } 772 } /// end of the for loop 773 block_map.push_back(blocks); 774 775 #if 0 //BlockMap check 776 if (fKt2->Check()) { 777 std::cout << "#D block_map.size: " << std::dec << block_map.size() << std::endl; 778 for (auto& blk : block_map) { 779 std::cout << "#D block " << std::setw(2) << blk.size() << " /"; 780 for (auto& b : blk) { 781 std::cout << " " << std::setw(3) << b.msg_index << " HB:" << b.is_HB; 782 } 783 std::cout << std::endl; 784 } 785 std::cout << std::endl; 786 } 787 #endif 788 789 size_t bsize_min = block_map[0].size(); 790 for (auto& blk : block_map) { 791 if (blk.size() < bsize_min) { 792 LOG(warn) << "Unmatched number of stf in TF " 793 << bsize_min << " " << blk.size(); 794 bsize_min = blk.size(); 795 } 796 } 797 798 #if 0 799 std::cout << "blocks: " << blocks.size() << std::endl; 800 #endif 801 802 return 0; 803 } 804 805 int LogicFilter::BuildHBF( 806 std::vector<struct HBFIndex> &hbf_list, 807 std::vector< std::vector<struct DataBlock> > &block_map, 808 FairMQParts &inParts, 809 int imsg 810 ) 811 { 812 813 for (size_t iifem = 0 ; iifem < block_map.size() ; iifem++) { 814 struct DataBlock *dbl = &block_map[iifem][imsg]; 815 uint64_t vfemid = dbl->femId; 816 if (!(dbl->is_HB)) { 817 int mindex = dbl->msg_index; 818 819 #if 0 820 std::cout << "#D1 Mark FEM ID: " << std::hex << vfemid 821 << " Type: " << dbl->Type << " HBFrame: " << std::hex 822 << dbl->HBFrame << std::endl; 823 std::cout << "#D1 mindex: " << std::dec << mindex 824 << " vfemid: " << vfemid << " type: " << dbl->Type 825 << std::endl; 826 std::cout << "#D1 inParts.size: " << inParts[mindex].GetSize() 827 << std::endl; 828 #endif 829 830 if (mindex > 0) { 831 struct HBFIndex hbf_seg; 832 hbf_seg.msg_index = imsg; 833 hbf_seg.timeFrameId = dbl->timeFrameId; 834 hbf_seg.femType = dbl->Type; 835 hbf_seg.femId = vfemid; 836 hbf_seg.data = reinterpret_cast<unsigned char *>( 837 inParts[mindex].GetData()); 838 hbf_seg.size = inParts[mindex].GetSize(); 839 840 hbf_list.emplace_back(hbf_seg); 841 } 842 } 843 } 844 845 #if 0 846 std::cout << "# HB: " << std::dec << i; 847 for (size_t iifem = 0 ; iifem < block_map.size() ; iifem++) { 848 struct DataBlock *dbl = &block_map[iifem][i]; 849 uint64_t vfemid = dbl->femId; 850 uint64_t vhbframe = dbl->HBFrame; 851 //std::cout << "# HB: " << std::dec << i 852 //<< " FEM: " << std::hex << vfemid 853 std::cout << " " << std::dec << (vfemid & 0xff) << ":" << vhbframe; 854 } 855 std::cout << std::endl; 856 #endif 857 858 return 0; 859 } 860 861 862 int LogicFilter::AddFilterMessage( 863 FairMQParts &outParts, 864 std::vector< std::vector<uint32_t> > &fltdata, 865 uint32_t elapse, 866 uint32_t tf_id) 867 { 868 int flt_data_len = 0; 869 870 auto tp = std::chrono::system_clock::now() ; 871 auto d = tp.time_since_epoch(); 872 uint64_t sec = std::chrono::duration_cast 873 <std::chrono::seconds>(d).count(); 874 uint64_t usec = std::chrono::duration_cast 875 <std::chrono::microseconds>(d).count(); 876 877 uint32_t flt_datasize = 0; 878 uint32_t totalhits = 0; 879 for (auto &v : fltdata) { 880 flt_datasize += v.size() * sizeof(uint32_t) 881 + sizeof(struct Filter::TrgTimeHeader); 882 totalhits += v.size(); 883 } 884 uint64_t flt_len = flt_datasize + sizeof(struct Filter::Header); 885 886 887 #if 0 // memo to make FairMQ Message 888 gQdepth++; 889 FairMQMessagePtr fltheadermsg( 890 NewMessage( 891 reinterpret_cast<char *>(&fltHeader), 892 sizeof(fltHeader), 893 [](void* pdata, void*) { 894 char *p = reinterpret_cast<char *>(pdata); 895 delete p; 896 //gQdepth--; 897 }, 898 nullptr)); 899 outParts.AddPart(std::move(fltheadermsg)); 900 #endif 901 902 //add FLT Header 903 auto fltHeader = std::make_unique<struct Filter::Header>(); 904 fltHeader->magic = Filter::MAGIC; 905 fltHeader->length = flt_len; 906 fltHeader->timeFrameId = tf_id; 907 fltHeader->numTrigs = totalhits; 908 fltHeader->workerId = fId; 909 fltHeader->numMessages = 1 + fltdata.size(); 910 fltHeader->elapseTime = elapse; 911 fltHeader->timeSec = sec; 912 fltHeader->timeUSec = usec; 913 outParts.AddPart(MessageUtil::NewMessage(*this, std::move(fltHeader))); 914 915 flt_data_len += sizeof(struct Filter::Header); 916 917 //add FLT data 918 for (auto &v : fltdata) { 919 //auto trgtdc = std::make_unique<std::vector<uint32_t>>(std::move(v)); 920 //outParts.AddPart(MessageUtil::NewMessage(*this, std::move(trgtdc))); 921 922 //std::cout << "#D v.size : " << v.size() << std::endl; 923 // << " msg.size" << (*outParts.end()).get()->GetSize() << std::endl; 924 925 struct Filter::TrgTimeHeader trg_time_header; 926 trg_time_header.length = sizeof(Filter::TrgTimeHeader) 927 + v.size() * sizeof(Filter::TrgTime); 928 //std::cout << "#D TrgTimeHeader.size: " << sizeof(struct Filter::TrgTimeHeader) 929 // << " Time.size: " << v.size() * sizeof(Filter::TrgTime) << std::endl; 930 //std::cout << "#D TrgTimeHeader.uint32_t.size: " << sizeof(trg_time_header.u32data) << std::endl; 931 932 std::vector<uint32_t> vv; 933 for (auto &h : trg_time_header.u32data) vv.emplace_back(h); 934 for (auto &tdc : v) { 935 uint32_t trg_type = 0xaa000000; 936 vv.emplace_back(tdc); 937 vv.emplace_back(trg_type); 938 } 939 //std::cout << "#D fltmsg.size: " << vv.size() 940 // << " TrgTime.len: " << trg_time_header.length << std::endl; 941 942 outParts.AddPart(MessageUtil::NewMessage 943 (*this, std::make_unique<std::vector<uint32_t>>(std::move(vv)))); 944 945 flt_data_len += outParts[outParts.Size() - 1].GetSize(); 946 } 947 948 // std::cout << "#D fltH.len: " << std::dec << flt_len 949 // << " flt_data_len: " << flt_data_len << std::endl; 950 951 return flt_data_len; 952 } 953 954 int LogicFilter::MarkFlagSending( 955 std::vector< std::vector<struct DataBlock> > &block_map, 956 int i, 957 int nhits, 958 std::vector<bool> &flag_sending) 959 { 960 for (size_t iifem = 0 ; iifem < block_map.size() ; iifem++) { 961 block_map[iifem][i].nTrig = nhits; 962 if (nhits == 0) { 963 int mindex = block_map[iifem][i].msg_index; 964 bool is_HB = block_map[iifem][i].is_HB; 965 uint32_t dtype = block_map[iifem][i].Type; 966 967 //std::cout << "#D mindex: " << mindex 968 // << " h: " << is_HB 969 // << " t: " << block_map[iifem][i].Type 970 // << " i: " << std::hex << block_map[iifem][i].femId 971 // << std::dec << std::endl;; 972 973 if ((mindex > 0) && (! is_HB)) { 974 if (dtype != SubTimeFrame::NULDEV) { 975 flag_sending[mindex] = false; 976 } 977 978 if ((fSplitMethod == 0) || (fSplitMethod == 1)) { 979 //// kokoha mondai gaaru //// 980 if (fIsRemoveHB) { 981 flag_sending[mindex + 1] = false; 982 } 983 } 984 } 985 986 } else { 987 //int mindex = block_map[iifem][i].msg_index; 988 //bool is_HB = block_map[iifem][i].is_HB; 989 //std::cout << "#D H mindex: " << mindex 990 // << " h: " << is_HB 991 // << " t: " << block_map[iifem][i].Type 992 // << " i: " << std::hex << block_map[iifem][i].femId 993 // << " Nhits: " << std::dec << nhits 994 // << std::endl;; 995 } 996 } 997 998 return 0; 999 } 1000 1001 1002 bool LogicFilter::ConditionalRun() 1003 { 1004 //Receive 1005 FairMQParts inParts; 1006 1007 FairMQMessagePtr msg_header(fTransportFactory->CreateMessage()); 1008 struct Filter::Header fltheader; 1009 // struct TimeFrame::Header *i_tfHeader = nullptr; 1010 int i_tf_msg_index = 0; 1011 uint32_t tf_tf_id = 0; 1012 std::chrono::system_clock::time_point sw_start, sw_end; 1013 1014 if (Receive(inParts, fInputChannelName, 0, 1000) > 0) { 1015 assert(inParts.Size() >= 2); 1016 sw_start = std::chrono::system_clock::now(); 1017 1018 #if 1 1019 if (fKt1->Check()) CheckMultiPart(inParts); 1020 #endif 1021 1022 std::vector<struct DataBlock> blocks; 1023 std::vector< std::vector<struct DataBlock> > block_map; 1024 std::vector<bool> flag_sending; 1025 for(int i = 0 ; i < inParts.Size() ; i++) flag_sending.push_back(true); 1026 1027 for(int i = 0 ; i < inParts.Size() ; i++) { 1028 auto tfHeader = reinterpret_cast<struct TimeFrame::Header *>(inParts[i].GetData()); 1029 if (tfHeader->magic == TimeFrame::MAGIC) { 1030 // i_tfHeader = tfHeader; 1031 i_tf_msg_index = i; 1032 tf_tf_id = tfHeader->timeFrameId; 1033 } 1034 } 1035 MakeBlockMap(inParts, block_map); 1036 1037 #if 0 //BlockMap check 1038 if (fKt2->Check()) { 1039 std::cout << "#D block_map.size: " << std::dec << block_map.size() << std::endl; 1040 for (auto& blk : block_map) { 1041 std::cout << "#D block " << std::setw(2) << blk.size() << " /"; 1042 for (auto& b : blk) { 1043 std::cout << " " << std::setw(3) << b.msg_index << " HB:" << b.is_HB; 1044 } 1045 std::cout << std::endl; 1046 } 1047 std::cout << std::endl; 1048 } 1049 #endif 1050 1051 size_t bsize_min = block_map[0].size(); 1052 for (auto& blk : block_map) { 1053 if (blk.size() < bsize_min) { 1054 LOG(warn) << "Unmatched number of stf in TF " 1055 << bsize_min << " " << blk.size(); 1056 bsize_min = blk.size(); 1057 } 1058 } 1059 1060 // Trigger processing in unit of HBF 1061 std::vector< std::vector<uint32_t> > fltdata; 1062 int totalhits = 0; 1063 for (size_t i = 0 ; i < bsize_min ; i++) { 1064 1065 //HBF building 1066 std::vector<struct HBFIndex> hbf_list; 1067 BuildHBF(hbf_list, block_map, inParts, i); 1068 1069 //Trigger process 1070 std::vector<uint32_t> *hits = fTrig->Exec(hbf_list); 1071 fltdata.emplace_back(*hits); 1072 int nhits = hits->size(); 1073 1074 #if 0 1075 if (nhits > 0) { 1076 std::cout << "#D1 Hits : " << hits->size() << " "; 1077 for (unsigned int ii = 0 ; ii < hits->size() ; ii++) { 1078 std::cout << " " << std::dec << (*hits)[ii]; 1079 if (ii > 10) { 1080 std::cout << "..."; 1081 break; 1082 } 1083 } 1084 std::cout << std::endl; 1085 } 1086 #endif 1087 1088 // mark sending flag for STF 1089 MarkFlagSending(block_map, i, nhits, flag_sending); 1090 1091 totalhits += nhits; 1092 } 1093 1094 #if 0 1095 if (fKt2->Check()) { 1096 //std::cout << "#block_map size: " << block_map.size() << std::endl; 1097 for (unsigned int i = 0 ; i < block_map.size() ; i++) { 1098 std::cout << "#HBFrame: " << i << "/"; 1099 for (unsigned int j = 0 ; j < block_map[i].size(); j++) { 1100 std::cout << " " << j << ":" 1101 << std::setw(5) << block_map[i][j].HBFrame; 1102 } 1103 std::cout << std::endl; 1104 } 1105 //std::cout << std::endl; 1106 1107 if (totalhits > 0) { 1108 std::cout << "#D TotalHits: " << totalhits; 1109 std::cout << " Flag: "; 1110 for (const auto& v : flag_sending) std::cout << " " << v; 1111 std::cout << std::endl; 1112 } 1113 } 1114 #endif 1115 1116 #if 1 1117 if (totalhits > 0) { 1118 std::cout << "#D TotalHits: " << std::setw(4) << totalhits << " :"; 1119 for (int i = 0 ; i < totalhits / 50 ; i++) std::cout << "."; 1120 // std::cout << " Flag: "; 1121 // for (const auto& v : flag_sending) std::cout << " " << v; 1122 std::cout << std::endl; 1123 } 1124 #endif 1125 1126 #if 0 1127 auto tfHeader = reinterpret_cast<TimeFrame::Header*>(inParts.At(0)->GetData()); 1128 auto stfHeader = reinterpret_cast<SubTimeFrame::Header*>(inParts.At(0)->GetData()); 1129 auto stfId = stfHeader->timeFrameId; 1130 #endif 1131 1132 sw_end = std::chrono::system_clock::now(); 1133 uint32_t elapse = std::chrono::duration_cast<std::chrono::microseconds>( 1134 sw_end - sw_start).count(); 1135 static uint64_t int_hits = 0; 1136 static uint64_t int_processed_hbf = 0; 1137 double trig_ratio; 1138 int_hits += totalhits; 1139 int_processed_hbf += bsize_min / 2; 1140 if (fKt3->Check()) { 1141 if (int_processed_hbf > 0) { 1142 trig_ratio = static_cast<double>(int_hits) 1143 / static_cast<double>(int_processed_hbf); 1144 } 1145 1146 std::cout << "#Elapse: " << std::dec << elapse << " us" 1147 << " Hits: " << totalhits 1148 << " T.Ratio: " << trig_ratio 1149 << " Hits(inte): " << int_hits 1150 << " HBF(inte): " << int_processed_hbf 1151 << std::endl; 1152 1153 //int_hits = 0; 1154 //int_processed_hbf = 0; 1155 } 1156 1157 #if 0 1158 //Modify SubTimeFrameHeader 1159 if (fIsDataSuppress) { 1160 for (int ii = 0 ; ii < inParts.Size() ; ii++) { 1161 auto stfh = reinterpret_cast<struct SubTimeFrame::Header *> 1162 (inParts[ii].GetData()); 1163 if (stfh->magic == SubTimeFrame::MAGIC) { 1164 uint32_t len_stf = 0; 1165 uint32_t nmsg_stf = 0; 1166 int kk = ii + 1; 1167 for (int jj = ii + 1 ; jj < inParts.Size() ; jj++) { 1168 auto sstf = 1169 reinterpret_cast<struct SubTimeFrame::Header *> 1170 (inParts[jj].GetData()); 1171 if (sstf->magic == SubTimeFrame::MAGIC) { 1172 kk = jj; 1173 break; 1174 } else 1175 if (flag_sending[jj]) { 1176 len_stf += inParts[jj].GetSize(); 1177 nmsg_stf++; 1178 } 1179 } 1180 // if (len_stf == 0) { // STF header kara nankunaru? OK? 1181 // flag_sending[ii] = false; 1182 // } 1183 stfh->length 1184 = len_stf + sizeof(struct SubTimeFrame::Header); 1185 stfh->numMessages = nmsg_stf; 1186 ii = kk - 1; 1187 } 1188 } 1189 } 1190 #endif 1191 1192 #if 0 1193 // Modify TimeFrameHeader Length 1194 uint32_t tf_len = sizeof(TimeFrame::Header); 1195 if (fIsDataSuppress) { 1196 // SubTimeFrameHeader 1197 for (int ii = 0 ; ii < inParts.Size() ; ii++) { 1198 if (flag_sending[ii] || (! fIsDataSuppress)) { 1199 tf_len += (inParts.AtRef(ii)).GetSize(); 1200 } 1201 } 1202 } else { 1203 tf_len = i_tfHeader->length; 1204 } 1205 #endif 1206 1207 FairMQParts outParts; 1208 1209 //Copy TimeFrameHeader 1210 if (i_tf_msg_index != 0) { 1211 LOG(error) << "Wrong TimeFrame Header locataion : " << i_tf_msg_index; 1212 } 1213 FairMQMessagePtr msgCopyTF(fTransportFactory->CreateMessage()); 1214 msgCopyTF->Copy(inParts.AtRef(0)); 1215 outParts.AddPart(std::move(msgCopyTF)); 1216 1217 //FilterHeader 1218 //tf_len += AddFilterMessage(outParts, fltdata, elapse, tf_tf_id); 1219 AddFilterMessage(outParts, fltdata, elapse, tf_tf_id); 1220 1221 //Copy SubTimeFrame 1222 unsigned int msg_size = inParts.Size(); 1223 for (unsigned int ii = 1 ; ii < msg_size ; ii++) { 1224 if (flag_sending[ii] || (! fIsDataSuppress)) { 1225 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 1226 msgCopy->Copy(inParts.AtRef(ii)); 1227 outParts.AddPart(std::move(msgCopy)); 1228 } else { 1229 // add blank SubTime Frame 1230 struct HeartbeatFrame::Header *hbf 1231 = reinterpret_cast<struct HeartbeatFrame::Header *>(inParts[ii].GetData()); 1232 //uint64_t *hbf_u64 1233 // = reinterpret_cast<uint64_t *>(inParts[ii].GetData()); 1234 if (hbf->magic == HeartbeatFrame::MAGIC) { 1235 //if (hbf_u64[0] == HeartbeatFrame::MAGIC) { 1236 #pragma pack(4) 1237 struct BlankHBF{ 1238 struct HeartbeatFrame::Header Header; 1239 uint64_t Delimiter[2]; 1240 }; 1241 #pragma pack() 1242 auto blank = std::make_unique<struct BlankHBF>(); 1243 uint64_t *d = reinterpret_cast<uint64_t *>( 1244 reinterpret_cast<char *>(inParts[ii].GetData()) 1245 + inParts[ii].GetSize() - (sizeof(uint64_t) * 2)); 1246 std::memcpy(&(blank->Header), inParts[ii].GetData(), 1247 sizeof(struct HeartbeatFrame::Header)); 1248 blank->Header.length = sizeof(HeartbeatFrame::Header) 1249 + (sizeof(uint64_t) * 2); 1250 std::memcpy(&(blank->Delimiter), d, sizeof(uint64_t) * 2); 1251 outParts.AddPart(MessageUtil::NewMessage(*this, std::move(blank))); 1252 //flag_sending[ii] = true; 1253 //tf_len += sizeof(outParts[outParts.Size() - 1]); 1254 1255 #if 0 1256 std::cout << "#D null HBF size: " << std::dec 1257 << outParts[outParts.Size() - 1].GetSize() 1258 << " " << outParts.Size() << std::endl; 1259 uint64_t *b = reinterpret_cast<uint64_t *>(outParts[outParts.Size() - 1].GetData()); 1260 std::cout << "#D : " << std::hex << std::flush; 1261 for (unsigned int i = 0 ; i < outParts[outParts.Size() -1].GetSize() / sizeof(uint64_t) ; i++) { 1262 std::cout << " " << b[i] << std::flush; 1263 } 1264 std::cout << std::dec << std::endl; 1265 #endif 1266 1267 } 1268 } 1269 } 1270 1271 //Modify SubTimeFrameHeader 1272 if (fIsDataSuppress) { 1273 for (int ii = 0 ; ii < outParts.Size() ; ii++) { 1274 auto stfh = reinterpret_cast<struct SubTimeFrame::Header *> 1275 (outParts[ii].GetData()); 1276 if (stfh->magic == SubTimeFrame::MAGIC) { 1277 uint32_t len_stf = 0; 1278 uint32_t nmsg_stf = 0; 1279 int kk = ii + 1; 1280 for (int jj = ii + 1 ; jj < outParts.Size() ; jj++) { 1281 auto sstf = 1282 reinterpret_cast<struct SubTimeFrame::Header *> 1283 (outParts[jj].GetData()); 1284 if (sstf->magic == SubTimeFrame::MAGIC) { 1285 kk = jj; 1286 break; 1287 } else 1288 if (flag_sending[jj]) { 1289 len_stf += outParts[jj].GetSize(); 1290 nmsg_stf++; 1291 } 1292 } 1293 // if (len_stf == 0) { // STF header kara nankunaru? OK? 1294 // flag_sending[ii] = false; 1295 // } 1296 stfh->length 1297 = len_stf + sizeof(struct SubTimeFrame::Header); 1298 stfh->numMessages = nmsg_stf; 1299 ii = kk - 1; 1300 } 1301 } 1302 } 1303 1304 // rewrite TimeFrameHeader length 1305 auto tfHeader = reinterpret_cast<struct TimeFrame::Header *>( 1306 //outParts[i_tf_msg_index].GetData()); 1307 outParts[0].GetData()); 1308 // Calcurate TimeFrame.length 1309 uint32_t tf_len = 0; 1310 for (auto &msg : outParts) tf_len += msg->GetSize(); 1311 tfHeader->length = tf_len; 1312 1313 #if 0 1314 std::cout << "#D Blocks" << std::endl; 1315 for (unsigned int ii = 0 ; ii < block_map.size() ; ii++) { 1316 auto lblocks = block_map[ii]; 1317 for (unsigned int jj = 0 ; jj < lblocks.size() ; jj++) { 1318 std::cout << " " << lblocks[jj].is_HB << ":" << lblocks[jj].HBFrame; 1319 } 1320 std::cout << std::endl; 1321 } 1322 std::cout << "#DD outParts.Size: " << outParts.Size() << std::endl; 1323 std::cout << "#DD flag_sending: "; 1324 int flagcount = 0; 1325 for (unsigned int ii = 0 ; ii < msg_size ; ii++) { 1326 auto stfHeader = reinterpret_cast<struct SubTimeFrame::Header *> 1327 (inParts[ii].GetData()); 1328 if (stfHeader->magic == SubTimeFrame::MAGIC) std::cout << std::endl; 1329 std::cout << flag_sending[ii]; 1330 if (flag_sending[ii]) flagcount++; 1331 } 1332 std::cout << " : " << flagcount << std::endl; 1333 #endif 1334 1335 1336 1337 #if 1 1338 FairMQParts dqmParts; 1339 bool dqmSocketExists = fChannels.count(fDQMChannelName); 1340 if (dqmSocketExists) { 1341 for (auto & m : outParts) { 1342 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 1343 msgCopy->Copy(*m); 1344 dqmParts.AddPart(std::move(msgCopy)); 1345 } 1346 1347 1348 if (Send(dqmParts, fDQMChannelName) < 0) { 1349 if (NewStatePending()) { 1350 LOG(info) << "Device is not RUNNING"; 1351 } else { 1352 LOG(error) << "Failed to enqueue dqm-channel"; 1353 } 1354 } else { 1355 //std::cout << "+" << std::flush; 1356 } 1357 } else { 1358 #if 0 1359 std::cout << "NoDQM socket" << std::endl; 1360 #endif 1361 } 1362 #endif 1363 1364 1365 //Send 1366 auto poller = NewPoller(fOutputChannelName); 1367 while (!NewStatePending()) { 1368 auto direction = (fDirection++) % fNumDestination; 1369 poller->Poll(fPollTimeoutMS); 1370 if (poller->CheckOutput(fOutputChannelName, direction)) { 1371 if (Send(outParts, fOutputChannelName, direction) > 0) { 1372 // successfully sent 1373 break; 1374 } else { 1375 LOG(error) << "Failed to queue output-channel"; 1376 } 1377 } 1378 if (fNumDestination==1) { 1379 std::this_thread::sleep_for(std::chrono::milliseconds(1)); 1380 } 1381 } 1382 1383 } 1384 1385 return true; 1386 } 1387 1388 void LogicFilter::PostRun() 1389 { 1390 LOG(info) << "Post Run"; 1391 return; 1392 } 1393 1394 1395 #if 0 1396 bool LogicFilter::HandleData(fair::mq::MessagePtr& msg, int val) 1397 { 1398 (void)val; 1399 #if 0 1400 LOG(info) << "Received: \"" 1401 << std::string(static_cast<char*>(msg->GetData()), msg->GetSize()) 1402 << "\""; 1403 #endif 1404 1405 #if 0 1406 LOG(info) << "Received: " << msg->GetSize() << " : " << val; 1407 #endif 1408 1409 if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) { 1410 LOG(info) << "Configured maximum number of iterations reached." 1411 << " Leaving RUNNING state."; 1412 return false; 1413 } 1414 1415 CheckData(msg); 1416 1417 1418 // return true if you want the handler to be called again 1419 // (otherwise return false go to the Ready state) 1420 return true; 1421 } 1422 #endif 1423 1424 1425 void addCustomOptions(bpo::options_description& options) 1426 { 1427 using opt = LogicFilter::OptionKey; 1428 1429 options.add_options() 1430 //("max-iterations", bpo::value<uint64_t>()->default_value(0), 1431 //"Maximum number of iterations of Run/ConditionalRun/OnData (0 - infinite)") 1432 (opt::InputChannelName.data(), 1433 bpo::value<std::string>()->default_value("in"), 1434 "Name of the input channel") 1435 (opt::OutputChannelName.data(), 1436 bpo::value<std::string>()->default_value("out"), 1437 "Name of the output channel") 1438 (opt::DQMChannelName.data(), 1439 bpo::value<std::string>()->default_value("dqm"), 1440 "Name of the data quality monitoring channel") 1441 (opt::DataSuppress.data(), 1442 bpo::value<std::string>()->default_value("false"), 1443 "Data suppression enable") 1444 (opt::RemoveHB.data(), 1445 bpo::value<std::string>()->default_value("false"), 1446 "Remove HB without hit") 1447 (opt::PollTimeout.data(), 1448 bpo::value<std::string>()->default_value("1"), 1449 "Timeout of polling (in msec)") 1450 (opt::SplitMethod.data(), 1451 bpo::value<std::string>()->default_value("2"), 1452 "STF split method") 1453 1454 (opt::TriggerSignals.data(), 1455 bpo::value<std::string>()->default_value( 1456 "(0xc0a802a9 0 0) (0xc0a802a9 1 0)"), 1457 "Triger signals (module_IP Channel_number Offset)") 1458 (opt::TriggerExpression.data(), 1459 bpo::value<std::string>()->default_value("RPN 0 1 &"), 1460 "Trigger expression") 1461 (opt::TriggerWidth.data(), 1462 bpo::value<std::string>()->default_value("10"), 1463 "Trigger window width (4 ns unit)") 1464 1465 ; 1466 } 1467 1468 1469 std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /*config*/) 1470 { 1471 return std::make_unique<LogicFilter>(); 1472 }
STFBuilder.cxx
SHA-256: d0b239d64b934d1ad15f2e2433a6ba2c8514e5e3e9a4ffbdd7c3c888ebd22dfe
1 #include <algorithm> 2 #include <iomanip> 3 #include <iostream> 4 #include <chrono> 5 #include <stdexcept> 6 #include <functional> 7 #include <thread> 8 #include <cassert> 9 #include <numeric> 10 #include <unordered_map> 11 #include <sstream> 12 #include <sys/time.h> 13 14 #include <fairmq/runDevice.h> 15 16 #include "utility/HexDump.h" 17 #include "utility/MessageUtil.h" 18 19 #include "SubTimeFrameHeader.h" 20 #include "STFBuilder.h" 21 #include "AmQStrTdcData.h" 22 23 namespace STF = SubTimeFrame; 24 25 //______________________________________________________________________________ 26 AmQStrTdcSTFBuilder::AmQStrTdcSTFBuilder() 27 : fair::mq::Device() 28 { 29 mdebug = false; 30 } 31 32 //______________________________________________________________________________ 33 void AmQStrTdcSTFBuilder::BuildFrame(FairMQMessagePtr& msg, int index) 34 { 35 namespace Data = AmQStrTdc::Data; 36 using Word = Data::Word; 37 std::size_t offset = 0; 38 39 (void)index; 40 41 if(mdebug) { 42 LOG(debug) 43 << " buildframe STF = " << fSTFSequenceNumber << " HBF = " << fHBFCounter << "\n" 44 << " input payload entries = " << fInputPayloads.size() 45 << " offset " << offset << std::endl; 46 } 47 48 auto msgBegin = reinterpret_cast<Word*>(msg->GetData()); 49 auto msgSize = msg->GetSize(); 50 auto nWord = msgSize / sizeof(Word); 51 52 if(mdebug) { 53 54 LOG(debug) << " msg size " << msgSize << " bytes " << nWord << " words" << std::endl; 55 { 56 std::for_each(reinterpret_cast<Word*>(msgBegin), 57 msgBegin+nWord, 58 ::HexDump{4}); 59 } 60 } 61 62 for (long unsigned int i = 0 ; i < nWord ; ++i) { 63 auto word = reinterpret_cast<Data::Bits*>(msgBegin+i); 64 65 uint8_t h = word->head; 66 67 if(mdebug) 68 LOG(debug) << " head = " << std::hex << static_cast<uint16_t>(h) << std::dec << std::endl; 69 70 bool isHeadValid = false; 71 for (auto validHead : { Data::Data, Data::Heartbeat, Data::Heartbeat2nd, Data::Trailer, 72 Data::ThrottlingT1Start, Data::ThrottlingT1End, 73 Data::ThrottlingT2Start, Data::ThrottlingT2End }) { 74 if (h == validHead) isHeadValid = true; 75 } 76 77 // LOG(warn) << std::hex << "raw = " << word->raw << std::dec << std::endl; 78 79 if (!isHeadValid) { 80 static int limit_counter = 0; 81 if (limit_counter < 10) { 82 LOG(warn) << " " << i << " " << offset << " unknown head = " << std::hex << static_cast<uint16_t>(h) 83 << " " << word->raw << std::dec << std::endl; 84 limit_counter++; 85 } 86 // continue; 87 } 88 89 // in case of one delimiter... 90 91 if(hbf_flag == 1){ 92 93 if ( h != Data::Heartbeat2nd ) { 94 LOG(warn) << "Only the first delimiter comes...: " << std::hex << word->raw; 95 offset = i+1; 96 hbf_flag = 0; 97 continue; 98 } 99 } 100 ///// 101 102 //check Heartbeat delimiter 103 if ((h == Data::Heartbeat) || (h == Data::Heartbeat2nd)) { 104 105 if (h == Data::Heartbeat) { 106 fdelimiterFrameId = word->hbframe & 0xFFFFFF; 107 hbf_flag++; 108 109 //LOG(debug) << " 1st delimiter comes " << std::hex << word->hbframe << ", raw = " << word->raw; 110 111 uint16_t throttlingIn1 = (word->hbflag >> 7) & 0x01; 112 uint16_t throttlingIn2 = (word->hbflag >> 6) & 0x01; 113 fThrottlingIn = fThrottlingIn || (throttlingIn1 || throttlingIn2); // 01 114 fThrottlingOut = fThrottlingOut || ((word->hbflag >> 4) & 0x02); // 10 115 116 continue; 117 118 } else if ((h == Data::Heartbeat2nd) && (hbf_flag == 1)) { 119 //LOG(debug) << " 2nd delimiter comes h = " << std::hex << static_cast<uint16_t>(h) 120 // << ", raw = " << word->raw; 121 hbf_flag++; 122 } else if ((h == Data::Heartbeat2nd) && (hbf_flag == 0)) { 123 LOG(warn) << "Only second HBF delimiter comes..."; 124 offset = i+1; 125 hbf_flag = 0; 126 continue; 127 } 128 129 /// 130 if( (h == Data::Heartbeat2nd) && (fHBFCounter==0) ){ 131 auto rdelim = fdelimiterFrameId % fMaxHBF; 132 // LOG(debug) << "rdelim: "<< rdelim; 133 // LOG(debug) << "hbfframe : "<< fdelimiterFrameId; 134 // LOG(debug) << "fMaxHBF: "<< fMaxHBF; 135 136 if(rdelim != 0){ 137 offset = i+1; 138 hbf_flag = 0; 139 LOG(warn) << "(HBF % fMaxHBF) != 0: HBF = " << fdelimiterFrameId 140 << " fMaxHBF = " << fMaxHBF; 141 continue; 142 } 143 } 144 /// 145 146 if (fTimeFrameIdType == TimeFrameIdType::FirstHeartbeatDelimiter) { 147 // first heartbeat delimiter or first spill-off delimiter 148 if (fSTFId<0) { 149 fSTFId = fdelimiterFrameId; 150 } 151 } else { // last heartbeat delimiter or last spill-off delimiter, or sequence number 152 fSTFId = fdelimiterFrameId; 153 } 154 155 if(mdebug) { 156 LOG(debug) << " Fill " << std::setw(10) << offset << " -> " << std::setw(10) << i << " : " << std::hex << word->raw << std::dec; 157 } 158 159 auto first = msgBegin + offset; 160 auto last = msgBegin + i; 161 offset = i+1; 162 163 FillData(first, last); 164 hbf_flag = 0; 165 166 if ( h == Data::Heartbeat2nd ) { 167 ++fHBFCounter; 168 169 if (fSplitMethod == 0) { 170 171 if ((fHBFCounter % fMaxHBF == 0) && (fHBFCounter>0)) { 172 FinalizeSTF(); 173 } 174 } 175 } 176 } 177 } 178 179 if (offset < nWord) { 180 fInputPayloads.insert(fInputPayloads.end(), 181 std::make_move_iterator(msgBegin + offset), 182 std::make_move_iterator(msgBegin + nWord)); 183 } 184 } 185 186 //______________________________________________________________________________ 187 void 188 AmQStrTdcSTFBuilder::FillData(AmQStrTdc::Data::Word* first, AmQStrTdc::Data::Word* last) 189 { 190 namespace Data = AmQStrTdc::Data; 191 // construct send buffer with remained data on heap buf + first...last 192 auto buf = std::make_unique<decltype(fInputPayloads)>(std::move(fInputPayloads)); 193 194 /* for data */ 195 if ( (last != first) && (hbf_flag == 2) ) { 196 // buf[data] + first...last[first delimiter, second delimiter] 197 // or buf[data] + first...last[data + first delimiter, second delimiter] 198 // insert new data to send buffer 199 buf->insert(buf->end(), std::make_move_iterator(first), std::make_move_iterator(last)); 200 buf->insert(buf->end(), *last); 201 202 if(mdebug){ 203 for(unsigned int a=0; a<buf->size(); ++a) 204 LOG(debug) << "buf[" << a << "]" << std::hex << reinterpret_cast<Data::Bits*>(&(buf->at(a)))->raw; 205 206 LOG(debug) << " last != first 1 : first = " << std::hex << reinterpret_cast<Data::Bits*>(first)->raw; 207 LOG(debug) << " last != first > 1 : last = " << std::hex << reinterpret_cast<Data::Bits*>(last)->raw; 208 } 209 210 }else if ((last == first) && (hbf_flag ==2)) { // buf[data + first delimiter] + last[second delimiter] 211 buf->insert(buf->end(), *last); 212 } 213 214 auto sbuf = std::make_unique<decltype(fInputDataloads)>(std::move(fInputDataloads)); 215 HBFHeader hbfHeader; 216 hbfHeader.length = sizeof(hbfHeader) + buf->size()*sizeof(uint64_t); 217 hbfHeader.hLength = sizeof(hbfHeader); // byte 218 hbfHeader.reserve = 0; 219 220 const auto whbfHeader = reinterpret_cast<Data::Word*>(&hbfHeader); 221 sbuf->insert(sbuf->end(), *whbfHeader); 222 sbuf->insert(sbuf->end(), *(whbfHeader+1)); 223 sbuf->insert(sbuf->end(), buf->begin(), buf->end()); 224 225 if(mdebug){ 226 for(unsigned int a=0; a<sbuf->size(); ++a) 227 LOG(debug) << "sbuf[" << a << "]" << std::hex << reinterpret_cast<Data::Bits*>(&(sbuf->at(a)))->raw; 228 } 229 230 NewData(); 231 if (!buf->empty()) { 232 fWorkingPayloads->emplace_back(MessageUtil::NewMessage(*this, std::move(sbuf))); 233 } 234 235 if (fSplitMethod!=0) { 236 if ((fHBFCounter % fMaxHBF == 0) && (fHBFCounter>0)) { 237 LOG(debug) << " calling FinalizeSTF() from FillData()"; 238 FinalizeSTF(); 239 NewData(); 240 } 241 } 242 243 } 244 245 //______________________________________________________________________________ 246 void AmQStrTdcSTFBuilder::FinalizeSTF() 247 { 248 //LOG(debug) << " FinalizeSTF()"; 249 auto stfHeader = std::make_unique<STF::Header>(); 250 if (fTimeFrameIdType==TimeFrameIdType::SequenceNumberOfTimeFrames) { 251 stfHeader->timeFrameId = fSTFSequenceNumber; 252 } else { 253 stfHeader->timeFrameId = fSTFId; 254 } 255 fSTFId = -1; 256 stfHeader->femType = fFEMType; 257 stfHeader->femId = fFEMId; 258 stfHeader->length = std::accumulate(fWorkingPayloads->begin(), fWorkingPayloads->end(), sizeof(STF::Header), 259 [](auto init, auto& m) { 260 return (!m) ? init : init + m->GetSize(); 261 }); 262 263 stfHeader->hLength = sizeof(*stfHeader); 264 stfHeader->numMessages = fWorkingPayloads->size() - 1; 265 266 struct timeval curtime; 267 gettimeofday(&curtime, NULL); 268 stfHeader->timeSec = curtime.tv_sec; 269 stfHeader->timeUSec = curtime.tv_usec; 270 271 stfHeader->type = fThrottlingOut || fThrottlingIn; 272 273 fThrottlingOut = 0; 274 fThrottlingIn = 0; 275 276 // std::cout << "femtype: "<< stfHeader->femType << std::endl; 277 // std::cout << "sec: "<< stfHeader->timeSec << std::endl; 278 // std::cout << "usec: "<< stfHeader->timeUSec << std::endl; 279 280 // replace first element with STF header 281 fWorkingPayloads->at(0) = MessageUtil::NewMessage(*this, std::move(stfHeader)); 282 283 fOutputPayloads.emplace(std::move(fWorkingPayloads)); 284 285 ++fSTFSequenceNumber; 286 fHBFCounter = 0; 287 288 } 289 290 291 //______________________________________________________________________________ 292 bool AmQStrTdcSTFBuilder::HandleData(FairMQMessagePtr& msg, int index) 293 { 294 namespace Data = AmQStrTdc::Data; 295 296 if(mdebug) 297 std::cout << "HandleData() HBF " << fHBFCounter << " input message " << msg->GetSize() << std::endl; 298 299 //Reporter::AddInputMessageSize(msg->GetSize()); 300 301 302 // std::cout << "============ data in ============= "<< std::endl; 303 // auto indata_size = msg->GetSize(); 304 // std::for_each(reinterpret_cast<uint64_t*>(msg->GetData()), 305 // reinterpret_cast<uint64_t*>(msg->GetData() + msg->GetSize()), 306 // ::HexDump{4}); 307 308 BuildFrame(msg, index); 309 310 while (!fOutputPayloads.empty()) { 311 // create a multipart message and move ownership of messages to the multipart message 312 FairMQParts parts; 313 FairMQParts dqmParts; 314 315 bool dqmSocketExists = fChannels.count(fDQMChannelName); 316 317 if(mdebug) 318 std::cout << " send data " << fOutputPayloads.size() << std::endl; 319 320 auto& payload = fOutputPayloads.front(); 321 322 for (auto& tmsg : *payload) { 323 324 if (dqmSocketExists) { 325 if (tmsg->GetSize()==sizeof(STF::Header)) { 326 327 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 328 msgCopy->Copy(*tmsg); 329 dqmParts.AddPart(std::move(msgCopy)); 330 } else { 331 /* 332 auto b = reinterpret_cast<Bits*>(tmsg->GetData()); 333 if (b->head == Data::Heartbeat || 334 // b->head == Data::ErrorRecovery || 335 b->head == Data::Data || 336 b->head == Data::SpillEnd) { 337 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 338 msgCopy->Copy(*tmsg); 339 dqmParts.AddPart(std::move(msgCopy)); 340 } 341 */ 342 343 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 344 msgCopy->Copy(*tmsg); 345 dqmParts.AddPart(std::move(msgCopy)); 346 } 347 } 348 349 parts.AddPart(std::move(tmsg)); 350 } 351 352 fOutputPayloads.pop(); 353 354 /* 355 #if 1 356 { // for debug-begin 357 std::cout << " parts size = " << parts.Size() << std::endl; 358 for (int i=0; i<parts.Size(); ++i){ 359 const auto& msg = parts.At(i); 360 361 if (i==0) { 362 auto stfh = reinterpret_cast<STF::Header*>(msg->GetData()); 363 LOG(debug) << "STF " << stfh->timeFrameId << " length " << stfh->length << " header " << msg->GetSize() << std::endl; 364 auto msize = msg->GetSize(); 365 std::for_each(reinterpret_cast<uint64_t*>(msg->GetData()), 366 reinterpret_cast<uint64_t*>(msg->GetData() + msize), 367 ::HexDump{4}); 368 } else { 369 LOG(debug) << " body " << i << " " << msg->GetSize() << " " 370 << std::showbase << std::hex << msg->GetSize() << std::noshowbase<< std::dec << std::endl; 371 auto n = msg->GetSize()/sizeof(Data::Word); 372 373 std::for_each(reinterpret_cast<Data::Word*>(msg->GetData()), 374 reinterpret_cast<Data::Word*>(msg->GetData()) + n, 375 ::HexDump{4}); 376 } 377 } 378 379 } // for debug-end 380 // fH_flag = 0; 381 #endif 382 */ 383 // Push multipart message into send queue 384 // LOG(debug) << "send multipart message "; 385 386 auto h = reinterpret_cast<STF::Header*>(parts.At(0)->GetData()); 387 388 if (dqmSocketExists) { 389 if (Send(dqmParts, fDQMChannelName) < 0) { 390 391 // timeout 392 if (NewStatePending()) { 393 LOG(info) << "Device is not RUNNING"; 394 return false; 395 } 396 LOG(error) << "Failed to enqueue sub time frame (DQM) : FEM = " << std::hex << h->femId 397 << std::dec << " STF = " << h->timeFrameId << std::endl; 398 } 399 } 400 401 auto direction = (fTimeFrameIdType==TimeFrameIdType::SequenceNumberOfTimeFrames) 402 ? (h->timeFrameId % fNumDestination) 403 : ((h->timeFrameId/fMaxHBF) % fNumDestination); 404 405 unsigned int err_count = 0; 406 407 while (Send(parts, fOutputChannelName, direction, 0) < 0) { 408 // timeout 409 if (NewStatePending()) { 410 LOG(info) << "Device is not RUNNING"; 411 return false; 412 } 413 if (err_count < 10) 414 LOG(warn) << "Failed to enqueue sub time frame (data) : FEM = " << std::hex << h->femId << std::dec 415 << " STF = " << h->timeFrameId 416 << " direction = " << direction << std::endl; 417 else if (fDiscardOutput && (err_count == 10)) { 418 LOG(error) << "Failed to enqueue sub time frame after 10 attempts. Discarding data : FEM = " 419 << std::hex << h->femId << std::dec << " STF = " << h->timeFrameId 420 << " direction = " << direction << std::endl; 421 break; // parts will be discarded because the parts is declared within the while loop scope. 422 } 423 err_count++; 424 } 425 } 426 427 return true; 428 } 429 430 //______________________________________________________________________________ 431 void AmQStrTdcSTFBuilder::Init() 432 { 433 //Reporter::Instance(fConfig); 434 } 435 436 //______________________________________________________________________________ 437 void AmQStrTdcSTFBuilder::InitTask() 438 { 439 440 using opt = OptionKey; 441 /* 442 { 443 // convert IP address to 32 bit. e.g. 192.168.10.16 -> FEM id = 0xC0A80A10 444 // fFEMId = 0; 445 446 auto femid = fConfig->GetProperty<std::string>(opt::FEMId.data()); 447 std::istringstream iss(femid); 448 std::string token; 449 int i = 3; 450 while (std::getline(iss, token, '.')) { 451 if (i<0) break; 452 uint32_t v = (std::stoul(token) & 0xff) << (8*i); 453 // std::cout << " i = " << i << " token = " << token << " v = " << v << std::endl; 454 fFEMId |= v; 455 --i; 456 } 457 LOG(debug) << "FEM ID " << std::hex << fFEMId << std::dec << std::endl; 458 } 459 */ 460 461 fInputChannelName = fConfig->GetProperty<std::string>(opt::InputChannelName.data()); 462 fOutputChannelName = fConfig->GetProperty<std::string>(opt::OutputChannelName.data()); 463 fDQMChannelName = fConfig->GetProperty<std::string>(opt::DQMChannelName.data()); 464 465 fTimeFrameIdType = static_cast<TimeFrameIdType>(std::stoi(fConfig->GetProperty<std::string>(opt::TimeFrameIdType.data()))); 466 467 ////// 468 FairMQMessagePtr msginfo(NewMessage()); 469 int nrecv=0; 470 while(true) { 471 if (Receive(msginfo, fInputChannelName) <= 0) { 472 LOG(debug) << __func__ << " Trying to get FEMIfo " << nrecv; 473 nrecv++; 474 } else { 475 break; 476 } 477 } 478 479 // fromFEMInfo feminfo; 480 auto femInfo = reinterpret_cast<FEMInfo*>(msginfo->GetData()); 481 auto femID = femInfo->femId; 482 auto femType = femInfo->femType; 483 484 LOG(info) << "magic: "<< std::hex << femInfo->magic << std::dec; 485 LOG(info) << "femId: "<< std::hex << femID << std::dec; 486 LOG(info) << "femType: "<< std::hex << femType << std::dec; 487 //// 488 489 fFEMId = femID; 490 fFEMType = femType; 491 492 auto s_maxHBF = fConfig->GetProperty<std::string>(opt::MaxHBF.data()); 493 fMaxHBF = std::stoi(s_maxHBF); 494 if (fMaxHBF<1) { 495 LOG(warn) << "fMaxHBF: non-positive value was specified = " << fMaxHBF; 496 fMaxHBF = 1; 497 } 498 LOG(debug) << "fMaxHBF = " <<fMaxHBF; 499 500 auto s_splitMethod = fConfig->GetProperty<std::string>(opt::SplitMethod.data()); 501 fSplitMethod = std::stoi(s_splitMethod); 502 503 504 LOG(debug) << " output channels: name = " << fOutputChannelName 505 << " num = " << GetNumSubChannels(fOutputChannelName); 506 fNumDestination = GetNumSubChannels(fOutputChannelName); 507 LOG(debug) << " number of desntination = " << fNumDestination; 508 if (fNumDestination<1) { 509 LOG(warn) << " number of destination is non-positive"; 510 } 511 512 513 LOG(debug) << " data quality monitoring channels: name = " << fDQMChannelName; 514 // << " num = " << fChannels.count(fDQMChannelName); 515 516 // if (fChannels.count(fDQMChannelName)) { 517 // LOG(debug) << " data quality monitoring channels: name = " << fDQMChannelName 518 // << " num = " << fChannels.at(fDQMChannelName).size(); 519 // } 520 521 auto s_DiscardOutput = fConfig->GetProperty<std::string>(opt::DiscardOutput.data()); 522 fDiscardOutput = ((s_DiscardOutput == "1") || (s_DiscardOutput == "true") || (s_DiscardOutput == "yes")); 523 LOG(debug) << " Discard output option: fDiscardOutput = " << fDiscardOutput; 524 525 OnData(fInputChannelName, &AmQStrTdcSTFBuilder::HandleData); 526 527 //Reporter::Reset(); 528 } 529 530 //______________________________________________________________________________ 531 void AmQStrTdcSTFBuilder::NewData() 532 { 533 if (!fWorkingPayloads) { 534 fWorkingPayloads = std::make_unique<std::vector<FairMQMessagePtr>>(); 535 fWorkingPayloads->reserve(fMaxHBF*2+1); 536 // add an empty message, which will be replaced with sub-time-frame header later. 537 fWorkingPayloads->push_back(nullptr); 538 } 539 } 540 541 //______________________________________________________________________________ 542 void AmQStrTdcSTFBuilder::PostRun() 543 { 544 fInputPayloads.clear(); 545 fInputDataloads.clear(); 546 fWorkingPayloads.reset(); 547 SendBuffer ().swap(fOutputPayloads); 548 549 fThrottlingOut = 0; 550 fThrottlingIn = 0; 551 552 fLastHeader = 0; 553 hbf_flag = 0; 554 555 int nrecv = 0; 556 557 while(true) { 558 FairMQMessagePtr msg(NewMessage()); 559 560 if (Receive(msg, fInputChannelName) <= 0) { 561 // LOG(debug) << __func__ << " no data received " << nrecv; 562 ++nrecv; 563 if (nrecv>10) { 564 break; 565 } 566 std::this_thread::sleep_for(std::chrono::milliseconds(200)); 567 } else { 568 // LOG(debug) << __func__ << " print data"; 569 // HandleData(msg, 0); 570 } 571 } 572 573 LOG(debug) << __func__ << " done"; 574 575 } 576 577 namespace bpo = boost::program_options; 578 579 //______________________________________________________________________________ 580 void AmQStrTdcSTFBuilder::PreRun() 581 { 582 fHBFCounter = 0; 583 hbf_flag = 0; 584 fdelimiterFrameId = 0; 585 fSTFSequenceNumber = 0; 586 fLastHeader = 0; 587 fSTFId = -1; 588 } 589 590 //______________________________________________________________________________ 591 void addCustomOptions(bpo::options_description& options) 592 { 593 using opt = AmQStrTdcSTFBuilder::OptionKey; 594 options.add_options() 595 (opt::FEMId.data(), bpo::value<std::string>(), "FEM ID") 596 (opt::InputChannelName.data(), bpo::value<std::string>()->default_value("in"), "Name of the input channel") 597 (opt::OutputChannelName.data(), bpo::value<std::string>()->default_value("out"), "Name of the output channel") 598 (opt::DQMChannelName.data(), bpo::value<std::string>()->default_value("dqm"), "Name of the data quality monitoring") 599 (opt::MaxHBF.data(), bpo::value<std::string>()->default_value("1"), "maximum number of heartbeat frame in one sub time frame") 600 (opt::SplitMethod.data(), bpo::value<std::string>()->default_value("0"), "STF split method") 601 (opt::TimeFrameIdType.data(), bpo::value<std::string>()->default_value("0"), "Time frame ID type: 0 = first HB delimiter, 1 = last HB delimiter, 2 = sequence number of time frames") 602 (opt::DiscardOutput.data(), bpo::value<std::string>()->default_value("false"),"Discard Output Data during conditional run caused by back pressure. true: discard the data, false: not discard the data.") 603 ; 604 } 605 606 //______________________________________________________________________________ 607 std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions&) 608 { 609 return std::make_unique<AmQStrTdcSTFBuilder>(); 610 }
STFBuilder.h
SHA-256: 290970dcf71f2731777100175afc92011b03c77310a781ec7ac25d791c90d8b1
1 #ifndef STFBuilder_H 2 #define STFBuilder_H 3 4 #include <string> 5 #include <string_view> 6 #include <cstdint> 7 #include <deque> 8 #include <memory> 9 #include <queue> 10 11 #include <fairmq/Device.h> 12 13 #include "AmQStrTdcData.h" 14 15 /* HBFRAME */ 16 //constexpr uint64_t HBFMagic = 0x00454d4152464248; 17 // HRTBEAT 18 constexpr uint64_t HBFMagic = 0x0054414542545248; 19 #pragma pack(4) 20 struct HBFHeader { 21 uint64_t magic {HBFMagic}; 22 uint32_t length {0}; 23 uint16_t hLength {0}; 24 uint16_t reserve {0}; 25 }; 26 #pragma pack() 27 28 struct FEMInfo { 29 uint64_t magic {0}; 30 uint32_t femType {0}; 31 uint32_t femId {0}; 32 uint64_t reserved {0}; 33 }; 34 35 enum class TimeFrameIdType : int { 36 FirstHeartbeatDelimiter = 0, 37 LastHeartbeatDelimiter, 38 SequenceNumberOfTimeFrames 39 }; 40 41 42 class AmQStrTdcSTFBuilder : public fair::mq::Device 43 { 44 public: 45 using WorkBuffer = std::unique_ptr<std::vector<FairMQMessagePtr>>; 46 using RecvBuffer = std::vector<AmQStrTdc::Data::Word>; //64 bits 47 using SendBuffer = std::queue<WorkBuffer>; 48 49 struct OptionKey { 50 static constexpr std::string_view FEMId {"fem-id"}; 51 static constexpr std::string_view InputChannelName {"in-chan-name"}; 52 static constexpr std::string_view OutputChannelName {"out-chan-name"}; 53 static constexpr std::string_view DQMChannelName {"dqm-chan-name"}; 54 static constexpr std::string_view StripHBF {"strip-hbf"}; 55 static constexpr std::string_view MaxHBF {"max-hbf"}; 56 static constexpr std::string_view SplitMethod {"split"}; 57 static constexpr std::string_view TimeFrameIdType {"time-frame-id-type"}; 58 static constexpr std::string_view DiscardOutput {"discard-output"}; 59 }; 60 61 AmQStrTdcSTFBuilder(); 62 AmQStrTdcSTFBuilder(const AmQStrTdcSTFBuilder&) = delete; 63 AmQStrTdcSTFBuilder& operator=(const AmQStrTdcSTFBuilder&) = delete; 64 ~AmQStrTdcSTFBuilder() = default; 65 66 private: 67 void BuildFrame(FairMQMessagePtr& msg, int index); 68 void FillData(AmQStrTdc::Data::Word* first, 69 AmQStrTdc::Data::Word* last); 70 void FinalizeSTF(); 71 bool HandleData(FairMQMessagePtr&, int index); 72 void Init() override; 73 void InitTask() override; 74 void NewData(); 75 void PostRun() override; 76 void PreRun() override; 77 78 uint64_t fFEMId {0}; 79 uint64_t fFEMType {0}; 80 int fNumDestination {0}; 81 std::string fInputChannelName; 82 std::string fOutputChannelName; 83 std::string fDQMChannelName; 84 int fMaxHBF {1}; 85 int fHBFCounter {0}; 86 int hbf_flag {0}; 87 88 int32_t fdelimiterFrameId {0}; 89 uint32_t fSTFSequenceNumber {0}; 90 int fSplitMethod {0}; 91 uint8_t fLastHeader {0}; 92 int fH_flag {0}; 93 uint16_t fThrottlingIn {0}; 94 uint16_t fThrottlingOut {0}; 95 96 97 TimeFrameIdType fTimeFrameIdType; 98 int32_t fSTFId{-1}; // 8-bit spill counter and 16-bit HB frame from heartbeat delimiter 99 bool fDiscardOutput; 100 101 bool mdebug; 102 RecvBuffer fInputPayloads; 103 RecvBuffer fInputDataloads; 104 WorkBuffer fWorkingPayloads; 105 SendBuffer fOutputPayloads; 106 }; 107 108 #endif
SubTimeFrameHeader.h
SHA-256: c04e42238114e70b72a1904ddc87a058b965f4bfee4acdb26ee0af6859a01b94
1 #ifndef SubTimeFrameHeader_h 2 #define SubTimeFrameHeader_h 3 4 #include <cstdint> 5 6 namespace SubTimeFrame { 7 8 // This format is temporary and should be updated. 9 namespace v0 { 10 11 // "DAEH-FTS" : little endian of "STF-HEAD" 12 constexpr uint64_t MAGIC {0x444145482d465453}; 13 //constexpr uint32_t TDC64H {0x48434454}; 14 //constexpr uint32_t TDC64L {0x4c434454}; 15 constexpr uint32_t TDC64H {2}; 16 constexpr uint32_t TDC64H_V1 {2}; 17 constexpr uint32_t TDC64L {3}; 18 constexpr uint32_t TDC64L_V1 {1}; 19 constexpr uint32_t TDC64L_V2 {3}; 20 constexpr uint32_t TDC64H_V3 {5}; 21 constexpr uint32_t TDC64L_V3 {6}; 22 23 constexpr uint32_t FLT_TDC {0x0000'1000}; 24 constexpr uint32_t NULDEV {0}; 25 26 struct Header { 27 uint64_t magic {MAGIC}; 28 uint32_t timeFrameId {0}; 29 uint32_t reserved {0}; 30 uint32_t femType {0}; 31 uint32_t femId {0}; 32 uint32_t length {0}; 33 uint32_t numMessages {0}; 34 uint64_t timeSec {0}; 35 uint64_t timeUSec {0}; 36 }; 37 38 } // namespace v0 39 40 inline namespace v1 { 41 42 // " EMITBUS" : little endian of "SUBTIME " 43 constexpr uint64_t MAGIC {0x00454d4954425553}; 44 constexpr uint32_t TDC64H {2}; 45 constexpr uint32_t TDC64H_V1 {2}; 46 constexpr uint32_t TDC64L {3}; 47 constexpr uint32_t TDC64L_V1 {1}; 48 constexpr uint32_t TDC64L_V2 {3}; 49 //constexpr uint32_t TDC64H_V3 {4}; 50 //constexpr uint32_t TDC64L_V3 {5}; 51 constexpr uint32_t TDC64H_V3 {5}; 52 constexpr uint32_t TDC64L_V3 {6}; 53 constexpr uint32_t INPUT_REG {21}; 54 constexpr uint32_t FLT_TDC {0x0000'1000}; 55 56 constexpr uint32_t NULDEV {0}; 57 58 constexpr uint16_t META {1}; 59 60 struct Header { 61 uint64_t magic {MAGIC}; 62 uint32_t length {0}; 63 uint16_t hLength {48}; 64 uint16_t type {0}; 65 uint32_t timeFrameId {0}; 66 uint32_t femType {0}; 67 uint32_t femId {0}; 68 uint32_t numMessages {0}; 69 uint64_t timeSec {0}; 70 uint64_t timeUSec {0}; 71 }; 72 73 } // namespace v1 74 75 } // namespace SubTimeFrame 76 77 #endif
TimeFrameBuilder.cxx
SHA-256: 46f543a47c0ad5d2bd8713e48a1eb3ef1877294258e645d1ec7ad1238df23ec2
1 #include <chrono> 2 #include <functional> 3 #include <thread> 4 #include <cassert> 5 #include <algorithm> 6 #include <numeric> 7 #include <sstream> 8 #include <ctime> 9 10 #include <fairmq/Poller.h> 11 #include <fairmq/runDevice.h> 12 13 #include "utility/HexDump.h" 14 #include "utility/MessageUtil.h" 15 16 #include "RedisDataStore.h" 17 18 #include "SubTimeFrameHeader.h" 19 #include "TimeFrameHeader.h" 20 #include "TimeFrameBuilder.h" 21 22 23 //#include "AmQStrTdcData.h" 24 25 namespace bpo = boost::program_options; 26 27 //______________________________________________________________________________ 28 void addCustomOptions(bpo::options_description& options) 29 { 30 using opt = TimeFrameBuilder::OptionKey; 31 options.add_options() 32 (opt::BufferTimeoutInMs.data(), bpo::value<std::string>()->default_value("10000"), "Buffer timeout in milliseconds") 33 (opt::BufferDepthLimit.data(), bpo::value<std::string>()->default_value("10000"), "Buffer depth limit") 34 (opt::InputChannelName.data(), bpo::value<std::string>()->default_value("in"), "Name of the input channel") 35 (opt::OutputChannelName.data(), bpo::value<std::string>()->default_value("out"), "Name of the output channel") 36 (opt::DQMChannelName.data(), bpo::value<std::string>()->default_value("dqm"), "Name of the data quality monitoring channel") 37 (opt::DecimatorChannelName.data(), bpo::value<std::string>()->default_value("decimator"), "Name of the decimated output channel") 38 (opt::PollTimeout.data(), bpo::value<std::string>()->default_value("0"), "Timeout (in msec) of polling") 39 (opt::DecimationFactor.data(), bpo::value<std::string>()->default_value("0"), "Decimation factor for decimated output channel") 40 (opt::DecimationOffset.data(), bpo::value<std::string>()->default_value("0"), "Decimation offset for decimated output channel") 41 (opt::DiscardOutput.data(), bpo::value<std::string>()->default_value("false"), "Discard output option to eliminate the back pressure for upstream FairMQ device. true: discard, false (default): no discard causing back pressure") 42 (opt::OutputIncompleteTF.data(), bpo::value<std::string>()->default_value("false"), "Output incomplete Time Frame") 43 (opt::ObjectDbNumber.data(), bpo::value<std::string>()->default_value("3"), "DB number for DQM objects") 44 (opt::EnableCheckHBF.data(), bpo::value<std::string>()->default_value("true"), "Check HBF delimiter") 45 ; 46 } 47 48 49 //______________________________________________________________________________ 50 std::unique_ptr<fair::mq::Device> getDevice(FairMQProgOptions&) 51 { 52 return std::make_unique<TimeFrameBuilder>(); 53 } 54 55 56 //______________________________________________________________________________ 57 TimeFrameBuilder::TimeFrameBuilder() 58 : fair::mq::Device(), fKt(1000) 59 { 60 } 61 62 63 void TimeFrameBuilder::TFBFailDump(std::vector<STFBuffer> &tfBuf, uint32_t stfId) 64 { 65 66 std::vector<uint32_t> femid; 67 std::vector<uint32_t> expected = { 68 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 69 170, 171, 172, 173, 174, 175, 176, 177, 178, 179 70 }; 71 std::vector<uint64_t> hb; 72 for (auto & stfBuf : tfBuf) { 73 auto & msg = stfBuf.parts[0]; 74 SubTimeFrame::Header *stfheader 75 = reinterpret_cast<SubTimeFrame::Header *>(msg.GetData()); 76 femid.push_back(stfheader->femId); 77 for (auto it = expected.begin() ; it != expected.end() ;) { 78 if (*it == (stfheader->femId && 0xff)) { 79 it = expected.erase(it); 80 } else { 81 it++; 82 } 83 } 84 //std::for_each(reinterpret_cast<uint64_t*>(msg.GetData()), 85 // reinterpret_cast<uint64_t*>(msg.GetData() + msg.GetSize()), 86 // ::HexDump{4}); 87 88 if (stfBuf.parts.Size() > 2) { 89 auto & hb0 = stfBuf.parts[2]; 90 uint64_t hb00 = (reinterpret_cast<uint64_t *>(hb0.GetData()))[0]; 91 hb.push_back(hb00); 92 } 93 } 94 95 #if 1 96 std::cout << "#D TFB Fail. TFid: " << stfId << "(0x" << std::hex << stfId << "), " 97 << std::dec << "N: " << femid.size() << "/" << fNumSource << ", FEid:"; 98 for (auto & i : femid) std::cout << " " << (i & 0xff); 99 std::cout << std::endl; 100 #endif 101 102 #if 0 103 std::cout << "#D HB :" << stfId << ":"; 104 for (auto & i : hb) std::cout << " " << std::hex <<i; 105 std::cout << std::dec << std::endl; 106 #endif 107 108 return; 109 } 110 111 112 void TimeFrameBuilder::TFBSegmentCheck(std::vector<STFBuffer> &tfBuf) 113 { 114 std::vector<uint32_t> expected = { 115 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 116 170, 171, 172, 173, 174, 175, 176, 177, 178, 179 117 }; 118 std::vector<uint64_t> hb; 119 for (auto & stfBuf : tfBuf) { 120 auto & msg = stfBuf.parts[0]; 121 SubTimeFrame::Header *stfheader 122 = reinterpret_cast<SubTimeFrame::Header *>(msg.GetData()); 123 uint32_t femid = stfheader->femId; 124 if (fSegmentCounts.count(femid) > 0) { 125 fSegmentCounts[femid]++; 126 } else { 127 fSegmentCounts[femid] = 1; 128 } 129 } 130 131 //check segment counts 132 133 return; 134 } 135 136 void TimeFrameBuilder::StoreSuccessfulSegmentIDs(const std::vector<STFBuffer> &stfBuf) 137 { 138 fSuccessfulSegmentIDs.clear(); 139 fSuccessfulSegmentIDs.resize(0); 140 141 for (auto & stf : stfBuf) { 142 auto & msg = stf.parts[0]; 143 SubTimeFrame::Header *stfheader 144 = reinterpret_cast<SubTimeFrame::Header *>(msg.GetData()); 145 uint32_t femid = stfheader->femId; 146 fSuccessfulSegmentIDs.emplace_back(femid); 147 } 148 #if 0 149 std::sort(fSuccessfulSegmentIDs.begin(), fSuccessfulSegmentIDs.end()); 150 #endif 151 152 return; 153 } 154 155 std::vector<uint32_t> TimeFrameBuilder::CheckLostSegmentIDs(const std::vector<STFBuffer> &stfBuf) 156 { 157 std::vector<uint32_t> lostSegmentIds; 158 std::vector<uint32_t> existedSegmentIds; 159 160 for (auto & stf : stfBuf) { 161 auto & msg = stf.parts[0]; 162 SubTimeFrame::Header *stfheader 163 = reinterpret_cast<SubTimeFrame::Header *>(msg.GetData()); 164 uint32_t femid = stfheader->femId; 165 existedSegmentIds.emplace_back(femid); 166 } 167 168 for (auto & femid : fSuccessfulSegmentIDs) { 169 170 #if 0 171 std::cout << "#D Succeeded femid: " << femid << " (" << (femid & 0xff) << ") : " 172 << [&existedSegmentIds](uint32_t x) { 173 if (std::find( 174 existedSegmentIds.begin(), existedSegmentIds.end(), x) 175 == existedSegmentIds.end()) { 176 return "lost"; 177 } else { 178 return "exists"; 179 } 180 }(femid) 181 << std::endl; 182 #endif 183 184 if (std::find(existedSegmentIds.begin(), existedSegmentIds.end(), femid) 185 == existedSegmentIds.end()) { 186 lostSegmentIds.emplace_back(femid); 187 fLostSegmentCounts[femid]++; 188 } 189 } 190 191 return lostSegmentIds; 192 } 193 194 195 int TimeFrameBuilder::CheckHBFDelimitor(fair::mq::Parts &inParts, uint32_t stfId) 196 { 197 auto nmsg = inParts.Size(); 198 const auto& msg =inParts.At(nmsg -1); 199 200 auto msgSize = msg->GetSize(); 201 auto nWord = msgSize / sizeof(uint64_t); 202 auto msgBegin = reinterpret_cast<uint64_t *>(msg->GetData()); 203 204 auto firstHBF = reinterpret_cast<uint64_t *>(msgBegin + (nWord - 2)); 205 //auto secondHBF = reinterpret_cast<uint64_t *>(msgBegin + (nWord - 1)); 206 207 auto fem = reinterpret_cast<SubTimeFrame::Header*>(inParts.At(0)->GetData())->femId; 208 auto lastmsg = reinterpret_cast<uint64_t *>(inParts.At(inParts.Size() - 1)->GetData()); 209 unsigned int type = (firstHBF[0] & 0xfc00'0000'0000'0000) >> 58; 210 //if ((type == 0x1c) || (type == 0x18) || (type == 0x14) || (type == 0x1e)) { 211 if ((type == 0x1c) || (type == 0x1e)) { 212 } else { 213 LOG(warn) << "BAD delimitor " << std::hex << lastmsg[0] 214 << " FEM: " << std::dec << (fem & 0xff) 215 << " TFid: " << stfId; 216 return -1; 217 } 218 219 return 0; 220 } 221 222 223 void TimeFrameBuilder::MakeOutPartsFromSTFBuffers( 224 std::vector<STFBuffer>& tfBuf, 225 uint32_t stfId, 226 uint16_t type, 227 fair::mq::Parts& outParts) 228 { 229 auto h = std::make_unique<TimeFrame::Header>(); 230 h->magic = TimeFrame::MAGIC; 231 #if 0 232 h->timeFrameId = reinterpret_cast<SubTimeFrame::Header*>(tfBuf[0].parts.At(0)->GetData())->timeFrameId; 233 #else 234 h->timeFrameId = stfId; 235 #endif 236 h->numSource = fNumSource; 237 h->type = type; 238 h->length = std::accumulate(tfBuf.begin(), tfBuf.end(), sizeof(TimeFrame::Header), 239 [](auto init, auto& stfBuf) { 240 return init + std::accumulate(stfBuf.parts.begin(), stfBuf.parts.end(), 0, 241 [] (auto jinit, auto& m) { 242 return (!m) ? jinit : jinit + m->GetSize(); 243 }); 244 }); 245 246 outParts.AddPart(MessageUtil::NewMessage(*this, std::move(h))); 247 for (auto& stfBuf: tfBuf) { 248 for (auto& m: stfBuf.parts) { 249 outParts.AddPart(std::move(m)); 250 } 251 } 252 253 return; 254 } 255 256 257 void TimeFrameBuilder::SendTimeFrameForDQM( 258 fair::mq::Parts& outParts) 259 { 260 auto poller = NewPoller(fDQMChannelName); 261 poller->Poll(fPollTimeoutMS); 262 int numDestDQM = GetNumSubChannels(fDQMChannelName); 263 for (int i = 0 ; i < numDestDQM ; i++) { 264 if (poller->CheckOutput(fDQMChannelName, i)) { 265 fair::mq::Parts dqmParts = MessageUtil::Copy(*this, outParts); 266 if (Send(dqmParts, fDQMChannelName, i) < 0) { 267 if (NewStatePending()) { 268 LOG(info) << "Device is not RUNNING"; 269 } else { 270 LOG(error) << "Failed to enqueue TFB (DQM) : " << i; 271 } 272 } 273 } 274 } 275 276 #if 0 277 if (trySend) { 278 fair::mq::Parts dqmParts = MessageUtil::Copy(*this, outParts); 279 if (Send(dqmParts, fDQMChannelName) < 0) { 280 if (NewStatePending()) { 281 LOG(info) << "Device is not RUNNING"; 282 } else { 283 LOG(error) << "Failed to enqueue TFB (DQM) "; 284 } 285 } 286 } 287 #endif 288 289 return; 290 } 291 292 293 void TimeFrameBuilder::SendTimeFrameForDecimator( 294 fair::mq::Parts& outParts) 295 { 296 297 if ((fDecimatorNumberOfConnectedPeers > 0) 298 && (fDecimationFactor > 0) 299 && (fNumSend % fDecimationFactor == fDecimationOffset)) { 300 301 unsigned int decimatorNumSubChannels = 0; 302 decimatorNumSubChannels = GetNumSubChannels(fDecimatorChannelName); 303 304 auto poller = NewPoller(fDecimatorChannelName); 305 poller->Poll(fPollTimeoutMS); 306 for (auto iSubChannel=0u; iSubChannel<decimatorNumSubChannels; ++iSubChannel) { 307 while (!NewStatePending()) { 308 if (poller->CheckOutput(fDecimatorChannelName, iSubChannel)) { 309 fair::mq::Parts decimatorParts = MessageUtil::Copy(*this, outParts); 310 if (Send(decimatorParts, fDecimatorChannelName, iSubChannel) > 0) { 311 // LOG(debug) << " successfully send to decimator " 312 // << iSubChannel << " " << fNumSend << " " << fDecimationFactor; 313 break; 314 } else { 315 auto h = reinterpret_cast<TimeFrame::Header*>(decimatorParts.At(0)->GetData()); 316 LOG(warn) << "Failed to enqueue to decimator output iSubChannel = " 317 << iSubChannel << " : TF = " << h->timeFrameId; 318 } 319 } 320 } 321 } 322 } 323 324 return; 325 } 326 327 328 void TimeFrameBuilder::SendTimeFrame( 329 fair::mq::Parts& outParts) 330 { 331 // send for Output 332 auto poller = NewPoller(fOutputChannelName); 333 int attempts = 0; 334 while (!NewStatePending()) { 335 poller->Poll(fPollTimeoutMS); 336 auto direction = fDirection % fNumDestination; 337 fDirection = direction + 1; 338 if (poller->CheckOutput(fOutputChannelName, direction)) { 339 // output ready 340 if (Send(outParts, fOutputChannelName, direction) > 0) { 341 ++fNumSend; 342 break; 343 } else { 344 auto h = reinterpret_cast<TimeFrame::Header*>(outParts.At(0)->GetData()); 345 LOG(warn) << "Failed to enqueue time frame : TF = " << h->timeFrameId; 346 } 347 } 348 if (fDiscardOutput) { 349 attempts++; 350 if (attempts >= fNumDestination) { 351 LOG(error) << "Failed to send time frame after checking all destinations." 352 << "Discarding data. Direction = " << direction; 353 break; 354 } 355 } 356 if (fNumDestination==1) { 357 std::this_thread::sleep_for(std::chrono::milliseconds(1)); 358 } 359 } 360 361 return; 362 } 363 364 365 void TimeFrameBuilder::SendTimeFrameToEvery( 366 fair::mq::Parts& outParts) 367 { 368 369 // for DQM 370 bool dqmSocketExists = fChannels.count(fDQMChannelName); 371 if (dqmSocketExists) { 372 SendTimeFrameForDQM(outParts); 373 } 374 375 // for Decimator 376 if (fDecimatorNumberOfConnectedPeers > 0) { 377 SendTimeFrameForDecimator(outParts); 378 } 379 380 // for out ; outParts will be moved to ZeroMQ. 381 SendTimeFrame(outParts); 382 fNumSccesssfulTFB++; 383 384 return; 385 } 386 387 388 //______________________________________________________________________________ 389 bool TimeFrameBuilder::ConditionalRun() 390 { 391 namespace STF = SubTimeFrame; 392 namespace TF = TimeFrame; 393 394 // receive 395 fair::mq::Parts inParts; 396 if (Receive(inParts, fInputChannelName, 0, 1) > 0) { 397 assert(inParts.Size() >= 2); 398 399 auto stfHeader = reinterpret_cast<STF::Header*>(inParts.At(0)->GetData()); 400 auto stfId = stfHeader->timeFrameId; 401 402 #if 1 403 LOG(debug4) << "msg size: " << inParts.Size() 404 << ", TFid: "<< stfId 405 << " FEid: 0x" << std::hex << stfHeader->femId 406 << " Type: " << std::dec << stfHeader->type; 407 #endif 408 409 if (fEnableCheckHBF) { 410 CheckHBFDelimitor(inParts, stfId); 411 } 412 413 if (fTFBuffer.find(stfId) == fTFBuffer.end()) { 414 fTFBuffer[stfId].reserve(fNumSource); 415 } 416 fTFBuffer[stfId].emplace_back(STFBuffer {std::move(inParts), std::chrono::steady_clock::now()}); 417 418 #if 0 419 LOG(debug4) << "TFid: " << stfId 420 << ", Fragments: " << fTFBuffer[stfId].size() << "/" << fNumSource; 421 #endif 422 423 } 424 425 // send 426 if (!fTFBuffer.empty()) { 427 428 // find time frame in ready 429 for (auto itr = fTFBuffer.begin(); itr != fTFBuffer.end();) { 430 auto stfId = itr->first; 431 auto& tfBuf = itr->second; 432 433 if (tfBuf.size() == static_cast<long unsigned int>(fNumSource)) { 434 LOG(debug4) << "All comes : " << tfBuf.size() << " stfId: "<< stfId ; 435 436 if (fFirstTFB) { 437 fFirstTFB = false; 438 StoreSuccessfulSegmentIDs(tfBuf); 439 } 440 441 fair::mq::Parts outParts; 442 MakeOutPartsFromSTFBuffers(tfBuf, stfId, TimeFrame::TF_COMPLETE, outParts); 443 tfBuf.clear(); 444 445 SendTimeFrameToEvery(outParts); 446 447 } else { 448 // Timeout processing for incomplete time frame 449 auto dt = std::chrono::steady_clock::now() - tfBuf.front().start; 450 if (std::chrono::duration_cast<std::chrono::milliseconds>(dt).count() > fBufferTimeoutInMs) { 451 // output debug info 452 //TFBFailDump(tfBuf, stfId); 453 454 #if 0 455 std::cout << "#D Succeeded segments for TF " << stfId << ": " << std::dec; 456 for (auto & id : fSuccessfulSegmentIDs) { 457 std::cout << id << " (" << (id & 0xff) << ") "; 458 } 459 std::cout << std::endl; 460 #endif 461 462 std::vector<uint32_t> lostSegmentIds = CheckLostSegmentIDs(tfBuf); 463 std::string lostSegmentIdsStr; 464 for (auto & id : lostSegmentIds) { 465 lostSegmentIdsStr += std::to_string(id) + "(" + std::to_string(id & 0xff) + ") "; 466 } 467 LOG(warn) << "TFB Timeout: TF ID: " << stfId << ", Lost Segment ID: " << lostSegmentIdsStr; 468 469 // output incomplete TF 470 if (fOutputIncompleteTF) { 471 472 std::cout << "#D Output incomplete TF: " << stfId << std::endl; 473 474 fair::mq::Parts outParts; 475 MakeOutPartsFromSTFBuffers(tfBuf, stfId, 476 TimeFrame::TF_INCOMPLETE | TimeFrame::TF_TIMEOUT, 477 outParts); 478 SendTimeFrameToEvery(outParts); 479 } 480 481 fNumFailedTFB++; 482 tfBuf.clear(); 483 } 484 } 485 486 // remove empty buffer 487 if (tfBuf.empty()) { 488 itr = fTFBuffer.erase(itr); 489 } 490 else { 491 ++itr; 492 } 493 } 494 } // send 495 496 // erase a depth limint exceeded TF. 497 if (fBufferDepthLimit > 0) { 498 if (fTFBuffer.size() > fBufferDepthLimit) { 499 500 #if 1 501 std::cout << "#D Buffer Size: " << fTFBuffer.size() 502 << ": Top TFN: " << fTFBuffer.begin()->first 503 << " Size: " << fTFBuffer.begin()->second.size(); 504 auto& tfBuf = fTFBuffer.begin()->second; 505 506 for (auto & stfBuf : tfBuf) { 507 auto & msg = stfBuf.parts[0]; 508 SubTimeFrame::Header *stfheader 509 = reinterpret_cast<SubTimeFrame::Header *>(msg.GetData()); 510 std::cout << " " << std::dec << (stfheader->femId & 0xff); 511 } 512 std::cout << std::endl; 513 #endif 514 515 std::vector<uint32_t> lostSegmentIds = CheckLostSegmentIDs(tfBuf); 516 std::string lostSegmentIdsStr; 517 for (auto & id : lostSegmentIds) { 518 lostSegmentIdsStr += std::to_string(id) + "(" + std::to_string(id & 0xff) + ") "; 519 } 520 LOG(warn) << "Buffer overflow: TF ID: " << fTFBuffer.begin()->first 521 << ", Lost segment ID: " << lostSegmentIdsStr; 522 523 if (fOutputIncompleteTF) { 524 uint32_t stfId = fTFBuffer.begin()->first; 525 auto & tfBuf = fTFBuffer.begin()->second; 526 fair::mq::Parts outParts; 527 MakeOutPartsFromSTFBuffers(tfBuf, 528 stfId, TimeFrame::TF_INCOMPLETE | TimeFrame::TF_DEPTH_LIMIT, 529 outParts); 530 SendTimeFrameToEvery(outParts); 531 } 532 533 fNumFailedTFB++; 534 fTFBuffer.begin()->second.clear(); 535 fTFBuffer.erase(fTFBuffer.begin()); 536 } 537 } 538 539 if (fKt.Check()) { 540 static int lcounts = 0; 541 //if ((lcounts++ % 5) == 0) { 542 if ((lcounts++ % 1) == 0) { 543 544 LOG(debug) << "TimeFrameBuilding Successful: " << fNumSccesssfulTFB 545 << ", Failed: " << fNumFailedTFB; 546 547 double successfulRatio = 0; 548 if (fNumSccesssfulTFB + fNumFailedTFB > 0) { 549 successfulRatio 550 = fNumSccesssfulTFB / static_cast<double>(fNumSccesssfulTFB + fNumFailedTFB) 551 * 100.0; 552 } 553 #if 0 554 time_t now = std::time(nullptr); 555 std::cout << "#D " 556 << fKeyPrefixMetric + gKeySuccessfulRatio + " " 557 << now 558 << " TFB Successful Ratio: " 559 << std::to_string(successfulRatio) << std::endl; 560 #endif 561 562 fDbMetric->ts_add( 563 fKeyPrefixMetricTs + gKeySuccessfulRatio, 564 std::to_string( 565 std::chrono::duration_cast<std::chrono::milliseconds>( 566 std::chrono::system_clock::now().time_since_epoch() 567 ).count()), 568 std::to_string(successfulRatio)); 569 570 fNumSccesssfulTFB = 0; 571 fNumFailedTFB = 0; 572 } 573 574 static bool flagLostSegment = false; 575 if ((lcounts++ % 1) == 0) { 576 if (fLostSegmentCounts.size() > 0) { 577 LOG(debug) << "Lost segment counts: "; 578 fDbMetric->del(fKeyPrefixMetric + gKeyLostSegments); 579 for (const auto& [femid, lostcounts] : fLostSegmentCounts) { 580 #if 0 581 LOG(debug) << " FEM ID: " << femid << " (" << (femid & 0xff) << ")" 582 << ", Counts: " << lostcounts; 583 #endif 584 std::string femid_str = std::to_string(femid); 585 std::string lostcounts_str = std::to_string(lostcounts); 586 try { 587 fDbMetric->hset(fKeyPrefixMetric + gKeyLostSegments, femid_str, lostcounts_str); 588 } catch(const sw::redis::Error &e) { 589 LOG(error) << "Lost segment, DB write fail : " << e.what(); 590 } 591 flagLostSegment = true; 592 } 593 fLostSegmentCounts.clear(); 594 } else { 595 if (flagLostSegment) { 596 fDbMetric->del(fKeyPrefixMetric + gKeyLostSegments); 597 flagLostSegment = false; 598 } 599 } 600 } 601 602 } 603 604 return true; 605 } 606 607 608 //______________________________________________________________________________ 609 void TimeFrameBuilder::Init() 610 { 611 } 612 613 void TimeFrameBuilder::SetKeyPrefix() 614 { 615 std::string serverUri; 616 serverUri = fConfig->GetProperty<std::string>("metrics-uri"); 617 if (serverUri.empty()) { 618 serverUri = fConfig->GetProperty<std::string>("registry-uri"); 619 } 620 if (serverUri.empty()) { 621 LOG(error) << "DB server URI for metrics is not specified."; 622 throw std::runtime_error("DB server URI for metrics is not specified."); 623 } 624 625 std::string serviceRegistryUri = fConfig->GetProperty<std::string>("registry-uri"); 626 LOG(debug) << "DB Server URI: " << serverUri; 627 LOG(debug) << "DB serviceRegistoryUri: " << serviceRegistryUri; 628 629 std::string service_name = fConfig->GetProperty<std::string>("service-name"); 630 std::string separator = fConfig->GetProperty<std::string>("separator"); 631 fKeyPrefixMetricTs = "ts" + separator + fId + separator; 632 fKeyPrefixMetric = "metrics" + separator + fId + separator; 633 fKeyPrefixObjects = "dqm" + separator + fId + separator; 634 635 636 if (fDbMetric == nullptr) { 637 fDbUriMetric = serverUri; 638 fDbMetric = std::make_unique<RedisDataStore>(fDbUriMetric); 639 } 640 641 if (fDbObjects == nullptr) { 642 fDbUriObjects = serverUri.substr(0, serverUri.rfind("/") + 1) 643 + fConfig->GetProperty<std::string>(OptionKey::ObjectDbNumber.data()); 644 fDbObjects = std::make_unique<RedisDataStore>(fDbUriObjects); 645 } 646 647 LOG(info) << "Metric DB: " << fDbUriMetric << " Prefix: " 648 << fKeyPrefixMetric << " " << fKeyPrefixMetricTs; 649 LOG(info) << "DQM DB: " << fDbUriObjects << " Prefix:" << fKeyPrefixObjects; 650 651 /* 652 if (!serverUri.empty()) { 653 fClient = std::make_shared<sw::redis::Redis>(serverUri); 654 } 655 */ 656 657 return; 658 } 659 660 661 //______________________________________________________________________________ 662 void TimeFrameBuilder::InitTask() 663 { 664 using opt = OptionKey; 665 auto sBufferTimeoutInMs = fConfig->GetProperty<std::string>(opt::BufferTimeoutInMs.data()); 666 fBufferTimeoutInMs = std::stoi(sBufferTimeoutInMs); 667 668 auto sBufferDepthLimit = fConfig->GetProperty<std::string>(opt::BufferDepthLimit.data()); 669 fBufferDepthLimit = std::stoi(sBufferDepthLimit); 670 671 fInputChannelName = fConfig->GetProperty<std::string>(opt::InputChannelName.data()); 672 fOutputChannelName = fConfig->GetProperty<std::string>(opt::OutputChannelName.data()); 673 fDecimatorChannelName = fConfig->GetProperty<std::string>(opt::DecimatorChannelName.data()); 674 675 auto numSubChannels = GetNumSubChannels(fInputChannelName); 676 LOG(debug) << " Nsubchnnael[" << fInputChannelName << "]: " << numSubChannels; 677 678 fNumSource = 0; 679 for (auto i=0u; i<numSubChannels; ++i) { 680 fNumSource += GetNumberOfConnectedPeers(fInputChannelName, i); 681 } 682 683 fDQMChannelName = fConfig->GetProperty<std::string>(opt::DQMChannelName.data()); 684 685 LOG(debug) << " input channel : name = " << fInputChannelName 686 << " num = " << GetNumSubChannels(fInputChannelName) 687 << " num peer = " << GetNumberOfConnectedPeers(fInputChannelName,0); 688 689 LOG(debug) << " number of source = " << fNumSource; 690 691 fNumDestination = GetNumSubChannels(fOutputChannelName); 692 fPollTimeoutMS = std::stoi(fConfig->GetProperty<std::string>(opt::PollTimeout.data())); 693 fDecimationFactor = std::stoi(fConfig->GetProperty<std::string>(opt::DecimationFactor.data())); 694 fDecimationOffset = std::stoi(fConfig->GetProperty<std::string>(opt::DecimationOffset.data())); 695 if ((fDecimationFactor > 0) && (fDecimationOffset >= fDecimationFactor)) { 696 LOG(warn) << "invalid decimation-offset = " << fDecimationOffset 697 << " (< decimation-factor). Set to 0"; 698 fDecimationOffset = 0; 699 } 700 LOG(debug) << " decimation-factor = " << fDecimationFactor 701 << " decimation-offset = " << fDecimationOffset; 702 703 fDecimatorNumberOfConnectedPeers = 0; 704 if (fChannels.count(fDecimatorChannelName) > 0) { 705 auto nsub = GetNumSubChannels(fDecimatorChannelName); 706 for (auto i = 0u; i<nsub; ++i) { 707 fDecimatorNumberOfConnectedPeers += GetNumberOfConnectedPeers(fDecimatorChannelName,i); 708 } 709 } 710 711 auto sDiscardOutput = fConfig->GetProperty<std::string>(opt::DiscardOutput.data()); 712 fDiscardOutput = ((sDiscardOutput == "1") || (sDiscardOutput == "true") || (sDiscardOutput == "yes")); 713 LOG(debug) << " discard-output = " << fDiscardOutput; 714 715 std::string sOutputIncompleteTF = fConfig->GetProperty<std::string>(opt::OutputIncompleteTF.data()); 716 fOutputIncompleteTF = ((sOutputIncompleteTF == "1") || (sOutputIncompleteTF == "true") || (sOutputIncompleteTF == "yes")); 717 LOG(debug) << " output-incomplete-tf = " << fOutputIncompleteTF; 718 719 std::string sEnableCheckHBF = fConfig->GetProperty<std::string>(opt::EnableCheckHBF.data()); 720 fEnableCheckHBF = ((sEnableCheckHBF == "1") || (sEnableCheckHBF == "true") || (sEnableCheckHBF == "yes")); 721 LOG(debug) << " enable-check-hbf = " << fEnableCheckHBF; 722 723 SetKeyPrefix(); 724 725 } 726 727 728 //______________________________________________________________________________ 729 void TimeFrameBuilder::PostRun() 730 { 731 fTFBuffer.clear(); 732 733 int nrecv=0; 734 if (fChannels.count(fInputChannelName) > 0) { 735 auto n = fChannels.count(fInputChannelName); 736 737 for (auto i = 0u; i < n; ++i) { 738 std::cout << "#D SubChannel : "<< i << std::endl; 739 while(true) { 740 fair::mq::Parts part; 741 if (Receive(part, fInputChannelName, i, 1000) <= 0) { 742 LOG(debug) << __func__ << " no data received " << nrecv; 743 ++nrecv; 744 if (nrecv > 10) { 745 break; 746 } 747 std::this_thread::sleep_for(std::chrono::milliseconds(200)); 748 } else { 749 // LOG(debug) << __func__ << " data comes.."; 750 } 751 } 752 } 753 }// for clean up 754 } 755 756 757 //_____________________________________________________________________________ 758 void TimeFrameBuilder::PreRun() 759 { 760 fDirection = 0; 761 fNumSend = 0; 762 fFirstTFB = true; 763 fNumSccesssfulTFB = 0; 764 fNumFailedTFB = 0; 765 fLostSegmentCounts.clear(); 766 }
TimeFrameHeader.h
SHA-256: 6990b35bf8f19c749c2aeeab09571d7f1f487318a1f273d0d1539144958e7375
1 #ifndef TimeFrameHeader_h 2 #define TimeFrameHeader_h 3 4 #include <cstdint> 5 6 namespace TimeFrame { 7 8 // This format is temporary and should be updated. 9 namespace v0 { 10 // "DAEH-FT@" : little endian of "@TF-HEAD" 11 constexpr uint64_t MAGIC {0x444145482d465440}; 12 struct Header { 13 uint64_t magic {MAGIC}; 14 uint32_t timeFrameId {0}; 15 uint32_t numSource {0}; 16 uint64_t length {0}; 17 }; 18 19 } // namespace v0 20 21 inline 22 namespace v1 { 23 // " MRFEMIT" : little endian of "TIMEFRM " 24 constexpr uint64_t MAGIC {0x004d5246454d4954}; 25 // TYPE : upper 8 bits are used for the framework, 26 // lower 8 bits are used for experimental usage. 27 constexpr uint16_t META {0x0001}; 28 constexpr uint16_t SLICE {0x0002}; 29 constexpr uint16_t TF_COMPLETE {0x0000}; 30 constexpr uint16_t TF_INCOMPLETE {0x8000}; 31 constexpr uint16_t TF_TIMEOUT {0x4000}; 32 constexpr uint16_t TF_DEPTH_LIMIT {0x2000}; 33 34 #pragma pack(2) 35 struct Header { 36 uint64_t magic {MAGIC}; 37 uint32_t length {0}; 38 uint16_t hLength {24}; 39 uint16_t type {0}; 40 uint32_t timeFrameId {0}; 41 uint32_t numSource {0}; 42 43 44 void Print() { 45 printf("TimeFrameHeader\n"); 46 printf("Length = %d\n",length); 47 printf("Header Lenght = %d\n",hLength); 48 printf("Type = %d\n",type); 49 printf("timeframeid = %d\n",timeFrameId); 50 printf("numSource = %d\n",numSource); 51 } 52 }; 53 54 } // namespace v1 55 56 } // namespace TimeFrame 57 58 #endif
TimeFrameSlicerByLogicTiming.cxx
SHA-256: d18453b8409d6bbcaf61e5ec662655ff5122e50e3a5acfee24b3dd5ec2ac7f68
1 /* 2 * @file TimeFrameSlicerByLogicTiming 3 * @brief Slice Timeframe by Logic timing for NestDAQ 4 * @date Created : 2024-05-04 12:31:55 JST 5 * Last Modified : 2024-07-04 19:38:05 JST 6 * 7 * @author Shinsuke OTA <ota@rcnp.osaka-u.ac.jp> 8 * 9 */ 10 #include "TimeFrameSlicerByLogicTiming.h" 11 #include "fairmq/runDevice.h" 12 13 #include "utility/MessageUtil.h" 14 #include "UnpackTdc.h" 15 16 #define DEBUG 0 17 18 using nestdaq::TimeFrameSlicerByLogicTiming; 19 namespace bpo = boost::program_options; 20 21 22 TimeFrameSlicerByLogicTiming::TimeFrameSlicerByLogicTiming() 23 : fKTimer(1,KTimer(1000)) 24 { 25 } 26 27 28 void TimeFrameSlicerByLogicTiming::InitTask() 29 { 30 using opt = OptionKey; 31 32 fInputChannelName = fConfig->GetValue<std::string>(opt::InputChannelName.data()); 33 fOutputChannelName = fConfig->GetValue<std::string>(opt::OutputChannelName.data()); 34 fDQMChannelName = fConfig->GetValue<std::string>(opt::DQMChannelName.data()); 35 //fOffset[0] = fConfig->GetValue<int>(opt::TimeOffsetBegin.data()); 36 //fOffset[1] = fConfig->GetValue<int>(opt::TimeOffsetEnd.data()); 37 fOffset[0] = std::stoi(fConfig->GetValue<std::string>(opt::TimeOffsetBegin.data())); 38 fOffset[1] = std::stoi(fConfig->GetValue<std::string>(opt::TimeOffsetEnd.data())); 39 LOG(info) << "InitTask : Input Channel = " << fInputChannelName ; 40 LOG(info) << "InitTask : Output Channel = " << fOutputChannelName; 41 LOG(info) << "InitTask : DQM Channel = " << fDQMChannelName; 42 LOG(info) << "Time Window [" << fOffset[0] << " : " << fOffset[1] << "]"; 43 44 45 // identity 46 fName = fConfig->GetProperty<std::string>("id"); 47 std::istringstream ss(fName.substr(fName.rfind("-") + 1)); 48 ss >> fId; 49 50 51 fNumDestination = GetNumSubChannels(fOutputChannelName); 52 fPollTimeoutMS = std::stoi(fConfig->GetProperty<std::string>(opt::PollTimeout.data())); 53 54 } 55 56 void TimeFrameSlicerByLogicTiming::PreRun() 57 { 58 // fTF = nullptr; 59 // fSTF.resize(0); 60 // fHBF.resize(0); 61 } 62 63 bool TimeFrameSlicerByLogicTiming::ConditionalRun() 64 { 65 fEOM = false; 66 fNextIdx = 0; 67 fNumHBF = INT_MAX; 68 fIdxHBF = 0; 69 fDoCheck = fKTimer[0].Check(); 70 #if DEBUG 71 if (fDoCheck) LOG(info) << "Let's build"; 72 #endif 73 std::chrono::system_clock::time_point sw_start, sw_end; 74 75 76 FairMQParts inParts; 77 FairMQParts outParts; 78 if (Receive(inParts, fInputChannelName,0,1) <= 0) return true; 79 // inParts should have at least two messages 80 81 #if DEBUG 82 if (fDoCheck) { 83 LOG(info) << "inParts : " << inParts.Size(); 84 } 85 #endif 86 ParseMessages(inParts); 87 88 using copyUnit = uint32_t; 89 //---------------------------------------------------------------------- 90 // make slices 91 //---------------------------------------------------------------------- 92 while (fIdxHBF < fNumHBF) { 93 auto outdataptr = std::make_unique<std::vector<copyUnit>>(); 94 auto outdata = outdataptr.get(); 95 96 /*********************************************************************** 97 * Structure of timeframe slices 98 * TFH (meta info) 99 * FltH 100 * TDCs 101 * STFH 102 * HBFH 103 * HBDs 104 * STFH 105 * HBFH 106 * HBDs 107 * TFH (sliced data) 108 * STFH 109 * HBFH 110 * TDCs 111 * STFH 112 * HBFH 113 * TDCs 114 ***********************************************************************/ 115 116 //---------------------------------------------------------------------- 117 // make summary block (meta) 118 //---------------------------------------------------------------------- 119 120 // copy time frame header 121 fTFHidx = 0; 122 fTF.CopyHeaderTo<copyUnit>(outdata); 123 // copy filter frame 124 auto lfhidx = outdata->size(); 125 fLF.CopyHeaderTo<copyUnit>(outdata); 126 127 // use only one frame 128 fLF[fIdxHBF]->CopyHeaderTo<copyUnit>(outdata); 129 fLF[fIdxHBF]->CopyDataTo<copyUnit>(outdata); 130 131 auto lfh = (Filter::Header*) &((*outdata)[lfhidx]); 132 lfh->length = (outdata->size() - lfhidx) * sizeof(copyUnit); 133 lfh->numTrigs = fLF[fIdxHBF]->GetNumData(); 134 135 // copy heart beat delimiters 136 for (uint32_t is = 0, ns = fTF.size(); is < ns; ++is) { 137 auto stfhidx = outdata->size(); 138 fTF[is]->CopyHeaderTo<copyUnit>(outdata); 139 auto hbfhidx = outdata->size(); 140 fTF[is]->at(fIdxHBF)->CopyHeaderTo<copyUnit>(outdata); 141 // store hbds 142 auto hbf = fTF[is]->at(fIdxHBF); 143 if (fTF[is]->GetHeader()->type == SubTimeFrame::META) { 144 hbf->CopyDataTo<copyUnit>(outdata); 145 } else { 146 hbf->CopyDataTo<copyUnit>(outdata,hbf->GetNumData()-2); 147 hbf->CopyDataTo<copyUnit>(outdata,hbf->GetNumData()-1); 148 } 149 // modify size of subtime frame 150 auto stfh = (SubTimeFrame::Header*) &((*outdata)[stfhidx]); 151 stfh->length = (outdata->size() - stfhidx) * sizeof(copyUnit); 152 stfh->hLength = sizeof(struct SubTimeFrame::Header); 153 stfh->type = SubTimeFrame::META; 154 // modify size of heartbeat frame 155 auto hbfh = (HeartbeatFrame::Header*) &((*outdata)[hbfhidx]); 156 hbfh->length = (outdata->size() - hbfhidx) * sizeof(copyUnit); 157 } 158 159 160 // modify header information 161 auto tfh = (TimeFrame::Header*) &((*outdata)[0]); 162 tfh->type = TimeFrame::META; 163 tfh->length = outdata->size()*sizeof(copyUnit); 164 165 /*********************************************************************** 166 * Prepare the slices 167 ***********************************************************************/ 168 #if DEBUG 169 if (fDoCheck) { 170 LOG(info) << "Num Sources = " << fTF.size() << " / " << fTF.GetHeader()->numSource; 171 LOG(info) << " Num HB = " << fNumHBF << " / " << fTF[0]->GetHeader()->numMessages; 172 LOG(info) << " Num LF = " << fLF.size(); 173 LOG(info) << " Num Data in " << fIdxHBF << " = " << fLF[fIdxHBF]->GetNumData(); 174 for (int i = 0,n = std::min(fLF[fIdxHBF]->GetNumData(),(uint64_t)4); i<n; ++i) { 175 LOG(info) << " Trg[" << i << "] = " << fLF[fIdxHBF]->UncheckedAt(i); 176 } 177 } 178 #endif 179 //---------------------------------------------------------------------- 180 // start analyzing tdc data 181 //---------------------------------------------------------------------- 182 auto lf = fLF[fIdxHBF]; 183 auto nTrig = lf->GetNumData(); 184 std::vector<uint64_t> tdcidxs(fTF.size(),0); 185 std::vector<THBF*> hbfs(fTF.size()); 186 std::vector<uint64_t> femtype(fTF.size()); 187 struct TDC64L_V3::tdc64 tdc64l; 188 struct TDC64H_V3::tdc64 tdc64h; 189 190 for (int is = 0, ns = fTF.size(); is < ns; ++is) { 191 // skip meta frame 192 if (fTF[is]->GetHeader()->type == SubTimeFrame::META) continue; 193 194 hbfs[is] = fTF[is]->at(fIdxHBF); 195 femtype[is] = fTF[is]->GetHeader()->femType; 196 #if 1 // sort data 197 auto data = hbfs[is]->GetData(); 198 auto nData = hbfs[is]->GetNumData(); 199 if (femtype[is] == SubTimeFrame::TDC64H_V3) { 200 std::sort(data,data+nData-2, 201 [](auto const& lhs, auto const& rhs) { 202 struct TDC64H_V3::tdc64 tdc64h; 203 TDC64H_V3::Unpack(lhs,&tdc64h); 204 auto t1 = tdc64h.tdc; 205 TDC64H_V3::Unpack(rhs,&tdc64h); 206 auto t2 = tdc64h.tdc; 207 return t1 < t2; 208 }); 209 } else if (femtype[is] == SubTimeFrame::TDC64L_V3) { 210 std::sort(data,data+nData-2, 211 [](auto const& lhs, auto const& rhs) { 212 struct TDC64L_V3::tdc64 tdc64l; 213 TDC64L_V3::Unpack(lhs,&tdc64l); 214 auto t1 = tdc64l.tdc; 215 TDC64L_V3::Unpack(rhs,&tdc64l); 216 auto t2 = tdc64l.tdc; 217 return t1 < t2; 218 }); 219 } 220 221 #endif 222 } 223 224 // check overlap of two neighboring search window 225 for (uint32_t iTrig = 0; iTrig < nTrig; ++iTrig) { 226 auto trig = lf->UncheckedAt(iTrig).time; 227 auto trigBegin = trig + fOffset[0]; 228 auto trigEnd = trig + fOffset[1]; 229 bool hasOverlapWithNextTrigger = false; 230 if (iTrig < nTrig - 1) { 231 hasOverlapWithNextTrigger = ((lf->UncheckedAt(iTrig+1).time + fOffset[0]) < trigEnd); 232 } 233 234 #if DEBUG 235 if (fDoCheck) { 236 LOG(info) << "trigger window [" << trigBegin << "," << trigEnd << "] " << trig; 237 } 238 if (hasOverlapWithNextTrigger) { 239 LOG(info) << "Overlap in triggers" << (lf->UncheckedAt(iTrig+1).time + fOffset[0]) << "<" << trigEnd; 240 } 241 #endif 242 //---------------------------------------------------------------------- 243 // make timeframe for each slice 244 //---------------------------------------------------------------------- 245 auto tfidx = outdata->size(); 246 fTF.CopyHeaderTo<copyUnit>(outdata); 247 uint32_t numSource = 0; 248 for (int is = 0, ns = fTF.size(); is < ns; ++is) { 249 // skip meta subtime frame 250 if (fTF[is]->GetHeader()->type == SubTimeFrame::META) continue; 251 // otherwise increment numSources 252 numSource++; 253 //---------------------------------------------------------------------- 254 // make subtime frame header 255 //---------------------------------------------------------------------- 256 auto stfhidx = outdata->size(); 257 fTF[is]->CopyHeaderTo<copyUnit>(outdata); 258 259 260 auto hbf = hbfs[is]; 261 auto fem = femtype[is]; 262 263 //---------------------------------------------------------------------- 264 // make heartbeat frame header 265 //---------------------------------------------------------------------- 266 auto hbfhidx = outdata->size(); 267 hbf->CopyHeaderTo<copyUnit>(outdata); 268 269 //---------------------------------------------------------------------- 270 // comparison of tdc 271 //---------------------------------------------------------------------- 272 auto iKeep = tdcidxs[is]; 273 // nt ignores HBD 274 auto& it = tdcidxs[is], nt = hbf->GetNumData() - 2; 275 // it = 0; 276 while ( it < nt) { 277 int tdc4n = 0; 278 int ch = 0; 279 // decode data 280 if (fem == SubTimeFrame::TDC64H_V3) { 281 TDC64H_V3::Unpack(hbf->UncheckedAt(it),&tdc64h); 282 tdc4n = tdc64h.tdc4n; 283 ch = tdc64h.ch; 284 } else if (fem == SubTimeFrame::TDC64L_V3) { 285 TDC64L_V3::Unpack(hbf->UncheckedAt(it),&tdc64l); 286 tdc4n = tdc64l.tdc4n; 287 ch = tdc64l.ch; 288 } 289 // validate hits 290 if (tdc4n < trigBegin) { 291 iKeep++; 292 it++; 293 continue; 294 } 295 if (tdc4n > trigEnd) { 296 break; 297 } 298 // if (trigBegin <= tdc4n && tdc4n <= trigEnd) { 299 hbf->CopyDataTo<copyUnit>(outdata,it); 300 301 // } 302 it++; 303 } 304 // copy heartbeat frmae (20250205) 305 hbf->CopyDataTo<copyUnit>(outdata,nt); 306 hbf->CopyDataTo<copyUnit>(outdata,nt+1); 307 if (hasOverlapWithNextTrigger) { 308 // if the search window is overlap with next trigger, 309 // the index should be rewineded to indicate the first hit in this trigger window. 310 tdcidxs[is] = iKeep; 311 } 312 auto hbfh = (SubTimeFrame::Header*) &((*outdata)[hbfhidx]); 313 hbfh->length = (outdata->size() - hbfhidx) * sizeof(copyUnit); 314 auto stfh = (SubTimeFrame::Header*) &((*outdata)[stfhidx]); 315 stfh->length = (outdata->size() - stfhidx) * sizeof(copyUnit); 316 stfh->hLength = sizeof(struct SubTimeFrame::Header); 317 stfh->numMessages = 2; // by definition the number of heartbeat frame is one and then num message is one 318 } 319 320 auto tfsh = (TimeFrame::Header*) &((*outdata)[tfidx]); 321 tfsh->type = TimeFrame::SLICE; 322 tfsh->numSource = numSource; 323 tfsh->length = (outdata->size() - tfidx) * sizeof(copyUnit); 324 } 325 326 //---------------------------------------------------------------------- 327 // store to output parts 328 //---------------------------------------------------------------------- 329 FairMQMessagePtr msgall(MessageUtil::NewMessage(*this,std::move(outdataptr))); 330 outParts.AddPart(std::move(msgall)); 331 // go to next 332 fIdxHBF++; 333 } // while 334 335 //////////////////////////////////////////////////// 336 // Transfer the data to dqm port 337 //////////////////////////////////////////////////// 338 if (fDoCheck) { 339 LOG(info) << "numCount(" << fDQMChannelName << ") = " << fChannels.count(fDQMChannelName); 340 LOG(info) << "numCount(" << fOutputChannelName << ") = " << fChannels.count(fOutputChannelName); 341 } 342 343 FairMQParts dqmParts; 344 if (int nSubChan = fChannels.count(fDQMChannelName)) { 345 for (auto& m: outParts) { 346 FairMQMessagePtr msgCopy(fTransportFactory->CreateMessage()); 347 msgCopy->Copy(*m); 348 dqmParts.AddPart(std::move(msgCopy)); 349 } 350 if (fDoCheck) { 351 LOG(info) << "Seinding " << dqmParts.Size() << " parts"; 352 for (int i = 0, n = dqmParts.Size(); i < n; ++i) { 353 LOG(info) << "Part[ " << i << "] :" << dqmParts[i].GetSize() << " bytes"; 354 } 355 } 356 if (Send(dqmParts,fDQMChannelName) < 0) { 357 if (NewStatePending()) { 358 LOG(info) << "Device is not RUNNING"; 359 } else { 360 LOG(error) << "Failed to enqueue time frame slice (DQM) " << std::endl; 361 } 362 } 363 } 364 365 //////////////////////////////////////////////////// 366 // Transfer the data to all of output channel 367 //////////////////////////////////////////////////// 368 auto poller = NewPoller(fOutputChannelName); 369 while (!NewStatePending()) { 370 auto direction = (fDirection++) % fNumDestination; 371 poller->Poll(fPollTimeoutMS); 372 if (poller->CheckOutput(fOutputChannelName, direction)) { 373 if (Send(outParts, fOutputChannelName, direction) > 0) { 374 // successfully sent 375 break; 376 } else { 377 LOG(error) << "Failed to queue output-channel"; 378 } 379 } 380 if (fNumDestination==1) { 381 std::this_thread::sleep_for(std::chrono::milliseconds(1)); 382 } 383 } 384 385 // for (uint32_t ipt = 0, npt = inParts.Size(); ipt < npt; ++ipt) { 386 // auto& part = inParts[ipt]; 387 // auto magic = *reinterpret_cast<uint64_t*>(part.GetData()); 388 // if (fDoCheck) { 389 // if (magic == TimeFrame::MAGIC) { 390 // LOG(info) << "part[" << std::setw(3) << ipt << "] = " << std::hex << magic; 391 // auto tfbhdr = reinterpret_cast<struct TimeFrame::Header*>(part.GetData()); 392 // tfbhdr->Print(); 393 // } else if (magic == Filter::MAGIC) { 394 // LOG(info) << "part[" << std::setw(3) << ipt << "] = " << std::hex << magic; 395 // auto hdr = reinterpret_cast<struct Filter::Header*>(part.GetData()); 396 // LOG(info) << "length = " << hdr->length; 397 // LOG(info) << "hLength = " << hdr->hLength; 398 // LOG(info) << "nTiming = " << (hdr->length - hdr->hLength)/sizeof(uint32_t); 399 // LOG(info) << "data ptr = " << (hdr + 1) << " ?? " << reinterpret_cast<uint8_t*>(part.GetData()) + hdr->hLength << " ?? " << part.GetData(); 400 // } 401 // } 402 // } 403 404 405 return true; 406 } 407 408 void TimeFrameSlicerByLogicTiming::PostRun() 409 { 410 int nrecv = 0; 411 while(true) { 412 FairMQMessagePtr msg(NewMessage()); 413 if (Receive(msg, fInputChannelName) <= 0) { 414 // LOG(debug) << __func__ << " no data received " << nrecv; 415 ++nrecv; 416 if (nrecv>10) { 417 break; 418 } 419 std::this_thread::sleep_for(std::chrono::milliseconds(200)); 420 } else { 421 // LOG(debug) << __func__ << " print data"; 422 // HandleData(msg, 0); 423 } 424 } 425 } 426 427 428 bool TimeFrameSlicerByLogicTiming::ParseMessages(FairMQParts& inParts) 429 { 430 #if DEBUG 431 432 if (fDoCheck) { 433 LOG(info) << "GetNextHBF"; 434 LOG(info) << "nParts = " << inParts.Size(); 435 } 436 #endif 437 if (fNextIdx >= inParts.Size()) { 438 LOG(info) << "no data is available"; 439 return !(fEOM = true); 440 } 441 int stfIdx = 0; // array index for stf 442 int hbfIdx = 0; // array index for hbf 443 int fltIdx = 0; // array index for flt 444 //for (auto& i = fNextIdx, n = inParts.Size(); i < n; ++i) { 445 auto n = inParts.Size(); 446 for (auto& i = fNextIdx; i < n; ++i) { 447 auto& part = inParts[i]; 448 auto magic = *reinterpret_cast<uint64_t*>(part.GetData()); 449 if (magic == TimeFrame::MAGIC) { 450 #if DEBUG 451 if (fDoCheck) { 452 LOG(info) << "analyze timeframe"; 453 } 454 #endif 455 stfIdx = -1; // reset stf index 456 fTF.SetHeader(part.GetData()); 457 auto ns = fTF.GetHeader()->numSource; 458 #if DEBUG 459 if (fDoCheck) { 460 LOG(info) << "timeframeid = " << fTF.GetHeader()->timeFrameId; 461 LOG(info) << "numSource = " << ns; 462 } 463 #endif 464 if (fTF.size() < ns) { 465 fTF.resize(ns); 466 for (decltype(ns) is = 0; is < ns; ++is) { 467 if (fTF[is] == nullptr) { 468 fTF[is] = new TSTF; 469 } 470 } 471 } 472 #if DEBUG 473 if (fDoCheck) { 474 LOG(info) << "analyze timeframe done"; 475 } 476 #endif 477 } else if (magic == Filter::MAGIC) { 478 fLF.SetHeader(part.GetData()); 479 auto nm = fLF.GetHeader()->numMessages; 480 #if DEBUG 481 if (fDoCheck) { 482 auto hdr = fLF.GetHeader(); 483 LOG(info) << "analyze filter tdc"; 484 LOG(info) << nm; 485 LOG(info) << hdr->length; 486 LOG(info) << hdr->hLength; 487 LOG(info) << hdr->numTrigs; 488 LOG(info) << hdr->workerId; 489 LOG(info) << hdr->numMessages; 490 LOG(info) << hdr->elapseTime; 491 } 492 #endif 493 if (fLF.size() < nm - 1) { 494 fLF.resize(nm); 495 for (decltype(nm) im = 0; im < nm; ++im) { 496 if (fLF[im] == nullptr) fLF[im] = new TTT; 497 } 498 } 499 } else if (magic == Filter::TDC_MAGIC) { 500 #if DEBUG 501 if (fDoCheck) { 502 LOG(info) << "analyze filter tdc"; 503 LOG(info) << fLF.size(); 504 LOG(info) << *(uint64_t*)((char*)part.GetData()); 505 LOG(info) << *(uint32_t*)((char*)part.GetData()+8); 506 LOG(info) << *(uint16_t*)((char*)part.GetData()+12); 507 LOG(info) << *(uint16_t*)((char*)part.GetData()+14); 508 LOG(info) << *(uint32_t*)((char*)part.GetData()+16); 509 LOG(info) << *(uint32_t*)((char*)part.GetData()+20); 510 LOG(info) << *(uint32_t*)((char*)part.GetData()+24); 511 LOG(info) << *(uint32_t*)((char*)part.GetData()+28); 512 } 513 #endif 514 auto& tdcf = *(fLF[fltIdx]); 515 tdcf.Set(part.GetData()); 516 fltIdx++; 517 } else if (magic == SubTimeFrame::MAGIC) { 518 hbfIdx = 0; // reset hbf index 519 stfIdx ++; 520 auto& stf = *(fTF[stfIdx]); 521 stf.SetHeader(part.GetData()); 522 uint32_t nh = stf.GetHeader()->numMessages; 523 #if DEBUG 524 if (fDoCheck) { 525 LOG(info) << stf.size() << " " << nh; 526 } 527 #endif 528 if (stf.size() < nh) { 529 stf.resize(nh); 530 #if DEBUG 531 if (fDoCheck) { 532 LOG(info) << stf.size() << " " << nh; 533 } 534 #endif 535 for (decltype(nh) ih = 0; ih < nh; ++ih) { 536 if (stf[ih] == nullptr) { 537 stf[ih] = new THBF; 538 } 539 } 540 } 541 #if DEBUG 542 if (fDoCheck) { 543 auto hdr = stf.GetHeader(); 544 LOG(info) << "stfIdx = " << stfIdx; 545 LOG(info) << "femtype = " << hdr->femType; 546 LOG(info) << "femId = " << hdr->femId; 547 LOG(info) << "numMessages = " << hdr->numMessages; 548 } 549 #endif 550 } else if (magic == HeartbeatFrame::MAGIC) { 551 auto& hbf = *(fTF[stfIdx]->at(hbfIdx)); 552 hbf.Set(part.GetData()); 553 hbfIdx++; 554 } 555 } 556 fNumHBF = hbfIdx; 557 return true; 558 } 559 560 561 //////////////////////////////////////////////////// 562 // override runDevice 563 //////////////////////////////////////////////////// 564 565 void addCustomOptions(bpo::options_description& options) 566 { 567 using opt = TimeFrameSlicerByLogicTiming::OptionKey; 568 569 options.add_options() 570 (opt::InputChannelName.data(), 571 bpo::value<std::string>()->default_value("in"), 572 "Name of the input channel") 573 (opt::OutputChannelName.data(), 574 bpo::value<std::string>()->default_value("out"), 575 "Name of the output channel") 576 (opt::DQMChannelName.data(), 577 bpo::value<std::string>()->default_value("dqm"), 578 "Name of the data quality monitoring channel") 579 (opt::TimeOffsetBegin.data(), 580 //bpo::value<int>()->default_value(-100), 581 bpo::value<std::string>()->default_value("-100"), 582 "offset where window begins") 583 (opt::TimeOffsetEnd.data(), 584 //bpo::value<int>()->default_value(100), 585 bpo::value<std::string>()->default_value("100"), 586 "offset where window ends") 587 (opt::PollTimeout.data(), 588 bpo::value<std::string>()->default_value("1"), 589 "Timeout of polling (in msec)") 590 (opt::SplitMethod.data(), 591 bpo::value<std::string>()->default_value("1"), 592 "STF split method") 593 ; 594 595 } 596 597 598 std::unique_ptr<fair::mq::Device> getDevice(fair::mq::ProgOptions& /*config*/) 599 { 600 return std::make_unique<TimeFrameSlicerByLogicTiming>(); 601 } 602
UnpackTdc.h
SHA-256: 01c634487af5d527a9625e89af01b82ac83599e8f1d305cf854174b0f57e41c4
1 /* 2 * 3 * 4 */ 5 #ifndef UNPACK_TDC_H 6 #define UNPACK_TDC_H 7 #include <iostream> 8 #include <iomanip> 9 10 11 namespace TDC40 { 12 13 static constexpr unsigned int T_TDC = 0xd; 14 static constexpr unsigned int T_HB = 0xf; 15 static constexpr unsigned int T_ERROR = 0xe; 16 static constexpr unsigned int T_SPL_START = 0x1; 17 static constexpr unsigned int T_SPL_END = 0x4; 18 19 struct tdc40 { 20 int type; 21 int tot; 22 int flag; 23 int ch; 24 int tdc; 25 int hartbeat; 26 }; 27 28 void Rev5(unsigned char *val, unsigned char *rval) 29 { 30 rval[4] = val[0]; 31 rval[3] = val[1]; 32 rval[2] = val[2]; 33 rval[1] = val[3]; 34 rval[0] = val[4]; 35 36 return; 37 } 38 39 int Unpack(unsigned char *data, struct tdc40 *tdc) 40 { 41 tdc->type = (data[4] & 0xf0) >> 4; 42 if (tdc->type == T_TDC) { 43 tdc->tot = ((data[4] & 0x07) << 5) | ((data[3] & 0xf8) >> 3); 44 tdc->flag = (data[3] & 0x06) >> 1; 45 tdc->ch = ((data[3] & 0x01) << 5) | (data[2] & 0xf8) >> 3; 46 tdc->tdc = ((data[2] & 0x07) << 16) | (data[1] << 8) | data[0]; 47 tdc->hartbeat = 0; 48 } else 49 if ((tdc->type == T_HB) || (tdc->type == T_ERROR)) { 50 tdc->tot = -1; 51 tdc->flag = -1; 52 tdc->ch = -1; 53 tdc->tdc = -1; 54 tdc->hartbeat = ((data[1] << 8) | data[0]); 55 } else { 56 tdc->tot = -1; 57 tdc->flag = -1; 58 tdc->ch = -1; 59 tdc->tdc = -1; 60 tdc->hartbeat = 0; 61 } 62 63 return tdc->type; 64 } 65 } //namespace TDC40 66 67 68 namespace TDC64H { 69 70 struct tdc64 { 71 int type; 72 int ch; 73 int tot; 74 int tdc; 75 int tdc4n; 76 int flag; 77 int spill; 78 int hartbeat; 79 }; 80 81 static constexpr unsigned int T_TDC = (0x2c >> 2); 82 static constexpr unsigned int T_TDC_L = (0x2c >> 2); 83 static constexpr unsigned int T_TDC_T = (0x34 >> 2); 84 static constexpr unsigned int T_HB = (0x70 >> 2); 85 static constexpr unsigned int T_THR1_START = (0x64 >> 2); 86 static constexpr unsigned int T_THR1_END = (0x44 >> 2); 87 static constexpr unsigned int T_THR2_START = (0x68 >> 2); 88 static constexpr unsigned int T_THR2_END = (0x48 >> 2); 89 static constexpr unsigned int T_SPL_START = (0x60 >> 2); 90 static constexpr unsigned int T_SPL_END = (0x50 >> 2); 91 92 int Unpack(uint64_t data, struct tdc64 *tdc) 93 { 94 //unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 95 96 tdc->type = (data & 0xfc00'0000'0000'0000) >> 58; 97 if (tdc->type == T_TDC) { 98 tdc->ch = (data & 0x03f8'0000'0000'0000) >> 51; 99 tdc->tot = (data & 0x0007'ffff'e000'0000) >> 29; 100 tdc->tdc = (data & 0x0000'0000'1fff'ffff); 101 tdc->tdc4n = (data & 0x0000'0000'1fff'ffff) >> 12; 102 tdc->spill = -1; 103 tdc->hartbeat = -1; 104 } else 105 if (tdc->type == T_HB) { 106 tdc->ch = -1; 107 tdc->tot = -1; 108 tdc->tdc = -1; 109 tdc->tdc4n = -1; 110 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 111 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 112 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 113 } else 114 if ((tdc->type == T_SPL_START) || (tdc->type == T_SPL_END)) { 115 tdc->ch = -1; 116 tdc->tot = -1; 117 tdc->tdc = -1; 118 tdc->tdc4n = -1; 119 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 120 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 121 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 122 } else { 123 tdc->ch = -1; 124 tdc->tot = -1; 125 tdc->tdc = -1; 126 tdc->tdc4n = -1; 127 tdc->flag = -1; 128 tdc->spill = -1; 129 tdc->hartbeat = -1; 130 } 131 132 return tdc->type; 133 } 134 135 #if 0 136 int Unpack(uint64_t data, struct tdc64 *tdc) 137 { 138 unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 139 return Unpack(cdata, tdc); 140 } 141 #endif 142 143 int Unpack(unsigned char *data, struct tdc64 *tdc) 144 { 145 uint64_t *pdata = reinterpret_cast<uint64_t *>(data); 146 return Unpack(*pdata, tdc); 147 } 148 149 int GetHBFrame(unsigned char *pdata, unsigned char *pend, unsigned char **ppnext) 150 { 151 uint64_t *tdcword = reinterpret_cast<uint64_t *>(pdata); 152 struct tdc64 tdc; 153 int i = 0; 154 while (true) { 155 if (Unpack(tdcword[i++], &tdc) == T_HB) break; 156 if (reinterpret_cast<unsigned char*>(&(tdcword[i])) > pend) { 157 *ppnext = nullptr; 158 return 0; 159 } 160 } 161 i++; 162 *ppnext = reinterpret_cast<unsigned char *>(tdcword + i); 163 164 return i * sizeof(uint64_t); 165 } 166 } //namespace TDC64H 167 168 namespace TDC64L { 169 170 struct tdc64 { 171 int type; 172 int ch; 173 int tot; 174 int tdc; 175 int tdc4n; 176 int flag; 177 int spill; 178 int hartbeat; 179 }; 180 181 namespace v1 { 182 static constexpr unsigned int T_TDC = (0x2c >> 2); 183 static constexpr unsigned int T_HB = (0x70 >> 2); 184 static constexpr unsigned int T_SPL_START = (0x60 >> 2); 185 static constexpr unsigned int T_SPL_END = (0x50 >> 2); 186 187 int Unpack(uint64_t data, struct tdc64 *tdc) 188 { 189 //unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 190 191 tdc->type = (data & 0xfc00'0000'0000'0000) >> 58; 192 if (tdc->type == T_TDC) { 193 tdc->ch = (data & 0x03f8'0000'0000'0000) >> 51; 194 tdc->tot = (data & 0x0007'f800'0000'0000) >> 43; 195 tdc->tdc = (data & 0x0000'07ff'ff00'0000) >> 24; 196 tdc->tdc4n = (data & 0x0000'07ff'ff00'0000) >> 26; 197 tdc->flag = -1; 198 tdc->spill = -1; 199 tdc->hartbeat = -1; 200 } else 201 if (tdc->type == T_HB) { 202 tdc->ch = -1; 203 tdc->tot = -1; 204 tdc->tdc = -1; 205 tdc->tdc4n = -1; 206 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 207 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 208 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 209 } else 210 if ((tdc->type == T_SPL_START) || (tdc->type == T_SPL_END)) { 211 tdc->ch = -1; 212 tdc->tot = -1; 213 tdc->tdc = -1; 214 tdc->tdc4n = -1; 215 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 216 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 217 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 218 } else { 219 tdc->ch = -1; 220 tdc->tot = -1; 221 tdc->tdc = -1; 222 tdc->tdc4n = -1; 223 tdc->flag = -1; 224 tdc->spill = -1; 225 tdc->hartbeat = -1; 226 } 227 228 return tdc->type; 229 } 230 } //namespace v1 231 232 inline namespace v2 { 233 static constexpr unsigned int T_TDC = (0x2c >> 2); 234 static constexpr unsigned int T_TDC_L = (0x2c >> 2); 235 static constexpr unsigned int T_TDC_T = (0x34 >> 2); 236 static constexpr unsigned int T_THR1_START = (0x64 >> 2); 237 static constexpr unsigned int T_THR1_END = (0x44 >> 2); 238 static constexpr unsigned int T_THR2_START = (0x68 >> 2); 239 static constexpr unsigned int T_THR2_END = (0x48 >> 2); 240 static constexpr unsigned int T_HB = (0x70 >> 2); 241 static constexpr unsigned int T_HB1 = (0x70 >> 2); 242 static constexpr unsigned int T_HB2 = (0x78 >> 2); 243 static constexpr unsigned int T_SPL_START = (0x60 >> 2); 244 static constexpr unsigned int T_SPL_END = (0x50 >> 2); 245 246 int Unpack(uint64_t data, struct tdc64 *tdc) 247 { 248 //unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 249 250 tdc->type = (data & 0xfc00'0000'0000'0000) >> 58; 251 if ( (tdc->type == T_TDC_L) || (tdc->type == T_TDC_T) 252 || (tdc->type == T_THR1_START) || (tdc->type == T_THR1_END) 253 || (tdc->type == T_THR2_START) || (tdc->type == T_THR2_END) ) { 254 tdc->ch = (data & 0x03f8'0000'0000'0000) >> 51; 255 tdc->tot = (data & 0x0007'fff8'0000'0000) >> 35; 256 tdc->tdc = (data & 0x0000'0007'ffff'0000) >> 16; 257 tdc->tdc4n = (data & 0x0000'0007'fffc'0000) >> 18; 258 tdc->flag = -1; 259 tdc->spill = -1; 260 tdc->hartbeat = -1; 261 } else 262 if (tdc->type == T_HB) { 263 tdc->ch = -1; 264 tdc->tot = -1; 265 tdc->tdc = -1; 266 tdc->tdc4n = -1; 267 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 268 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 269 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 270 } else 271 if ((tdc->type == T_SPL_START) || (tdc->type == T_SPL_END)) { 272 tdc->ch = -1; 273 tdc->tot = -1; 274 tdc->tdc = -1; 275 tdc->tdc4n = -1; 276 tdc->flag = (data & 0x03ff'0000'0000'0000) >> 48; 277 tdc->spill = (data & 0x0000'ff00'0000'0000) >> 40; 278 tdc->hartbeat = (data & 0x0000'00ff'ff00'0000) >> 24; 279 } else { 280 tdc->ch = -1; 281 tdc->tot = -1; 282 tdc->tdc = -1; 283 tdc->tdc4n = -1; 284 tdc->flag = -1; 285 tdc->spill = -1; 286 tdc->hartbeat = -1; 287 } 288 289 return tdc->type; 290 } 291 } //namespace v2 292 293 } //namespace TDC64L 294 295 296 namespace TDC64H_V3 { 297 298 struct tdc64 { 299 int type; 300 int ch; 301 int tot; 302 int tdc; 303 int tdc4n; 304 int flag; 305 int toffset; 306 int genesize; 307 int transize; 308 int hartbeat; 309 }; 310 311 inline namespace v2 { 312 static constexpr unsigned int T_TDC = (0x2c >> 2); 313 static constexpr unsigned int T_TDC_L = (0x2c >> 2); 314 static constexpr unsigned int T_TDC_T = (0x34 >> 2); 315 static constexpr unsigned int T_THR1_START = (0x64 >> 2); 316 static constexpr unsigned int T_THR1_END = (0x44 >> 2); 317 static constexpr unsigned int T_THR2_START = (0x68 >> 2); 318 static constexpr unsigned int T_THR2_END = (0x48 >> 2); 319 static constexpr unsigned int T_HB = (0x70 >> 2); 320 static constexpr unsigned int T_HB1 = (0x70 >> 2); 321 static constexpr unsigned int T_HB2 = (0x78 >> 2); 322 static constexpr unsigned int T_SPL_START = (0x60 >> 2); 323 static constexpr unsigned int T_SPL_END = (0x50 >> 2); 324 325 int Unpack(uint64_t data, struct tdc64 *tdc) 326 { 327 //unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 328 329 tdc->type = (data & 0xfc00'0000'0000'0000) >> 58; 330 if ( (tdc->type == T_TDC_L) || (tdc->type == T_TDC_T) 331 || (tdc->type == T_THR1_START) || (tdc->type == T_THR1_END) 332 || (tdc->type == T_THR2_START) || (tdc->type == T_THR2_END) ) { 333 tdc->ch = (data & 0x03f8'0000'0000'0000) >> 51; 334 tdc->tot = (data & 0x0007'ffff'e000'0000) >> 29; 335 tdc->tdc = (data & 0x0000'0000'1fff'ffff); 336 tdc->tdc4n = (data & 0x0000'0000'1fff'ffff) >> 12; 337 tdc->flag = -1; 338 tdc->toffset = -1; 339 tdc->hartbeat = -1; 340 tdc->genesize = -1; 341 tdc->transize = -1; 342 } else 343 if (tdc->type == T_HB1) { 344 tdc->ch = -1; 345 tdc->tot = -1; 346 tdc->tdc = -1; 347 tdc->tdc4n = -1; 348 tdc->flag = (data & 0x00ff'ff00'0000'0000) >> 40; 349 tdc->toffset = (data & 0x0000'00ff'ff00'0000) >> 24; 350 tdc->hartbeat = (data & 0x0000'0000'00ff'ffff) >> 0; 351 tdc->genesize = -1; 352 tdc->transize = -1; 353 } else 354 if (tdc->type == T_HB2) { 355 tdc->ch = -1; 356 tdc->tot = -1; 357 tdc->tdc = -1; 358 tdc->tdc4n = -1; 359 tdc->flag = (data & 0x00ff'ff00'0000'0000) >> 40; 360 tdc->toffset = -1; 361 tdc->hartbeat = -1; 362 tdc->genesize = (data & 0x0000'00ff'fff0'0000) >> 20; 363 tdc->transize = (data & 0x0000'0000'000f'ffff) >> 0; 364 } else 365 { 366 tdc->ch = -1; 367 tdc->tot = -1; 368 tdc->tdc = -1; 369 tdc->tdc4n = -1; 370 tdc->flag = -1; 371 tdc->toffset = -1; 372 tdc->hartbeat = -1; 373 tdc->genesize = -1; 374 tdc->transize = -1; 375 } 376 377 return tdc->type; 378 } 379 } //namespace v2 380 381 } //namespace TDC64H_V3 382 383 namespace TDC64L_V3 { 384 385 struct tdc64 { 386 int type; 387 int ch; 388 int tot; 389 int tdc; 390 int tdc4n; 391 int flag; 392 int toffset; 393 int genesize; 394 int transize; 395 int hartbeat; 396 }; 397 398 inline namespace v2 { 399 static constexpr unsigned int T_TDC = (0x2c >> 2); 400 static constexpr unsigned int T_TDC_L = (0x2c >> 2); 401 static constexpr unsigned int T_TDC_T = (0x34 >> 2); 402 static constexpr unsigned int T_THR1_START = (0x64 >> 2); 403 static constexpr unsigned int T_THR1_END = (0x44 >> 2); 404 static constexpr unsigned int T_THR2_START = (0x68 >> 2); 405 static constexpr unsigned int T_THR2_END = (0x48 >> 2); 406 static constexpr unsigned int T_HB = (0x70 >> 2); 407 static constexpr unsigned int T_HB1 = (0x70 >> 2); 408 static constexpr unsigned int T_HB2 = (0x78 >> 2); 409 static constexpr unsigned int T_SPL_START = (0x60 >> 2); 410 static constexpr unsigned int T_SPL_END = (0x50 >> 2); 411 412 int Unpack(uint64_t data, struct tdc64 *tdc) 413 { 414 //unsigned char *cdata = reinterpret_cast<unsigned char *>(&data); 415 416 tdc->type = (data & 0xfc00'0000'0000'0000) >> 58; 417 if ( (tdc->type == T_TDC_L) || (tdc->type == T_TDC_T) 418 || (tdc->type == T_THR1_START) || (tdc->type == T_THR1_END) 419 || (tdc->type == T_THR2_START) || (tdc->type == T_THR2_END) ) { 420 tdc->ch = (data & 0x03fc'0000'0000'0000) >> 50; 421 tdc->tot = (data & 0x0003'fffc'0000'0000) >> 34; 422 tdc->tdc = (data & 0x0000'0003'ffff'8000) >> 15; 423 tdc->tdc4n = (data & 0x0000'0003'fffe'0000) >> 17; 424 tdc->flag = -1; 425 tdc->toffset = -1; 426 tdc->genesize = -1; 427 tdc->transize = -1; 428 tdc->hartbeat = -1; 429 } else 430 if (tdc->type == T_HB1) { 431 tdc->ch = -1; 432 tdc->tot = -1; 433 tdc->tdc = -1; 434 tdc->tdc4n = -1; 435 tdc->flag = (data & 0x00ff'ff00'0000'0000) >> 40; 436 tdc->toffset = (data & 0x0000'00ff'ff00'0000) >> 24; 437 tdc->hartbeat = (data & 0x0000'0000'00ff'ffff) >> 0; 438 tdc->genesize = -1; 439 tdc->transize = -1; 440 } else 441 if (tdc->type == T_HB2) { 442 tdc->ch = -1; 443 tdc->tot = -1; 444 tdc->tdc = -1; 445 tdc->tdc4n = -1; 446 tdc->flag = (data & 0x00ff'ff00'0000'0000) >> 40; 447 tdc->toffset = -1; 448 tdc->hartbeat = -1; 449 tdc->genesize = (data & 0x0000'00ff'fff0'0000) >> 20; 450 tdc->transize = (data & 0x0000'0000'000f'ffff) >> 0; 451 } else 452 { 453 tdc->ch = -1; 454 tdc->tot = -1; 455 tdc->tdc = -1; 456 tdc->tdc4n = -1; 457 tdc->flag = -1; 458 tdc->toffset = -1; 459 tdc->hartbeat = -1; 460 tdc->genesize = -1; 461 tdc->transize = -1; 462 } 463 464 return tdc->type; 465 } 466 } //namespace v2 467 468 } //namespace TDC64L3 469 470 471 #ifdef TEST_MAIN_UNPACKTDC 472 //#if 0 473 474 int tdc64h_dump(uint64_t data) 475 { 476 struct TDC64H::tdc64 tdc; 477 int type = TDC64H::Unpack(data, &tdc); 478 if (type == TDC64H::T_TDC) { 479 std::cout << "TDC " << std::dec 480 << " CH:" << tdc.ch 481 << " TOT:" << tdc.tot 482 << " TDC:" << tdc.tdc 483 << " TDC4n:" << tdc.tdc4n 484 << " : " << std::hex << data 485 << std::endl; 486 } else 487 if ((type == TDC64H::T_HB) 488 || (tdc.type == TDC64H::T_SPL_START) 489 || (tdc.type == TDC64H::T_SPL_END)) { 490 if (type ==TDC64H::T_HB) std::cout << "HB "; 491 if (type ==TDC64H::T_SPL_START) std::cout << "S_STA "; 492 if (type ==TDC64H::T_SPL_END) std::cout << "S_END "; 493 std::cout << std::hex 494 << " FLAG: " << tdc.flag 495 << " SPILL: " << tdc.spill 496 << " HERTBEAT: " << tdc.hartbeat 497 << " : " << std::hex << data 498 << std::dec << std::endl; 499 } else { 500 std::cerr << "Invalid data : " 501 << std::hex << data << std::dec << std::endl; 502 } 503 504 return type; 505 } 506 507 int main(int argc, char* argv[]) 508 { 509 static char cbuf[16]; 510 uint64_t *pdata = reinterpret_cast<uint64_t *>(cbuf); 511 512 uint64_t *buf = new uint64_t[1024*1024*8]; 513 514 int i = 0; 515 while (true) { 516 std::cin.read(cbuf, 8); 517 //std::cin >> *pdata; 518 if (std::cin.eof()) break; 519 buf[i++] = *pdata; 520 #if 0 521 std::cout << "\r " << i << ": " << *pdata << " " << std::flush; 522 #endif 523 524 tdc64h_dump(*pdata); 525 526 } 527 528 unsigned char *pcurr = reinterpret_cast<unsigned char *>(buf); 529 unsigned char *pend = reinterpret_cast<unsigned char *>(buf + i); 530 unsigned char *pnext = nullptr; 531 while (true) { 532 int len = TDC64H::GetHBFrame(pcurr, pend, &pnext); 533 if (len <= 0) break; 534 535 /* 536 std::cout << "#D HB frame size: " << std::dec << len 537 << " curr: " << std::hex 538 << reinterpret_cast<uintptr_t>(pcurr) 539 << " next: " 540 << reinterpret_cast<uintptr_t>(pnext) << std::endl; 541 */ 542 543 pdata = reinterpret_cast<uint64_t *>(pcurr); 544 545 for (unsigned int j = 0 ; j < len / sizeof(uint64_t) ; j++) { 546 tdc64h_dump(pdata[j]); 547 } 548 pcurr = pnext; 549 } 550 551 return 0; 552 } 553 #endif 554 555 #if 0 556 int main(int argc, char* argv[]) 557 { 558 static char cbuf[16]; 559 uint64_t *pdata = reinterpret_cast<uint64_t *>(cbuf); 560 struct TDC64H::tdc64 tdc; 561 562 while (true) { 563 std::cin.read(cbuf, 8); 564 //std::cin >> *pdata; 565 if (std::cin.eof()) break; 566 567 int type = TDC64H::Unpack(*pdata, &tdc); 568 if (type == TDC64H::T_TDC) { 569 std::cout << "TDC " << std::dec 570 << " CH:" << tdc.ch 571 << " TOT:" << tdc.tot 572 << " TDC:" << tdc.tdc 573 << " : " << std::hex << *pdata 574 << std::endl; 575 } else 576 if ((type == TDC64H::T_HB) 577 || (tdc.type == TDC64H::T_SPL_START) 578 || (tdc.type == TDC64H::T_SPL_END)) { 579 if (type ==TDC64H::T_HB) std::cout << "HB "; 580 if (type ==TDC64H::T_SPL_START) std::cout << "S_STA "; 581 if (type ==TDC64H::T_SPL_END) std::cout << "S_END "; 582 std::cout << std::hex 583 << " FLAG: " << tdc.flag 584 << " SPILL: " << tdc.spill 585 << " HERTBEAT: " << tdc.hartbeat 586 << " : " << std::hex << *pdata 587 << std::dec << std::endl; 588 } else { 589 std::cerr << "Invalid data : " 590 << std::hex << *pdata << std::dec << std::endl; 591 break; 592 } 593 } 594 595 return 0; 596 } 597 #endif 598 599 #if 0 600 int main(int argc, char* argv[]) 601 { 602 static unsigned char cbuf[16]; 603 //uint64_t *pdata = reinterpret_cast<uint64_t *>(cbuf); 604 struct TDC40::tdc40 tdc; 605 606 std::cin.read(reinterpret_cast<char *>(cbuf), 13); 607 while (true) { 608 std::cin.read(reinterpret_cast<char *>(cbuf), 5); 609 //std::cin >> *pdata; 610 if (std::cin.eof()) break; 611 612 int type = TDC40::Unpack(cbuf, &tdc); 613 if (type == TDC40::T_TDC) { 614 std::cout << "TDC " << std::dec 615 << " CH:" << std::setw(2) << tdc.ch 616 << " TOT:" << std::setw(3) << tdc.tot 617 << " TDC:" << std::setw(7) << tdc.tdc 618 << " : " << std::hex 619 << std::setw(2) << static_cast<unsigned int>(cbuf[4]) 620 << std::setw(2) << static_cast<unsigned int>(cbuf[3]) 621 << std::setw(2) << static_cast<unsigned int>(cbuf[2]) 622 << std::setw(2) << static_cast<unsigned int>(cbuf[1]) 623 << std::setw(2) << static_cast<unsigned int>(cbuf[0]) 624 << std::endl; 625 } else 626 if ((type == TDC40::T_HB) 627 || (tdc.type == TDC40::T_ERROR) 628 || (tdc.type == TDC40::T_SPL_START) 629 || (tdc.type == TDC40::T_SPL_END)) { 630 if (type ==TDC40::T_HB) std::cout << "HB "; 631 if (type ==TDC40::T_ERROR) std::cout << "ERR "; 632 if (type ==TDC40::T_SPL_START) std::cout << "S_STA "; 633 if (type ==TDC40::T_SPL_END) std::cout << "S_END "; 634 std::cout << std::dec 635 //<< " FLAG: " << tdc.flag 636 //<< " SPILL: " << tdc.spill 637 << " HERTBEAT: " << tdc.hartbeat 638 << " : " << std::hex 639 << std::setw(2) << static_cast<unsigned int>(cbuf[4]) 640 << std::setw(2) << static_cast<unsigned int>(cbuf[3]) 641 << std::setw(2) << static_cast<unsigned int>(cbuf[2]) 642 << std::setw(2) << static_cast<unsigned int>(cbuf[1]) 643 << std::setw(2) << static_cast<unsigned int>(cbuf[0]) 644 << std::dec << std::endl; 645 } else { 646 std::cout << "Invalid data : " << std::hex 647 << std::setw(2) << static_cast<unsigned int>(cbuf[4]) 648 << std::setw(2) << static_cast<unsigned int>(cbuf[3]) 649 << std::setw(2) << static_cast<unsigned int>(cbuf[2]) 650 << std::setw(2) << static_cast<unsigned int>(cbuf[1]) 651 << std::setw(2) << static_cast<unsigned int>(cbuf[0]) 652 << std::dec << std::endl; 653 //break; 654 } 655 } 656 657 return 0; 658 } 659 #endif 660 661 #if 0 662 int main(int argc, char* argv[]) 663 { 664 struct TDC40::tdc40 tdc; 665 666 unsigned char rawdata[] = { 667 0x00, 0x00, 0x80, 0x40, 0xd0, 668 0x11, 0x00, 0x80, 0x40, 0xd0, 669 0x22, 0x00, 0x80, 0x40, 0xd0, 670 0x33, 0x00, 0x80, 0x40, 0xd0, 671 0x44, 0x00, 0x80, 0x40, 0xd0, 672 0x02, 0x00, 0x00, 0x00, 0xf0 673 }; 674 675 676 for (int i = 0 ; i < 30 ; i += 5) { 677 TDC40::Unpack((rawdata + i), &tdc); 678 std::cout << " Head: " << std::hex << tdc.type 679 << " TOT: " << std::hex << tdc.tot 680 << " TYPE: " << std::hex << tdc.flag 681 << " CH: " << std::hex << tdc.ch 682 << " TDC: " << std::hex << tdc.tdc 683 << " HB: " << std::hex << tdc.hartbeat 684 << std::endl; 685 } 686 687 688 return 0; 689 } 690 #endif 691 #endif // UNPACK_TDC_H